docs/builtin.rst

@@ -729,14 +729,6 @@ Available
      - | Compute a brain mask for the images by guessing the value of the
        | background from the border of the image.
        | See :func:`nilearn.masking.compute_background_mask`
-   * - ``nilearn``'s ICBM152 template gray-matter mask
-     - | ``fetch_icbm152_brain_gm_mask``
-     - ``MNI152NLin2009aAsym``
-     - 0.0.2
-     - | Compute a gray-matter mask from the asymmetrical ICBM152 2009 template,
-       | release a.
-       | See :func:`nilearn.datasets.fetch_icbm152_brain_gm_mask`
-
 
 ..
    Planned

docs/changes/newsfragments/336.removal

@@ -0,0 +1 @@
+Remove ``Power`` coordinates, ``fetch_icbm152_brain_gm_mask`` mask, ``BOLDWarper`` Preprocessor by `Synchon Mandal`_

docs/whats_new.rst

@@ -79,7 +79,7 @@ API Changes
 - Add a positional argument ``using`` for Markers and Preprocessors having
   implementation-based variations, in particular :class:`.ReHoParcels`,
   :class:`.ReHoSpheres`, :class:`.ALFFParcels`, :class:`.ALFFSpheres` and
-  :class:`.BOLDWarper` by `Synchon Mandal`_ (:gh:`311`)
+  ``BOLDWarper`` by `Synchon Mandal`_ (:gh:`311`)
 - Change all ``probseg_`` types to ``VBM_`` types by `Fede Raimondo`_
   (:gh:`320`)
 - Change the subject and session patterns for :class:`.DataladAOMICID1000`,
@@ -120,13 +120,13 @@ Enhancements
   coordinates by `Synchon Mandal`_ and `Fede Raimondo`_ (:gh:`268`)
 - Add ``mode`` as an aggregation function option in
   :func:`.get_aggfunc_by_name` by `Synchon Mandal`_ (:gh:`287`)
-- Adapt :class:`.BOLDWarper` to use FSL or ANTs depending on warp file
+- Adapt ``BOLDWarper`` to use FSL or ANTs depending on warp file
   extension by `Synchon Mandal`_ (:gh:`293`)
 - Rewrite :func:`.compute_brain_mask` to allow variable template fetching via
   templateflow, according to target data by `Synchon Mandal`_ (:gh:`299`)
 - Replace ``requests`` with ``httpx`` for fetching parcellations by `Synchon
   Mandal`_ (:gh:`300`)
-- Allow :class:`.BOLDWarper` to warp BOLD data to other MNI spaces by `Synchon
+- Allow ``BOLDWarper`` to warp BOLD data to other MNI spaces by `Synchon
   Mandal`_ (:gh:`302`)
 - Add support for local ``junifer queue`` via GNU Parallel by `Synchon Mandal`_
   (:gh:`306`)
@@ -168,7 +168,7 @@ Features
   by `Synchon Mandal`_ (:gh:`265`)
 - Introduce ``junifer.preprocess.fsl.apply_warper._ApplyWarper`` to wrap FSL's
   ``applywarp`` by `Synchon Mandal`_ (:gh:`266`)
-- Introduce :class:`.BOLDWarper` for warping BOLD data via FSL's ``applywarp``
+- Introduce ``BOLDWarper`` for warping BOLD data via FSL's ``applywarp``
   by `Synchon Mandal`_ (:gh:`267`)
 - Introduce :class:`.DMCC13Benchmark` to access `DMCC13benchmark dataset
   <https://openneuro.org/datasets/ds003452/versions/1.0.1>`_ by `Synchon
@@ -216,7 +216,7 @@ Miscellaneous
   `Synchon Mandal`_ (:gh:`279`)
 - Add support for accessing FSL ``img2imgcoord`` via Docker wrapper command by
   `Synchon Mandal`_ (:gh:`281`)
-- Make :class:`.BOLDWarper` tool-agnostic by moving it from
+- Make ``BOLDWarper`` tool-agnostic by moving it from
   ``junifer.preprocess.fsl`` to :mod:`junifer.preprocess` by `Synchon Mandal`_
   (:gh:`288`)
 - Add support for accessing ANTs' ``ResampleImage`` via Docker wrapper by
@@ -232,7 +232,7 @@ Miscellaneous
 Deprecations and Removals
 ^^^^^^^^^^^^^^^^^^^^^^^^^
 
