[ENH] Improve BIDS compatibility of PetLinear CAPS output

clinica/pipelines/pet_linear/pet_linear_pipeline.py

@@ -194,18 +194,22 @@ def build_output_node(self):
 
         # Other nodes
         rename_file_node_inputs = [
-            "in_bids_pet",
-            "fname_pet",
-            "fname_trans",
+            "pet_filename_bids",
+            "pet_filename_raw",
+            "transformation_filename_raw",
             "suvr_reference_region",
             "uncropped_image",
         ]
         if self.parameters.get("save_PETinT1w"):
-            rename_file_node_inputs.append("fname_pet_in_t1w")
+            rename_file_node_inputs.append("pet_filename_in_t1w_raw")
         rename_files = npe.Node(
             interface=nutil.Function(
                 input_names=rename_file_node_inputs,
-                output_names=["out_caps_pet", "out_caps_trans", "out_caps_pet_in_T1w"],
+                output_names=[
+                    "pet_filename_caps",
+                    "transformation_filename_caps",
+                    "pet_filename_in_t1w_caps",
+                ],
                 function=rename_into_caps,
             ),
             name="renameFileCAPS",
@@ -229,30 +233,30 @@ def build_output_node(self):
             [
                 (self.input_node, container_path, [("pet", "bids_or_caps_filename")]),
                 (container_path, write_node, [(("container", fix_join, "pet_linear"), "container")]),
-                (self.input_node, rename_files, [("pet", "in_bids_pet")]),
-                (self.output_node, rename_files, [("affine_mat", "fname_trans")]),
-                (rename_files, write_node, [("out_caps_trans", "@transform_mat")]),
+                (self.input_node, rename_files, [("pet", "pet_filename_bids")]),
+                (self.output_node, rename_files, [("affine_mat", "transformation_filename_raw")]),
+                (rename_files, write_node, [("transformation_filename_caps", "@transform_mat")]),
             ]
         )
         if not (self.parameters.get("uncropped_image")):
             self.connect(
                 [
-                    (self.output_node, rename_files, [("outfile_crop", "fname_pet")]),
-                    (rename_files, write_node, [("out_caps_pet", "@registered_pet")]),
+                    (self.output_node, rename_files, [("outfile_crop", "pet_filename_raw")]),
+                    (rename_files, write_node, [("pet_filename_caps", "@registered_pet")]),
                 ]
             )
         else:
             self.connect(
                 [
-                    (self.output_node, rename_files, [("suvr_pet", "fname_pet")]),
-                    (rename_files, write_node, [("out_caps_pet", "@registered_pet")]),
+                    (self.output_node, rename_files, [("suvr_pet", "pet_filename_raw")]),
+                    (rename_files, write_node, [("pet_filename_caps", "@registered_pet")]),
                 ]
             )
         if self.parameters.get("save_PETinT1w"):
             self.connect(
                 [
-                    (self.output_node, rename_files, [("PETinT1w", "fname_pet_in_t1w")]),
-                    (rename_files, write_node, [("out_caps_pet_in_T1w", "@registered_pet_in_t1w")]),
+                    (self.output_node, rename_files, [("PETinT1w", "pet_filename_in_t1w_raw")]),
+                    (rename_files, write_node, [("pet_filename_in_t1w_caps", "@registered_pet_in_t1w")]),
                 ]
             )
         # fmt: on

clinica/pipelines/pet_linear/pet_linear_utils.py

@@ -134,61 +134,145 @@ def crop_nifti(input_img: str, ref_img: str) -> str:
 
 
 def rename_into_caps(
-    in_bids_pet,
-    fname_pet,
-    fname_trans,
-    suvr_reference_region,
-    uncropped_image,
-    fname_pet_in_t1w=None,
+    pet_filename_bids: str,
+    pet_filename_raw: str,
+    transformation_filename_raw: str,
+    suvr_reference_region: str,
+    uncropped_image: bool,
+    pet_filename_in_t1w_raw: str = None,
+    output_dir: str = None,
 ):
     """
     Rename the outputs of the pipelines into CAPS format.
-    Args:
-        in_bids_pet (str): Input BIDS PET to extract the <source_file>
-        fname_pet (str): Preprocessed PET file.
-        fname_trans (str): Transformation file from PET to MRI space
-        suvr_reference_region (str): SUVR mask name for file name output
-        uncropped_image (bool): Pipeline argument for image cropping
-        fname_pet_in_t1w (bool): Pipeline argument for saving intermediate file
-    Returns:
-        The different outputs in CAPS format
+
+    More precisely, the following files will be renamed:
+
+        - 'pet_filename_raw' will be renamed to 'pet_filename_caps'
+        - 'transformation_filename_raw' will be renamed to 'transformation_filename_caps'
+        - 'pet_filename_in_t1w_raw' will be renamed (if provided) to 'pet_filename_in_t1w_caps'
+
+    Parameters
+    ----------
+    pet_filename_bids : str
+        Input PET image file from the BIDS dataset.
+        This file is used to extract the entities from the source
+        file like subject, session, run, tracer...
+
+    pet_filename_raw : str
+        Preprocessed PET file as outputted by Nipype.
+
+    transformation_filename_raw : str
+        Transformation file from PET to MRI space as outputted by Nipype.
+
+    suvr_reference_region : str
+        SUVR mask name for file name output.
+        This will be used to derive the prefix of the PET image.
+
+    uncropped_image : bool
+        Pipeline argument for image cropping.
+        This will be used to derive the prefix of the PET image.
+
+    pet_filename_in_t1w_raw : str, optional
+        Intermediate PET in T1w MRI space.
+        If not provided, no renaming will be done.
+
+    output_dir : str, optional
+        Specify the output folder where the renamed files should
+        be written. This is mostly used for testing purposes in order
+        to enable this function to write to Pytest's temporary folders.
+        When used in the pipeline, this is left as None as other nodes
+        are responsible for adding the missing pieces to the output paths
+        and writing the files to disk.
+
+    Returns
+    -------
+    pet_filename_caps : str
+        The renamed preprocessed PET file to match CAPS conventions.
+
+    transformation_filename_caps : str
+        The transformation file from PET to MRI space renamed to match CAPS conventions.
+
+    pet_filename_in_t1w_caps : str or None
+        Intermediate PET in T1w MRI space renamed to match CAPS conventions.
+        If 'pet_filename_in_t1w_raw' is None, this will be None.
     """
     import os
 
-    from nipype.interfaces.utility import Rename
-    from nipype.utils.filemanip import split_filename
+    from clinica.pipelines.pet_linear.pet_linear_utils import (  # noqa
+        _get_bids_entities_without_suffix,
+        _rename_intermediate_pet_in_t1w_space_into_caps,
+        _rename_pet_into_caps,
+        _rename_transformation_into_caps,
+    )
 
