Update to list of tuples form #439

Author: peterhollender

src/openlifu/xdc/element.py

@@ -2,68 +2,23 @@
 
 import copy
 from dataclasses import dataclass, field
-from typing import Annotated
+from typing import Annotated, List
 
 import numpy as np
 
 from openlifu.util.annotations import OpenLIFUFieldData
 from openlifu.util.units import getunitconversion
 
-SENS_FREQ_KEY = "freq_Hz"
-SENS_VALUE_KEY = "values_Pa_per_V"
 
-
-def normalize_sensitivity(sensitivity: float | dict) -> float | dict[str, list[float]]:
-    """Normalize sensitivity to a canonical representation.
-
-    Canonical frequency-dependent representation is:
-        {"freq_Hz": [...], "values_Pa_per_V": [...]}
-
-    Backward-compatible legacy representation with frequency keys is accepted and
-    converted to the canonical representation.
-    """
-    if isinstance(sensitivity, dict):
-        if SENS_FREQ_KEY in sensitivity or SENS_VALUE_KEY in sensitivity:
-            if SENS_FREQ_KEY not in sensitivity or SENS_VALUE_KEY not in sensitivity:
-                raise ValueError("Sensitivity dictionary must include both 'freq_Hz' and 'values_Pa_per_V'.")
-            freqs = np.asarray(sensitivity[SENS_FREQ_KEY], dtype=np.float64).reshape(-1)
-            values = np.asarray(sensitivity[SENS_VALUE_KEY], dtype=np.float64).reshape(-1)
-        else:
-            # Legacy format: {frequency_hz: sensitivity}
-            if len(sensitivity) == 0:
-                raise ValueError("Sensitivity dictionary must not be empty.")
-            mapping = {float(k): float(v) for k, v in sensitivity.items()}
-            freqs = np.array(list(mapping.keys()), dtype=np.float64)
-            values = np.array(list(mapping.values()), dtype=np.float64)
-
-        if len(freqs) == 0:
-            raise ValueError("Sensitivity frequency list must not be empty.")
-        if len(freqs) != len(values):
-            raise ValueError("Sensitivity frequency and value lists must have the same length.")
-
-        order = np.argsort(freqs)
-        freqs = freqs[order]
-        values = values[order]
-        if np.any(np.diff(freqs) <= 0):
-            raise ValueError("Sensitivity frequencies must be strictly increasing.")
-
-        return {
-            SENS_FREQ_KEY: [float(f) for f in freqs],
-            SENS_VALUE_KEY: [float(v) for v in values],
-        }
-
-    return float(sensitivity)
-
-
-def sensitivity_at_frequency(sensitivity: float | dict, frequency: float) -> float:
-    sensitivity = normalize_sensitivity(sensitivity)
-    if isinstance(sensitivity, dict):
-        if frequency in sensitivity[SENS_FREQ_KEY]:
-            idx = sensitivity[SENS_FREQ_KEY].index(frequency)
-            return float(sensitivity[SENS_VALUE_KEY][idx])
+def sensitivity_at_frequency(sensitivity: float | List[tuple[float, float]], frequency: float) -> float:
+    if isinstance(sensitivity, list):
+        freqs, values = zip(*sensitivity)
+        freqs = np.array(freqs, dtype=np.float64)
+        values = np.array(values, dtype=np.float64)
+        if frequency in freqs:
+            idx = np.where(freqs == frequency)[0][0]
+            return float(values[idx])
         else:
-            freqs = np.array(sensitivity[SENS_FREQ_KEY], dtype=np.float64)
-            values = np.array(sensitivity[SENS_VALUE_KEY], dtype=np.float64)
             return float(np.interp(frequency, freqs, values, left=values[0], right=values[-1]))
     return float(sensitivity)
 
@@ -114,7 +69,7 @@ class Element:
     size: Annotated[np.ndarray, OpenLIFUFieldData("Size", "Size of the element in 2D")] = field(default_factory=lambda: np.array([1., 1.]))
     """ Size of the element in 2D as a numpy array [width, length]."""
 
