diff --git a/docs/changes/newsfragments/452.doc b/docs/changes/newsfragments/452.doc new file mode 100644 index 000000000..7635ee2ee --- /dev/null +++ b/docs/changes/newsfragments/452.doc @@ -0,0 +1 @@ +Add documentation on :class:`.ConfigManager`, dumping data object and extending data dump assets by `Synchon Mandal`_ diff --git a/docs/changes/newsfragments/452.feature b/docs/changes/newsfragments/452.feature new file mode 100644 index 000000000..dff9d2ea0 --- /dev/null +++ b/docs/changes/newsfragments/452.feature @@ -0,0 +1 @@ +Allow pipeline data object to be dumped by introducing :class:`.DataObjectDumper` and available dumping / loading assets to be extended by introducing :func:`.register_data_dump_asset` decorator by `Synchon Mandal`_ diff --git a/docs/extending/data_dump_asset.rst b/docs/extending/data_dump_asset.rst new file mode 100644 index 000000000..e92843aac --- /dev/null +++ b/docs/extending/data_dump_asset.rst @@ -0,0 +1,46 @@ +.. include:: ../links.inc + +.. _adding_data_dump_assets: + +Adding Data Dump Assets +======================= + +``junifer`` supports dumping ``nibabel.Nifti1Image`` and ``pandas.DataFrame`` which should cover most fMRI use cases. But, in case you work with other modalities like EEG, you can register your own data dump asset. + +How to add a data dump asset +---------------------------- + +This example shows how to create a data dump asset for ``mne.io.Raw``. + +#. Check :ref:`extending junifer ` on how to create a + *junifer extension* if you have not done so. +#. Create the data registry in the *extension script* like so: + + .. code-block:: python + + from pathlib import Path + + from junifer.api.decorators import register_data_dump_asset + from junifer.pipeline import BaseDataDumpAsset + + import mne + + @register_data_dump_asset([mne.io.Raw], [".fif", ".fif.gz"]) + class RawAsset(BaseDataDumpAsset): + """Class for ``mne.io.Raw`` dumper.""" + + def dump(self) -> None: + self.data.save(self.path_without_ext.with_suffix(".raw.fif.gz")) + + @classmethod + def load(cls: "RawAsset", path: Path) -> mne.io.Raw: + return mne.io.Raw(path) + + * :func:`.register_data_dump_asset` registers a class. The first argument is + a list of types that the class is responsible for saving and the second + argument is a list of file extensions that the class is responsible for + loading. + * Inheriting from ``junifer.pipeline.BaseDumpAsset`` takes care of the class + acting as a data dump asset. + * Method ``junifer.pipeline.BaseDumpAsset.dump`` and class method + ``junifer.pipeline.BaseDumpAsset.load`` need to be implemented. diff --git a/docs/extending/index.rst b/docs/extending/index.rst index a6ed8ca12..2f3348dfb 100644 --- a/docs/extending/index.rst +++ b/docs/extending/index.rst @@ -33,3 +33,4 @@ DataGrabbers, Preprocessors, Markers, etc., following the *junifer* way. plugins data_registries data_types + data_dump_asset diff --git a/docs/using/configuring.rst b/docs/using/configuring.rst new file mode 100644 index 000000000..94fc5fcbb --- /dev/null +++ b/docs/using/configuring.rst @@ -0,0 +1,65 @@ +.. include:: ../links.inc + +.. _configuring: + +Configuring Pipeline Behaviour +============================== + +It is also possible to configure some internal :ref:`pipeline ` behaviour via :obj:`.ConfigManager`. +It can be done either via the command-line interface (CLI) or the application programming +interface (API). + +To use via the CLI, one would do: + +.. code-block:: bash + + = junifer run ... + +and via the API like so: + +.. code-block:: python + + from junifer.utils import config + + # Add config + config.set(key="", val=) + + # do feature extraction + ... + + # Remove config + config.delete("") + +.. _available_configurations: + +Available Configurations +------------------------ + +.. list-table:: + :widths: auto + :header-rows: 1 + + * - CLI Key + - API Key + - Value + - Description + * - ``JUNIFER_DATA_LOCATION`` + - ``data.location`` + - str + - Alternative location for ``junifer-data`` + * - ``JUNIFER_DATAGRABBER_SKIPIDCHECK`` + - ``datagrabber.skipidcheck`` + - bool + - Skip DataLad-based DataGrabber's ID check + * - ``JUNIFER_DATAGRABBER_SKIPDIRTYCHECK`` + - ``datagrabber.skipdirtycheck`` + - bool + - Skip Git "dirty" check for a DataLad dataset clone of a DataGrabber + * - ``JUNIFER_PREPROCESSING_DUMP_LOCATION`` + - ``preprocessing.dump.location`` + - str + - Dump location of pre-processed data for debugging purposes + * - ``JUNIFER_PREPROCESSING_DUMP_GRANULARITY`` + - ``preprocessing.dump.granularity`` + - "full" or "final" + - Dump all pre-processing steps or just the final pre-processed data diff --git a/docs/using/dumping.rst b/docs/using/dumping.rst new file mode 100644 index 000000000..9145419ea --- /dev/null +++ b/docs/using/dumping.rst @@ -0,0 +1,18 @@ +.. include:: ../links.inc + +.. _dumping: + +Dumping Pipeline Data +===================== + +It is usually not required to check the :ref:`data object ` but +for debugging purposes one can do that. Since, there is no direct way to interact +with it, ``junifer`` provides a way to dump the object before, in between and / or after +pre-processing. It is controlled by the ``JUNIFER_PREPROCESSING_DUMP_LOCATION`` and +``JUNIFER_PREPROCESSING_DUMP_GRANULARITY`` :ref:`configuration options `. + +``junifer`` dumps the ``"data"`` attribute of the data object as proper files in their respective formats, +for example, ``nibabel.Nifti1Image`` gets dumped as a ``.nii.gz`` file. As of now, ``junifer`` can dump +``nibabel.Nifti1Image`` and ``pandas.DataFrame`` (confound files) file formats. In case you +:ref:`add custom data types ` which support different file formats, you can create and +register a :ref:`custom dumper `. diff --git a/docs/using/index.rst b/docs/using/index.rst index 545f57035..338d9e72c 100644 --- a/docs/using/index.rst +++ b/docs/using/index.rst @@ -18,6 +18,8 @@ to interact with HPC and HTC systems. codeless running queueing + configuring + dumping .. _using_components: diff --git a/junifer/api/decorators.py b/junifer/api/decorators.py index aa0f377d5..3f8ea56f6 100644 --- a/junifer/api/decorators.py +++ b/junifer/api/decorators.py @@ -6,8 +6,13 @@ # License: AGPL from ..data import DataDispatcher -from ..pipeline import PipelineComponentRegistry +from ..pipeline import ( + AssetDumperDispatcher, + AssetLoaderDispatcher, + PipelineComponentRegistry, +) from ..typing import ( + DataDumpAssetLike, DataGrabberLike, DataRegistryLike, MarkerLike, @@ -17,6 +22,7 @@ from ..typing import ( __all__ = [ + "register_data_dump_asset", "register_data_registry", "register_datagrabber", "register_datareader", @@ -184,3 +190,49 @@ def register_data_registry(name: str) -> DataRegistryLike: return klass return decorator + + +def register_data_dump_asset( + types: list[type], exts: list[str] +) -> DataDumpAssetLike: + """Asset registration decorator. + + Registers the data dump asset for ``types`` with ``exts``. + + Parameters + ---------- + types : list of class + The classes to dump. + exts : list of str + The extensions to load. + + Returns + ------- + class + The unmodified input class. + + """ + + def decorator(klass: DataDumpAssetLike) -> DataDumpAssetLike: + """Actual decorator. + + Parameters + ---------- + klass : class + The class of the data dump asset to register. + + Returns + ------- + class + The unmodified input class. + + """ + # Add asset dumper + for t in types: + AssetDumperDispatcher()[t] = klass + # Add asset loader + for e in exts: + AssetLoaderDispatcher()[e] = klass + return klass + + return decorator diff --git a/junifer/api/tests/test_decorators.py b/junifer/api/tests/test_decorators.py index 8dec54483..2439cea2a 100644 --- a/junifer/api/tests/test_decorators.py +++ b/junifer/api/tests/test_decorators.py @@ -3,8 +3,18 @@ # Authors: Synchon Mandal # License: AGPL -from junifer.api.decorators import register_data_registry +import pickle + +from junifer.api.decorators import ( + register_data_dump_asset, + register_data_registry, +) from junifer.data import BasePipelineDataRegistry, DataDispatcher +from junifer.pipeline import ( + AssetDumperDispatcher, + AssetLoaderDispatcher, + BaseDataDumpAsset, +) def test_register_data_registry() -> None: @@ -30,3 +40,39 @@ def test_register_data_registry() -> None: assert "dumb" in DataDispatcher() _ = DataDispatcher().pop("dumb") assert "dumb" not in DataDispatcher() + + +def test_register_data_dump_asset() -> None: + """Test data dump asset registration.""" + + class Int(int): ... + + class Float(float): ... + + @register_data_dump_asset([Int, Float], [".int", ".float"]) + class DumAsset(BaseDataDumpAsset): + def dump(self): + suffix = "" + if isinstance(self.data, Int): + suffix = ".int" + else: + suffix = ".float" + pickle.dump(self.data, self.path_without_ext.with_suffix(suffix)) + + @classmethod + def load(cls, path): + return pickle.load(path) + + assert Int in AssetDumperDispatcher() + assert Float in AssetDumperDispatcher() + _ = AssetDumperDispatcher().pop(Int) + _ = AssetDumperDispatcher().pop(Float) + assert Int not in AssetDumperDispatcher() + assert Float not in AssetDumperDispatcher() + + assert ".int" in AssetLoaderDispatcher() + assert ".float" in AssetLoaderDispatcher() + _ = AssetLoaderDispatcher().pop(".int") + _ = AssetLoaderDispatcher().pop(".float") + assert ".int" not in AssetLoaderDispatcher() + assert ".float" not in AssetLoaderDispatcher() diff --git a/junifer/pipeline/__init__.pyi b/junifer/pipeline/__init__.pyi index f3b55a576..f2c45a38d 100644 --- a/junifer/pipeline/__init__.pyi +++ b/junifer/pipeline/__init__.pyi @@ -1,4 +1,8 @@ __all__ = [ + "AssetDumperDispatcher", + "AssetLoaderDispatcher", + "BaseDataDumpAsset", + "DataObjectDumper", "PipelineComponentRegistry", "PipelineStepMixin", "UpdateMetaMixin", @@ -6,6 +10,12 @@ __all__ = [ "MarkerCollection", ] +from ._data_object_dumper import ( + AssetDumperDispatcher, + AssetLoaderDispatcher, + BaseDataDumpAsset, + DataObjectDumper, +) from .pipeline_component_registry import PipelineComponentRegistry from .pipeline_step_mixin import PipelineStepMixin from .update_meta_mixin import UpdateMetaMixin diff --git a/junifer/pipeline/_data_object_dumper.py b/junifer/pipeline/_data_object_dumper.py new file mode 100644 index 000000000..ec13eb76c --- /dev/null +++ b/junifer/pipeline/_data_object_dumper.py @@ -0,0 +1,347 @@ +"""Provide pipeline data object dumper and data dump asset classes.""" + +# Authors: Synchon Mandal +# License: AGPL + +from abc import ABC, abstractmethod +from collections.abc import Iterator, MutableMapping +from copy import deepcopy +from pathlib import Path +from typing import Any + +import nibabel +import pandas + +from ..utils import raise_error, yaml + + +__all__ = [ + "AssetDumperDispatcher", + "AssetLoaderDispatcher", + "BaseDataDumpAsset", + "DataObjectDumper", +] + + +class BaseDataDumpAsset(ABC): + """Abstract base class for a data dump asset. + + Parameters + ---------- + data : Any + Data to save. + path_without_ext : pathlib.Path + Path to the asset without extension. + The subclass should add the extension when saving. + + """ + + def __init__(self, data: Any, path_without_ext: Path) -> None: + """Initialize the class.""" + self.data = data + self.path_without_ext = path_without_ext + + @abstractmethod + def dump(self) -> None: + """Dump asset.""" + raise_error( + msg="Concrete classes need to implement dump().", + klass=NotImplementedError, + ) + + @classmethod + @abstractmethod + def load(cls: type["BaseDataDumpAsset"], path: Path) -> Any: + """Load asset from path.""" + raise_error( + msg="Concrete classes need to implement load().", + klass=NotImplementedError, + ) + + +class Nifti1ImageAsset(BaseDataDumpAsset): + """Class for ``nibabel.Nifti1Image`` dumper.""" + + def dump(self) -> None: + nibabel.save(self.data, self.path_without_ext.with_suffix(".nii.gz")) + + @classmethod + def load(cls: "Nifti1ImageAsset", path: Path) -> nibabel.Nifti1Image: + return nibabel.load(path) + + +class PandasDataFrameAsset(BaseDataDumpAsset): + """Class for ``pandas.DataFrame`` dumper.""" + + def dump(self) -> None: + self.data.to_csv(self.path_without_ext.with_suffix(".csv")) + + @classmethod + def load(cls: "PandasDataFrameAsset", path: Path) -> pandas.DataFrame: + return pandas.read_csv(path, index_col=0) + + +class AssetDumperDispatcher(MutableMapping): + """Class for helping dynamic asset dumper dispatch.""" + + _instance = None + + def __new__(cls): + # Make class singleton + if cls._instance is None: + cls._instance = super().__new__(cls) + # Set dumpers + cls._dumpers: dict[type, type[BaseDataDumpAsset]] = {} + cls._builtin: dict[type, type[BaseDataDumpAsset]] = {} + cls._external: dict[type, type[BaseDataDumpAsset]] = {} + cls._builtin.update( + { + nibabel.Nifti1Image: Nifti1ImageAsset, + pandas.DataFrame: PandasDataFrameAsset, + } + ) + cls._dumpers.update(cls._builtin) + return cls._instance + + def __getitem__(self, key: type) -> type[BaseDataDumpAsset]: + return self._dumpers[key] + + def __iter__(self) -> Iterator[type]: + return iter(self._dumpers) + + def __len__(self) -> int: + return len(self._dumpers) + + def __delitem__(self, key: type) -> None: + # Internal check + if key in self._builtin: + raise_error(f"Cannot delete in-built key: {key}") + # Non-existing key + if key not in self._external: + raise_error(klass=KeyError, msg=str(key)) + # Update external + _ = self._external.pop(key) + # Update global + _ = self._dumpers.pop(key) + + def __setitem__(self, key: type, value: type[BaseDataDumpAsset]) -> None: + # Internal check + if key in self._builtin: + raise_error(f"Cannot set value for in-built key: {key}") + # Value type check + if not issubclass(value, BaseDataDumpAsset): + raise_error(f"Invalid value type: {type(value)}") + # Update external + self._external[key] = value + # Update global + self._dumpers[key] = value + + def popitem(): + """Not implemented.""" + pass + + def clear(self): + """Not implemented.""" + pass + + def setdefault(self, key: type, value=None): + """Not implemented.""" + pass + + +class AssetLoaderDispatcher(MutableMapping): + """Class for helping dynamic asset loader dispatch.""" + + _instance = None + + def __new__(cls): + # Make class singleton + if cls._instance is None: + cls._instance = super().__new__(cls) + # Set loaders + cls._loaders: dict[str, type[BaseDataDumpAsset]] = {} + cls._builtin: dict[str, type[BaseDataDumpAsset]] = {} + cls._external: dict[str, type[BaseDataDumpAsset]] = {} + cls._builtin.update( + { + ".nii.gz": Nifti1ImageAsset, + ".nii": Nifti1ImageAsset, + ".csv": PandasDataFrameAsset, + } + ) + cls._loaders.update(cls._builtin) + return cls._instance + + def __getitem__(self, key: str) -> type[BaseDataDumpAsset]: + return self._loaders[key] + + def __iter__(self) -> Iterator[str]: + return iter(self._loaders) + + def __len__(self) -> int: + return len(self._loaders) + + def __delitem__(self, key: str) -> None: + # Internal check + if key in self._builtin: + raise_error(f"Cannot delete in-built key: {key}") + # Non-existing key + if key not in self._external: + raise_error(klass=KeyError, msg=key) + # Update external + _ = self._external.pop(key) + # Update global + _ = self._loaders.pop(key) + + def __setitem__(self, key: str, value: type[BaseDataDumpAsset]) -> None: + # Internal check + if key in self._builtin: + raise_error(f"Cannot set value for in-built key: {key}") + # Value type check + if not issubclass(value, BaseDataDumpAsset): + raise_error(f"Invalid value type: {type(value)}") + # Update external + self._external[key] = value + # Update global + self._loaders[key] = value + + def popitem(): + """Not implemented.""" + pass + + def clear(self): + """Not implemented.""" + pass + + def setdefault(self, key: str, value=None): + """Not implemented.""" + pass + + +class DataObjectDumper: + """Class for pipeline data object dumping.""" + + _instance = None + + def __new__(cls): + """Overridden to make the class singleton.""" + # Make class singleton + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def dump(self, data: dict, path: Path, step: str) -> None: + """Dump data object at path. + + Parameters + ---------- + data : dict + The data object state to dump. + path : pathlib.Path + The path to dump the data object. + step : str + The step name. Also sets the dump directory. + + """ + # Make a deep copy of data + data_copy = deepcopy(data) + # Initialize list for storing assets to save + assets = [] + + dump_file_root = path / step + + for k, v in data_copy.items(): + # Conditional for Warp type; kept separate for low cognitive load + if isinstance(v, list): + for idx, _ in enumerate(v): + data_copy[k][idx]["path"] = str(data_copy[k][idx]["path"]) + continue + + # Transform Path to str + data_copy[k]["path"] = str(data_copy[k]["path"]) + # Pop out first level assets; some data types might not have + if "data" in v: + dumper = AssetDumperDispatcher()[type(v["data"])] + assets.append( + dumper( + data=v.pop("data"), + path_without_ext=dump_file_root / k, + ) + ) + for kk, vv in v.items(): + if isinstance(vv, dict) and kk != "meta": + # Transform Path to str + data_copy[k][kk]["path"] = str(data_copy[k][kk]["path"]) + # Pop out second level assets + if "data" in vv: + dumper = AssetDumperDispatcher()[type(vv["data"])] + assets.append( + dumper( + data=vv.pop("data"), + path_without_ext=dump_file_root / f"{k}_{kk}", + ) + ) + + # Save yaml + dump_file_path = dump_file_root / "data.yaml" + dump_file_path.parent.mkdir(parents=True, exist_ok=True) + yaml.dump(data_copy, stream=dump_file_path) + + # Save assets + for x in assets: + x.dump() + + def load(self, path: Path) -> dict: + """Load data object from path. + + Parameters + ---------- + path : pathlib.Path + The path to the dumped data object. + + Returns + ------- + dict + The restored data object dump. + + """ + data = yaml.load(path) + # Load assets; stem => path mapping + assets = { + child.stem.split(".")[0]: child + for child in path.parent.iterdir() + if "".join(child.suffixes) in AssetLoaderDispatcher() + } + + for k, v in data.items(): + # Conditional for Warp type; kept separate for low cognitive load + if isinstance(v, list): + for idx, _ in enumerate(v): + data[k][idx]["path"] = Path(data[k][idx]["path"]) + continue + + # Transform str to Path + data[k]["path"] = Path(data[k]["path"]) + # Insert first level assets if matching asset is found + if k in assets: + # Get path + p = assets[k] + data[k]["path"] = p + # Get correct loader using extension + loader = AssetLoaderDispatcher()["".join(p.suffixes)] + data[k]["data"] = loader.load(p) + for kk, vv in v.items(): + if isinstance(vv, dict) and kk != "meta": + # Transform str to Path + data[k][kk]["path"] = Path(data[k][kk]["path"]) + # Insert second level assets + key = f"{k}_{kk}" + if key in assets: + # Get path + pp = assets[key] + data[k][kk]["path"] = pp + # Get correct loader using extension + loader = AssetLoaderDispatcher()["".join(pp.suffixes)] + data[k][kk]["data"] = loader.load(pp) + + return data diff --git a/junifer/pipeline/marker_collection.py b/junifer/pipeline/marker_collection.py index f8ae07bba..9e25b40e8 100644 --- a/junifer/pipeline/marker_collection.py +++ b/junifer/pipeline/marker_collection.py @@ -5,12 +5,13 @@ # License: AGPL from collections import Counter +from pathlib import Path from typing import Optional from ..datareader import DefaultDataReader -from ..pipeline import PipelineStepMixin, WorkDirManager +from ..pipeline import DataObjectDumper, PipelineStepMixin, WorkDirManager from ..typing import DataGrabberLike, MarkerLike, PreprocessorLike, StorageLike -from ..utils import logger, raise_error +from ..utils import config, logger, raise_error __all__ = ["MarkerCollection"] @@ -80,16 +81,53 @@ class MarkerCollection: # Fetch actual data using datareader data = self._datareader.fit_transform(input) + # Conditional data dump + if ( + config.get("preprocessing.dump.location") is not None + and config.get("preprocessing.dump.granularity") == "full" + ): + DataObjectDumper().dump( + data=data, + path=Path(config.get("preprocessing.dump.location")), + step=f"0_datareader_{self._datareader.__class__.__name__}", + ) # Apply preprocessing steps if self._preprocessors is not None: - for preprocessor in self._preprocessors: + for idx, preprocessor in enumerate(self._preprocessors): logger.info( "Preprocessing data with " f"{preprocessor.__class__.__name__}" ) # Mutate data after every iteration data = preprocessor.fit_transform(data) + # Conditional data dump + if ( + config.get("preprocessing.dump.location") is not None + and config.get("preprocessing.dump.granularity") == "full" + ): + DataObjectDumper().dump( + data=data, + path=Path(config.get("preprocessing.dump.location")), + step=( + f"{idx + 1}_preprocessor_" + f"{preprocessor.__class__.__name__}" + ), + ) + + # Conditional data dump + if ( + config.get("preprocessing.dump.location") is not None + and config.get("preprocessing.dump.granularity") == "final" + ): + DataObjectDumper().dump( + data=data, + path=Path(config.get("preprocessing.dump.location")), + step=( + f"final_preprocessor_" + f"{self._preprocessors[-1].__class__.__name__}" + ), + ) # Compute markers out = {} diff --git a/junifer/pipeline/tests/test_data_object_dumper.py b/junifer/pipeline/tests/test_data_object_dumper.py new file mode 100644 index 000000000..7cfefa659 --- /dev/null +++ b/junifer/pipeline/tests/test_data_object_dumper.py @@ -0,0 +1,225 @@ +"""Provide tests for data object dumping.""" + +# Authors: Synchon Mandal +# License: AGPL + +import pickle +from pathlib import Path +from typing import Union + +import nibabel +import pytest + +from junifer.markers import FunctionalConnectivitySpheres +from junifer.pipeline import ( + AssetDumperDispatcher, + AssetLoaderDispatcher, + BaseDataDumpAsset, + DataObjectDumper, + MarkerCollection, +) +from junifer.preprocess import fMRIPrepConfoundRemover +from junifer.testing.datagrabbers import PartlyCloudyTestingDataGrabber +from junifer.utils import config + + +@pytest.mark.parametrize( + "dispatcher, inbuilt_key, ext_key, val", + [ + ( + AssetDumperDispatcher, + nibabel.Nifti1Image, + nibabel.Nifti2Image, + dict, + ), + (AssetLoaderDispatcher, ".nii", ".tsv", dict), + ], +) +def test_dispatcher_addition_errors( + dispatcher: Union[AssetDumperDispatcher, AssetLoaderDispatcher], + inbuilt_key: Union[str, type], + ext_key: Union[str, type], + val: type, +) -> None: + """Test asset dumper / loader addition errors. + + Parameters + ---------- + dispatcher : AssetDumperDispatcher or AssetLoaderDispatcher, + The parametrized dispatcher. + inbuilt_key : str or type + The parametrized in-built key. + ext_key : str or type + The parametrized external key. + val : type + The parametrized value. + + """ + with pytest.raises(ValueError, match="Cannot set"): + dispatcher()[inbuilt_key] = val + + with pytest.raises(ValueError, match="Invalid"): + dispatcher()[ext_key] = val + + +@pytest.mark.parametrize( + "dispatcher, inbuilt_key, ext_key", + [ + (AssetDumperDispatcher, nibabel.Nifti1Image, nibabel.Nifti2Image), + (AssetLoaderDispatcher, ".nii", ".tsv"), + ], +) +def test_dispatcher_removal_errors( + dispatcher: Union[AssetDumperDispatcher, AssetLoaderDispatcher], + inbuilt_key: Union[str, type], + ext_key: Union[str, type], +) -> None: + """Test asset dumper / loader removal errors. + + Parameters + ---------- + dispatcher : AssetDumperDispatcher or AssetLoaderDispatcher, + The parametrized dispatcher. + inbuilt_key : str or type + The parametrized in-built key. + ext_key : str or type + The parametrized external key. + + """ + with pytest.raises(ValueError, match="Cannot delete"): + _ = dispatcher().pop(inbuilt_key) + + with pytest.raises(KeyError, match=f"{ext_key}"): + del dispatcher()[ext_key] + + +def test_dispatcher() -> None: + """Test asset dumper / loader addition and removal.""" + + class Int(int): ... + + class Float(float): ... + + class DumAsset(BaseDataDumpAsset): + def dump(self): + suffix = "" + if isinstance(self.data, Int): + suffix = ".int" + else: + suffix = ".float" + pickle.dump(self.data, self.path_without_ext.with_suffix(suffix)) + + @classmethod + def load(cls, path): + return pickle.load(path) + + AssetDumperDispatcher().update({nibabel.Nifti2Image: DumAsset}) + assert nibabel.Nifti2Image in AssetDumperDispatcher() + _ = AssetDumperDispatcher().pop(nibabel.Nifti2Image) + assert nibabel.Nifti2Image not in AssetDumperDispatcher() + + AssetLoaderDispatcher().update({".n+2": DumAsset}) + assert ".n+2" in AssetLoaderDispatcher() + _ = AssetLoaderDispatcher().pop(".n+2") + assert ".n+2" not in AssetLoaderDispatcher() + + +@pytest.mark.parametrize( + "granularity, expected_dir_count", + [ + ("full", 2), + ("final", 1), + ], +) +def test_data_object_dumper( + tmp_path: Path, granularity: str, expected_dir_count: int +) -> None: + """Test data object dumper. + + Parameters + ---------- + tmp_path : pathlib.Path + The path to the test directory. + granularity : str + The parametrized granularity. + expected_dir_count : int + The parametrized expected directory count. + + """ + config.set(key="preprocessing.dump.location", val=tmp_path) + config.set(key="preprocessing.dump.granularity", val=granularity) + + mc = MarkerCollection( + preprocessors=[ + fMRIPrepConfoundRemover( + strategy={ + "motion": "full", + "wm_csf": "full", + }, + detrend=True, + standardize=True, + low_pass=0.08, + high_pass=0.01, + ), + ], + markers=[ + FunctionalConnectivitySpheres( + name="dmnbuckner_5mm_fc_spheres", + coords="DMNBuckner", + radius=5.0, + conn_method="correlation", + ), + ], + ) + dg = PartlyCloudyTestingDataGrabber() + + with dg: + mc.fit(dg["sub-01"]) + + dirs = list(tmp_path.iterdir()) + assert len(dirs) == expected_dir_count + + dump_load = DataObjectDumper().load(dirs[-1] / "data.yaml") + assert "BOLD" in dump_load + + config.delete("preprocessing.dump.location") + config.delete("preprocessing.dump.granularity") + + +def test_data_object_dumper_with_warp(tmp_path: Path) -> None: + """Test data object dumper with Warp data type. + + Parameters + ---------- + tmp_path : pathlib.Path + The path to the test directory. + + """ + DataObjectDumper().dump( + data={ + "Warp": [ + { + "path": ( + tmp_path / "from-MNI152NLin2009cAsym_to-T1w_" + "mode-image_xfm.h5" + ), + "src": "MNI152NLin2009cAsym", + "dst": "native", + "warper": "ants", + }, + { + "path": ( + tmp_path / "from-T1w_to-MNI152NLin2009cAsym_" + "mode-image_xfm.h5" + ), + "src": "native", + "dst": "MNI152NLin2009cAsym", + "warper": "ants", + }, + ], + }, + path=tmp_path, + step="warp_test", + ) + dump_load = DataObjectDumper().load(tmp_path / "warp_test" / "data.yaml") + assert "Warp" in dump_load diff --git a/junifer/typing/__init__.pyi b/junifer/typing/__init__.pyi index 7db1445e2..6cae82bfe 100644 --- a/junifer/typing/__init__.pyi +++ b/junifer/typing/__init__.pyi @@ -1,4 +1,5 @@ __all__ = [ + "DataDumpAssetLike", "DataGrabberLike", "DataRegistryLike", "PreprocessorLike", @@ -16,6 +17,7 @@ __all__ = [ ] from ._typing import ( + DataDumpAssetLike, DataGrabberLike, DataRegistryLike, PreprocessorLike, diff --git a/junifer/typing/_typing.py b/junifer/typing/_typing.py index 42d4e01bd..dbdc40226 100644 --- a/junifer/typing/_typing.py +++ b/junifer/typing/_typing.py @@ -22,6 +22,7 @@ if TYPE_CHECKING: __all__ = [ "ConditionalDependencies", "ConfigVal", + "DataDumpAssetLike", "DataGrabberLike", "DataGrabberPatterns", "DataRegistryLike", @@ -37,6 +38,7 @@ __all__ = [ ] +DataDumpAssetLike = type["DataDumpAssetLike"] DataRegistryLike = type["BasePipelineDataRegistry"] DataGrabberLike = type["BaseDataGrabber"] PreprocessorLike = type["BasePreprocessor"]