EDFRawIO: treat annotations without a duration as events

neo/rawio/edfrawio.py

@@ -184,7 +184,10 @@ def _parse_header(self):
             stream_index: self.edf_reader.getNSamples()[chidx][0]
             for stream_index, chidx in self.stream_idx_to_chidx.items()
         }
-        self._number_of_events = len(self.edf_reader.readAnnotations()[0])
+        # pyedflib reports a missing duration as -1, so anything without a positive duration is an event
+        annotation_durations = np.asarray(self._edf_annotations[1])
+        self._number_of_events = int(np.sum(annotation_durations <= 0))
+        self._number_of_epochs = int(np.sum(annotation_durations > 0))
 
         self.close()
 
@@ -258,6 +261,8 @@ def _get_spike_raw_waveforms(self, block_index, seg_index, spike_channel_index,
         return None
 
     def _event_count(self, block_index, seg_index, event_channel_index):
+        if self.header["event_channels"]["type"][event_channel_index] == b"epoch":
+            return self._number_of_epochs
         return self._number_of_events
 
     def _get_event_timestamps(self, block_index, seg_index, event_channel_index, t_start, t_stop):
@@ -269,12 +274,12 @@ def _get_event_timestamps(self, block_index, seg_index, event_channel_index, t_s
             t_stop = self.segment_t_stop(block_index, seg_index)
 
         # only consider events and epochs that overlap with t_start t_stop range
-        time_mask = ((t_start < timestamps) & (timestamps < t_stop)) | (
-            (t_start < (timestamps + durations)) & ((timestamps + durations) < t_stop)
-        )
+        stops = timestamps + np.maximum(durations, 0)
+        time_mask = ((t_start <= timestamps) & (timestamps <= t_stop)) | ((t_start <= stops) & (stops <= t_stop))
 
         # separate event from epoch times
-        event_mask = durations[time_mask] == 0
+        # pyedflib reports a missing duration as -1, so anything without a positive duration is an event
+        event_mask = durations[time_mask] <= 0
         if self.header["event_channels"]["type"][event_channel_index] == b"epoch":
             event_mask = ~event_mask
             durations = durations[time_mask][event_mask]

neo/test/rawiotest/test_edfrawio.py

@@ -2,9 +2,13 @@
 Tests of neo.rawio.edfrawio
 """
 
+import tempfile
 import unittest
+from pathlib import Path
 
-from neo.rawio.edfrawio import EDFRawIO
+import numpy as np
+
+from neo.rawio.edfrawio import EDFRawIO, HAS_PYEDF
 from neo.test.rawiotest.common_rawio_test import BaseTestRawIO
 
 
@@ -41,5 +45,50 @@ def test_close(self):
         io2.close()
 
 
+@unittest.skipUnless(HAS_PYEDF, "requires pyedflib")
+class TestEDFRawIOAnnotations(unittest.TestCase):
+    def test_annotations_without_duration_are_events(self):
+        import pyedflib
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            filename = str(Path(tmp_dir) / "annotations.edf")
+            writer = pyedflib.EdfWriter(filename, 1, file_type=pyedflib.FILETYPE_EDFPLUS)
+            writer.setSignalHeaders(
+                [
+                    dict(
+                        label="ch1",
+                        dimension="uV",
+                        sample_frequency=100,
+                        physical_min=-100,
+                        physical_max=100,
+                        digital_min=-32768,
+                        digital_max=32767,
+                    )
+                ]
+            )
+            writer.writeSamples([np.zeros(1000)])
+            # a duration of -1 writes an EDF+ annotation with no duration field
+            writer.writeAnnotation(0, -1, "start")
+            writer.writeAnnotation(2.5, -1, "stim")
+            writer.writeAnnotation(0, 1.5, "baseline")
+            writer.close()
+
+            with EDFRawIO(filename) as io:
+                io.parse_header()
+                event_index, epoch_index = 0, 1
+
+                self.assertEqual(io.event_count(event_channel_index=event_index), 2)
+                times, durations, labels = io.get_event_timestamps(event_channel_index=event_index)
+                np.testing.assert_array_equal(times, [0.0, 2.5])
+                self.assertIsNone(durations)
+                np.testing.assert_array_equal(labels, ["start", "stim"])
+
+                self.assertEqual(io.event_count(event_channel_index=epoch_index), 1)
+                times, durations, labels = io.get_event_timestamps(event_channel_index=epoch_index)
+                np.testing.assert_array_equal(times, [0.0])
+                np.testing.assert_array_equal(durations, [1.5])
+                np.testing.assert_array_equal(labels, ["baseline"])
+
+
 if __name__ == "__main__":
     unittest.main()