Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions src/probeinterface/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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 (
Expand Down
191 changes: 0 additions & 191 deletions src/probeinterface/io.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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.
Expand Down
Loading
Loading