Blackrock: read NEV 2.2 spikes on electrodes above 255

neo/rawio/blackrockrawio.py

@@ -2464,14 +2464,24 @@ def _is_set(self, flag, pos):
 # Used to create masks that filter raw data packets by their packet ID field.
 # Single values indicate equality check, tuples (min, max) indicate range check.
 # According to NEV spec: packet IDs < 32768 identify channels, IDs >= 32768 are system events.
+# Packets whose ID is not listed for a spec are not read. Known cases, ignored on purpose:
+# - 0x8000 + channel (e.g. 32769): preview packets of the live protocol (cbproto's "Preview Data Packet
+#   Definitions"), found in 3.0 files written by Central. They carry no neural data.
+# - 0xFFF9: recording events of 3.0 files written by Central (header type 0 start, 1 stop, 2 pause,
+#   3 resume; Central's CNevFile30::AddRecordingEvent). Not used to split segments, which come from
+#   resets and gaps.
 NEV_PACKET_IDENTIFIERS_BY_SPEC = {
     "2.1": {
         "NonNeural": 0,
         "Spikes": (1, 255),  # Packet IDs in this range identify spike events on electrodes
     },
     "2.2": {
         "NonNeural": 0,
-        "Spikes": (1, 255),  # Packet IDs in this range identify spike events on electrodes
+        # Packet IDs in this range identify spike events on electrodes. Wider than 2.1's range:
+        # Ripple Trellis writes 2.2 files with electrode IDs above 255 (its front-end ports are
+        # numbered 1-128, 129-256, 257-384...), and every spike on them was dropped without a
+        # warning (NeuralEnsemble/python-neo#1918). 2048 is the range of 2.3.
+        "Spikes": (1, 2048),
     },
     "2.3": {
         "NonNeural": 0,

neo/test/rawiotest/test_blackrockrawio.py

@@ -2,7 +2,9 @@
 Tests of neo.rawio.examplerawio
 """
 
+import tempfile
 import unittest
+from pathlib import Path
 
 from neo.rawio.blackrockrawio import BlackrockRawIO
 from neo.test.rawiotest.common_rawio_test import BaseTestRawIO
@@ -374,6 +376,41 @@ def test_gap_tolerance_ms_parameter(self):
         segments_strict = reader_strict.segment_count(0)
         self.assertEqual(segments_strict, 3)  #
 
+    def test_nev_22_reads_spikes_on_electrodes_above_255(self):
+        """
+        A FileSpec 2.2 NEV with spikes on an electrode above 255 (Ripple Trellis numbers its front-end ports
+        1-128, 129-256, 257-384...) keeps them, as 2.3 does (NeuralEnsemble/python-neo#1918). Built from neo's
+        2.3 test NEV: marked 2.2, with electrode 1 renumbered 300 in its NEUEVWAV header and spike packets.
+        """
+        source_base = self.get_local_path("blackrock/FileSpec2.3001")
+        original = BlackrockRawIO(filename=source_base, nsx_to_load=None)
+        original.parse_header()
+        units_on_1 = [i for i, (channel, _) in enumerate(original.internal_unit_ids) if channel == 1]
+        self.assertTrue(units_on_1)  # electrode 1 has spikes to move
+        expected = np.concatenate([original.get_spike_timestamps(0, 0, i, None, None) for i in units_on_1])
+
+        data = bytearray(Path(source_base + ".nev").read_bytes())
+        data[9] = 2  # FileSpec 2.3 -> 2.2
+        header_size, packet_size = np.frombuffer(bytes(data[12:20]), dtype="<u4")
+        for at in range(336, int(header_size), 32):  # extended headers follow the 336-byte basic header
+            if data[at : at + 8] == b"NEUEVWAV" and int.from_bytes(data[at + 8 : at + 10], "little") == 1:
+                data[at + 8 : at + 10] = (300).to_bytes(2, "little")
+        # 2.x packets: a uint32 timestamp, then the uint16 packet id
+        for at in range(int(header_size), len(data) - int(packet_size) + 1, int(packet_size)):
+            if int.from_bytes(data[at + 4 : at + 6], "little") == 1:
+                data[at + 4 : at + 6] = (300).to_bytes(2, "little")
+
+        with tempfile.TemporaryDirectory() as folder:
+            base = str(Path(folder) / "trellis_like")
+            Path(base + ".nev").write_bytes(bytes(data))
+            reader = BlackrockRawIO(filename=base, nsx_to_load=None)
+            reader.parse_header()
+        self.assertEqual(reader._nev_spec, "2.2")
+        units_on_300 = [i for i, (channel, _) in enumerate(reader.internal_unit_ids) if channel == 300]
+        self.assertTrue(units_on_300, "no spike channel on electrode 300")
+        got = np.concatenate([reader.get_spike_timestamps(0, 0, i, None, None) for i in units_on_300])
+        assert_equal(np.sort(got), np.sort(expected))
+
 
 if __name__ == "__main__":
     unittest.main()