Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import clinica.pipelines.t1_freesurfer_longitudinal.t1_freesurfer_template_utils as utils
# import t1_freesurfer_template_utils as utils
import nipype.interfaces.utility as nutil
import nipype.pipeline.engine as npe
# Per-subject creation of the template
# Step 1: create folder to store recon-all -base results
# ======
store_reconallbase_node_name = '1_store_reconallbase'
store_reconallbase_node = npe.Node(
name=store_reconallbase_node_name,
interface=nutil.Function(
input_names=['in_subject_list'],
output_names=['out_workdirstore_path'],
function=utils.store_reconallbase_results))
# Step 2: prepare unbiased template flags for all subjects
# (i.e. create flags for recon-all -base)
# ======
prepare_template_node_name = '2_prepare_template'
prepare_template_node = npe.MapNode(
name=prepare_template_node_name,
interface=nutil.Function(
input_names=['in_subject',
'in_session_list'],
output_names=['out_reconallbase_flags'],
function=utils.get_reconallbase_flags
),
iterfield=['in_subject',
'in_session_list'])
# get a unique list of subjects and associated per-subject list of
# sessions
unique_subject_list, persubject_session_list2 = clinica.utils.participant.get_unique_subjects(
subject_list, session_list)
# Check processed/unprocessed subjects
[
pcssd_sublist,
pcssd_seslist2,
pcssd_capstargetlist,
unpcssd_sublist,
unpcssd_seslist2,
unpcssd_capstargetlist,
all_sublist,
all_seslist2,
all_capstargetlist] = utils.check_caps_processing(
'template',
unique_subject_list,
persubject_session_list2,
caps_dir)
return [
pcssd_sublist,
pcssd_seslist2,
pcssd_capstargetlist,
unpcssd_sublist,
unpcssd_seslist2,
unpcssd_capstargetlist,
all_sublist,
all_seslist2,
all_capstargetlist]
) for subject_index in range(subject_number))
# change the relative path to be absolute path
abs_caps_path = os.path.abspath(caps_dir)
subjects_path = os.path.join(abs_caps_path, 'subjects')
if not os.path.isdir(subjects_path):
error_msg = 'Error: {0} does not exist.'.format(subjects_path)
raise IOError(error_msg)
for subject_index in range(subject_number):
subject = subject_list[subject_index]
session = session_list[subject_index]
subses = subses_list[subject_index]
xsectional_path = utils.get_capsxsectional_path(
caps_dir, subject, session)
if not os.path.isdir(xsectional_path):
error_msg = 'Error: {0} does not exist.'.format(xsectional_path)
raise IOError(error_msg)
# check the recon-all.log
log_file = os.path.join(
xsectional_path, subses, 'scripts', 'recon-all.log')
if os.path.isfile(log_file):
last_line = subprocess.check_output(['tail', '-1', log_file])
if b'finished without error' not in last_line:
error_msg = 'Subject {0} has not been processed yet.'.format(
subses)
error_msg = '{0} Re-run t1-freesurfer-cross-sectional.'.format(
error_msg)
raise IOError(error_msg)
checkinput_node = npe.Node(name=checkinput_node_name,
interface=nutil.Function(
input_names=['in_caps_dir',
'in_subject_list',
'in_session_list',
'in_overwrite_caps',
'in_working_directory',
'in_n_procs',
'in_rundir'],
output_names=['out_subject_list',
'out_session_list2',
'out_caps_target_list',
'out_unpcssd_sublist',
'out_pcssd_capstargetlist',
'out_overwrite_tsv'],
function=utils.process_input_node))
checkinput_node.inputs.in_caps_dir = self.caps_directory
checkinput_node.inputs.in_subject_list = self.subjects
checkinput_node.inputs.in_session_list = self.sessions
checkinput_node.inputs.in_working_directory = self.base_dir
checkinput_node.inputs.in_overwrite_caps = self.parameters[
'overwrite_caps']
checkinput_node.inputs.in_n_procs = self.parameters['n_procs']
checkinput_node.inputs.in_rundir = os.getcwd()
self.connect([
(
checkinput_node, self.input_node,
[('out_subject_list', 'subject_list'),
('out_session_list2', 'session_list2'),
('out_caps_target_list', 'caps_target_list'),
('out_unpcssd_sublist', 'unpcssd_sublist'),
the CAPS folder)
Args:
caps_dir (string): CAPS directory to contain the output
subject_list (list): a list containing all the participant_id
session_list (list): a list containing all the session_id
Returns:
N/A
"""
import clinica.pipelines.t1_freesurfer_longitudinal.t1_freesurfer_template_utils as template_utils
import clinica.pipelines.t1_freesurfer_longitudinal.t1_freesurfer_longitudinal_correction_utils as utils
# Check if t1-freesurfer-cross-sectional has been correctly run
# beforehand. Will crash if it has not
template_utils.check_xsectional_reconalled(
caps_dir, subject_list, session_list)
# get unique list of subjects and corresponding per-subject sessions
[unique_subject_list,
persubject_session_list2] = clinica.utils.participant.get_unique_subjects(
subject_list, session_list)
# Check if t1-freesurfer-template has been correctly run
# beforehand. Will crash if it has not
utils.check_template_reconalled(
caps_dir, unique_subject_list, persubject_session_list2)
Returns:
N/A
"""
import os
import subprocess
import clinica.pipelines.t1_freesurfer_longitudinal.t1_freesurfer_template_utils as template_utils
unique_subject_number = len(unique_subject_list)
# check recon_all template creation for each subject
for subject_index in range(unique_subject_number):
subject = unique_subject_list[subject_index]
session_list = persubject_session_list2[subject_index]
# get template location in the CAPS directory
caps_template_target = template_utils.get_capstemplate_path(
caps_dir, subject, session_list)
# check if template subfolder exists
if not os.path.isdir(caps_template_target):
error_msg = 'Error: {0} does not exist.'.format(
caps_template_target)
raise IOError(error_msg)
# Check that the recon-all template creation run was successful
log_file = os.path.join(caps_template_target, 'scripts', 'recon-all.log')
if os.path.isfile(log_file):
last_line = subprocess.check_output(['tail', '-1', log_file])
if b'finished without error' not in last_line:
error_msg = 'No template found for subject {0}.'.format(
subject)
error_msg = '{0} Re-run t1-freesurfer-template.'.format(
error_msg)
raise IOError(error_msg)
topcss_seslist2 (list of list): list of the list of sessions
that correspond to each subject to be processed
topcss_capstargetlist (list of strings): list, for all subjects
to be processed, of the associated CAPS folder where the
template will get stored
overwrite_tsv_path (string): path to the .tsv to use if the user
wishes to overwrite the subject that have already been
processed in the CAPS folder. None if no subjects already
processed.
"""
import os
import csv
import clinica.pipelines.t1_freesurfer_longitudinal.t1_freesurfer_template_utils as utils
# Convert overwrite-caps flag to boolean
force_overwrite = utils.boolean_overwrite_caps(in_overwrite_caps)
# Generate .tsv file. Will be used to re-run with --force-overwrite
# enabled.
pcssd_sub_number = len(pcssd_sublist)
if pcssd_sub_number > 0:
# only write the .tsv file if at least one subject already
# processed
overwrite_tsv_path = os.path.join(in_rundir, 'overwrite.tsv')
with open(overwrite_tsv_path, 'w') as overwrite_tsv_file:
# create tsv file
tsv_writer = csv.writer(
overwrite_tsv_file, delimiter='\t')
tsv_writer.writerow(['participant_id', 'session_id'])
# loop through all processed subject/session
for pcssd_sub_index in range(pcssd_sub_number):
pcssd_sub = pcssd_sublist[pcssd_sub_index]
tsv_writer.writerow(['participant_id', 'session_id'])
# loop through all processed subject/session
for pcssd_sub_index in range(pcssd_sub_number):
pcssd_sub = pcssd_sublist[pcssd_sub_index]
pcssd_seslist = pcssd_seslist2[pcssd_sub_index]
pcssd_seslist_number = len(pcssd_seslist)
for pcssd_seslist_index in range(pcssd_seslist_number):
pcssd_ses = pcssd_seslist[pcssd_seslist_index]
tsv_writer.writerow([pcssd_sub, pcssd_ses])
else:
overwrite_tsv_path = None
# if all subjects already processed and --force-overwrite option not
# provided, exit
if (not unpcssd_sublist) and (not force_overwrite):
cl_overwrite_caps = utils.get_cl_overwritecaps(
in_caps_dir,
overwrite_tsv_path,
in_working_directory,
in_n_procs)
error_msg = 'Error: all subjects are already processed. To overwrite'
error_msg = '{0} the data in CAPS folder, please run:\n'.format(
error_msg)
error_msg = '{0} {1}'.format(error_msg, cl_overwrite_caps)
raise ValueError(error_msg)
# check what subjects / sessions / CAPS target dirs to process
# depending on --force-overwrite flag
if force_overwrite:
topcss_sublist = all_sublist
topcss_seslist2 = all_seslist2
topcss_capstargetlist = all_capstargetlist