From 5a653d5c45dd5d01fd116cf0eadf15b040e9184b Mon Sep 17 00:00:00 2001 From: aaronkanzer Date: Mon, 9 Nov 2026 16:33:04 -0500 Subject: [PATCH] Treat EDF+ annotations without a duration as events pyedflib reports a missing annotation duration as -1. EDFRawIO only treated a duration of exactly 0 as an event, so every instantaneous EDF+ annotation ended up on the epoch channel with a duration of -1 s and the event channel stayed empty. The time window also used strict bounds, which dropped annotations at t=0, and event_count returned the total number of annotations for both channels. --- neo/rawio/edfrawio.py | 15 ++++++--- neo/test/rawiotest/test_edfrawio.py | 51 ++++++++++++++++++++++++++++- 2 files changed, 60 insertions(+), 6 deletions(-) diff --git a/neo/rawio/edfrawio.py b/neo/rawio/edfrawio.py index 79cc62eb8..3b69124b0 100644 --- a/neo/rawio/edfrawio.py +++ b/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] diff --git a/neo/test/rawiotest/test_edfrawio.py b/neo/test/rawiotest/test_edfrawio.py index 4b5a5e9bf..c5e90ff4d 100644 --- a/neo/test/rawiotest/test_edfrawio.py +++ b/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()