Pipeline

The core class that drives the preprocessing pipeline.

MEEGFlowPipeline

Source code in src/meegflow/pipeline.py

class MEEGFlowPipeline:
    def __init__(
        self, 
        reader: DatasetReader,
        output_root: Union[str, Path] = None, 
        config: Dict[str, Any] = None
    ):
        """Initialize MEEGFlow preprocessing pipeline.

        Parameters
        ----------
        reader : DatasetReader
            Reader instance for discovering data files. Use BIDSReader for BIDS datasets
            or GlobReader for custom directory structures.
        output_root : str or Path, optional
            Path to output derivatives root. If not provided, defaults to
            {dataset_root}/derivatives/meegflow
        config : dict, optional
            Configuration dictionary containing pipeline steps and parameters
        """
        self.config = config or {}
        self.output_root = Path(output_root) if output_root else None
        self.reader = reader

        # Validate the top-level datatype. If the configuration names one, a
        # BIDS reader that was not given its own datatype searches that one.
        self.datatype = resolve_datatype(self.config)
        if self.config.get('datatype') is not None and getattr(reader, 'datatype', False) is None:
            reader.datatype = self.datatype

        # Built-in steps come from the registry (populated by importing the
        # steps package); custom steps may add to or override them by name.
        self.step_functions = build_step_functions(self.config)

        # Validate pipeline steps if provided in config
        pipeline_cfg = self.config.get('pipeline', [])
        unknown = [s.get('name') for s in pipeline_cfg if s.get('name') not in self.step_functions]
        if unknown:
            raise ValueError(f"Unknown pipeline steps in config: {unknown}")

    @property
    def dataset_root(self) -> Path:
        """Get the dataset root path from the reader."""
        return self.reader.root

    def _load_custom_steps(self, custom_steps_folder: Union[str, Path]) -> Dict[str, Callable]:
        """
        Load custom preprocessing steps from Python files in the specified folder.

        This method discovers .py files in the custom_steps_folder and imports functions
        that follow the step function signature: func(data: Dict, step_config: Dict) -> Dict

        The function name will be used as the step name in the pipeline configuration.
        Custom steps can override built-in steps by using the same name.

        Parameters
        ----------
        custom_steps_folder : str or Path
            Path to folder containing Python files with custom step functions.

        Returns
        -------
        custom_steps : dict
            Dictionary mapping step names to their functions.

        Notes
        -----
        Custom step functions must:
        - Accept two parameters: data (Dict) and step_config (Dict)
        - Return a Dict (the updated data dictionary)
        - Be defined at module level (not inside classes)

        Example custom step file (my_steps.py):
        ```python
        def my_custom_filter(data, step_config):
            '''Apply custom filtering to raw data.'''
            if 'raw' not in data:
                raise ValueError("my_custom_filter requires 'raw' in data")

            # Get parameters from step_config
            cutoff_freq = step_config.get('cutoff_freq', 30.0)

            # Apply custom processing
            data['raw'].filter(h_freq=cutoff_freq, l_freq=None)

            # Record the step
            data['preprocessing_steps'].append({
                'step': 'my_custom_filter',
                'cutoff_freq': cutoff_freq
            })

            return data
        ```
        """
        custom_steps_folder = Path(custom_steps_folder)

        if not custom_steps_folder.exists():
            raise ValueError(f"Custom steps folder does not exist: {custom_steps_folder}")

        if not custom_steps_folder.is_dir():
            raise ValueError(f"Custom steps folder is not a directory: {custom_steps_folder}")

        custom_steps = {}
        python_files = list(custom_steps_folder.glob("*.py"))

        logger.info(f"Searching for custom steps in: {custom_steps_folder}")
        logger.info(f"Found {len(python_files)} Python file(s)")

        for py_file in python_files:
            # Skip __init__.py and files starting with underscore
            if py_file.name.startswith('_'):
                logger.debug(f"Skipping {py_file.name}")
                continue

            try:
                # Create a unique module name to avoid conflicts
                module_name = f"custom_steps.{py_file.stem}"

                # Load the module
                spec = importlib.util.spec_from_file_location(module_name, py_file)
                if spec is None or spec.loader is None:
                    logger.warning(f"Could not load module spec for {py_file}")
                    continue

                module = importlib.util.module_from_spec(spec)
                sys.modules[module_name] = module
                spec.loader.exec_module(module)

                # Find all functions in the module that match the step signature
                for name, obj in inspect.getmembers(module, inspect.isfunction):
                    # Skip private functions
                    if name.startswith('_'):
                        continue

                    # Check function signature
                    sig = inspect.signature(obj)
                    params = list(sig.parameters.keys())

                    # Step functions should accept exactly 2 parameters: data and step_config
                    if len(params) == 2:
                        custom_steps[name] = obj
                        logger.info(f"Loaded custom step '{name}' from {py_file.name}")
                    else:
                        logger.debug(f"Skipping function '{name}' in {py_file.name} - "
                                   f"expected 2 parameters, found {len(params)}")

            except Exception as e:
                logger.error(f"Error loading custom steps from {py_file}: {e}")
                # Continue loading other files even if one fails
                continue

        if not custom_steps:
            logger.warning(f"No valid custom steps found in {custom_steps_folder}")

        return custom_steps

    def run_pipeline(
        self,
        subjects: Union[str, List[str]] = None,
        sessions: Union[str, List[str]] = None,
        tasks: Union[str, List[str]] = None,
        acquisitions: Union[str, List[str]] = None,
        runs: Union[str, List[str]] = None,
        extension: str = '.vhdr',
        io_backend: str = 'read_raw_bids'
    ) -> Dict[str, Any]:
        """Run the pipeline using the configured reader to find files.

        Acts as the job manager: discovers the recording list via the
        reader, then dispatches one job per recording through
        :func:`meegflow.execution.dispatch`. The execution backend
        (sequential single-process, an in-process/local Dask cluster, or a
        ``dask-jobqueue`` HPC cluster) is selected via the ``execution``
        block of ``config`` (see
        :class:`~meegflow.execution.ExecutionConfig`); if absent, execution
        is sequential and single-process, matching previous behavior.

        Parameters
        ----------
        subjects : str | list of str | None
            Subject ID(s) to process. None matches all subjects.
        sessions : str | list of str | None
            Session ID(s) to process. None matches all sessions.
        tasks : str | list of str | None
            Task(s) to process. None matches all tasks.
        acquisitions : str | list of str | None
            Acquisition parameter(s). None matches all acquisitions.
        runs : str | list of str | None
            Run ID(s) to process. None matches all runs.
        extension : str
            File extension to match (default: ``'.vhdr'``).
        io_backend : str
            MNE IO function used to read each file (default:
            ``'read_raw_bids'``). Any function name resolvable via
            ``mne.io`` can be supplied (e.g. ``'read_raw_eeglab'``).

        Returns
        -------
        all_results : dict
            Dictionary mapping recording name -> result dict. Each result
            contains the keys set by whichever output steps ran (e.g.
            ``'raw_file'``, ``'epochs_file'``, ``'json_report'``,
            ``'html_report'``), or an ``'error'`` key with the exception if
            processing failed.
        """
        recordings = self.reader.find_recordings(
            subjects=subjects,
            sessions=sessions,
            tasks=tasks,
            acquisitions=acquisitions,
            runs=runs,
            extension=extension
        )

        logger.info(f"Found {len(recordings)} recording(s) to process")

        exec_config = ExecutionConfig.from_config(self.config)
        logger.info(
            f"Execution backend: {exec_config.backend}"
            + (f" (n_workers={exec_config.n_workers})" if exec_config.n_workers != 1 else "")
        )

        all_results = dispatch(
            recordings,
            reader=self.reader,
            output_root=self.output_root,
            config=self.config,
            step_functions=self.step_functions,
            io_backend=io_backend,
            exec_config=exec_config,
        )

        logger.info(f"Pipeline completed.")
        return all_results

