From 30cf6014bdd28a4b1bd291656a9b0345f1ddb7b4 Mon Sep 17 00:00:00 2001 From: chrishalcrow Date: Mon, 1 Sep 2025 12:25:05 +0100 Subject: [PATCH 1/5] changes needed for onix --- src/probeinterface/neuropixels_tools.py | 56 ++++++++++++++++--------- 1 file changed, 37 insertions(+), 19 deletions(-) diff --git a/src/probeinterface/neuropixels_tools.py b/src/probeinterface/neuropixels_tools.py index b2770b5..9639b00 100644 --- a/src/probeinterface/neuropixels_tools.py +++ b/src/probeinterface/neuropixels_tools.py @@ -744,6 +744,7 @@ def read_openephys( oe_version = parse(info_chain.find("VERSION").text) neuropix_pxi_processor = None onebox_processor = None + onix_processor = None for signal_chain in root.findall("SIGNALCHAIN"): for processor in signal_chain: if "PROCESSOR" == processor.tag: @@ -752,8 +753,10 @@ def read_openephys( neuropix_pxi_processor = processor if "OneBox" in name: onebox_processor = processor + if "ONIX" in name: + onix_processor = processor - if neuropix_pxi_processor is None and onebox_processor is None: + 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 when the Neuropix-PXI or the " "OneBox plugin is used.") return None @@ -769,6 +772,8 @@ def read_openephys( if onebox_processor is not None: assert neuropix_pxi_processor is None, "Only one processor should be present" processor = onebox_processor + if onix_processor is not None: + processor = onix_processor if "NodeId" in processor.attrib: node_id = processor.attrib["NodeId"] @@ -797,6 +802,9 @@ def read_openephys( has_streams = False probe_names_used = None + if onix_processor is not None: + probe_names_used = [probe_name for probe_name in probe_names_used if 'Probe' in probe_name] + # for Open Ephys version < 1.0 np_probes is in the EDITOR field. # for Open Ephys version >= 1.0 np_probes is in the CUSTOM_PARAMETERS field. editor = processor.find("EDITOR") @@ -804,7 +812,10 @@ def read_openephys( np_probes = editor.findall("NP_PROBE") else: custom_parameters = editor.find("CUSTOM_PARAMETERS") - np_probes = custom_parameters.findall("NP_PROBE") + if onix_processor is not None: + np_probes = custom_parameters.findall("NEUROPIXELSV1E") + else: + np_probes = custom_parameters.findall("NP_PROBE") if len(np_probes) == 0: if raise_error: @@ -839,13 +850,18 @@ def read_openephys( # now load probe info from NP_PROBE fields np_probes_info = [] for probe_idx, np_probe in enumerate(np_probes): - slot = np_probe.attrib["slot"] - port = np_probe.attrib["port"] - dock = np_probe.attrib["dock"] - probe_part_number = np_probe.attrib["probe_part_number"] - probe_serial_number = np_probe.attrib["probe_serial_number"] - # read channels - channels = np_probe.find("CHANNELS") + if onix_processor is not None: + slot, port, dock = None, None, None + probe_part_number, probe_serial_number = np_probe.attrib["probePartNumber"], np_probe.attrib["probeSerialNumber"] + channels = np_probe.find("SELECTED_CHANNELS") + else: + slot = np_probe.attrib["slot"] + port = np_probe.attrib["port"] + dock = np_probe.attrib["dock"] + probe_part_number = np_probe.attrib["probe_part_number"] + probe_serial_number = np_probe.attrib["probe_serial_number"] + channels = np_probe.find("CHANNELS") + channel_names = np.array(list(channels.attrib.keys())) channel_ids = np.array([int(ch[2:]) for ch in channel_names]) channel_order = np.argsort(channel_ids) @@ -862,18 +878,20 @@ def read_openephys( else: shank_ids = None - electrode_xpos = np_probe.find("ELECTRODE_XPOS") - electrode_ypos = np_probe.find("ELECTRODE_YPOS") + if onix_processor is None: + electrode_xpos = np_probe.find("ELECTRODE_XPOS") + electrode_ypos = np_probe.find("ELECTRODE_YPOS") - if electrode_xpos is None or electrode_ypos is None: - if raise_error: - raise Exception("ELECTRODE_XPOS or ELECTRODE_YPOS is not available in settings!") - return None - xpos = np.array([float(electrode_xpos.attrib[ch]) for ch in channel_names]) - ypos = np.array([float(electrode_ypos.attrib[ch]) for ch in channel_names]) - positions = np.array([xpos, ypos]).T + if electrode_xpos is None or electrode_ypos is None: + if raise_error: + raise Exception("ELECTRODE_XPOS or ELECTRODE_YPOS is not available in settings!") + return None + xpos = np.array([float(electrode_xpos.attrib[ch]) for ch in channel_names]) + ypos = np.array([float(electrode_ypos.attrib[ch]) for ch in channel_names]) + positions = np.array([xpos, ypos]).T + else: + positions = np.reshape(np.arange(0,384*2*20,20), shape=(384,2)) - probe_part_number = np_probe.get("probe_part_number", None) pt_metadata, _, mux_info = get_probe_metadata_from_probe_features(probe_features, probe_part_number) shank_pitch = pt_metadata["shank_pitch_um"] From 86698350df55a0cfc4604d8663430407a263ae94 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 1 Sep 2025 11:26:57 +0000 Subject: [PATCH 2/5] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/probeinterface/neuropixels_tools.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/probeinterface/neuropixels_tools.py b/src/probeinterface/neuropixels_tools.py index 9639b00..a0ab110 100644 --- a/src/probeinterface/neuropixels_tools.py +++ b/src/probeinterface/neuropixels_tools.py @@ -803,7 +803,7 @@ def read_openephys( probe_names_used = None if onix_processor is not None: - probe_names_used = [probe_name for probe_name in probe_names_used if 'Probe' in probe_name] + probe_names_used = [probe_name for probe_name in probe_names_used if "Probe" in probe_name] # for Open Ephys version < 1.0 np_probes is in the EDITOR field. # for Open Ephys version >= 1.0 np_probes is in the CUSTOM_PARAMETERS field. @@ -852,7 +852,10 @@ def read_openephys( for probe_idx, np_probe in enumerate(np_probes): if onix_processor is not None: slot, port, dock = None, None, None - probe_part_number, probe_serial_number = np_probe.attrib["probePartNumber"], np_probe.attrib["probeSerialNumber"] + probe_part_number, probe_serial_number = ( + np_probe.attrib["probePartNumber"], + np_probe.attrib["probeSerialNumber"], + ) channels = np_probe.find("SELECTED_CHANNELS") else: slot = np_probe.attrib["slot"] @@ -890,7 +893,7 @@ def read_openephys( ypos = np.array([float(electrode_ypos.attrib[ch]) for ch in channel_names]) positions = np.array([xpos, ypos]).T else: - positions = np.reshape(np.arange(0,384*2*20,20), shape=(384,2)) + positions = np.reshape(np.arange(0, 384 * 2 * 20, 20), shape=(384, 2)) pt_metadata, _, mux_info = get_probe_metadata_from_probe_features(probe_features, probe_part_number) From 37dbf9c48680f7505a616fd803cbd4bc055288af Mon Sep 17 00:00:00 2001 From: chrishalcrow Date: Sun, 28 Sep 2025 19:37:46 +0100 Subject: [PATCH 3/5] further onix fun --- src/probeinterface/neuropixels_tools.py | 418 +++++++++++++----------- 1 file changed, 233 insertions(+), 185 deletions(-) diff --git a/src/probeinterface/neuropixels_tools.py b/src/probeinterface/neuropixels_tools.py index a0ab110..871abed 100644 --- a/src/probeinterface/neuropixels_tools.py +++ b/src/probeinterface/neuropixels_tools.py @@ -813,7 +813,18 @@ def read_openephys( else: custom_parameters = editor.find("CUSTOM_PARAMETERS") if onix_processor is not None: - np_probes = custom_parameters.findall("NEUROPIXELSV1E") + possible_probe_names = ["NEUROPIXELSV1E", "NEUROPIXELSV1F", "NEUROPIXELSV2E"] + parent_np_probe = "" + for probe_name in possible_probe_names: + print(f"{custom_parameters.findall(probe_name)=}") + parent_np_probe = custom_parameters.findall(probe_name) + if len(parent_np_probe) > 0: + break + if probe_name == "NEUROPIXELSV2E": + np_probes = [parent_np_probe[0].findall(f"PROBE{a}")[0] for a in range(2)] + else: + np_probes = [parent_np_probe[0]] + else: np_probes = custom_parameters.findall("NP_PROBE") @@ -835,6 +846,8 @@ def read_openephys( if not has_streams: probe_names_used = [f"{node_id}.{stream_index}" for stream_index in range(len(np_probes))] + print(f"{probe_names_used=}") + # check consistency with stream names and other fields if has_streams: # make sure we have at least as many NP_PROBE as the number of used probes @@ -847,210 +860,245 @@ def read_openephys( probe_features = _load_np_probe_features() - # now load probe info from NP_PROBE fields + list_of_probes = [] np_probes_info = [] - for probe_idx, np_probe in enumerate(np_probes): - if onix_processor is not None: + if onix_processor is not None: + + + for probe_idx, np_probe in enumerate(np_probes[1:]): + slot, port, dock = None, None, None - probe_part_number, probe_serial_number = ( - np_probe.attrib["probePartNumber"], - np_probe.attrib["probeSerialNumber"], - ) - channels = np_probe.find("SELECTED_CHANNELS") - else: - slot = np_probe.attrib["slot"] - port = np_probe.attrib["port"] - dock = np_probe.attrib["dock"] - probe_part_number = np_probe.attrib["probe_part_number"] - probe_serial_number = np_probe.attrib["probe_serial_number"] - channels = np_probe.find("CHANNELS") + probe_part_number, probe_serial_number = np_probe.attrib["probePartNumber"], np_probe.attrib["probeSerialNumber"] + selected_channels = np_probe.find("SELECTED_CHANNELS").attrib - channel_names = np.array(list(channels.attrib.keys())) - channel_ids = np.array([int(ch[2:]) for ch in channel_names]) - channel_order = np.argsort(channel_ids) + pt_metadata, _, mux_info = get_probe_metadata_from_probe_features(probe_features, probe_part_number) - # sort channel_names and channel_values - channel_names = channel_names[channel_order] - channel_values = np.array(list(channels.attrib.values()))[channel_order] + num_shank = pt_metadata["num_shanks"] + contact_per_shank = pt_metadata["cols_per_shank"] * pt_metadata["rows_per_shank"] + + if num_shank == 1: + elec_ids = np.arange(contact_per_shank) + shank_ids = None + else: + elec_ids = np.concatenate([np.arange(contact_per_shank) for i in range(num_shank)]) + shank_ids = np.concatenate([np.zeros(contact_per_shank) + i for i in range(num_shank)]) - # check if shank ids is present - if all(":" in val for val in channel_values): - shank_ids = np.array([int(val.split(":")[1]) for val in channel_values]) - elif all("_" in val for val in channel_names): - shank_ids = np.array([int(val.split("_")[1]) for val in channel_names]) - else: - shank_ids = None - - if onix_processor is None: - electrode_xpos = np_probe.find("ELECTRODE_XPOS") - electrode_ypos = np_probe.find("ELECTRODE_YPOS") - - if electrode_xpos is None or electrode_ypos is None: - if raise_error: - raise Exception("ELECTRODE_XPOS or ELECTRODE_YPOS is not available in settings!") - return None - xpos = np.array([float(electrode_xpos.attrib[ch]) for ch in channel_names]) - ypos = np.array([float(electrode_ypos.attrib[ch]) for ch in channel_names]) - positions = np.array([xpos, ypos]).T - else: - positions = np.reshape(np.arange(0, 384 * 2 * 20, 20), shape=(384, 2)) + full_probe = _make_npx_probe_from_description(pt_metadata, probe_part_number, elec_ids, shank_ids) - pt_metadata, _, mux_info = get_probe_metadata_from_probe_features(probe_features, probe_part_number) + selected_channel_indices = [int(channel_index) for channel_index in selected_channels.values()] - shank_pitch = pt_metadata["shank_pitch_um"] + sliced_probe = full_probe.get_slice(selection=selected_channel_indices) + elec_ids = [int(contact_id.split('e')[1]) for contact_id in sliced_probe.contact_ids] + shank_ids = [0 for contact_id in sliced_probe.contact_ids] - if fix_x_position_for_oe_5 and oe_version < parse("0.6.0") and shank_ids is not None: - positions[:, 1] = positions[:, 1] - shank_pitch * shank_ids + one_probe_again = _make_npx_probe_from_description(pt_metadata, probe_part_number, elec_ids, shank_ids, mux_info=mux_info) - # x offset so that the first column is at 0x - offset = np.min(positions[:, 0]) - # if some shanks are not used, we need to adjust the offset - if shank_ids is not None: - offset -= np.min(shank_ids) * shank_pitch - positions[:, 0] -= offset - - # - y_pitch = pt_metadata["electrode_pitch_vert_um"] # Vertical spacing between the centers of adjacent contacts - x_pitch = pt_metadata[ - "electrode_pitch_horz_um" - ] # Horizontal spacing between the centers of contacts within the same row - number_of_columns = pt_metadata["cols_per_shank"] - probe_stagger = ( - pt_metadata["even_row_horz_offset_left_edge_to_leftmost_electrode_center_um"] - - pt_metadata["odd_row_horz_offset_left_edge_to_leftmost_electrode_center_um"] - ) - num_shanks = pt_metadata["num_shanks"] + list_of_probes.append(one_probe_again) + + return list_of_probes, full_probe + - description = pt_metadata.get("description") - elec_ids = [] - for i, pos in enumerate(positions): - # Do not calculate contact ids if the model name is not known - if description is None: - elec_ids = None - break - x_pos = pos[0] - y_pos = pos[1] + else: - # Adds a shift to rows in the staggered configuration - is_row_staggered = np.mod(y_pos / y_pitch + 1, 2) == 1 - row_stagger = probe_stagger if is_row_staggered else 0 + # now load probe info from NP_PROBE fields + np_probes_info = [] + for probe_idx, np_probe in enumerate(np_probes): + + slot = np_probe.attrib["slot"] + port = np_probe.attrib["port"] + dock = np_probe.attrib["dock"] + probe_part_number = np_probe.attrib["probe_part_number"] + probe_serial_number = np_probe.attrib["probe_serial_number"] + channels = np_probe.find("CHANNELS") + + channel_names = np.array(list(channels.attrib.keys())) + channel_ids = np.array([int(ch[2:]) for ch in channel_names]) + channel_order = np.argsort(channel_ids) - # Map the positions to the contacts ids - shank_id = shank_ids[i] if num_shanks > 1 else 0 + # sort channel_names and channel_values + channel_names = channel_names[channel_order] + channel_values = np.array(list(channels.attrib.values()))[channel_order] - # Electrode ids are computed from the positions of the electrodes. The computation - # is different for probes with one row of electrodes, or more than one. - if x_pitch == 0: - elec_id = int(number_of_columns * y_pos / y_pitch) + # check if shank ids is present + if all(":" in val for val in channel_values): + shank_ids = np.array([int(val.split(":")[1]) for val in channel_values]) + elif all("_" in val for val in channel_names): + shank_ids = np.array([int(val.split("_")[1]) for val in channel_names]) else: - elec_id = int( - (x_pos - row_stagger - shank_pitch * shank_id) / x_pitch + number_of_columns * y_pos / y_pitch - ) - elec_ids.append(elec_id) - - np_probe_dict = { - "shank_ids": shank_ids, - "elec_ids": elec_ids, - "pt_metadata": pt_metadata, - "slot": slot, - "port": port, - "dock": dock, - "serial_number": probe_serial_number, - "part_number": probe_part_number, - "mux_info": mux_info, - } - # Sequentially assign probe names - if "custom_probe_name" in np_probe.attrib and np_probe.attrib["custom_probe_name"] != probe_serial_number: - name = np_probe.attrib["custom_probe_name"] - else: - name = probe_names_used[probe_idx] - np_probe_dict.update({"name": name}) - np_probes_info.append(np_probe_dict) - - # now select correct probe (if multiple) - if len(np_probes) > 1: - found = False - probe_names = [p["name"] for p in np_probes_info] - - if stream_name is not None: - assert probe_name is None and serial_number is None, ( - "Use one of 'stream_name', 'probe_name', " "or 'serial_number'" - ) - for probe_idx, probe_info in enumerate(np_probes_info): - if probe_info["name"] in stream_name or probe_info["serial_number"] in stream_name: - found = True - break - if not found: - if raise_error: - raise Exception( - f"The stream {stream_name} is not associated to an available probe: {probe_names_used}" - ) - return None - elif probe_name is not None: - assert stream_name is None and serial_number is None, ( - "Use one of 'stream_name', 'probe_name', " "or 'serial_number'" - ) - for probe_idx, probe_info in enumerate(np_probes_info): - if probe_info["name"] == probe_name: - found = True - break - if not found: - if raise_error: - raise Exception(f"The provided {probe_name} is not in the available probes: {probe_names_used}") - return None - elif serial_number is not None: - assert stream_name is None and probe_name is None, ( - "Use one of 'stream_name', 'probe_name', " "or 'serial_number'" + shank_ids = None + + if onix_processor is None: + electrode_xpos = np_probe.find("ELECTRODE_XPOS") + electrode_ypos = np_probe.find("ELECTRODE_YPOS") + + if electrode_xpos is None or electrode_ypos is None: + if raise_error: + raise Exception("ELECTRODE_XPOS or ELECTRODE_YPOS is not available in settings!") + return None + xpos = np.array([float(electrode_xpos.attrib[ch]) for ch in channel_names]) + ypos = np.array([float(electrode_ypos.attrib[ch]) for ch in channel_names]) + positions = np.array([xpos, ypos]).T + else: + positions = np.reshape(np.arange(0,384*2*20,20), shape=(384,2)) + + pt_metadata, _, mux_info = get_probe_metadata_from_probe_features(probe_features, probe_part_number) + + shank_pitch = pt_metadata["shank_pitch_um"] + + if fix_x_position_for_oe_5 and oe_version < parse("0.6.0") and shank_ids is not None: + positions[:, 1] = positions[:, 1] - shank_pitch * shank_ids + + # x offset so that the first column is at 0x + offset = np.min(positions[:, 0]) + # if some shanks are not used, we need to adjust the offset + if shank_ids is not None: + offset -= np.min(shank_ids) * shank_pitch + positions[:, 0] -= offset + + # + y_pitch = pt_metadata["electrode_pitch_vert_um"] # Vertical spacing between the centers of adjacent contacts + x_pitch = pt_metadata[ + "electrode_pitch_horz_um" + ] # Horizontal spacing between the centers of contacts within the same row + number_of_columns = pt_metadata["cols_per_shank"] + probe_stagger = ( + pt_metadata["even_row_horz_offset_left_edge_to_leftmost_electrode_center_um"] + - pt_metadata["odd_row_horz_offset_left_edge_to_leftmost_electrode_center_um"] ) - for probe_idx, probe_info in enumerate(np_probes_info): - if probe_info["serial_number"] == str(serial_number): - found = True + num_shanks = pt_metadata["num_shanks"] + + description = pt_metadata.get("description") + + elec_ids = [] + for i, pos in enumerate(positions): + # Do not calculate contact ids if the model name is not known + if description is None: + elec_ids = None break - if not found: - np_serial_numbers = [p["serial_number"] for p in probe_info] - if raise_error: - raise Exception( - f"The provided {serial_number} is not in the available serial numbers: {np_serial_numbers}" + + x_pos = pos[0] + y_pos = pos[1] + + # Adds a shift to rows in the staggered configuration + is_row_staggered = np.mod(y_pos / y_pitch + 1, 2) == 1 + row_stagger = probe_stagger if is_row_staggered else 0 + + # Map the positions to the contacts ids + shank_id = shank_ids[i] if num_shanks > 1 else 0 + + # Electrode ids are computed from the positions of the electrodes. The computation + # is different for probes with one row of electrodes, or more than one. + if x_pitch == 0: + elec_id = int(number_of_columns * y_pos / y_pitch) + else: + elec_id = int( + (x_pos - row_stagger - shank_pitch * shank_id) / x_pitch + number_of_columns * y_pos / y_pitch ) - return None + elec_ids.append(elec_id) + + np_probe_dict = { + "shank_ids": shank_ids, + "elec_ids": elec_ids, + "pt_metadata": pt_metadata, + "slot": slot, + "port": port, + "dock": dock, + "serial_number": probe_serial_number, + "part_number": probe_part_number, + "mux_info": mux_info, + } + # Sequentially assign probe names + if "custom_probe_name" in np_probe.attrib and np_probe.attrib["custom_probe_name"] != probe_serial_number: + name = np_probe.attrib["custom_probe_name"] + else: + name = probe_names_used[probe_idx] + np_probe_dict.update({"name": name}) + np_probes_info.append(np_probe_dict) + + # now select correct probe (if multiple) + if len(np_probes) > 1: + found = False + probe_names = [p["name"] for p in np_probes_info] + + if stream_name is not None: + assert probe_name is None and serial_number is None, ( + "Use one of 'stream_name', 'probe_name', " "or 'serial_number'" + ) + for probe_idx, probe_info in enumerate(np_probes_info): + if probe_info["name"] in stream_name or probe_info["serial_number"] in stream_name: + found = True + break + if not found: + if raise_error: + raise Exception( + f"The stream {stream_name} is not associated to an available probe: {probe_names_used}" + ) + return None + elif probe_name is not None: + assert stream_name is None and serial_number is None, ( + "Use one of 'stream_name', 'probe_name', " "or 'serial_number'" + ) + for probe_idx, probe_info in enumerate(np_probes_info): + if probe_info["name"] == probe_name: + found = True + break + if not found: + if raise_error: + raise Exception(f"The provided {probe_name} is not in the available probes: {probe_names_used}") + return None + elif serial_number is not None: + assert stream_name is None and probe_name is None, ( + "Use one of 'stream_name', 'probe_name', " "or 'serial_number'" + ) + for probe_idx, probe_info in enumerate(np_probes_info): + if probe_info["serial_number"] == str(serial_number): + found = True + break + if not found: + np_serial_numbers = [p["serial_number"] for p in probe_info] + if raise_error: + raise Exception( + f"The provided {serial_number} is not in the available serial numbers: {np_serial_numbers}" + ) + return None + else: + raise Exception( + f"More than one probe found. Use one of 'stream_name', 'probe_name', or 'serial_number' " + f"to select the right probe.\nProbe names: {probe_names}" + ) else: - raise Exception( - f"More than one probe found. Use one of 'stream_name', 'probe_name', or 'serial_number' " - f"to select the right probe.\nProbe names: {probe_names}" - ) - else: - # in case of a single probe, make sure it is consistent with optional - # stream_name, probe_name, or serial number - available_probe_name = np_probes_info[0]["name"] - available_serial_number = np_probes_info[0]["serial_number"] - - if stream_name: - if available_probe_name not in stream_name: - if raise_error: - raise Exception( - f"Inconsistency between provided stream {stream_name} and available probe " - f"{available_probe_name}" - ) - return None - if probe_name: - if probe_name != available_probe_name: - if raise_error: - raise Exception( - f"Inconsistency between provided probe name {probe_name} and available probe " - f"{available_probe_name}" - ) - return None - if serial_number: - if str(serial_number) != available_serial_number: - if raise_error: - raise Exception( - f"Inconsistency between provided serial number {serial_number} and available serial numbers " - f"{available_serial_number}" - ) - return None - probe_idx = 0 + # in case of a single probe, make sure it is consistent with optional + # stream_name, probe_name, or serial number + available_probe_name = np_probes_info[0]["name"] + available_serial_number = np_probes_info[0]["serial_number"] + + if stream_name: + if available_probe_name not in stream_name: + if raise_error: + raise Exception( + f"Inconsistency between provided stream {stream_name} and available probe " + f"{available_probe_name}" + ) + return None + if probe_name: + if probe_name != available_probe_name: + if raise_error: + raise Exception( + f"Inconsistency between provided probe name {probe_name} and available probe " + f"{available_probe_name}" + ) + return None + if serial_number: + if str(serial_number) != available_serial_number: + if raise_error: + raise Exception( + f"Inconsistency between provided serial number {serial_number} and available serial numbers " + f"{available_serial_number}" + ) + return None + probe_idx = 0 np_probe_info = np_probes_info[probe_idx] np_probe = np_probes[probe_idx] From feaa386674db21f74c33988155e8ab07abc8e094 Mon Sep 17 00:00:00 2001 From: Alessio Buccino Date: Tue, 7 Oct 2025 16:26:42 +0200 Subject: [PATCH 4/5] Add NP2 onix data and support SELECTED_ELECTRODES --- src/probeinterface/neuropixels_tools.py | 15 +- .../openephys/OE_ONIX-NP/settings_NP2.xml | 382 ++++++++++++++++++ tests/test_io/test_openephys.py | 17 + 3 files changed, 406 insertions(+), 8 deletions(-) create mode 100644 tests/data/openephys/OE_ONIX-NP/settings_NP2.xml diff --git a/src/probeinterface/neuropixels_tools.py b/src/probeinterface/neuropixels_tools.py index 5a2de5b..8f6f817 100644 --- a/src/probeinterface/neuropixels_tools.py +++ b/src/probeinterface/neuropixels_tools.py @@ -860,23 +860,22 @@ def read_openephys( # now load probe info from NP_PROBE fields np_probes_info = [] for probe_idx, np_probe in enumerate(np_probes): - # selected_channels is the preferred way to instantiate the probe + # selected_electrodes is the preferred way to instantiate the probe # if this field is available, a full probe is created from the probe_part_number - # and then sliced using the selected channels + # and then sliced using the selected electrodes. # if not available, the xpos and ypos fields are used to create the probe - # if onix_processor is None: slot = np_probe.attrib.get("slot") port = np_probe.attrib.get("port") dock = np_probe.attrib.get("dock") probe_part_number = np_probe.attrib.get("probe_part_number") or np_probe.attrib.get("probePartNumber") probe_serial_number = np_probe.attrib.get("probe_serial_number") or np_probe.attrib.get("probeSerialNumber") - selected_channels = np_probe.find("SELECTED_CHANNELS") + selected_electrodes = np_probe.find("SELECTED_ELECTRODES") or np_probe.find("SELECTED_CHANNELS") channels = np_probe.find("CHANNELS") pt_metadata, _, mux_info = get_probe_metadata_from_probe_features(probe_features, probe_part_number) - if selected_channels is not None: - selected_channels_values = selected_channels.attrib.values() + if selected_electrodes is not None: + selected_electrodes_values = selected_electrodes.attrib.values() num_shank = pt_metadata["num_shanks"] contact_per_shank = pt_metadata["cols_per_shank"] * pt_metadata["rows_per_shank"] @@ -892,9 +891,9 @@ def read_openephys( pt_metadata, probe_part_number, elec_ids, shank_ids, mux_info=mux_info ) - selected_channel_indices = [int(channel_index) for channel_index in selected_channels_values] + selected_electrode_indices = [int(electrode_index) for electrode_index in selected_electrodes_values] - sliced_probe = full_probe.get_slice(selection=selected_channel_indices) + sliced_probe = full_probe.get_slice(selection=selected_electrode_indices) np_probe_dict = { "pt_metadata": pt_metadata, diff --git a/tests/data/openephys/OE_ONIX-NP/settings_NP2.xml b/tests/data/openephys/OE_ONIX-NP/settings_NP2.xml new file mode 100644 index 0000000..a88d41d --- /dev/null +++ b/tests/data/openephys/OE_ONIX-NP/settings_NP2.xml @@ -0,0 +1,382 @@ + + + + + 1.0.1 + 10 + 11 Sep 2025 11:52:51 + Windows 11 + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/tests/test_io/test_openephys.py b/tests/test_io/test_openephys.py index f9163ff..630330f 100644 --- a/tests/test_io/test_openephys.py +++ b/tests/test_io/test_openephys.py @@ -311,6 +311,23 @@ def test_onix(): # Verify each group has exactly 4 contacts assert unique_groups.shape[1] == 4, f"Tetrode groups should have 4 contacts, found {unique_groups.shape[1]}" + # NP2.0 + probe_np2_probe0 = read_openephys( + 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) + validate_probe_dict(probe_dict) + assert probe_np2_probe0.get_contact_count() == 384 + assert probe_np2_probe0.get_shank_count() == 1 + + probe_np2_probe1 = read_openephys( + 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) + validate_probe_dict(probe_dict) + assert probe_np2_probe1.get_contact_count() == 384 + assert probe_np2_probe1.get_shank_count() == 4 + if __name__ == "__main__": # test_multiple_probes() From c52aee11c6ce9d79f732a0c185e5dbeb2824dce8 Mon Sep 17 00:00:00 2001 From: Alessio Buccino Date: Tue, 7 Oct 2025 16:28:42 +0200 Subject: [PATCH 5/5] Only support SELECTED_ELECTRODES, not SELECTED_CHANNELS --- src/probeinterface/neuropixels_tools.py | 2 +- tests/data/openephys/OE_ONIX-NP/settings_NP2.xml | 4 ++-- tests/data/openephys/OE_ONIX-NP/settings_bankA.xml | 2 +- tests/data/openephys/OE_ONIX-NP/settings_bankB.xml | 2 +- tests/data/openephys/OE_ONIX-NP/settings_bankC.xml | 2 +- tests/data/openephys/OE_ONIX-NP/settings_tetrodes.xml | 2 +- 6 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/probeinterface/neuropixels_tools.py b/src/probeinterface/neuropixels_tools.py index 8f6f817..71eb370 100644 --- a/src/probeinterface/neuropixels_tools.py +++ b/src/probeinterface/neuropixels_tools.py @@ -869,7 +869,7 @@ def read_openephys( dock = np_probe.attrib.get("dock") probe_part_number = np_probe.attrib.get("probe_part_number") or np_probe.attrib.get("probePartNumber") probe_serial_number = np_probe.attrib.get("probe_serial_number") or np_probe.attrib.get("probeSerialNumber") - selected_electrodes = np_probe.find("SELECTED_ELECTRODES") or np_probe.find("SELECTED_CHANNELS") + selected_electrodes = np_probe.find("SELECTED_ELECTRODES") channels = np_probe.find("CHANNELS") pt_metadata, _, mux_info = get_probe_metadata_from_probe_features(probe_features, probe_part_number) diff --git a/tests/data/openephys/OE_ONIX-NP/settings_NP2.xml b/tests/data/openephys/OE_ONIX-NP/settings_NP2.xml index a88d41d..9794438 100644 --- a/tests/data/openephys/OE_ONIX-NP/settings_NP2.xml +++ b/tests/data/openephys/OE_ONIX-NP/settings_NP2.xml @@ -59,7 +59,7 @@ flexVersion="0.1" searchForCorrectionFiles="1" gainCorrectionFolder="C:\Users\anjal.doshi\Desktop\23176523171" gainCorrectionFile="C:\Users\anjal.doshi\Desktop\23176523171\23176523171_gainCalValues.csv"> - - - - - -