[REF] Refactor function clip_img (#1392)

* add generic clip_nifti function

* refactor clip_img to use clip_nifti

* provide output directory to clip_task to avoid writting in current folder

* add unit tests

* refactor and test threshold compute logic

clinica/pipelines/pet/linear/pipeline.py

@@ -298,10 +298,11 @@ def _build_core_nodes(self):
             name="clipping",
             interface=nutil.Function(
                 function=clip_task,
-                input_names=["input_pet"],
+                input_names=["input_pet", "output_dir"],
                 output_names=["output_image"],
             ),
         )
+        clipping_node.inputs.output_dir = self.base_dir
 
         # 2. `RegistrationSynQuick` by *ANTS*. It uses nipype interface.
         ants_registration_node = npe.Node(

clinica/pipelines/pet/linear/tasks.py

@@ -18,12 +18,16 @@ def perform_suvr_normalization_task(
 
 def clip_task(
     input_pet: str,
+    output_dir: str = None,
 ) -> str:
     from pathlib import Path
 
     from clinica.pipelines.pet.linear.utils import clip_img
 
-    return str(clip_img(Path(input_pet)))
+    if output_dir:
+        output_dir = Path(output_dir)
+
+    return str(clip_img(Path(input_pet), output_dir=output_dir))
 
 
 def rename_into_caps_task(

clinica/pipelines/pet/linear/utils.py

@@ -9,6 +9,7 @@
     "perform_suvr_normalization",
     "rename_into_caps",
     "print_end_pipeline",
+    "clip_img",
 ]
 
 
@@ -108,42 +109,41 @@ def perform_suvr_normalization(
     return output_image
 
 
-def clip_img(
-    pet_image_path: Path,
-) -> Path:
-    """Clip PET images to remove preprocessing artifacts leading to negative values
+def clip_img(pet_image_path: Path, output_dir: Optional[Path] = None) -> Path:
+    """Clip PET images to remove preprocessing artifacts leading to negative values.
 
     Parameters
     ----------
     pet_image_path : Path
         The path to the image to be processed.
 
+    output_dir : Path, optional
+        The path to the folder in which the output image should be written.
+        If None, the image will be written in the current directory.
+
     Returns
     -------
     output_img : Path
         The path to the normalized nifti image.
     """
-    import nibabel as nib
-    import numpy as np
-
-    from clinica.utils.filemanip import get_filename_no_ext
-
-    pet_image = nib.load(pet_image_path)
+    from clinica.utils.image import clip_nifti
 
-    unique, counts = np.unique(
-        pet_image.get_fdata().reshape(-1), axis=0, return_counts=True
+    return clip_nifti(
+        pet_image_path,
+        low=_compute_clipping_threshold(pet_image_path),
+        output_dir=output_dir,
     )
-    clip_threshold = max(0.0, unique[np.argmax(counts)])
 
-    data = np.clip(
-        pet_image.get_fdata(dtype="float32"), a_min=clip_threshold, a_max=None
-    )
 
-    output_image = Path.cwd() / f"{get_filename_no_ext(pet_image_path)}_clipped.nii.gz"
-    clipped_pet = nib.Nifti1Image(data, pet_image.affine, header=pet_image.header)
-    clipped_pet.to_filename(output_image)
+def _compute_clipping_threshold(pet_image_path: Path) -> float:
+    """Compute the clipping threshold for PET images background cleaning."""
+    import nibabel as nib
+    import numpy as np
 
-    return output_image
+    unique, counts = np.unique(
+        nib.load(pet_image_path).get_fdata().reshape(-1), axis=0, return_counts=True
+    )
+    return max(0.0, unique[np.argmax(counts)])
 
 
 def rename_into_caps(

clinica/utils/image.py

@@ -15,6 +15,7 @@
     "merge_nifti_images_in_time_dimension",
     "remove_dummy_dimension_from_image",
     "crop_nifti",
+    "clip_nifti",
     "get_mni_template",
     "get_mni_cropped_template",
 ]
@@ -96,6 +97,59 @@ def get_new_image_like(old_image: PathLike, new_image_data: np.ndarray) -> Nifti
     return nib.Nifti1Image(new_image_data, old_img.affine, hdr)
 
 
+def clip_nifti(
+    input_image: Path,
+    new_image_suffix: str = "_clipped",
+    low: Optional[float] = None,
+    high: Optional[float] = None,
+    output_dir: Optional[Path] = None,
+) -> Path:
+    """Build a new nifti image from the clipped values of the provided image.
+
+    Parameters
+    ----------
+    input_image : Path
+        The path to the nifti image to be clipped.
+
+    new_image_suffix : str, optional
+        The suffix for building the new image file name.
+        By default, it is '_clipped' such that the output image file name
+        of an image named 'image.nii.gz' will be 'image_clipped.nii.gz'.
+
+    low : float, optional
+        The low threshold for clipping.
+        If None, no thresholding will be applied.
+
+    high : float, optional
+        The high threshold for clipping.
+        If None, no thresholding will be applied.
+
+    output_dir : Path, optional
+        The path to the folder in which the output image should be written.
+        If None, the image will be written in the current directory.
+
+    Returns
+    -------
+    output_image : Path
+        The path to the output clipped image.
+    """
+    from .filemanip import get_filename_no_ext
+
+    clipped = get_new_image_like(
+        input_image,
+        np.clip(
+            nib.load(input_image).get_fdata(dtype="float32"),
+            a_min=low,
+            a_max=high,
+        ),
+    )
+    output_image = (
+        output_dir or Path.cwd()
+    ) / f"{get_filename_no_ext(input_image)}{new_image_suffix}.nii.gz"
+    clipped.to_filename(output_image)
+    return output_image
+
+
 def merge_nifti_images_in_time_dimension(
     images: Tuple[Union[str, PathLike], ...], out_file: Optional[PathLike] = None
 ) -> PathLike:

clinica/utils/testing_utils.py

@@ -11,6 +11,7 @@
 from clinica.pipelines.dwi.utils import DWIDataset
 
 __all__ = [
+    "build_test_image_cubic_object",
     "build_bids_directory",
     "build_caps_directory",
     "build_dwi_dataset",
@@ -21,6 +22,55 @@
 ]
 
 
+def build_test_image_cubic_object(
+    shape: tuple[int, int, int],
+    background_value: float,
+    object_value: float,
+    object_size: int,
+    affine: Optional[np.ndarray] = None,
+) -> nib.Nifti1Image:
+    """Returns a 3D nifti image of a centered cube.
+
+    Parameters
+    ----------
+    shape : (int, int, int)
+        The shape of the image to generate.
+
+    background_value : float
+        The value that should be put in the background.
+
+    object_value : float
+        The value that should be put in the cubic object.
+
+    object_size : int
+        The size of the cube (in number of voxels).
+
+    affine : np.ndarray, optional
+        The affine to use for the generated image.
+        If None, np.eye(4) is used.
+
+    Returns
+    -------
+    nib.Nifti1Image :
+        The test image.
+    """
+    from clinica.utils.image import Bbox3D
+
+    if object_size > min(shape):
+        raise ValueError(
+            f"Cannot generate an image of dimension {shape} with an object of size {object_size} in it..."
+        )
+    bbox_coordinates = []
+    for c in (dim / 2 for dim in shape):
+        bbox_coordinates.append(int(c - object_size / 2))
+        bbox_coordinates.append(int(c + object_size / 2))
+    bbox = Bbox3D.from_coordinates(*bbox_coordinates)
+    data = np.ones(shape, dtype=np.float32) * background_value
+    x, y, z = bbox.get_slices()
+    data[x, y, z] = object_value
+    return nib.Nifti1Image(data, affine=(affine or np.eye(4)))
+
+
 def build_bids_directory(
     directory: Path,
     subjects_sessions: dict,

test/unittests/pipelines/pet/test_pet_linear_utils.py

@@ -1,5 +1,3 @@
-from pathlib import Path
-
 import pytest
 
 from clinica.utils.pet import SUVRReferenceRegion
@@ -67,3 +65,34 @@ def test_rename_into_caps(tmp_path):
     )
     assert b == tmp_path / "sub-01_ses-M000_run-01_space-T1w_rigid.mat"
     assert c is None
+
+
+@pytest.mark.parametrize(
+    "background_value,object_value,expected_threshold",
+    [
+        (-0.01, 0.01, 0.0),
+        (0.0, 0.01, 0.0),
+        (0.01, 0.01, 0.01),
+        (0.6, 0.01, 0.6),
+        (-1.0, -1.0, 0.0),
+    ],
+)
+def test_compute_clipping_threshold(
+    tmp_path,
+    background_value: float,
+    object_value: float,
+    expected_threshold: float,
+):
+    from clinica.pipelines.pet.linear.utils import _compute_clipping_threshold
+    from clinica.utils.testing_utils import build_test_image_cubic_object
+
+    build_test_image_cubic_object(
+        shape=(10, 10, 10),
+        background_value=background_value,
+        object_value=object_value,
+        object_size=4,
+    ).to_filename(tmp_path / "test.nii.gz")
+
+    assert _compute_clipping_threshold(tmp_path / "test.nii.gz") == pytest.approx(
+        expected_threshold
+    )

test/unittests/utils/test_image.py

@@ -5,6 +5,11 @@
 import pytest
 from numpy.testing import assert_array_equal
 
+from clinica.utils.testing_utils import (
+    assert_nifti_equal,
+    build_test_image_cubic_object,
+)
+
 
 @pytest.mark.parametrize(
     "aggregator,expected_value",
@@ -281,3 +286,52 @@ def test_crop_nifti(tmp_path):
     assert_array_equal(
         cropped.get_fdata(), nib.load(get_mni_cropped_template()).get_fdata()
     )
+
+
+@pytest.mark.parametrize(
+    "input_background,input_object,threshold_low,threshold_high,expected_background,expected_object",
+    [
+        (-0.01, 1.0, 0.0, None, 0.0, 1.0),
+        (1.1, 1.0, 1.1, None, 1.1, 1.1),
+        (0.0, 0.0, 1.0, None, 1.0, 1.0),
+        (0.0, 0.0, 0.0, None, 0.0, 0.0),
+        (0.0, 0.0, -0.1, None, 0.0, 0.0),
+        (10.0, 1.0, 5.0, None, 10.0, 5.0),
+        (-0.01, 1.0, None, 0.0, -0.01, 0.0),
+        (-0.01, 1.0, None, 1.0, -0.01, 1.0),
+        (10.0, 1.0, None, 2.0, 2.0, 1.0),
+        (10.0, 1.0, 2.0, 4.0, 4.0, 2.0),
+        (0.0, 0.0, 2.0, 4.0, 2.0, 2.0),
+        (10.0, 1.0, 1.0, 10.0, 10.0, 1.0),
+    ],
+)
+def test_clip_nifti(
+    tmp_path,
+    input_background: float,
+    input_object: float,
+    threshold_low: float,
+    threshold_high: float,
+    expected_background: float,
+    expected_object: float,
+):
+    from clinica.utils.image import clip_nifti
+
+    build_test_image_cubic_object(
+        shape=(10, 10, 10),
+        background_value=input_background,
+        object_value=input_object,
+        object_size=2,
+    ).to_filename(tmp_path / "input.nii.gz")
+
+    clipped = clip_nifti(
+        tmp_path / "input.nii.gz", low=threshold_low, high=threshold_high
+    )
+
+    build_test_image_cubic_object(
+        shape=(10, 10, 10),
+        background_value=expected_background,
+        object_value=expected_object,
+        object_size=2,
+    ).to_filename(tmp_path / "expected.nii.gz")
+
+    assert_nifti_equal(clipped, tmp_path / "expected.nii.gz")