Re-order ADC contact annotations to match device_channel_index

Author: alejoe91

src/probeinterface/neuropixels_tools.py

@@ -1752,6 +1752,12 @@ def read_openephys(
     if oebin_file is not None:
         device_channel_indices = _compute_wiring_from_oebin(probe, oebin_file, stream_name, settings_file)
         probe.set_device_channel_indices(device_channel_indices)
+
+        # re-order contact annotations to match device channel order
+        ordered_adc_groups = [probe.contact_annotations["adc_group"][i] for i in device_channel_indices]
+        probe.annotate_contacts(adc_group=ordered_adc_groups)
+        ordered_adc_sample_orders = [probe.contact_annotations["adc_sample_order"][i] for i in device_channel_indices]
+        probe.annotate_contacts(adc_sample_order=ordered_adc_sample_orders)
     else:
         probe.set_device_channel_indices(np.arange(probe.get_contact_count()))
     return probe

tests/test_io/test_openephys.py

@@ -499,11 +499,9 @@ def _read_oebin_electrode_indices(oebin_file, stream_name):
     """Read electrode_index metadata from an oebin file for a given stream."""
     with open(oebin_file) as f:
         oebin = json.load(f)
-    is_neo_stream = "#" in stream_name
-    oebin_stream_name = stream_name.split("#")[-1] if is_neo_stream else stream_name
     for cs in oebin.get("continuous", []):
         folder_name = cs.get("folder_name", "").rstrip("/")
-        if folder_name == oebin_stream_name:
+        if folder_name == stream_name:
             indices = []
             for ch in cs.get("channels", []):
                 for m in ch.get("channel_metadata", []):
@@ -526,6 +524,8 @@ def test_read_openephys_with_oebin_wiring():
 
     assert probe.get_contact_count() == 384
     assert probe.device_channel_indices is not None
+    assert "adc_group" in probe.contact_annotations
+    assert "adc_sample_order" in probe.contact_annotations
 
     # Wiring invariant
     oebin_electrode_indices = _read_oebin_electrode_indices(oebin, stream_name)
@@ -708,7 +708,6 @@ def test_read_openephys_onebox_nonsequential_wiring():
             f"at column {column}, got {oebin_electrode_indices[column]}"
         )
 
-
 if __name__ == "__main__":
     # test_multiple_probes()
     # test_NP_Ultra()