processors

dreem.utils.processors

Base processor classes for DREEM processing pipelines.

Classes:

Name	Description
ProcessingStep	Abstract base class for processing steps.

ProcessingStep

Bases: ABC

Abstract base class for processing steps.

Each step accepts a state dict, modifies it, and returns it. Steps are instantiated once and can be called multiple times in a loop.

Methods:

Name	Description
__init__	Initialize a processing step.
run	Execute the processing step.

Source code in dreem/utils/processors.py

class ProcessingStep(ABC):
    """Abstract base class for processing steps.

    Each step accepts a state dict, modifies it, and returns it.
    Steps are instantiated once and can be called multiple times in a loop.
    """

    def __init__(self, name: Optional[str] = None):
        """Initialize a processing step.

        Args:
            name: Optional name for the step. Defaults to class name.
        """
        self.name = name or self.__class__.__name__

    @abstractmethod
    def run(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """Execute the processing step.

        Args:
            state: Dictionary containing all necessary data for processing.
                   Modified in place and returned.

        Returns:
            The modified state dictionary.
        """
        ...

__init__(name=None)

Initialize a processing step.

Parameters:

Name	Type	Description	Default
name	Optional[str]	Optional name for the step. Defaults to class name.	None

Source code in dreem/utils/processors.py

def __init__(self, name: Optional[str] = None):
    """Initialize a processing step.

    Args:
        name: Optional name for the step. Defaults to class name.
    """
    self.name = name or self.__class__.__name__

run(state) abstractmethod

Execute the processing step.

Parameters:

Name	Type	Description	Default
state	Dict[str, Any]	Dictionary containing all necessary data for processing. Modified in place and returned.	required

Returns:

Type	Description
Dict[str, Any]	The modified state dictionary.

Source code in dreem/utils/processors.py

@abstractmethod
def run(self, state: Dict[str, Any]) -> Dict[str, Any]:
    """Execute the processing step.

    Args:
        state: Dictionary containing all necessary data for processing.
               Modified in place and returned.

    Returns:
        The modified state dictionary.
    """
    ...