-    sensitivity: Annotated[float | dict, OpenLIFUFieldData("Sensitivity", "Sensitivity of the element (Pa/V), scalar or {'freq_Hz':[...], 'values_Pa_per_V':[...]}")] = 1.0
+    sensitivity: Annotated[float | List[tuple[float, float]], OpenLIFUFieldData("Sensitivity", "Sensitivity of the element (Pa/V), scalar or list of (frequency, value) tuples")] = 1.0
     """Sensitivity of the element (Pa/V)"""
 
     pin: Annotated[int, OpenLIFUFieldData("Pin", "Channel pin to which the element is connected")] = -1
@@ -135,7 +90,8 @@ def __post_init__(self):
             raise ValueError("Size must be a 2-element array.")
         if self.sensitivity is None:
             self.sensitivity = 1.0
-        self.sensitivity = normalize_sensitivity(self.sensitivity)
+        elif isinstance(self.sensitivity, list):
+            self.sensitivity = [(float(f), float(v)) for f, v in self.sensitivity]
 
     @property
     def x(self):

src/openlifu/xdc/transducer.py

@@ -14,7 +14,6 @@
 from openlifu.xdc.element import (
     Element,
     generate_drive_signal,
-    normalize_sensitivity,
     sensitivity_at_frequency,
 )
 
@@ -23,38 +22,28 @@
 
 
 def _combine_sensitivities(
-    base_sensitivity: float | dict,
-    scale_sensitivity: float | dict,
-) -> float | dict[str, list[float]]:
-    base_sensitivity = normalize_sensitivity(base_sensitivity)
-    scale_sensitivity = normalize_sensitivity(scale_sensitivity)
-
-    if isinstance(base_sensitivity, dict) and isinstance(scale_sensitivity, dict):
-        base_freqs = np.asarray(base_sensitivity["freq_Hz"], dtype=np.float64)
-        scale_freqs = np.asarray(scale_sensitivity["freq_Hz"], dtype=np.float64)
+    base_sensitivity: float | List[tuple[float, float]],
+    scale_sensitivity: float | List[tuple[float, float]],
+) -> float | List[tuple[float, float]]:
+
+    if isinstance(base_sensitivity, list) and isinstance(scale_sensitivity, list):
+        base_freqs = np.asarray([f for f, _ in base_sensitivity], dtype=np.float64)
+        scale_freqs = np.asarray([f for f, _ in scale_sensitivity], dtype=np.float64)
         if not np.array_equal(base_freqs, scale_freqs):
             raise ValueError("Cannot combine sensitivity dictionaries with different frequency keys.")
-        base_values = np.asarray(base_sensitivity["values_Pa_per_V"], dtype=np.float64)
-        scale_values = np.asarray(scale_sensitivity["values_Pa_per_V"], dtype=np.float64)
-        return {
-            "freq_Hz": [float(f) for f in base_freqs],
-            "values_Pa_per_V": [float(v) for v in (base_values * scale_values)],
-        }
-    if isinstance(base_sensitivity, dict):
+        base_values = np.asarray([v for _, v in base_sensitivity], dtype=np.float64)
+        scale_values = np.asarray([v for _, v in scale_sensitivity], dtype=np.float64)
+        return [(float(f), float(v)) for f, v in zip(base_freqs, base_values * scale_values)]
+    elif isinstance(base_sensitivity, list):
         factor = float(scale_sensitivity)
-        values = np.asarray(base_sensitivity["values_Pa_per_V"], dtype=np.float64)
-        return {
-            "freq_Hz": [float(f) for f in base_sensitivity["freq_Hz"]],
-            "values_Pa_per_V": [float(v) for v in (values * factor)],
-        }
-    if isinstance(scale_sensitivity, dict):
+        values = np.asarray([v for _, v in base_sensitivity], dtype=np.float64)
+        return [(float(f), float(v)) for (f, _), v in zip(base_sensitivity, values * factor)]
+    elif isinstance(scale_sensitivity, list):
         factor = float(base_sensitivity)
