[ENH]: Add support for multiple preprocessors in the pipeline #263
6 changed files with 58 additions and 30 deletions
1
docs/changes/newsfragments/263.change
Normal file
1
docs/changes/newsfragments/263.change
Normal file
|
|
@ -0,0 +1 @@
|
||||||
|
Modify ``preprocessor`` to ``preprocessors`` in :func:`.run` and ``preprocessing`` to ``preprocessors`` in :class:`.MarkerCollection` to accept multiple preprocessors by `Synchon Mandal`_
|
||||||
|
|
@ -200,13 +200,23 @@ def run(
|
||||||
|
|
||||||
"""
|
"""
|
||||||
configure_logging(level=verbose)
|
configure_logging(level=verbose)
|
||||||
# TODO: add validation
|
# TODO(synchon): add validation
|
||||||
|
# Parse YAML
|
||||||
config = parse_yaml(filepath) # type: ignore
|
config = parse_yaml(filepath) # type: ignore
|
||||||
|
# Retrieve working directory
|
||||||
workdir = config["workdir"]
|
workdir = config["workdir"]
|
||||||
|
# Fetch datagrabber
|
||||||
datagrabber = config["datagrabber"]
|
datagrabber = config["datagrabber"]
|
||||||
|
# Fetch markers
|
||||||
markers = config["markers"]
|
markers = config["markers"]
|
||||||
|
# Fetch storage
|
||||||
storage = config["storage"]
|
storage = config["storage"]
|
||||||
preprocessor = config.get("preprocess")
|
# Fetch preprocessors
|
||||||
|
preprocessors = config.get("preprocess")
|
||||||
|
# Convert to list if single preprocessor
|
||||||
|
if preprocessors is not None and not isinstance(preprocessors, list):
|
||||||
|
preprocessors = [preprocessors]
|
||||||
|
# Parse elements
|
||||||
elements = _parse_elements(element, config)
|
elements = _parse_elements(element, config)
|
||||||
# Perform operation
|
# Perform operation
|
||||||
api_run(
|
api_run(
|
||||||
|
|
@ -214,7 +224,7 @@ def run(
|
||||||
datagrabber=datagrabber,
|
datagrabber=datagrabber,
|
||||||
markers=markers,
|
markers=markers,
|
||||||
storage=storage,
|
storage=storage,
|
||||||
preprocessor=preprocessor,
|
preprocessors=preprocessors,
|
||||||
elements=elements,
|
elements=elements,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -81,7 +81,7 @@ def run(
|
||||||
datagrabber: Dict,
|
datagrabber: Dict,
|
||||||
markers: List[Dict],
|
markers: List[Dict],
|
||||||
storage: Dict,
|
storage: Dict,
|
||||||
preprocessor: Optional[Dict] = None,
|
preprocessors: Optional[List[Dict]] = None,
|
||||||
elements: Union[str, List[Union[str, Tuple]], Tuple, None] = None,
|
elements: Union[str, List[Union[str, Tuple]], Tuple, None] = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Run the pipeline on the selected element.
|
"""Run the pipeline on the selected element.
|
||||||
|
|
@ -104,10 +104,10 @@ def run(
|
||||||
Storage to use. Must have a key ``kind`` with the kind of
|
Storage to use. Must have a key ``kind`` with the kind of
|
||||||
storage to use. All other keys are passed to the storage
|
storage to use. All other keys are passed to the storage
|
||||||
init function.
|
init function.
|
||||||
preprocessor : dict, optional
|
preprocessors : list of dict, optional
|
||||||
Preprocessor to use. Must have a key ``kind`` with the kind of
|
List of preprocessors to use. Each preprocessor is a dict with at
|
||||||
preprocessor to use. All other keys are passed to the preprocessor
|
least a key ``kind`` specifying the preprocessor to use. All other keys
|
||||||
init function (default None).
|
are passed to the preprocessor init function (default None).
|
||||||
elements : str or tuple or list of str or tuple, optional
|
elements : str or tuple or list of str or tuple, optional
|
||||||
Element(s) to process. Will be used to index the DataGrabber
|
Element(s) to process. Will be used to index the DataGrabber
|
||||||
(default None).
|
(default None).
|
||||||
|
|
@ -152,15 +152,19 @@ def run(
|
||||||
storage_object = typing.cast(BaseFeatureStorage, storage_object)
|
storage_object = typing.cast(BaseFeatureStorage, storage_object)
|
||||||
|
|
||||||
# Get preprocessor to use (if provided)
|
# Get preprocessor to use (if provided)
|
||||||
if preprocessor is not None:
|
if preprocessors is not None:
|
||||||
|
_preprocessors = [x.copy() for x in preprocessors]
|
||||||
|
built_preprocessors = []
|
||||||
|
for preprocessor in _preprocessors:
|
||||||
preprocessor_object = _get_preprocessor(preprocessor)
|
preprocessor_object = _get_preprocessor(preprocessor)
|
||||||
|
built_preprocessors.append(preprocessor_object)
|
||||||
else:
|
else:
|
||||||
preprocessor_object = None
|
built_preprocessors = None
|
||||||
|
|
||||||
# Create new marker collection
|
# Create new marker collection
|
||||||
mc = MarkerCollection(
|
mc = MarkerCollection(
|
||||||
markers=built_markers,
|
markers=built_markers,
|
||||||
preprocessing=preprocessor_object,
|
preprocessors=built_preprocessors,
|
||||||
storage=storage_object,
|
storage=storage_object,
|
||||||
)
|
)
|
||||||
# Fit elements
|
# Fit elements
|
||||||
|
|
|
||||||
|
|
@ -117,9 +117,11 @@ def test_run_single_element_with_preprocessing(tmp_path: Path) -> None:
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
storage=storage,
|
storage=storage,
|
||||||
preprocessor={
|
preprocessors=[
|
||||||
|
{
|
||||||
"kind": "fMRIPrepConfoundRemover",
|
"kind": "fMRIPrepConfoundRemover",
|
||||||
},
|
}
|
||||||
|
],
|
||||||
elements=["sub-01"],
|
elements=["sub-01"],
|
||||||
)
|
)
|
||||||
# Check files
|
# Check files
|
||||||
|
|
|
||||||
|
|
@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Dict, List, Optional
|
||||||
from ..datareader.default import DefaultDataReader
|
from ..datareader.default import DefaultDataReader
|
||||||
from ..markers.base import BaseMarker
|
from ..markers.base import BaseMarker
|
||||||
from ..pipeline import PipelineStepMixin
|
from ..pipeline import PipelineStepMixin
|
||||||
|
from ..preprocess.base import BasePreprocessor
|
||||||
from ..storage.base import BaseFeatureStorage
|
from ..storage.base import BaseFeatureStorage
|
||||||
from ..utils import logger
|
from ..utils import logger
|
||||||
|
|
||||||
|
|
@ -27,8 +28,8 @@ class MarkerCollection:
|
||||||
The markers to compute.
|
The markers to compute.
|
||||||
datareader : DataReader-like object, optional
|
datareader : DataReader-like object, optional
|
||||||
The DataReader to use (default None).
|
The DataReader to use (default None).
|
||||||
preprocessing : preprocessing-like, optional
|
preprocessors : list of preprocessing-like, optional
|
||||||
The preprocessing steps to apply.
|
The preprocessors to apply (default None).
|
||||||
storage : storage-like, optional
|
storage : storage-like, optional
|
||||||
The storage to use (default None).
|
The storage to use (default None).
|
||||||
|
|
||||||
|
|
@ -38,7 +39,7 @@ class MarkerCollection:
|
||||||
self,
|
self,
|
||||||
markers: List[BaseMarker],
|
markers: List[BaseMarker],
|
||||||
datareader: Optional[PipelineStepMixin] = None,
|
datareader: Optional[PipelineStepMixin] = None,
|
||||||
preprocessing: Optional[PipelineStepMixin] = None,
|
preprocessors: Optional[List[BasePreprocessor]] = None,
|
||||||
storage: Optional[BaseFeatureStorage] = None,
|
storage: Optional[BaseFeatureStorage] = None,
|
||||||
):
|
):
|
||||||
# Check that the markers have different names
|
# Check that the markers have different names
|
||||||
|
|
@ -53,7 +54,7 @@ class MarkerCollection:
|
||||||
if datareader is None:
|
if datareader is None:
|
||||||
datareader = DefaultDataReader()
|
datareader = DefaultDataReader()
|
||||||
self._datareader = datareader
|
self._datareader = datareader
|
||||||
self._preprocessing = preprocessing
|
self._preprocessors = preprocessors
|
||||||
self._storage = storage
|
self._storage = storage
|
||||||
|
|
||||||
def fit(self, input: Dict[str, Dict]) -> Optional[Dict]:
|
def fit(self, input: Dict[str, Dict]) -> Optional[Dict]:
|
||||||
|
|
@ -79,9 +80,14 @@ class MarkerCollection:
|
||||||
data = self._datareader.fit_transform(input)
|
data = self._datareader.fit_transform(input)
|
||||||
|
|
||||||
# Apply preprocessing steps
|
# Apply preprocessing steps
|
||||||
if self._preprocessing is not None:
|
if self._preprocessors is not None:
|
||||||
logger.info("Preprocessing data")
|
for preprocessor in self._preprocessors:
|
||||||
data = self._preprocessing.fit_transform(data)
|
logger.info(
|
||||||
|
"Preprocessing data with "
|
||||||
|
f"{preprocessor.__class__.__name__}"
|
||||||
|
)
|
||||||
|
# Mutate data after every iteration
|
||||||
|
data = preprocessor.fit_transform(data)
|
||||||
|
|
||||||
# Compute markers
|
# Compute markers
|
||||||
out = {}
|
out = {}
|
||||||
|
|
@ -116,9 +122,14 @@ class MarkerCollection:
|
||||||
t_data = self._datareader.validate(t_data)
|
t_data = self._datareader.validate(t_data)
|
||||||
logger.info(f"Data Reader output type: {t_data}")
|
logger.info(f"Data Reader output type: {t_data}")
|
||||||
|
|
||||||
if self._preprocessing is not None:
|
if self._preprocessors is not None:
|
||||||
logger.info("Validating Preprocessor:")
|
for preprocessor in self._preprocessors:
|
||||||
t_data = self._preprocessing.validate(t_data)
|
logger.info(
|
||||||
|
"Validating Preprocessor: "
|
||||||
|
f"{preprocessor.__class__.__name__}"
|
||||||
|
)
|
||||||
|
# Validate preprocessor
|
||||||
|
t_data = preprocessor.validate(t_data)
|
||||||
logger.info(f"Preprocess output type: {t_data}")
|
logger.info(f"Preprocess output type: {t_data}")
|
||||||
|
|
||||||
for marker in self._markers:
|
for marker in self._markers:
|
||||||
|
|
|
||||||
|
|
@ -64,7 +64,7 @@ def test_marker_collection() -> None:
|
||||||
]
|
]
|
||||||
mc = MarkerCollection(markers=markers) # type: ignore
|
mc = MarkerCollection(markers=markers) # type: ignore
|
||||||
assert mc._markers == markers
|
assert mc._markers == markers
|
||||||
assert mc._preprocessing is None
|
assert mc._preprocessors is None
|
||||||
assert mc._storage is None
|
assert mc._storage is None
|
||||||
assert isinstance(mc._datareader, DefaultDataReader)
|
assert isinstance(mc._datareader, DefaultDataReader)
|
||||||
|
|
||||||
|
|
@ -97,7 +97,7 @@ def test_marker_collection() -> None:
|
||||||
|
|
||||||
mc2 = MarkerCollection(
|
mc2 = MarkerCollection(
|
||||||
markers=markers, # type: ignore
|
markers=markers, # type: ignore
|
||||||
preprocessing=BypassPreprocessing(),
|
preprocessors=[BypassPreprocessing()], # type: ignore
|
||||||
datareader=DefaultDataReader(),
|
datareader=DefaultDataReader(),
|
||||||
)
|
)
|
||||||
assert isinstance(mc2._datareader, DefaultDataReader)
|
assert isinstance(mc2._datareader, DefaultDataReader)
|
||||||
|
|
@ -128,10 +128,10 @@ def test_marker_collection_with_preprocessing() -> None:
|
||||||
]
|
]
|
||||||
mc = MarkerCollection(
|
mc = MarkerCollection(
|
||||||
markers=markers, # type: ignore
|
markers=markers, # type: ignore
|
||||||
preprocessing=fMRIPrepConfoundRemover(),
|
preprocessors=[fMRIPrepConfoundRemover()],
|
||||||
)
|
)
|
||||||
assert mc._markers == markers
|
assert mc._markers == markers
|
||||||
assert mc._preprocessing is not None
|
assert mc._preprocessors is not None
|
||||||
assert mc._storage is None
|
assert mc._storage is None
|
||||||
assert isinstance(mc._datareader, DefaultDataReader)
|
assert isinstance(mc._datareader, DefaultDataReader)
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue