wip: add option for sessionwise processing

Author: mgxd

fmriprep/cli/workflow.py

@@ -41,12 +41,11 @@ def build_workflow(config_file, retval):
     from niworkflows.utils.bids import collect_participants
     from niworkflows.utils.misc import check_valid_fs_license
 
+    from fmriprep import config, data
     from fmriprep.reports.core import generate_reports
     from fmriprep.utils.bids import check_pipeline_version
-
-    from .. import config, data
-    from ..utils.misc import check_deps
-    from ..workflows.base import init_fmriprep_wf
+    from fmriprep.utils.misc import check_deps, fmt_subjects_sessions
+    from fmriprep.workflows.base import init_fmriprep_wf
 
     config.load(config_file)
     build_log = config.loggers.workflow
@@ -85,16 +84,27 @@ def build_workflow(config_file, retval):
         config.execution.layout, participant_label=config.execution.participant_label
     )
     session_list = config.execution.session_label or []
-    subject_session_list = get_subject_session_list()
+    subject_session_list = create_processing_groups(
+        config.execution.layout,
+        subject_list,
+        session_list,
+        config.execution.subject_anatomical_reference,
+    )
+    config.execution.processing_groups = subject_session_list
 
     # Called with reports only
     if config.execution.reports_only:
-        build_log.log(25, 'Running --reports-only on participants %s', ', '.join(subject_list))
-        session_list = (
-            config.execution.bids_filters.get('bold', {}).get('session')
-            if config.execution.bids_filters
-            else None
+        build_log.log(
+            25,
+            'Running --reports-only on %s',
+            fmt_subjects_sessions(subject_session_list)
         )
+        if not session_list:
+            session_list = (
+                config.execution.bids_filters.get('bold', {}).get('session')
+                if config.execution.bids_filters
+                else None
+            )
 
         failed_reports = generate_reports(
             config.execution.participant_label,
@@ -115,7 +125,7 @@ def build_workflow(config_file, retval):
     init_msg = [
         "Building fMRIPrep's workflow:",
         f'BIDS dataset path: {config.execution.bids_dir}.',
-        f'Participant list: {subject_list}.',
+        f'Participants and sessions: {fmt_subjects_sessions(subject_session_list)}.',
         f'Run identifier: {config.execution.run_uuid}.',
         f'Output spaces: {config.execution.output_spaces}.',
     ]
@@ -240,7 +250,7 @@ def build_boilerplate(config_file, workflow):
             config.loggers.cli.warning('Could not generate CITATION.tex file:\n%s', ' '.join(cmd))
 
 
-def get_subject_session_list(
+def create_processing_groups(
     layout: 'BIDSLayout',
     subject_list: list,
     session_list: list | str | None,

fmriprep/config.py

@@ -434,6 +434,8 @@ class execution(_Config):
     """Only build the reports, based on the reportlets found in a cached working directory."""
     run_uuid = f'{strftime("%Y%m%d-%H%M%S")}_{uuid4()}'
     """Unique identifier of this particular run."""
+    processing_groups = None
+    """List of tuples (participant, session(s)) that will be preprocessed."""
     participant_label = None
     """List of participant identifiers that are to be preprocessed."""
     session_label = None

fmriprep/utils/misc.py

@@ -66,3 +66,28 @@ def estimate_bold_mem_usage(bold_fname: str) -> tuple[int, dict]:
     }
 
     return bold_tlen, mem_gb
+
+
+def fmt_subjects_sessions(subses: list[str], concat_limit: int = 1):
+    """
+    Format a list of subjects and sessions to be printed.
+
+    Example
+    -------
+    >>> fmt_subjects_sessions([('01', 'A'), ('02', ['A', 'B']), ('03', None), ('04', ['A'])])
+    'sub-01 ses-A, sub-02 (2 sessions), sub-03, sub-04 ses-A'
+    """
+    output = []
+    for subject, session in subses:
+        if isinstance(session, list):
+            if len(session) > concat_limit:
+                output.append(f'sub-{subject} ({len(session)} sessions)')
+                continue
+            session = session[0]
+
+        if session is None:
+            output.append(f'sub-{subject}')
+        else:
+            output.append(f'sub-{subject} ses-{session}')
+
+    return ', '.join(output)

fmriprep/workflows/base.py

@@ -52,7 +52,7 @@ def init_fmriprep_wf():
     Build *fMRIPrep*'s pipeline.
 
     This workflow organizes the execution of FMRIPREP, with a sub-workflow for
-    each subject.
+    each processing group.
 
     If FreeSurfer's ``recon-all`` is to be run, a corresponding folder is created
     and populated with any needed template subjects under the derivatives folder.
@@ -91,12 +91,27 @@ def init_fmriprep_wf():
         if config.execution.fs_subjects_dir is not None:
             fsdir.inputs.subjects_dir = str(config.execution.fs_subjects_dir.absolute())
 
-    for subject_id in config.execution.participant_label:
-        single_subject_wf = init_single_subject_wf(subject_id)
+    for subject_id, session_ids in config.execution.processing_groups:
+        log_dir = config.execution.fmriprep_dir / f'sub-{subject_id}'
+        sessions = listify(session_ids)
+        ses_str = ''
 
-        single_subject_wf.config['execution']['crashdump_dir'] = str(
-            config.execution.fmriprep_dir / f'sub-{subject_id}' / 'log' / config.execution.run_uuid
+        if isinstance(sessions, list):
+            from smriprep.utils.misc import stringify_sessions
+
+            ses_str = stringify_sessions(sessions)
+            log_dir /= f'ses-{ses_str}'
+
+        log_dir /= 'log' / config.execution.run_uuid
+
+        wf_name = '_'.join(
+            ('sub', subject_id,) +
+            (('ses', ses_str) if ses_str else ()) +
+            ('wf',)
         )
+        single_subject_wf = init_single_subject_wf(subject_id, sessions, name=wf_name)
+
+        single_subject_wf.config['execution']['crashdump_dir'] = str(log_dir)
         for node in single_subject_wf._get_all_nodes():
             node.config = deepcopy(single_subject_wf.config)
         if freesurfer:
@@ -105,16 +120,17 @@ def init_fmriprep_wf():
             fmriprep_wf.add_nodes([single_subject_wf])
 
         # Dump a copy of the config file into the log directory
-        log_dir = (
-            config.execution.fmriprep_dir / f'sub-{subject_id}' / 'log' / config.execution.run_uuid
-        )
         log_dir.mkdir(exist_ok=True, parents=True)
         config.to_filename(log_dir / 'fmriprep.toml')
 
     return fmriprep_wf
 
 
-def init_single_subject_wf(subject_id: str):
+def init_single_subject_wf(
+    subject_id: str,
+    session_id: str | list[str] | None = None,
+    name: str | None = None,
+):
     """
     Organize the preprocessing pipeline for a single subject.
 
@@ -139,6 +155,12 @@ def init_single_subject_wf(subject_id: str):
     ----------
     subject_id : :obj:`str`
         Subject label for this single-subject workflow.
+    session_id
+        Session label(s) for this workflow.
+    name
+        Name of the workflow.
+        If not provided, will be set to ``sub_{subject_id}_ses_{session_id}_wf``.
+
 
     Inputs
     ------
@@ -169,7 +191,10 @@ def init_single_subject_wf(subject_id: str):
 
     from fmriprep.workflows.bold.base import init_bold_wf
 
-    workflow = Workflow(name=f'sub_{subject_id}_wf')
+    if name is None:
+        name = f'sub_{subject_id}_wf'
+
+    workflow = Workflow(name=name)
     workflow.__desc__ = f"""
 Results included in this manuscript come from preprocessing
 performed using *fMRIPrep* {config.environment.version}
@@ -204,6 +229,7 @@ def init_single_subject_wf(subject_id: str):
     subject_data = collect_data(
         config.execution.layout,
         subject_id,
+        session_id=session_id,
         task=config.execution.task_id,
         echo=config.execution.echo_idx,
         bids_filters=config.execution.bids_filters,
@@ -255,6 +281,7 @@ def init_single_subject_wf(subject_id: str):
                     derivatives_dir=deriv_dir,
                     subject_id=subject_id,
                     std_spaces=std_spaces,
+                    session_id=session_id,
                 )
             )
 
@@ -265,7 +292,7 @@ def init_single_subject_wf(subject_id: str):
             subject_data=subject_data,
             anat_only=config.workflow.anat_only,
             subject_id=subject_id,
-            anat_derivatives=anatomical_cache if anatomical_cache else None,
+            anat_derivatives=anatomical_cache or None,
         ),
         name='bidssrc',
     )
@@ -363,7 +390,7 @@ def init_single_subject_wf(subject_id: str):
             ('roi', 'inputnode.roi'),
             ('flair', 'inputnode.flair'),
         ]),
