[ENH] Refactor Freesurfer pipelines architecture (#1120)

* refactor freesurfer pipelines

* add forgotten cli

* use absolute imports in task definitions

* allow string or Path for get_subject_id

* fix folder creation in init_input_node

clinica/pipelines/__init__.py

@@ -10,8 +10,7 @@
     statistics_surface,
     statistics_volume,
     statistics_volume_correction,
-    t1_freesurfer,
-    t1_freesurfer_longitudinal,
+    anatomical,
     t1_linear,
     t1_volume,
     t1_volume_create_dartel,

clinica/pipelines/anatomical/__init__.py

@@ -0,0 +1 @@
+from . import freesurfer

clinica/pipelines/anatomical/freesurfer/__init__.py

@@ -0,0 +1 @@
+from . import atlas, longitudinal, t1

clinica/pipelines/anatomical/freesurfer/atlas/__init__.py

@@ -0,0 +1 @@
+from . import cli

clinica/pipelines/t1_freesurfer_atlas/t1_freesurfer_atlas_cli.py → clinica/pipelines/anatomical/freesurfer/atlas/cli.py

@@ -26,19 +26,17 @@ def cli(
 
     from clinica.utils.ux import print_end_pipeline
 
-    from .t1_freeesurfer_atlas_pipeline import T1FreeSurferAtlas
+    from .pipeline import T1FreeSurferAtlas
 
     pipeline = T1FreeSurferAtlas(
         caps_directory=caps_directory,
         atlas_path=atlas_path,
     )
-
     exec_pipeline = (
         pipeline.run(plugin="MultiProc", plugin_args={"n_procs": n_procs})
         if n_procs
         else pipeline.run()
     )
-
     if isinstance(exec_pipeline, Graph):
         print_end_pipeline(
             pipeline_name, pipeline.base_dir, pipeline.base_dir_was_specified

clinica/pipelines/t1_freesurfer_atlas/info.json → clinica/pipelines/anatomical/freesurfer/atlas/info.json

@@ -11,4 +11,4 @@
             "version": ">=6.0.0"
         }
     ]
-}
+}

clinica/pipelines/t1_freesurfer_atlas/t1_freeesurfer_atlas_pipeline.py → clinica/pipelines/anatomical/freesurfer/atlas/pipeline.py

@@ -162,7 +162,7 @@ def _build_core_nodes(self):
         import nipype.interfaces.utility as nutil
         import nipype.pipeline.engine as npe
 
-        from .t1_freesurfer_atlas_utils import compute_atlas, write_tsv_files
+        from .tasks import compute_atlas_task, write_tsv_files_task
 
         # Run an additional two Freesurfer commands if there is an atlas_path specified
         compute_other_atlases = npe.Node(
@@ -173,31 +173,29 @@ def _build_core_nodes(self):
                     "path_to_atlas",
                 ],
                 output_names=["subject_dir", "image_id", "atlas"],
-                function=compute_atlas,
+                function=compute_atlas_task,
             ),
             name="1-ComputeOtherAtlases",
         )
