herostools.flow

Submodules

• herostools.flow._subscriber
• herostools.flow.dummy
• herostools.flow.join
• herostools.flow.node

Classes

Join	Node that collects one message per input and then processes them together.
EventInputNode	Node that feeds a flow pipeline from an arbitrary HEROS event.
Node	Base flow pipeline node.

Functions

only(inputs) → Any

Return the single value of a one-entry inputs dict.

Package Contents

class herostools.flow.Join(inputs: collections.abc.Mapping[str, heros.HERORef | str] | collections.abc.Sequence[heros.HERORef | str], *, max_open_rounds: int = 10, **kwargs)

Bases: herostools.flow.node.Node

Node that collects one message per input and then processes them together.

Messages are grouped into rounds by their identifier. _process runs once per completed round, with the data of every input keyed by input name, in config order. The default _process emits that dict. The rules:

• A round completes when every configured input has delivered. The metadata of the inputs is merged in config order. Their flow paths are joined in config order without duplicates. Completed rounds are processed in completion order.

• At most max_open_rounds incomplete rounds are open. A message with a new identifier at the limit drops the oldest incomplete round and logs a warning that names its identifier and missing inputs. The warning is logged once per crossing of the limit, and again after the number of open rounds fell below the limit.

• The last max_open_rounds finished or dropped identifiers are remembered. Later messages for them are ignored.

• A repeated message from an input for an open round replaces the earlier one.

• A message that fails validation does not count.

• feed processes its message alone, as input “manual”. It never joins a round.

Rounds are only touched on the worker, so they need no lock.

_hero_implements: ClassVar[list[str]] = ['herostools.flow.Node', 'herostools.flow.Join']

_max_open_rounds = 10

_rounds: dict[Any, dict[str, tuple[Any, dict]]]

_finished: collections.deque

_warned = False

_process(inputs: dict[str, Any], metadata: dict) → tuple[Any, dict] | None

Emit the dict of all inputs. Override in subclasses to combine them.

_collect(input_name: str, data: Any, metadata: dict) → tuple[dict, dict] | None

Add the message to its round and return the round once every input delivered.

_drop_oldest_round() → None

Drop the oldest incomplete round, with a warning once per crossing of the limit.

class herostools.flow.EventInputNode(object_selector: str, event_name: str, *args, identifier_key: str = 'identifier', queue_warn_threshold: int = 100, **kwargs)

Bases: Node, heros.EventObserver

Node that feeds a flow pipeline from an arbitrary HEROS event.

The emitting HERO of each observed event names the input of its messages. Every event is processed individually. The node has no configured inputs. Events without an identifier in their metadata are rejected like any message.

_teardown() → None

Undeclare the event subscription, then tear the node down. Called by BOSS on shutdown.

handle_event(source: str, payload: Any) → None

Queue an incoming HEROS event payload for conversion into flow data on the worker.

This method is local only and not exposed over RPC.

Parameters:

• source – Name of the emitting HERO.

• payload – Event payload, either a (data, metadata) pair or plain data.

_convert_event(source: str, payload: Any) → None

Convert and process an event on the worker.

This must not be called _handle_event, which EventObserver uses as its zenoh callback. Events of this node itself are ignored, so a selector that matches it does not loop.

class herostools.flow.Node(inputs: collections.abc.Mapping[str, heros.HERORef | str] | collections.abc.Sequence[heros.HERORef | str] | None = None, identifier_key: str = 'identifier', subscriber: herostools.flow._subscriber.FlowSubscriber | None = None, queue_warn_threshold: int = 100)

Base flow pipeline node.

A node has named inputs. Each input is fed by one source HERO. The inputs argument declares them in the node config, as a mapping from input name to source or as a list of sources. A source is a heros.HERORef or a HERO name string. The inputs are fixed at construction. _setup, which BOSS calls after creating the node, subscribes to every configured source. Every incoming message must carry metadata[identifier_key]. Every emitted message lists the nodes it passed in metadata[“_flow_path”]. A message that already passed this node is dropped with a warning, so a loop in the flow ends there.

The node runs _process once per message, with a one-entry dict keyed by the name of the input that delivered it. Join runs it once per completed round instead.

Messages arrive by pub/sub. The node subscribes to the flow_data event of every source and processes the messages on one worker thread, so emitting upstream never waits for this node. The subscriber is created on first use from the zenoh session and realm that the HERO mixin provides, so it only works once BOSS has turned the node into a HERO.

The identifier_key and a copy of the inputs are readable remotely.

_boss_heroref_arguments makes BOSS pass “$name” strings in inputs as heros.HERORef values without looking the HERO up. Subclasses inherit it.

The queue is unbounded. The subscriber logs a warning each time its depth crosses queue_warn_threshold.

Subclasses override _validate and _process. _process may return None to emit nothing for that call. _collect decides which messages are processed together; Join overrides it. Subclasses with their own __init__ must call super().__init__().

A subclass can call _submit(func) to run func on the worker, in order with the messages, for example to flush accumulated state on a timeout. It does nothing after teardown. Submitted work bypasses process and the metadata merge. It emits by calling self.flow_data(data, metadata) with the complete metadata. The worker logs an exception raised by func and keeps running.

_hero_implements: ClassVar[list[str]] = ['herostools.flow.Node']

_boss_heroref_arguments: ClassVar[frozenset[str]]

