Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from clinica.iotools.utils.data_handling import check_volume_location_in_world_coordinate_system
from clinica.utils.inputs import clinica_file_reader
from clinica.utils.exceptions import ClinicaBIDSError, ClinicaException
from clinica.utils.input_files import T1W_NII
# Reading BIDS
# ============
read_node = npe.Node(name="read_node",
interface=nutil.IdentityInterface(fields=['bids_images'],
mandatory_inputs=True))
try:
read_node.inputs.bids_images = clinica_file_reader(self.subjects,
self.sessions,
self.bids_directory,
T1W_NII)
except ClinicaException as e:
error_str_final = 'Clinica faced error(s) while trying to read files in your CAPS directory.\n'
error_str_final += str(e)
raise ClinicaBIDSError(error_str_final)
check_volume_location_in_world_coordinate_system(read_node.inputs.bids_images, self.bids_directory)
self.connect([
(read_node, self.input_node, [('bids_images', 'input_images')])
])
def get_processed_images(caps_directory, subjects, sessions):
"""Extract processed image IDs in `caps_directory` based on `subjects`_`sessions`.
Todo:
[ ] Implement this static method in all pipelines
[ ] Make it abstract to force overload in future pipelines
"""
from clinica.utils.exceptions import ClinicaException
import datetime
from colorama import Fore
from clinica.utils.stream import cprint
now = datetime.datetime.now().strftime('%H:%M:%S')
cprint('\n%s[%s] Pipeline finished with errors.%s\n' % (Fore.RED, now, Fore.RESET))
cprint('%sCAPS outputs were not found for some image(s):%s' % (Fore.RED, Fore.RESET))
raise ClinicaException('Implementation on which image(s) failed will appear soon.')
interface=nutil.IdentityInterface(fields=self.get_input_fields(),
mandatory_inputs=True))
all_errors = []
d_input = []
for tissue_number in self.parameters['dartel_tissues']:
try:
current_file = clinica_file_reader(self.subjects,
self.sessions,
self.caps_directory,
{'pattern': 't1/spm/segmentation/dartel_input/*_*_T1w_segm-'
+ tissue_names[tissue_number] + '_dartelinput.nii*',
'description': 'Dartel input for tissue ' + tissue_names[tissue_number]
+ ' from T1w MRI',
'needed_pipeline': 't1-volume-tissue-segmentation'})
d_input.append(current_file)
except ClinicaException as e:
all_errors.append(e)
# Raise all errors if some happened
if len(all_errors) > 0:
error_message = 'Clinica faced errors while trying to read files in your BIDS or CAPS directories.\n'
for msg in all_errors:
error_message += str(msg)
raise RuntimeError(error_message)
# d_input is a list of size len(self.parameters['dartel_tissues'])
# Each element of this list is a list of size len(self.subjects)
read_parameters_node.inputs.dartel_inputs = d_input
self.connect([
(read_parameters_node, self.input_node, [('dartel_inputs', 'dartel_input_images')])
])
+ '/*_pet_space-Ixi549Space_suvr-pons_pet.nii.gz',
'description': pet_type + ' PET in Ixi549 space',
'needed_pipeline': 'pet-volume'})
except ClinicaException as e:
all_errors.append(e)
elif no_pvc.lower() == 'false':
try:
input_image = clinica_file_reader(self.subjects,
self.sessions,
self.caps_directory,
{'pattern': 'pet/preprocessing/group-' + self.parameters['group_id']
+ '_pet_space-Ixi549Space_pvc-rbv_suvr-pons_pet.nii.gz',
'description': pet_type + ' PET partial volume corrected (RBV) in Ixi549 space',
'needed_pipeline': 'pet-volume with PVC'})
except ClinicaException as e:
all_errors.append(e)
else:
raise ValueError(no_pvc + ' is not a valid keyword for -no_pvc'
+ ':only True or False are accepted (string)')
else:
raise ValueError('Image type ' + image_type + ' unknown')
try:
dartel_input = clinica_group_reader(self.caps_directory,
{'pattern': 'group-' + self.parameters['group_id'] + '_template.nii*',
'description': 'template file of group ' + self.parameters['group_id'],
'needed_pipeline': 't1-volume or t1-volume-create-dartel'})
except ClinicaException as e:
all_errors.append(e)
# Raise all errors if some happened
def custom_traceback(exc_type, exc_value, exc_traceback):
import traceback
import math
from colorama import Fore
from clinica.utils.exceptions import ClinicaException
from clinica.utils.stream import cprint
if issubclass(exc_type, ClinicaException):
cprint(exc_value)
elif issubclass(exc_type, KeyboardInterrupt):
cprint('\n%s[Error] Program interrupted by the user. Clinica will now exit...%s' %
(Fore.RED, Fore.RESET))
else:
cprint(Fore.RED + '\n' + '*' * 23 + '\n*** Clinica crashed ***\n' + '*' * 23 + '\n' + Fore.RESET)
cprint('%sException type:%s %s' % (Fore.YELLOW, Fore.RESET, exc_type.__name__))
cprint('%sException value:%s %s' % (Fore.YELLOW, Fore.RESET, exc_value))
cprint('Below are displayed information that were gathered when Clinica crashed. This will help to understand '
'what happened if you transfer those information to the Clinica development team.\n')
frames = traceback.extract_tb(exc_traceback)
framewidth = int(math.ceil(math.log(len(frames)) / math.log(10)))
filewidth = 0
linewidth = 0
functionwidth = 0
from colorama import Fore
from clinica.utils.exceptions import ClinicaException
if not os.path.isfile(tsv_file):
raise ClinicaException(
"\n%s[Error] The TSV file you gave is not a file.%s\n"
"\n%sError explanations:%s\n"
" - Clinica expected the following path to be a file: %s%s%s\n"
" - If you gave relative path, did you run Clinica on the good folder?" %
(Fore.RED, Fore.RESET,
Fore.YELLOW, Fore.RESET,
Fore.BLUE, tsv_file, Fore.RESET)
)
ss_df = pd.io.parsers.read_csv(tsv_file, sep='\t')
if 'participant_id' not in list(ss_df.columns.values):
raise ClinicaException(
"\n%s[Error] The TSV file does not contain participant_id column (path: %s)%s" %
(Fore.RED, tsv_file, Fore.RESET)
)
if 'session_id' not in list(ss_df.columns.values):
raise ClinicaException(
"\n%s[Error] The TSV file does not contain session_id column (path: %s)%s" %
(Fore.RED, tsv_file, Fore.RESET)
)
participants = list(ss_df.participant_id)
sessions = list(ss_df.session_id)
# Remove potential whitespace in participant_id or session_id
return [sub.strip(' ') for sub in participants], [ses.strip(' ') for ses in sessions]
"""
class ClinicaException(Exception):
""" Base class for Clinica exceptions """
class ClinicaMissingDependencyError(ClinicaException):
""" Base class for Clinica dependencies errors """
class ClinicaBIDSError(ClinicaException):
""" Base class for BIDS errors """
class ClinicaCAPSError(ClinicaException):
""" Base class for CAPS errors """
args.feature_label = 'RD'
elif args.feature_type == 'dti_projection_ad':
args.custom_file = '@subject/@session/noddi/postprocessing/dti-register-vertex-fsaverage/cortex-projection/@subject_@session_OnFsaverage_fwhm-@fwhm_measure-ad_hemi-@hemi.mgh'
args.feature_label = 'AD'
else:
raise ClinicaException('Feature type ' + args.feature_type + ' not recognized. Use --custom_file '
'to specify your own files (without --feature_type).')
elif args.feature_type is None:
if args.custom_file is None:
cprint('No feature type selected: using cortical thickness as default value')
args.custom_file = '@subject/@session/t1/freesurfer_cross_sectional/@subject_@session/surf/@hemi.thickness.fwhm@fwhm.fsaverage.mgh'
args.feature_label = 'ct'
else:
cprint('Using custom features.')
if args.feature_label is None:
raise ClinicaException('You must specify a --feature_label when using the --custom_files flag.')
# Check if the group label has been existed, if yes, give an error to the users
# Note(AR): if the user wants to compare Cortical Thickness measure with PET measure
# using the group_id, Clinica won't allow it.
# TODO: Modify this behaviour
if os.path.exists(os.path.join(os.path.abspath(self.absolute_path(args.caps_directory)), 'groups', 'group-' + args.group_id)):
error_message = 'group_id: ' + args.group_id + ' already exists, please choose another one or delete ' \
'the existing folder and also the working directory and rerun the pipeline'
raise ClinicaException(error_message)
pipeline = StatisticsSurface(
caps_directory=self.absolute_path(args.caps_directory),
tsv_file=self.absolute_path(args.subject_visits_with_covariates_tsv),
base_dir=self.absolute_path(args.working_directory),
)
pipeline.parameters = {
# coding: utf8
"""
Clinica exceptions
"""
class ClinicaException(Exception):
""" Base class for Clinica exceptions """
class ClinicaMissingDependencyError(ClinicaException):
""" Base class for Clinica dependencies errors """
class ClinicaBIDSError(ClinicaException):
""" Base class for BIDS errors """
class ClinicaCAPSError(ClinicaException):
""" Base class for CAPS errors """
from clinica.utils.dwi import check_dwi_volume
from clinica.utils.inputs import clinica_file_reader
import clinica.pipelines.dwi_preprocessing_using_phasediff_fieldmap.dwi_preprocessing_using_phasediff_fieldmap_utils as utils
from clinica.utils.exceptions import ClinicaBIDSError, ClinicaException
import clinica.utils.input_files as input_files
from clinica.utils.stream import cprint
all_errors = []
# DWI
try:
dwi_bids = clinica_file_reader(self.subjects,
self.sessions,
self.bids_directory,
input_files.DWI_NII)
except ClinicaException as e:
all_errors.append(e)
# DWI json
try:
dwi_json = clinica_file_reader(self.subjects,
self.sessions,
self.bids_directory,
input_files.DWI_JSON)
# Create list_eff_echo_spacings and list_enc_directions
list_eff_echo_spacings = []
list_enc_directions = []
for json in dwi_json:
[eff_echo_spacing, enc_direction] = utils.parameters_from_dwi_metadata(json)
list_eff_echo_spacings.append(eff_echo_spacing)
list_enc_directions.append(enc_direction)