ezmsg.qt.chain#

ProcessorGraph - Fluent DAG API for compiled processing graphs.

Functions

normalize_topic(topic)[source]#
Return type:

str

Parameters:

topic (str | Enum)

Classes

class BoundProcessor(processor, name=None, input_name=None, output_name=None, inputs=None)[source]#

Bases: object

Wrap a processor spec with optional metadata and stream overrides.

Parameters:
processor: type[Unit] | tuple[type[Unit], Settings] | Any#
name: str | None = None#
input_name: str | None = None#
output_name: str | None = None#
inputs: dict[str, Enum | str] | None = None#
__init__(processor, name=None, input_name=None, output_name=None, inputs=None)#
Parameters:
Return type:

None

class GraphSink(sink_id, source_ref, slot)[source]#

Bases: object

A Qt-facing observer attached to a tail in the graph.

Parameters:
sink_id: str#
source_ref: str#
slot: Callable[[Any], None]#
__init__(sink_id, source_ref, slot)#
Parameters:
Return type:

None

class ProcessorGraph(source_topic, parent=None, auto_gate=False, auto_gate_position='input')[source]#

Bases: ProcessorPath

Fluent builder for Qt-facing processor graphs.

Parameters:

  • source_topic (Enum | str)

  • parent (QtWidgets.QWidget | None)

  • auto_gate (bool)

  • auto_gate_position (AutoGatePosition)

__init__(source_topic, parent=None, auto_gate=False, auto_gate_position='input')[source]#
Parameters:

  • source_topic (Enum | str)

  • parent (QWidget | None)

  • auto_gate (bool)

  • auto_gate_position (Literal['input', 'output'])

property source_topic: Enum | str#
property parent_widget: QWidget | None#
property auto_gate: bool#
property auto_gate_position: Literal['input', 'output']#
property stages: list[ProcessorStage]#
property sinks: list[GraphSink]#
property session: EzSession | None#
property attached: bool#
attach(session)[source]#
Return type:

ProcessorGraph

Parameters:

session (EzSession)

settings_bindings()[source]#
Return type:

list[ProcessorSettingsBinding]

input_bindings()[source]#
Return type:

list[ProcessorInputBinding]

class ProcessorInputBinding(node_id, processor_index, input_name, topic)[source]#

Bases: object

Bind a public topic to a named processor input stream.

Parameters:

  • node_id (str)

  • processor_index (int)

  • input_name (str)

  • topic (str)

node_id: str#
processor_index: int#
input_name: str#
topic: str#
__init__(node_id, processor_index, input_name, topic)#
Parameters:

  • node_id (str)

  • processor_index (int)

  • input_name (str)

  • topic (str)

Return type:

None

class ProcessorPath(graph, tail_ref)[source]#

Bases: object

Mutable builder for a path in a processor graph.

Parameters:
__init__(graph, tail_ref)[source]#
Parameters:
property graph: ProcessorGraph#
property tail_ref: str#
parallel(*processors)[source]#
Return type:

Self

Parameters:

processors (type[Unit] | tuple[type[Unit], Settings] | BoundProcessor | Any)

local(*processors)[source]#
Return type:

Self

Parameters:

processors (type[Unit] | tuple[type[Unit], Settings] | BoundProcessor | Any)

connect(slot)[source]#
Return type:

Self

Parameters:

slot (Callable[[Any], None])

apply(recipe)[source]#
Return type:

Self

Parameters:

recipe (Callable[[ProcessorPath], Any])

branch(recipe)[source]#
Return type:

Self

Parameters:

recipe (Callable[[ProcessorPath], Any])

class ProcessorSettingsBinding(name, title, node_id, processor_index, settings_type, topic, initial_settings)[source]#

Bases: object

Runtime settings metadata for a processor in the graph.

Parameters:

  • name (str)

  • title (str)

  • node_id (str)

  • processor_index (int)

  • settings_type (type[Any])

  • topic (str)

  • initial_settings (Any | None)

name: str#
title: str#
node_id: str#
processor_index: int#
settings_type: type[Any]#
topic: str#
initial_settings: Any | None#
__init__(name, title, node_id, processor_index, settings_type, topic, initial_settings)#
Parameters:

  • name (str)

  • title (str)

  • node_id (str)

  • processor_index (int)

  • settings_type (type[Any])

  • topic (str)

  • initial_settings (Any | None)

Return type:

None

class ProcessorStage(stage_id, source_ref, processors, mode)[source]#

Bases: object

A linear group of processors sourced from an upstream tail.

Parameters:
stage_id: str#
source_ref: str#
processors: tuple[type[Unit] | tuple[type[Unit], Settings] | BoundProcessor | Any, ...]#
mode: Literal['shared', 'process']#
__init__(stage_id, source_ref, processors, mode)#
Parameters:
Return type:

None