Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def write_derivative_description(bids_dir, deriv_dir):
from fitlins import __version__
desc = {
'Name': 'Fitlins output',
'BIDSVersion': '1.1.0',
'PipelineDescription': {
'Name': 'FitLins',
'Version': __version__,
'CodeURL': 'https://github.com/poldracklab/fitlins',
},
'CodeURL': 'https://github.com/poldracklab/fitlins',
'HowToAcknowledge': 'https://github.com/poldracklab/fitlins',
}
# Keys that can only be set by environment
if 'FITLINS_DOCKER_TAG' in os.environ:
desc['DockerHubContainerTag'] = os.environ['FITLINS_DOCKER_TAG']
if 'FITLINS_SINGULARITY_URL' in os.environ:
singularity_url = os.environ['FITLINS_SINGULARITY_URL']
desc['SingularityContainerURL'] = singularity_url
try:
desc['SingularityContainerMD5'] = _get_shub_version(singularity_url)
except ValueError:
pass
def get_parser():
"""Build parser object"""
verstr = 'fitlins v{}'.format(__version__)
parser = ArgumentParser(description='FitLins: Workflows for Fitting Linear models to fMRI',
formatter_class=RawTextHelpFormatter)
# Arguments as specified by BIDS-Apps
# required, positional arguments
# IMPORTANT: they must go directly with the parser object
parser.add_argument('bids_dir', action='store', type=op.abspath,
help='the root folder of a BIDS valid dataset (sub-XXXXX folders should '
'be found at the top level in this folder).')
parser.add_argument('output_dir', action='store', type=op.abspath,
help='the output path for the outcomes of preprocessing and visual '
'reports')
parser.add_argument('analysis_level', choices=['run', 'session', 'participant', 'dataset'],
help='processing stage to be runa (see BIDS-Apps specification).')
if opts.work_dir:
# dump crashes in working directory (non /tmp)
fitlins_wf.config['execution']['crashdump_dir'] = opts.work_dir
# easy to read crashfiles
fitlins_wf.config['execution']['crashfile_format'] = 'txt'
retcode = 0
if not opts.reports_only:
try:
fitlins_wf.run(**plugin_settings)
except Exception:
retcode = 1
models = auto_model(layout) if model == 'default' else [model]
run_context = {'version': __version__,
'command': ' '.join(sys.argv),
'timestamp': time.strftime('%Y-%m-%d %H:%M:%S %z'),
}
selectors = {'desc': opts.desc_label}
if opts.space is not None:
selectors['space'] = opts.space
if subject_list is not None:
selectors['subject'] = subject_list
for model in models:
analysis = Analysis(layout, model=model)
analysis.setup(**selectors)
report_dict = build_report_dict(deriv_dir, work_dir, analysis)
write_full_report(report_dict, run_context, deriv_dir)
layout = BIDSLayout(opts.bids_dir,
derivatives=derivatives,
ignore=ignore,
force_index=force_index,
database_path=database_path,
reset_database=reset_database)
subject_list = None
if opts.participant_label is not None:
subject_list = bids.collect_participants(
layout, participant_label=opts.participant_label)
# Build main workflow
logger.log(25, INIT_MSG(
version=__version__,
subject_list=subject_list)
)
fitlins_wf = init_fitlins_wf(
database_path, deriv_dir,
analysis_level=opts.analysis_level, model=model,
space=opts.space, desc=opts.desc_label,
participants=subject_list, base_dir=work_dir,
smoothing=opts.smoothing, drop_missing=opts.drop_missing,
)
if opts.work_dir:
# dump crashes in working directory (non /tmp)
fitlins_wf.config['execution']['crashdump_dir'] = opts.work_dir
# easy to read crashfiles
fitlins_wf.config['execution']['crashfile_format'] = 'txt'