Task

The Task class is an abstract base class that defines the interface for all EEG processing tasks in AutoClean.

class autoclean.core.task.Task(config)

Bases: ABC, CombinedAutocleanMixins

Base class for all EEG processing tasks.

This class defines the interface that all specific EEG tasks must implement. It provides the basic structure for: 1. Loading and validating configuration 2. Importing raw EEG data 3. Running preprocessing steps 4. Applying task-specific processing 5. Saving results

It should be inherited from to create new tasks in the autoclean.tasks module.

Notes

Abstract base class that enforces a consistent interface across all EEG processing tasks through abstract methods and strict type checking. Manages state through MNE objects (Raw and Epochs) while maintaining processing history in a dictionary.

Initialization

__init__(config)

Initialize a new task instance.

Parameters:

configDict[str, Any]

A dictionary containing all configuration settings for the task. Must include:

• run_id (str): Unique identifier for this processing run

• unprocessed_file (Path): Path to the raw EEG data file

• task (str): Name of the task (e.g., “rest_eyesopen”)

The base class automatically detects a module-level ‘config’ variable and uses it for self.settings in Python-based tasks.

Examples

>>> # Python task file approach - no __init__ needed!
>>> config = {'resample': {'enabled': True, 'value': 250}}
>>> class MyTask(Task):
...     def run(self):
...         self.import_raw()
...         # Processing steps here

Core Methods

abstract run()

Run the standard EEG preprocessing pipeline.

Notes

Defines interface for MNE-based preprocessing operations including filtering, resampling, and artifact detection. Maintains processing state through self.raw modifications.

The specific parameters for each preprocessing step should be defined in the task configuration and validated before use.

import_raw()

Import the raw EEG data from file.

Notes

Imports data using the configured import function and flags files with duration less than 60 seconds. Saves the imported data as a post-import stage file.

validate_config(config)

Validate the complete task configuration.

Parameters:

configDict[str, Any]

The configuration dictionary to validate. See __init__ docstring for required fields.

Returns:

Dict[str, Any]

The validated configuration dictionary. May contain additional fields added during validation.

Notes

Implements two-stage validation pattern with base validation followed by task-specific checks. Uses type annotations and runtime checks to ensure configuration integrity before processing begins.

Examples

>>> config = {...}  # Your configuration dictionary
>>> validated_config = task.validate_config(config)
>>> print(f"Validation successful: {validated_config['task']}")

Getter Methods

get_raw()

Get the raw data of the task.

Returns:

mne.io.Raw

The raw data of the task.

get_epochs()

Get the epochs of the task.

Returns:

mne.Epochs

The epochs of the task.

get_flagged_status()

Get the flagged status of the task.

Returns:

tuple of (bool, list of str)

A tuple containing a boolean flag and a list of reasons for flagging.

get_epochs()

Get the epochs of the task.

Returns:

mne.Epochs

The epochs of the task.

get_flagged_status()

Get the flagged status of the task.

Returns:

tuple of (bool, list of str)

A tuple containing a boolean flag and a list of reasons for flagging.

get_raw()

Get the raw data of the task.

Returns:

mne.io.Raw

The raw data of the task.

import_epochs()

Import the epochs from file.

Notes

Imports data using the configured import function and saves the imported data as a post-import stage file.

import_raw()

Import the raw EEG data from file.

Notes

Imports data using the configured import function and flags files with duration less than 60 seconds. Saves the imported data as a post-import stage file.

abstract run()

Run the standard EEG preprocessing pipeline.

Notes

Defines interface for MNE-based preprocessing operations including filtering, resampling, and artifact detection. Maintains processing state through self.raw modifications.

The specific parameters for each preprocessing step should be defined in the task configuration and validated before use.

validate_config(config)

Validate the complete task configuration.

Parameters:

configDict[str, Any]

The configuration dictionary to validate. See __init__ docstring for required fields.

Returns:

Dict[str, Any]

The validated configuration dictionary. May contain additional fields added during validation.

Notes

Implements two-stage validation pattern with base validation followed by task-specific checks. Uses type annotations and runtime checks to ensure configuration integrity before processing begins.

Examples

>>> config = {...}  # Your configuration dictionary
>>> validated_config = task.validate_config(config)
>>> print(f"Validation successful: {validated_config['task']}")

Creating Custom Tasks

To create a custom task, subclass the Task class and implement the required methods:

from autoclean.core.task import Task

class MyCustomTask(Task):
    """Custom task implementation for special EEG preprocessing."""

    def __init__(self, config):
        self.raw = None
        self.epochs = None
        super().__init__(config)

    def run(self):
        """Run the complete processing pipeline."""
        # Implement your custom processing steps here
        pass

    def _validate_task_config(self, config):
        """Validate task-specific configuration."""
        # Perform validation of task-specific parameters
        return config