-
Notifications
You must be signed in to change notification settings - Fork 6
/
Copy pathstep3a_validation_animacy_size.py
156 lines (127 loc) · 5.96 KB
/
step3a_validation_animacy_size.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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
#!/usr/bin/env python3
"""
@ Lina Teichmann
INPUTS:
call from command line with following inputs:
-bids_dir
OUTPUTS:
Runs a linear regression, using the MEG data at every timepoint to predict animacy and size ratings for each image.
NOTES:
The plot was made in matlab so it looks the same as the decoding plots (see Step3aa)
If the output directory does not exist, this script makes an output folder in BIDS/derivatives
"""
import numpy as np
import mne,os,itertools,sys
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LinearRegression
from joblib import Parallel, delayed
#*****************************#
### PARAMETERS ###
#*****************************#
n_participants = 4
n_sessions = 12
#*****************************#
### HELPER FUNCTIONS ###
#*****************************#
class load_data:
def __init__(self,dat,sourcedata_dir,trial_type='exp'):
self.dat = dat
self.trial_type = trial_type
self.sourcedata_dir = sourcedata_dir
def load_animacy_size(self):
ani_csv = f'{self.sourcedata_dir}/ratings_animacy.csv'
size_csv = f'{self.sourcedata_dir}/ratings_size.csv'
# load with pandas
ani_df = pd.read_csv(ani_csv)[['uniqueID', 'lives_mean']]
ani_df = ani_df.rename(columns={'lives_mean':'animacy'})
size_df = pd.read_csv(size_csv, sep=';')[['uniqueID', 'meanSize']]
size_df = size_df.rename(columns={'meanSize':'size'})
# ani_df has "_", size_df " " as separator in multi-word concepts
size_df['uniqueID'] = size_df.uniqueID.str.replace(' ', '_')
# merge
anisize_df = pd.merge(left=ani_df, right=size_df, on='uniqueID', how='outer')
assert anisize_df.shape[0] == ani_df.shape[0] == size_df.shape[0]
return anisize_df
def load_meg(self):
# select exp trails & sort the trials based on things_category_nr
epochs_exp = self.dat[(self.dat.metadata['trial_type']=='exp')]
sort_order = np.argsort(epochs_exp.metadata['things_category_nr'])
dat_sorted=epochs_exp[sort_order]
# getting data from each session and load it
self.n_categories = len(dat_sorted.metadata.things_category_nr.unique())
self.n_sessions = len(dat_sorted.metadata.session_nr.unique())
self.n_channels = len(dat_sorted.ch_names)
self.n_time = len(dat_sorted.times)
self.sess_data = np.empty([self.n_categories,self.n_channels,self.n_time,self.n_sessions])
for sess in range(self.n_sessions):
print('loading data for session ' + str(sess+1))
curr_data = dat_sorted[dat_sorted.metadata['session_nr']==sess+1]
curr_data = curr_data.load_data()
self.sess_data[:,:,:,sess]= curr_data._data
return self.sess_data
class linear_regression:
def __init__(self,dat,label):
self.dat = dat
self.label = label
self.n_categories = dat.shape[0]
self.n_channels = dat.shape[1]
self.n_time = dat.shape[2]
self.n_sessions = dat.shape[3]
def train_test_splits(self):
self.train_splits,self.test_splits = [],[]
for comb in itertools.combinations(np.arange(self.n_sessions), self.n_sessions-1):
self.train_splits.append(comb)
self.test_splits.append(list(set(np.arange(self.n_sessions)) - set(comb)))
return self.train_splits,self.test_splits
def run(self):
sess_dat = self.dat
train_splits,test_splits = self.train_test_splits()
pipe = Pipeline([('scaler', StandardScaler()),
('regression', LinearRegression())])
corr_coef = np.empty([self.n_time,self.n_sessions])
def fit_predict(pipe,train_x,train_y,test_x,test_y):
pipe.fit(train_x,train_y)
y_pred = pipe.predict(test_x)
return np.corrcoef(y_pred,test_y)[0,1]
for split in range(self.n_sessions):
print('cv-split ' + str(split))
training_x = np.take(sess_dat,train_splits[split],axis=3)
training_x = np.concatenate(tuple(training_x[:,:,:,i] for i in range(training_x.shape[3])),axis=0)
training_y = self.label
training_y = np.tile(training_y,self.n_sessions-1)
testing_x=np.take(sess_dat,test_splits[split][0],axis=3)
testing_y = self.label
corr_coef_time = Parallel(n_jobs=24)(delayed(fit_predict)(pipe,training_x[:,:,t],training_y,testing_x[:,:,t],testing_y) for t in range(self.n_time))
corr_coef[:,split] = corr_coef_time
return corr_coef
def run(p,preproc_dir):
epochs = mne.read_epochs(f'{preproc_dir}/preprocessed_P{str(p)}-epo.fif', preload=False)
anisize_df = load_data(epochs,sourcedata_dir,'exp').load_animacy_size()
data = load_data(epochs,sourcedata_dir,'exp').load_meg()
animacy_corr_coeff = linear_regression(data,anisize_df['animacy'].to_numpy()).run()
size_corr_coeff = linear_regression(data,anisize_df['size'].to_numpy()).run()
pd.DataFrame(animacy_corr_coeff).to_csv(f'{res_dir}/validation-animacy-P{str(p)}.csv')
pd.DataFrame(size_corr_coeff).to_csv(f'{res_dir}/validation-size-P{str(p)}.csv')
#*****************************#
### COMMAND LINE INPUTS ###
#*****************************#
if __name__=='__main__':
import argparse
parser = argparse.ArgumentParser()
parser.add_argument(
"-bids_dir",
required=True,
help='path to bids root',
)
args = parser.parse_args()
bids_dir = args.bids_dir
preproc_dir = f'{bids_dir}/derivatives/preprocessed/'
sourcedata_dir = f'{bids_dir}/sourcedata/'
res_dir = f'{bids_dir}/derivatives/output/'
if not os.path.exists(res_dir):
os.makedirs(res_dir)
####### Run analysis ########
for p in range(1,n_participants+1):
run(p,preproc_dir)