herostools.flow.node ==================== .. py:module:: herostools.flow.node .. autoapi-nested-parse:: Flow pipeline node primitives. Attributes ---------- .. autoapisummary:: herostools.flow.node._MANUAL_INPUT herostools.flow.node._MIN_HEROS_BOSS herostools.flow.node._FORBIDDEN_SOURCE_CHARS herostools.flow.node._FLOW_PATH_KEY Exceptions ---------- .. autoapisummary:: herostools.flow.node._NodeClosedError Classes ------- .. autoapisummary:: herostools.flow.node.Node herostools.flow.node.EventInputNode Functions --------- .. autoapisummary:: herostools.flow.node._is_hashable herostools.flow.node._flow_path herostools.flow.node._source_name herostools.flow.node._check_source herostools.flow.node._normalize_inputs herostools.flow.node.only Module Contents --------------- .. py:data:: _MANUAL_INPUT :value: 'manual' .. py:data:: _MIN_HEROS_BOSS :value: '2026.2.0' .. py:data:: _FORBIDDEN_SOURCE_CHARS .. py:data:: _FLOW_PATH_KEY :value: '_flow_path' .. py:function:: _is_hashable(value: Any) -> bool Return whether `hash(value)` succeeds (tuples can hold unhashable items). .. py:function:: _flow_path(metadata: dict) -> list Return the names of the nodes a message passed, or an empty list if it carries none. .. py:function:: _source_name(source: heros.HERORef | str) -> str Return the HERO name of a source. .. py:function:: _check_source(source: Any) -> None Raise if a source is neither a `HERORef` nor a plain HERO name string. :raises TypeError: If `source` is not a `HERORef` or a str. :raises ValueError: If `source` is an empty str or contains one of `/ * ? # $`. .. py:function:: _normalize_inputs(inputs: collections.abc.Mapping[str, heros.HERORef | str] | collections.abc.Sequence[heros.HERORef | str] | None) -> dict[str, heros.HERORef | str] Return the configured inputs as an ordered dict from input name to source. :param inputs: Mapping from input name to source, or a list of sources that name their own inputs. A source is a `HERORef` or a HERO name. :returns: The inputs keyed by input name, with the sources as given. Empty if `inputs` is `None`. :raises TypeError: If `inputs` is not a mapping, list or tuple, or a source is neither a `HERORef` nor a str. :raises ValueError: If a source is invalid, two sources have the same name or an input is called `"manual"`. .. py:function:: only(inputs: dict[str, Any]) -> Any Return the single value of a one-entry inputs dict. :param inputs: Inputs keyed by input name. :returns: The only value in `inputs`. :raises ValueError: If `inputs` does not contain exactly one entry. .. py:exception:: _NodeClosedError Bases: :py:obj:`RuntimeError` Raised when a torn-down node is asked for its subscriber. .. py:class:: 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. .. py:attribute:: _hero_implements :type: ClassVar[list[str]] :value: ['herostools.flow.Node'] .. py:attribute:: _boss_heroref_arguments :type: ClassVar[frozenset[str]] .. py:attribute:: _input_refs .. py:attribute:: _inputs .. py:attribute:: _input_of_source .. py:attribute:: _identifier_key :value: 'identifier' .. py:attribute:: _subscriber :value: None .. py:attribute:: _queue_warn_threshold :value: 100 .. py:attribute:: _subscriber_lock .. py:attribute:: _closed :value: False .. py:property:: identifier_key :type: str Metadata key holding the identifier of each message. Read-only. .. py:property:: inputs :type: dict[str, str] Copy of the inputs, mapping each input name to its source HERO name. Read-only. .. py:method:: _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. .. py:property:: name :type: str the HERO name once BOSS created it, otherwise the class name. :type: Name of the node .. py:method:: _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. .. py:method:: _open_subscriber() -> herostools.flow._subscriber.FlowSubscriber | None Return the subscriber, or `None` (logged at debug level) once the node is torn down. .. py:method:: _submit(func: collections.abc.Callable[[], None]) -> None Run a callable on the worker, in order with the messages. Does nothing after teardown. .. py:method:: _on_flow_data(source: str, payload: Any) -> None Handle a decoded `flow_data` payload on the worker. :param source: Name of the HERO the message was received from. :param payload: Tuple of data, metadata and the emitting node's name. .. py:method:: _teardown() -> None Stop the subscriber. Called by BOSS on shutdown. .. py:method:: 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. :param data: Input data. :param metadata: Input metadata. Must contain `identifier_key`. :raises ValueError: If the identifier is missing or not hashable. .. py:method:: _feed(data: Any, metadata: dict) -> None Validate, process and emit a fed message. Runs on the worker. .. py:method:: _identifier_error(metadata: dict) -> str | None Return why the metadata has no usable identifier, or `None`. .. py:method:: _is_valid(input_name: str, data: Any, metadata: dict, source: str) -> bool Run `validate` for one message, logging failures. .. py:method:: validate(inputs: dict[str, Any], metadata: dict) -> bool Validate incoming flow data. Local only, not exposed over RPC. Subclasses override `_validate`. :param inputs: Input data keyed by input name. :param metadata: Input metadata. :returns: Whether the data is valid. .. py:method:: _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. :param inputs: One-entry dict with the message's data keyed by its input name. :param metadata: Input metadata. :returns: Whether the message is valid. .. py:method:: process(inputs: dict[str, Any], metadata: dict) -> tuple[Any, dict] | None Process collected flow data. Local only, not exposed over RPC. Subclasses override `_process`. :param inputs: Input data keyed by input name. One entry, the input that delivered, except in a `Join`, which passes every input. :param metadata: Input metadata. :returns: Tuple of processed data and additional metadata, or None to emit nothing for this call. .. py:method:: _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. :param inputs: Input data keyed by input name. One entry, the input that delivered, except in a `Join`, which passes every input. :param 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. .. py:method:: _handle_flow_data(data: Any, metadata: dict, source: str) -> None Map a message to its input and handle it. Runs on the worker. :param data: Input data. :param metadata: Input metadata. :param source: Name of the upstream node. .. py:method:: _handle_input(input_name: str, data: Any, metadata: dict, source: str) -> None Validate, process and emit one message. Runs on the worker. :param input_name: Name of the input that delivered the message. :param data: Input data. :param metadata: Input metadata. :param source: Name of the upstream node. .. py:method:: _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. :param input_name: Name of the input that delivered the message. :param data: Input data. :param 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. .. py:method:: _fire(ready: tuple[dict, dict]) -> None Process and emit one message, with this node appended to its flow path. .. py:method:: flow_data(data: Any, metadata: dict) -> tuple[Any, dict, str] Emit processed flow data as a HEROS event. :param data: Output data. :param metadata: Output metadata. :returns: Tuple of data, metadata and this node's name. .. py:class:: EventInputNode(object_selector: str, event_name: str, *args, identifier_key: str = 'identifier', queue_warn_threshold: int = 100, **kwargs) Bases: :py:obj:`Node`, :py:obj:`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. .. py:method:: _teardown() -> None Undeclare the event subscription, then tear the node down. Called by BOSS on shutdown. .. py:method:: 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. :param source: Name of the emitting HERO. :param payload: Event payload, either a `(data, metadata)` pair or plain data. .. py:method:: _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.