1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
|
# Author: Joan Massich <mailsik@gmail.com>
#
# License: BSD (3-clause)
import numpy as np
import os.path as op
from io import BytesIO
from ...annotations import Annotations
from .res4 import _read_res4
def _get_markers(fname):
def consume(fid, predicate): # just a consumer to move around conveniently
while(predicate(fid.readline())):
pass
def parse_marker(string): # XXX: there should be a nicer way to do that
data = np.genfromtxt(
BytesIO(string.encode()), dtype=[('trial', int), ('sync', float)])
return int(data['trial']), float(data['sync'])
markers = dict()
with open(fname) as fid:
consume(fid, lambda l: not l.startswith('NUMBER OF MARKERS:'))
num_of_markers = int(fid.readline())
for _ in range(num_of_markers):
consume(fid, lambda l: not l.startswith('NAME:'))
label = fid.readline().strip('\n')
consume(fid, lambda l: not l.startswith('NUMBER OF SAMPLES:'))
n_markers = int(fid.readline())
consume(fid, lambda l: not l.startswith('LIST OF SAMPLES:'))
next(fid) # skip the samples header
markers[label] = [
parse_marker(next(fid)) for _ in range(n_markers)
]
return markers
def _get_res4_info_needed_by_markers(directory):
"""Get required information from CTF res4 information file."""
# we only need 3 values from res4. Maybe we can read them directly instead
# of parsing the entire res4 file.
res4 = _read_res4(directory)
total_offset_duration = res4['pre_trig_pts'] / res4['sfreq']
trial_duration = res4['nsamp'] / res4['sfreq']
return total_offset_duration, trial_duration
def _read_annotations_ctf(directory):
total_offset, trial_duration = _get_res4_info_needed_by_markers(directory)
return _read_annotations_ctf_call(directory, total_offset, trial_duration)
def _read_annotations_ctf_call(directory, total_offset, trial_duration):
fname = op.join(directory, 'MarkerFile.mrk')
if not op.exists(fname):
return Annotations(list(), list(), list(), orig_time=None)
else:
markers = _get_markers(fname)
onset = [synctime + (trialnum * trial_duration) + total_offset
for _, m in markers.items() for (trialnum, synctime) in m]
description = np.concatenate([
np.repeat(label, len(m)) for label, m in markers.items()
])
return Annotations(onset=onset, duration=np.zeros_like(onset),
description=description, orig_time=None)
|