-    _, source_file_pet, _ = split_filename(in_bids_pet)
-
-    # Rename into CAPS PET:
-    rename_pet = Rename()
-    rename_pet.inputs.in_file = fname_pet
-    if not uncropped_image:
-        suffix = f"_space-MNI152NLin2009cSym_desc-Crop_res-1x1x1_suvr-{suvr_reference_region}_pet.nii.gz"
-        rename_pet.inputs.format_string = source_file_pet + suffix
-    else:
-        suffix = f"_space-MNI152NLin2009cSym_res-1x1x1_suvr-{suvr_reference_region}_pet.nii.gz"
-        rename_pet.inputs.format_string = source_file_pet + suffix
-    out_caps_pet = rename_pet.run().outputs.out_file
-
-    # Rename into CAPS transformation file:
-    rename_trans = Rename()
-    rename_trans.inputs.in_file = fname_trans
-    rename_trans.inputs.format_string = source_file_pet + "_space-T1w_rigid.mat"
-    out_caps_trans = rename_trans.run().outputs.out_file
-
-    # Rename intermediate PET in T1w MRI space
-    if fname_pet_in_t1w is not None:
-        rename_pet_in_t1w = Rename()
-        rename_pet_in_t1w.inputs.in_file = fname_pet_in_t1w
-        rename_pet_in_t1w.inputs.format_string = (
-            source_file_pet + "_space-T1w_pet.nii.gz"
+    bids_entities = _get_bids_entities_without_suffix(pet_filename_bids, suffix="pet")
+    if output_dir:
+        bids_entities = os.path.join(output_dir, bids_entities)
+    pet_filename_caps = _rename_pet_into_caps(
+        bids_entities, pet_filename_raw, not uncropped_image, suvr_reference_region
+    )
+    transformation_filename_caps = _rename_transformation_into_caps(
+        bids_entities, transformation_filename_raw
+    )
+    pet_filename_in_t1w_caps = None
+    if pet_filename_in_t1w_raw is not None:
+        pet_filename_in_t1w_caps = _rename_intermediate_pet_in_t1w_space_into_caps(
+            bids_entities, pet_filename_in_t1w_raw
         )
-        out_caps_pet_in_t1w = rename_pet_in_t1w.run().outputs.out_file
-    else:
-        out_caps_pet_in_t1w = None
 
-    return out_caps_pet, out_caps_trans, out_caps_pet_in_t1w
+    return pet_filename_caps, transformation_filename_caps, pet_filename_in_t1w_caps
+
+
+def _get_bids_entities_without_suffix(filename: str, suffix: str) -> str:
+    """Return the BIDS entities without the suffix from a BIDS path."""
+    from nipype.utils.filemanip import split_filename
+
+    _, stem, _ = split_filename(filename)
+    return stem.rstrip(f"_{suffix}")
+
+
+def _rename_pet_into_caps(
+    entities: str, filename: str, cropped: bool, suvr_reference_region: str
+) -> str:
+    """Rename into CAPS PET."""
+    return _rename(
+        filename, entities, _get_pet_bids_components(cropped, suvr_reference_region)
+    )
+
+
+def _rename_transformation_into_caps(entities: str, filename: str) -> str:
+    """Rename into CAPS transformation file."""
+    return _rename(filename, entities, "_space-T1w_rigid.mat")
+
+
+def _rename_intermediate_pet_in_t1w_space_into_caps(
+    entities: str, filename: str
+) -> str:
+    """Rename intermediate PET in T1w MRI space."""
+    return _rename(filename, entities, "_space-T1w_pet.nii.gz")
+
+
+def _rename(filename: str, entities: str, suffix: str):
+    """Rename 'filename' into '{entities}{suffix}'."""
+    from nipype.interfaces.utility import Rename
+
+    rename = Rename()
+    rename.inputs.in_file = filename
+    rename.inputs.format_string = entities + suffix
+
+    return rename.run().outputs.out_file
+
+
+def _get_pet_bids_components(cropped: bool, suvr_reference_region: str) -> str:
+    """Return a string composed of the PET-specific entities (space, resolution,
+    desc, and suvr), suffix, and extension.
+    """
+    space = "_space-MNI152NLin2009cSym"
+    resolution = "_res-1x1x1"
+    desc = "_desc-Crop" if cropped else ""
+    suvr = f"_suvr-{suvr_reference_region}"
+
+    return f"{space}{desc}{resolution}{suvr}_pet.nii.gz"
 
 
 def print_end_pipeline(pet, final_file):

test/unittests/pipelines/pet_linear/test_pet_linear_utils.py

@@ -0,0 +1,63 @@
+from pathlib import Path
+
+import pytest
+
+
+@pytest.mark.parametrize(
+    "cropped,suvr_reference_region,expected",
+    [
+        (False, "foo", "_space-MNI152NLin2009cSym_res-1x1x1_suvr-foo_pet.nii.gz"),
+        (
+            True,
+            "bar",
+            "_space-MNI152NLin2009cSym_desc-Crop_res-1x1x1_suvr-bar_pet.nii.gz",
+        ),
+    ],
+)
+def test_get_pet_bids_components(cropped, suvr_reference_region, expected):
+    from clinica.pipelines.pet_linear.pet_linear_utils import _get_pet_bids_components
+
+    assert _get_pet_bids_components(cropped, suvr_reference_region) == expected
+
+
+def test_rename(tmp_path):
+    from clinica.pipelines.pet_linear.pet_linear_utils import _rename
+
+    file_to_rename = tmp_path / "foo.nii.gz"
+    file_to_rename.touch()
+    source_file = tmp_path / "sub-01_ses-M000_run-01_pet.json"
+    source_file.touch()
+
+    output_file = _rename(
+        str(file_to_rename), str(tmp_path / "sub-01_ses-M000_run-01"), "_pet.nii.gz"
+    )
+    assert file_to_rename.exists()
+    assert source_file.exists()
+    assert Path(output_file) == tmp_path / "sub-01_ses-M000_run-01_pet.nii.gz"
+    assert Path(output_file).exists()
+
+
+def test_rename_into_caps(tmp_path):
+    from clinica.pipelines.pet_linear.pet_linear_utils import rename_into_caps
+
+    pet_file_to_rename = tmp_path / "foobarbaz.nii.gz"
+    pet_file_to_rename.touch()
+    transformation_file_to_rename = tmp_path / "transformation.mat"
+    transformation_file_to_rename.touch()
+    source_file = tmp_path / "sub-01_ses-M000_run-01_pet.nii.gz"
+    source_file.touch()
+    a, b, c = rename_into_caps(
+        str(source_file),
+        str(pet_file_to_rename),
+        str(transformation_file_to_rename),
+        "suvrfoo",
+        True,
+        output_dir=tmp_path,  # Force the writing to tmp_path instead of current folder...
+    )
+    assert (
+        Path(a)
+        == tmp_path
+        / "sub-01_ses-M000_run-01_space-MNI152NLin2009cSym_res-1x1x1_suvr-suvrfoo_pet.nii.gz"
+    )
+    assert Path(b) == tmp_path / "sub-01_ses-M000_run-01_space-T1w_rigid.mat"
+    assert c is None