-        values = np.asarray(scale_sensitivity["values_Pa_per_V"], dtype=np.float64)
-        return {
-            "freq_Hz": [float(f) for f in scale_sensitivity["freq_Hz"]],
-            "values_Pa_per_V": [float(v) for v in (factor * values)],
-        }
-    return float(base_sensitivity) * float(scale_sensitivity)
+        values = np.asarray([v for _, v in scale_sensitivity], dtype=np.float64)
+        return [(float(f), float(v)) for (f, _), v in zip(scale_sensitivity, factor * values)]
+    else:
+        return float(base_sensitivity) * float(scale_sensitivity)
 
 @dataclass
 class Transducer:
@@ -95,8 +84,8 @@ class Transducer:
     The units of this transform are assumed to be the native units of the transducer, the `Transducer.units` field.
     """
 
-    sensitivity: Annotated[float | dict, OpenLIFUFieldData("Sensitivity", "Sensitivity of the transducer (Pa/V), scalar or {'freq_Hz':[...], 'values_Pa_per_V':[...]}")] = 1.0
-    """Sensitivity of the transducer (Pa/V), scalar or frequency-dependent dictionary."""
+    sensitivity: Annotated[float | List[tuple[float, float]], OpenLIFUFieldData("Sensitivity", "Sensitivity of the transducer (Pa/V), scalar or list of (frequency, value) tuples")] = 1.0
+    """Sensitivity of the transducer (Pa/V), scalar or frequency-dependent list of tuples."""
 
     crosstalk_frac: Annotated[float, OpenLIFUFieldData("Crosstalk fraction", "Fraction of the signal that leaks into other elements due to crosstalk")] = 0.0
     """Fraction of the signal that leaks into other elements due to crosstalk"""
@@ -115,21 +104,19 @@ def __post_init__(self):
             element.rescale(self.units)
         if self.sensitivity is None:
             self.sensitivity = 1.0
-        self.sensitivity = normalize_sensitivity(self.sensitivity)
-
-    def get_sensitivity(self, frequency: float) -> float:
-        return sensitivity_at_frequency(self.sensitivity, frequency)
+        elif isinstance(self.sensitivity, list):
+            self.sensitivity = [(float(f), float(v)) for f, v in self.sensitivity]
 
     def calc_output(self, cycles: float, frequency: float, dt: float, delays: np.ndarray = None, apod: np.ndarray = None, amplitude: float = 1.0) -> np.ndarray:
         if delays is None:
             delays = np.zeros(self.numelements())
         if apod is None:
             apod = np.ones(self.numelements())
         drive_signal = generate_drive_signal(cycles=cycles, frequency=frequency, dt=dt, amplitude=amplitude)
-        base_output = drive_signal * self.get_sensitivity(frequency)
+        base_output = drive_signal * sensitivity_at_frequency(self.sensitivity, frequency)
         outputs = [
             np.concatenate(
-                [np.zeros(int(delay / dt)), a * element.get_sensitivity(frequency) * base_output],
+                [np.zeros(int(delay / dt)), a * sensitivity_at_frequency(element.sensitivity, frequency) * base_output],
                 axis=0,
             )
             for element, delay, a, in zip(self.elements, delays, apod)
@@ -260,22 +247,20 @@ def merge(list_of_transducers:List[Transducer], offset_pins:bool=False, offset_i
         array_copies = [arr.copy() for arr in list_of_transducers]
         dict_key_sets = set()
         for array in array_copies:
-            array.sensitivity = normalize_sensitivity(array.sensitivity)
-            if isinstance(array.sensitivity, dict):
-                dict_key_sets.add(tuple(array.sensitivity["freq_Hz"]))
+            if isinstance(array.sensitivity, list):
+                dict_key_sets.add(tuple(f for f, _ in array.sensitivity))
             for el in array.elements:
-                el.sensitivity = normalize_sensitivity(el.sensitivity)
-                if isinstance(el.sensitivity, dict):
-                    dict_key_sets.add(tuple(el.sensitivity["freq_Hz"]))
+                if isinstance(el.sensitivity, list):
+                    dict_key_sets.add(tuple(f for f, _ in el.sensitivity))
         if len(dict_key_sets) > 1:
             raise ValueError("Cannot merge sensitivities with different frequency keys.")
 
         sensitivity_signatures = []
         for array in array_copies:
-            if isinstance(array.sensitivity, dict):
+            if isinstance(array.sensitivity, list):
                 sensitivity_signatures.append((
-                    tuple(array.sensitivity["freq_Hz"]),
-                    tuple(array.sensitivity["values_Pa_per_V"]),
+                    tuple(f for f, _ in array.sensitivity),
+                    tuple(v for _, v in array.sensitivity),
                 ))
             else:
                 sensitivity_signatures.append(float(array.sensitivity))

src/openlifu/xdc/util.py

@@ -300,10 +300,7 @@ def report_to_matrix_dict(report_df: pd.DataFrame, focal_gain_lut=FOCAL_GAIN_LUT
     freq_df["Frequency"] = freq_df['Item'].apply(lambda x: float(re.search(r"(?<=^PNP \()\d+(?= kHz\)$)", x).group(0)))
     freq_df['focal_gain'] = freq_df['Frequency'].apply(lambda f: focal_gain_lut.interp(f0=f*1e3, crosstalk=matrix_dict['crosstalk_frac']).item())
     freq_df['Sensitivity'] = freq_df['PNP'].astype(float)*1e6/freq_df['focal_gain']/voltage
-    matrix_dict['sensitivity'] = {
-        'freq_Hz': [float(f * 1e3) for f in freq_df["Frequency"]],
-        'values_Pa_per_V': [float(sens) for sens in freq_df['Sensitivity']],
-    }
+    matrix_dict['sensitivity'] = [(f*1e3, sens) for f, sens in zip(freq_df["Frequency"], freq_df['Sensitivity'])]
     matrix_dict['id'] = matrix_dict['id'].format(sn=sn.lower())
     matrix_dict['name'] = matrix_dict['name'].format(sn=sn)
     return matrix_dict

tests/test_transducer.py

@@ -121,7 +121,7 @@ def test_transducer_calc_output_interpolates_dictionary_sensitivity():
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [1.0, 3.0]},
+        sensitivity=[(100e3, 1.0), (300e3, 3.0)],
     )
     transducer.elements[0].sensitivity = 1.0
     cycles = 3
@@ -141,16 +141,6 @@ def test_transducer_calc_output_interpolates_dictionary_sensitivity():
     np.testing.assert_allclose(output_low[0], expected_low)
 
 
-def test_legacy_sensitivity_mapping_is_normalized_to_schema():
-    transducer = Transducer(
-        sensitivity={"100000.0": 1.0, "300000.0": 3.0},
-    )
-    assert transducer.sensitivity == {
-        "freq_Hz": [100000.0, 300000.0],
-        "values_Pa_per_V": [1.0, 3.0],
-    }
-
-
 def test_element_calc_output_generates_signal_from_scalar_input():
     element = Element(sensitivity=2.0)
     cycles = 4
@@ -184,39 +174,39 @@ def test_merge_pushes_transducer_sensitivity_into_elements():
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [2.0, 4.0]},
+        sensitivity=[(100e3, 2.0), (300e3, 4.0)],
     )
     transducer_b = Transducer.gen_matrix_array(
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [3.0, 6.0]},
+        sensitivity=[(100e3, 3.0), (300e3, 6.0)],
     )
     transducer_a.elements[0].sensitivity = 5.0
     transducer_b.elements[0].sensitivity = 7.0
 
     merged = Transducer.merge([transducer_a, transducer_b], merge_mismatched_sensitivity=True)
 
     assert merged.sensitivity == 1.0
-    assert merged.elements[0].sensitivity == {"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [10.0, 20.0]}
-    assert merged.elements[1].sensitivity == {"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [21.0, 42.0]}
+    assert merged.elements[0].sensitivity == [(100e3, 10.0),(300e3, 20.0)]
+    assert merged.elements[1].sensitivity == [(100e3, 21.0),(300e3, 42.0)]
 
 
 def test_merge_rejects_mismatched_sensitivity_keys():
     transducer_a = Transducer.gen_matrix_array(
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [2.0, 4.0]},
+        sensitivity=[(100e3, 2.0), (300e3, 4.0)],
     )
     transducer_b = Transducer.gen_matrix_array(
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={"freq_Hz": [100e3, 400e3], "values_Pa_per_V": [3.0, 6.0]},
+        sensitivity=[(100e3, 2.0), (300e3, 4.0)],
     )
-    transducer_a.elements[0].sensitivity = {"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [5.0, 7.0]}
-    transducer_b.elements[0].sensitivity = {"freq_Hz": [100e3, 400e3], "values_Pa_per_V": [11.0, 13.0]}
+    transducer_a.elements[0].sensitivity = [(100e3, 5.0), (300e3, 7.0)]
+    transducer_b.elements[0].sensitivity = [(100e3, 11.0), (400e3, 13.0)]
 
     with pytest.raises(ValueError, match="different frequency keys"):
         Transducer.merge([transducer_a, transducer_b], merge_mismatched_sensitivity=True)
@@ -307,9 +297,9 @@ def test_transducer_calc_output_combines_frequency_dependent_sensitivities():
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [2.0, 4.0]},
+        sensitivity=[(100e3, 2.0), (300e3, 4.0)],
     )
-    transducer.elements[0].sensitivity = {"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [5.0, 9.0]}
+    transducer.elements[0].sensitivity = [(100e3, 5.0), (300e3, 9.0)]
 
     frequency = 200e3
     dt = 1e-7
@@ -328,13 +318,13 @@ def test_transducer_array_to_transducer_preserves_frequency_dependent_sensitivit
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [2.0, 4.0]},
+        sensitivity=[(100e3, 2.0), (300e3, 4.0)],
     )
     transducer_b = Transducer.gen_matrix_array(
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [1.0, 3.0]},
+        sensitivity=[(100e3, 1.0), (300e3, 3.0)],
     )
     transducer_a.elements[0].sensitivity = 5.0
     transducer_b.elements[0].sensitivity = 7.0
@@ -360,3 +350,48 @@ def test_transducer_array_to_transducer_preserves_frequency_dependent_sensitivit
 
     np.testing.assert_allclose(output[0], 15.0 * expected_drive)
     np.testing.assert_allclose(output[1], 14.0 * expected_drive)
+
+
+def test_element_sensitivity_from_json_is_list_of_tuples():
+    """Sensitivity read from a JSON dict (list-of-lists) is converted to List[tuple[float, float]]."""
+    d = {
+        "index": 1,
+        "position": [0.0, 0.0, 0.0],
+        "orientation": [0.0, 0.0, 0.0],
+        "size": [1.0, 1.0],
+        "pin": 1,
+        "units": "mm",
+        "sensitivity": [[100e3, 1.0], [300e3, 3.0]],  # JSON encodes tuples as lists
+    }
+    element = Element.from_dict(d)
+    assert isinstance(element.sensitivity, list)
+    assert all(isinstance(pair, tuple) for pair in element.sensitivity)
+    assert all(isinstance(f, float) and isinstance(v, float) for f, v in element.sensitivity)
+    assert element.sensitivity == [(100e3, 1.0), (300e3, 3.0)]
+
+
+def test_transducer_sensitivity_from_json_is_list_of_tuples():
+    """Transducer-level sensitivity survives a to_json/from_json round-trip as List[tuple[float, float]]."""
+    transducer = Transducer.gen_matrix_array(
+        nx=1,
+        ny=1,
+        units="mm",
+        sensitivity=[(100e3, 2.0), (300e3, 4.0)],
+    )
+    reconstructed = Transducer.from_json(transducer.to_json())
+    assert isinstance(reconstructed.sensitivity, list)
+    assert all(isinstance(pair, tuple) for pair in reconstructed.sensitivity)
+    assert all(isinstance(f, float) and isinstance(v, float) for f, v in reconstructed.sensitivity)
+    assert reconstructed.sensitivity == [(100e3, 2.0), (300e3, 4.0)]
+
+
+def test_element_in_transducer_sensitivity_from_json_is_list_of_tuples():
+    """Element-level sensitivity inside a Transducer survives a to_json/from_json round-trip as List[tuple[float, float]]."""
+    transducer = Transducer.gen_matrix_array(nx=1, ny=1, units="mm")
+    transducer.elements[0].sensitivity = [(100e3, 5.0), (300e3, 9.0)]
+    reconstructed = Transducer.from_json(transducer.to_json())
+    el_sensitivity = reconstructed.elements[0].sensitivity
+    assert isinstance(el_sensitivity, list)
+    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)]