diff --git a/docs/changes/newsfragments/267.feature b/docs/changes/newsfragments/267.feature new file mode 100644 index 000000000..73b7626e5 --- /dev/null +++ b/docs/changes/newsfragments/267.feature @@ -0,0 +1 @@ +Introduce :class:`.BOLDWarper` for warping BOLD data via FSL's ``applywarp`` by `Synchon Mandal`_ diff --git a/junifer/preprocess/__init__.py b/junifer/preprocess/__init__.py index d73020c9e..9eed4d5f2 100644 --- a/junifer/preprocess/__init__.py +++ b/junifer/preprocess/__init__.py @@ -7,3 +7,4 @@ from .base import BasePreprocessor from .confounds import fMRIPrepConfoundRemover +from .fsl import BOLDWarper diff --git a/junifer/preprocess/fsl/__init__.py b/junifer/preprocess/fsl/__init__.py index 01c947063..7ae2765d1 100644 --- a/junifer/preprocess/fsl/__init__.py +++ b/junifer/preprocess/fsl/__init__.py @@ -2,3 +2,5 @@ # Authors: Synchon Mandal # License: AGPL + +from .bold_warper import BOLDWarper diff --git a/junifer/preprocess/fsl/bold_warper.py b/junifer/preprocess/fsl/bold_warper.py new file mode 100644 index 000000000..63d1c20a4 --- /dev/null +++ b/junifer/preprocess/fsl/bold_warper.py @@ -0,0 +1,124 @@ +"""Provide class for warping BOLD via FSL FLIRT.""" + +# Authors: Synchon Mandal +# 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 diff --git a/junifer/preprocess/fsl/tests/test_bold_warper.py b/junifer/preprocess/fsl/tests/test_bold_warper.py new file mode 100644 index 000000000..71cfa5160 --- /dev/null +++ b/junifer/preprocess/fsl/tests/test_bold_warper.py @@ -0,0 +1,58 @@ +"""Provide tests for BOLDWarper.""" + +# Authors: Synchon Mandal +# 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