herostools.flow =============== .. py:module:: herostools.flow Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/herostools/flow/_subscriber/index /autoapi/herostools/flow/dummy/index /autoapi/herostools/flow/join/index /autoapi/herostools/flow/node/index Classes ------- .. autoapisummary:: herostools.flow.Join herostools.flow.EventInputNode herostools.flow.Node Functions --------- .. autoapisummary:: herostools.flow.only Package Contents ---------------- .. py:class:: Join(inputs: collections.abc.Mapping[str, heros.HERORef | str] | collections.abc.Sequence[heros.HERORef | str], *, max_open_rounds: int = 10, **kwargs) Bases: :py:obj:`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. .. py:attribute:: _hero_implements :type: ClassVar[list[str]] :value: ['herostools.flow.Node', 'herostools.flow.Join'] .. py:attribute:: _max_open_rounds :value: 10 .. py:attribute:: _rounds :type: dict[Any, dict[str, tuple[Any, dict]]] .. py:attribute:: _finished :type: collections.deque .. py:attribute:: _warned :value: False .. py:method:: _process(inputs: dict[str, Any], metadata: dict) -> tuple[Any, dict] | None Emit the dict of all inputs. Override in subclasses to combine them. .. py:method:: _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. .. py:method:: _drop_oldest_round() -> None Drop the oldest incomplete round, with a warning once per crossing of the limit. .. 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. .. 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: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.