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

docs/changes/newsfragments/263.change

@@ -0,0 +1 @@
+Modify ``preprocessor`` to ``preprocessors`` in :func:`.run` and ``preprocessing`` to ``preprocessors`` in :class:`.MarkerCollection` to accept multiple preprocessors by `Synchon Mandal`_

junifer/api/cli.py

@@ -200,21 +200,31 @@ 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(
         workdir=workdir,
         datagrabber=datagrabber,
         markers=markers,
         storage=storage,
-        preprocessor=preprocessor,
+        preprocessors=preprocessors,
         elements=elements,
     )
 

junifer/api/functions.py

@@ -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:
-        preprocessor_object = _get_preprocessor(preprocessor)
+    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

junifer/api/tests/test_functions.py

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

junifer/markers/collection.py

@@ -10,6 +10,7 @@
 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 @@ def __init__(
         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 @@ def __init__(
         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 @@ def fit(self, input: Dict[str, Dict]) -> Optional[Dict]:
         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,10 +122,15 @@ def validate(self, datagrabber: "BaseDataGrabber") -> None:
         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)
-            logger.info(f"Preprocess output type: {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:
             logger.info(f"Validating Marker: {marker.name}")

junifer/markers/tests/test_collection.py

@@ -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 fit_transform(self, input):
 
     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)