-- Deprecate :class:`.BOLDWarper` and mark for removal in v0.0.4 by `Synchon
+- Deprecate ``BOLDWarper`` and mark for removal in v0.0.4 by `Synchon
   Mandal`_ (:gh:`301`)
 
 

junifer/data/coordinates.py

@@ -13,7 +13,7 @@ import pandas as pd
 from numpy.typing import ArrayLike
 
 from ..pipeline import WorkDirManager
-from ..utils import logger, raise_error, run_ext_cmd, warn_with_log
+from ..utils import logger, raise_error, run_ext_cmd
 
 
 # Path to the VOIs
@@ -377,11 +377,6 @@ def load_coordinates(name: str) -> Tuple[ArrayLike, List[str], str]:
     ValueError
         If ``name`` is invalid.
 
-    Warns
-    -----
-    DeprecationWarning
-        If ``Power`` is provided as the ``name``.
-
     """
     # Check for valid coordinates name
     if name not in _available_coordinates:
@@ -389,18 +384,6 @@ def load_coordinates(name: str) -> Tuple[ArrayLike, List[str], str]:
             f"Coordinates {name} not found. "
             f"Valid options are: {list_coordinates()}"
         )
-
-    # Put up deprecation notice
-    if name == "Power":
-        warn_with_log(
-            msg=(
-                "`Power` has been replaced with `Power2011` and will be "
-                "removed in the next release. For now, it's available for "
-                "backward compatibility."
-            ),
-            category=DeprecationWarning,
-        )
-
     # Load coordinates
     t_coord = _available_coordinates[name]
     if isinstance(t_coord.get("path"), Path):

junifer/data/masks.py

@@ -19,7 +19,6 @@ from typing import (
 
 import nibabel as nib
 import numpy as np
-from nilearn.datasets import fetch_icbm152_brain_gm_mask
 from nilearn.image import get_data, new_img_like, resample_to_img
 from nilearn.masking import (
     compute_background_mask,
@@ -28,7 +27,7 @@ from nilearn.masking import (
 )
 
 from ..pipeline import WorkDirManager
-from ..utils import logger, raise_error, run_ext_cmd, warn_with_log
+from ..utils import logger, raise_error, run_ext_cmd
 from .template_spaces import get_template, get_xfm
 from .utils import closest_resolution
 
@@ -121,46 +120,6 @@ def compute_brain_mask(
     return new_img_like(target_img, mask)  # type: ignore
 
 
-def _fetch_icbm152_brain_gm_mask(
-    target_img: "Nifti1Image",
-    **kwargs,
-) -> "Nifti1Image":
-    """Fetch ICBM152 brain mask and resample.
-
-    Parameters
-    ----------
-    target_img : nibabel.Nifti1Image
-        The image to which the mask will be resampled.
-    **kwargs : dict
-        Keyword arguments to be passed to
-        :func:`nilearn.datasets.fetch_icbm152_brain_gm_mask`.
-
-    Returns
-    -------
-    nibabel.Nifti1Image
-        The resampled mask.
-
-    Warns
-    -----
-    DeprecationWarning
-        If this function is used.
-
-    """
-    warn_with_log(
-        msg=(
-            "It is recommended to use ``compute_brain_mask`` with "
-            "``mask_type='gm'``. This function will be removed in the next "
-            "release. For now, it's available for backward compatibility."
-        ),
-        category=DeprecationWarning,
-    )
-    mask = fetch_icbm152_brain_gm_mask(**kwargs)
-    mask = resample_to_img(
-        mask, target_img, interpolation="nearest", copy=True
-    )
-    return mask
-
-
 # A dictionary containing all supported masks and their respective file or
 # data.
 
@@ -194,11 +153,6 @@ _available_masks: Dict[str, Dict[str, Any]] = {
         "func": compute_epi_mask,
         "space": "inherit",
     },
-    "fetch_icbm152_brain_gm_mask": {
-        "family": "Callable",
-        "func": _fetch_icbm152_brain_gm_mask,
-        "space": "MNI152NLin2009aAsym",
-    },
 }
 
 
@@ -292,9 +246,7 @@ def get_mask(  # noqa: C901
     Raises
     ------
     RuntimeError
-        If warp / transformation file extension is not ".mat" or ".h5" or
+        If warp / transformation file extension is not ".mat" or ".h5".
-        if fetch_icbm152_brain_gm_mask is used and requires warping to
-        other template space.
     ValueError
         If extra key is provided in addition to mask name in ``masks`` or
         if no mask is provided or
@@ -402,21 +354,6 @@ def get_mask(  # noqa: C901
             mask_img = extra_input[inherited_mask_item]["data"]
         # Starting with new mask
         else:
-            # Restrict fetch_icbm152_brain_gm_mask if target std space doesn't
-            # match
-            if (
-                mask_name == "fetch_icbm152_brain_gm_mask"
-                and target_std_space != "MNI152NLin2009aAsym"
-            ):
-                raise_error(
-                    (
-                        "``fetch_icbm152_brain_gm_mask`` is deprecated and "
-                        "space transformation to any other template space is "
-                        "prohibited as it will lead to unforeseen errors. "
-                        "``compute_brain_mask`` is a better alternative."
-                    ),
-                    klass=RuntimeError,
-                )
             # Load mask
             mask_object, _, mask_space = load_mask(
                 mask_name, path_only=False, resolution=resolution

junifer/data/tests/test_masks.py

@@ -386,14 +386,6 @@ def test_get_mask_errors() -> None:
                 masks="inherit", target_data=vbm_gm, extra_input=extra_input
             )
 
-        # Block fetch_icbm152_brain_gm_mask space transformation
-        with pytest.raises(RuntimeError, match="prohibited"):
-            get_mask(
-                masks="fetch_icbm152_brain_gm_mask",
-                target_data=vbm_gm,
-                extra_input=extra_input,
-            )
-
 
 @pytest.mark.parametrize(
     "mask_name,function,params,resample",

junifer/preprocess/__init__.py

@@ -7,6 +7,5 @@
 
 from .base import BasePreprocessor
 from .confounds import fMRIPrepConfoundRemover
-from .bold_warper import BOLDWarper
 from .warping import SpaceWarper
 from .smoothing import Smoothing

junifer/preprocess/ants/__init__.py

@@ -1,4 +0,0 @@
-"""Provide imports for ants sub-package."""
-
-# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
-# License: AGPL

junifer/preprocess/ants/ants_apply_transforms_warper.py

@@ -1,185 +0,0 @@
-"""Provide class for warping via ANTs antsApplyTransforms."""
-
-# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
-# License: AGPL
-
-from pathlib import Path
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    ClassVar,
-    Dict,
-    List,
-    Optional,
-    Tuple,
-    Union,
-)
-
-import nibabel as nib
-import numpy as np
-
-from ...pipeline import WorkDirManager
-from ...utils import logger, raise_error, run_ext_cmd
-
-
-if TYPE_CHECKING:
-    from nibabel import Nifti1Image
-
-
-class _AntsApplyTransformsWarper:
-    """Class for warping NIfTI images via ANTs antsApplyTransforms.
-
-    Warps ANTs ``antsApplyTransforms``.
-
-    Parameters
-    ----------
-    reference : str
-        The data type to use as reference for warping.
-    on : str
-        The data type to use for warping.
-
-    Raises
-    ------
-    ValueError
-        If a list was passed for ``on``.
-
-    """
-
-    _EXT_DEPENDENCIES: ClassVar[List[Dict[str, Union[str, List[str]]]]] = [
-        {
-            "name": "ants",
-            "commands": ["ResampleImage", "antsApplyTransforms"],
-        },
-    ]
-
-    def __init__(self, reference: str, on: str) -> None:
-        """Initialize the class."""
-        self.ref = reference
-        # Check only single data type is passed
-        if isinstance(on, list):
-            raise_error("Can only work on single data type, list was passed.")
-        self.on = on
-
-    def _run_apply_transforms(
-        self,
-        input_data: Dict,
-        ref_path: Path,
-        warp_path: Path,
-    ) -> Tuple["Nifti1Image", Path]:
-        """Run ``antsApplyTransforms``.
-
-        Parameters
-        ----------
-        input_data : dict
-            The input data.
-        ref_path : pathlib.Path
-            The path to the reference file.
-        warp_path : pathlib.Path
-            The path to the warp file.
-
-        Returns
-        -------
-        Niimg-like object
-            The warped input image.
-        pathlib.Path
-            The path to the resampled reference image.
-
-        """
-        # Get the min of the voxel sizes from input and use it as the
-        # resolution
-        resolution = np.min(input_data["data"].header.get_zooms()[:3])
-
-        # Create element-specific tempdir for storing post-warping assets
-        tempdir = WorkDirManager().get_element_tempdir(
-            prefix="applytransforms"
-        )
-
-        # Create a tempfile for resampled reference output
-        resample_image_out_path = tempdir / "reference_resampled.nii.gz"
-        # Set ResampleImage command
-        resample_image_cmd = [
-            "ResampleImage",
-            "3",  # image dimension
-            f"{ref_path.resolve()}",
-            f"{resample_image_out_path.resolve()}",
-            f"{resolution}x{resolution}x{resolution}",
-            "0",  # option for spacing and not size
-            "3 3",  # Lanczos windowed sinc
-        ]
-        # Call ResampleImage
-        run_ext_cmd(name="ResampleImage", cmd=resample_image_cmd)
-
-        # Create a tempfile for warped output
-        apply_transforms_out_path = tempdir / "input_warped.nii.gz"
-        # Set antsApplyTransforms command
-        apply_transforms_cmd = [
-            "antsApplyTransforms",
-            "-d 3",
-            "-e 3",
-            "-n LanczosWindowedSinc",
-            f"-i {input_data['path'].resolve()}",
-            # use resampled reference
-            f"-r {resample_image_out_path.resolve()}",
-            f"-t {warp_path.resolve()}",
-            f"-o {apply_transforms_out_path.resolve()}",
-        ]
-        # Call antsApplyTransforms
-        run_ext_cmd(name="antsApplyTransforms", cmd=apply_transforms_cmd)
-
-        # Load nifti
-        output_img = nib.load(apply_transforms_out_path)
-
-        return output_img, resample_image_out_path  # type: ignore
-
-    def preprocess(
-        self,
-        input: Dict[str, Any],
-        extra_input: Optional[Dict[str, Any]] = None,
-    ) -> Tuple[str, Dict[str, Any]]:
-        """Preprocess.
-
-        Parameters
-        ----------
-        input : dict
-            A single input from the Junifer Data object in which to preprocess.
-        extra_input : dict, optional
-            The other fields in the Junifer Data object. Must include the
-            ``Warp`` and ``ref`` value's keys.
-
-        Returns
-        -------
-        str
-            The key to store the output in the Junifer Data object.
-        dict
-            The computed result as dictionary. This will be stored in the
-            Junifer Data object under the key ``data`` of the data type.
-
-        Raises
-        ------
-        ValueError
-            If ``extra_input`` is None.
-
-        """
-        logger.debug("Warping via ANTs using antsApplyTransforms")
-        # Check for extra inputs
-        if extra_input is None:
-            raise_error(
-                f"No extra input provided, requires `Warp` and `{self.ref}` "
-                "data types in particular."
-            )
-        # Retrieve data type info to warp
-        to_warp_input = input
-        # Retrieve data type info to use as reference
-        ref_input = extra_input[self.ref]
-        # Retrieve Warp data
-        warp = extra_input["Warp"]
-        # Replace original data with warped data and add resampled reference
-        # path
-        input["data"], input["reference_path"] = self._run_apply_transforms(
-            input_data=to_warp_input,
-            ref_path=ref_input["path"],
-            warp_path=warp["path"],
-        )
-        # Use reference input's space as warped input's space
-        input["space"] = ref_input["space"]
-        return self.on, input

junifer/preprocess/ants/tests/test_ants_apply_transforms_warper.py

@@ -1,56 +0,0 @@
-"""Provide tests for AntsApplyTransformsWarper."""
-
-# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
-# License: AGPL
-
-import socket
-
-import pytest
-
-from junifer.datagrabber import DMCC13Benchmark
-from junifer.datareader import DefaultDataReader
-from junifer.pipeline.utils import _check_ants
-from junifer.preprocess.ants.ants_apply_transforms_warper import (
-    _AntsApplyTransformsWarper,
-)
-
-
-def test_AntsApplyTransformsWarper_init() -> None:
-    """Test AntsApplyTransformsWarper init."""
-    ants_apply_transforms_warper = _AntsApplyTransformsWarper(
-        reference="T1w", on="BOLD"
-    )
-    assert ants_apply_transforms_warper.ref == "T1w"
-    assert ants_apply_transforms_warper.on == "BOLD"
-
-
-@pytest.mark.skipif(
-    _check_ants() is False, reason="requires ANTs to be in PATH"
-)
-@pytest.mark.skipif(
-    socket.gethostname() != "juseless",
-    reason="only for juseless",
-)
-def test_AntsApplyTransformsWarper_preprocess() -> None:
-    """Test AntsApplyTransformsWarper preprocess."""
-    with DMCC13Benchmark(
-        types=["BOLD", "T1w", "Warp"],
-        sessions=["ses-wave1bas"],
-        tasks=["Rest"],
-        phase_encodings=["AP"],
-        runs=["1"],
-        native_t1w=True,
-    ) as dg:
-        # Read data
-        element_data = DefaultDataReader().fit_transform(
-            dg[("sub-f9057kp", "ses-wave1bas", "Rest", "AP", "1")]
-        )
-        # Preprocess data
-        data_type, data = _AntsApplyTransformsWarper(
-            reference="T1w", on="BOLD"
-        ).preprocess(
-            input=element_data["BOLD"],
-            extra_input=element_data,
-        )
-        assert isinstance(data_type, str)
-        assert isinstance(data, dict)

junifer/preprocess/bold_warper.py

@@ -1,265 +0,0 @@
-"""Provide class for warping BOLD to other template spaces."""
-
-# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
-# License: AGPL
-
-from typing import (
-    Any,
-    ClassVar,
-    Dict,
-    List,
-    Optional,
-    Tuple,
-    Type,
-    Union,
-)
-
-import nibabel as nib
-from templateflow import api as tflow
-
-from ..api.decorators import register_preprocessor
-from ..data import get_template, get_xfm
-from ..pipeline import WorkDirManager
-from ..utils import logger, raise_error, run_ext_cmd
-from .ants.ants_apply_transforms_warper import _AntsApplyTransformsWarper
-from .base import BasePreprocessor
-from .fsl.apply_warper import _ApplyWarper
-
-
-@register_preprocessor
-class BOLDWarper(BasePreprocessor):
-    """Class for warping BOLD NIfTI images.
-
-    .. deprecated:: 0.0.3
-              `BOLDWarper` will be removed in v0.0.4, it is replaced by
-              `SpaceWarper` because the latter works also with T1w data.
-
-    Parameters
-    ----------
-    using : {"fsl", "ants"}
-        Implementation to use for warping:
-
-        * "fsl" : Use FSL's ``applywarp``
-        * "afni" : Use ANTs' ``antsApplyTransforms``
-
-    reference : str
-        The data type to use as reference for warping, can be either a data
-        type like "T1w" or a template space like "MNI152NLin2009cAsym".
-
-    Raises
-    ------
-    ValueError
-        If ``using`` is invalid or
-        if ``reference`` is invalid.
-
-    Notes
-    -----
-    If you are setting ``reference`` to a template space like
-    "MNI152NLin2009cAsym", make sure ANTs is available for the
-    transformation else it will fail during runtime. It is tricky to validate
-    this beforehand and difficult to enforce this as a requirement, hence the
-    heads-up.
-
-    """
-
-    _CONDITIONAL_DEPENDENCIES: ClassVar[List[Dict[str, Union[str, Type]]]] = [
-        {
-            "using": "fsl",
-            "depends_on": _ApplyWarper,
-        },
-        {
-            "using": "ants",
-            "depends_on": _AntsApplyTransformsWarper,
-        },
-    ]
-
-    def __init__(self, using: str, reference: str) -> None:
-        """Initialize the class."""
-        # Validate `using` parameter
-        valid_using = [dep["using"] for dep in self._CONDITIONAL_DEPENDENCIES]
-        if using not in valid_using:
-            raise_error(
-                f"Invalid value for `using`, should be one of: {valid_using}"
-            )
-        self.using = using
-        self.ref = reference
-        # Initialize superclass based on reference
-        if self.ref == "T1w":
-            super().__init__(
-                on="BOLD", required_data_types=["BOLD", self.ref, "Warp"]
-            )
-        elif self.ref in tflow.templates():
-            super().__init__(on="BOLD", required_data_types=["BOLD"])
-        else:
-            raise_error(f"Unknown reference: {self.ref}")
-
-    def get_valid_inputs(self) -> List[str]:
-        """Get valid data types for input.
-
-        Returns
-        -------
-        list of str
-            The list of data types that can be used as input for this
-            preprocessor.
-
-        """
-        return ["BOLD"]
-
-    def get_output_type(self, input_type: str) -> str:
-        """Get output type.
-
-        Parameters
-        ----------
-        input_type : str
-            The data type input to the preprocessor.
-
-        Returns
-        -------
-        str
-            The data type output by the preprocessor.
-
-        """
-        # Does not add any new keys
-        return input_type
-
-    def preprocess(
-        self,
-        input: Dict[str, Any],
-        extra_input: Optional[Dict[str, Any]] = None,
-    ) -> Tuple[Dict[str, Any], Optional[Dict[str, Dict[str, Any]]]]:
-        """Preprocess.
-
-        Parameters
-        ----------
-        input : dict
-            The BOLD input from the Junifer Data object.
-        extra_input : dict, optional
-            The other fields in the Junifer Data object. Must include the
-            ``Warp`` and ``ref`` value's keys if native space transformation is
-            needed.
-
-        Returns
-        -------
-        dict
-            The computed result as dictionary.
-        None
-            Extra "helper" data types as dictionary to add to the Junifer Data
-            object.
-
-        Raises
-        ------
-        ValueError
-            If ``extra_input`` is None when transforming to native space
-            i.e., using "T1w" as reference.
-        RuntimeError
-            If warp / transformation file extension is not ".mat" or ".h5"
-            when transforming to native space or
-            if the BOLD data is in the correct space and does not require
-            warping.
-
-        """
-        logger.info(f"Warping BOLD to {self.ref} space using BOLDWarper")
-        # Transform to native space
-        if self.ref == "T1w":
-            # Check for extra inputs
-            if extra_input is None:
-                raise_error(
-                    "No extra input provided, requires `Warp` and "
-                    f"`{self.ref}` data types in particular."
-                )
-            # Check for warp file type to use correct tool
-            warp_file_ext = extra_input["Warp"]["path"].suffix
-            if warp_file_ext == ".mat":
-                logger.debug("Using FSL with BOLDWarper")
-                # Initialize ApplyWarper for computation
-                apply_warper = _ApplyWarper(reference=self.ref, on="BOLD")
-                # Replace original BOLD data with warped BOLD data
-                _, input = apply_warper.preprocess(
-                    input=input,
-                    extra_input=extra_input,
-                )
-            elif warp_file_ext == ".h5":
-                logger.debug("Using ANTs with BOLDWarper")
-                # Initialize AntsApplyTransformsWarper for computation
-                ants_apply_transforms_warper = _AntsApplyTransformsWarper(
-                    reference=self.ref, on="BOLD"
-                )
-                # Replace original BOLD data with warped BOLD data
-                _, input = ants_apply_transforms_warper.preprocess(
-                    input=input,
-                    extra_input=extra_input,
-                )
-            else:
-                raise_error(
-                    msg=(
-                        "Unknown warp / transformation file extension: "
-                        f"{warp_file_ext}"
-                    ),
-                    klass=RuntimeError,
-                )
-        # Transform to template space
-        else:
-            # Check pre-requirements for space manipulation
-            if self.ref == input["space"]:
-                raise_error(
-                    (
-                        f"Skipped warping as the BOLD data is in {self.ref} "
-                        "space which would mean that you can remove the "
-                        "BOLDWarper from the preprocess step."
-                    ),
-                    klass=RuntimeError,
-                )
-            else:
-                # Get xfm file
-                xfm_file_path = get_xfm(src=input["space"], dst=self.ref)
-                # Get template space image
-                template_space_img = get_template(
-                    space=self.ref,
-                    target_data=input,
-                    extra_input=None,
-                )
-
-                # Create component-scoped tempdir
-                tempdir = WorkDirManager().get_tempdir(prefix="boldwarper")
-                # Create element-scoped tempdir so that warped BOLD is
-                # available later as nibabel stores file path reference for
-                # loading on computation
-                element_tempdir = WorkDirManager().get_element_tempdir(
-                    prefix="boldwarper"
-                )
-
-                # Save template
-                template_space_img_path = tempdir / f"{self.ref}_T1w.nii.gz"
-                nib.save(template_space_img, template_space_img_path)
-
-                # Create a tempfile for warped output
-                warped_bold_path = (
-                    element_tempdir
-                    / f"bold_warped_from_{input['space']}_to_{self.ref}.nii.gz"
-                )
-
-                logger.debug(
-                    f"Using ANTs to warp BOLD "
-                    f"from {input['space']} to {self.ref}"
-                )
-                # Set antsApplyTransforms command
-                apply_transforms_cmd = [
-                    "antsApplyTransforms",
-                    "-d 3",
-                    "-e 3",
-                    "-n LanczosWindowedSinc",
-                    f"-i {input['path'].resolve()}",
-                    f"-r {template_space_img_path.resolve()}",
-                    f"-t {xfm_file_path.resolve()}",
-                    f"-o {warped_bold_path.resolve()}",
-                ]
-                # Call antsApplyTransforms
-                run_ext_cmd(
-                    name="antsApplyTransforms", cmd=apply_transforms_cmd
-                )
-
-                # Modify target data
-                input["data"] = nib.load(warped_bold_path)
-                input["space"] = self.ref
-
-        return input, None

junifer/preprocess/fsl/__init__.py

@@ -1,4 +0,0 @@
-"""Provide imports for fsl sub-package."""
-
-# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
-# License: AGPL

junifer/preprocess/fsl/apply_warper.py

@@ -1,179 +0,0 @@
-"""Provide class for warping via FSL FLIRT."""
-
-# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
-# License: AGPL
-
-from pathlib import Path
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    ClassVar,
-    Dict,
-    List,
-    Optional,
-    Tuple,
-    Union,
-)
-
-import nibabel as nib
-import numpy as np
-
-from ...pipeline import WorkDirManager
-from ...utils import logger, raise_error, run_ext_cmd
-
-
-if TYPE_CHECKING:
-    from nibabel import Nifti1Image
-
-
-class _ApplyWarper:
-    """Class for warping NIfTI images via FSL FLIRT.
-
-    Wraps FSL FLIRT ``applywarp``.
-
-    Parameters
-    ----------
-    reference : str
-        The data type to use as reference for warping.
-    on : str
-        The data type to use for warping.
-
-    Raises
-    ------
-    ValueError
-        If a list was passed for ``on``.
-
-    """
-
-    _EXT_DEPENDENCIES: ClassVar[List[Dict[str, Union[str, List[str]]]]] = [
-        {
-            "name": "fsl",
-            "commands": ["flirt", "applywarp"],
-        },
-    ]
-
-    def __init__(self, reference: str, on: str) -> None:
-        """Initialize the class."""
-        self.ref = reference
-        # Check only single data type is passed
-        if isinstance(on, list):
-            raise_error("Can only work on single data type, list was passed.")
-        self.on = on
-
-    def _run_applywarp(
-        self,
-        input_data: Dict,
-        ref_path: Path,
-        warp_path: Path,
-    ) -> Tuple["Nifti1Image", Path]:
-        """Run ``applywarp``.
-
-        Parameters
-        ----------
-        input_data : dict
-            The input data.
-        ref_path : pathlib.Path
-            The path to the reference file.
-        warp_path : pathlib.Path
-            The path to the warp file.
-
-        Returns
-        -------
-        Niimg-like object
-            The warped input image.
-        pathlib.Path
-            The path to the resampled reference image.
-
-        """
-        # Get the min of the voxel sizes from input and use it as the
-        # resolution
-        resolution = np.min(input_data["data"].header.get_zooms()[:3])
-
-        # Create element-specific tempdir for storing post-warping assets
-        tempdir = WorkDirManager().get_element_tempdir(prefix="applywarp")
-
-        # Create a tempfile for resampled reference output
-        flirt_out_path = tempdir / "reference_resampled.nii.gz"
-        # Set flirt command
-        flirt_cmd = [
-            "flirt",
-            "-interp spline",
-            f"-in {ref_path.resolve()}",
-            f"-ref {ref_path.resolve()}",
-            f"-applyisoxfm {resolution}",
-            f"-out {flirt_out_path.resolve()}",
-        ]
-        # Call flirt
-        run_ext_cmd(name="flirt", cmd=flirt_cmd)
-
-        # Create a tempfile for warped output
-        applywarp_out_path = tempdir / "input_warped.nii.gz"
-        # Set applywarp command
-        applywarp_cmd = [
-            "applywarp",
-            "--interp=spline",
-            f"-i {input_data['path'].resolve()}",
-            f"-r {flirt_out_path.resolve()}",  # use resampled reference
-            f"-w {warp_path.resolve()}",
-            f"-o {applywarp_out_path.resolve()}",
-        ]
-        # Call applywarp
-        run_ext_cmd(name="applywarp", cmd=applywarp_cmd)
-
-        # Load nifti
-        output_img = nib.load(applywarp_out_path)
-
-        return output_img, flirt_out_path  # type: ignore
-
-    def preprocess(
-        self,
-        input: Dict[str, Any],
-        extra_input: Optional[Dict[str, Any]] = None,
-    ) -> Tuple[str, Dict[str, Any]]:
-        """Preprocess.
-
-        Parameters
-        ----------
-        input : dict
-            A single input from the Junifer Data object in which to preprocess.
-        extra_input : dict, optional
-            The other fields in the Junifer Data object. Must include the
-            ``Warp`` and ``ref`` value's keys.
-
-        Returns
-        -------
-        str
-            The key to store the output in the Junifer Data object.
-        dict
-            The computed result as dictionary. This will be stored in the
-            Junifer Data object under the key ``data`` of the data type.
-
-        Raises
-        ------
-        ValueError
-            If ``extra_input`` is None.
-
-        """
-        logger.debug("Warping via FSL using ApplyWarper")
-        # Check for extra inputs
-        if extra_input is None:
-            raise_error(
-                f"No extra input provided, requires `Warp` and `{self.ref}` "
-                "data types in particular."
-            )
-        # Retrieve data type info to warp
-        to_warp_input = input
-        # Retrieve data type info to use as reference
-        ref_input = extra_input[self.ref]
-        # Retrieve Warp data
-        warp = extra_input["Warp"]
-        # Replace original data with warped data and add resampled reference
-        # path
-        input["data"], input["reference_path"] = self._run_applywarp(
-            input_data=to_warp_input,
-            ref_path=ref_input["path"],
-            warp_path=warp["path"],
-        )
-        # Use reference input's space as warped input's space
-        input["space"] = ref_input["space"]
-        return self.on, input

junifer/preprocess/fsl/tests/test_apply_warper.py

@@ -1,45 +0,0 @@
-"""Provide tests for ApplyWarper."""
-
-# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
-# License: AGPL
-
-import socket
-
-import pytest
-
-from junifer.datagrabber import DataladHCP1200
-from junifer.datareader import DefaultDataReader
-from junifer.pipeline.utils import _check_fsl
-from junifer.preprocess.fsl.apply_warper import _ApplyWarper
-
-
-def test_ApplyWarper_init() -> None:
-    """Test ApplyWarper init."""
-    apply_warper = _ApplyWarper(reference="T1w", on="BOLD")
-    assert apply_warper.ref == "T1w"
-    assert apply_warper.on == "BOLD"
-
-
-@pytest.mark.skipif(_check_fsl() is False, reason="requires FSL to be in PATH")
-@pytest.mark.skipif(
-    socket.gethostname() != "juseless",
-    reason="only for juseless",
-)
-def test_ApplyWarper_preprocess() -> None:
-    """Test ApplyWarper preprocess."""
-    with DataladHCP1200(
-        tasks=["REST1"],
-        phase_encodings=["LR"],
-        ica_fix=True,
-    ) as dg:
-        # Read data
-        element_data = DefaultDataReader().fit_transform(
-            dg[("100206", "REST1", "LR")]
-        )
-        # Preprocess data
-        data_type, data = _ApplyWarper(reference="T1w", on="BOLD").preprocess(
-            input=element_data["BOLD"],
-            extra_input=element_data,
-        )
-        assert isinstance(data_type, str)
-        assert isinstance(data, dict)

junifer/preprocess/tests/test_bold_warper.py

@@ -1,159 +0,0 @@
-"""Provide tests for BOLDWarper."""
-
-# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
-# License: AGPL
-
-import socket
-from typing import TYPE_CHECKING, Tuple
-
-import pytest
-from numpy.testing import assert_array_equal, assert_raises
-
-from junifer.datagrabber import DataladHCP1200, DMCC13Benchmark
-from junifer.datareader import DefaultDataReader
-from junifer.pipeline.utils import _check_ants, _check_fsl
-from junifer.preprocess import BOLDWarper
-
-
-if TYPE_CHECKING:
-    from junifer.datagrabber import BaseDataGrabber
-
-
-def test_BOLDWarper_init() -> None:
-    """Test BOLDWarper init."""
-    bold_warper = BOLDWarper(using="ants", reference="T1w")
-    assert bold_warper._on == ["BOLD"]
-
-
-def test_BOLDWarper_get_valid_inputs() -> None:
-    """Test BOLDWarper get_valid_inputs."""
-    bold_warper = BOLDWarper(using="ants", reference="T1w")
-    assert bold_warper.get_valid_inputs() == ["BOLD"]
-
-
-def test_BOLDWarper_get_output_type() -> None:
-    """Test BOLDWarper get_output_type."""
-    bold_warper = BOLDWarper(using="ants", reference="T1w")
-    assert bold_warper.get_output_type("BOLD") == "BOLD"
-
-
-@pytest.mark.parametrize(
-    "datagrabber, element",
-    [
-        [
-            DMCC13Benchmark(
-                types=["BOLD", "T1w", "Warp"],
-                sessions=["ses-wave1bas"],
-                tasks=["Rest"],
-                phase_encodings=["AP"],
-                runs=["1"],
-                native_t1w=True,
-            ),
-            ("sub-f9057kp", "ses-wave1bas", "Rest", "AP", "1"),
-        ],
-        [
-            DataladHCP1200(
-                tasks=["REST1"],
-                phase_encodings=["LR"],
-                ica_fix=True,
-            ),
-            ("100206", "REST1", "LR"),
-        ],
-    ],
-)
-@pytest.mark.skipif(_check_fsl() is False, reason="requires FSL to be in PATH")
-@pytest.mark.skipif(
-    _check_ants() is False, reason="requires ANTs to be in PATH"
-)
-@pytest.mark.skipif(
-    socket.gethostname() != "juseless",
-    reason="only for juseless",
-)
-def test_BOLDWarper_preprocess_to_native(
-    datagrabber: "BaseDataGrabber", element: Tuple[str, ...]
-) -> None:
-    """Test BOLDWarper preprocess.
-
-    Parameters
-    ----------
-    datagrabber : DataGrabber-like object
-        The parametrized DataGrabber objects.
-    element : tuple of str
-        The parametrized elements.
-
-    """
-    with datagrabber as dg:
-        # Read data
-        element_data = DefaultDataReader().fit_transform(dg[element])
-        # Preprocess data
-        data, _ = BOLDWarper(reference="T1w").preprocess(
-            input=element_data["BOLD"],
-            extra_input=element_data,
-        )
-        assert isinstance(data, dict)
-
-
-@pytest.mark.parametrize(
-    "datagrabber, element, space",
-    [
-        [
-            DMCC13Benchmark(
-                types=["BOLD"],
-                sessions=["ses-wave1bas"],
-                tasks=["Rest"],
-                phase_encodings=["AP"],
-                runs=["1"],
-                native_t1w=False,
-            ),
-            ("sub-f9057kp", "ses-wave1bas", "Rest", "AP", "1"),
-            "MNI152NLin2009aAsym",
-        ],
-        [
-            DMCC13Benchmark(
-                types=["BOLD"],
-                sessions=["ses-wave1bas"],
-                tasks=["Rest"],
-                phase_encodings=["AP"],
-                runs=["1"],
-                native_t1w=False,
-            ),
-            ("sub-f9057kp", "ses-wave1bas", "Rest", "AP", "1"),
-            "MNI152NLin6Asym",
-        ],
-    ],
-)
-@pytest.mark.skipif(
-    _check_ants() is False, reason="requires ANTs to be in PATH"
-)
-@pytest.mark.skipif(
-    socket.gethostname() != "juseless",
-    reason="only for juseless",
-)
-def test_BOLDWarper_preprocess_to_multi_mni(
-    datagrabber: "BaseDataGrabber", element: Tuple[str, ...], space: str
-) -> None:
-    """Test BOLDWarper preprocess.
-
-    Parameters
-    ----------
-    datagrabber : DataGrabber-like object
-        The parametrized DataGrabber objects.
-    element : tuple of str
-        The parametrized elements.
-    space : str
-        The parametrized template space to transform to.
-
-    """
-    with datagrabber as dg:
-        # Read data
-        element_data = DefaultDataReader().fit_transform(dg[element])
-        pre_xfm_data = element_data["BOLD"]["data"].get_fdata().copy()
-        # Preprocess data
-        data, _ = BOLDWarper(reference=space).preprocess(
-            input=element_data["BOLD"],
-            extra_input=element_data,
-        )
-        assert isinstance(data, dict)
-        assert data["space"] == space
-        with assert_raises(AssertionError):
-            assert_array_equal(pre_xfm_data, data["data"])