-        compute_other_atlases.inputs.path_to_atlas = self.atlas_path
-        compute_other_atlases.inputs.caps_directory = self.caps_directory
+        compute_other_atlases.inputs.path_to_atlas = str(self.atlas_path)
+        compute_other_atlases.inputs.caps_directory = str(self.caps_directory)
 
         create_tsv = npe.Node(
             interface=nutil.Function(
                 input_names=["subject_dir", "image_id", "atlas"],
                 output_names=["image_id"],
-                function=write_tsv_files,
+                function=write_tsv_files_task,
             ),
             name="2-CreateTsvFiles",
         )
 
         self.connect(
             [
-                # Run compute_atlases command
                 (
                     self.input_node,
                     compute_other_atlases,
                     [("to_process_with_atlases", "to_process_with_atlases")],
                 ),
-                # Generate TSV files
                 (compute_other_atlases, create_tsv, [("subject_dir", "subject_dir")]),
                 (compute_other_atlases, create_tsv, [("image_id", "image_id")]),
                 (compute_other_atlases, create_tsv, [("atlas", "atlas")]),

clinica/pipelines/anatomical/freesurfer/atlas/tasks.py

@@ -0,0 +1,26 @@
+__all__ = ["compute_atlas_task", "write_tsv_files_task"]
+
+
+def compute_atlas_task(
+    caps_directory: str, to_process_with_atlases: tuple, path_to_atlas: str
+) -> tuple:
+    """Adapter for Nipype."""
+    from pathlib import Path
+
+    from clinica.pipelines.anatomical.freesurfer.atlas.utils import compute_atlas
+
+    subject_dir, image_id, atlas = compute_atlas(
+        Path(caps_directory),
+        to_process_with_atlases,
+        Path(path_to_atlas),
+    )
+    return str(subject_dir), image_id, atlas
+
+
+def write_tsv_files_task(subject_dir: str, image_id: str, atlas: str) -> str:
+    """Adapter for Nipype."""
+    from pathlib import Path
+
+    from clinica.pipelines.anatomical.freesurfer.atlas.utils import write_tsv_files
+
+    return write_tsv_files(Path(subject_dir), image_id, atlas)

clinica/pipelines/anatomical/freesurfer/atlas/utils.py

@@ -0,0 +1,125 @@
+from pathlib import Path
+
+__all__ = ["compute_atlas", "write_tsv_files"]
+
+
+def compute_atlas(
+    caps_directory: Path,
+    to_process_with_atlases: tuple,
+    path_to_atlas: Path,
+) -> tuple[Path, str, str]:
+    import subprocess
+    from pathlib import Path
+    from tempfile import mkdtemp
+
+    from clinica.compat import errno
+    from clinica.utils.stream import cprint
+
+    subject_dir = Path()
+    image_id = ""
+    atlas = ""
+
+    for path in path_to_atlas.rglob(f"*{to_process_with_atlases[0]}_6p0.gcs"):
+        hemisphere = Path(path.stem).stem
+        atlas_name = Path(path.stem).suffix[1:].split("_")[0]
+        if "long" not in to_process_with_atlases[1]:
+            sub, ses = to_process_with_atlases[1].split("_")
+            subject_dir = (
+                caps_directory
+                / "subjects"
+                / sub
+                / ses
+                / "t1"
+                / "freesurfer_cross_sectional"
+            )
+            substitute_dir = subject_dir
+            freesurfer_cross = subject_dir / to_process_with_atlases[1]
+            output_path_annot = (
+                freesurfer_cross / "label" / f"{hemisphere}.{atlas_name}.annot"
+            )
+            sphere_reg = freesurfer_cross / "surf" / f"{hemisphere}.sphere.reg"
+            output_path_stats = (
+                freesurfer_cross / "stats" / f"{hemisphere}.{atlas_name}.stats"
+            )
+            subject_id = to_process_with_atlases[1]
+        else:
+            sub, ses, long = to_process_with_atlases[1].split("_")
+            substitute_dir = Path(mkdtemp())
+            substitute_dir.mkdir(exist_ok=True)
+            subject_id = f"{sub}_{ses}.long.{sub}_{long}"
+            subject_dir = (
+                caps_directory
+                / "subjects"
+                / sub
+                / ses
+                / "t1"
+                / long
+                / "freesurfer_longitudinal"
+            )
+            freesurfer_cross = subject_dir / subject_id
+            try:
+                (substitute_dir / subjid).symlink_to(freesurfer_cross)
+            except FileExistsError as e:
+                if e.errno != errno.EEXIST:  # EEXIST: folder already exists
+                    raise e
+            substitute_dir_id = substitute_dir / subject_id
+            output_path_annot = (
+                substitute_dir_id / "label" / f"{hemisphere}.{atlas_name}.annot"
+            )
+            sphere_reg = substitute_dir_id / "surf" / f"{hemisphere}.sphere.reg"
+            output_path_stats = (
+                substitute_dir_id / "stats" / f"{hemisphere}.{atlas_name}.stats"
+            )
+        if not sphere_reg.is_file():
+            cprint(
+                f"The {hemisphere}.sphere.reg file appears to be missing. "
+                f"The data for {to_process_with_atlases[1]} will not be "
+                f"processed with {to_process_with_atlases[0]}",
+                lvl="warning",
+            )
+        mris_ca_label_command = (
+            f"mris_ca_label -sdir {substitute_dir} {subject_id} {hemisphere} "
+            f"{freesurfer_cross}/surf/{hemisphere}.sphere.reg {path} {output_path_annot}"
+        )
+        subprocess.run(mris_ca_label_command, shell=True, capture_output=True)
+        mris_anatomical_stats_command = (
+            f"export SUBJECTS_DIR={substitute_dir}; "
+            f"mris_anatomical_stats -a {output_path_annot} -f {output_path_stats} -b {subject_id} {hemisphere}"
+        )
+        subprocess.run(mris_anatomical_stats_command, shell=True, capture_output=True)
+        image_id, atlas = (
+            to_process_with_atlases[1],
+            to_process_with_atlases[0],
+        )
+    return subject_dir, image_id, atlas
+
+
+def write_tsv_files(subject_dir: Path, image_id: str, atlas: str) -> str:
+    """
+    Generate statistics TSV files in `subjects_dir`/regional_measures folder for `image_id`.
+
+    Notes
+    -----
+    We do not need to check the line "finished without error" in scripts/recon-all.log.
+    If an error occurs, it will be detected by Nipype and the next nodes (including
+    write_tsv_files will not be called).
+    """
+    from clinica.pipelines.anatomical.freesurfer.utils import generate_regional_measures
+    from clinica.utils.stream import cprint
+
+    if "long" in image_id:
+        sub, ses, long = image_id.split("_")
+        folder = f"{sub}_{ses}.long.{sub}_{long}"
+    else:
+        folder = image_id
+    if (subject_dir / folder / "mri" / "aparc+aseg.mgz").is_file():
+        generate_regional_measures(subject_dir, folder, [atlas])
+    else:
+        cprint(
+            msg=(
+                f"{image_id.replace('_', ' | ')} does not contain "
+                f"mri/aseg+aparc.mgz file. Creation of regional_measures/ folder will be skipped."
+            ),
+            lvl="warning",
+        )
+    return image_id

clinica/pipelines/anatomical/freesurfer/longitudinal/__init__.py

@@ -0,0 +1 @@
+from . import cli, correction, template

clinica/pipelines/t1_freesurfer_longitudinal/t1_freesurfer_longitudinal_cli.py → clinica/pipelines/anatomical/freesurfer/longitudinal/cli.py

@@ -34,33 +34,34 @@ def cli(
     https://aramislab.paris.inria.fr/clinica/docs/public/latest/Pipelines/T1_FreeSurfer_Longitudinal/
     """
     import datetime
-    import os
+    from pathlib import Path
 
+    import clinica.pipelines.anatomical.freesurfer.longitudinal.correction.cli as correction_cli
+    import clinica.pipelines.anatomical.freesurfer.longitudinal.template.cli as template_cli
     from clinica.utils.longitudinal import get_participants_long_id
     from clinica.utils.participant import get_subject_session_list
     from clinica.utils.stream import cprint
 
-    from . import t1_freesurfer_longitudinal_correction_cli, t1_freesurfer_template_cli
-    from .longitudinal_utils import save_part_sess_long_ids_to_tsv
+    from .utils import save_part_sess_long_ids_to_tsv
 
     cprint(
         "The t1-freesurfer-longitudinal pipeline is divided into 2 parts:\n"
         "\tt1-freesurfer-unbiased-template pipeline: Creation of unbiased template\n"
         "\tt1-freesurfer-longitudinal-correction pipeline: Longitudinal correction."
     )
-
+    caps_directory = Path(caps_directory)
     if not subjects_sessions_tsv:
         l_part, l_sess = get_subject_session_list(caps_directory, is_bids_dir=False)
         l_long = get_participants_long_id(l_part, l_sess)
         now = datetime.datetime.now().strftime("%H%M%S")
-        subjects_sessions_tsv = now + "_participants.tsv"
+        subjects_sessions_tsv = f"{now}_participants.tsv"
         save_part_sess_long_ids_to_tsv(
-            l_part, l_sess, l_long, os.getcwd(), subjects_sessions_tsv
+            l_part, l_sess, l_long, Path.cwd(), subjects_sessions_tsv
         )
 
     cprint("Part 1/2: Running t1-freesurfer-unbiased-template pipeline.")
     ctx.invoke(
-        t1_freesurfer_template_cli.cli,
+        template_cli.cli,
         caps_directory=caps_directory,
         subjects_sessions_tsv=subjects_sessions_tsv,
         working_directory=working_directory,
@@ -70,7 +71,7 @@ def cli(
 
     cprint("Part 2/2 Running t1-freesurfer-longitudinal-correction pipeline.")
     ctx.invoke(
-        t1_freesurfer_longitudinal_correction_cli.cli,
+        correction_cli.cli,
         caps_directory=caps_directory,
         subjects_sessions_tsv=subjects_sessions_tsv,
         working_directory=working_directory,

clinica/pipelines/anatomical/freesurfer/longitudinal/correction/__init__.py

@@ -0,0 +1 @@
+from . import cli

clinica/pipelines/t1_freesurfer_longitudinal/t1_freesurfer_longitudinal_correction_cli.py → clinica/pipelines/anatomical/freesurfer/longitudinal/correction/cli.py

@@ -35,12 +35,12 @@ def cli(
     """
     from networkx import Graph
 
+    from clinica.pipelines.anatomical.freesurfer.atlas.cli import (
+        cli as t1_freesurfer_atlas_cli,
+    )
     from clinica.utils.ux import print_end_pipeline
 
-    from ..t1_freesurfer_atlas import t1_freesurfer_atlas_cli
-    from .t1_freesurfer_longitudinal_correction_pipeline import (
-        T1FreeSurferLongitudinalCorrection,
-    )
+    from .pipeline import T1FreeSurferLongitudinalCorrection
 
     pipeline = T1FreeSurferLongitudinalCorrection(
         caps_directory=caps_directory,
@@ -49,18 +49,15 @@ def cli(
         name=pipeline_name,
         overwrite_caps=overwrite_outputs,
     )
-
     exec_pipeline = (
         pipeline.run(plugin="MultiProc", plugin_args={"n_procs": n_procs})
         if n_procs
         else pipeline.run()
     )
-
     if isinstance(exec_pipeline, Graph):
         print_end_pipeline(
             pipeline_name, pipeline.base_dir, pipeline.base_dir_was_specified
         )
-
     if atlas_path is not None:
         ctx.invoke(
             t1_freesurfer_atlas_cli.cli,