[ENH]: Smoothing images as a preprocessing step

docs/changes/newsfragments/161.feature

@@ -0,0 +1 @@
+Introduce :class:`.Smoothing` for smoothing / blurring images as a preprocessing step by `Synchon Mandal`_

junifer/preprocess/__init__.py

@@ -9,3 +9,4 @@
 from .confounds import fMRIPrepConfoundRemover
 from .bold_warper import BOLDWarper
 from .warping import SpaceWarper
+from .smoothing import Smoothing

junifer/preprocess/smoothing/__init__.py

@@ -0,0 +1,6 @@
+"""Provide imports for smoothing sub-package."""
+
+# Authors: Synchon Mandal <[email protected]>
+# License: AGPL
+
+from .smoothing import Smoothing

junifer/preprocess/smoothing/_afni_smoothing.py

@@ -0,0 +1,119 @@
+"""Provide class for smoothing via AFNI."""
+
+# Authors: Synchon Mandal <[email protected]>
+# License: AGPL
+
+from typing import (
+    TYPE_CHECKING,
+    ClassVar,
+    Dict,
+    List,
+    Set,
+    Union,
+)
+
+import nibabel as nib
+
+from ...pipeline import WorkDirManager
+from ...utils import logger, run_ext_cmd
+
+
+if TYPE_CHECKING:
+    from nibabel import Nifti1Image
+
+
+__all__ = ["AFNISmoothing"]
+
+
+class AFNISmoothing:
+    """Class for smoothing via AFNI.
+
+    This class uses AFNI's 3dBlurToFWHM.
+
+    """
+
+    _EXT_DEPENDENCIES: ClassVar[List[Dict[str, Union[str, List[str]]]]] = [
+        {
+            "name": "afni",
+            "commands": ["3dBlurToFWHM"],
+        },
+    ]
+
+    _DEPENDENCIES: ClassVar[Set[str]] = {"nibabel"}
+
+    def preprocess(
+        self,
+        data: "Nifti1Image",
+        fwhm: Union[int, float],
+    ) -> "Nifti1Image":
+        """Preprocess using AFNI.
+
+        Parameters
+        ----------
+        data : Niimg-like object
+            Image(s) to preprocess.
+        fwhm : int or float
+            Smooth until the value. AFNI estimates the smoothing and then
+            applies smoothing to reach ``fwhm``.
+
+        Returns
+        -------
+        Niimg-like object
+            The preprocessed image(s).
+
+        Notes
+        -----
+        For more information on ``3dBlurToFWHM``, check:
+        https://afni.nimh.nih.gov/pub/dist/doc/program_help/3dBlurToFWHM.html
+
+        As the process also depends on the conversion of AFNI files to NIfTI
+        via AFNI's ``3dAFNItoNIFTI``, the help for that can be found at:
+        https://afni.nimh.nih.gov/pub/dist/doc/program_help/3dAFNItoNIFTI.html
+
+        """
+        logger.info("Smoothing using AFNI")
+
+        # Create component-scoped tempdir
+        tempdir = WorkDirManager().get_tempdir(prefix="afni_smoothing")
+
+        # Save target data to a component-scoped tempfile
+        nifti_in_file_path = tempdir / "input.nii"  # needs to be .nii
+        nib.save(data, nifti_in_file_path)
+
+        # Set 3dBlurToFWHM command
+        blur_out_path_prefix = tempdir / "blur"
+        blur_cmd = [
+            "3dBlurToFWHM",
+            f"-input {nifti_in_file_path.resolve()}",
+            f"-prefix {blur_out_path_prefix.resolve()}",
+            "-automask",
+            f"-FWHM {fwhm}",
+        ]
+        # Call 3dBlurToFWHM
+        run_ext_cmd(name="3dBlurToFWHM", cmd=blur_cmd)
+
+        # Create element-scoped tempdir so that the blurred output is
+        # available later as nibabel stores file path reference for
+        # loading on computation
+        element_tempdir = WorkDirManager().get_element_tempdir(
+            prefix="afni_blur"
+        )
+        # Convert afni to nifti
+        blur_afni_to_nifti_out_path = (
+            element_tempdir / "output.nii"  # needs to be .nii
+        )
+        convert_cmd = [
+            "3dAFNItoNIFTI",
+            f"-prefix {blur_afni_to_nifti_out_path.resolve()}",
+            f"{blur_out_path_prefix}+tlrc.BRIK",
+        ]
+        # Call 3dAFNItoNIFTI
+        run_ext_cmd(name="3dAFNItoNIFTI", cmd=convert_cmd)
+
+        # Load nifti
+        output_data = nib.load(blur_afni_to_nifti_out_path)
+
+        # Delete tempdir
+        WorkDirManager().delete_tempdir(tempdir)
+
+        return output_data  # type: ignore

junifer/preprocess/smoothing/_fsl_smoothing.py

