ezmsg.baseproc.composite

Composite processor classes for building pipelines.

Classes

class CompositeProcessor(*args, **kwargs)

Bases: BaseProcessor[SettingsType, MessageInType, MessageOutType], CompositeStateful[SettingsType, MessageOutType], ABC, Generic[SettingsType, MessageInType, MessageOutType]

A processor that chains multiple processor together in a feedforward non-branching graph. The individual processors may be stateless or stateful. The last processor may be a consumer, otherwise processors must be transformers. Use CompositeProducer if you want the first processor to be a producer. Concrete subclasses must implement _initialize_processors. Optionally override _reset_state if you want adaptive state behaviour. Example implementation:

class CustomCompositeProcessor(CompositeProcessor[CustomSettings, AxisArray, AxisArray]):

@staticmethod def _initialize_processors(settings: CustomSettings) -> dict[str, BaseProcessor]:

return {

“stateful_transformer”: CustomStatefulProducer(**settings), “transformer”: CustomTransformer(**settings),

}

Where **settings should be replaced with initialisation arguments for each processor.

__init__(*args, **kwargs)

Return type:

None

stateful_op(state, message)

Return type:

tuple[dict[str, tuple[Any, int]], Optional[TypeVar(MessageOutType)]]

Parameters:

• state (dict[str, tuple[Any, int]] | None)

• message (MessageInType | None)

class CompositeProducer(*args, **kwargs)

Bases: BaseProducer[SettingsType, MessageOutType], CompositeStateful[SettingsType, MessageOutType], ABC, Generic[SettingsType, MessageOutType]

A producer that chains multiple processors (starting with a producer) together in a feedforward non-branching graph. The individual processors may be stateless or stateful. The first processor must be a producer, the last processor may be a consumer, otherwise processors must be transformers.

__init__(*args, **kwargs)

Return type:

None

stateful_op(state)

Return type:

tuple[dict[str, tuple[Any, int]], Optional[TypeVar(MessageOutType)]]

Parameters:

state (dict[str, tuple[Any, int]] | None)

class CompositeStateful

Bases: Stateful[dict[str, Any]], ABC, Generic[SettingsType, MessageOutType]

Mixin class for composite processor/producer chains. DO NOT use this class directly. Used to enforce statefulness of the composite processor/producer chain and provide initialization and validation methods.

property state: dict[str, Any]

abstractmethod stateful_op(state, *args, **kwargs)

Return type:

tuple[dict[str, tuple[Any, int]], Optional[TypeVar(MessageOutType)]]

Parameters:

• state (dict[str, tuple[Any, int]] | None)

• args (Any)

• kwargs (Any)

class CompositeStateful

Bases: Stateful[dict[str, Any]], ABC, Generic[SettingsType, MessageOutType]

Mixin class for composite processor/producer chains. DO NOT use this class directly. Used to enforce statefulness of the composite processor/producer chain and provide initialization and validation methods.

property state: dict[str, Any]

abstractmethod stateful_op(state, *args, **kwargs)

Return type:

tuple[dict[str, tuple[Any, int]], Optional[TypeVar(MessageOutType)]]

Parameters:

• state (dict[str, tuple[Any, int]] | None)

• args (Any)

• kwargs (Any)

class CompositeProcessor(*args, **kwargs)

Bases: BaseProcessor[SettingsType, MessageInType, MessageOutType], CompositeStateful[SettingsType, MessageOutType], ABC, Generic[SettingsType, MessageInType, MessageOutType]

A processor that chains multiple processor together in a feedforward non-branching graph. The individual processors may be stateless or stateful. The last processor may be a consumer, otherwise processors must be transformers. Use CompositeProducer if you want the first processor to be a producer. Concrete subclasses must implement _initialize_processors. Optionally override _reset_state if you want adaptive state behaviour. Example implementation:

class CustomCompositeProcessor(CompositeProcessor[CustomSettings, AxisArray, AxisArray]):

@staticmethod def _initialize_processors(settings: CustomSettings) -> dict[str, BaseProcessor]:

return {

“stateful_transformer”: CustomStatefulProducer(**settings), “transformer”: CustomTransformer(**settings),

}

Where **settings should be replaced with initialisation arguments for each processor.

__init__(*args, **kwargs)

Return type:

None

stateful_op(state, message)

Return type:

tuple[dict[str, tuple[Any, int]], Optional[TypeVar(MessageOutType)]]

Parameters:

• state (dict[str, tuple[Any, int]] | None)

• message (MessageInType | None)

class CompositeProducer(*args, **kwargs)

Bases: BaseProducer[SettingsType, MessageOutType], CompositeStateful[SettingsType, MessageOutType], ABC, Generic[SettingsType, MessageOutType]

A producer that chains multiple processors (starting with a producer) together in a feedforward non-branching graph. The individual processors may be stateless or stateful. The first processor must be a producer, the last processor may be a consumer, otherwise processors must be transformers.

__init__(*args, **kwargs)

Return type:

None

stateful_op(state)

Return type:

tuple[dict[str, tuple[Any, int]], Optional[TypeVar(MessageOutType)]]

Parameters:

state (dict[str, tuple[Any, int]] | None)