adjust frequency-dependent sensitivity (#439)

Author: peterhollender

src/openlifu/xdc/element.py

@@ -114,7 +114,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[float, float], OpenLIFUFieldData("Sensitivity", "Sensitivity of the element (Pa/V), scalar or {frequency_hz: sensitivity}")] = 1.0
+    sensitivity: Annotated[float | dict, OpenLIFUFieldData("Sensitivity", "Sensitivity of the element (Pa/V), scalar or {'freq_Hz':[...], 'values_Pa_per_V':[...]}")] = 1.0
     """Sensitivity of the element (Pa/V)"""
 
     pin: Annotated[int, OpenLIFUFieldData("Pin", "Channel pin to which the element is connected")] = -1
@@ -135,16 +135,7 @@ def __post_init__(self):
             raise ValueError("Size must be a 2-element array.")
         if self.sensitivity is None:
             self.sensitivity = 1.0
-        elif isinstance(self.sensitivity, dict):
-            if len(self.sensitivity) == 0:
-                raise ValueError("Sensitivity dictionary must not be empty.")
-            mapping = {float(k): float(v) for k, v in self.sensitivity.items()}
-            freqs = np.array(sorted(mapping.keys()), dtype=np.float64)
-            if np.any(np.diff(freqs) <= 0):
-                raise ValueError("Sensitivity dictionary frequencies must be strictly increasing.")
-            self.sensitivity = {float(f): mapping[float(f)] for f in freqs}
-        else:
-            self.sensitivity = float(self.sensitivity)
+        self.sensitivity = normalize_sensitivity(self.sensitivity)
 
     @property
     def x(self):

src/openlifu/xdc/transducer.py

@@ -14,6 +14,7 @@
 from openlifu.xdc.element import (
     Element,
     generate_drive_signal,
+    normalize_sensitivity,
     sensitivity_at_frequency,
 )
 
@@ -22,21 +23,37 @@
 
 
 def _combine_sensitivities(
-    base_sensitivity: float | dict[float, float],
-    scale_sensitivity: float | dict[float, float],
-) -> float | dict[float, float]:
+    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_keys = tuple(base_sensitivity.keys())
-        scale_keys = tuple(scale_sensitivity.keys())
-        if base_keys != scale_keys:
+        base_freqs = np.asarray(base_sensitivity["freq_Hz"], dtype=np.float64)
+        scale_freqs = np.asarray(scale_sensitivity["freq_Hz"], dtype=np.float64)
+        if not np.array_equal(base_freqs, scale_freqs):
             raise ValueError("Cannot combine sensitivity dictionaries with different frequency keys.")
-        return {frequency: base_sensitivity[frequency] * scale_sensitivity[frequency] for frequency in base_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):
         factor = float(scale_sensitivity)
-        return {frequency: value * factor for frequency, value in base_sensitivity.items()}
+        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):
         factor = float(base_sensitivity)
-        return {frequency: factor * value for frequency, value in scale_sensitivity.items()}
+        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)
 
 @dataclass
@@ -78,7 +95,7 @@ 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[float, float], OpenLIFUFieldData("Sensitivity", "Sensitivity of the element (Pa/V), scalar or {frequency_hz: sensitivity}")] = 1.0
+    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."""
 
     crosstalk_frac: Annotated[float, OpenLIFUFieldData("Crosstalk fraction", "Fraction of the signal that leaks into other elements due to crosstalk")] = 0.0
@@ -98,16 +115,7 @@ def __post_init__(self):
             element.rescale(self.units)
         if self.sensitivity is None:
             self.sensitivity = 1.0
-        elif isinstance(self.sensitivity, dict):
-            if len(self.sensitivity) == 0:
-                raise ValueError("Sensitivity dictionary must not be empty.")
-            mapping = {float(k): float(v) for k, v in self.sensitivity.items()}
-            freqs = np.array(sorted(mapping.keys()), dtype=np.float64)
-            if np.any(np.diff(freqs) <= 0):
-                raise ValueError("Sensitivity dictionary frequencies must be strictly increasing.")
-            self.sensitivity = {float(f): mapping[float(f)] for f in freqs}
-        else:
-            self.sensitivity = float(self.sensitivity)
+        self.sensitivity = normalize_sensitivity(self.sensitivity)
 
     def get_sensitivity(self, frequency: float) -> float:
         return sensitivity_at_frequency(self.sensitivity, frequency)
@@ -252,18 +260,23 @@ 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.keys()))
+                dict_key_sets.add(tuple(array.sensitivity["freq_Hz"]))
             for el in array.elements:
+                el.sensitivity = normalize_sensitivity(el.sensitivity)
                 if isinstance(el.sensitivity, dict):
-                    dict_key_sets.add(tuple(el.sensitivity.keys()))
+                    dict_key_sets.add(tuple(el.sensitivity["freq_Hz"]))
         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):
-                sensitivity_signatures.append((tuple(array.sensitivity.keys()), tuple(array.sensitivity.values())))
+                sensitivity_signatures.append((
+                    tuple(array.sensitivity["freq_Hz"]),
+                    tuple(array.sensitivity["values_Pa_per_V"]),
+                ))
             else:
                 sensitivity_signatures.append(float(array.sensitivity))
 

src/openlifu/xdc/util.py

@@ -300,7 +300,10 @@ 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'] = {f*1e3:sens for f, sens, in zip(freq_df["Frequency"], freq_df['Sensitivity'])}
+    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['id'] = matrix_dict['id'].format(sn=sn.lower())
     matrix_dict['name'] = matrix_dict['name'].format(sn=sn)
     return matrix_dict

tests/test_transducer.py

@@ -116,7 +116,7 @@ def test_transducer_calc_output_interpolates_dictionary_sensitivity():
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={100e3: 1.0, 300e3: 3.0},
+        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [1.0, 3.0]},
     )
     transducer.elements[0].sensitivity = 1.0
     input_signal = np.array([1.0, -1.0, 0.5], dtype=float)
@@ -128,6 +128,16 @@ def test_transducer_calc_output_interpolates_dictionary_sensitivity():
     np.testing.assert_allclose(output_low[0, :len(input_signal)], 1.0 * input_signal)
 
 
+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
@@ -161,39 +171,39 @@ def test_merge_pushes_transducer_sensitivity_into_elements():
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={100e3: 2.0, 300e3: 4.0},
+        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [2.0, 4.0]},
     )
     transducer_b = Transducer.gen_matrix_array(
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={100e3: 3.0, 300e3: 6.0},
+        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [3.0, 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 == {100e3: 10.0, 300e3: 20.0}
-    assert merged.elements[1].sensitivity == {100e3: 21.0, 300e3: 42.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]}
 
 
 def test_merge_rejects_mismatched_sensitivity_keys():
     transducer_a = Transducer.gen_matrix_array(
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={100e3: 2.0, 300e3: 4.0},
+        sensitivity={"freq_Hz": [100e3, 300e3], "values_Pa_per_V": [2.0, 4.0]},
     )
     transducer_b = Transducer.gen_matrix_array(
         nx=1,
         ny=1,
         units="mm",
-        sensitivity={100e3: 3.0, 400e3: 6.0},
+        sensitivity={"freq_Hz": [100e3, 400e3], "values_Pa_per_V": [3.0, 6.0]},
     )
-    transducer_a.elements[0].sensitivity = {100e3: 5.0, 300e3: 7.0}
-    transducer_b.elements[0].sensitivity = {100e3: 11.0, 400e3: 13.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]}
 
     with pytest.raises(ValueError, match="different frequency keys"):
         Transducer.merge([transducer_a, transducer_b], merge_mismatched_sensitivity=True)