From 81f37f10564ef04e666e72a57a41c4e17fb329d6 Mon Sep 17 00:00:00 2001 From: Roberto <37729096+RobertoDF@users.noreply.github.com> Date: Wed, 20 May 2026 12:09:42 +0200 Subject: [PATCH 1/8] Generalize to fit also NPX2. Use only data block. --- src/probeinterface/io.py | 129 ++++++++++++++------------------------- 1 file changed, 46 insertions(+), 83 deletions(-) diff --git a/src/probeinterface/io.py b/src/probeinterface/io.py index 286c5787..84fb7ddc 100644 --- a/src/probeinterface/io.py +++ b/src/probeinterface/io.py @@ -734,101 +734,64 @@ def write_csv(file, probe): def read_spikegadgets(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. - - - 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" + PROBE_SPECS = { + "NeuroPixels1": {"width": 12, "height": 12}, + "NeuroPixels2": {"width": 12, "height": 12} + } 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 + probe_devices = [d for d in hconf if d.attrib.get("name") in PROBE_SPECS.keys()] - # 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"] + # Group by probe ID + probes_dict = {} + for ntrode in sconf: + ch_info = ntrode.find("SpikeChannel") + if ch_info is None: continue + probe_id = int(ntrode.attrib["id"][0]) + if probe_id not in probes_dict: + probes_dict[probe_id] = [] + probes_dict[probe_id].append((ntrode.attrib["id"], ch_info)) + + # --- Print Summary Information --- + print(f"Number of probes detected: {len(probes_dict)}") + for pid in sorted(probes_dict.keys()): + print(f"Probe {pid}: {len(probes_dict[pid])} active channels") + print(f"Total active channels: {sum(len(ch) for ch in probes_dict.values())}") probe_group = ProbeGroup() - for curr_probe in range(1, n_probes + 1): - # SpikeNTrode elements are the authoritative list of recorded electrodes. - # Each id is "<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) + for probe_id in sorted(probes_dict.keys()): + data = probes_dict[probe_id] + + device_name = probe_devices[probe_id - 1].attrib["name"] + print(f"Processing Probe {probe_id} identified as: {device_name}") + + if device_name not in PROBE_SPECS: + raise ValueError(f"Unknown probe type: {device_name}.") - # 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]) + specs = PROBE_SPECS[device_name] + + contact_ids = [item[0] for item in data] + device_channels = [item[1].attrib["hwChan"] for item in data] + + positions = np.zeros((len(data), 2)) + for i, (_, ch_info) in enumerate(data): + positions[i, 0] = float(ch_info.attrib["coord_ml"]) + positions[i, 1] = float(ch_info.attrib["coord_dv"]) + + probe = Probe(ndim=2, si_units="um", model_name=device_name, manufacturer="IMEC") + probe.set_contacts( + contact_ids=contact_ids, positions=positions, shapes="square", + shape_params={"width": specs["width"], "height": specs["height"]}, + ) + probe.set_device_channel_indices(device_channels) + # + probe.move([250 * (probe_id - 1), 0]) probe_group.add_probe(probe) return probe_group From b1a890c5ee72123e0d93d9dabef9143e09bad282 Mon Sep 17 00:00:00 2001 From: Roberto <37729096+RobertoDF@users.noreply.github.com> Date: Wed, 20 May 2026 16:42:30 +0200 Subject: [PATCH 2/8] update according to new catalogue pattern --- src/probeinterface/io.py | 153 +++++++++++++++++++++++++++------------ 1 file changed, 105 insertions(+), 48 deletions(-) diff --git a/src/probeinterface/io.py b/src/probeinterface/io.py index 84fb7ddc..f333d91e 100644 --- a/src/probeinterface/io.py +++ b/src/probeinterface/io.py @@ -734,64 +734,121 @@ def write_csv(file, probe): def read_spikegadgets(file: str | Path, raise_error: bool = True) -> ProbeGroup: - PROBE_SPECS = { - "NeuroPixels1": {"width": 12, "height": 12}, - "NeuroPixels2": {"width": 12, "height": 12} - } + """ + 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. + + + 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. header_txt = parse_spikegadgets_header(file) root = ElementTree.fromstring(header_txt) hconf = root.find("HardwareConfiguration") sconf = root.find("SpikeConfiguration") - probe_devices = [d for d in hconf if d.attrib.get("name") in PROBE_SPECS.keys()] + # Detect devices present in the header + probe_configs = [d for d in hconf if d.attrib.get("name") in ["NeuroPixels1", "NeuroPixels2"]] + n_probes = len(probe_configs) - # Group by probe ID - probes_dict = {} - for ntrode in sconf: - ch_info = ntrode.find("SpikeChannel") - if ch_info is None: continue - probe_id = int(ntrode.attrib["id"][0]) - if probe_id not in probes_dict: - probes_dict[probe_id] = [] - probes_dict[probe_id].append((ntrode.attrib["id"], ch_info)) - - # --- Print Summary Information --- - print(f"Number of probes detected: {len(probes_dict)}") - for pid in sorted(probes_dict.keys()): - print(f"Probe {pid}: {len(probes_dict[pid])} active channels") - print(f"Total active channels: {sum(len(ch) for ch in probes_dict.values())}") + if n_probes == 0: + if raise_error: + raise Exception("No supported Neuropixels probes found") probe_group = ProbeGroup() - for probe_id in sorted(probes_dict.keys()): - data = probes_dict[probe_id] - - device_name = probe_devices[probe_id - 1].attrib["name"] - print(f"Processing Probe {probe_id} identified as: {device_name}") - - if device_name not in PROBE_SPECS: - raise ValueError(f"Unknown probe type: {device_name}.") - - specs = PROBE_SPECS[device_name] - - contact_ids = [item[0] for item in data] - device_channels = [item[1].attrib["hwChan"] for item in data] - - positions = np.zeros((len(data), 2)) - for i, (_, ch_info) in enumerate(data): - positions[i, 0] = float(ch_info.attrib["coord_ml"]) - positions[i, 1] = float(ch_info.attrib["coord_dv"]) - - probe = Probe(ndim=2, si_units="um", model_name=device_name, manufacturer="IMEC") - probe.set_contacts( - contact_ids=contact_ids, positions=positions, shapes="square", - shape_params={"width": specs["width"], "height": specs["height"]}, - ) - probe.set_device_channel_indices(device_channels) - - # - probe.move([250 * (probe_id - 1), 0]) + for curr_probe_idx, probe_config in enumerate(probe_configs): + + device_name = probe_config.attrib["name"] + + # 1. Collect all used probeColumns for this probe index, this is needed to understand how many shanks are present + curr_probe = curr_probe_idx + 1 + used_columns = set() + for ntrode in sconf: + if int(ntrode.attrib["id"][0]) == curr_probe: + # Assuming SpikeChannel follows the structure where probeColumn is defined + for channel in ntrode.findall("SpikeChannel"): + used_columns.add(int(channel.attrib["probeColumn"])) + + # 2. Determine part number based on shank count + if device_name == "NeuroPixels1": + part_number = "NP1000" + elif device_name == "NeuroPixels2": + # NP2.0: 1 shank = columns 0-1; 4 shanks = columns 0-7 + num_shanks = 4 if max(used_columns) == 7 else 1 + part_number = "NP2000" if num_shanks == 1 else "NP2010" + + channel_data = [] + for ntrode in sconf: + electrode_id = ntrode.attrib["id"] + if int(ntrode.attrib["id"][0]) == curr_probe: + chan_data = ntrode[0].attrib + channel_data.append({ + "hw": int(chan_data["hwChan"]), + "col": int(chan_data["probeColumn"]), + "ap": int(chan_data["coord_ap"]), + "ml": int(chan_data["coord_ml"]), + "dv": int(chan_data["coord_dv"]), + "probe_n": int(electrode_id[0]), + "channel": int(electrode_id[1:]) + }) + + # 2. Extract indices + + device_channels = np.array([c["channel"] for c in channel_data]) + active_channels = np.array([c["hw"] for c in channel_data]) + + full_probe = build_neuropixels_probe(part_number) + + full_probe_df = pd.concat([pd.DataFrame(full_probe.contact_positions, columns=["ml", "dv"]), + pd.Series(full_probe.contact_ids, name="contact_ids")], axis=1).sort_values( + ["dv"]).reset_index(drop=True) + + device_channels = pd.DataFrame(full_probe.contact_positions, columns=["ml", "dv"]).sort_values(["dv"]).iloc[ + device_channels].index # the ids in trodes are assigned according to a dv sorted probe + + probe = full_probe.get_slice(device_channels) + probe.set_device_channel_indices(active_channels) + + probe.model_name = "" + probe.description = "" + + # Annotate ADC info + adc_sampling_table = probe.annotations.get("adc_sampling_table") + if adc_sampling_table is not None: + _annotate_probe_with_adc_sampling_info(probe, adc_sampling_table) + + # Handle gain settings dynamically + source_options_blocks = [s for s in hconf.findall("SourceOptions") if s.attrib.get("name") == device_name] + if curr_probe_idx < len(source_options_blocks): + custom_options = { + opt.attrib["name"]: opt.attrib["data"].strip() + for opt in source_options_blocks[curr_probe_idx].findall("CustomOption") + } + if "APGainMode" in custom_options: + probe.annotate(ap_gain=float(custom_options["APGainMode"])) + if probe.annotations.get("lf_sample_frequency_hz", 0) > 0 and "LFPGainMode" in custom_options: + probe.annotate(lf_gain=float(custom_options["LFPGainMode"])) + + # Spatial shift for multiple probes + probe.move([250 * curr_probe_idx, 0]) probe_group.add_probe(probe) return probe_group From 409047e7d122cf05f59908d71760b927edc396d6 Mon Sep 17 00:00:00 2001 From: Roberto <37729096+RobertoDF@users.noreply.github.com> Date: Wed, 20 May 2026 16:55:20 +0200 Subject: [PATCH 3/8] everything in numpy Refactor probe channel sorting logic for clarity and efficiency. --- src/probeinterface/io.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/src/probeinterface/io.py b/src/probeinterface/io.py index f333d91e..6d25e495 100644 --- a/src/probeinterface/io.py +++ b/src/probeinterface/io.py @@ -817,12 +817,11 @@ def read_spikegadgets(file: str | Path, raise_error: bool = True) -> ProbeGroup: full_probe = build_neuropixels_probe(part_number) - full_probe_df = pd.concat([pd.DataFrame(full_probe.contact_positions, columns=["ml", "dv"]), - pd.Series(full_probe.contact_ids, name="contact_ids")], axis=1).sort_values( - ["dv"]).reset_index(drop=True) - - device_channels = pd.DataFrame(full_probe.contact_positions, columns=["ml", "dv"]).sort_values(["dv"]).iloc[ - device_channels].index # the ids in trodes are assigned according to a dv sorted probe + contact_positions = full_probe.contact_positions # shape (n_contacts, 2) + dv_col = contact_positions[:, 1] # assume dv is the second column + sorted_order = np.argsort(dv_col) # indices that sort by dv ascending + device_channels = sorted_order[ + device_channels] # the ids in trodes are assigned according to a dv sorted probe, we have to check which ml direction though! probe = full_probe.get_slice(device_channels) probe.set_device_channel_indices(active_channels) From 6094c812d08f9a448ad2ca1349cbc7175e83f98b Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Wed, 20 May 2026 15:29:17 +0000 Subject: [PATCH 4/8] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/probeinterface/io.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/src/probeinterface/io.py b/src/probeinterface/io.py index 6d25e495..cd5fb90f 100644 --- a/src/probeinterface/io.py +++ b/src/probeinterface/io.py @@ -800,15 +800,17 @@ def read_spikegadgets(file: str | Path, raise_error: bool = True) -> ProbeGroup: electrode_id = ntrode.attrib["id"] if int(ntrode.attrib["id"][0]) == curr_probe: chan_data = ntrode[0].attrib - channel_data.append({ - "hw": int(chan_data["hwChan"]), - "col": int(chan_data["probeColumn"]), - "ap": int(chan_data["coord_ap"]), - "ml": int(chan_data["coord_ml"]), - "dv": int(chan_data["coord_dv"]), - "probe_n": int(electrode_id[0]), - "channel": int(electrode_id[1:]) - }) + channel_data.append( + { + "hw": int(chan_data["hwChan"]), + "col": int(chan_data["probeColumn"]), + "ap": int(chan_data["coord_ap"]), + "ml": int(chan_data["coord_ml"]), + "dv": int(chan_data["coord_dv"]), + "probe_n": int(electrode_id[0]), + "channel": int(electrode_id[1:]), + } + ) # 2. Extract indices @@ -821,7 +823,8 @@ def read_spikegadgets(file: str | Path, raise_error: bool = True) -> ProbeGroup: dv_col = contact_positions[:, 1] # assume dv is the second column sorted_order = np.argsort(dv_col) # indices that sort by dv ascending device_channels = sorted_order[ - device_channels] # the ids in trodes are assigned according to a dv sorted probe, we have to check which ml direction though! + device_channels + ] # the ids in trodes are assigned according to a dv sorted probe, we have to check which ml direction though! probe = full_probe.get_slice(device_channels) probe.set_device_channel_indices(active_channels) From 063557466bad122f33dc2340d3fbaa1d4e843a48 Mon Sep 17 00:00:00 2001 From: Roberto <37729096+RobertoDF@users.noreply.github.com> Date: Wed, 20 May 2026 17:36:48 +0200 Subject: [PATCH 5/8] add "return None" after checking probe detected --- src/probeinterface/io.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/probeinterface/io.py b/src/probeinterface/io.py index cd5fb90f..f49d13c4 100644 --- a/src/probeinterface/io.py +++ b/src/probeinterface/io.py @@ -771,7 +771,8 @@ def read_spikegadgets(file: str | Path, raise_error: bool = True) -> ProbeGroup: if n_probes == 0: if raise_error: raise Exception("No supported Neuropixels probes found") - + return None + probe_group = ProbeGroup() for curr_probe_idx, probe_config in enumerate(probe_configs): From 3f330e3d4d1ab7c0bdf3541474404d5c4f7f9b9e Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Wed, 20 May 2026 15:36:58 +0000 Subject: [PATCH 6/8] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/probeinterface/io.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/probeinterface/io.py b/src/probeinterface/io.py index f49d13c4..df8b4c56 100644 --- a/src/probeinterface/io.py +++ b/src/probeinterface/io.py @@ -772,7 +772,7 @@ def read_spikegadgets(file: str | Path, raise_error: bool = True) -> ProbeGroup: if raise_error: raise Exception("No supported Neuropixels probes found") return None - + probe_group = ProbeGroup() for curr_probe_idx, probe_config in enumerate(probe_configs): From a1e3d3441eb196cff99b59bc7958d8bbebfe82e1 Mon Sep 17 00:00:00 2001 From: Roberto <37729096+RobertoDF@users.noreply.github.com> Date: Wed, 20 May 2026 20:02:38 +0200 Subject: [PATCH 7/8] fix for ids starting at 1, and ml order inverted --- src/probeinterface/io.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/src/probeinterface/io.py b/src/probeinterface/io.py index df8b4c56..f1cd47fd 100644 --- a/src/probeinterface/io.py +++ b/src/probeinterface/io.py @@ -809,25 +809,26 @@ def read_spikegadgets(file: str | Path, raise_error: bool = True) -> ProbeGroup: "ml": int(chan_data["coord_ml"]), "dv": int(chan_data["coord_dv"]), "probe_n": int(electrode_id[0]), - "channel": int(electrode_id[1:]), + "channel": int(electrode_id[1:])-1 # trodes channels start at 1 not 0 } ) # 2. Extract indices - device_channels = np.array([c["channel"] for c in channel_data]) + device_channels = np.array([c["channel"] for c in channel_data]) - 1 # channel ids start at 1 active_channels = np.array([c["hw"] for c in channel_data]) full_probe = build_neuropixels_probe(part_number) contact_positions = full_probe.contact_positions # shape (n_contacts, 2) - dv_col = contact_positions[:, 1] # assume dv is the second column - sorted_order = np.argsort(dv_col) # indices that sort by dv ascending - device_channels = sorted_order[ - device_channels - ] # the ids in trodes are assigned according to a dv sorted probe, we have to check which ml direction though! - - probe = full_probe.get_slice(device_channels) + ml = contact_positions[:, 0] + dv = contact_positions[:, 1] + + sorted_order = np.lexsort((-ml, dv)) + device_channels_indexes = sorted_order[ + device_channels] # the ids in trodes are assigned according to whats seen in trodes, id 0 is tip of the probe (min dv) and max ml. + + probe = full_probe.get_slice(device_channels_indexes) probe.set_device_channel_indices(active_channels) probe.model_name = "" From 63a355e89c27b14241197f9337d790337de4415a Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Wed, 20 May 2026 18:02:50 +0000 Subject: [PATCH 8/8] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/probeinterface/io.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/probeinterface/io.py b/src/probeinterface/io.py index f1cd47fd..22a2a298 100644 --- a/src/probeinterface/io.py +++ b/src/probeinterface/io.py @@ -809,25 +809,26 @@ def read_spikegadgets(file: str | Path, raise_error: bool = True) -> ProbeGroup: "ml": int(chan_data["coord_ml"]), "dv": int(chan_data["coord_dv"]), "probe_n": int(electrode_id[0]), - "channel": int(electrode_id[1:])-1 # trodes channels start at 1 not 0 + "channel": int(electrode_id[1:]) - 1, # trodes channels start at 1 not 0 } ) # 2. Extract indices - device_channels = np.array([c["channel"] for c in channel_data]) - 1 # channel ids start at 1 + device_channels = np.array([c["channel"] for c in channel_data]) - 1 # channel ids start at 1 active_channels = np.array([c["hw"] for c in channel_data]) full_probe = build_neuropixels_probe(part_number) contact_positions = full_probe.contact_positions # shape (n_contacts, 2) - ml = contact_positions[:, 0] - dv = contact_positions[:, 1] - - sorted_order = np.lexsort((-ml, dv)) + ml = contact_positions[:, 0] + dv = contact_positions[:, 1] + + sorted_order = np.lexsort((-ml, dv)) device_channels_indexes = sorted_order[ - device_channels] # the ids in trodes are assigned according to whats seen in trodes, id 0 is tip of the probe (min dv) and max ml. - + device_channels + ] # the ids in trodes are assigned according to whats seen in trodes, id 0 is tip of the probe (min dv) and max ml. + probe = full_probe.get_slice(device_channels_indexes) probe.set_device_channel_indices(active_channels)