ENH: Treat session as first-class identifier (#193)

* ENH: Re-enable session-level processing

* ENH: Treat session as first-class identifier

* FIX: Account for session when working with layout

* FIX: Patch niworkflows `Report` out_filename generation

* PIN: Test sdcflows session change

* CI: Remove deprecated pip flag

* STY: black

* FIX: Use empty list if no sessions

* FIX: Ensure product is not empty if no sessions

* FIX: use altered parameter name

Author: mgxd

.github/workflows/pytest.yml

@@ -28,7 +28,7 @@ jobs:
         python -m venv /tmp/venv
         source /tmp/venv/bin/activate
         python -m pip install -U pip
-        python -m pip install --use-feature=in-tree-build ".[test]"
+        python -m pip install ".[test]"
     - name: Run Pytest
       run: |
         source /tmp/venv/bin/activate

nibabies/cli/parser.py

@@ -46,6 +46,9 @@ def _to_gb(value):
     def _drop_sub(value):
         return value[4:] if value.startswith("sub-") else value
 
+    def _drop_ses(value):
+        return value[4:] if value.startswith("ses-") else value
+
     def _filter_pybids_none_any(dct):
         import bids
 
@@ -134,9 +137,14 @@ def _slice_time_ref(value, parser):
         help="a space delimited list of participant identifiers or a single "
         "identifier (the sub- prefix can be removed)",
     )
-    # Re-enable when option is actually implemented
-    # g_bids.add_argument('-s', '--session-id', action='store', default='single_session',
-    #                     help='select a specific session to be processed')
+    g_bids.add_argument(
+        "-s",
+        "--session-id",
+        action="store",
+        nargs="+",
+        type=_drop_ses,
+        help="a space delimited list of session identifiers or a single identifier",
+    )
     # Re-enable when option is actually implemented
     # g_bids.add_argument('-r', '--run-id', action='store', default='single_run',
     #                     help='select a specific run to be processed')

nibabies/cli/run.py

