diff --git a/neo/rawio/intanrawio.py b/neo/rawio/intanrawio.py index 00a7a2035..1a7ab0212 100644 --- a/neo/rawio/intanrawio.py +++ b/neo/rawio/intanrawio.py @@ -190,16 +190,18 @@ def _parse_header(self): elif self.file_format == "one-file-per-signal": self._raw_data = {} for stream_name, stream_dtype in memmap_data_dtype.items(): - num_channels = channel_number_dict[stream_name] + # Digital streams pack all channels into one 16-bit word per sample (see #1853). + stream_is_digital = stream_name in digital_stream_names + file_channels = 1 if stream_is_digital else channel_number_dict[stream_name] file_path = raw_file_paths_dict[stream_name] size_in_bytes = file_path.stat().st_size dtype_size = np.dtype(stream_dtype).itemsize - n_samples = size_in_bytes // (dtype_size * num_channels) + n_samples = size_in_bytes // (dtype_size * file_channels) signal_stream_memmap = np.memmap( file_path, dtype=stream_dtype, mode="r", - shape=(n_samples, num_channels), + shape=(n_samples, file_channels), order="C", ) self._raw_data[stream_name] = signal_stream_memmap diff --git a/neo/test/rawiotest/test_intanrawio.py b/neo/test/rawiotest/test_intanrawio.py index e97966242..186c10073 100644 --- a/neo/test/rawiotest/test_intanrawio.py +++ b/neo/test/rawiotest/test_intanrawio.py @@ -219,5 +219,25 @@ def test_correct_decoding_of_stimulus_current(self): assert np.isclose(duration_of_positive_pulse, expected_duration) +class TestIntanDigitalDemultiplexShape(unittest.TestCase): + """Regression coverage for https://github.com/NeuralEnsemble/python-neo/issues/1853.""" + + def test_demultiplex_handles_packed_digital_buffer(self): + # Digital streams pack all channels into one uint16 word per timestamp. + n_samples = 8 + packed = np.array([0, 1, 16, 17, 0, 16, 1, 17], dtype=np.uint16).reshape(n_samples, 1) + expected_bit0 = np.array([0, 1, 0, 1, 0, 0, 1, 1], dtype=np.uint16) + expected_bit4 = np.array([0, 0, 1, 1, 0, 1, 0, 1], dtype=np.uint16) + + reader = IntanRawIO.__new__(IntanRawIO) + reader.native_channel_order = {"DIN-00": 0, "DIN-04": 4} + + chunk = reader._demultiplex_digital_data(packed, ["DIN-00", "DIN-04"], 0, n_samples) + + self.assertEqual(chunk.shape, (n_samples, 2)) + np.testing.assert_array_equal(chunk[:, 0], expected_bit0) + np.testing.assert_array_equal(chunk[:, 1], expected_bit4) + + if __name__ == "__main__": unittest.main()