Add Neuropixels 2.0 support to read_spikegadgets_neuropixels (#441)

Author: h-mayorquin

src/probeinterface/__init__.py

@@ -14,9 +14,6 @@
     write_csv,
     read_BIDS_probe,
     write_BIDS_probe,
-    read_spikegadgets,
-    read_spikegadgets_neuropixels,
-    has_spikegadgets_neuropixels_probes,
     read_mearec,
     read_nwb,
     read_maxwell,
@@ -33,6 +30,10 @@
     read_openephys_neuropixels,
     has_neuropixels_probes,
     get_saved_channel_indices_from_openephys_settings,
+    read_spikegadgets,
+    read_spikegadgets_neuropixels,
+    has_spikegadgets_neuropixels_probes,
+    parse_spikegadgets_header,
 )
 from .utils import combine_probes
 from .generator import (

src/probeinterface/io.py

@@ -11,17 +11,13 @@
 
 from pathlib import Path
 import re
-import warnings
 import json
 from collections import OrderedDict
 from packaging.version import parse
 import numpy as np
-from xml.etree import ElementTree
-
 from . import __version__
 from .probe import Probe
 from .probegroup import ProbeGroup
-from .neuropixels_tools import build_neuropixels_probe, _annotate_probe_with_adc_sampling_info
 from .utils import import_safely
 
 
@@ -733,193 +729,6 @@ def write_csv(file, probe):
     raise NotImplementedError
 
 
-def read_spikegadgets_neuropixels(file: str | Path, raise_error: bool = True) -> ProbeGroup:
-    """
-    Find active channels of the given Neuropixels probe from a SpikeGadgets .rec file.
-    SpikeGadgets headstages support up to three Neuropixels 1.0 probes (as of March 28, 2024),
-    and information for all probes will be returned in a ProbeGroup object.
-
-    This function only supports Neuropixels probes recorded with SpikeGadgets
-    headstages (``HardwareConfiguration`` entries with ``name == "NeuroPixels1"``).
-    It does not handle tetrodes or other probe types that SpikeGadgets can
-    record. Use :func:`has_spikegadgets_neuropixels_probes` to check whether a
-    ``.rec`` file contains Neuropixels probe geometry before calling this reader.
-
-    Parameters
-    ----------
-    file : Path or str
-        The .rec file path
-
-    Returns
-    -------
-    probe_group : ProbeGroup object
-
-    """
-    # The SpikeGadgets .rec XML does not include a probe part number. The NP1.0
-    # catalogue variants (NP1000, NP1001, PRB_1_2_0480_2, PRB_1_4_0480_1,
-    # PRB_1_4_0480_1_C) share identical 2D geometry in the probeinterface
-    # catalogue (contact positions, pitch, stagger, shank width), differing only
-    # in metadata that probeinterface does not consume (ADC resolution, databus
-    # phase, gain, on-shank reference, shank thickness). So hardcoding NP1000
-    # produces correct geometry; `model_name` and `description` are cleared on
-    # the sliced probe to avoid claiming a specific variant.
-    PART_NUMBER = "NP1000"
-
-    header_txt = parse_spikegadgets_header(file)
-    root = ElementTree.fromstring(header_txt)
-    hconf = root.find("HardwareConfiguration")
-    sconf = root.find("SpikeConfiguration")
-
-    probe_configs = [d for d in hconf if d.attrib.get("name") == "NeuroPixels1"]
-    n_probes = len(probe_configs)
-
-    if n_probes == 0:
-        if raise_error:
-            raise Exception("No Neuropixels 1.0 probes found")
-        return None
-
-    # NeuroPixels1 SourceOptions blocks carry the per-probe AP/LF gain settings.
-    # They appear in the same order as the SpikeNTrode probe digits (1, 2, 3).
-    source_options_blocks = [s for s in hconf.findall("SourceOptions") if s.attrib.get("name") == "NeuroPixels1"]
-
-    probe_group = ProbeGroup()
-
-    for curr_probe in range(1, n_probes + 1):
-        # SpikeNTrode elements are the authoritative list of recorded electrodes.
-        # Each id is "<probe_digit><1-based electrode number>" for up to 960
-        # electrodes on NP1.0; the catalogue uses 0-based indices, so
-        # catalogue_index = electrode_number - 1. The probe number is assumed
-        # to be a single digit (1, 2, or 3), matching the documented
-        # SpikeGadgets limit of three simultaneous Neuropixels probes.
-        electrode_to_hwchan = {}
-        for ntrode in sconf:
-            electrode_id = ntrode.attrib["id"]
-            if int(electrode_id[0]) == curr_probe:
-                catalogue_index = int(electrode_id[1:]) - 1
-                hw_chan = int(ntrode[0].attrib["hwChan"])
-                electrode_to_hwchan[catalogue_index] = hw_chan
-
-        active_indices = np.array(sorted(electrode_to_hwchan.keys()))
-
-        full_probe = build_neuropixels_probe(PART_NUMBER)
-        probe = full_probe.get_slice(active_indices)
-
-        # Clear part-number-specific metadata since we don't know the actual part number.
-        probe.model_name = ""
-        probe.description = ""
-
-        device_channels = np.array([electrode_to_hwchan[idx] for idx in active_indices])
-        probe.set_device_channel_indices(device_channels)
-
-        # Per-contact ADC group and sample order from the catalogue MUX table plus
-        # the hwChan mapping (which is the readout-channel index for each contact).
-        adc_sampling_table = probe.annotations.get("adc_sampling_table")
-        _annotate_probe_with_adc_sampling_info(probe, adc_sampling_table)
-
-        # NP1.0 gain is programmable. Read APGainMode and LFPGainMode from the
-        # SourceOptions block matching this probe (blocks appear in probe order).
-        if "ap_gain" not in probe.annotations and curr_probe - 1 < len(source_options_blocks):
-            custom_options = {
-                opt.attrib["name"]: opt.attrib["data"].strip()
-                for opt in source_options_blocks[curr_probe - 1].findall("CustomOption")
-            }
-            ap_gain_str = custom_options.get("APGainMode")
-            if ap_gain_str:
-                probe.annotate(ap_gain=float(ap_gain_str))
-            if probe.annotations.get("lf_sample_frequency_hz", 0) > 0:
-                lf_gain_str = custom_options.get("LFPGainMode")
-                if lf_gain_str:
-                    probe.annotate(lf_gain=float(lf_gain_str))
-
-        # Shift multiple probes so they don't overlap when plotted
-        probe.move([250 * (curr_probe - 1), 0])
-
-        probe_group.add_probe(probe)
-
-    return probe_group
-
-
-def read_spikegadgets(*args, **kwargs) -> ProbeGroup:
-    """
-    Deprecated alias for :func:`read_spikegadgets_neuropixels`.
-
-    The name ``read_spikegadgets`` is misleading because the function only reads
-    Neuropixels probe geometry, not arbitrary SpikeGadgets ``.rec`` recordings.
-    Use :func:`read_spikegadgets_neuropixels` instead, and
-    :func:`has_spikegadgets_neuropixels_probes` to check whether a ``.rec`` file
-    has Neuropixels geometry before calling it.
-    """
-    warnings.warn(
-        "read_spikegadgets is deprecated and will be removed in a future release. "
-        "Use read_spikegadgets_neuropixels instead.",
-        category=DeprecationWarning,
-        stacklevel=2,
-    )
-    return read_spikegadgets_neuropixels(*args, **kwargs)
-
-
-def has_spikegadgets_neuropixels_probes(file: str | Path) -> bool:
-    """
-    Return True if the SpikeGadgets ``.rec`` file describes at least one
-    Neuropixels probe.
-
-    Detection scans the ``HardwareConfiguration`` block of the ``.rec`` XML
-    header for ``Device`` entries whose ``name`` attribute matches a known
-    Neuropixels source name (currently ``"NeuroPixels1"``). The presence of
-    any such entry is the ground-truth signal that the file contains
-    Neuropixels probe geometry, independent of what other hardware the
-    headstage is also streaming.
-
-    Intended use: callers that route heterogeneous SpikeGadgets recordings
-    (mixing tetrodes, Neuropixels, etc.) can gate the call to
-    :func:`read_spikegadgets_neuropixels` on this helper and skip probe
-    attachment for non-Neuropixels recordings.
-
-    Parameters
-    ----------
-    file : str or Path
-        Path to the SpikeGadgets ``.rec`` file.
-
-    Returns
-    -------
-    bool
-    """
-    try:
-        header_txt = parse_spikegadgets_header(file)
-        root = ElementTree.fromstring(header_txt)
-    except Exception:
-        return False
-
-    hconf = root.find("HardwareConfiguration")
-    if hconf is None:
-        return False
-
-    for device in hconf:
-        if device.attrib.get("name") == "NeuroPixels1":
-            return True
-    return False
-
-
-def parse_spikegadgets_header(file: str | Path) -> str:
-    """
-    Parse file (SpikeGadgets .rec format) into a string until "</Configuration>",
-    which is the last tag of the header, after which the binary data begins.
-    """
-    header_size = None
-    with open(file, mode="rb") as f:
-        while True:
-            line = f.readline()
-            if b"</Configuration>" in line:
-                header_size = f.tell()
-                break
-
-        if header_size is None:
-            ValueError("SpikeGadgets: the xml header does not contain '</Configuration>'")
-
-        f.seek(0)
-        return f.read(header_size).decode("utf8")
-
-
 def read_mearec(file: str | Path) -> Probe:
     """
     Read probe position, and contact shape from a MEArec file.