diff --git a/neo/rawio/micromedrawio.py b/neo/rawio/micromedrawio.py index 018783574..651d58eac 100644 --- a/neo/rawio/micromedrawio.py +++ b/neo/rawio/micromedrawio.py @@ -68,7 +68,7 @@ def _parse_header(self): # header version (header_version,) = f.read_f("b", offset=175) - if header_version != 4: + if header_version not in (4, 5): raise NotImplementedError(f"`header_version {header_version} is not implemented in neo yet") # area @@ -94,7 +94,8 @@ def _parse_header(self): for zname in zone_names: zname2, pos, length = f.read_f("8sII") zones[zname] = zname2, pos, length - if zname != zname2.decode("ascii").strip(" "): + # zone names are padded with spaces in header version 4 and with NULs in version 5 + if zname != zname2.decode("ascii").strip(" \x00"): raise NeoReadWriteError("expected the zone name to match") # "TRONCA" zone define segments @@ -115,6 +116,10 @@ def _parse_header(self): if len(self.info_segments) == 0: # one unique segment = general case self.info_segments.append((0, 0)) + elif self.info_segments[0][1] > 0: + # The table lists the segments after the first cut. When its first trace_offset is above 0 the + # samples before it are a segment of their own, where the recording clock and the file agree. + self.info_segments.insert(0, (0, 0)) nb_segment = len(self.info_segments) @@ -214,7 +219,7 @@ def _parse_header(self): dtype = np.dtype(ev_dtype) rawevent = np.memmap(self.filename, dtype=dtype, mode="r", offset=pos, shape=length // dtype.itemsize) - # important : all events timing are related to the first segment t_start + # event positions are indices in the data block, like trace_offset (see _get_event_timestamps) self._raw_events.append([]) for seg_index in range(nb_segment): left_lim = seg_limits[seg_index] @@ -274,18 +279,18 @@ def _event_count(self, block_index, seg_index, event_channel_index): def _get_event_timestamps(self, block_index, seg_index, event_channel_index, t_start, t_stop): raw_event = self._raw_events[event_channel_index][seg_index] - # important : all events timing are related to the first segment t_start - seg_start0, _ = self.info_segments[0] + # An event start is an index in the data block, the coordinate of trace_offset, and the time axis of a segment + # starts at its seg_start: the sample at index s of segment k is at clock sample seg_start + s - trace_offset. + seg_start, trace_offset = self.info_segments[seg_index] + timestamp = raw_event["start"].astype("int64") + (seg_start - trace_offset) if t_start is not None: - keep = raw_event["start"] + seg_start0 >= int(t_start * self._sampling_rate) - raw_event = raw_event[keep] + keep = timestamp >= int(t_start * self._sampling_rate) + raw_event, timestamp = raw_event[keep], timestamp[keep] if t_stop is not None: - keep = raw_event["start"] + seg_start0 <= int(t_stop * self._sampling_rate) - raw_event = raw_event[keep] - - timestamp = raw_event["start"] + seg_start0 + keep = timestamp <= int(t_stop * self._sampling_rate) + raw_event, timestamp = raw_event[keep], timestamp[keep] if event_channel_index < 2: durations = None diff --git a/neo/test/rawiotest/test_micromedrawio.py b/neo/test/rawiotest/test_micromedrawio.py index 1c8262860..d5f144d98 100644 --- a/neo/test/rawiotest/test_micromedrawio.py +++ b/neo/test/rawiotest/test_micromedrawio.py @@ -1,7 +1,12 @@ """ """ +import struct +import tempfile import unittest +from pathlib import Path +from neo.core import NeoReadWriteError +from neo.io import MicromedIO from neo.rawio.micromedrawio import MicromedRawIO from neo.test.rawiotest.common_rawio_test import BaseTestRawIO @@ -52,5 +57,265 @@ def test_micromed_multi_segments(self): assert np.array_equal(traces2[:-100], traces1_chunk[:-100]) +ZONE_NAMES = [ + "ORDER", + "LABCOD", + "NOTE", + "FLAGS", + "TRONCA", + "IMPED_B", + "IMPED_E", + "MONTAGE", + "COMPRESS", + "AVERAGE", + "HISTORY", + "DVIDEO", + "EVENT A", + "EVENT B", + "TRIGGER", +] + + +def write_micromed_trc( + path, codes, sampling_rate=256, header_version=4, tronca=(), notes=(), triggers=(), epochs=(), zone_names=None +): + """ + Write a minimal Micromed .trc file, for tests that do not need a real recording. + + The header has the 15 zones neo reads, and no patient data (the name and the other free text fields stay empty). + + Parameters + ---------- + path: str or Path + The file to write. + codes: np.ndarray of shape (n_samples, n_channels) + The ADC codes (unsigned 16 bits) of the data block. The channels are in uV, with the logical ground at 32768. + sampling_rate: int, default: 256 + Rate_Min of the header, and the rate of every channel. + header_version: int, default: 4 + Version 4 pads the zone names with spaces, version 5 with NULs. + tronca: list of (seg_start, trace_offset) + The TRONCA table: the sample at index trace_offset of the data block is at sample seg_start of the recording clock. + notes: list of (start, text) + NOTE events, start is an index in the data block. + triggers: list of (start, label) + TRIGGER events, start is an index in the data block. + epochs: list of (label, start, stop) + EVENT A entries, start and stop are indices in the data block. + zone_names: dict, default: None + Names to write in the zone table instead of the standard ones, by standard name. + """ + codes = np.asarray(codes, dtype="