Merge pull request #267 from juaml/feat/bold-warper

[ENH]: Introduce `BOLDWarper`

docs/changes/newsfragments/267.feature

@@ -0,0 +1 @@
+Introduce :class:`.BOLDWarper` for warping BOLD data via FSL's ``applywarp`` by `Synchon Mandal`_

junifer/preprocess/__init__.py

@@ -7,3 +7,4 @@
 
 from .base import BasePreprocessor
 from .confounds import fMRIPrepConfoundRemover
+from .fsl import BOLDWarper

junifer/preprocess/fsl/__init__.py

@@ -2,3 +2,5 @@
 
 # Authors: Synchon Mandal <s.mandal@fz-juelich.de>
 # License: AGPL
+
+from .bold_warper import BOLDWarper

junifer/preprocess/fsl/bold_warper.py

@@ -0,0 +1,124 @@
+"""Provide class for warping BOLD via FSL FLIRT."""
+
+# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
+# License: AGPL
+
+from typing import (
+    Any,
+    ClassVar,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+    Union,
+)
+
+from ...api.decorators import register_preprocessor
+from ...utils import logger, raise_error
+from ..base import BasePreprocessor
+from .apply_warper import _ApplyWarper
+
+
+@register_preprocessor
+class BOLDWarper(BasePreprocessor):
+    """Class for warping BOLD NIfTI images via FSL FLIRT.
+
+    Parameters
+    ----------
+    reference : str
+        The data type to use as reference for warping.
+
+    """
+
+    _EXT_DEPENDENCIES: ClassVar[
+        List[Dict[str, Union[str, bool, List[str]]]]
+    ] = [
+        {
+            "name": "fsl",
+            "optional": True,
+            "commands": ["applywarp"],
+        },
+    ]
+
+    def __init__(self, reference: str) -> None:
+        """Initialize the class."""
+        self.ref = reference
+        super().__init__(
+            on="BOLD", required_data_types=["BOLD", self.ref, "Warp"]
+        )
+
+    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: List[str]) -> List[str]:
+        """Get output type.
+
+        Parameters
+        ----------
+        input : list of str
+            The input to the preprocessor. The list must contain the
+            available Junifer Data dictionary keys.
+
+        Returns
+        -------
+        list of str
+            The updated list of available Junifer Data object keys after
+            the pipeline step.
+
+        """
+        # Does not add any new keys
+        return input
+
+    def preprocess(
+        self,
+        input: Dict[str, Any],
+        extra_input: Optional[Dict[str, Any]] = None,
+    ) -> Tuple[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.
+
+        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 BOLD using BOLDWarper")
+        # 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."
+            )
+        # 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,
+        )
+        return "BOLD", input

junifer/preprocess/fsl/tests/test_bold_warper.py

@@ -0,0 +1,58 @@
+"""Provide tests for BOLDWarper."""
+
+# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
+# License: AGPL
+
+from typing import List
+
+import pytest
+
+# from junifer.datareader import DefaultDataReader
+# from junifer.pipeline.utils import _check_fsl
+from junifer.preprocess.fsl import BOLDWarper
+
+
+def test_BOLDWarper_init() -> None:
+    """Test BOLDWarper init."""
+    bold_warper = BOLDWarper(reference="T1w")
+    assert bold_warper._on == ["BOLD"]
+
+
+def test_BOLDWarper_get_valid_inputs() -> None:
+    """Test BOLDWarper get_valid_inputs."""
+    bold_warper = BOLDWarper(reference="T1w")
+    assert bold_warper.get_valid_inputs() == ["BOLD"]
+
+
+@pytest.mark.parametrize(
+    "input_",
+    [
+        ["BOLD", "T1w", "Warp"],
+        ["BOLD", "T1w"],
+        ["BOLD"],
+    ],
+)
+def test_BOLDWarper_get_output_type(input_: List[str]) -> None:
+    """Test BOLDWarper get_output_type.
+
+    Parameters
+    ----------
+    input_ : list of str
+        The input data types.
+
+    """
+    bold_warper = BOLDWarper(reference="T1w")
+    assert bold_warper.get_output_type(input_) == input_
+
+
+@pytest.mark.skip(reason="requires testing dataset")
+# @pytest.mark.skipif(
+#     _check_fsl() is False, reason="requires fsl to be in PATH"
+# )
+def test_BOLDWarper_preprocess() -> None:
+    """Test BOLDWarper preprocess."""
+    # Initialize datareader
+    # reader = DefaultDataReader()
+    # Initialize preprocessor
+    # bold_warper = BOLDWarper(reference="T1w")
+    # TODO(synchon): setup datagrabber and run pipeline