Fix neuroexplorer version 106

neo/rawio/neuroexplorerrawio.py

@@ -60,6 +60,9 @@ def _parse_header(self):
 
         self._memmap = np.memmap(self.filename, dtype="u1", mode="r")
 
+        # NexFile.h: PrethresholdTimeInSeconds can only be non-zero from file version 106 on
+        file_version = self.global_header["version"]
+
         self._sig_lengths = []
         self._sig_t_starts = []
         sig_channels = []
@@ -82,8 +85,15 @@ def _parse_header(self):
                 wf_units = "mV"
                 wf_gain = entity_header["ADtoMV"]
                 wf_offset = entity_header["MVOffset"]
-                wf_left_sweep = 0
                 wf_sampling_rate = entity_header["WFrequency"]
+                # PrethresholdTimeInSeconds is the specification's own definition of neo's left
+                # sweep: if a waveform timestamp is t, its first point is at t - Prethreshold.
+                # neo counts the left sweep in samples. The variable's own version gates the
+                # field independently of the file version, 102 being where it is valid.
+                if file_version >= 106 and entity_header["varVersion"] >= 102:
+                    wf_left_sweep = int(round(entity_header["PrethresholdTimeInSeconds"] * wf_sampling_rate))
+                else:
+                    wf_left_sweep = 0
                 spike_channels.append((name, _id, wf_units, wf_gain, wf_offset, wf_left_sweep, wf_sampling_rate))
 
             elif entity_header["type"] == 4:
@@ -324,7 +334,9 @@ def read_as_dict(fid, dtype, offset=None):
     ("NMarkers", "int32"),
     ("MarkerLength", "int32"),
     ("MVOffset", "float64"),
-    ("dummy", "S60"),
+    ("PrethresholdTimeInSeconds", "float64"),
+    # Unused padding that fills the variable header to its fixed 208 bytes
+    ("dummy", "S52"),
 ]
 
 MarkerHeader = [

neo/test/rawiotest/test_neuroexplorerrawio.py

@@ -14,9 +14,10 @@ class TestNeuroExplorerRawIO(
 ):
     rawioclass = NeuroExplorerRawIO
     entities_to_download = ["neuroexplorer"]
-    files_to_download = [
+    entities_to_test = [
         "neuroexplorer/File_neuroexplorer_1.nex",
         "neuroexplorer/File_neuroexplorer_2.nex",
+        "neuroexplorer/nex_version_106/nonzero_prethreshold.nex",
     ]
 
     def test_spike_waveforms(self):
@@ -57,6 +58,40 @@ def test_data_offset_above_two_gigabytes(self):
         parsed = np.frombuffer(bytes(buffer), dtype=entity_dtype)[0]
         assert parsed["offset"] == true_offset
 
+    def test_version_106_prethreshold(self):
+        """A nonzero PrethresholdTimeInSeconds used to make the file unopenable.
+
+        The variable header ends with that double followed by 52 bytes of padding, but the
+        reader declared the last 60 bytes as a single string field and decodes string fields
+        as UTF-8, so a nonzero double raised UnicodeDecodeError before any data was read.
+        The value is the left sweep, which neo counts in samples.
+        """
+        assert np.dtype(EntityHeader).itemsize == 208
+
+        filename = self.get_local_path("neuroexplorer/nex_version_106/nonzero_prethreshold.nex")
+        reader = NeuroExplorerRawIO(filename=filename)
+        reader.parse_header()
+
+        waveform_channels = [channel for channel in reader.header["spike_channels"] if channel["wf_sampling_rate"] > 0]
+        assert len(waveform_channels) == 6
+        for channel in waveform_channels:
+            # 0.0002 s of prethreshold time at 40 kHz, so 8 of the 32 points precede the crossing
+            assert channel["wf_left_sweep"] == 8
+
+    def test_prethreshold_ignored_before_version_106(self):
+        """PrethresholdTimeInSeconds is only meaningful from file version 106.
+
+        Before that the bytes are padding, so a version 104 file must report a left sweep of
+        zero whatever they hold.
+        """
+        filename = self.get_local_path("neuroexplorer/File_neuroexplorer_2.nex")
+        reader = NeuroExplorerRawIO(filename=filename)
+        reader.parse_header()
+        assert reader.global_header["version"] == 104
+
+        names = [channel["name"] for channel in reader.header["spike_channels"]]
+        assert reader.header["spike_channels"][names.index("sig01i_wf")]["wf_left_sweep"] == 0
+
 
 if __name__ == "__main__":
     unittest.main()