1
|
|
|
import unittest |
2
|
|
|
from mock import Mock, patch |
3
|
|
|
|
4
|
|
|
|
5
|
|
|
class MnefunTests(unittest.TestCase): |
6
|
|
|
|
7
|
|
|
def setUp(self): |
8
|
|
|
self.listener = Mock() |
9
|
|
|
self.libs = Mock() |
10
|
|
|
self.dependencies = Mock() |
11
|
|
|
self.dependencies.getListener.return_value = self.listener |
12
|
|
|
self.dependencies.getLibraries.return_value = self.libs |
13
|
|
|
|
14
|
|
|
@unittest.skip("Needs to be fixed for new mnefun support code.") |
15
|
|
|
def test_Discovers_fetched_raw_files(self): |
16
|
|
|
import niprov.mnefunsupport |
17
|
|
|
class MockParams(object): |
18
|
|
|
pass |
19
|
|
|
params = MockParams() |
20
|
|
|
params.work_dir = '/root' |
21
|
|
|
params.subjects = ['johndoe','janedoe'] |
22
|
|
|
params.raw_dir = 'rawdir' |
23
|
|
|
def fetch_raw_files(): |
24
|
|
|
pass |
25
|
|
|
with patch('niprov.mnefunsupport.discover') as discover: |
26
|
|
|
niprov.mnefunsupport.handler('Pulling raw files from acquisition machine', |
27
|
|
|
fetch_raw_files, None, params, dependencies=self.dependencies) |
28
|
|
|
discover.assert_any_call('/root/johndoe/rawdir') |
29
|
|
|
discover.assert_any_call('/root/janedoe/rawdir') |
30
|
|
|
self.listener.mnefunEventReceived.assert_called_with('fetch_raw_files') |
31
|
|
|
|
32
|
|
View Code Duplication |
@unittest.skip("Needs to be fixed for new mnefun support code.") |
|
|
|
|
33
|
|
|
def test_Logs_sss_operation(self): |
34
|
|
|
import niprov.mnefunsupport |
35
|
|
|
class MockParams(object): |
36
|
|
|
pass |
37
|
|
|
params = MockParams() |
38
|
|
|
params.subjects = ['s1','s2'] |
39
|
|
|
fnames = {'raw':{'s1':['subj 1 raw file 1','subj 1 raw file 2'], |
40
|
|
|
's2':['subj 2 raw file 1','subj 2 raw file 2']}, |
41
|
|
|
'sss': {'s1':['subj 1 sss file 1','subj 1 sss file 2'], |
42
|
|
|
's2':['subj 2 sss file 1','subj 2 sss file 2']}} |
43
|
|
|
self.libs.mnefun.get_raw_fnames.side_effect = lambda p, s, t: fnames[t][s] |
44
|
|
|
def fetch_sss_files(): |
45
|
|
|
pass |
46
|
|
|
with patch('niprov.mnefunsupport.log') as log: |
47
|
|
|
niprov.mnefunsupport.handler('Pulling SSS files from remote workstation', |
48
|
|
|
fetch_sss_files, None, params, dependencies=self.dependencies) |
49
|
|
|
log.assert_any_call(fnames['sss']['s1'][0], |
50
|
|
|
'Signal Space Separation', |
51
|
|
|
fnames['raw']['s1'][0], provenance = {'mnefun':{}}) |
52
|
|
|
log.assert_any_call(fnames['sss']['s2'][1], |
53
|
|
|
'Signal Space Separation', |
54
|
|
|
fnames['raw']['s2'][1], provenance = {'mnefun':{}}) |
55
|
|
|
|
56
|
|
View Code Duplication |
@unittest.skip("Needs to be fixed for new mnefun support code.") |
|
|
|
|
57
|
|
|
def test_Logs_ssp_operation(self): |
58
|
|
|
import niprov.mnefunsupport |
59
|
|
|
class MockParams(object): |
60
|
|
|
pass |
61
|
|
|
params = MockParams() |
62
|
|
|
params.subjects = ['s1','s2'] |
63
|
|
|
fnames = {'pca':{'s1':['subj 1 pca file 1','subj 1 pca file 2'], |
64
|
|
|
's2':['subj 2 pca file 1','subj 2 pca file 2']}, |
65
|
|
|
'sss': {'s1':['subj 1 sss file 1','subj 1 sss file 2'], |
66
|
|
|
's2':['subj 2 sss file 1','subj 2 sss file 2']}} |
67
|
|
|
self.libs.mnefun.get_raw_fnames.side_effect = lambda p, s, t: fnames[t][s] |
68
|
|
|
def apply_preprocessing_combined(): |
69
|
|
|
pass |
70
|
|
|
with patch('niprov.mnefunsupport.log') as log: |
71
|
|
|
niprov.mnefunsupport.handler('Apply SSP vectors and filtering.', |
72
|
|
|
apply_preprocessing_combined, None, params, |
73
|
|
|
dependencies=self.dependencies) |
74
|
|
|
log.assert_any_call(fnames['pca']['s1'][0], |
75
|
|
|
'Signal Space Projection', |
76
|
|
|
fnames['sss']['s1'][0], provenance = {'mnefun':{}}) |
77
|
|
|
log.assert_any_call(fnames['pca']['s2'][1], |
78
|
|
|
'Signal Space Projection', |
79
|
|
|
fnames['sss']['s2'][1], provenance = {'mnefun':{}}) |
80
|
|
|
|
81
|
|
|
@unittest.skip("Needs to be fixed for new mnefun support code.") |
82
|
|
|
def test_Logs_epoch_operation(self): |
83
|
|
|
import niprov.mnefunsupport |
84
|
|
|
class MockParams(object): |
85
|
|
|
pass |
86
|
|
|
params = MockParams() |
87
|
|
|
params.subjects = ['s1','s2'] |
88
|
|
|
params.analyses = ['a'] |
89
|
|
|
fnames = {'pca':{'s1':['subj 1 pca file 1','subj 1 pca file 2'], |
90
|
|
|
's2':['subj 2 pca file 1','subj 2 pca file 2']}} |
91
|
|
|
epochfnames = {'s1':['s1 evt 1','s1 evt 2','s1 evt 3'], |
92
|
|
|
's2':['s2 evt 1','s2 evt 2','s2 evt 3']} |
93
|
|
|
self.libs.mnefun.get_raw_fnames.side_effect = \ |
94
|
|
|
lambda p, s, t: fnames[t][s] |
95
|
|
|
self.libs.mnefun._paths.get_epochs_evokeds_fnames.side_effect = \ |
96
|
|
|
lambda p, s, a: epochfnames[s] |
97
|
|
|
def save_epochs(): |
98
|
|
|
pass |
99
|
|
|
with patch('niprov.mnefunsupport.log') as log: |
100
|
|
|
niprov.mnefunsupport.handler('Doing epoch EQ/DQ', |
101
|
|
|
save_epochs, None, params, dependencies=self.dependencies) |
102
|
|
|
log.assert_any_call(epochfnames['s1'][0], #any event file |
103
|
|
|
'Epoching', |
104
|
|
|
fnames['pca']['s1'], provenance = {'mnefun':{}}) #all raw files for subj |
105
|
|
|
log.assert_any_call(epochfnames['s2'][1], #any event file |
106
|
|
|
'Epoching', |
107
|
|
|
fnames['pca']['s2'], provenance = {'mnefun':{}}) #all raw files for subj |
108
|
|
|
|
109
|
|
|
@unittest.skip("Needs to be fixed for new mnefun support code.") |
110
|
|
|
def test_Log_seeded_with_params_based_custom_provenance(self): |
111
|
|
|
import niprov.mnefunsupport |
112
|
|
|
class MockParams(object): |
113
|
|
|
pass |
114
|
|
|
params = MockParams() |
115
|
|
|
params.tmin = -0.2 |
116
|
|
|
params.quat_tol = 5e-2 |
117
|
|
|
params.subjects = ['s1','s2'] |
118
|
|
|
fnames = {'raw':{'s1':['subj 1 raw file 1','subj 1 raw file 2'], |
119
|
|
|
's2':['subj 2 raw file 1','subj 2 raw file 2']}, |
120
|
|
|
'sss': {'s1':['subj 1 sss file 1','subj 1 sss file 2'], |
121
|
|
|
's2':['subj 2 sss file 1','subj 2 sss file 2']}} |
122
|
|
|
self.libs.mnefun.get_raw_fnames.side_effect = lambda p, s, t: fnames[t][s] |
123
|
|
|
def fetch_sss_files(): |
124
|
|
|
pass |
125
|
|
|
with patch('niprov.mnefunsupport.log') as log: |
126
|
|
|
niprov.mnefunsupport.handler('Pulling SSS files from remote workstation', |
127
|
|
|
fetch_sss_files, None, params, dependencies=self.dependencies) |
128
|
|
|
log.assert_called_with(fnames['sss']['s2'][1], |
129
|
|
|
'Signal Space Separation', |
130
|
|
|
fnames['raw']['s2'][1], |
131
|
|
|
provenance={'mnefun':{ |
132
|
|
|
'tmin':-0.2, |
133
|
|
|
'quat_tol':5e-2}}) |
134
|
|
|
|
135
|
|
|
|
136
|
|
|
# write_epochs : bool |
137
|
|
|
# Write epochs to disk. |
138
|
|
|
# gen_covs : bool |
139
|
|
|
# Generate covariances. |
140
|
|
|
# gen_fwd : bool |
141
|
|
|
# Generate forward solutions. |
142
|
|
|
# get_inv : bool |
143
|
|
|
# Generate inverses. |
144
|
|
|
# gen_report : bool |
145
|
|
|
# Generate HTML reports. |
146
|
|
|
|
147
|
|
|
|