# -*- coding: utf-8 -*- """ Class for reading/writing data from micromed (.trc). Inspired by the Matlab code for EEGLAB from Rami K. Niazy. Completed with matlab Guillaume BECQ code. Author: Samuel Garcia """ from __future__ import print_function, division, absolute_import # from __future__ import unicode_literals is not compatible with numpy.dtype both py2 py3 from .baserawio import (BaseRawIO, _signal_channel_dtype, _unit_channel_dtype, _event_channel_dtype) import numpy as np import datetime import os import struct import io class StructFile(io.BufferedReader): def read_f(self, fmt, offset=None): if offset is not None: self.seek(offset) return struct.unpack(fmt, self.read(struct.calcsize(fmt))) class MicromedRawIO(BaseRawIO): """ Class for reading data from micromed (.trc). """ extensions = ['trc', 'TRC'] rawmode = 'one-file' def __init__(self, filename=''): BaseRawIO.__init__(self) self.filename = filename def _parse_header(self): with io.open(self.filename, 'rb') as fid: f = StructFile(fid) # Name f.seek(64) surname = f.read(22).strip(b' ') firstname = f.read(20).strip(b' ') # Date day, month, year, hour, minute, sec = f.read_f('bbbbbb', offset=128) rec_datetime = datetime.datetime(year + 1900, month, day, hour, minute, sec) Data_Start_Offset, Num_Chan, Multiplexer, Rate_Min, Bytes = f.read_f( 'IHHHH', offset=138) # header version header_version, = f.read_f('b', offset=175) assert header_version == 4 # area f.seek(176) zone_names = ['ORDER', 'LABCOD', 'NOTE', 'FLAGS', 'TRONCA', 'IMPED_B', 'IMPED_E', 'MONTAGE', 'COMPRESS', 'AVERAGE', 'HISTORY', 'DVIDEO', 'EVENT A', 'EVENT B', 'TRIGGER'] zones = {} for zname in zone_names: zname2, pos, length = f.read_f('8sII') zones[zname] = zname2, pos, length assert zname == zname2.decode('ascii').strip(' ') # raw signals memmap sig_dtype = 'u' + str(Bytes) self._raw_signals = np.memmap(self.filename, dtype=sig_dtype, mode='r', offset=Data_Start_Offset).reshape(-1, Num_Chan) # Reading Code Info zname2, pos, length = zones['ORDER'] f.seek(pos) code = np.frombuffer(f.read(Num_Chan * 2), dtype='u2') units_code = {-1: 'nV', 0: 'uV', 1: 'mV', 2: 1, 100: 'percent', 101: 'dimensionless', 102: 'dimensionless'} sig_channels = [] sig_grounds = [] for c in range(Num_Chan): zname2, pos, length = zones['LABCOD'] f.seek(pos + code[c] * 128 + 2, 0) chan_name = f.read(6).strip(b"\x00").decode('ascii') ground = f.read(6).strip(b"\x00").decode('ascii') sig_grounds.append(ground) logical_min, logical_max, logical_ground, physical_min, physical_max = f.read_f( 'iiiii') k, = f.read_f('h') units = units_code.get(k, 'uV') factor = float(physical_max - physical_min) / float( logical_max - logical_min + 1) gain = factor offset = -logical_ground * factor f.seek(8, 1) sampling_rate, = f.read_f('H') sampling_rate *= Rate_Min chan_id = c group_id = 0 sig_channels.append((chan_name, chan_id, sampling_rate, sig_dtype, units, gain, offset, group_id)) sig_channels = np.array(sig_channels, dtype=_signal_channel_dtype) assert np.unique(sig_channels['sampling_rate']).size == 1 self._sampling_rate = float(np.unique(sig_channels['sampling_rate'])[0]) # Event channels event_channels = [] event_channels.append(('Trigger', '', 'event')) event_channels.append(('Note', '', 'event')) event_channels.append(('Event A', '', 'epoch')) event_channels.append(('Event B', '', 'epoch')) event_channels = np.array(event_channels, dtype=_event_channel_dtype) # Read trigger and notes self._raw_events = [] ev_dtypes = [('TRIGGER', [('start', 'u4'), ('label', 'u2')]), ('NOTE', [('start', 'u4'), ('label', 'S40')]), ('EVENT A', [('label', 'u4'), ('start', 'u4'), ('stop', 'u4')]), ('EVENT B', [('label', 'u4'), ('start', 'u4'), ('stop', 'u4')]), ] for zname, ev_dtype in ev_dtypes: zname2, pos, length = zones[zname] dtype = np.dtype(ev_dtype) rawevent = np.memmap(self.filename, dtype=dtype, mode='r', offset=pos, shape=length // dtype.itemsize) keep = (rawevent['start'] >= rawevent['start'][0]) & ( rawevent['start'] < self._raw_signals.shape[0]) & ( rawevent['start'] != 0) rawevent = rawevent[keep] self._raw_events.append(rawevent) # No spikes unit_channels = [] unit_channels = np.array(unit_channels, dtype=_unit_channel_dtype) # fille into header dict self.header = {} self.header['nb_block'] = 1 self.header['nb_segment'] = [1] self.header['signal_channels'] = sig_channels self.header['unit_channels'] = unit_channels self.header['event_channels'] = event_channels # insert some annotation at some place self._generate_minimal_annotations() bl_annotations = self.raw_annotations['blocks'][0] seg_annotations = bl_annotations['segments'][0] for d in (bl_annotations, seg_annotations): d['rec_datetime'] = rec_datetime d['firstname'] = firstname d['surname'] = surname d['header_version'] = header_version for c in range(sig_channels.size): anasig_an = seg_annotations['signals'][c] anasig_an['ground'] = sig_grounds[c] channel_an = self.raw_annotations['signal_channels'][c] channel_an['ground'] = sig_grounds[c] def _source_name(self): return self.filename def _segment_t_start(self, block_index, seg_index): return 0. def _segment_t_stop(self, block_index, seg_index): t_stop = self._raw_signals.shape[0] / self._sampling_rate return t_stop def _get_signal_size(self, block_index, seg_index, channel_indexes): return self._raw_signals.shape[0] def _get_signal_t_start(self, block_index, seg_index, channel_indexes): return 0. def _get_analogsignal_chunk(self, block_index, seg_index, i_start, i_stop, channel_indexes): if channel_indexes is None: channel_indexes = slice(channel_indexes) raw_signals = self._raw_signals[slice(i_start, i_stop), channel_indexes] return raw_signals def _spike_count(self, block_index, seg_index, unit_index): return 0 def _event_count(self, block_index, seg_index, event_channel_index): n = self._raw_events[event_channel_index].size return n def _get_event_timestamps(self, block_index, seg_index, event_channel_index, t_start, t_stop): raw_event = self._raw_events[event_channel_index] if t_start is not None: keep = raw_event['start'] >= int(t_start * self._sampling_rate) raw_event = raw_event[keep] if t_stop is not None: keep = raw_event['start'] <= int(t_stop * self._sampling_rate) raw_event = raw_event[keep] timestamp = raw_event['start'] if event_channel_index < 2: durations = None else: durations = raw_event['stop'] - raw_event['start'] labels = raw_event['label'].astype('U') return timestamp, durations, labels def _rescale_event_timestamp(self, event_timestamps, dtype): event_times = event_timestamps.astype(dtype) / self._sampling_rate return event_times def _rescale_epoch_duration(self, raw_duration, dtype): durations = raw_duration.astype(dtype) // self._sampling_rate return durations