@@ -149,6 +149,7 @@ def main():
         # Generate reports phase
         generate_reports(
             config.execution.participant_label,
+            config.execution.session_id,
             config.execution.nibabies_dir,
             config.execution.run_uuid,
             config=pkgrf("nibabies", "data/reports-spec.yml"),

nibabies/cli/workflow.py

@@ -46,6 +46,12 @@ def build_workflow(config_file, retval):
     subject_list = collect_participants(
         config.execution.layout, participant_label=config.execution.participant_label
     )
+    subjects_sessions = {
+        subject: config.execution.session_id
+        or config.execution.layout.get_sessions(scope='raw', subject=subject)
+        or [None]
+        for subject in subject_list
+    }
 
     # Called with reports only
     if config.execution.reports_only:
@@ -78,7 +84,7 @@ def build_workflow(config_file, retval):
       * Pre-run FreeSurfer's SUBJECTS_DIR: {config.execution.fs_subjects_dir}."""
     build_log.log(25, init_msg)
 
-    retval["workflow"] = init_nibabies_wf()
+    retval["workflow"] = init_nibabies_wf(subjects_sessions)
 
     # Check for FS license after building the workflow
     if not check_valid_fs_license():

nibabies/config.py

@@ -404,6 +404,8 @@ class execution(_Config):
     """Unique identifier of this particular run."""
     segmentation_atlases_dir = None
     """Directory with atlases to use for JLF segmentations"""
+    session_id = None
+    """List of session identifiers that are to be preprocessed."""
     participant_label = None
     """List of participant identifiers that are to be preprocessed."""
     task_id = None

nibabies/reports/core.py

@@ -1,9 +1,44 @@
+from itertools import product
 from pathlib import Path
 
 from niworkflows.reports.core import Report as _Report
+from pkg_resources import resource_filename as pkgrf
 
 
 class Report(_Report):
+    # niworkflows patch to preserve `out_filename` even if subject_id is present
+    def __init__(
+        self,
+        out_dir,
+        run_uuid,
+        config=None,
+        out_filename=None,
+        packagename=None,
+        reportlets_dir=None,
+        subject_id=None,
+    ):
+        self.root = Path(reportlets_dir or out_dir)
+
+        # Initialize structuring elements
+        self.sections = []
+        self.errors = []
+        self.out_dir = Path(out_dir)
+        self.run_uuid = run_uuid
+        self.packagename = packagename
+        self.subject_id = subject_id
+        if subject_id is not None:
+            self.subject_id = subject_id[4:] if subject_id.startswith("sub-") else subject_id
+            # ensure set output filename is preserved
+            if not out_filename:
+                out_filename = f"sub-{self.subject_id}.html"
+
+        self.out_filename = out_filename or "report.html"
+
+        # Default template from niworkflows
+        self.template_path = Path(pkgrf("niworkflows", "reports/report.tpl"))
+        self._load_config(Path(config or pkgrf("niworkflows", "reports/default.yml")))
+        assert self.template_path.exists()
+
     # TODO: Upstream ``Report._load_config`` to niworkflows
     def _load_config(self, config):
         from yaml import safe_load as load
@@ -33,6 +68,7 @@ def run_reports(
     subject_label,
     run_uuid,
     config=None,
+    out_filename='report.html',
     reportlets_dir=None,
     packagename=None,
 ):
@@ -43,30 +79,47 @@ def run_reports(
         out_dir,
         run_uuid,
         config=config,
+        out_filename=out_filename,
         subject_id=subject_label,
         packagename=packagename,
         reportlets_dir=reportlets_dir,
     ).generate_report()
 
 
 def generate_reports(
-    subject_list, output_dir, run_uuid, config=None, work_dir=None, packagename=None
+    subject_list,
+    sessions_list,
+    output_dir,
+    run_uuid,
+    config=None,
+    work_dir=None,
+    packagename=None,
 ):
     """Execute run_reports on a list of subjects."""
     reportlets_dir = None
     if work_dir is not None:
         reportlets_dir = Path(work_dir) / "reportlets"
-    report_errors = [
-        run_reports(
-            output_dir,
-            subject_label,
-            run_uuid,
-            config=config,
-            packagename=packagename,
-            reportlets_dir=reportlets_dir,
+
+    if sessions_list is None:
+        sessions_list = [None]
+
+    report_errors = []
+    for subject_label, session in product(subject_list, sessions_list):
+        html_report = f"sub-{subject_label}"
+        if session:
+            html_report += f"_ses-{session}"
+        html_report += ".html"
+        report_errors.append(
+            run_reports(
+                output_dir,
+                subject_label,
+                run_uuid,
+                config=config,
+                out_filename=html_report,
+                packagename=packagename,
+                reportlets_dir=reportlets_dir,
+            )
         )
-        for subject_label in subject_list
-    ]
 
     errno = sum(report_errors)
     if errno:

nibabies/utils/bids.py

@@ -107,7 +107,7 @@ def _unique(inlist):
     return {k: _unique(v) for k, v in entities.items()}
 
 
-def group_bolds_ref(*, layout, subject):
+def group_bolds_ref(*, layout, subject, session_id):
     """
     Extracts BOLD files from a BIDS dataset and combines them into buckets.
     Files in a bucket share:
@@ -123,6 +123,8 @@ def group_bolds_ref(*, layout, subject):
         Initialized BIDSLayout
     subject : str
         The subject ID
+    session_id : :obj:`str`, None, or ``False``
+        The session identifier. If ``False``, all sessions will be used (default).
 
     Outputs
     -------
@@ -153,14 +155,14 @@ def group_bolds_ref(*, layout, subject):
     # list of all BOLDS encountered
     all_bolds = []
 
-    for ses, suffix in sorted(
-        product(
-            layout.get_sessions(subject=subject, scope="raw") or (None,),
-            {
-                "bold",
-            },
-        )
-    ):
+    # TODO: Simplify with pybids.layout.Query.OPTIONAL
+    sessions = (
+        [session_id]
+        if session_id is not False
+        else layout.get_sessions(subject=subject, scope="raw")
+    )
+
+    for ses, suffix in sorted(product(sessions or (None,), {"bold"})):
         # bold files same session
         bolds = layout.get(suffix=suffix, session=ses, **base_entities)
         # some sessions may not have BOLD scans

nibabies/workflows/base.py

@@ -1,5 +1,36 @@
 # emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
 # vi: set ft=python sts=4 ts=4 sw=4 et:
+#
+# STATEMENT OF CHANGES: This file is derived from sources licensed under the Apache-2.0 terms,
+# and this file has been changed.
+# The original file this work derives from is found at:
+# https://github.com/nipreps/fmriprep/blob/a4fd718/fmriprep/workflows/bold/base.py
+#
+# [January 2022] CHANGES:
+#   * `init_nibabies_wf` now takes in a dictionary composed of participant/session key/values.
+#   * `init_single_subject_wf` now differentiates between participant sessions.
+#     This change is to treat sessions as a "first-class" identifier, to better handle the
+#     potential rapid changing of brain morphometry.
+#
+# Copyright 2021 The NiPreps Developers <[email protected]>
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# We support and encourage derived works from this project, please read
+# about our expectations at
+#
+#     https://www.nipreps.org/community/licensing/
+#
 """
 NiBabies base processing workflows
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -24,7 +55,7 @@
 from .bold import init_func_preproc_wf
 
 
-def init_nibabies_wf():
+def init_nibabies_wf(participants_table):
     """
     Build *NiBabies*'s pipeline.
 
@@ -44,6 +75,10 @@ def init_nibabies_wf():
             with mock_config():
                 wf = init_nibabies_wf()
 
+    Parameters
+    ----------
+    participants_table: :obj:`dict`
+        Keys of participant labels and values of the sessions to process.
     """
     from niworkflows.engine.workflows import LiterateWorkflow as Workflow
     from niworkflows.interfaces.bids import BIDSFreeSurferDir
@@ -66,32 +101,40 @@ def init_nibabies_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, sessions in participants_table.items():
+        for session_id in sessions:
+            single_subject_wf = init_single_subject_wf(subject_id, session_id=session_id)
 
-        single_subject_wf.config["execution"]["crashdump_dir"] = str(
-            config.execution.nibabies_dir / f"sub-{subject_id}" / "log" / config.execution.run_uuid
-        )
-        for node in single_subject_wf._get_all_nodes():
-            node.config = deepcopy(single_subject_wf.config)
-        if freesurfer:
-            nibabies_wf.connect(fsdir, "subjects_dir", single_subject_wf, "inputnode.subjects_dir")
-        else:
-            nibabies_wf.add_nodes([single_subject_wf])
-
-        # Dump a copy of the config file into the log directory
-        log_dir = (
-            config.execution.nibabies_dir / f"sub-{subject_id}" / "log" / config.execution.run_uuid
-        )
-        log_dir.mkdir(exist_ok=True, parents=True)
-        config.to_filename(log_dir / "nibabies.toml")
+            bids_level = [f"sub-{subject_id}"]
+            if session_id:
+                bids_level.append(f"ses-{session_id}")
+
+            log_dir = (
+                config.execution.nibabies_dir.joinpath(*bids_level)
+                / "log"
+                / config.execution.run_uuid
+            )
+
+            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:
+                nibabies_wf.connect(
+                    fsdir, "subjects_dir", single_subject_wf, "inputnode.subjects_dir"
+                )
+            else:
+                nibabies_wf.add_nodes([single_subject_wf])
+
+            # Dump a copy of the config file into the log directory
+            log_dir.mkdir(exist_ok=True, parents=True)
+            config.to_filename(log_dir / "nibabies.toml")
 
     return nibabies_wf
 
 
-def init_single_subject_wf(subject_id):
+def init_single_subject_wf(subject_id, session_id=None):
     """
-    Organize the preprocessing pipeline for a single subject.
+    Organize the preprocessing pipeline for a single subject, at a single session.
 
     It collects and reports information about the subject, and prepares
     sub-workflows to perform anatomical and functional preprocessing.
@@ -114,6 +157,8 @@ def init_single_subject_wf(subject_id):
     ----------
     subject_id : :obj:`str`
         Subject label for this single-subject workflow.
+    session_id : :obj:`str` or None
+        Session identifier.
 
     Inputs
     ------
@@ -130,10 +175,15 @@ def init_single_subject_wf(subject_id):
     from ..utils.misc import fix_multi_source_name
     from .anatomical import init_infant_anat_wf
 
-    name = "single_subject_%s_wf" % subject_id
+    name = (
+        f"single_subject_{subject_id}_{session_id}_wf"
+        if session_id
+        else f"single_subject_{subject_id}_wf"
+    )
     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,
@@ -350,6 +400,7 @@ def init_single_subject_wf(subject_id):
         fmap_estimators = find_estimators(
             layout=config.execution.layout,
             subject=subject_id,
+            sessions=session_id,
             fmapless=False,  # config.workflow.use_syn,
             force_fmapless=False,  # config.workflow.force_syn,
         )
@@ -374,7 +425,11 @@ def init_single_subject_wf(subject_id):
     # 3) total readout time
     from niworkflows.workflows.epi.refmap import init_epi_reference_wf
 
-    _, bold_groupings = group_bolds_ref(layout=config.execution.layout, subject=subject_id)
+    _, bold_groupings = group_bolds_ref(
+        layout=config.execution.layout,
+        subject=subject_id,
+        session_id=session_id,
+    )
     if any(not x for x in bold_groupings):
         print("No BOLD files found for one or more reference groupings")
         return workflow