docs/changes/newsfragments/452.doc

@@ -0,0 +1 @@
+Add documentation on :class:`.ConfigManager`, dumping data object and extending data dump assets by `Synchon Mandal`_

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`_

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 <extending_extension>` 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.

docs/extending/index.rst

@@ -33,3 +33,4 @@ DataGrabbers, Preprocessors, Markers, etc.,  following the *junifer* way.
    plugins
    data_registries
    data_types
+   data_dump_asset

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 <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
+
+   <CONFIG-KEY>=<CONFIG-VAL> junifer run ...
+
+and via the API like so:
+
+.. code-block:: python
+
+    from junifer.utils import config
+
+    # Add config
+    config.set(key="<config-key>", val=<config-val>)
+
+    # do feature extraction
+    ...
+
+    # Remove config
+    config.delete("<config-key>")
+
+.. _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

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 <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 <available_configurations>`.
+
+``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 <adding_data_types>` which support different file formats, you can create and
+register a :ref:`custom dumper <adding_data_dump_assets>`.

docs/using/index.rst

@@ -18,6 +18,8 @@ to interact with HPC and HTC systems.
    codeless
    running
    queueing
+   configuring
+   dumping
 
 
 .. _using_components:

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

junifer/api/tests/test_decorators.py

@@ -3,8 +3,18 @@
 # Authors: Synchon Mandal <s.mandal@fz-juelich.de>
 # 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()

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

junifer/pipeline/_data_object_dumper.py

@@ -0,0 +1,347 @@
+"""Provide pipeline data object dumper and data dump asset classes."""
+
+# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
+# 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

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 = {}

junifer/pipeline/tests/test_data_object_dumper.py

@@ -0,0 +1,225 @@
+"""Provide tests for data object dumping."""
+
+# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
+# 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

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,

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"]