ezmsg.qt.sidecar#

Sidecar runtime compilation for processor graphs.

Functions

build_sidecar_components(graphs, topic_prefix='_qt')[source]#
Return type:

tuple[dict[str, Component], list[tuple[Any, Any]], tuple[Component, ...], list[CompiledGraph]]

Parameters:
normalize_topic(topic)[source]#
Return type:

str

Parameters:

topic (str | Enum)

Classes

class CompiledGraph(graph, gate_component_name, stage_component_names, source_topic, gate_topic, sinks, settings_bindings=(), input_bindings=())[source]#

Bases: object

Parameters:
graph: ProcessorGraph#
gate_component_name: str | None#
stage_component_names: tuple[str, ...]#
source_topic: str#
gate_topic: str | None#
sinks: tuple[CompiledSink, ...]#
settings_bindings: tuple[ProcessorSettingsBinding, ...] = ()#
input_bindings: tuple[ProcessorInputBinding, ...] = ()#
__init__(graph, gate_component_name, stage_component_names, source_topic, gate_topic, sinks, settings_bindings=(), input_bindings=())#
Parameters:
Return type:

None

class CompiledSink(sink_id, topic, slot, gate_component_name=None)[source]#

Bases: object

Parameters:
sink_id: str#
topic: str#
slot: Callable[[Any], None]#
gate_component_name: str | None = None#
__init__(sink_id, topic, slot, gate_component_name=None)#
Parameters:
Return type:

None

class ProcessorStageCollection(processors)[source]#

Bases: Collection

Collection wrapper for a linear processor stage.

Parameters:

processors (tuple[Any, ...])

INPUT = InputTopicStream:unlocated[Any]()#
OUTPUT = OutputTopicStream:unlocated[Any]()#
__init__(processors)[source]#

Initialize an Addressable object.

The name and location are initially None and must be set before the object can be properly addressed. This is achieved through the _set_name() and _set_location() methods.

Parameters:

processors (tuple[Any, ...])

network()[source]#

Override this method and have the definition return a NetworkDefinition which defines how InputStreams and OutputStreams from member Units will be connected.

The NetworkDefinition specifies the message routing between components by connecting output streams to input streams.

Returns:

Network definition specifying stream connections

Return type:

Iterable[tuple[Stream | str, Stream | str]]

process_components()[source]#

Override this method and have the definition return a tuple which contains Units and Collections which should run in their own processes.

This method allows you to specify which components should be isolated in separate processes for performance or isolation requirements.

Returns:

Collection of components that should run in separate processes

Return type:

tuple[Component, ...]