@@ -0,0 +1,116 @@
+"""Provide class for smoothing via FSL."""
+
+# Authors: Synchon Mandal <[email protected]>
+# License: AGPL
+
+from typing import (
+    TYPE_CHECKING,
+    ClassVar,
+    Dict,
+    List,
+    Set,
+    Union,
+)
+
+import nibabel as nib
+
+from ...pipeline import WorkDirManager
+from ...utils import logger, run_ext_cmd
+
+
+if TYPE_CHECKING:
+    from nibabel import Nifti1Image
+
+
+__all__ = ["FSLSmoothing"]
+
+
+class FSLSmoothing:
+    """Class for smoothing via FSL.
+
+    This class uses FSL's susan.
+
+    """
+
+    _EXT_DEPENDENCIES: ClassVar[List[Dict[str, Union[str, List[str]]]]] = [
+        {
+            "name": "fsl",
+            "commands": ["susan"],
+        },
+    ]
+
+    _DEPENDENCIES: ClassVar[Set[str]] = {"nibabel"}
+
+    def preprocess(
+        self,
+        data: "Nifti1Image",
+        brightness_threshold: float,
+        fwhm: float,
+    ) -> "Nifti1Image":
+        """Preprocess using FSL.
+
+        Parameters
+        ----------
+        data : Niimg-like object
+            Image(s) to preprocess.
+        brightness_threshold : float
+            Threshold to discriminate between noise and the underlying image.
+            The value should be set greater than the noise level and less than
+            the contrast of the underlying image.
+        fwhm : float
+            Spatial extent of smoothing.
+
+        Returns
+        -------
+        Niimg-like object
+            The preprocessed image(s).
+
+        Notes
+        -----
+        For more information on ``SUSAN``, check [1]_
+
+        References
+        ----------
+        .. [1] Smith, S.M. and Brady, J.M. (1997).
+               SUSAN - a new approach to low level image processing.
+               International Journal of Computer Vision, Volume 23(1),
+               Pages 45-78.
+
+        """
+        logger.info("Smoothing using FSL")
+
+        # Create component-scoped tempdir
+        tempdir = WorkDirManager().get_tempdir(prefix="fsl_smoothing")
+
+        # Save target data to a component-scoped tempfile
+        nifti_in_file_path = tempdir / "input.nii.gz"
+        nib.save(data, nifti_in_file_path)
+
+        # Create element-scoped tempdir so that the output is
+        # available later as nibabel stores file path reference for
+        # loading on computation
+        element_tempdir = WorkDirManager().get_element_tempdir(
+            prefix="fsl_susan"
+        )
+        susan_out_path = element_tempdir / "output.nii.gz"
+        # Set susan command
+        susan_cmd = [
+            "susan",
+            f"{nifti_in_file_path.resolve()}",
+            f"{brightness_threshold}",
+            f"{fwhm}",
+            "3",  # dimension
+            "1",  # use median when no neighbourhood is found
+            "0",  # use input image to find USAN
+            f"{susan_out_path.resolve()}",
+        ]
+        # Call susan
+        run_ext_cmd(name="susan", cmd=susan_cmd)
+
+        # Load nifti
+        output_data = nib.load(susan_out_path)
+
+        # Delete tempdir
+        WorkDirManager().delete_tempdir(tempdir)
+
+        return output_data  # type: ignore

junifer/preprocess/smoothing/_nilearn_smoothing.py

@@ -0,0 +1,69 @@
+"""Provide class for smoothing via nilearn."""
+
+# Authors: Synchon Mandal <[email protected]>
+# License: AGPL
+
+from typing import (
+    TYPE_CHECKING,
+    ClassVar,
+    Literal,
+    Set,
+    Union,
+)
+
+from nilearn import image as nimg
+from numpy.typing import ArrayLike
+
+from ...utils import logger
+
+
+if TYPE_CHECKING:
+    from nibabel import Nifti1Image
+
+
+__all__ = ["NilearnSmoothing"]
+
+
+class NilearnSmoothing:
+    """Class for smoothing via nilearn.
+
+    This class uses :func:`nilearn.image.smooth_img` to smooth image(s).
+
+    """
+
+    _DEPENDENCIES: ClassVar[Set[str]] = {"nilearn"}
+
+    def preprocess(
+        self,
+        data: "Nifti1Image",
+        fwhm: Union[int, float, ArrayLike, Literal["fast"], None],
+    ) -> "Nifti1Image":
+        """Preprocess using nilearn.
+
+        Parameters
+        ----------
+        data : Niimg-like object
+            Image(s) to preprocess.
+        fwhm : scalar, ``numpy.ndarray``, tuple or list of scalar, "fast" or \
+               None
+            Smoothing strength, as a full-width at half maximum, in
+            millimeters:
+
+            * If nonzero scalar, width is identical in all 3 directions.
+            * If ``numpy.ndarray``, tuple, or list, it must have 3 elements,
+              giving the FWHM along each axis. If any of the elements is 0 or
+              None, smoothing is not performed along that axis.
+            * If ``"fast"``, a fast smoothing will be performed with a filter
+              ``[0.2, 1, 0.2]`` in each direction and a normalisation to
+              preserve the local average value.
+            * If None, no filtering is performed (useful when just removal of
+              non-finite values is needed).
+
+        Returns
+        -------
+        Niimg-like object
+            The preprocessed image(s).
+
+        """
+        logger.info("Smoothing using nilearn")
+        return nimg.smooth_img(imgs=data, fwhm=fwhm)  # type: ignore