[ENH]: Allow for dumping data object to disk for debugging purposes #452
15 changed files with 861 additions and 5 deletions
1
docs/changes/newsfragments/452.doc
Normal file
1
docs/changes/newsfragments/452.doc
Normal file
|
|
@ -0,0 +1 @@
|
|||
Add documentation on :class:`.ConfigManager`, dumping data object and extending data dump assets by `Synchon Mandal`_
|
||||
1
docs/changes/newsfragments/452.feature
Normal file
1
docs/changes/newsfragments/452.feature
Normal file
|
|
@ -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`_
|
||||
46
docs/extending/data_dump_asset.rst
Normal file
46
docs/extending/data_dump_asset.rst
Normal file
|
|
@ -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.
|
||||
|
|
@ -33,3 +33,4 @@ DataGrabbers, Preprocessors, Markers, etc., following the *junifer* way.
|
|||
plugins
|
||||
data_registries
|
||||
data_types
|
||||
data_dump_asset
|
||||
|
|
|
|||
65
docs/using/configuring.rst
Normal file
65
docs/using/configuring.rst
Normal file
|
|
@ -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
|
||||
18
docs/using/dumping.rst
Normal file
18
docs/using/dumping.rst
Normal file
|
|
@ -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>`.
|
||||
|
|
@ -18,6 +18,8 @@ to interact with HPC and HTC systems.
|
|||
codeless
|
||||
running
|
||||
queueing
|
||||
configuring
|
||||
dumping
|
||||
|
||||
|
||||
.. _using_components:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
347
junifer/pipeline/_data_object_dumper.py
Normal file
347
junifer/pipeline/_data_object_dumper.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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 = {}
|
||||
|
|
|
|||
225
junifer/pipeline/tests/test_data_object_dumper.py
Normal file
225
junifer/pipeline/tests/test_data_object_dumper.py
Normal file
|
|
@ -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
|
||||
|
|
@ -1,4 +1,5 @@
|
|||
__all__ = [
|
||||
"DataDumpAssetLike",
|
||||
"DataGrabberLike",
|
||||
"DataRegistryLike",
|
||||
"PreprocessorLike",
|
||||
|
|
@ -16,6 +17,7 @@ __all__ = [
|
|||
]
|
||||
|
||||
from ._typing import (
|
||||
DataDumpAssetLike,
|
||||
DataGrabberLike,
|
||||
DataRegistryLike,
|
||||
PreprocessorLike,
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
|
|||
Loading…
Reference in a new issue