dataset_root property

dataset_root

Get the dataset root path from the reader.

__init__

__init__(reader, output_root=None, config=None)

Initialize MEEGFlow preprocessing pipeline.

Parameters

reader : DatasetReader Reader instance for discovering data files. Use BIDSReader for BIDS datasets or GlobReader for custom directory structures. output_root : str or Path, optional Path to output derivatives root. If not provided, defaults to {dataset_root}/derivatives/meegflow config : dict, optional Configuration dictionary containing pipeline steps and parameters

Source code in src/meegflow/pipeline.py

def __init__(
    self, 
    reader: DatasetReader,
    output_root: Union[str, Path] = None, 
    config: Dict[str, Any] = None
):
    """Initialize MEEGFlow preprocessing pipeline.

    Parameters
    ----------
    reader : DatasetReader
        Reader instance for discovering data files. Use BIDSReader for BIDS datasets
        or GlobReader for custom directory structures.
    output_root : str or Path, optional
        Path to output derivatives root. If not provided, defaults to
        {dataset_root}/derivatives/meegflow
    config : dict, optional
        Configuration dictionary containing pipeline steps and parameters
    """
    self.config = config or {}
    self.output_root = Path(output_root) if output_root else None
    self.reader = reader

    # Validate the top-level datatype. If the configuration names one, a
    # BIDS reader that was not given its own datatype searches that one.
    self.datatype = resolve_datatype(self.config)
    if self.config.get('datatype') is not None and getattr(reader, 'datatype', False) is None:
        reader.datatype = self.datatype

    # Built-in steps come from the registry (populated by importing the
    # steps package); custom steps may add to or override them by name.
    self.step_functions = build_step_functions(self.config)

    # Validate pipeline steps if provided in config
    pipeline_cfg = self.config.get('pipeline', [])
    unknown = [s.get('name') for s in pipeline_cfg if s.get('name') not in self.step_functions]
    if unknown:
        raise ValueError(f"Unknown pipeline steps in config: {unknown}")

