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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186 | class Exp:
#############################################################################################
# EXPERIMENTAL PARAMETERS
#############################################################################################
# Volumes processing information
n_heatup_vols = int(config["experiment"]["n_heatup_vols"]) # Number of volumes needed for heating up MRI scanner (heatup duration = TR * n_heatup_vols)
#(ie, n_heatup_vols = 0 --> 1 heatup volume)
n_baseline_vols = int(config["experiment"]["n_baseline_vols"]) # Number of baseline volumes after heatup to sample before beggining with the experimental task (baseline duration = TR * n_baseline_vols)
HRF_peak_onset = int(config["experiment"]["HRF_peak_onset"]) # HRF peak onset threshold (seconds from trial onset). Default HRF_peak_onset = 5 for decoding procedure = 'average_hrf_peak_vols' or 'average_probs'.
# Set to 0 if do you want to decode volumes from trial onset in dynamic neurofeedback experiments.
HRF_peak_offset = int(config["experiment"]["HRF_peak_offset"]) #float("inf")# HRF peak offset threshold (seconds from trial onset). Default HRF_peak_offset = 11 for decoding procedure = 'average_hrf_peak_vols' or 'average_probs'.
# Set to float("inf") if using an undetermined number of HRF peak volumes within each trial of dynamic neurofeedback experiments.
TR = int(config["experiment"]["TR"]) # Repetition time. Volumes sampling rate (in seconds)
# Volumes tracking
first_vol_idx = int(config["experiment"]["first_vol_idx"]) # First volume index to track in EXP.raw_volumes_folder
index_format = config["experiment"]["index_format"] # How volumes indexes are left-zero padded in fMRI cls.raw_volumes_folder folder? (ex., index_format = '04d' for IM-0001.dcm)
#index format size only matters if the number of 0 at the end of the .dcm file is smaller than the specified by index_format (ie, 4)
# Z-scoring procedure
zscoring_procedure = config["experiment"]["zscoring_procedure"] # 'to_baseline' (each task volume will be z-scored relative to data from that run baseline in specific R.O.I. For example, volumes 51 will be z-scored to n_baseline_vols data)
# 'to_timeseries' (each task volume will be z-scored relative to that run previous volumes in specific R.O.I.. For example, volume 51 will be z-scored using data from volume 0 to that volume)
# 'to_model_session' (default) (each task volume will be z-scored relative to model construction session data in specific R.O.I. using its mean and standard deviation).
# Decoding settings
decoding_procedure = config["experiment"]["decoding_procedure"] # 'average_hrf_peak_vols' (average volumes within a trial HRF peak before decoding a single averaged volume to increase signal-to-noise ratio)
# 'average_probs' (default) (average decoding probabilities of volumes within a trial HRF peak to increase feedbacks variability)
# 'dynamic' (all volumes within a trial HRF peak, will be decoded independently and sent individually to experimental software as feedback)
coadaptation_active = string_to_bool(config["experiment"]["coadaptation_active"]) # activating co-adaptation, a.k.a real time training of the decoder while conducting the DecNef
coadaptation_background_warmup = string_to_bool(config["experiment"]["coadaptation_background_warmup"]) # Only relevant if "coadaptation_active = False". Training the coadaptation decoder with the DecNef data but not in real time and without using it in predictions and participants feedback (the predictions are made with the original decoder)
coadaptation_vol_acceptance_criteria = float(config["experiment"]["coadaptation_vol_acceptance_criteria"]) # The threshold of decoder prediction in selecting the volumes used for co-adapatation.
classifier_type = config["experiment"]["classifier_type"] # The classifier used (must be the same as the one used for the training process) [logisticregression, svmlinear, decisiontree, extratree, randomforest, extratrees, bagging, gradientboosting, adaboost, naivebayes, kneighbors, mlp, sgd]
@classmethod # This method ensures participants data and directory routes can be inherited by all other classes
def _new_participant(cls):
""" Request new participant data (participant, session, run) each time main.py runs to set directories routes """
def check_file(file):
""" Check if a essential file exists or not. If not then cancel script execution. """
if not os.path.exists(file):
sys.exit(Fore.RED + f'[ERROR] File/Directory "{file}" does not exist. Check that you are pointing to a correct path. Breaking main.py execution.')
#############################################################################################
# DIRECTORIES & DATA
#############################################################################################
# First, ask for participant info to find/generate its corresponding folders
print(Fore.YELLOW + 'Specify participants data before initialization:')
cls.subject = input(Fore.YELLOW + '\nParticipant number: ')
cls.session = input(Fore.YELLOW + 'Session number: ')
cls.run = input(Fore.YELLOW + 'Run number: ')
print('\n')
#PATH TO COPY IN THE TERMINAL: cd /firmm/rt-fmri/META-BRAIN/pyDecNef-static/2.neurofeedback_training/1.realtime_fMRI_scripts'''
# Package directory
cls.moduledir = Path(__file__).absolute().parent.parent.parent
# fMRI raw volumes output folder
if string_to_bool(config["experiment"]["simulated_experiment"]): # Whether doing Real-time DecNef or simulated experiment
cls.raw_volumes_folder_path = os.path.join(cls.moduledir.parent,'2.MRI_simulator','outputs') # To use with fMRI simulator
else:
cls.raw_volumes_folder_path = config["files_and_dir"]["raw_volumes_folder_path"] # To use in a real experiment setting (the path to the raw volumes folder generated by the MRI scanner)
cls.raw_volumes_folder = Path(cls.raw_volumes_folder_path)
check_file(cls.raw_volumes_folder)
# Required resources directory
# Contains pretrained model, region of interest mask, reference functional volume & z-scoring information
cls.resources_dir = os.path.join(cls.moduledir.parent,'required_resources' ,f'sub-{cls.subject}')
check_file(cls.resources_dir)
# Pretrained model path
cls.model_name = config["files_and_dir"]["model_name"] # The model file name in the required_resources folder
cls.model_file = os.path.join(cls.resources_dir , 'models',cls.model_name)
check_file(cls.model_file)
if cls.coadaptation_active:
cls.coadaptation_model_name = cls.model_name + "_coadapted"
cls.coadaptation_model_file = os.path.join(cls.resources_dir,'models',cls.coadaptation_model_name)
if not os.path.exists(cls.coadaptation_model_file):
model = load(cls.model_file)
dump(model,cls.coadaptation_model_file)
cls.model_file = cls.coadaptation_model_file
# Region of interest mask path (as .nii to maximize load speed)
cls.mask_name = config["files_and_dir"]["mask_name"] # The mask file name in the required_resources folder
cls.mask_file = os.path.join(cls.resources_dir, 'masks',cls.mask_name)
check_file(cls.mask_file)
# Reference functional volume path (from model construction session. As .nii to maximize load speed)
cls.ref_vol_name = config["files_and_dir"]["ref_vol_name"] # The reference volume file name in the required_resources folder
cls.ref_vol_file = os.path.join(cls.resources_dir, 'training_session_ref_image',cls.ref_vol_name)
check_file(cls.ref_vol_file)
# ROI reference data for z-scoring (if zscoring_procedure is 'to_model_session')
if cls.zscoring_procedure == 'to_model_session':
cls.zscoring_mean_file_name = config["files_and_dir"]["zscoring_mean_file_name"] # The zscore mean file name in the required_resources folder
cls.zscoring_std_file_name = config["files_and_dir"]["zscoring_std_file_name"] # The zscore std file name in the required_resources folder
cls.zscoring_mean = os.path.join(cls.resources_dir , 'training_zscoring_data',cls.zscoring_mean_file_name) # Numpy array containing mean of model construction session data
cls.zscoring_std = os.path.join(cls.resources_dir, 'training_zscoring_data',cls.zscoring_std_file_name) # Numpy array containing standard deviation of model construction session data
if (cls.coadaptation_active or cls.coadaptation_background_warmup):
cls.coadaptation_base_training_data_dir_name = config["files_and_dir"]["coadaptation_base_training_data_dir_name"] # The directory used to dump the preprocessed data and construct the dataset for the co-adaptation
cls.coadaptation_training_data_file_name = config["files_and_dir"]["coadaptation_training_data_file_name"] # The file name of the original decoder training data to be used for co-adaptation
cls.coadaptation_training_data_labels_file_name = config["files_and_dir"]["coadaptation_training_data_labels_file_name"] # The file name of the original decoder training labels to be used for co-adaptation
cls.coadaptation_training_data_dir = os.path.join(cls.resources_dir,cls.coadaptation_base_training_data_dir_name)
cls.coadaptation_training_data_file = os.path.join(cls.coadaptation_training_data_dir,cls.coadaptation_training_data_file_name)
cls.coadaptation_training_data_labels_file = os.path.join(cls.coadaptation_training_data_dir,cls.coadaptation_training_data_labels_file_name)
check_file(cls.coadaptation_training_data_dir)
check_file(cls.coadaptation_training_data_file)
check_file(cls.coadaptation_training_data_labels_file)
# Create an outputs directory to store participant session log files and preprocessed volumes
cls.outputs_dir = os.path.join(cls.moduledir,'outputs',f'sub-{cls.subject}_session-{cls.session}')
Path(cls.outputs_dir).mkdir(parents=True, exist_ok=True)
# main.py script run time
script_run_time = time.strftime('%Y-%m-%d_%H-%M-%S') # Get main.py script run time, to create an unique run folder
# and avoid folder replacement problems when wrongly typing runs number
# Make a run directory inside outputs dir to store all participant log files and preprocessed volumes
cls.run_dir = os.path.join(cls.outputs_dir, f'run-{cls.run}_{script_run_time}')
Path(cls.run_dir).mkdir(parents=True, exist_ok=True)
# Make a trials directory inside run directory to store all masked volumes and information classified by trial
cls.trials_dir = os.path.join(cls.run_dir , 'trials')
Path(cls.trials_dir).mkdir(parents=True, exist_ok=True)
# Make a logs directory inside run directory to store run logs data
cls.logs_dir = os.path.join(cls.run_dir , 'logs_dir')
Path(cls.logs_dir).mkdir(parents=True, exist_ok=True)
# Make a preprocessed volumes directory inside run directory to store all outputs corresponding to preprocessed volumes in that run
cls.preprocessed_dir = os.path.join(cls.run_dir, 'preprocessed')
Path(cls.preprocessed_dir).mkdir(parents=True, exist_ok=True)
|