diff --git a/AGENTS.md b/AGENTS.md index 3a67b1d9..e6e7674a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -83,6 +83,8 @@ Every commit must reference a relevant GitHub issue number in the title or body (e.g. `Fix target placement crash (#42)` or `Fixes #42` in the body). CI enforces this on PRs. +Commits should also be signed off (`git commit -s`). + ## Architecture ### Subpackages diff --git a/pyproject.toml b/pyproject.toml index e91397f1..b580b6f5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,7 +59,7 @@ io = [ "crc", "crcmod", "pyserial", - "openlifu-sdk>=2.0.12" + "openlifu-sdk>=2.0.14" ] sim = [ "k-wave-python==0.4.0", diff --git a/src/openlifu/xdc/__init__.py b/src/openlifu/xdc/__init__.py index 1e941740..b6c4e718 100644 --- a/src/openlifu/xdc/__init__.py +++ b/src/openlifu/xdc/__init__.py @@ -2,7 +2,13 @@ from .element import Element from .transducer import Transducer, TransformedTransducer -from .transducerarray import TransducerArray, get_angle_from_gap, get_roc_from_angle +from .transducerarray import ( + DeviceConfigMismatchError, + TransducerArray, + arrays_structurally_equal, + get_angle_from_gap, + get_roc_from_angle, +) __all__ = [ "element", @@ -11,6 +17,8 @@ "Transducer", "TransformedTransducer", "TransducerArray", + "DeviceConfigMismatchError", + "arrays_structurally_equal", "get_angle_from_gap", "get_roc_from_angle" ] diff --git a/src/openlifu/xdc/transducer.py b/src/openlifu/xdc/transducer.py index 529180f0..a40b83b4 100644 --- a/src/openlifu/xdc/transducer.py +++ b/src/openlifu/xdc/transducer.py @@ -95,10 +95,16 @@ class Transducer: module_invert: Annotated[List[bool], OpenLIFUFieldData("Invert polarity", "Whether to invert the polarity of the transducer output, per module")] = field(default_factory=lambda: [False]) """Whether to invert the polarity of the transducer output""" + def _normalize_standoff_transform(self) -> None: + self.standoff_transform = np.array(self.standoff_transform, dtype=float) + if self.standoff_transform.shape != (4, 4): + raise ValueError("standoff_transform must be a 4x4 matrix.") + def __post_init__(self): logging.info("Initializing transducer array") if self.name == "": self.name = self.id + self._normalize_standoff_transform() for element in self.elements: element.rescale(self.units) if self.sensitivity is None: @@ -288,6 +294,7 @@ def merge(list_of_transducers:List[Transducer], offset_pins:bool=False, offset_i merged_array.module_invert += xform_array.module_invert for k, v in merged_attrs.items(): merged_array.__setattr__(k, v) + merged_array._normalize_standoff_transform() # pylint: disable=protected-access return merged_array def numelements(self): @@ -312,7 +319,7 @@ def sort_by_pin(self): def to_dict(self): d = self.__dict__.copy() d["elements"] = [element.to_dict() for element in d["elements"]] - d["standoff_transform"] = d["standoff_transform"].tolist() + d["standoff_transform"] = np.array(d["standoff_transform"], dtype=float).tolist() return d def to_file(self, filename): @@ -393,6 +400,27 @@ def to_json(self, compact:bool=False) -> str: else: return json.dumps(self.to_dict(), indent=4) + @classmethod + def from_module_user_config(cls, user_config: dict) -> Transducer: + """Build a module from its SDK user configuration. + + The nonempty ``module`` dictionary supplies the geometry and calibration + arguments to :py:meth:`gen_matrix_array`. The top-level ``hwid`` is + preserved in the resulting transducer's ``attrs``. Mutable configuration + values are copied so the module can be edited independently. + + A template supplied to :py:meth:`TransducerArray.from_module_user_configs` + provides array placement and mesh metadata. + """ + module_cfg = user_config.get("module") + if not isinstance(module_cfg, dict) or not module_cfg: + raise ValueError("user_config has no 'module' sub-dict (expected a nonempty dictionary)") + transducer = cls.gen_matrix_array(**copy.deepcopy(module_cfg)) + hwid = user_config.get("hwid") + if hwid is not None: + transducer.attrs["hwid"] = copy.deepcopy(hwid) + return transducer + @staticmethod def gen_matrix_array(nx=2, ny=2, pitch=1, kerf=0, units="mm", **kwargs): """Generate a 2D flat matrix array diff --git a/src/openlifu/xdc/transducerarray.py b/src/openlifu/xdc/transducerarray.py index e552f20b..efa26c7e 100644 --- a/src/openlifu/xdc/transducerarray.py +++ b/src/openlifu/xdc/transducerarray.py @@ -1,6 +1,11 @@ from __future__ import annotations +import copy import json +import os +import warnings +from collections import Counter +from collections.abc import Sequence from dataclasses import dataclass, field import numpy as np @@ -9,6 +14,306 @@ from openlifu.util.units import getunitconversion from openlifu.xdc import Transducer, TransformedTransducer +# Mapping from (num_connected_modules, freq_khz) to the canonical +# template id consumed by :py:meth:`TransducerArray.get_connected`. +_DEFAULT_TEMPLATE_IDS: dict[tuple[int, int], str] = { + (1, 155): "openlifu_1x155", + (1, 400): "openlifu_1x400", + (2, 155): "openlifu_2x155", + (2, 400): "openlifu_2x400", +} + +# Locally-embedded per-module transforms and array-level standoff for +# the canonical default templates. Used as a meshless fallback when no +# database is provided to :py:meth:`TransducerArray.get_connected`. +# Translations are in millimeters. +# The 2x155 entries currently mirror the openlifu_2x180_evt1 template +# as a stand-in until a dedicated 155 kHz template ships. +_DEFAULT_TEMPLATE_DATA: dict[str, dict] = { + "openlifu_1x155": { + "name": "OpenLIFU 1x 155kHz", + "module_transforms": [np.eye(4, dtype=float)], + "standoff_transform": np.eye(4, dtype=float), + }, + "openlifu_1x400": { + "name": "OpenLIFU 1x 400kHz", + "module_transforms": [np.eye(4, dtype=float)], + "standoff_transform": np.eye(4, dtype=float), + }, + "openlifu_2x155": { + "name": "OpenLIFU 2x 155kHz", + "module_transforms": [ + np.array([ + [0.9697859993972769, 0.0, -0.2439571998794554, 25.84571998794554], + [0.0, 1.0, 0.0, 0.0], + [0.24395719987945538, 0.0, 0.9697859993972772, 3.20098197421292], + [0.0, 0.0, 0.0, 1.0], + ], dtype=float), + np.array([ + [0.9697859993972769, 0.0, 0.2439571998794554, -25.84571998794554], + [0.0, 1.0, 0.0, 0.0], + [-0.24395719987945538, 0.0, 0.9697859993972772, 3.20098197421292], + [0.0, 0.0, 0.0, 1.0], + ], dtype=float), + ], + "standoff_transform": np.array([ + [1.0, 0.0, 0.0, 0.0], + [0.0, 0.997684, -0.0680153, 0.0], + [0.0, 0.0680153, 0.997684, -8.0], + [0.0, 0.0, 0.0, 1.0], + ], dtype=float), + }, + "openlifu_2x400": { + "name": "OpenLIFU 2x 400kHz", + "module_transforms": [ + np.array([ + [0.9659258262890683, 0.0, -0.25881904510252074, 25.84571998794554], + [0.0, 1.0, 0.0, 0.0], + [0.25881904510252074, 0.0, 0.9659258262890683, 3.20098197421292], + [0.0, 0.0, 0.0, 1.0], + ], dtype=float), + np.array([ + [0.9659258262890683, 0.0, 0.25881904510252074, -25.84571998794554], + [0.0, 1.0, 0.0, 0.0], + [-0.25881904510252074, 0.0, 0.9659258262890683, 3.20098197421292], + [0.0, 0.0, 0.0, 1.0], + ], dtype=float), + ], + "standoff_transform": np.array([ + [1.0, 0.0, 0.0, 0.0], + [0.0, 1.0, 0.0, 0.0], + [0.0, 0.0, 1.0, -8.0], + [0.0, 0.0, 0.0, 1.0], + ], dtype=float), + }, +} + + +class DeviceConfigMismatchError(ValueError): + """Raised when stored device identity differs from the module configurations. + + Both :py:meth:`TransducerArray.get_connected` and + :py:meth:`TransducerArray.from_module_user_configs` validate that block + against the supplied module configurations before assembling an array. + """ + + +def _validate_device_config_against_connected( + device_cfg: dict, + user_configs: Sequence[dict], +) -> None: + """Check a stored ``device`` block matches the connected modules. + + The recorded module count must match. When any expected HWIDs are + recorded, compare their set with all reported HWIDs, ignoring order. + Partially populated HWIDs can therefore fail this check. With no expected + HWIDs, only the count is checked. + + Raises: + DeviceConfigMismatchError: if the count or the HWID sets disagree. + """ + expected_modules = list(device_cfg.get("modules") or []) + if len(expected_modules) != len(user_configs): + raise DeviceConfigMismatchError( + f"Device config '{device_cfg.get('id')}' lists {len(expected_modules)} " + f"module(s) but {len(user_configs)} module(s) are connected." + ) + + expected_hwids = { + m["hwid"] + for m in expected_modules + if isinstance(m, dict) and m.get("hwid") + } + if not expected_hwids: + # No HWIDs to compare; count match alone is acceptable. + return + connected_hwids = { + c.get("hwid") for c in user_configs if c.get("hwid") + } + missing = expected_hwids - connected_hwids + extra = connected_hwids - expected_hwids + if missing or extra: + raise DeviceConfigMismatchError( + f"Device config '{device_cfg.get('id')}' HWIDs do not match connected " + f"hardware. Missing from connected: {sorted(missing)!r}; " + f"unexpected on connected: {sorted(extra)!r}." + ) + + +def _validate_template_mesh_units( + template: TransducerArray | None, + modules: Sequence[TransformedTransducer], + attrs: dict, +) -> None: + """Reject inherited mesh coordinates that would require rescaling.""" + if template is None: + return + mesh_fields = ("registration_surface_filename", "transducer_body_filename") + for index, (source, destination) in enumerate(zip(template.modules, modules)): + for mesh_field in mesh_fields: + if getattr(source, mesh_field) and getunitconversion(source.units, destination.units) != 1: + raise ValueError( + f"Mesh {mesh_field} for module {index} cannot be inherited across units " + f"({source.units} to {destination.units}); mesh rescaling is not supported." + ) + for mesh_field in mesh_fields: + if not template.attrs.get(mesh_field) or not attrs.get(mesh_field): + continue + if not template.modules: + raise ValueError(f"Mesh {mesh_field} units require a template module.") + source_units = template.modules[0].units + destination_units = modules[0].units + if getunitconversion(source_units, destination_units) != 1: + raise ValueError( + f"Mesh {mesh_field} for the array cannot be inherited across units " + f"({source_units} to {destination_units}); mesh rescaling is not supported." + ) + + +def _find_module_matching(candidates: list[list[int]], fixed: dict[int, int]) -> list[int] | None: + """Find a one-to-one reported-to-recorded assignment, retaining fixed pairs.""" + if len(set(fixed.values())) != len(fixed): + return None + if any(recorded not in candidates[reported] for reported, recorded in fixed.items()): + return None + owners = {recorded: reported for reported, recorded in fixed.items()} + + def assign(reported, visited): + available = candidates[reported] + for recorded in available: + if recorded not in visited and recorded not in owners: + owners[recorded] = reported + return True + for recorded in available: + if recorded in visited: + continue + visited.add(recorded) + owner = owners[recorded] + if owner not in fixed and assign(owner, visited): + owners[recorded] = reported + return True + return False + + for reported in range(len(candidates)): + if reported not in fixed and not assign(reported, set()): + return None + matches = [0] * len(candidates) + for recorded, reported in owners.items(): + matches[reported] = recorded + return matches + + +def _associate_device_modules(device_modules: list[dict], user_configs: Sequence[dict]) -> tuple[list[int], list[list[int]]]: + """Return an assignment and every feasible origin of each recorded module. + + Known IDs must agree; absent IDs permit positional fallback. IDs unique in + both lists are reserved before assigning other entries. Candidate order + prefers the same position, but no recorded entry may be assigned twice. + """ + if any(not isinstance(module, dict) for module in device_modules): + raise DeviceConfigMismatchError("Cannot associate device modules: entries must be dictionaries.") + reported_ids = [cfg.get("hwid") for cfg in user_configs] + recorded_ids = [module.get("hwid") for module in device_modules] + reported_counts, recorded_counts = Counter(reported_ids), Counter(recorded_ids) + fixed = { + i: recorded_ids.index(hwid) + for i, hwid in enumerate(reported_ids) + if hwid and reported_counts[hwid] == 1 and recorded_counts[hwid] == 1 + } + candidates = [ + sorted( + [j for j, recorded in enumerate(recorded_ids) if not reported or not recorded or reported == recorded], + key=lambda j, i=i: (j != i, j), + ) + for i, reported in enumerate(reported_ids) + ] + matches = _find_module_matching(candidates, fixed) + if matches is None: + raise DeviceConfigMismatchError("Cannot associate device modules without reusing entries or mismatching hardware IDs.") + + origins: list[list[int]] = [[] for _ in device_modules] + for recorded in range(len(device_modules)): + for reported, possible in enumerate(candidates): + if recorded not in possible or (reported in fixed and fixed[reported] != recorded): + continue + if _find_module_matching(candidates, {**fixed, reported: recorded}) is not None: + origins[recorded].append(reported) + return matches, origins + + +def _recorded_module_units(recorded: int, origins: list[list[int]], modules: list[TransformedTransducer]) -> str: + """Resolve units only when all feasible origins use the same scale.""" + units = modules[origins[recorded][0]].units + if any(getunitconversion(modules[i].units, units) != 1 for i in origins[recorded]): + raise DeviceConfigMismatchError(f"Ambiguous units for recorded device module {recorded}.") + return units + + +def _build_meshless_default_template(template_id: str) -> TransducerArray: + """Build a meshless template :class:`TransducerArray` from embedded transforms.""" + spec = _DEFAULT_TEMPLATE_DATA[template_id] + modules: list[TransformedTransducer] = [] + for tform in spec["module_transforms"]: + t = Transducer(id=template_id, elements=[], units="mm") + modules.append(TransformedTransducer.from_transducer(t, transform=np.array(tform, dtype=float))) + attrs = {"standoff_transform": np.array(spec["standoff_transform"], dtype=float)} + return TransducerArray(id=template_id, name=spec["name"], modules=modules, attrs=attrs) + + +def _canonicalize_array_for_compare(arr: TransducerArray) -> dict: + """Produce a structure suitable for equality-comparing two :class:`TransducerArray`. + + Normalizations applied: + + * NumPy arrays are converted to nested lists and rounded so trivial + floating-point noise does not trigger spurious mismatches. + * Mesh filename fields (``registration_surface_filename``, + ``transducer_body_filename``) are reduced to their basename so absolute + vs database-relative paths are treated as equivalent. + * Fields that legitimately vary between a reconstructed array and a + database-stored one (e.g. ``impulse_response`` / ``impulse_dt`` from + calibration) are dropped. + + Used by :py:meth:`TransducerArray.get_connected` to warn when the array + assembled from connected hardware disagrees with the same-id array in + the supplied database. + """ + def _norm(obj): + if isinstance(obj, np.ndarray): + return _norm(obj.tolist()) + if isinstance(obj, list | tuple): + return [_norm(x) for x in obj] + if isinstance(obj, dict): + return {k: _norm(v) for k, v in obj.items()} + if isinstance(obj, float): + return round(obj, 6) + return obj + + raw = _norm(arr.to_dict()) + # Strip per-module fields that do not need to round-trip identically. + for m in raw.get("modules", []): + for k in ("registration_surface_filename", "transducer_body_filename"): + v = m.get(k) + if isinstance(v, str) and v: + m[k] = os.path.basename(v) + attrs = m.get("attrs") or {} + attrs.pop("impulse_response", None) + attrs.pop("impulse_dt", None) + # Strip array-level mesh paths likewise. + arr_attrs = raw.get("attrs") or {} + for k in ("registration_surface_filename", "transducer_body_filename"): + v = arr_attrs.get(k) + if isinstance(v, str) and v: + arr_attrs[k] = os.path.basename(v) + arr_attrs.pop("impulse_response", None) + arr_attrs.pop("impulse_dt", None) + return raw + + +def arrays_structurally_equal(a: TransducerArray, b: TransducerArray) -> bool: + """Return ``True`` if two arrays are equal after :func:`_canonicalize_array_for_compare`.""" + return _canonicalize_array_for_compare(a) == _canonicalize_array_for_compare(b) + def get_angle_from_gap(width, gap, roc): a = roc @@ -47,10 +352,10 @@ def to_transducer(self, offset_pins=True, offset_indices=True): @staticmethod def from_dict(data: dict): - d = data.copy() + d = copy.deepcopy(data) if "type" in d: d.pop("type") - d["modules"] = [TransformedTransducer.from_dict(t) for t in data["modules"]] + d["modules"] = [TransformedTransducer.from_dict(t) for t in d["modules"]] if "attrs" in d: if "standoff_transform" in d["attrs"] and d["attrs"]["standoff_transform"] is not None: d["attrs"]["standoff_transform"] = np.array(d["attrs"]["standoff_transform"]) @@ -63,7 +368,8 @@ def to_dict(self): d = {"type": "TransducerArray"} d.update(self.__dict__) d["modules"] = [t.to_dict() for t in self.modules] - for k, v in self.attrs.items(): + d = copy.deepcopy(d) + for k, v in d["attrs"].items(): if isinstance(v, np.ndarray): d["attrs"][k] = v.tolist() return d @@ -151,6 +457,369 @@ def from_file(filename: str) -> TransducerArray: data = json.load(f) return TransducerArray.from_dict(data) + @classmethod + def from_module_user_configs( + cls, + user_configs: Sequence[dict], + template: TransducerArray | None = None, + module_transforms: Sequence[np.ndarray] | None = None, + arr_id: str | None = None, + arr_name: str | None = None, + ) -> TransducerArray: + """Construct a :class:`TransducerArray` from one or more module ``user_config`` dicts. + + Each ``user_config`` describes a single physical module as reported by + the SDK (``hwid``, ``module`` sub-dict suitable for + :py:meth:`Transducer.gen_matrix_array`, optional ``device`` sub-dict on + the lead module). User configs cannot carry mesh data, so a + ``template`` :class:`TransducerArray` is normally supplied to inject + per-module mesh filenames / standoff transforms / placement transforms + and array-level metadata (id, name, attrs). + + Sources of array-level metadata, lowest priority first: + + 1. ``template``: provides ``id``, ``name``, ``attrs``, and per-module + ``transform``, ``standoff_transform``, ``registration_surface_filename``, + ``transducer_body_filename``. Modules are matched to ``user_configs`` + positionally. + 2. ``user_configs[0]["device"]`` (if present): overrides ``id``, + ``name``, merges into ``attrs``, and supplies per-module transforms + keyed by ``hwid`` when unique in both the configs and device + entries. Remaining entries are matched one-to-one, preferring + position when IDs agree or either ID is absent. + 3. ``module_transforms`` (if given): per-module 4x4 transforms that + override everything else. Length must match ``user_configs``. + 4. ``arr_id`` / ``arr_name`` (if given): explicit array id/name that + override the values picked up from the device config or template. + + The per-module ``Transducer`` is always rebuilt from the user_config's + ``module`` field (this is the on-device truth for nx/ny/pitch/kerf/ + frequency/sensitivity/etc.); only metadata that cannot live in the + user_config is taken from the template. + + A nonempty lead-module ``device`` block must record a matching module + count. If any expected HWIDs are recorded, their set must match all + reported HWIDs. A metadata-only block without module entries fails + count validation; an absent or empty block is valid. Associations that + require reusing an entry or pairing different known IDs are rejected. + + Inherited placement and standoff translations are converted from each + template module's units to the corresponding configuration's units. + Array-level standoff uses the first module's units in each array. + A translated template array standoff requires a template module to + establish its units. Device array standoff uses the recorded first + module's units, inferred from its possible matches. Ambiguous stored + translations with possible origins in different units are rejected. + Each physical module must retain its units since the device block was + recorded; explicit transforms use the destination module units. + + Mesh references cannot be inherited across different unit scales + because their coordinates are not rescaled. Renaming a template array + mesh through device attributes does not bypass this check. An explicit + ``None`` array standoff means identity and overrides any template standoff. + + Args: + user_configs: ordered list of user_config dicts. Order corresponds + to module index as reported by the device. + template: optional template array; see above for what it supplies. + module_transforms: optional list of explicit 4x4 transforms, + one per user_config, that override template/device transforms. + arr_id: optional explicit array id. Highest-priority source for the + resulting ``TransducerArray.id`` (overrides device/template/default). + arr_name: optional explicit array name. Highest-priority source for + the resulting ``TransducerArray.name``. + + Returns: + A :class:`TransducerArray` whose ``modules`` are + :class:`TransformedTransducer` instances built from the + user_configs. + """ + if not user_configs: + raise ValueError("user_configs must contain at least one user_config dict") + if module_transforms is not None and len(module_transforms) != len(user_configs): + raise ValueError( + f"module_transforms length ({len(module_transforms)}) does not match " + f"user_configs length ({len(user_configs)})" + ) + + resolved_id: str = "transducer_array" + resolved_name: str = "Transducer Array" + arr_attrs: dict = {} + if template is not None: + resolved_id = template.id + resolved_name = template.name + arr_attrs = copy.deepcopy(template.attrs) + + device_cfg = user_configs[0].get("device") or None + device_attrs = (device_cfg or {}).get("attrs") or {} + device_modules_in_order: list = [] + device_module_indices: list[int] = [] + device_module_origins: list[list[int]] = [] + if device_cfg: + _validate_device_config_against_connected(device_cfg, user_configs) + resolved_id = device_cfg.get("id", resolved_id) + resolved_name = device_cfg.get("name", resolved_name) + for k, v in device_attrs.items(): + arr_attrs[k] = copy.deepcopy(v) + device_modules_in_order = list(device_cfg.get("modules") or []) + device_module_indices, device_module_origins = _associate_device_modules(device_modules_in_order, user_configs) + + if arr_id is not None: + resolved_id = arr_id + if arr_name is not None: + resolved_name = arr_name + + template_modules: list = list(template.modules) if template is not None else [] + modules: list[TransformedTransducer] = [] + for i, cfg in enumerate(user_configs): + t = Transducer.from_module_user_config(cfg) + template_mod = template_modules[i] if i < len(template_modules) else None + + if template_mod is not None: + t.registration_surface_filename = template_mod.registration_surface_filename + t.transducer_body_filename = template_mod.transducer_body_filename + if template_mod.standoff_transform is not None: + t.standoff_transform = t.convert_transform( + np.array(template_mod.standoff_transform, dtype=float), template_mod.units, + ) + if template_mod.module_invert: + t.module_invert = list(template_mod.module_invert) + + # Resolve transform: template < device < explicit override + transform = np.eye(4) + if template_mod is not None: + transform = t.convert_transform(np.array(template_mod.transform, dtype=float), template_mod.units) + + device_mod = device_modules_in_order[device_module_indices[i]] if device_cfg else None + if device_mod is not None and device_mod.get("transform") is not None: + transform = np.array(device_mod["transform"], dtype=float) + + if module_transforms is not None: + transform = np.array(module_transforms[i], dtype=float) + + modules.append(TransformedTransducer.from_transducer(t, transform=transform)) + + _validate_template_mesh_units(template, modules, arr_attrs) + + if device_cfg: + if module_transforms is None: + for recorded, entry in enumerate(device_modules_in_order): + transform = entry.get("transform") + if transform is not None and np.any(np.asarray(transform)[:3, 3]): + _recorded_module_units(recorded, device_module_origins, modules) + if any(device_attrs.get(key) for key in ("registration_surface_filename", "transducer_body_filename")): + mesh_units = _recorded_module_units(0, device_module_origins, modules) + if getunitconversion(mesh_units, modules[0].units) != 1: + raise ValueError("Cannot inherit device mesh references across different units.") + + if "standoff_transform" in arr_attrs: + st = arr_attrs["standoff_transform"] + st = np.eye(4) if st is None else np.array(st, dtype=float) + if st.shape != (4, 4): + raise ValueError("standoff_transform must be a 4x4 matrix.") + if "standoff_transform" in device_attrs: + if np.any(st[:3, 3]): + units = _recorded_module_units(0, device_module_origins, modules) + st = modules[0].convert_transform(st, units) + elif template is not None: + if template_modules: + st = modules[0].convert_transform(st, template_modules[0].units) + elif np.any(st[:3, 3]): + raise ValueError("Cannot infer standoff units from a template without modules.") + arr_attrs["standoff_transform"] = st + + return cls(id=resolved_id, name=resolved_name, modules=modules, attrs=arr_attrs) + + @classmethod + def get_connected( + cls, + interface=None, + db=None, + arr_id: str | None = None, + arr_name: str | None = None, + module_transforms: Sequence[np.ndarray] | None = None, + use_default_template: bool = True, + ) -> TransducerArray: + """Read ``user_config`` from every connected TX module and build a :class:`TransducerArray`. + + If the lead module's ``user_config`` contains a nonempty ``device`` + block, it is validated before template selection: the number of + modules listed must match the + number of connected modules, and the recorded base58 ``hwid`` values + must match the reported HWID set when any expected HWIDs are recorded. + Without expected HWIDs, only the module count is checked. A mismatch + raises :class:`DeviceConfigMismatchError`. When the ``device`` block + carries a ``"template"`` field, that template id is preferred for the + ``db`` lookup over the default ``(n_modules, freq)`` mapping below. + + Otherwise, picks a default template based on the number of connected + modules and the per-module ``freq`` value (which must agree across + modules when more than one is connected). The mapping is: + + ====================== ===================== + ``(n_modules, freq)`` template id + ====================== ===================== + ``(1, 155)`` ``openlifu_1x155`` + ``(1, 400)`` ``openlifu_1x400`` + ``(2, 155)`` ``openlifu_2x155`` + ``(2, 400)`` ``openlifu_2x400`` + ====================== ===================== + + When ``db`` is provided, the template (with its meshes) is loaded + from the database via ``db.load_transducer(template_id, convert_array=False)``. + If no database is provided (or the lookup fails) and + ``use_default_template`` is ``True``, a meshless fallback template + is constructed from the transforms embedded in this module, without + mesh filenames. The 2x155 fallback uses stand-in geometry from the + 2x180 EVT1 template. + + Args: + interface: an :py:class:`openlifu_sdk.io.LIFUInterface`-like + object exposing ``txdevice.get_tx_module_count()`` and + ``txdevice.read_config(module=i)``. A fresh + :py:class:`LIFUInterface` is constructed when omitted + (requires ``openlifu_sdk`` to be installed). An interface + created here is closed on success or failure; an injected + interface remains open. + db: optional :py:class:`openlifu.db.Database` used to load the + template by id (so the resulting array references the + database's mesh files). + arr_id: optional explicit override for the resulting array id. + arr_name: optional explicit override for the resulting array name. + module_transforms: optional explicit per-module 4x4 transforms + (e.g. from a per-module calibration step) that override + both the template and any device-config transforms. + use_default_template: when ``True`` (default), fall back to a + meshless embedded template if no database template can be + found. ``False`` skips only the embedded fallback: a database + template can still be used, or construction can proceed + without a template. + + Returns: + A :class:`TransducerArray` representing the connected device. + """ + owns_interface = interface is None + if owns_interface: + try: + from openlifu_sdk.io import LIFUInterface + except ModuleNotFoundError as exc: + if exc.name != "openlifu_sdk": + raise + raise ImportError( + "openlifu_sdk is required to auto-create a LIFUInterface; " + "install it or pass an explicit `interface=` argument." + ) from exc + interface = LIFUInterface() + + try: + txdevice = interface.txdevice + count = int(txdevice.get_tx_module_count()) + if count <= 0: + raise RuntimeError("No TX modules are connected.") + + user_configs: list[dict] = [] + for i in range(count): + cfg = txdevice.read_config(module=i) + if cfg is None: + raise RuntimeError(f"Failed to read user_config from module {i}.") + user_configs.append(json.loads(cfg.get_json_str())) + + # All connected modules must report the same frequency for the + # template lookup to be unambiguous. + freqs = {c.get("freq") for c in user_configs} + if len(freqs) > 1: + raise ValueError( + f"Connected modules have mismatched frequencies: " + f"{sorted(f for f in freqs if f is not None)}" + ) + freq = next(iter(freqs)) if freqs else None + + # Validate recorded identity before loading its template. + device_cfg = user_configs[0].get("device") or None + device_template_id: str | None = None + if device_cfg: + _validate_device_config_against_connected(device_cfg, user_configs) + tid = device_cfg.get("template") + if isinstance(tid, str) and tid: + device_template_id = tid + + # Resolve a template: prefer db lookup, fall back to embedded transforms. + template: TransducerArray | None = None + template_id: str | None = device_template_id + if template_id is None and freq is not None: + template_id = _DEFAULT_TEMPLATE_IDS.get((count, int(freq))) + if template_id is not None: + if db is not None: + try: + loaded = db.load_transducer(template_id, convert_array=False) + except Exception: # pylint: disable=broad-exception-caught + # The optional database is only a source of template geometry. + # If it cannot supply one, use the configured fallback below. + loaded = None + if isinstance(loaded, TransducerArray): + template = loaded + if template is None and use_default_template and template_id in _DEFAULT_TEMPLATE_DATA: + template = _build_meshless_default_template(template_id) + + arr = cls.from_module_user_configs( + user_configs, + template=template, + module_transforms=module_transforms, + arr_id=arr_id, + arr_name=arr_name, + ) + + # Callers use this warning to ask about database overwrites. + if db is not None: + try: + known_ids = list(db.get_transducer_ids() or []) + except Exception: # pylint: disable=broad-exception-caught + known_ids = [] + if arr.id in known_ids: + try: + db_arr = db.load_transducer(arr.id, convert_array=False) + except Exception: # pylint: disable=broad-exception-caught + db_arr = None + if isinstance(db_arr, TransducerArray) and not arrays_structurally_equal(arr, db_arr): + warnings.warn( + f"Connected transducer '{arr.id}' differs from the version " + f"stored in the database. The database version was not used.", + stacklevel=2, + ) + + return arr + finally: + if owns_interface: + interface.close() + + def to_device_config(self) -> dict: + """Serialize array-level info to a ``device`` dict for the lead module's user_config. + + Captures the array ``id``, ``name``, and ``attrs`` (mesh filenames and + array-level ``standoff_transform``) along with per-module ``hwid`` + + ``transform`` entries. Mesh files themselves are not stored; consumers + must combine this with a template :class:`TransducerArray` (which + provides the mesh files via :py:attr:`Transducer.registration_surface_filename` + / :py:attr:`Transducer.transducer_body_filename`) when reconstructing + the array via :py:meth:`from_module_user_configs`. + """ + attrs_serialized: dict = {} + for k, v in self.attrs.items(): + attrs_serialized[k] = v.tolist() if isinstance(v, np.ndarray) else copy.deepcopy(v) + modules_entries: list[dict] = [] + for m in self.modules: + entry = { + "hwid": (m.attrs or {}).get("hwid"), + "transform": np.array(m.transform).tolist(), + } + modules_entries.append(entry) + return { + "id": self.id, + "name": self.name, + "modules": modules_entries, + "attrs": attrs_serialized, + } + @property def registration_surface_filename(self): if "registration_surface_filename" in self.attrs: diff --git a/tests/test_transducer.py b/tests/test_transducer.py index 5b34382f..cbea7380 100644 --- a/tests/test_transducer.py +++ b/tests/test_transducer.py @@ -1,13 +1,16 @@ from __future__ import annotations +import copy +import json from pathlib import Path import numpy as np import pytest from helpers import dataclasses_are_equal -from openlifu.xdc import Element, Transducer, TransducerArray +from openlifu.xdc import DeviceConfigMismatchError, Element, Transducer, TransducerArray from openlifu.xdc.transducerarray import ( + _build_meshless_default_template, get_angle_from_gap, get_gap_from_angle, get_roc_from_angle, @@ -395,3 +398,567 @@ def test_element_in_transducer_sensitivity_from_json_is_list_of_tuples(): assert all(isinstance(pair, tuple) for pair in el_sensitivity) assert all(isinstance(f, float) and isinstance(v, float) for f, v in el_sensitivity) assert el_sensitivity == [(100e3, 5.0), (300e3, 9.0)] + + +def _example_module_user_config(hwid: str = "ABCD1234") -> dict: + return { + "sn": "EVT2B-400K-TEST", + "hwid": hwid, + "freq": 400, + "module": { + "id": f"txm_400_{hwid.lower()}", + "name": f"TXM 400kHz ({hwid})", + "nx": 8, + "ny": 8, + "pitch": 5, + "frequency": 400000.0, + "kerf": 0.3, + "crosstalk_frac": 0.12, + "crosstalk_dist": 0.00505, + "sensitivity": [(400e3, 2800.0), (405e3, 1950.0)], + }, + "device": {}, + } + + +def test_transducer_from_module_user_config(): + cfg = _example_module_user_config(hwid="HW1") + t = Transducer.from_module_user_config(cfg) + assert isinstance(t, Transducer) + assert t.numelements() == 64 + assert t.id == "txm_400_hw1" + assert t.frequency == 400000.0 + assert t.attrs["hwid"] == "HW1" + assert t.sensitivity == [(400e3, 2800.0), (405e3, 1950.0)] + + +def test_transducer_from_module_user_config_missing_module(): + with pytest.raises(ValueError, match="no 'module'"): + Transducer.from_module_user_config({"hwid": "X"}) + + +def test_transducer_array_from_module_user_configs_bare(): + cfgs = [_example_module_user_config("HW1"), _example_module_user_config("HW2")] + arr = TransducerArray.from_module_user_configs(cfgs) + assert isinstance(arr, TransducerArray) + assert len(arr.modules) == 2 + assert arr.id == "transducer_array" + for m in arr.modules: + np.testing.assert_allclose(m.transform, np.eye(4)) + assert {m.attrs.get("hwid") for m in arr.modules} == {"HW1", "HW2"} + + +def test_transducer_array_from_module_user_configs_with_device_field(): + cfg1 = _example_module_user_config("HW1") + cfg2 = _example_module_user_config("HW2") + cfg1["device"] = { + "id": "test_array", + "name": "Test Array", + "modules": [ + {"hwid": "HW2", "transform": np.diag([1, 1, 1, 1]).tolist()}, + {"hwid": "HW1", + "transform": [[1, 0, 0, 10.0], [0, 1, 0, 0], [0, 0, 1, 0], [0, 0, 0, 1]]}, + ], + "attrs": {"registration_surface_filename": "x.obj"}, + } + arr = TransducerArray.from_module_user_configs([cfg1, cfg2]) + assert arr.id == "test_array" + assert arr.name == "Test Array" + assert arr.attrs["registration_surface_filename"] == "x.obj" + np.testing.assert_allclose(arr.modules[0].transform[0, 3], 10.0) + np.testing.assert_allclose(arr.modules[1].transform, np.eye(4)) + + +def test_transducer_array_from_module_user_configs_with_template(): + cfgs = [_example_module_user_config("HW1"), _example_module_user_config("HW2")] + base_template = TransducerArray.get_concave_cylinder( + Transducer.gen_matrix_array(nx=8, ny=8, pitch=5, kerf=0.3, units="mm"), + rows=1, cols=2, width=40, gap=0.0, units="mm", + id="template_array", name="Template Array", + attrs={"registration_surface_filename": "tpl.obj"}, + ) + for m in base_template.modules: + m.registration_surface_filename = "module.surf.obj" + m.transducer_body_filename = "module.body.obj" + + arr = TransducerArray.from_module_user_configs(cfgs, template=base_template) + assert arr.id == "template_array" + assert arr.attrs["registration_surface_filename"] == "tpl.obj" + for m in arr.modules: + assert m.registration_surface_filename == "module.surf.obj" + assert m.transducer_body_filename == "module.body.obj" + np.testing.assert_allclose(arr.modules[0].transform, base_template.modules[0].transform) + + +def test_transducer_array_from_module_user_configs_module_transforms_override(): + cfgs = [_example_module_user_config("HW1"), _example_module_user_config("HW2")] + cfgs[0]["device"] = { + "id": "x", + "name": "x", + "modules": [ + {"hwid": "HW1", "transform": np.eye(4).tolist()}, + {"hwid": "HW2", "transform": np.eye(4).tolist()}, + ], + "attrs": {}, + } + overrides = [ + np.array([[1, 0, 0, 1.0], [0, 1, 0, 0], [0, 0, 1, 0], [0, 0, 0, 1]], dtype=float), + np.array([[1, 0, 0, 2.0], [0, 1, 0, 0], [0, 0, 1, 0], [0, 0, 0, 1]], dtype=float), + ] + arr = TransducerArray.from_module_user_configs(cfgs, module_transforms=overrides) + np.testing.assert_allclose(arr.modules[0].transform[0, 3], 1.0) + np.testing.assert_allclose(arr.modules[1].transform[0, 3], 2.0) + + +def test_transducer_array_from_module_user_configs_empty_raises(): + with pytest.raises(ValueError, match="at least one user_config"): + TransducerArray.from_module_user_configs([]) + + +def test_transducer_array_from_module_user_configs_length_mismatch_raises(): + cfgs = [_example_module_user_config("HW1")] + with pytest.raises(ValueError, match="module_transforms length"): + TransducerArray.from_module_user_configs(cfgs, module_transforms=[np.eye(4), np.eye(4)]) + + +def test_transducer_array_from_module_user_configs_explicit_arr_id_name_override(): + cfg1 = _example_module_user_config("HW1") + cfg2 = _example_module_user_config("HW2") + cfg1["device"] = { + "id": "from_device", + "name": "From Device", + "modules": [ + {"hwid": "HW1", "transform": np.eye(4).tolist()}, + {"hwid": "HW2", "transform": np.eye(4).tolist()}, + ], + "attrs": {}, + } + arr = TransducerArray.from_module_user_configs( + [cfg1, cfg2], arr_id="explicit_id", arr_name="Explicit Name", + ) + assert arr.id == "explicit_id" + assert arr.name == "Explicit Name" + + +def test_transducer_array_from_module_user_configs_arr_id_falls_through(): + cfgs = [_example_module_user_config("HW1"), _example_module_user_config("HW2")] + template = TransducerArray.get_concave_cylinder( + Transducer.gen_matrix_array(nx=8, ny=8, pitch=5, kerf=0.3, units="mm"), + rows=1, cols=2, width=40, gap=0.0, units="mm", + id="tpl_id", name="Tpl Name", + ) + arr = TransducerArray.from_module_user_configs(cfgs, template=template) + assert arr.id == "tpl_id" + assert arr.name == "Tpl Name" + + +@pytest.mark.parametrize("module", [None, {}, [], [1], "module"]) +def test_transducer_from_module_user_config_requires_nonempty_dict(module): + with pytest.raises(ValueError, match="no 'module'"): + Transducer.from_module_user_config({"module": module}) + + +def test_transducer_from_module_user_config_geometry_and_independence(): + cfg = _example_module_user_config("HW1") + cfg["module"].update({ + "nx": 3, + "ny": 2, + "pitch": 2, + "kerf": 0.5, + "units": "cm", + "attrs": {"calibration": {"values": [1, 2]}, "hwid": "OLD"}, + "module_invert": [True], + }) + original = copy.deepcopy(cfg) + + transducer = Transducer.from_module_user_config(cfg) + + assert transducer.numelements() == 6 + assert transducer.units == "cm" + assert transducer.name == cfg["module"]["name"] + assert transducer.frequency == 400e3 + assert transducer.crosstalk_frac == 0.12 + assert transducer.crosstalk_dist == 0.00505 + assert transducer.attrs["hwid"] == "HW1" + np.testing.assert_allclose(transducer.elements[0].get_position(), [-2, 1, 0]) + np.testing.assert_allclose(transducer.elements[-1].get_position(), [2, -1, 0]) + np.testing.assert_allclose(transducer.elements[0].get_size(), [1.5, 1.5]) + assert [el.pin for el in transducer.elements] == list(range(1, 7)) + assert [el.index for el in transducer.elements] == list(range(1, 7)) + assert transducer.registration_surface_filename is None + assert transducer.transducer_body_filename is None + np.testing.assert_array_equal(transducer.standoff_transform, np.eye(4)) + transducer.to_json() + transducer.attrs["calibration"]["values"].append(3) + transducer.module_invert[0] = False + assert cfg == original + cfg["module"]["sensitivity"].append((410e3, 1000.0)) + assert transducer.sensitivity == original["module"]["sensitivity"] + + +def test_transducer_from_module_user_config_without_hwid(): + cfg = _example_module_user_config() + cfg.pop("hwid") + assert "hwid" not in Transducer.from_module_user_config(cfg).attrs + + +def _translation(x): + transform = np.eye(4) + transform[0, 3] = x + return transform + + +def test_transducer_array_from_module_user_configs_precedence_and_independence(): + cfgs = [_example_module_user_config("HW1"), _example_module_user_config("HW2")] + cfgs[0]["module"]["attrs"] = {"calibration": {"values": [1, 2]}} + template = TransducerArray.get_concave_cylinder( + Transducer.gen_matrix_array(nx=1, ny=1, frequency=155e3, sensitivity=10), + cols=2, id="template", name="Template", + attrs={ + "registration_surface_filename": "template.obj", + "retained": {"values": [1]}, + "overridden": "template", + "standoff_transform": _translation(3), + }, + ) + for i, module in enumerate(template.modules): + module.registration_surface_filename = f"module{i}.surface.obj" + module.transducer_body_filename = f"module{i}.body.obj" + module.standoff_transform = _translation(i + 1) + module.module_invert = [True] + cfgs[0]["device"] = { + "id": "device", "name": "Device", + "attrs": {"overridden": {"values": [2]}, "standoff_transform": _translation(7).tolist()}, + "modules": [ + {"hwid": "HW2", "transform": _translation(20).tolist()}, + {"hwid": "HW1", "transform": _translation(10).tolist()}, + ], + } + cfgs[1]["device"] = {"id": "ignored", "name": "Ignored"} + original_cfgs = copy.deepcopy(cfgs) + original_template = copy.deepcopy(template) + overrides = [_translation(100), _translation(200)] + + array = TransducerArray.from_module_user_configs(cfgs, template=template) + explicit = TransducerArray.from_module_user_configs( + cfgs, template=template, module_transforms=overrides, + arr_id="explicit", arr_name="Explicit", + ) + + assert (array.id, array.name) == ("device", "Device") + assert (explicit.id, explicit.name) == ("explicit", "Explicit") + assert array.attrs["registration_surface_filename"] == "template.obj" + assert array.attrs["overridden"] == {"values": [2]} + np.testing.assert_array_equal(array.attrs["standoff_transform"], _translation(7)) + for i, module in enumerate(array.modules): + assert module.numelements() == 64 + assert module.frequency == 400e3 + assert module.sensitivity == [(400e3, 2800.0), (405e3, 1950.0)] + assert module.crosstalk_frac == 0.12 + assert module.crosstalk_dist == 0.00505 + assert module.attrs["hwid"] == f"HW{i + 1}" + assert module.registration_surface_filename == f"module{i}.surface.obj" + assert module.transducer_body_filename == f"module{i}.body.obj" + assert module.module_invert == [True] + np.testing.assert_array_equal(module.standoff_transform, _translation(i + 1)) + np.testing.assert_array_equal(module.transform, _translation((i + 1) * 10)) + np.testing.assert_array_equal(explicit.modules[i].transform, overrides[i]) + + array.to_json() + array.attrs["retained"]["values"].append(9) + array.attrs["overridden"]["values"].append(9) + array.attrs["standoff_transform"][0, 3] = 9 + array.modules[0].attrs["calibration"]["values"].append(9) + array.modules[0].module_invert[0] = False + array.modules[0].standoff_transform[0, 3] = 9 + array.modules[0].transform[0, 3] = 9 + explicit.modules[0].transform[0, 3] = 9 + assert cfgs == original_cfgs + assert dataclasses_are_equal(template, original_template) + np.testing.assert_array_equal(overrides[0], _translation(100)) + + +def test_transducer_array_device_transforms_use_position_without_hwids(): + cfgs = [_example_module_user_config("HW1"), _example_module_user_config("HW2")] + cfgs[0]["device"] = {"modules": [ + {"transform": _translation(1).tolist()}, + {"transform": _translation(2).tolist()}, + ]} + array = TransducerArray.from_module_user_configs(cfgs) + for i, module in enumerate(array.modules): + np.testing.assert_array_equal(module.transform, _translation(i + 1)) + + +@pytest.mark.parametrize( + ("recorded_hwids", "connected_hwids", "valid"), + [ + (["HW1", "HW2"], ["HW1", "HW2"], True), + (["HW2", "HW1"], ["HW1", "HW2"], True), + (["HW1"], ["HW1", "HW2"], False), + (["HW1", "HW2"], ["HW1", "OTHER"], False), + (["HW1", "HW2"], ["HW1", None], False), + ([None, None], ["HW1", "HW2"], True), + ([None, None], [None, None], True), + (["HW1", None], ["HW1", "HW2"], False), + (["HW1", None], ["HW1", None], True), + (["HW1", "HW1"], ["HW1", "HW1"], True), + (["HW1", "HW1"], ["HW1", "HW2"], False), + (["HW1", None], ["HW1", "HW1"], True), + ], +) +def test_transducer_array_pure_constructor_validates_device_identity(recorded_hwids, connected_hwids, valid): + cfgs = [_example_module_user_config(str(i)) for i in range(len(connected_hwids))] + for cfg, hwid in zip(cfgs, connected_hwids): + if hwid is None: + cfg.pop("hwid") + else: + cfg["hwid"] = hwid + cfgs[0]["device"] = {"modules": [ + {"hwid": hwid} if hwid is not None else {} for hwid in recorded_hwids + ]} + if valid: + array = TransducerArray.from_module_user_configs(cfgs) + assert [m.attrs.get("hwid") for m in array.modules] == connected_hwids + else: + with pytest.raises(DeviceConfigMismatchError): + TransducerArray.from_module_user_configs(cfgs) + + +@pytest.mark.parametrize("device", [None, {}]) +def test_transducer_array_pure_constructor_accepts_no_device_metadata(device): + cfg = _example_module_user_config() + cfg["device"] = device + assert len(TransducerArray.from_module_user_configs([cfg]).modules) == 1 + + +@pytest.mark.parametrize("device", [{"id": "metadata_only"}, {"modules": []}]) +def test_transducer_array_pure_constructor_rejects_device_without_modules(device): + cfg = _example_module_user_config() + cfg["device"] = device + with pytest.raises(DeviceConfigMismatchError, match="lists 0 module"): + TransducerArray.from_module_user_configs([cfg]) + + +@pytest.mark.parametrize("as_list", [True, False]) +def test_standoff_construction_merge_and_roundtrips(as_list): + expected = _translation(8) + value = expected.tolist() if as_list else expected.copy() + transducer = Transducer.gen_matrix_array(nx=1, ny=1, standoff_transform=value) + assert isinstance(transducer.standoff_transform, np.ndarray) + np.testing.assert_array_equal(transducer.standoff_transform, expected) + + merged = Transducer.merge([transducer], merged_attrs={"standoff_transform": value}) + assert isinstance(merged.standoff_transform, np.ndarray) + np.testing.assert_array_equal(merged.standoff_transform, expected) + merged.standoff_transform[0, 3] = 99 + np.testing.assert_array_equal(value, expected) + + transducer.standoff_transform = value + serialized = transducer.to_dict() + assert isinstance(serialized["standoff_transform"], list) + for restored in [Transducer.from_dict(serialized), Transducer.from_json(transducer.to_json())]: + assert isinstance(restored.standoff_transform, np.ndarray) + np.testing.assert_array_equal(restored.standoff_transform, expected) + + array = TransducerArray.from_module_user_configs([_example_module_user_config()]) + array.attrs["standoff_transform"] = value + for restored in [array, TransducerArray.from_dict(json.loads(array.to_json()))]: + flattened = restored.to_transducer() + assert isinstance(flattened.standoff_transform, np.ndarray) + np.testing.assert_array_equal(flattened.standoff_transform, expected) + assert flattened.numelements() == 64 + + +@pytest.mark.parametrize("standoff", [None, [], np.eye(3), np.ones((4, 3))]) +def test_transducer_rejects_invalid_standoff_shape(standoff): + with pytest.raises(ValueError, match="4x4"): + Transducer(standoff_transform=standoff) + with pytest.raises(ValueError, match="4x4"): + Transducer.merge([Transducer()], merged_attrs={"standoff_transform": standoff}) + + +def test_transducer_array_to_device_config_shape_and_independence(): + cfgs = [_example_module_user_config("HW2"), _example_module_user_config("HW1")] + transforms = [_translation(2), _translation(1)] + array = TransducerArray.from_module_user_configs( + cfgs, arr_id="custom_array", arr_name="Custom Array", module_transforms=transforms, + ) + array.attrs = { + "standoff_transform": _translation(8), + "weights": np.array([1.0, 2.0]), + "metadata": {"labels": ["custom"]}, + "registration_surface_filename": "surface.obj", + } + original = copy.deepcopy(array) + + device = array.to_device_config() + + assert set(device) == {"id", "name", "modules", "attrs"} + assert (device["id"], device["name"]) == ("custom_array", "Custom Array") + assert device["modules"] == [ + {"hwid": "HW2", "transform": transforms[0].tolist()}, + {"hwid": "HW1", "transform": transforms[1].tolist()}, + ] + assert device["attrs"]["standoff_transform"] == _translation(8).tolist() + assert device["attrs"]["weights"] == [1.0, 2.0] + assert device["attrs"]["registration_surface_filename"] == "surface.obj" + assert json.loads(json.dumps(device)) == device + assert dataclasses_are_equal(array, original) + + reconstructed_cfgs = copy.deepcopy(cfgs) + reconstructed_cfgs[0]["device"] = device + reconstructed = TransducerArray.from_module_user_configs(reconstructed_cfgs) + assert reconstructed.to_device_config() == device + original_reconstructed = copy.deepcopy(reconstructed) + device["modules"][0]["transform"][0][3] = 99 + device["attrs"]["standoff_transform"][0][3] = 99 + device["attrs"]["metadata"]["labels"].append("changed") + assert dataclasses_are_equal(array, original) + assert dataclasses_are_equal(reconstructed, original_reconstructed) + + +def test_transducer_array_dict_serialization_does_not_alias_inputs(): + cfg = _example_module_user_config() + cfg["module"]["attrs"] = {"calibration": {"values": [1]}} + array = TransducerArray.from_module_user_configs([cfg]) + array.attrs = {"standoff_transform": _translation(8), "metadata": {"labels": ["custom"]}} + original = copy.deepcopy(array) + serialized = array.to_dict() + assert dataclasses_are_equal(array, original) + assert isinstance(serialized["attrs"]["standoff_transform"], list) + serialized["attrs"]["impulse_response"] = [1, 2] + serialized["attrs"]["impulse_dt"] = 1e-6 + original_serialized = copy.deepcopy(serialized) + + restored = TransducerArray.from_dict(serialized) + + assert serialized == original_serialized + assert "impulse_response" not in restored.attrs + assert "impulse_dt" not in restored.attrs + assert isinstance(restored.attrs["standoff_transform"], np.ndarray) + restored.attrs["metadata"]["labels"].append("changed") + restored.modules[0].attrs["calibration"]["values"].append(2) + restored.modules[0].module_invert[0] = True + assert serialized == original_serialized + serialized["attrs"]["standoff_transform"][0][3] = 99 + serialized["attrs"]["metadata"]["labels"].append("changed") + serialized["modules"][0]["attrs"]["calibration"]["values"].append(2) + serialized["modules"][0]["module_invert"][0] = True + assert dataclasses_are_equal(array, original) + + +def _physical_module_configs(units): + mm_per_unit = {"mm": 1, "cm": 10, "m": 1000} + configs = [_example_module_user_config(f"HW{i}") for i in range(len(units))] + for cfg, unit in zip(configs, units): + cfg["module"].update( + nx=2, ny=1, pitch=4 / mm_per_unit[unit], kerf=0.2 / mm_per_unit[unit], units=unit, + ) + return configs + + +@pytest.mark.parametrize("template_units", [("mm", "mm"), ("cm", "cm"), ("mm", "cm")]) +@pytest.mark.parametrize("module_units", [("mm", "mm"), ("cm", "cm"), ("cm", "mm")]) +@pytest.mark.parametrize("as_lists", [False, True]) +def test_template_geometry_preserves_physical_units(template_units, module_units, as_lists): + template = TransducerArray.from_module_user_configs(_physical_module_configs(template_units)) + standoff = np.array([[1, 0, 0, 2], [0, 0, -1, 4], [0, 1, 0, 8], [0, 0, 0, 1]], dtype=float) + array_standoff = standoff.copy() + array_standoff[2, 3] = 18 + array_standoff[:3, 3] /= 1 if template_units[0] == "mm" else 10 + template.attrs["standoff_transform"] = array_standoff + for i, module in enumerate(template.modules): + mm_per_template_unit = 1 if module.units == "mm" else 10 + module.transform = np.array( + [[0, -1, 0, 10 * (-1) ** i], [1, 0, 0, 4], [0, 0, 1, 6], [0, 0, 0, 1]], dtype=float, + ) + module.transform[:3, 3] /= mm_per_template_unit + module.standoff_transform = standoff.copy() + module.standoff_transform[:3, 3] /= mm_per_template_unit + reference = copy.deepcopy(template) + if as_lists: + template.attrs["standoff_transform"] = array_standoff.tolist() + for module in template.modules: + module.transform = module.transform.tolist() + module.standoff_transform = module.standoff_transform.tolist() + original_template = copy.deepcopy(template) + configs = _physical_module_configs(module_units) + original_configs = copy.deepcopy(configs) + + array = TransducerArray.from_module_user_configs(configs, template=template) + + for module, expected in zip(array.modules, reference.modules): + np.testing.assert_allclose(module.bake().get_positions(units="mm"), expected.bake().get_positions(units="mm")) + np.testing.assert_allclose(module.get_standoff_transform_in_units("mm"), expected.get_standoff_transform_in_units("mm")) + np.testing.assert_array_equal(module.transform[:3, :3], expected.transform[:3, :3]) + assert module.frequency == expected.frequency + assert module.sensitivity == expected.sensitivity + assert [module.units for module in array.modules] == list(module_units) + flattened = array.to_transducer() + expected_flattened = reference.to_transducer() + np.testing.assert_allclose(flattened.get_positions(units="mm"), expected_flattened.get_positions(units="mm")) + np.testing.assert_allclose(flattened.get_standoff_transform_in_units("mm"), expected_flattened.get_standoff_transform_in_units("mm")) + replay_configs = copy.deepcopy(configs) + replay_configs[0]["device"] = array.to_device_config() + replay = TransducerArray.from_module_user_configs(replay_configs, template=template) + np.testing.assert_allclose(replay.to_transducer().get_positions(units="mm"), flattened.get_positions(units="mm")) + np.testing.assert_allclose(replay.attrs["standoff_transform"], array.attrs["standoff_transform"]) + assert configs == original_configs + assert dataclasses_are_equal(template, original_template) + + +@pytest.mark.parametrize("standoff_override", [None, _translation(3).tolist()]) +def test_template_unit_conversion_preserves_device_and_explicit_overrides(standoff_override): + template = TransducerArray.from_module_user_configs(_physical_module_configs(["mm"])) + template.modules[0].transform = _translation(10) + template.modules[0].standoff_transform = _translation(8) + template.attrs["standoff_transform"] = _translation(8) + configs = _physical_module_configs(["cm"]) + configs[0]["device"] = { + "modules": [{"hwid": "HW0", "transform": _translation(2).tolist()}], + "attrs": {"standoff_transform": standoff_override}, + } + original_configs = copy.deepcopy(configs) + array = TransducerArray.from_module_user_configs(configs, template=template) + explicit_transform = _translation(4) + explicit = TransducerArray.from_module_user_configs(configs, template=template, module_transforms=[explicit_transform]) + np.testing.assert_array_equal(array.modules[0].transform, _translation(2)) + np.testing.assert_array_equal(explicit.modules[0].transform, explicit_transform) + for result in (array, explicit): + np.testing.assert_array_equal(result.modules[0].standoff_transform, _translation(0.8)) + expected_standoff = np.eye(4) if standoff_override is None else np.array(standoff_override) + for restored in (result, TransducerArray.from_dict(result.to_dict()), TransducerArray.from_dict(json.loads(result.to_json()))): + np.testing.assert_array_equal(restored.attrs["standoff_transform"], expected_standoff) + flattened = restored.to_transducer() + np.testing.assert_array_equal(flattened.standoff_transform, expected_standoff) + np.testing.assert_allclose(flattened.get_positions(units="mm"), result.modules[0].bake().get_positions(units="mm")) + assert configs == original_configs + + +def test_translated_array_standoff_requires_template_units(): + template = TransducerArray(attrs={"standoff_transform": _translation(8)}) + configs = _physical_module_configs(["cm"]) + with pytest.raises(ValueError, match="standoff.*template.*module"): + TransducerArray.from_module_user_configs(configs, template=template) + template.attrs["standoff_transform"] = np.eye(4) + array = TransducerArray.from_module_user_configs(configs, template=template) + np.testing.assert_array_equal(array.attrs["standoff_transform"], np.eye(4)) + + +@pytest.mark.parametrize("frequency", [155, 400]) +def test_embedded_template_translations_are_in_millimeters(frequency): + template = _build_meshless_default_template(f"openlifu_2x{frequency}") + assert [module.units for module in template.modules] == ["mm", "mm"] + array = TransducerArray.from_module_user_configs(_physical_module_configs(["cm", "cm"]), template=template) + for module, template_module in zip(array.modules, template.modules): + np.testing.assert_allclose(module.transform[:3, 3] * 10, template_module.transform[:3, 3]) + np.testing.assert_array_equal(module.transform[:3, :3], template_module.transform[:3, :3]) + assert array.modules[0].transform[0, 3] == pytest.approx(2.584571998794554) + assert array.to_transducer().get_standoff_transform_in_units("mm")[2, 3] == pytest.approx(-8) + + +def test_template_geometry_rejects_incompatible_units(): + template = TransducerArray.from_module_user_configs(_physical_module_configs(["mm"])) + template.modules[0].units = "s" + with pytest.raises(ValueError, match="Unit type mismatch"): + TransducerArray.from_module_user_configs(_physical_module_configs(["cm"]), template=template) diff --git a/tests/test_transducer_array_device_config.py b/tests/test_transducer_array_device_config.py new file mode 100644 index 00000000..7c708629 --- /dev/null +++ b/tests/test_transducer_array_device_config.py @@ -0,0 +1,715 @@ +from __future__ import annotations + +import copy +import json +import subprocess +import sys +import textwrap +from types import ModuleType, SimpleNamespace +from unittest.mock import Mock + +import numpy as np +import pytest +from helpers import dataclasses_are_equal + +from openlifu.db import Database +from openlifu.xdc import ( + DeviceConfigMismatchError, + Transducer, + TransducerArray, + arrays_structurally_equal, +) + + +def _module_user_config(hwid: str | None, freq: int = 400) -> dict: + return { + "hwid": hwid, + "freq": freq, + "module": { + "nx": 2, + "ny": 2, + "pitch": 1.0, + "kerf": 0.0, + "units": "mm", + "frequency": freq * 1000.0, + }, + } + + +class _FakeTxDevice: + def __init__(self, configs): + self.configs = configs + self.count_calls = 0 + self.read_calls = [] + + def get_tx_module_count(self): + self.count_calls += 1 + return len(self.configs) + + def read_config(self, *, module): + self.read_calls.append(module) + config = self.configs[module] + if isinstance(config, Exception): + raise config + if config is None: + return None + payload = config if isinstance(config, str) else json.dumps(config) + return SimpleNamespace(get_json_str=lambda: payload) + + +class _FakeInterface: + def __init__(self, configs): + self.txdevice = _FakeTxDevice(configs) + self.close = Mock() + + +class _FakeDB: + def __init__(self, templates=None, stored=None): + self.templates = templates or {} + self.stored = stored or {} + self.load_calls = [] + + def get_transducer_ids(self): + return list(self.stored) + + def load_transducer(self, transducer_id, convert_array=True): + self.load_calls.append((transducer_id, convert_array)) + value = self.templates.get(transducer_id, self.stored.get(transducer_id)) + if isinstance(value, Exception): + raise value + return value + + +@pytest.mark.parametrize( + ("expected", "connected", "valid"), + [ + (["AAA", "BBB"], ["AAA", "BBB"], True), + (["BBB", "AAA"], ["AAA", "BBB"], True), + (["AAA", "BBB"], ["AAA"], False), + (["AAA", "BBB"], ["AAA", "ZZZ"], False), + (["AAA", "BBB"], ["AAA", None], False), + ([None, None], ["AAA", "BBB"], True), + (["AAA", None], ["AAA", "BBB"], False), + (["AAA", None], ["AAA", None], True), + (["AAA", "AAA"], ["AAA", "AAA"], True), + (["AAA", "AAA"], ["AAA", "BBB"], False), + (["AAA", None], ["AAA", "AAA"], True), + ([], ["AAA"], False), + ], +) +def test_get_connected_validates_device_identity_before_database_lookup(expected, connected, valid): + configs = [_module_user_config(hwid) for hwid in connected] + configs[0]["device"] = { + "id": "device", + "modules": [{"hwid": hwid} if hwid else {} for hwid in expected], + } + interface = _FakeInterface(configs) + db = _FakeDB() + if valid: + array = TransducerArray.get_connected(interface=interface, db=db) + assert [module.attrs.get("hwid") for module in array.modules] == connected + else: + with pytest.raises(DeviceConfigMismatchError, match="lists .* module|HWIDs do not match"): + TransducerArray.get_connected(interface=interface, db=db) + assert db.load_calls == [] + assert interface.txdevice.read_calls == list(range(len(configs))) + interface.close.assert_not_called() + + +@pytest.mark.parametrize("device", [None, {}]) +def test_get_connected_accepts_absent_or_empty_device(device): + config = _module_user_config("AAA") + if device is not None: + config["device"] = device + array = TransducerArray.get_connected(interface=_FakeInterface([config])) + assert array.id == "openlifu_1x400" + + +def test_get_connected_rejects_metadata_only_device(): + config = _module_user_config("AAA") + config["device"] = {"id": "metadata-only", "name": "Metadata only"} + with pytest.raises(DeviceConfigMismatchError, match="lists 0 module"): + TransducerArray.get_connected(interface=_FakeInterface([config])) + + +@pytest.mark.parametrize( + ("count", "freq", "template_id"), + [(1, 155, "openlifu_1x155"), (1, 400, "openlifu_1x400"), + (2, 155, "openlifu_2x155"), (2, 400, "openlifu_2x400")], +) +def test_get_connected_infers_embedded_template(count, freq, template_id): + interface = _FakeInterface([_module_user_config(str(i), freq) for i in range(count)]) + array = TransducerArray.get_connected(interface=interface) + assert array.id == template_id + assert len(array.modules) == count + assert array.registration_surface_filename is None + assert array.transducer_body_filename is None + assert isinstance(array.attrs["standoff_transform"], np.ndarray) + assert all(module.frequency == freq * 1000 for module in array.modules) + if count == 1: + np.testing.assert_array_equal(array.modules[0].transform, np.eye(4)) + else: + np.testing.assert_allclose(array.modules[0].transform[0, 3], 25.84571998794554) + np.testing.assert_allclose(array.modules[1].transform[0, 3], -25.84571998794554) + assert interface.txdevice.count_calls == 1 + assert interface.txdevice.read_calls == list(range(count)) + interface.close.assert_not_called() + + +@pytest.mark.parametrize("use_default_template", [True, False]) +def test_get_connected_prefers_recorded_database_template(use_default_template): + config = _module_user_config("AAA", 400) + config["device"] = { + "id": "device", + "name": "Device", + "template": "openlifu_1x155", + "modules": [{"hwid": "AAA"}], + } + template = TransducerArray.from_module_user_configs([_module_user_config("template", 155)]) + template.modules[0].transform[0, 3] = 17.0 + template.modules[0].registration_surface_filename = "module-surface.obj" + template.transducer_body_filename = "array-body.obj" + original = copy.deepcopy(template) + db = _FakeDB(templates={"openlifu_1x155": template}) + array = TransducerArray.get_connected( + interface=_FakeInterface([config]), db=db, use_default_template=use_default_template, + ) + assert (array.id, array.name) == ("device", "Device") + assert array.modules[0].frequency == 400000.0 + assert array.modules[0].registration_surface_filename == "module-surface.obj" + assert array.transducer_body_filename == "array-body.obj" + assert array.modules[0].transform[0, 3] == 17.0 + assert db.load_calls == [("openlifu_1x155", False)] + assert dataclasses_are_equal(template, original) + + +@pytest.mark.parametrize("use_default_template", [True, False]) +@pytest.mark.parametrize("loaded", [None, FileNotFoundError("missing template"), Transducer()]) +def test_get_connected_database_template_unavailable(loaded, use_default_template): + db = _FakeDB(templates={"openlifu_1x400": loaded}) + array = TransducerArray.get_connected( + interface=_FakeInterface([_module_user_config("AAA")]), db=db, + use_default_template=use_default_template, + ) + assert array.id == ("openlifu_1x400" if use_default_template else "transducer_array") + assert ("standoff_transform" in array.attrs) is use_default_template + assert db.load_calls == [("openlifu_1x400", False)] + + +@pytest.mark.parametrize("use_default_template", [True, False]) +def test_get_connected_unknown_recorded_template_does_not_infer_replacement(use_default_template): + config = _module_user_config("AAA") + config["device"] = { + "id": "device", "template": "custom-template", "modules": [{"hwid": "AAA"}], + } + db = _FakeDB() + array = TransducerArray.get_connected( + interface=_FakeInterface([config]), db=db, use_default_template=use_default_template, + ) + assert db.load_calls == [("custom-template", False)] + assert array.id == "device" + assert "standoff_transform" not in array.attrs + np.testing.assert_array_equal(array.modules[0].transform, np.eye(4)) + + +@pytest.mark.parametrize("freq", [None, 250]) +def test_get_connected_unknown_frequency_constructs_without_template(freq): + config = _module_user_config("AAA") + config["freq"] = freq + db = _FakeDB() + array = TransducerArray.get_connected(interface=_FakeInterface([config]), db=db) + assert array.id == "transducer_array" + assert db.load_calls == [] + np.testing.assert_array_equal(array.modules[0].transform, np.eye(4)) + + +def test_get_connected_forwards_explicit_overrides(): + configs = [_module_user_config("AAA"), _module_user_config("BBB")] + transforms = [np.eye(4), np.eye(4)] + transforms[0][0, 3] = 1.0 + transforms[1][0, 3] = 2.0 + array = TransducerArray.get_connected( + interface=_FakeInterface(configs), arr_id="custom", arr_name="Custom array", + module_transforms=transforms, + ) + assert (array.id, array.name) == ("custom", "Custom array") + for module, transform in zip(array.modules, transforms): + np.testing.assert_array_equal(module.transform, transform) + + +@pytest.mark.parametrize("recorded_template", [None, "openlifu_2x400"]) +def test_get_connected_checks_frequencies_before_template_lookup(recorded_template): + configs = [_module_user_config("AAA", 400), _module_user_config("BBB", 155)] + if recorded_template: + configs[0]["device"] = { + "template": recorded_template, "modules": [{"hwid": "AAA"}, {"hwid": "BBB"}], + } + db = _FakeDB() + with pytest.raises(ValueError, match="mismatched frequencies"): + TransducerArray.get_connected(interface=_FakeInterface(configs), db=db) + assert db.load_calls == [] + + +@pytest.mark.parametrize("owned", [True, False]) +@pytest.mark.parametrize("outcome", ["success", "empty", "count-error", "none", "read-error", "json", "frequency", "identity", "module"]) +def test_get_connected_closes_only_owned_interface(monkeypatch, owned, outcome): + configs = [_module_user_config("AAA")] + error = RuntimeError + message = "" + if outcome == "empty": + configs = [] + message = "No TX modules" + elif outcome == "none": + configs = [None] + message = "module 0" + elif outcome == "read-error": + configs = [OSError("USB read failed")] + error, message = OSError, "USB read failed" + elif outcome == "json": + configs = ["{malformed"] + error, message = json.JSONDecodeError, "Expecting" + elif outcome == "frequency": + configs.append(_module_user_config("BBB", 155)) + error, message = ValueError, "mismatched frequencies" + elif outcome == "identity": + configs[0]["device"] = {"modules": [{"hwid": "ZZZ"}]} + error, message = DeviceConfigMismatchError, "HWIDs do not match" + elif outcome == "module": + configs[0]["module"] = {} + error, message = ValueError, "module" + interface = _FakeInterface(configs) + if outcome == "count-error": + interface.txdevice.get_tx_module_count = Mock(side_effect=OSError("Module count failed")) + error, message = OSError, "Module count failed" + factory = Mock(return_value=interface) + sdk_io = ModuleType("openlifu_sdk.io") + sdk_io.LIFUInterface = factory + monkeypatch.setitem(sys.modules, "openlifu_sdk.io", sdk_io) + kwargs = {} if owned else {"interface": interface} + if outcome == "success": + array = TransducerArray.get_connected(**kwargs) + assert array.modules[0].attrs["hwid"] == "AAA" + else: + with pytest.raises(error, match=message): + TransducerArray.get_connected(**kwargs) + if owned: + factory.assert_called_once_with() + interface.close.assert_called_once_with() + else: + factory.assert_not_called() + interface.close.assert_not_called() + + +@pytest.mark.parametrize("import_failure", ["absent", "dependency", "internal"]) +def test_optional_sdk_import_in_fresh_process(import_failure): + script = textwrap.dedent(""" + import importlib.abc + import json + import sys + from types import SimpleNamespace + + mode = sys.argv[1] + attempts = [] + + class BlockSDK(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if fullname == "openlifu_sdk" or fullname.startswith("openlifu_sdk."): + attempts.append(fullname) + if mode == "absent": + raise ModuleNotFoundError("SDK is absent", name="openlifu_sdk") + if mode == "dependency": + raise ModuleNotFoundError("SDK dependency is absent", name="sdk_dependency") + raise ImportError("SDK internal failure") + + sys.meta_path.insert(0, BlockSDK()) + from openlifu.xdc import Transducer, TransducerArray + + config = json.loads(sys.argv[2]) + module = Transducer.from_module_user_config(config) + pure = TransducerArray.from_module_user_configs([config]) + pure.to_device_config() + interface = SimpleNamespace(txdevice=SimpleNamespace( + get_tx_module_count=lambda: 1, + read_config=lambda module: SimpleNamespace(get_json_str=lambda: json.dumps(config)), + )) + connected = TransducerArray.get_connected(interface=interface) + assert module.numelements() == 4 + assert connected.modules[0].attrs["hwid"] == "AAA" + assert attempts == [], attempts + try: + TransducerArray.get_connected() + except ImportError as exc: + if mode == "absent": + assert "openlifu_sdk" in str(exc), str(exc) + assert "interface" in str(exc), str(exc) + elif mode == "dependency": + assert isinstance(exc, ModuleNotFoundError), repr(exc) + assert exc.name == "sdk_dependency", repr(exc) + assert str(exc) == "SDK dependency is absent", str(exc) + else: + assert str(exc) == "SDK internal failure", str(exc) + else: + raise AssertionError("SDK import failure was swallowed") + assert attempts + """) + result = subprocess.run( + [sys.executable, "-c", script, import_failure, json.dumps(_module_user_config("AAA"))], + capture_output=True, text=True, check=False, timeout=60, + ) + assert result.returncode == 0, result.stdout + result.stderr + + +@pytest.mark.parametrize("different", [True, False]) +def test_get_connected_database_comparison_warning(different): + configs = [_module_user_config("AAA")] + configs[0]["device"] = { + "id": "device", "name": "Device", "template": "openlifu_1x400", + "modules": [{"hwid": "AAA"}], + } + stored = TransducerArray.get_connected(interface=_FakeInterface(configs)) + if different: + stored.name = "Other array" + original = copy.deepcopy(stored) + db = _FakeDB(stored={stored.id: stored}) + if different: + with pytest.warns(UserWarning, match="differs from the version"): + array = TransducerArray.get_connected(interface=_FakeInterface(configs), db=db) + else: + array = TransducerArray.get_connected(interface=_FakeInterface(configs), db=db) + assert arrays_structurally_equal(array, stored) is not different + assert dataclasses_are_equal(stored, original) + + +def test_arrays_structurally_equal_normalizes_without_mutating_inputs(): + array = TransducerArray.get_connected(interface=_FakeInterface([_module_user_config("AAA")])) + array.registration_surface_filename = "surface.obj" + array.transducer_body_filename = "body.obj" + array.modules[0].registration_surface_filename = "module-surface.obj" + array.modules[0].transducer_body_filename = "module-body.obj" + array.attrs["samples"] = np.array([1.0, 2.0]) + other = copy.deepcopy(array) + other.registration_surface_filename = "/database/device/surface.obj" + other.transducer_body_filename = "/database/device/body.obj" + other.modules[0].registration_surface_filename = "/database/module/module-surface.obj" + other.modules[0].transducer_body_filename = "/database/module/module-body.obj" + other.attrs["samples"] = [1.0000001, 2.0000001] + other.attrs["standoff_transform"] = other.attrs["standoff_transform"].tolist() + other.modules[0].transform[0, 3] += 0.0000001 + for attrs in (other.attrs, other.modules[0].attrs): + attrs["impulse_response"] = [123.0] + attrs["impulse_dt"] = 0.123 + original_array, original_other = copy.deepcopy((array, other)) + assert arrays_structurally_equal(array, other) + assert arrays_structurally_equal(other, array) + assert dataclasses_are_equal(array, original_array) + assert dataclasses_are_equal(other, original_other) + + +@pytest.mark.parametrize("difference", ["name", "hwid", "geometry", "transform", "mesh", "attrs"]) +def test_arrays_structurally_equal_detects_meaningful_differences(difference): + array = TransducerArray.from_module_user_configs([_module_user_config("AAA")]) + other = copy.deepcopy(array) + if difference == "name": + other.name = "Different name" + elif difference == "hwid": + other.modules[0].attrs["hwid"] = "BBB" + elif difference == "geometry": + other.modules[0].elements[0].position[0] += 0.001 + elif difference == "transform": + other.modules[0].transform[0, 3] += 0.000001 + elif difference == "mesh": + other.registration_surface_filename = "different.obj" + else: + other.attrs["custom"] = True + assert not arrays_structurally_equal(array, other) + + +def test_to_device_config_has_independent_json_compatible_data(): + configs = [_module_user_config("BBB"), _module_user_config("AAA")] + array = TransducerArray.get_connected(interface=_FakeInterface(configs), arr_id="device", arr_name="Device") + array.registration_surface_filename = "surface.obj" + array.transducer_body_filename = "body.obj" + array.attrs["metadata"] = {"labels": ["original"]} + array.attrs["samples"] = np.array([1.0, 2.0]) + original = copy.deepcopy(array) + serialized = array.to_device_config() + assert set(serialized) == {"id", "name", "modules", "attrs"} + assert (serialized["id"], serialized["name"]) == ("device", "Device") + assert [module["hwid"] for module in serialized["modules"]] == ["BBB", "AAA"] + assert all(set(module) == {"hwid", "transform"} for module in serialized["modules"]) + assert serialized["attrs"]["samples"] == [1.0, 2.0] + assert json.loads(json.dumps(serialized)) == serialized + for entry, module in zip(serialized["modules"], array.modules): + np.testing.assert_array_equal(entry["transform"], module.transform) + np.testing.assert_array_equal(serialized["attrs"]["standoff_transform"], array.attrs["standoff_transform"]) + configs[0]["device"] = copy.deepcopy(serialized) + rebuilt = TransducerArray.from_module_user_configs(configs) + assert arrays_structurally_equal(array, rebuilt) + serialized["attrs"]["metadata"]["labels"].append("changed") + serialized["attrs"]["standoff_transform"][0][3] = 999.0 + serialized["modules"][0]["transform"][0][3] = 999.0 + assert dataclasses_are_equal(array, original) + + +@pytest.mark.parametrize("connected", [False, True]) +def test_device_config_round_trip_preserves_placements_with_duplicate_hwids(connected): + configs = [_module_user_config("ABCDEFGH"), _module_user_config("ABCDEFGH")] + transforms = [np.eye(4), np.eye(4)] + transforms[0][0, 3] = 25.0 + transforms[1][0, 3] = -25.0 + original = TransducerArray.from_module_user_configs(configs, module_transforms=transforms) + configs[0]["device"] = json.loads(json.dumps(original.to_device_config())) + if connected: + rebuilt = TransducerArray.get_connected( + interface=_FakeInterface(configs), use_default_template=False, + ) + else: + rebuilt = TransducerArray.from_module_user_configs(configs) + assert arrays_structurally_equal(original, rebuilt) + np.testing.assert_allclose(rebuilt.to_transducer().get_positions(), original.to_transducer().get_positions()) + + +@pytest.mark.parametrize("connected", [False, True]) +@pytest.mark.parametrize( + ("recorded_hwids", "reported_hwids", "expected_indices"), + [ + (["BBB", "AAA"], ["AAA", "BBB"], [1, 0]), + (["BBB", "DUP", "AAA", "DUP"], ["AAA", "DUP", "BBB", "DUP"], [2, 1, 0, 3]), + (["AAA", None], [None, "AAA"], [1, 0]), + (["AAA", None], ["AAA", "AAA"], [0, 1]), + (["AAA", "AAA"], ["AAA", None], [0, 1]), + ], + ids=["unique-reordered", "mixed-unique-and-duplicate", "missing-id-reordered", "reported-duplicate", "recorded-duplicate"], +) +def test_device_transforms_match_only_unambiguous_hwids(connected, recorded_hwids, reported_hwids, expected_indices): + configs = [_module_user_config(hwid) for hwid in reported_hwids] + transforms = [np.eye(4) for _ in recorded_hwids] + for i, transform in enumerate(transforms): + transform[0, 3] = 10.0 * (i + 1) + configs[0]["device"] = {"modules": [ + {"hwid": hwid, "transform": transform.tolist()} + for hwid, transform in zip(recorded_hwids, transforms) + ]} + original_configs = copy.deepcopy(configs) + if connected: + array = TransducerArray.get_connected(interface=_FakeInterface(configs), use_default_template=False) + else: + array = TransducerArray.from_module_user_configs(configs) + for module, expected_index in zip(array.modules, expected_indices): + np.testing.assert_array_equal(module.transform, transforms[expected_index]) + assert configs == original_configs + + +@pytest.mark.parametrize("connected", [False, True]) +def test_device_transform_matching_reserves_entries_before_duplicate_fallback(connected): + configs = [_module_user_config(hwid) for hwid in ["DUP", "DUP", "AAA"]] + transforms = [np.eye(4) for _ in configs] + for i, transform in enumerate(transforms): + transform[0, 3] = 10.0 * (i + 1) + configs[0]["device"] = {"modules": [ + {"hwid": hwid, "transform": transform.tolist()} + for hwid, transform in zip(["AAA", "DUP", "DUP"], transforms) + ]} + if connected: + array = TransducerArray.get_connected(interface=_FakeInterface(configs), use_default_template=False) + else: + array = TransducerArray.from_module_user_configs(configs) + + assert sorted(module.transform[0, 3] for module in array.modules[:2]) == [20, 30] + np.testing.assert_array_equal(array.modules[2].transform, transforms[0]) + + +@pytest.mark.parametrize("connected", [False, True]) +def test_device_transform_matching_rejects_incompatible_id_multiplicities(connected): + configs = [_module_user_config(hwid) for hwid in ["AAA", "BBB", "BBB"]] + configs[0]["device"] = {"modules": [ + {"hwid": hwid, "transform": np.eye(4).tolist()} + for hwid in ["AAA", "AAA", "BBB"] + ]} + if connected: + with pytest.raises(DeviceConfigMismatchError, match="(?i)associate|match"): + TransducerArray.get_connected(interface=_FakeInterface(configs), use_default_template=False) + else: + with pytest.raises(DeviceConfigMismatchError, match="(?i)associate|match"): + TransducerArray.from_module_user_configs(configs) + + +def test_translated_device_placements_reject_ambiguous_mixed_units(): + configs = [_module_user_config("DUP"), _module_user_config("DUP")] + configs[1]["module"].update(units="cm", pitch=0.1) + transforms = [np.eye(4), np.eye(4)] + transforms[0][0, 3] = 10 + transforms[1][0, 3] = -2 + configs[0]["device"] = {"modules": [ + {"hwid": "DUP", "transform": transform.tolist()} for transform in transforms + ]} + + with pytest.raises(DeviceConfigMismatchError, match="(?i)ambiguous.*units"): + TransducerArray.from_module_user_configs(configs) + + +def test_explicit_placements_override_ambiguous_mixed_unit_device_placements(): + configs = [_module_user_config("DUP"), _module_user_config("DUP")] + configs[1]["module"].update(units="cm", pitch=0.1) + stored_transforms = [np.eye(4), np.eye(4)] + stored_transforms[0][0, 3] = 10 + stored_transforms[1][0, 3] = -2 + configs[0]["device"] = {"modules": [ + {"hwid": "DUP", "transform": transform.tolist()} for transform in stored_transforms + ]} + explicit_transforms = [np.eye(4), np.eye(4)] + explicit_transforms[0][0, 3] = 25 + explicit_transforms[1][0, 3] = -2.5 + original_configs = copy.deepcopy(configs) + + array = TransducerArray.from_module_user_configs(configs, module_transforms=explicit_transforms) + + for module, transform in zip(array.modules, explicit_transforms): + np.testing.assert_array_equal(module.transform, transform) + positions = array.to_transducer().get_positions(units="mm") + assert positions[:4, 0].mean() == pytest.approx(-25) + assert positions[4:, 0].mean() == pytest.approx(25) + assert configs == original_configs + + +@pytest.mark.parametrize("entry", [None, "AAA", []]) +def test_device_transform_matching_rejects_non_dictionary_entry(entry): + configs = [_module_user_config("AAA")] + configs[0]["device"] = {"modules": [entry]} + + with pytest.raises(DeviceConfigMismatchError, match="(?i)associate.*dictionaries"): + TransducerArray.from_module_user_configs(configs) + + +@pytest.mark.parametrize("connected", [False, True]) +@pytest.mark.parametrize("reordered", [False, True]) +def test_device_standoff_preserves_units_when_modules_are_reordered(connected, reordered): + configs = [_module_user_config("AAA"), _module_user_config("BBB")] + configs[1]["module"].update(units="cm", pitch=0.1) + transforms = [np.eye(4), np.eye(4)] + transforms[0][0, 3] = 10 + transforms[1][0, 3] = -2 + original = TransducerArray.from_module_user_configs(configs, module_transforms=transforms) + original.attrs["standoff_transform"] = np.array( + [[1, 0, 0, 0], [0, 0, -1, 0], [0, 1, 0, 8], [0, 0, 0, 1]], dtype=float, + ) + serialized = json.loads(json.dumps(original.to_device_config())) + if reordered: + configs.reverse() + configs[0]["device"] = serialized + original_configs = copy.deepcopy(configs) + + if connected: + rebuilt = TransducerArray.get_connected(interface=_FakeInterface(configs), use_default_template=False) + else: + rebuilt = TransducerArray.from_module_user_configs(configs) + + assert rebuilt.attrs["standoff_transform"][2, 3] == pytest.approx(0.8 if reordered else 8) + expected = original.to_transducer() + flattened = rebuilt.to_transducer() + np.testing.assert_allclose( + flattened.get_standoff_transform_in_units("mm"), expected.get_standoff_transform_in_units("mm"), + ) + expected_modules = original.modules[::-1] if reordered else original.modules + for module, expected_module in zip(rebuilt.modules, expected_modules): + np.testing.assert_allclose(module.bake().get_positions(units="mm"), expected_module.bake().get_positions(units="mm")) + assert configs == original_configs + + +@pytest.mark.parametrize("recorded_hwids", [["AAA", "AAA"], [None, None]]) +@pytest.mark.parametrize("connected", [False, True]) +def test_translated_device_standoff_rejects_ambiguous_mixed_units(recorded_hwids, connected): + configs = [_module_user_config("AAA"), _module_user_config("AAA")] + configs[1]["module"].update(units="cm", pitch=0.1) + standoff = np.eye(4) + standoff[2, 3] = 8 + configs[0]["device"] = { + "modules": [{"hwid": hwid} for hwid in recorded_hwids], + "attrs": {"standoff_transform": standoff.tolist()}, + } + if connected: + with pytest.raises(DeviceConfigMismatchError, match="(?i)ambiguous|units"): + TransducerArray.get_connected(interface=_FakeInterface(configs), use_default_template=False) + else: + with pytest.raises(DeviceConfigMismatchError, match="(?i)ambiguous|units"): + TransducerArray.from_module_user_configs(configs) + + +@pytest.mark.parametrize("mixed_units", [False, True]) +def test_device_standoff_allows_ambiguous_ids_when_unit_conversion_is_unnecessary(mixed_units): + configs = [_module_user_config("AAA"), _module_user_config("AAA")] + standoff = np.eye(4) + if mixed_units: + configs[1]["module"].update(units="cm", pitch=0.1) + else: + standoff[2, 3] = 8 + configs[0]["device"] = { + "modules": [{"hwid": "AAA"}, {"hwid": "AAA"}], + "attrs": {"standoff_transform": standoff.tolist()}, + } + + array = TransducerArray.from_module_user_configs(configs) + + np.testing.assert_array_equal(array.attrs["standoff_transform"], standoff) + np.testing.assert_array_equal(array.to_transducer().get_standoff_transform_in_units("mm"), standoff) + + +def test_device_standoff_infers_missing_recorded_id_units_from_remaining_module(): + configs = [_module_user_config(None), _module_user_config("AAA"), _module_user_config("BBB")] + configs[0]["module"].update(units="m", pitch=0.001) + configs[2]["module"].update(units="cm", pitch=0.1) + original = TransducerArray.from_module_user_configs(configs) + original.attrs["standoff_transform"] = np.eye(4) + original.attrs["standoff_transform"][2, 3] = 0.008 + configs = [configs[1], configs[2], configs[0]] + configs[0]["device"] = original.to_device_config() + + rebuilt = TransducerArray.from_module_user_configs(configs) + + assert rebuilt.modules[0].units == "mm" + assert rebuilt.attrs["standoff_transform"][2, 3] == pytest.approx(8) + np.testing.assert_allclose( + rebuilt.to_transducer().get_standoff_transform_in_units("mm"), + original.to_transducer().get_standoff_transform_in_units("mm"), + ) + + +def test_device_standoff_accepts_matching_origin_units_in_otherwise_mixed_array(): + configs = [_module_user_config(hwid) for hwid in ["AAA", "DUP", "DUP"]] + configs[0]["module"].update(units="cm", pitch=0.1) + standoff = np.eye(4) + standoff[2, 3] = 8 + configs[0]["device"] = { + "modules": [{"hwid": hwid} for hwid in ["DUP", "AAA", "DUP"]], + "attrs": {"standoff_transform": standoff.tolist()}, + } + + array = TransducerArray.from_module_user_configs(configs) + + assert array.attrs["standoff_transform"][2, 3] == pytest.approx(0.8) + np.testing.assert_allclose(array.to_transducer().get_standoff_transform_in_units("mm"), standoff) + + +def test_connected_array_saves_loads_and_flattens_with_temporary_mesh_files(tmp_path): + db = Database.initialize_empty_database(tmp_path / "db") + configs = [_module_user_config("AAA"), _module_user_config("BBB")] + array = TransducerArray.get_connected(interface=_FakeInterface(configs), arr_id="device") + mesh_text = "v 0 0 0\nv 1 0 0\nv 0 1 0\nf 1 2 3\n" + surface_path = tmp_path / "surface.obj" + body_path = tmp_path / "body.obj" + surface_path.write_text(mesh_text) + body_path.write_text(mesh_text) + db.write_transducer(array, surface_path, body_path) + loaded = db.load_transducer(array.id, convert_array=False) + assert isinstance(loaded, TransducerArray) + assert arrays_structurally_equal(array, loaded) + paths = db.get_transducer_absolute_filepaths(array.id) + assert (tmp_path / paths["registration_surface_abspath"]).read_text() == mesh_text + assert (tmp_path / paths["transducer_body_abspath"]).read_text() == mesh_text + flattened = db.load_transducer(array.id) + assert isinstance(flattened, Transducer) + assert flattened.numelements() == 8 + assert flattened.registration_surface_filename == surface_path.name + np.testing.assert_array_equal(flattened.standoff_transform, array.attrs["standoff_transform"]) + expected_positions = np.concatenate([module.bake().get_positions() for module in array.modules]) + np.testing.assert_allclose(flattened.get_positions(), expected_positions) diff --git a/tests/test_transducer_array_mesh_units.py b/tests/test_transducer_array_mesh_units.py new file mode 100644 index 00000000..1f51deb3 --- /dev/null +++ b/tests/test_transducer_array_mesh_units.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +import copy + +import numpy as np +import pytest +from vtkmodules.vtkIOGeometry import vtkOBJReader + +from openlifu.db import Database +from openlifu.util.units import getunitconversion +from openlifu.xdc import TransducerArray + + +def _config(units="mm"): + scale = getunitconversion("mm", units) + return { + "hwid": "AAA", + "module": {"nx": 2, "ny": 1, "pitch": 20 * scale, "kerf": scale, "units": units}, + } + + +@pytest.mark.parametrize("mesh_field", ["registration_surface_filename", "transducer_body_filename"]) +@pytest.mark.parametrize("mesh_level", ["array", "module"]) +@pytest.mark.parametrize("units", ["mm", "millimeter"]) +def test_inherited_mesh_matches_elements_after_database_roundtrip(tmp_path, mesh_field, mesh_level, units): + db = Database.initialize_empty_database(tmp_path / "db") + mesh = tmp_path / "surface.obj" + mesh.write_text("v -10 0 0\nv 10 0 0\nv -10 1 0\nf 1 2 3\n", encoding="utf-8") + template = TransducerArray.from_module_user_configs([_config()], arr_id="template") + if mesh_level == "module": + setattr(template.modules[0], mesh_field, mesh.name) + write_kwarg = mesh_field.replace("_filename", "_model_filepath") + db.write_transducer(template, **{write_kwarg: mesh}) + template = db.load_transducer("template", convert_array=False) + if mesh_level == "module": + template.attrs.pop(mesh_field) + + array = TransducerArray.from_module_user_configs([_config(units)], template=template, arr_id="device") + db.write_transducer(array, **{write_kwarg: mesh}) + loaded = db.load_transducer("device") + mesh_path = db.get_transducer_absolute_filepaths("device")[mesh_field.replace("_filename", "_abspath")] + reader = vtkOBJReader() + reader.SetFileName(mesh_path) + reader.Update() + vertex_mm = np.array(reader.GetOutput().GetPoint(0)) * getunitconversion(loaded.units, "mm") + np.testing.assert_allclose(vertex_mm, loaded.get_positions(units="mm")[0]) + np.testing.assert_allclose(vertex_mm, [-10, 0, 0]) + + original_template = copy.deepcopy(template.to_dict()) + with pytest.raises(ValueError, match="[Mm]esh.*units"): + TransducerArray.from_module_user_configs([_config("cm")], template=template, arr_id="invalid") + assert template.to_dict() == original_template + assert "invalid" not in db.get_transducer_ids() + + +@pytest.mark.parametrize("mesh_field", ["registration_surface_filename", "transducer_body_filename"]) +@pytest.mark.parametrize("override", [None, "", "surface.obj", "renamed.obj"]) +def test_array_mesh_override_preserves_unit_guard(mesh_field, override): + template = TransducerArray.from_module_user_configs([_config()]) + template.attrs[mesh_field] = "surface.obj" + config = _config("cm") + config["device"] = {"modules": [{"hwid": "AAA"}], "attrs": {mesh_field: override}} + if override: + with pytest.raises(ValueError, match="[Mm]esh.*units"): + TransducerArray.from_module_user_configs([config], template=template) + else: + array = TransducerArray.from_module_user_configs([config], template=template) + assert array.attrs[mesh_field] == override + np.testing.assert_allclose(array.to_transducer().get_positions(units="mm")[0], [-10, 0, 0]) + + +@pytest.mark.parametrize("mesh_field", ["registration_surface_filename", "transducer_body_filename"]) +def test_array_mesh_requires_template_module_units(mesh_field): + template = TransducerArray(attrs={mesh_field: "surface.obj"}) + with pytest.raises(ValueError, match="[Mm]esh.*template.*module"): + TransducerArray.from_module_user_configs([_config()], template=template) + + +def test_array_mesh_uses_first_module_unit_basis(): + first, second = _config(), _config("cm") + second["hwid"] = "BBB" + template = TransducerArray.from_module_user_configs([first, second]) + template.attrs["transducer_body_filename"] = "body.obj" + array = TransducerArray.from_module_user_configs([first, first], template=template) + assert array.transducer_body_filename == "body.obj" + with pytest.raises(ValueError, match="[Mm]esh.*units"): + TransducerArray.from_module_user_configs([second, first], template=template) + + +@pytest.mark.parametrize("mesh_field", ["registration_surface_filename", "transducer_body_filename"]) +@pytest.mark.parametrize("mesh_reference", [None, "", "surface.obj"]) +def test_device_mesh_rejects_reordered_module_units(mesh_field, mesh_reference): + first, second = _config(), _config("cm") + second["hwid"] = "BBB" + array = TransducerArray.from_module_user_configs([first, second]) + array.attrs[mesh_field] = mesh_reference + device = array.to_device_config() + unchanged = copy.deepcopy([first, second]) + unchanged[0]["device"] = device + restored = TransducerArray.from_module_user_configs(unchanged) + assert restored.attrs[mesh_field] == mesh_reference + assert restored.to_transducer().units == "mm" + + reordered = copy.deepcopy([second, first]) + reordered[0]["device"] = device + if mesh_reference: + with pytest.raises(ValueError, match="[Mm]esh.*units"): + TransducerArray.from_module_user_configs(reordered) + else: + restored = TransducerArray.from_module_user_configs(reordered) + assert restored.attrs[mesh_field] == mesh_reference + assert restored.to_transducer().units == "cm"