diff --git a/.gitignore b/.gitignore index 0ee5de6..52a45ce 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,4 @@ uv.lock # libraries **/neuropixels_library_generated **/cambridgeneurotech_library +.codex diff --git a/src/probeinterface/__init__.py b/src/probeinterface/__init__.py index 3317c79..50b7dc1 100644 --- a/src/probeinterface/__init__.py +++ b/src/probeinterface/__init__.py @@ -28,6 +28,8 @@ parse_spikeglx_snsGeomMap, get_saved_channel_indices_from_spikeglx_meta, read_openephys, + read_openephys_neuropixels, + has_neuropixels_probes, get_saved_channel_indices_from_openephys_settings, ) from .utils import combine_probes diff --git a/src/probeinterface/neuropixels_tools.py b/src/probeinterface/neuropixels_tools.py index 3e9e513..9dd37cc 100644 --- a/src/probeinterface/neuropixels_tools.py +++ b/src/probeinterface/neuropixels_tools.py @@ -997,6 +997,11 @@ def _parse_openephys_settings( - settings_channel_keys: np.array of str, or None - elec_ids, shank_ids: for legacy fallback """ + if not has_neuropixels_probes(settings_file): + if raise_error: + raise Exception("No Neuropixels probe geometry found in settings file") + return None + ET = import_safely("xml.etree.ElementTree") tree = ET.parse(str(settings_file)) root = tree.getroot() @@ -1035,11 +1040,6 @@ def _parse_openephys_settings( and channel_map_position < record_node_position ) - if neuropix_pxi_processor is None and onebox_processor is None and onix_processor is None: - if raise_error: - raise Exception("Open Ephys can only be read from Neuropix-PXI, OneBox or ONIX plugins.") - return None - if neuropix_pxi_processor is not None: assert onebox_processor is None, "Only one processor should be present" processor = neuropix_pxi_processor @@ -1484,7 +1484,7 @@ def _annotate_openephys_probe(probe: Probe, probe_info: dict) -> None: _annotate_probe_with_adc_sampling_info(probe, adc_sampling_table) -def read_openephys( +def read_openephys_neuropixels( settings_file: str | Path, stream_name: str | None = None, probe_name: str | None = None, @@ -1495,6 +1495,14 @@ def read_openephys( """ Read a Neuropixels probe geometry from an Open Ephys settings.xml file. + This function only supports Neuropixels probes (those with ```` + or ```` / ```` / ```` + elements in the settings file). It does not handle other Open Ephys + hardware such as Intan acquisition boards, tetrodes, NI-DAQmx, etc. + Use :func:`has_neuropixels_probes` to check whether a settings file (or + a specific stream within it) has Neuropixels probe geometry before calling + this reader. + A single settings.xml can describe multiple probes (one ```` element per probe). When the file contains more than one probe, use one of the three mutually exclusive selectors (``stream_name``, ``probe_name``, or @@ -1575,6 +1583,83 @@ def read_openephys( return probe +def read_openephys(*args, **kwargs) -> Probe: + """ + Deprecated alias for :func:`read_openephys_neuropixels`. + + The name ``read_openephys`` is misleading because the function only reads + Neuropixels probe geometry, not arbitrary Open Ephys recordings. Use + :func:`read_openephys_neuropixels` instead, and :func:`has_neuropixels_probes` + to check whether a settings file has Neuropixels geometry before calling it. + """ + warnings.warn( + "read_openephys is deprecated and will be removed in a future release. " + "Use read_openephys_neuropixels instead.", + category=DeprecationWarning, + stacklevel=2, + ) + return read_openephys_neuropixels(*args, **kwargs) + + +_NP_PROBE_ELEMENT_TAGS = frozenset({"NP_PROBE", "NEUROPIXELSV1E", "NEUROPIXELSV1F", "NEUROPIXELSV2E"}) + + +def has_neuropixels_probes(settings_file: str | Path, stream_name: str | None = None) -> bool: + """ + Return True if the Open Ephys settings file contains Neuropixels probe + geometry elements. + + Detection is element-based: the function scans the settings XML for + ```` (Neuropix-PXI / OneBox) or the ONIX equivalents + ```` / ```` / ````. The + presence of any of these is the ground-truth signal that Neuropixels + geometry is described in the file, independent of processor names. This + is robust to ONIX streams that can carry non-Neuropixels probes and to + new Neuropixels-capable plugins. + + Intended use: callers that route heterogeneous streams (e.g. Open Ephys + recordings mixing Intan / NI-DAQmx / Neuropixels) can gate the call to + :func:`read_openephys_neuropixels` on this helper and skip probe + attachment for non-Neuropixels streams. + + Parameters + ---------- + settings_file : str or Path + Path to the Open Ephys settings.xml file. + stream_name : str or None + If provided, only return True when a Neuropixels probe element lives + under a processor whose STREAM names match ``stream_name``. Matching + mirrors the selection logic in :func:`read_openephys_neuropixels`: a + probe's STREAM name (with ``-AP`` / ``-LFP`` stripped) must appear as + a substring of ``stream_name`` (so ``"ProbeC"`` matches + ``"Neuropix-PXI-100.ProbeC-AP"``). If None, returns True whenever any + Neuropixels probe element is present. + + Returns + ------- + bool + """ + ET = import_safely("xml.etree.ElementTree") + try: + root = ET.parse(str(settings_file)).getroot() + except Exception: + return False + + for processor in root.iter("PROCESSOR"): + if not any(e.tag in _NP_PROBE_ELEMENT_TAGS for e in processor.iter()): + continue + if stream_name is None: + return True + for stream_field in processor.findall("STREAM"): + name = stream_field.attrib.get("name", "") + if "ADC" in name: + continue + probe_name = name.replace("-AP", "").replace("-LFP", "") + if probe_name and probe_name in stream_name: + return True + return False + + def get_saved_channel_indices_from_openephys_settings(settings_file: str | Path, stream_name: str) -> np.ndarray | None: """ Returns an array with the subset of saved channels indices (if used) diff --git a/tests/test_io/test_openephys.py b/tests/test_io/test_openephys.py index ca3363f..de51616 100644 --- a/tests/test_io/test_openephys.py +++ b/tests/test_io/test_openephys.py @@ -7,7 +7,7 @@ import json -from probeinterface import read_openephys +from probeinterface import read_openephys, read_openephys_neuropixels, has_neuropixels_probes from probeinterface.neuropixels_tools import _parse_openephys_settings, _select_openephys_probe_info from probeinterface.neuropixels_tools import _slice_openephys_catalogue_probe, build_neuropixels_probe from probeinterface.testing import validate_probe_dict @@ -38,7 +38,9 @@ def _assert_contact_ids_match_canonical_pattern(probe, label=""): ### TESTS ### def test_NP2_OE_1_0(): # NP2 1-shank - probeA = read_openephys(data_path / "OE_1.0_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="ProbeA") + probeA = read_openephys_neuropixels( + data_path / "OE_1.0_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="ProbeA" + ) probe_dict = probeA.to_dict(array_as_list=True) validate_probe_dict(probe_dict) assert probeA.get_shank_count() == 1 @@ -49,7 +51,7 @@ def test_NP2_OE_1_0(): def test_NP2(): # NP2 - probe = read_openephys(data_path / "OE_Neuropix-PXI" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI" / "settings.xml") probe_dict = probe.to_dict(array_as_list=True) validate_probe_dict(probe_dict) assert probe.get_shank_count() == 1 @@ -59,7 +61,7 @@ def test_NP2(): def test_NP2_four_shank(): # NP2 - probe = read_openephys(data_path / "OE_Neuropix-PXI-NP2-4shank" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI-NP2-4shank" / "settings.xml") probe_dict = probe.to_dict(array_as_list=True) validate_probe_dict(probe_dict) # on this case, only shanks 2-3 are used @@ -70,7 +72,7 @@ def test_NP2_four_shank(): def test_NP_Ultra(): # ProbeD (NP1121) matches its catalogue geometry - probeD = read_openephys( + probeD = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-NP-Ultra" / "settings.xml", probe_name="ProbeD", ) @@ -93,7 +95,7 @@ def test_probe_part_number_mismatch_with_catalogue(): "See https://github.com/SpikeInterface/probeinterface/issues/407 for details." ) with pytest.raises(ValueError, match=re.escape(expected_error)): - read_openephys( + read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-NP-Ultra" / "settings.xml", probe_name="ProbeA", ) @@ -101,7 +103,9 @@ def test_probe_part_number_mismatch_with_catalogue(): def test_NP1_subset(): # NP1 - 200 channels selected by recording_state in Record Node - probe_ap = read_openephys(data_path / "OE_Neuropix-PXI-subset" / "settings.xml", stream_name="ProbeA-AP") + probe_ap = read_openephys_neuropixels( + data_path / "OE_Neuropix-PXI-subset" / "settings.xml", stream_name="ProbeA-AP" + ) probe_dict = probe_ap.to_dict(array_as_list=True) validate_probe_dict(probe_dict) @@ -111,7 +115,9 @@ def test_NP1_subset(): assert "adc_group" in probe_ap.contact_annotations assert "adc_sample_order" in probe_ap.contact_annotations - probe_lf = read_openephys(data_path / "OE_Neuropix-PXI-subset" / "settings.xml", stream_name="ProbeA-LFP") + probe_lf = read_openephys_neuropixels( + data_path / "OE_Neuropix-PXI-subset" / "settings.xml", stream_name="ProbeA-LFP" + ) probe_dict = probe_lf.to_dict(array_as_list=True) validate_probe_dict(probe_dict) @@ -125,12 +131,12 @@ def test_NP1_subset(): # Not specifying the stream_name should raise an Exception, because both the ProbeA-AP and # ProbeA-LFP have custom channel selections with pytest.raises(AssertionError): - probe = read_openephys(data_path / "OE_Neuropix-PXI-subset" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI-subset" / "settings.xml") def test_multiple_probes(): # multiple probes - probeA = read_openephys(data_path / "OE_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="ProbeA") + probeA = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="ProbeA") probe_dict = probeA.to_dict(array_as_list=True) validate_probe_dict(probe_dict) @@ -139,7 +145,7 @@ def test_multiple_probes(): assert "adc_group" in probeA.contact_annotations assert "adc_sample_order" in probeA.contact_annotations - probeB = read_openephys( + probeB = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-multi-probe" / "settings.xml", stream_name="RecordNode#ProbeB", ) @@ -150,7 +156,7 @@ def test_multiple_probes(): assert "adc_group" in probeB.contact_annotations assert "adc_sample_order" in probeB.contact_annotations - probeC = read_openephys( + probeC = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-multi-probe" / "settings.xml", serial_number="20403311714", ) @@ -161,7 +167,7 @@ def test_multiple_probes(): assert "adc_group" in probeC.contact_annotations assert "adc_sample_order" in probeC.contact_annotations - probeD = read_openephys(data_path / "OE_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="ProbeD") + probeD = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="ProbeD") probe_dict = probeD.to_dict(array_as_list=True) validate_probe_dict(probe_dict) @@ -174,7 +180,7 @@ def test_multiple_probes(): assert probeC.serial_number == "20403311714" assert probeD.serial_number == "21144108671" - probeA2 = read_openephys( + probeA2 = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-multi-probe" / "settings_2.xml", probe_name="ProbeA", ) @@ -183,7 +189,7 @@ def test_multiple_probes(): ypos = probeA2.contact_positions[:, 1] assert np.min(ypos) >= 0 - probeB2 = read_openephys( + probeB2 = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-multi-probe" / "settings_2.xml", probe_name="ProbeB", ) @@ -197,7 +203,7 @@ def test_multiple_probes(): def test_multiple_probes_enabled(): # multiple probes, all enabled: - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_6.7_enabled_disabled_Neuropix-PXI" / "settings_enabled-enabled.xml", probe_name="ProbeA" ) probe_dict = probe.to_dict(array_as_list=True) @@ -207,7 +213,7 @@ def test_multiple_probes_enabled(): assert "adc_group" in probe.contact_annotations assert "adc_sample_order" in probe.contact_annotations - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_6.7_enabled_disabled_Neuropix-PXI" / "settings_enabled-enabled.xml", probe_name="ProbeB" ) probe_dict = probe.to_dict(array_as_list=True) @@ -219,7 +225,7 @@ def test_multiple_probes_enabled(): def test_multiple_probes_disabled(): # multiple probes, some disabled - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_6.7_enabled_disabled_Neuropix-PXI" / "settings_enabled-disabled.xml", probe_name="ProbeA" ) probe_dict = probe.to_dict(array_as_list=True) @@ -230,7 +236,7 @@ def test_multiple_probes_disabled(): # Fail as this is disabled: with pytest.raises(Exception) as e: - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_6.7_enabled_disabled_Neuropix-PXI" / "settings_enabled-disabled.xml", probe_name="ProbeB" ) @@ -238,7 +244,7 @@ def test_multiple_probes_disabled(): def test_np_opto_with_sync(): - probe = read_openephys(data_path / "OE_Neuropix-PXI-opto-with-sync" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI-opto-with-sync" / "settings.xml") probe_dict = probe.to_dict(array_as_list=True) validate_probe_dict(probe_dict) assert probe.get_shank_count() == 1 @@ -250,7 +256,7 @@ def test_np_opto_with_sync(): def test_older_than_06_format(): ## Test with the open ephys < 0.6 format - probe = read_openephys(data_path / "OE_5_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="100.0") + probe = read_openephys_neuropixels(data_path / "OE_5_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="100.0") probe_dict = probe.to_dict(array_as_list=True) validate_probe_dict(probe_dict) assert probe.get_shank_count() == 4 @@ -264,7 +270,7 @@ def test_older_than_06_format(): def test_multiple_signal_chains(): # tests that the probe information can be loaded even if the Neuropix-PXI plugin # is not in the first signalchain - probe = read_openephys(data_path / "OE_Neuropix-PXI-multiple-signalchains" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI-multiple-signalchains" / "settings.xml") probe_dict = probe.to_dict(array_as_list=True) validate_probe_dict(probe_dict) assert "adc_group" in probe.contact_annotations @@ -278,7 +284,7 @@ def test_quadbase_no_custom_name(): quadbase_probe_name = "ProbeC" for i in range(4): shank_offset = i * shank_pitch - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-QuadBase" / "settings.xml", probe_name=f"{quadbase_probe_name}-{i+1}" ) probe_dict = probe.to_dict(array_as_list=True) @@ -302,7 +308,7 @@ def test_quadbase_custom_name(): for probe_name in quadbase_custom_names: for i in range(4): shank_offset = i * shank_pitch - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-QuadBase" / "settings_custom_names.xml", probe_name=f"{probe_name}-{i+1}" ) probe_dict = probe.to_dict(array_as_list=True) @@ -325,7 +331,7 @@ def test_quadbase_custom_names(): shank_pitch = 250 for i in range(4): shank_offset = i * shank_pitch - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-QuadBase" / "settings_custom_names_w_shank.xml", probe_name=f"{sn}-{i+1}" ) probe_dict = probe.to_dict(array_as_list=True) @@ -344,7 +350,7 @@ def test_quadbase_custom_names(): def test_onebox(): # This dataset has a Neuropixels Ultra probe with a onebox - probe = read_openephys(data_path / "OE_OneBox-NP-Ultra" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_OneBox-NP-Ultra" / "settings.xml") probe_dict = probe.to_dict(array_as_list=True) validate_probe_dict(probe_dict) assert probe.get_shank_count() == 1 @@ -355,7 +361,7 @@ def test_onebox(): def test_onix_np1(): # This dataset has a multiple settings with different banks and configs - probe_bankA = read_openephys(data_path / "OE_ONIX-NP" / "settings_bankA.xml") + probe_bankA = read_openephys_neuropixels(data_path / "OE_ONIX-NP" / "settings_bankA.xml") probe_dict = probe_bankA.to_dict(array_as_list=True) validate_probe_dict(probe_dict) assert probe_bankA.get_shank_count() == 1 @@ -367,7 +373,7 @@ def test_onix_np1(): assert "adc_group" in probe_bankA.contact_annotations assert "adc_sample_order" in probe_bankA.contact_annotations - probe_bankB = read_openephys(data_path / "OE_ONIX-NP" / "settings_bankB.xml") + probe_bankB = read_openephys_neuropixels(data_path / "OE_ONIX-NP" / "settings_bankB.xml") probe_dict = probe_bankB.to_dict(array_as_list=True) validate_probe_dict(probe_dict) assert probe_bankB.get_shank_count() == 1 @@ -379,7 +385,7 @@ def test_onix_np1(): assert "adc_group" in probe_bankB.contact_annotations assert "adc_sample_order" in probe_bankB.contact_annotations - probe_bankC = read_openephys(data_path / "OE_ONIX-NP" / "settings_bankC.xml") + probe_bankC = read_openephys_neuropixels(data_path / "OE_ONIX-NP" / "settings_bankC.xml") probe_dict = probe_bankC.to_dict(array_as_list=True) validate_probe_dict(probe_dict) assert probe_bankC.get_shank_count() == 1 @@ -392,7 +398,7 @@ def test_onix_np1(): assert "adc_sample_order" in probe_bankC.contact_annotations # for the tetrode configuration, we expect to have 96 tetrodes - probe_tetrodes = read_openephys(data_path / "OE_ONIX-NP" / "settings_tetrodes.xml") + probe_tetrodes = read_openephys_neuropixels(data_path / "OE_ONIX-NP" / "settings_tetrodes.xml") probe_dict = probe_tetrodes.to_dict(array_as_list=True) assert "adc_group" in probe_tetrodes.contact_annotations @@ -419,7 +425,7 @@ def test_onix_np1(): def test_onix_np2(): # NP2.0 - probe_np2_probe0 = read_openephys( + probe_np2_probe0 = read_openephys_neuropixels( data_path / "OE_ONIX-NP" / "settings_NP2.xml", probe_name="PortA-Neuropixels2.0eHeadstage-Neuropixels2.0-Probe0" ) probe_dict = probe_np2_probe0.to_dict(array_as_list=True) @@ -429,7 +435,7 @@ def test_onix_np2(): assert "adc_group" in probe_np2_probe0.contact_annotations assert "adc_sample_order" in probe_np2_probe0.contact_annotations - probe_np2_probe1 = read_openephys( + probe_np2_probe1 = read_openephys_neuropixels( data_path / "OE_ONIX-NP" / "settings_NP2.xml", probe_name="PortA-Neuropixels2.0eHeadstage-Neuropixels2.0-Probe1" ) probe_dict = probe_np2_probe1.to_dict(array_as_list=True) @@ -440,13 +446,13 @@ def test_onix_np2(): assert "adc_sample_order" in probe_np2_probe1.contact_annotations for i in range(4): - probe_0 = read_openephys( + probe_0 = read_openephys_neuropixels( data_path / "OE_ONIX-NP" / f"settings_NP2_{i+1}.xml", probe_name=f"PortA-Neuropixels2.0eHeadstage-Neuropixels2.0-Probe0", ) probe_dict = probe_0.to_dict(array_as_list=True) validate_probe_dict(probe_dict) - probe_1 = read_openephys( + probe_1 = read_openephys_neuropixels( data_path / "OE_ONIX-NP" / f"settings_NP2_{i+1}.xml", probe_name=f"PortA-Neuropixels2.0eHeadstage-Neuropixels2.0-Probe1", ) @@ -492,39 +498,41 @@ def test_read_openephys_contact_ids_match_canonical_pattern(): (see https://github.com/SpikeInterface/probeinterface/pull/383#discussion_r2650588006). """ # Path A (SELECTED_ELECTRODES): OE 1.0 dataset - probe = read_openephys(data_path / "OE_1.0_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="ProbeA") + probe = read_openephys_neuropixels( + data_path / "OE_1.0_Neuropix-PXI-multi-probe" / "settings.xml", probe_name="ProbeA" + ) _assert_contact_ids_match_canonical_pattern(probe, "OE_1.0 ProbeA") # Path B (CHANNELS): NP2 dataset (single shank) - probe = read_openephys(data_path / "OE_Neuropix-PXI" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI" / "settings.xml") _assert_contact_ids_match_canonical_pattern(probe, "NP2") # Path B (CHANNELS): NP2 4-shank dataset (multi-shank) - probe = read_openephys(data_path / "OE_Neuropix-PXI-NP2-4shank" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI-NP2-4shank" / "settings.xml") _assert_contact_ids_match_canonical_pattern(probe, "NP2 4-shank") # Path B (CHANNELS): NP-Opto dataset - probe = read_openephys(data_path / "OE_Neuropix-PXI-opto-with-sync" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI-opto-with-sync" / "settings.xml") _assert_contact_ids_match_canonical_pattern(probe, "NP-Opto") # Path B (CHANNELS): OneBox NP-Ultra (NP1110) dataset - probe = read_openephys(data_path / "OE_OneBox-NP-Ultra" / "settings.xml") + probe = read_openephys_neuropixels(data_path / "OE_OneBox-NP-Ultra" / "settings.xml") _assert_contact_ids_match_canonical_pattern(probe, "OneBox NP1110") # Datasets identified as inconsistent in PR #383 discussion: # NP-Ultra: NP1100 probes error due to catalogue mismatch (see issue #407), NP1121 should match - probe = read_openephys(data_path / "OE_Neuropix-PXI-NP-Ultra" / "settings.xml", probe_name="ProbeD") + probe = read_openephys_neuropixels(data_path / "OE_Neuropix-PXI-NP-Ultra" / "settings.xml", probe_name="ProbeD") _assert_contact_ids_match_canonical_pattern(probe, "NP-Ultra ProbeD") # enabled/disabled: NP1 and NP2014 - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_6.7_enabled_disabled_Neuropix-PXI" / "settings_enabled-enabled.xml", probe_name="ProbeA", ) _assert_contact_ids_match_canonical_pattern(probe, "enabled-enabled ProbeA") - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_6.7_enabled_disabled_Neuropix-PXI" / "settings_enabled-enabled.xml", probe_name="ProbeB", ) @@ -532,7 +540,9 @@ def test_read_openephys_contact_ids_match_canonical_pattern(): # QuadBase: NP2020 (4 probes) for i in range(4): - probe = read_openephys(data_path / "OE_Neuropix-PXI-QuadBase" / "settings.xml", probe_name=f"ProbeC-{i+1}") + probe = read_openephys_neuropixels( + data_path / "OE_Neuropix-PXI-QuadBase" / "settings.xml", probe_name=f"ProbeC-{i+1}" + ) _assert_contact_ids_match_canonical_pattern(probe, f"QuadBase ProbeC-{i+1}") @@ -561,7 +571,7 @@ def test_read_openephys_against_oebin_wiring(): ) stream_name = "Neuropix-PXI-100.ProbeA" - probe = read_openephys(settings, stream_name=stream_name) + probe = read_openephys_neuropixels(settings, stream_name=stream_name) assert probe.get_contact_count() == 384 assert probe.device_channel_indices is not None @@ -582,14 +592,14 @@ def test_read_openephys_against_oebin_wiring(): def test_read_openephys_with_oebin_contact_ids_match_canonical_pattern(): """Verify that contact_ids with oebin are consistent with SpikeGLX (issue #394).""" # NP2014 single-shank - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-NP1-binary" / "Record_Node_101" / "settings.xml", stream_name="Neuropix-PXI-100.ProbeA", ) _assert_contact_ids_match_canonical_pattern(probe, "NP2014 binary") # NP1032 4-shank - probe = read_openephys( + probe = read_openephys_neuropixels( data_path / "OE_Neuropix-PXI-NP2-4shank-binary" / "Record_Node_101" / "settings.xml", stream_name="Neuropix-PXI-100.ProbeA-AP", ) @@ -600,7 +610,7 @@ def test_read_openephys_with_oebin_sync_channel_filtered(): """Verify that the oebin sync channel (385 channels) is filtered, producing 384 contacts.""" settings = data_path / "OE_Neuropix-PXI-NP2-4shank-binary" / "Record_Node_101" / "settings.xml" - probe = read_openephys(settings, stream_name="Neuropix-PXI-100.ProbeA-AP") + probe = read_openephys_neuropixels(settings, stream_name="Neuropix-PXI-100.ProbeA-AP") assert probe.get_contact_count() == 384 assert "adc_group" in probe.contact_annotations assert "adc_sample_order" in probe.contact_annotations @@ -611,7 +621,7 @@ def test_read_openephys_with_oebin_settings_channel_key(): settings = data_path / "OE_Neuropix-PXI-NP1-binary" / "Record_Node_101" / "settings.xml" stream_name = "Neuropix-PXI-100.ProbeA" - probe = read_openephys(settings, stream_name=stream_name) + probe = read_openephys_neuropixels(settings, stream_name=stream_name) keys = probe.contact_annotations.get("settings_channel_key", None) assert keys is not None, "settings_channel_key annotation not set" assert len(keys) == probe.get_contact_count() @@ -639,7 +649,7 @@ def test_read_openephys_multishank_wiring(): ) stream_name = "Neuropix-PXI-103.ProbeA" - probe = read_openephys(settings, stream_name=stream_name) + probe = read_openephys_neuropixels(settings, stream_name=stream_name) assert probe.get_contact_count() == 384 assert probe.device_channel_indices is not None @@ -678,7 +688,7 @@ def test_read_openephys_onebox_nonsequential_wiring(): ) stream_name = "OneBox-111.ProbeA" - probe = read_openephys(settings, stream_name=stream_name) + probe = read_openephys_neuropixels(settings, stream_name=stream_name) assert probe.get_contact_count() == 384 assert probe.device_channel_indices is not None @@ -707,6 +717,26 @@ def test_read_openephys_onebox_nonsequential_wiring(): ) +def test_read_openephys_deprecation_warning(): + # Old read_openephys name must still work but emit DeprecationWarning pointing at the new name. + settings = data_path / "OE_Neuropix-PXI" / "settings.xml" + with pytest.warns(DeprecationWarning, match="read_openephys_neuropixels"): + read_openephys(settings) + + +def test_has_neuropixels_probes_positive(): + # A real Neuropixels settings.xml should report True. + settings = data_path / "OE_Neuropix-PXI" / "settings.xml" + assert has_neuropixels_probes(settings) is True + + +def test_has_neuropixels_probes_stream_match(): + # Stream-name filter: matching stream returns True, non-matching returns False. + settings = data_path / "OE_Neuropix-PXI-subset" / "settings.xml" + assert has_neuropixels_probes(settings, stream_name="ProbeA-AP") is True + assert has_neuropixels_probes(settings, stream_name="Rhythm_FPGA-100.0") is False + + if __name__ == "__main__": # test_multiple_probes() # test_NP_Ultra()