Skip to content
Open
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
8 changes: 5 additions & 3 deletions neo/rawio/intanrawio.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
20 changes: 20 additions & 0 deletions neo/test/rawiotest/test_intanrawio.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Loading