From 20c7b1832130981a57bc610944a02cb69dc3e0d5 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 13 Apr 2026 11:21:04 +0000 Subject: [PATCH 1/2] Add NIDQ (NI-DAQ) stream support to neuropixelsGLX reader The neuropixelsGLX reader previously only handled Imec AP-band streams (.ap.meta/.ap.bin). This adds full support for NIDQ streams (.nidq.meta/.nidq.bin), matching the NDR-matlab reference implementation. Changes: - header.py: Parse snsMnMaXaDw with full MN/MA/XA/DW breakdown. Compute per-bit digital line mapping from niXDBytes1/niXDBytes2. Add niAiRangeMax/Min, niMaxInt, and NI-DAQ gain fields. Also add digital line mapping for IMEC sync words (16 bits per sync column). - neuropixelsGLX.py: Rewrite filenamefromepochfiles() to find .bin first and derive .meta (matching MATLAB), allowing mixed AP+NIDQ epoch lists. Update getchannelsepoch() to expose individual digital lines via n_digital_lines. Update readchannels_epochsamples() digital_in to extract single bits from packed digital words using bitwise operations. - Add 39 new tests covering header parsing, channel enumeration, sample reading, and bit extraction for both NIDQ and AP streams. https://claude.ai/code/session_01HEWqAySjSJ6dFeNWJrdsrJ --- src/ndr/format/neuropixelsGLX/header.py | 96 ++++- src/ndr/reader/neuropixelsGLX.py | 152 +++++--- tests/test_neuropixelsGLX.py | 464 ++++++++++++++++++++++++ 3 files changed, 662 insertions(+), 50 deletions(-) create mode 100644 tests/test_neuropixelsGLX.py diff --git a/src/ndr/format/neuropixelsGLX/header.py b/src/ndr/format/neuropixelsGLX/header.py index 96b73d9..ef48b16 100644 --- a/src/ndr/format/neuropixelsGLX/header.py +++ b/src/ndr/format/neuropixelsGLX/header.py @@ -8,6 +8,8 @@ from pathlib import Path from typing import Any +import numpy as np + from ndr.format.neuropixelsGLX.readmeta import readmeta @@ -49,8 +51,13 @@ def header(metafilename: str | Path) -> dict[str, Any]: Dictionary with fields: - sample_rate : float - n_saved_chans : int - - n_neural_chans : int - - n_sync_chans : int + - n_neural_chans : int — for NIDQ this is MN + MA + XA + - n_sync_chans : int — number of digital word int16 columns + - n_digital_word_cols : int — same as n_sync_chans + - n_digital_lines : int — number of individual bit lines exposed + - digital_line_col : ndarray — 0-based DW column offset per line + - digital_line_bit : ndarray — 0-based bit position per line + - digital_line_label : list of str — label per line - saved_chan_list : list of int (1-based) - voltage_range : tuple of (float, float) - max_int : int @@ -79,7 +86,13 @@ def header(metafilename: str | Path) -> dict[str, Any]: # Number of saved channels info["n_saved_chans"] = int(meta["nSavedChans"]) - # Parse snsApLfSy or snsMnMaXaDw to determine neural vs sync channels + # Parse snsApLfSy or snsMnMaXaDw to determine neural vs sync channels. + # Also compute digital line mapping: + # n_digital_word_cols : number of int16 columns holding digital data + # n_digital_lines : number of single-bit digital lines exposed + # digital_line_col : (n_digital_lines,) 0-based DW column offset + # digital_line_bit : (n_digital_lines,) 0-based bit position + # digital_line_label : list of str labels per line if "snsApLfSy" in meta: counts = [int(x) for x in meta["snsApLfSy"].split(",")] # counts[0] = AP chans, counts[1] = LF chans, counts[2] = SY chans @@ -92,15 +105,76 @@ def header(metafilename: str | Path) -> dict[str, Any]: else: info["stream_type"] = "ap" info["n_neural_chans"] = counts[0] + # IMEC sync: each sync column provides 16 bits + info["n_digital_word_cols"] = info["n_sync_chans"] + n_lines = 16 * info["n_sync_chans"] + info["n_digital_lines"] = n_lines + cols = np.zeros(n_lines, dtype=int) + bits = np.zeros(n_lines, dtype=int) + labels: list[str] = [] + idx = 0 + for c in range(info["n_sync_chans"]): + for b in range(16): + cols[idx] = c + bits[idx] = b + labels.append(f"SY{c}.{b}") + idx += 1 + info["digital_line_col"] = cols + info["digital_line_bit"] = bits + info["digital_line_label"] = labels elif "snsMnMaXaDw" in meta: info["stream_type"] = "nidq" counts = [int(x) for x in meta["snsMnMaXaDw"].split(",")] - info["n_neural_chans"] = counts[0] - info["n_sync_chans"] = 0 + info["n_mn_chans"] = counts[0] # multiplexed neural + info["n_ma_chans"] = counts[1] # multiplexed analog + info["n_xa_chans"] = counts[2] # non-multiplexed analog + info["n_dw_chans"] = counts[3] # digital word int16 columns + info["n_neural_chans"] = counts[0] + counts[1] + counts[2] + info["n_sync_chans"] = counts[3] + info["n_digital_word_cols"] = counts[3] + + # Bytes saved per port — each byte = 8 active digital lines + n_bytes_p0 = int(meta.get("niXDBytes1", "0")) + n_bytes_p1 = int(meta.get("niXDBytes2", "0")) + + if n_bytes_p0 == 0 and n_bytes_p1 == 0: + # Fall back: assume all 16 bits of every DW column are active + n_lines_p0 = 16 * info["n_dw_chans"] + n_lines_p1 = 0 + else: + n_lines_p0 = 8 * n_bytes_p0 + n_lines_p1 = 8 * n_bytes_p1 + + n_lines = n_lines_p0 + n_lines_p1 + info["n_digital_lines"] = n_lines + cols = np.zeros(n_lines, dtype=int) + bits = np.zeros(n_lines, dtype=int) + labels = [] + idx = 0 + for k in range(n_lines_p0): + abs_bit = k + cols[idx] = abs_bit // 16 + bits[idx] = abs_bit % 16 + labels.append(f"XD{k}") + idx += 1 + for k in range(n_lines_p1): + abs_bit = n_bytes_p0 * 8 + k + cols[idx] = abs_bit // 16 + bits[idx] = abs_bit % 16 + labels.append(f"XD1.{k}") + idx += 1 + info["digital_line_col"] = cols + info["digital_line_bit"] = bits + info["digital_line_label"] = labels else: info["stream_type"] = "unknown" info["n_neural_chans"] = info["n_saved_chans"] - 1 info["n_sync_chans"] = 1 + info["n_digital_word_cols"] = 1 + info["n_digital_lines"] = 16 + info["digital_line_col"] = np.zeros(16, dtype=int) + info["digital_line_bit"] = np.arange(16, dtype=int) + info["digital_line_label"] = [f"bit{b}" for b in range(16)] # Parse saved channel subset if "snsSaveChanSubset" in meta: @@ -115,15 +189,27 @@ def header(metafilename: str | Path) -> dict[str, Any]: vmax = float(meta["imAiRangeMax"]) vmin = float(meta["imAiRangeMin"]) info["voltage_range"] = (vmin, vmax) + elif "niAiRangeMax" in meta: + vmax = float(meta["niAiRangeMax"]) + vmin = float(meta["niAiRangeMin"]) + info["voltage_range"] = (vmin, vmax) else: info["voltage_range"] = (-0.6, 0.6) # Neuropixels 1.0 default # Max integer value if "imMaxInt" in meta: info["max_int"] = int(meta["imMaxInt"]) + elif "niMaxInt" in meta: + info["max_int"] = int(meta["niMaxInt"]) else: info["max_int"] = 512 # Neuropixels 1.0 default + # NI-DAQ gains + if "niMNGain" in meta: + info["ni_mn_gain"] = float(meta["niMNGain"]) + if "niMAGain" in meta: + info["ni_ma_gain"] = float(meta["niMAGain"]) + # Bits per sample info["bits_per_sample"] = 16 diff --git a/src/ndr/reader/neuropixelsGLX.py b/src/ndr/reader/neuropixelsGLX.py index e6825ed..2a19001 100644 --- a/src/ndr/reader/neuropixelsGLX.py +++ b/src/ndr/reader/neuropixelsGLX.py @@ -18,21 +18,27 @@ class ndr_reader_neuropixelsGLX(ndr_reader_base): - """Reader for Neuropixels SpikeGLX AP-band data. + """Reader for SpikeGLX data (AP, LF, and NIDQ streams). - This class reads action-potential band data from Neuropixels probes - acquired with the SpikeGLX software. Each instance handles one probe's - AP stream (one .ap.bin / .ap.meta file pair per epoch). + This class reads data from Neuropixels probes and NI-DAQ devices + acquired with the SpikeGLX software. Each instance handles one + stream (one .bin / .meta file pair per epoch). - SpikeGLX saves Neuropixels data as flat interleaved int16 binary files - with companion .meta text files. The binary files have no header. - Channel count, sample rate, and gain information are read from the - .meta file. + SpikeGLX saves data as flat interleaved int16 binary files with + companion .meta text files. The binary files have no header. + Channel count, sample rate, and gain information are read from + the .meta file. Channel mapping: - - Neural channels are exposed as 'analog_in' (ai1..aiN) - - The sync word is exposed as 'digital_in' (di1) - - A single time channel 't1' is always present + - Analog channels are exposed as 'analog_in' (ai1..aiN). + For AP streams these are neural probe channels; for NIDQ + streams these are NI-DAQ analog inputs (MN + MA + XA). + - Digital lines are exposed as 'digital_in' (di1..diM), + where each di channel is a single bit of the packed digital + word(s). For NIDQ streams the count comes from + ``8 * (niXDBytes1 + niXDBytes2)``; for IMEC streams it is + ``16 * n_sync_chans``. + - A single time channel 't1' is always present. Data is returned as int16 to preserve native precision. Use :func:`ndr.format.neuropixelsGLX.samples2volts` for voltage conversion. @@ -75,8 +81,10 @@ def getchannelsepoch( ) -> list[dict[str, Any]]: """List channels available for a given epoch. - Neural channels are 'analog_in' (ai1..aiN), the sync channel is - 'digital_in' (di1), and a time channel 't1' is always present. + Analog channels are 'analog_in' (ai1..aiN), digital lines are + 'digital_in' (di1..diM) with one entry per single-bit line in + the packed digital word(s), and a time channel 't1' is always + present. """ metafile = self.filenamefromepochfiles(epochstreams) info = header(metafile) @@ -86,13 +94,13 @@ def getchannelsepoch( # Time channel channels.append({"name": "t1", "type": "time", "time_channel": 1}) - # Neural channels (analog_in) + # Analog channels (analog_in) for i in range(1, info["n_neural_chans"] + 1): channels.append({"name": f"ai{i}", "type": "analog_in", "time_channel": 1}) - # Sync channel (digital_in) - if info["n_sync_chans"] > 0: - channels.append({"name": "di1", "type": "digital_in", "time_channel": 1}) + # Digital lines (digital_in) — one per bit of the packed digital word(s) + for i in range(1, info["n_digital_lines"] + 1): + channels.append({"name": f"di{i}", "type": "digital_in", "time_channel": 1}) return channels @@ -145,9 +153,12 @@ def readchannels_epochsamples( Reads data between sample s0 and s1 (inclusive, 1-based). - For 'analog_in': returns int16 neural data. + For 'analog_in': returns int16 data (neural for AP, NI-DAQ analog + inputs for NIDQ). For 'time': returns float64 time stamps in seconds. - For 'digital_in': returns int16 sync word values. + For 'digital_in': returns int16 single-bit values (0 or 1) + extracted from the packed digital word(s). ``channel`` gives + the 1-based digital line(s). """ metafile = self.filenamefromepochfiles(epochstreams) info = header(metafile) @@ -178,18 +189,46 @@ def readchannels_epochsamples( return data elif ct in ("digital_in", "di"): - # Sync channel is the last channel in the file - sync_chan = np.array([info["n_saved_chans"]], dtype=np.int64) - data, _, _, _ = binarymatrix_read( - binfile, - info["n_saved_chans"], - sync_chan, - float(s0), - float(s1), - dataType="int16", - byteOrder="ieee-le", - headerSkip=0, - ) + if isinstance(channel, int): + channel = [channel] + line_idx = np.array(channel, dtype=int) + + if np.any(line_idx < 1) or np.any(line_idx > info["n_digital_lines"]): + raise ValueError( + f"Digital line out of range; valid lines are 1..{info['n_digital_lines']}." + ) + + # Digital word columns are at the end of each sample row + first_dw_col = info["n_saved_chans"] - info["n_digital_word_cols"] + 1 + + # Look up the (column, bit) position for each requested line + # line_idx is 1-based; digital_line_col/bit arrays are 0-indexed + col_offsets = info["digital_line_col"][line_idx - 1] + bit_pos = info["digital_line_bit"][line_idx - 1] + + n_samples = int(s1) - int(s0) + 1 + data = np.zeros((n_samples, len(channel)), dtype=np.int16) + + unique_cols = np.unique(col_offsets) + for uc in unique_cols: + file_col = np.array([first_dw_col + uc], dtype=np.int64) + raw, _, _, _ = binarymatrix_read( + binfile, + info["n_saved_chans"], + file_col, + float(s0), + float(s1), + dataType="int16", + byteOrder="ieee-le", + headerSkip=0, + ) + # raw is (n_samples, 1) int16 — treat as uint16 for bit extraction + raw_uint = raw[:, 0].view(np.uint16) + mask = col_offsets == uc + indices = np.where(mask)[0] + for idx in indices: + data[:, idx] = ((raw_uint >> bit_pos[idx]) & 1).astype(np.int16) + return data else: @@ -204,9 +243,9 @@ def readevents_epochsamples_native( t0: float, t1: float, ) -> tuple[np.ndarray, np.ndarray]: - """Read events or markers. Neuropixels has no native event channels. + """Read events or markers. SpikeGLX has no native event channels. - Returns empty arrays since Neuropixels data is purely + Returns empty arrays since SpikeGLX data is purely regularly-sampled. """ return np.array([]), np.array([]) @@ -220,8 +259,8 @@ def samplerate( ) -> np.ndarray | float: """Get the sample rate for specified channels. - For Neuropixels AP-band, all channels share the same sample rate - (typically 30 kHz). + All channels in a single SpikeGLX binary file share one sample + rate: typically 30 kHz for AP, 2.5 kHz for LF, or ~25 kHz for NIDQ. """ metafile = self.filenamefromepochfiles(epochstreams) info = header(metafile) @@ -230,18 +269,41 @@ def samplerate( return info["sample_rate"] def filenamefromepochfiles(self, filename_array: list[str]) -> str: - """Identify the .ap.meta file from epoch file list. + """Identify the companion .meta file from the epoch file list. - Searches the list for a file matching *.ap.meta. Errors if zero - or more than one match is found. - """ - matches = [f for f in filename_array if f.lower().endswith(".ap.meta")] + First searches for a ``.bin`` file and derives the ``.meta`` path + from it (replacing the last 3 characters). This allows multiple + ``.meta`` files (e.g. both AP and NIDQ) to coexist in the same + epoch file list — the ``.bin`` file disambiguates which stream + this reader instance handles. - if len(matches) == 0: - raise FileNotFoundError("No .ap.meta file found in the epoch file list.") - if len(matches) > 1: - raise ValueError("Multiple .ap.meta files found. Each epoch should have exactly one.") - return matches[0] + If no ``.bin`` file is present, falls back to finding a single + ``.meta`` file. Errors if zero or more than one ``.meta`` match + is found without a ``.bin`` to disambiguate. + """ + # Primary path: find the .bin file and derive .meta from it + binfile = "" + for f in filename_array: + if f.lower().endswith(".bin"): + binfile = f + break + + if binfile: + metafile = binfile[:-3] + "meta" + # Verify the .meta exists in the file list or on disk + if metafile not in filename_array and not Path(metafile).is_file(): + raise FileNotFoundError(f"No companion .meta file found for {binfile}.") + return metafile + + # Fallback: find a single .meta file + meta_matches = [f for f in filename_array if f.lower().endswith(".meta")] + if len(meta_matches) == 0: + raise FileNotFoundError("No .meta file found in the epoch file list.") + if len(meta_matches) > 1: + raise ValueError( + "Multiple .meta files found and no .bin file to disambiguate." + ) + return meta_matches[0] def daqchannels2internalchannels( self, diff --git a/tests/test_neuropixelsGLX.py b/tests/test_neuropixelsGLX.py new file mode 100644 index 0000000..497d98c --- /dev/null +++ b/tests/test_neuropixelsGLX.py @@ -0,0 +1,464 @@ +"""Tests for Neuropixels SpikeGLX reader — AP and NIDQ stream support.""" + +from __future__ import annotations + +import struct +from pathlib import Path + +import numpy as np +import pytest + +from ndr.format.neuropixelsGLX.header import header +from ndr.reader.neuropixelsGLX import ndr_reader_neuropixelsGLX + + +# --------------------------------------------------------------------------- +# Fixtures: create minimal .meta / .bin files on disk +# --------------------------------------------------------------------------- + + +NIDQ_META_CONTENT = """\ +acqMnMaXaDw=0,0,8,1 +appVersion=20231207 +fileCreateTime=2026-02-10T12:35:42 +fileSizeBytes=618204852 +fileTimeSecs=3242.1410015948127 +firstSample=501549 +nSavedChans=9 +niAiRangeMax=5 +niAiRangeMin=-5 +niMaxInt=32768 +niSampRate=10593.220339 +niXDBytes1=1 +niXDChans1=0:7 +snsMnMaXaDw=0,0,8,1 +snsSaveChanSubset=all +typeThis=nidq +~snsChanMap=(0,0,32,8,1)(XA0;0:0)(XA1;1:1)(XA2;2:2)(XA3;3:3)(XA4;4:4)(XA5;5:5)(XA6;6:6)(XA7;7:7)(XD0;8:8) +""" + +AP_META_CONTENT = """\ +appVersion=20231207 +fileCreateTime=2026-02-10T12:35:42 +fileSizeBytes=1000000 +fileTimeSecs=1.0 +imAiRangeMax=0.6 +imAiRangeMin=-0.6 +imDatPrb_type=0 +imDatPrb_sn=12345 +imMaxInt=512 +imSampRate=30000 +nSavedChans=385 +snsApLfSy=384,0,1 +snsSaveChanSubset=all +typeThis=imec +""" + + +def _write_meta(directory: Path, basename: str, content: str) -> Path: + meta_path = directory / basename + meta_path.write_text(content) + return meta_path + + +def _write_bin(directory: Path, basename: str, n_chans: int, n_samples: int) -> Path: + """Write a minimal int16 binary file with known data.""" + bin_path = directory / basename + data = np.zeros((n_samples, n_chans), dtype=np.int16) + # Fill analog channels with ascending values per channel + for c in range(n_chans - 1): # last column(s) = digital word + data[:, c] = np.arange(1, n_samples + 1, dtype=np.int16) * (c + 1) + # Digital word column: pack known bit pattern + # Set bit 0 on odd samples, bit 1 on every-4th sample, etc. + for s in range(n_samples): + dw = 0 + if s % 2 == 1: + dw |= 1 # bit 0 + if s % 4 == 0: + dw |= 2 # bit 1 + if s % 8 == 0: + dw |= 4 # bit 2 + data[s, n_chans - 1] = np.int16(dw) + data.tofile(bin_path) + return bin_path + + +@pytest.fixture +def nidq_files(tmp_path): + """Create NIDQ .meta + .bin in a temp directory (9 channels: 8 XA + 1 DW).""" + n_chans = 9 + n_samples = 100 + meta = _write_meta(tmp_path, "run_g0_t0.nidq.meta", NIDQ_META_CONTENT) + binf = _write_bin(tmp_path, "run_g0_t0.nidq.bin", n_chans, n_samples) + return str(meta), str(binf), n_chans, n_samples + + +@pytest.fixture +def ap_files(tmp_path): + """Create AP .meta + .bin in a temp directory (385 channels: 384 AP + 1 SY).""" + n_chans = 385 + n_samples = 50 + meta = _write_meta(tmp_path, "run_g0_t0.imec0.ap.meta", AP_META_CONTENT) + binf = _write_bin(tmp_path, "run_g0_t0.imec0.ap.bin", n_chans, n_samples) + return str(meta), str(binf), n_chans, n_samples + + +# --------------------------------------------------------------------------- +# header.py tests +# --------------------------------------------------------------------------- + + +class TestHeaderNIDQ: + def test_stream_type(self, nidq_files): + meta_path, _, _, _ = nidq_files + info = header(meta_path) + assert info["stream_type"] == "nidq" + + def test_sample_rate(self, nidq_files): + meta_path, _, _, _ = nidq_files + info = header(meta_path) + assert info["sample_rate"] == pytest.approx(10593.220339) + + def test_channel_counts(self, nidq_files): + meta_path, _, _, _ = nidq_files + info = header(meta_path) + assert info["n_saved_chans"] == 9 + assert info["n_mn_chans"] == 0 + assert info["n_ma_chans"] == 0 + assert info["n_xa_chans"] == 8 + assert info["n_dw_chans"] == 1 + assert info["n_neural_chans"] == 8 # MN + MA + XA + assert info["n_sync_chans"] == 1 # DW count + + def test_digital_lines(self, nidq_files): + meta_path, _, _, _ = nidq_files + info = header(meta_path) + # niXDBytes1=1 → 8 digital lines + assert info["n_digital_lines"] == 8 + assert info["n_digital_word_cols"] == 1 + assert len(info["digital_line_col"]) == 8 + assert len(info["digital_line_bit"]) == 8 + assert len(info["digital_line_label"]) == 8 + # All 8 lines should be in column 0 + np.testing.assert_array_equal(info["digital_line_col"], 0) + # Bits should be 0..7 + np.testing.assert_array_equal(info["digital_line_bit"], np.arange(8)) + # Labels + assert info["digital_line_label"][0] == "XD0" + assert info["digital_line_label"][7] == "XD7" + + def test_voltage_range(self, nidq_files): + meta_path, _, _, _ = nidq_files + info = header(meta_path) + assert info["voltage_range"] == (-5.0, 5.0) + + def test_max_int(self, nidq_files): + meta_path, _, _, _ = nidq_files + info = header(meta_path) + assert info["max_int"] == 32768 + + def test_ni_gains(self, nidq_files): + """NI gains are not in the minimal fixture, but check they don't error.""" + meta_path, _, _, _ = nidq_files + info = header(meta_path) + # The example meta doesn't have niMNGain/niMAGain (stripped in fixture) + # but the field should not be present rather than erroring + assert info["stream_type"] == "nidq" + + +class TestHeaderAP: + def test_stream_type(self, ap_files): + meta_path, _, _, _ = ap_files + info = header(meta_path) + assert info["stream_type"] == "ap" + + def test_sample_rate(self, ap_files): + meta_path, _, _, _ = ap_files + info = header(meta_path) + assert info["sample_rate"] == 30000.0 + + def test_channel_counts(self, ap_files): + meta_path, _, _, _ = ap_files + info = header(meta_path) + assert info["n_neural_chans"] == 384 + assert info["n_sync_chans"] == 1 + assert info["n_saved_chans"] == 385 + + def test_digital_lines(self, ap_files): + meta_path, _, _, _ = ap_files + info = header(meta_path) + assert info["n_digital_lines"] == 16 # 16 bits * 1 sync chan + assert info["n_digital_word_cols"] == 1 + assert info["digital_line_label"][0] == "SY0.0" + assert info["digital_line_label"][6] == "SY0.6" # SMA sync input + + def test_voltage_range(self, ap_files): + meta_path, _, _, _ = ap_files + info = header(meta_path) + assert info["voltage_range"] == (-0.6, 0.6) + + +# --------------------------------------------------------------------------- +# Reader: filenamefromepochfiles tests +# --------------------------------------------------------------------------- + + +class TestFilenamefromepochfiles: + def test_nidq_bin_derives_meta(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + result = reader.filenamefromepochfiles([bin_path, meta_path]) + assert result == meta_path + + def test_nidq_bin_only(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + # .bin file present, .meta on disk but not in list + result = reader.filenamefromepochfiles([bin_path]) + assert result.endswith(".nidq.meta") + + def test_ap_meta_only(self, ap_files): + meta_path, _, _, _ = ap_files + reader = ndr_reader_neuropixelsGLX() + result = reader.filenamefromepochfiles([meta_path]) + assert result == meta_path + + def test_ap_bin_derives_meta(self, ap_files): + meta_path, bin_path, _, _ = ap_files + reader = ndr_reader_neuropixelsGLX() + result = reader.filenamefromepochfiles([bin_path, meta_path]) + assert result == meta_path + + def test_no_files_raises(self): + reader = ndr_reader_neuropixelsGLX() + with pytest.raises(FileNotFoundError, match="No .meta file"): + reader.filenamefromepochfiles(["/path/to/data.txt"]) + + def test_multiple_meta_no_bin_raises(self, tmp_path): + reader = ndr_reader_neuropixelsGLX() + m1 = str(tmp_path / "a.ap.meta") + m2 = str(tmp_path / "b.nidq.meta") + Path(m1).touch() + Path(m2).touch() + with pytest.raises(ValueError, match="Multiple .meta files"): + reader.filenamefromepochfiles([m1, m2]) + + def test_bin_with_missing_meta_raises(self, tmp_path): + reader = ndr_reader_neuropixelsGLX() + binf = str(tmp_path / "missing.nidq.bin") + Path(binf).touch() + with pytest.raises(FileNotFoundError, match="No companion .meta"): + reader.filenamefromepochfiles([binf]) + + +# --------------------------------------------------------------------------- +# Reader: getchannelsepoch tests +# --------------------------------------------------------------------------- + + +class TestGetchannelsepochNIDQ: + def test_channel_count(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + channels = reader.getchannelsepoch([bin_path, meta_path]) + # 1 time + 8 analog + 8 digital = 17 + assert len(channels) == 17 + + def test_channel_types(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + channels = reader.getchannelsepoch([bin_path, meta_path]) + types = [ch["type"] for ch in channels] + assert types.count("time") == 1 + assert types.count("analog_in") == 8 + assert types.count("digital_in") == 8 + + def test_channel_names(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + channels = reader.getchannelsepoch([bin_path, meta_path]) + names = [ch["name"] for ch in channels] + assert names[0] == "t1" + assert names[1] == "ai1" + assert names[8] == "ai8" + assert names[9] == "di1" + assert names[16] == "di8" + + +class TestGetchannelsepochAP: + def test_channel_count(self, ap_files): + meta_path, bin_path, _, _ = ap_files + reader = ndr_reader_neuropixelsGLX() + channels = reader.getchannelsepoch([bin_path, meta_path]) + # 1 time + 384 analog + 16 digital = 401 + assert len(channels) == 401 + + def test_channel_types(self, ap_files): + meta_path, bin_path, _, _ = ap_files + reader = ndr_reader_neuropixelsGLX() + channels = reader.getchannelsepoch([bin_path, meta_path]) + types = [ch["type"] for ch in channels] + assert types.count("time") == 1 + assert types.count("analog_in") == 384 + assert types.count("digital_in") == 16 + + +# --------------------------------------------------------------------------- +# Reader: samplerate tests +# --------------------------------------------------------------------------- + + +class TestSamplerate: + def test_nidq_samplerate(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + sr = reader.samplerate([bin_path, meta_path], 1, "analog_in", 1) + assert sr == pytest.approx(10593.220339) + + def test_ap_samplerate(self, ap_files): + meta_path, bin_path, _, _ = ap_files + reader = ndr_reader_neuropixelsGLX() + sr = reader.samplerate([bin_path, meta_path], 1, "analog_in", 1) + assert sr == 30000.0 + + def test_samplerate_array(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + sr = reader.samplerate([bin_path, meta_path], 1, "analog_in", [1, 2, 3]) + assert len(sr) == 3 + np.testing.assert_allclose(sr, 10593.220339) + + +# --------------------------------------------------------------------------- +# Reader: t0_t1 tests +# --------------------------------------------------------------------------- + + +class TestT0T1: + def test_nidq_t0_t1(self, nidq_files): + meta_path, bin_path, n_chans, n_samples = nidq_files + reader = ndr_reader_neuropixelsGLX() + t0t1 = reader.t0_t1([bin_path, meta_path]) + assert len(t0t1) == 1 + assert t0t1[0][0] == 0.0 + expected_t_end = (n_samples - 1) / 10593.220339 + assert t0t1[0][1] == pytest.approx(expected_t_end, rel=1e-6) + + +# --------------------------------------------------------------------------- +# Reader: readchannels_epochsamples tests +# --------------------------------------------------------------------------- + + +class TestReadchannelsNIDQ: + def test_read_analog(self, nidq_files): + meta_path, bin_path, _, n_samples = nidq_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + data = reader.readchannels_epochsamples("analog_in", [1], files, 1, 1, 10) + assert data.shape == (10, 1) + assert data.dtype == np.int16 + # First analog channel was filled with arange(1, n_samples+1)*1 + np.testing.assert_array_equal(data[:, 0], np.arange(1, 11, dtype=np.int16)) + + def test_read_multiple_analog(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + data = reader.readchannels_epochsamples("analog_in", [1, 3], files, 1, 1, 5) + assert data.shape == (5, 2) + # Channel 1: arange*1, Channel 3: arange*3 + np.testing.assert_array_equal(data[:, 0], np.arange(1, 6, dtype=np.int16)) + np.testing.assert_array_equal(data[:, 1], np.arange(1, 6, dtype=np.int16) * 3) + + def test_read_digital_bit0(self, nidq_files): + """Bit 0 is set on odd samples (0-indexed: s=1,3,5,...).""" + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + # Read samples 1-8 (1-based) + data = reader.readchannels_epochsamples("digital_in", [1], files, 1, 1, 8) + assert data.shape == (8, 1) + # 0-indexed samples 0..7 → bit0 set at odd indices (1,3,5,7) + # 1-based samples 1..8 → file rows 0..7 + expected = np.array([0, 1, 0, 1, 0, 1, 0, 1], dtype=np.int16) + np.testing.assert_array_equal(data[:, 0], expected) + + def test_read_digital_bit1(self, nidq_files): + """Bit 1 is set on every-4th sample (0-indexed: s=0,4,8,...).""" + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + data = reader.readchannels_epochsamples("digital_in", [2], files, 1, 1, 8) + expected = np.array([1, 0, 0, 0, 1, 0, 0, 0], dtype=np.int16) + np.testing.assert_array_equal(data[:, 0], expected) + + def test_read_digital_bit2(self, nidq_files): + """Bit 2 is set on every-8th sample (0-indexed: s=0,8,...).""" + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + data = reader.readchannels_epochsamples("digital_in", [3], files, 1, 1, 8) + expected = np.array([1, 0, 0, 0, 0, 0, 0, 0], dtype=np.int16) + np.testing.assert_array_equal(data[:, 0], expected) + + def test_read_digital_multiple_lines(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + data = reader.readchannels_epochsamples("digital_in", [1, 2, 3], files, 1, 1, 4) + assert data.shape == (4, 3) + # bit 0 at samples 1-4: [0,1,0,1] + np.testing.assert_array_equal(data[:, 0], [0, 1, 0, 1]) + # bit 1 at samples 1-4: [1,0,0,0] + np.testing.assert_array_equal(data[:, 1], [1, 0, 0, 0]) + # bit 2 at samples 1-4: [1,0,0,0] + np.testing.assert_array_equal(data[:, 2], [1, 0, 0, 0]) + + def test_read_digital_out_of_range(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + with pytest.raises(ValueError, match="Digital line out of range"): + reader.readchannels_epochsamples("digital_in", [9], files, 1, 1, 5) + + def test_read_time(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + data = reader.readchannels_epochsamples("time", 1, files, 1, 1, 3) + assert data.shape[0] == 3 + assert data.dtype == np.float64 + + +class TestReadchannelsAP: + def test_read_analog(self, ap_files): + meta_path, bin_path, _, _ = ap_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + data = reader.readchannels_epochsamples("analog_in", [1], files, 1, 1, 5) + assert data.shape == (5, 1) + assert data.dtype == np.int16 + + def test_read_digital_sync_bit6(self, ap_files): + """For AP, digital_in di7 = bit 6 of the sync word (SMA sync input).""" + meta_path, bin_path, _, _ = ap_files + reader = ndr_reader_neuropixelsGLX() + files = [bin_path, meta_path] + data = reader.readchannels_epochsamples("digital_in", [7], files, 1, 1, 5) + assert data.shape == (5, 1) + assert data.dtype == np.int16 + + +# --------------------------------------------------------------------------- +# Reader: epochclock tests +# --------------------------------------------------------------------------- + + +class TestEpochclock: + def test_returns_dev_local_time(self, nidq_files): + meta_path, bin_path, _, _ = nidq_files + reader = ndr_reader_neuropixelsGLX() + ec = reader.epochclock([bin_path, meta_path]) + assert len(ec) == 1 + assert ec[0].type == "dev_local_time" From 4d0c18406324d47a8c0251bd05764b65b9dd3c27 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 13 Apr 2026 12:38:48 +0000 Subject: [PATCH 2/2] Fix CI: fall back to main when NDR-matlab branch doesn't exist The test-symmetry workflow tried to check out NDR-matlab at the same branch name as the PR, failing when no matching branch existed. Add a resolve step that checks if the branch exists via git ls-remote and falls back to main. Also fix black formatting and remove unused import. https://claude.ai/code/session_01HEWqAySjSJ6dFeNWJrdsrJ --- .github/workflows/test-symmetry.yml | 17 +++++++++++++---- src/ndr/reader/neuropixelsGLX.py | 4 +--- tests/test_neuropixelsGLX.py | 2 -- 3 files changed, 14 insertions(+), 9 deletions(-) diff --git a/.github/workflows/test-symmetry.yml b/.github/workflows/test-symmetry.yml index 36ade01..cdc1c79 100644 --- a/.github/workflows/test-symmetry.yml +++ b/.github/workflows/test-symmetry.yml @@ -26,15 +26,24 @@ jobs: # Check out NDR-matlab at the same branch name as this workflow run # (e.g. the PR's head branch or the branch being pushed to main). # Developers are expected to keep matching branch names across the - # two repos when making cross-language changes, the same way the - # symmetry branches work. Falls back to main via the default so - # `main` pushes also work. + # two repos when making cross-language changes. Falls back to main + # when the matching branch does not exist in NDR-matlab. + - name: Resolve NDR-matlab branch + id: matlab-ref + run: | + BRANCH="${{ github.head_ref || github.ref_name }}" + if git ls-remote --exit-code --heads https://github.com/VH-Lab/NDR-matlab.git "$BRANCH" >/dev/null 2>&1; then + echo "ref=$BRANCH" >> "$GITHUB_OUTPUT" + else + echo "ref=main" >> "$GITHUB_OUTPUT" + fi + - name: Check out NDR-matlab (matching branch) uses: actions/checkout@v4 with: repository: VH-Lab/NDR-matlab path: NDR-matlab - ref: ${{ github.head_ref || github.ref_name }} + ref: ${{ steps.matlab-ref.outputs.ref }} # -- Runtime setup ---------------------------------------------------- - name: Start virtual display server diff --git a/src/ndr/reader/neuropixelsGLX.py b/src/ndr/reader/neuropixelsGLX.py index 2a19001..8169d5a 100644 --- a/src/ndr/reader/neuropixelsGLX.py +++ b/src/ndr/reader/neuropixelsGLX.py @@ -300,9 +300,7 @@ def filenamefromepochfiles(self, filename_array: list[str]) -> str: if len(meta_matches) == 0: raise FileNotFoundError("No .meta file found in the epoch file list.") if len(meta_matches) > 1: - raise ValueError( - "Multiple .meta files found and no .bin file to disambiguate." - ) + raise ValueError("Multiple .meta files found and no .bin file to disambiguate.") return meta_matches[0] def daqchannels2internalchannels( diff --git a/tests/test_neuropixelsGLX.py b/tests/test_neuropixelsGLX.py index 497d98c..322d903 100644 --- a/tests/test_neuropixelsGLX.py +++ b/tests/test_neuropixelsGLX.py @@ -2,7 +2,6 @@ from __future__ import annotations -import struct from pathlib import Path import numpy as np @@ -11,7 +10,6 @@ from ndr.format.neuropixelsGLX.header import header from ndr.reader.neuropixelsGLX import ndr_reader_neuropixelsGLX - # --------------------------------------------------------------------------- # Fixtures: create minimal .meta / .bin files on disk # ---------------------------------------------------------------------------