[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)
|
||||
# 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,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue