-
Notifications
You must be signed in to change notification settings - Fork 66
/
03-extract_events.py
87 lines (65 loc) · 2.82 KB
/
03-extract_events.py
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
78
79
80
81
82
83
84
85
86
87
"""
============================================
03. Extract events from the stimulus channel
============================================
Here, all events present in the stimulus channel indicated in
config.stim_channel are extracted.
The events are saved to the subject's MEG directory.
This is done early in the pipeline to avoid distorting event-time,
for instance by resampling.
"""
import os.path as op
import itertools
import numpy as np
import matplotlib.pyplot as plt
import mne
from mne.parallel import parallel_func
from mne_bids import make_bids_basename
import config
def run_events(subject, run=None, session=None):
"""Extract events and save as fif."""
print("Processing subject: %s" % subject)
# Construct the search path for the data file. `sub` is mandatory
subject_path = op.join('sub-{}'.format(subject))
# `session` is optional
if session is not None:
subject_path = op.join(subject_path, 'ses-{}'.format(session))
subject_path = op.join(subject_path, config.kind)
bids_basename = make_bids_basename(subject=subject,
session=session,
task=config.task,
acquisition=config.acq,
run=run,
processing=config.proc,
recording=config.rec,
space=config.space
)
# Prepare a name to save the data
fpath_deriv = op.join(config.bids_root, 'derivatives',
config.PIPELINE_NAME, subject_path)
raw_fname_in = \
op.join(fpath_deriv, bids_basename + '_filt_raw.fif')
eve_fname_out = op.join(fpath_deriv, bids_basename + '-eve.fif')
raw = mne.io.read_raw_fif(raw_fname_in)
events, event_id = mne.events_from_annotations(raw)
if config.trigger_time_shift:
events = mne.event.shift_time_events(events,
np.unique(events[:, 2]),
config.trigger_time_shift,
raw.info['sfreq'])
print("Input: ", raw_fname_in)
print("Output: ", eve_fname_out)
mne.write_events(eve_fname_out, events)
if config.plot:
# plot events
mne.viz.plot_events(events, sfreq=raw.info['sfreq'],
first_samp=raw.first_samp)
plt.show()
def main():
"""Run events."""
parallel, run_func, _ = parallel_func(run_events, n_jobs=config.N_JOBS)
parallel(run_func(subject, run, session) for subject, run, session in
itertools.product(config.subjects_list, config.runs,
config.sessions))
if __name__ == '__main__':
main()