forked from mne-tools/mne-bids-pipeline
-
Notifications
You must be signed in to change notification settings - Fork 0
/
05b-apply_ssp.py
84 lines (60 loc) · 2.63 KB
/
05b-apply_ssp.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
"""
===============
06b. Apply SSP
===============
Blinks and ECG artifacts are automatically detected and the corresponding SSP
projections components are removed from the data.
"""
import os.path as op
import itertools
import logging
import mne
from mne.parallel import parallel_func
from mne_bids import make_bids_basename
import config
from config import gen_log_message, on_error, failsafe_run
logger = logging.getLogger('mne-study-template')
@failsafe_run(on_error=on_error)
def apply_ssp(subject, session=None):
# load epochs to reject ICA components
# compute SSP on first run of raw
deriv_path = config.get_subject_deriv_path(subject=subject,
session=session,
kind=config.get_kind())
bids_basename = make_bids_basename(subject=subject,
session=session,
task=config.get_task(),
acquisition=config.acq,
run=None,
processing=config.proc,
recording=config.rec,
space=config.space)
fname_in = op.join(deriv_path, bids_basename + '-epo.fif')
fname_out = op.join(deriv_path, bids_basename + '_cleaned-epo.fif')
epochs = mne.read_epochs(fname_in, preload=True)
msg = f'Input: {fname_in}, Output: {fname_out}'
logger.info(gen_log_message(message=msg, step=5, subject=subject,
session=session))
proj_fname_in = op.join(deriv_path, bids_basename + '_ssp-proj.fif')
msg = f'Reading SSP projections from : {proj_fname_in}'
logger.info(gen_log_message(message=msg, step=5, subject=subject,
session=session))
projs = mne.read_proj(proj_fname_in)
epochs.add_proj(projs).apply_proj()
msg = 'Saving epochs'
logger.info(gen_log_message(message=msg, step=5, subject=subject,
session=session))
epochs.save(fname_out, overwrite=True)
def main():
"""Apply ssp."""
if not config.use_ssp:
return
msg = 'Running Step 5: Apply SSP'
logger.info(gen_log_message(step=5, message=msg))
parallel, run_func, _ = parallel_func(apply_ssp, n_jobs=config.N_JOBS)
parallel(run_func(subject, session) for subject, session in
itertools.product(config.get_subjects(), config.get_sessions()))
msg = 'Completed Step 5: Apply SSP'
logger.info(gen_log_message(step=5, message=msg))
if __name__ == '__main__':
main()