[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) 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,
) )

View file

@ -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:
preprocessor_object = _get_preprocessor(preprocessor) _preprocessors = [x.copy() for x in preprocessors]
built_preprocessors = []
for preprocessor in _preprocessors:
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

View file

@ -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

View file

@ -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,10 +122,15 @@ 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(
logger.info(f"Preprocess output type: {t_data}") "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: for marker in self._markers:
logger.info(f"Validating Marker: {marker.name}") logger.info(f"Validating Marker: {marker.name}")

View file

@ -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)