-        (bids_info, anat_fit_wf, [(('subject', _prefix), 'inputnode.subject_id')]),
+        (bids_info, anat_fit_wf, [(('subject', _prefix, 'session'), 'inputnode.subject_id')]),
         # Reporting connections
         (inputnode, summary, [('subjects_dir', 'subjects_dir')]),
         (bidssrc, summary, [('t2w', 't2w'), ('bold', 'bold')]),
@@ -929,8 +956,21 @@ def map_fieldmap_estimation(
     return fmap_estimators, estimator_map
 
 
-def _prefix(subid):
-    return subid if subid.startswith('sub-') else f'sub-{subid}'
+def _prefix(subject_id, session_id=None):
+    """Create FreeSurfer subject ID."""
+    if not subject_id.startswith('sub-'):
+        subject_id = f'sub-{subject_id}'
+
+    if session_id:
+        ses_str = session_id
+        if isinstance(session_id, list):
+            from smriprep.utils.misc import stringify_sessions
+
+            ses_str = stringify_sessions(session_id)
+        if not ses_str.startswith('ses-'):
+            ses_str = f'ses-{ses_str}'
+        subject_id += f'_{ses_str}'
+    return subject_id
 
 
 def clean_datasinks(workflow: pe.Workflow) -> pe.Workflow: