[ENH]: Add support for multiple preprocessors in the pipeline #263

Merged
synchon merged 6 commits from update/multi-preprocessor into main 2023-10-18 13:13:39 +00:00
6 changed files with 58 additions and 30 deletions

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

View file

@ -200,13 +200,23 @@ def run(
"""
configure_logging(level=verbose)
# TODO: add validation
# TODO(synchon): add validation
# Parse YAML
config = parse_yaml(filepath) # type: ignore
# Retrieve working directory
workdir = config["workdir"]
# Fetch datagrabber
datagrabber = config["datagrabber"]
# Fetch markers
markers = config["markers"]
# Fetch 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)
# Perform operation
api_run(
@ -214,7 +224,7 @@ def run(
datagrabber=datagrabber,
markers=markers,
storage=storage,
preprocessor=preprocessor,
preprocessors=preprocessors,
elements=elements,
)

View file

@ -81,7 +81,7 @@ def run(
datagrabber: Dict,
markers: List[Dict],
storage: Dict,
preprocessor: Optional[Dict] = None,
preprocessors: Optional[List[Dict]] = None,
elements: Union[str, List[Union[str, Tuple]], Tuple, None] = None,
) -> None:
"""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. All other keys are passed to the storage
init function.
preprocessor : dict, optional
Preprocessor to use. Must have a key ``kind`` with the kind of
preprocessor to use. All other keys are passed to the preprocessor
init function (default None).
preprocessors : list of dict, optional
List of preprocessors to use. Each preprocessor is a dict with at
least a key ``kind`` specifying the preprocessor to use. All other keys
are passed to the preprocessor init function (default None).
elements : str or tuple or list of str or tuple, optional
Element(s) to process. Will be used to index the DataGrabber
(default None).
@ -152,15 +152,19 @@ def run(
storage_object = typing.cast(BaseFeatureStorage, storage_object)
# 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)
built_preprocessors.append(preprocessor_object)
else:
preprocessor_object = None
built_preprocessors = None
# Create new marker collection
mc = MarkerCollection(
markers=built_markers,
preprocessing=preprocessor_object,
preprocessors=built_preprocessors,
storage=storage_object,
)
# Fit elements

View file

@ -117,9 +117,11 @@ def test_run_single_element_with_preprocessing(tmp_path: Path) -> None:
}
],
storage=storage,
preprocessor={
preprocessors=[
{
"kind": "fMRIPrepConfoundRemover",
},
}
],
elements=["sub-01"],
)
# Check files

View file

@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Dict, List, Optional
from ..datareader.default import DefaultDataReader
from ..markers.base import BaseMarker
from ..pipeline import PipelineStepMixin
from ..preprocess.base import BasePreprocessor
from ..storage.base import BaseFeatureStorage
from ..utils import logger
@ -27,8 +28,8 @@ class MarkerCollection:
The markers to compute.
datareader : DataReader-like object, optional
The DataReader to use (default None).
preprocessing : preprocessing-like, optional
The preprocessing steps to apply.
preprocessors : list of preprocessing-like, optional
The preprocessors to apply (default None).
storage : storage-like, optional
The storage to use (default None).
@ -38,7 +39,7 @@ class MarkerCollection:
self,
markers: List[BaseMarker],
datareader: Optional[PipelineStepMixin] = None,
preprocessing: Optional[PipelineStepMixin] = None,
preprocessors: Optional[List[BasePreprocessor]] = None,
storage: Optional[BaseFeatureStorage] = None,
):
# Check that the markers have different names
@ -53,7 +54,7 @@ class MarkerCollection:
if datareader is None:
datareader = DefaultDataReader()
self._datareader = datareader
self._preprocessing = preprocessing
self._preprocessors = preprocessors
self._storage = storage
def fit(self, input: Dict[str, Dict]) -> Optional[Dict]:
@ -79,9 +80,14 @@ class MarkerCollection:
data = self._datareader.fit_transform(input)
# Apply preprocessing steps
if self._preprocessing is not None:
logger.info("Preprocessing data")
data = self._preprocessing.fit_transform(data)
if self._preprocessors is not None:
for preprocessor in self._preprocessors:
logger.info(
"Preprocessing data with "
f"{preprocessor.__class__.__name__}"
)
# Mutate data after every iteration
data = preprocessor.fit_transform(data)
# Compute markers
out = {}
@ -116,9 +122,14 @@ class MarkerCollection:
t_data = self._datareader.validate(t_data)
logger.info(f"Data Reader output type: {t_data}")
if self._preprocessing is not None:
logger.info("Validating Preprocessor:")
t_data = self._preprocessing.validate(t_data)
if self._preprocessors is not None:
for preprocessor in self._preprocessors:
logger.info(
"Validating Preprocessor: "
f"{preprocessor.__class__.__name__}"
)
# Validate preprocessor
t_data = preprocessor.validate(t_data)
logger.info(f"Preprocess output type: {t_data}")
for marker in self._markers:

View file

@ -64,7 +64,7 @@ def test_marker_collection() -> None:
]
mc = MarkerCollection(markers=markers) # type: ignore
assert mc._markers == markers
assert mc._preprocessing is None
assert mc._preprocessors is None
assert mc._storage is None
assert isinstance(mc._datareader, DefaultDataReader)
@ -97,7 +97,7 @@ def test_marker_collection() -> None:
mc2 = MarkerCollection(
markers=markers, # type: ignore
preprocessing=BypassPreprocessing(),
preprocessors=[BypassPreprocessing()], # type: ignore
datareader=DefaultDataReader(),
)
assert isinstance(mc2._datareader, DefaultDataReader)
@ -128,10 +128,10 @@ def test_marker_collection_with_preprocessing() -> None:
]
mc = MarkerCollection(
markers=markers, # type: ignore
preprocessing=fMRIPrepConfoundRemover(),
preprocessors=[fMRIPrepConfoundRemover()],
)
assert mc._markers == markers
assert mc._preprocessing is not None
assert mc._preprocessors is not None
assert mc._storage is None
assert isinstance(mc._datareader, DefaultDataReader)