run_pipeline

run_pipeline(subjects=None, sessions=None, tasks=None, acquisitions=None, runs=None, extension='.vhdr', io_backend='read_raw_bids')

Run the pipeline using the configured reader to find files.

Acts as the job manager: discovers the recording list via the reader, then dispatches one job per recording through :func:meegflow.execution.dispatch. The execution backend (sequential single-process, an in-process/local Dask cluster, or a dask-jobqueue HPC cluster) is selected via the execution block of config (see :class:~meegflow.execution.ExecutionConfig); if absent, execution is sequential and single-process, matching previous behavior.

Parameters

subjects : str | list of str | None Subject ID(s) to process. None matches all subjects. sessions : str | list of str | None Session ID(s) to process. None matches all sessions. tasks : str | list of str | None Task(s) to process. None matches all tasks. acquisitions : str | list of str | None Acquisition parameter(s). None matches all acquisitions. runs : str | list of str | None Run ID(s) to process. None matches all runs. extension : str File extension to match (default: '.vhdr'). io_backend : str MNE IO function used to read each file (default: 'read_raw_bids'). Any function name resolvable via mne.io can be supplied (e.g. 'read_raw_eeglab').

Returns

all_results : dict Dictionary mapping recording name -> result dict. Each result contains the keys set by whichever output steps ran (e.g. 'raw_file', 'epochs_file', 'json_report', 'html_report'), or an 'error' key with the exception if processing failed.

Source code in src/meegflow/pipeline.py

def run_pipeline(
    self,
    subjects: Union[str, List[str]] = None,
    sessions: Union[str, List[str]] = None,
    tasks: Union[str, List[str]] = None,
    acquisitions: Union[str, List[str]] = None,
    runs: Union[str, List[str]] = None,
    extension: str = '.vhdr',
    io_backend: str = 'read_raw_bids'
) -> Dict[str, Any]:
    """Run the pipeline using the configured reader to find files.

    Acts as the job manager: discovers the recording list via the
    reader, then dispatches one job per recording through
    :func:`meegflow.execution.dispatch`. The execution backend
    (sequential single-process, an in-process/local Dask cluster, or a
    ``dask-jobqueue`` HPC cluster) is selected via the ``execution``
    block of ``config`` (see
    :class:`~meegflow.execution.ExecutionConfig`); if absent, execution
    is sequential and single-process, matching previous behavior.

    Parameters
    ----------
    subjects : str | list of str | None
        Subject ID(s) to process. None matches all subjects.
    sessions : str | list of str | None
        Session ID(s) to process. None matches all sessions.
    tasks : str | list of str | None
        Task(s) to process. None matches all tasks.
    acquisitions : str | list of str | None
        Acquisition parameter(s). None matches all acquisitions.
    runs : str | list of str | None
        Run ID(s) to process. None matches all runs.
    extension : str
        File extension to match (default: ``'.vhdr'``).
    io_backend : str
        MNE IO function used to read each file (default:
        ``'read_raw_bids'``). Any function name resolvable via
        ``mne.io`` can be supplied (e.g. ``'read_raw_eeglab'``).

    Returns
    -------
    all_results : dict
        Dictionary mapping recording name -> result dict. Each result
        contains the keys set by whichever output steps ran (e.g.
        ``'raw_file'``, ``'epochs_file'``, ``'json_report'``,
        ``'html_report'``), or an ``'error'`` key with the exception if
        processing failed.
    """
    recordings = self.reader.find_recordings(
        subjects=subjects,
        sessions=sessions,
        tasks=tasks,
        acquisitions=acquisitions,
        runs=runs,
        extension=extension
    )

    logger.info(f"Found {len(recordings)} recording(s) to process")

    exec_config = ExecutionConfig.from_config(self.config)
    logger.info(
        f"Execution backend: {exec_config.backend}"
        + (f" (n_workers={exec_config.n_workers})" if exec_config.n_workers != 1 else "")
    )

    all_results = dispatch(
        recordings,
        reader=self.reader,
        output_root=self.output_root,
        config=self.config,
        step_functions=self.step_functions,
        io_backend=io_backend,
        exec_config=exec_config,
    )

    logger.info(f"Pipeline completed.")
    return all_results