_input_refs

_inputs

_input_of_source

_identifier_key = 'identifier'

_subscriber = None

_queue_warn_threshold = 100

_subscriber_lock

_closed = False

property identifier_key: str

Metadata key holding the identifier of each message. Read-only.

property inputs: dict[str, str]

Copy of the inputs, mapping each input name to its source HERO name. Read-only.

_setup() → None

Subscribe to the source of every input. Called by BOSS after creation.

Every heros.HERORef source must be in the realm of this node. The check is skipped for a node without a HERO realm, such as one that was given a subscriber and is no HERO. Plain name sources are not checked.

Raises:

ValueError – If a heros.HERORef source is from another realm, or a source is this node.

property name: str

the HERO name once BOSS created it, otherwise the class name.

Type:

Name of the node

_get_subscriber() → herostools.flow._subscriber.FlowSubscriber

Return the subscriber, creating it from the HERO session and realm on first use.

Raises:

RuntimeError – If the node was torn down, or is no HERO and was given no subscriber.

_open_subscriber() → herostools.flow._subscriber.FlowSubscriber | None

Return the subscriber, or None (logged at debug level) once the node is torn down.

_submit(func: collections.abc.Callable[[], None]) → None

Run a callable on the worker, in order with the messages. Does nothing after teardown.

_on_flow_data(source: str, payload: Any) → None

Handle a decoded flow_data payload on the worker.

Parameters:

• source – Name of the HERO the message was received from.

• payload – Tuple of data, metadata and the emitting node’s name.

_teardown() → None

Stop the subscriber. Called by BOSS on shutdown.

feed(data: Any, metadata: dict) → None

Process one message as if it came from an input called “manual”.

Returns at once. The message is validated, processed and emitted on the worker, in order with the subscribed messages. It does not belong to any configured input.

Parameters:

• data – Input data.

• metadata – Input metadata. Must contain identifier_key.

Raises:

ValueError – If the identifier is missing or not hashable.

_feed(data: Any, metadata: dict) → None

Validate, process and emit a fed message. Runs on the worker.

_identifier_error(metadata: dict) → str | None

Return why the metadata has no usable identifier, or None.

_is_valid(input_name: str, data: Any, metadata: dict, source: str) → bool

Run validate for one message, logging failures.

validate(inputs: dict[str, Any], metadata: dict) → bool

Validate incoming flow data.

Local only, not exposed over RPC. Subclasses override _validate.

Parameters:

• inputs – Input data keyed by input name.

• metadata – Input metadata.

Returns:

Whether the data is valid.

_validate(inputs: dict[str, Any], metadata: dict) → bool

Check one incoming message. Override point for subclasses.

The default accepts everything. A message that fails validation is dropped. A plain False is logged at debug level, an exception as an error.

Parameters:

• inputs – One-entry dict with the message’s data keyed by its input name.

• metadata – Input metadata.

Returns:

Whether the message is valid.

process(inputs: dict[str, Any], metadata: dict) → tuple[Any, dict] | None

Process collected flow data.

Local only, not exposed over RPC. Subclasses override _process.

Parameters:

• inputs – Input data keyed by input name. One entry, the input that delivered, except in a Join, which passes every input.

• metadata – Input metadata.

Returns:

Tuple of processed data and additional metadata, or None to emit nothing for this call.

_process(inputs: dict[str, Any], metadata: dict) → tuple[Any, dict] | None

Compute the output of one message or one Join round. Override point for subclasses.

The default forwards the data unchanged. An exception is logged and nothing is emitted for that call.

Parameters:

• inputs – Input data keyed by input name. One entry, the input that delivered, except in a Join, which passes every input.

• metadata – Input metadata.

Returns:

Tuple of output data and additional metadata, which is merged over metadata. None emits nothing for this call, for nodes that accumulate messages over several calls.

_handle_flow_data(data: Any, metadata: dict, source: str) → None

Map a message to its input and handle it. Runs on the worker.

Parameters:

• data – Input data.

• metadata – Input metadata.

• source – Name of the upstream node.

_handle_input(input_name: str, data: Any, metadata: dict, source: str) → None

Validate, process and emit one message. Runs on the worker.

Parameters:

• input_name – Name of the input that delivered the message.

• data – Input data.

• metadata – Input metadata.

• source – Name of the upstream node.

_collect(input_name: str, data: Any, metadata: dict) → tuple[dict, dict] | None

Decide what to process after one valid message. Override point for subclasses.

The default hands every message on alone.

Parameters:

• input_name – Name of the input that delivered the message.

• data – Input data.

• metadata – Input metadata, with a hashable identifier.

Returns:

Tuple of inputs keyed by input name and their metadata, to pass to _process. None waits for more messages.

_fire(ready: tuple[dict, dict]) → None

Process and emit one message, with this node appended to its flow path.

flow_data(data: Any, metadata: dict) → tuple[Any, dict, str]

Emit processed flow data as a HEROS event.

Parameters:

• data – Output data.

• metadata – Output metadata.

Returns:

Tuple of data, metadata and this node’s name.

herostools.flow.only(inputs: dict[str, Any]) → Any

Return the single value of a one-entry inputs dict.

Parameters:

inputs – Inputs keyed by input name.

Returns:

The only value in inputs.

Raises:

ValueError – If inputs does not contain exactly one entry.