"""Flow pipeline node primitives."""
import numbers
import reprlib
import threading
from collections.abc import Callable, Mapping, Sequence
from typing import Any, ClassVar
from heros import EventObserver, HERORef, event
from heros.inspect import local_only
from herostools.flow._subscriber import FlowSubscriber
from herostools.helper import log, merge_dicts
_MANUAL_INPUT = "manual"
_MIN_HEROS_BOSS = "2026.2.0"
_FORBIDDEN_SOURCE_CHARS = frozenset("/*?#$")
_FLOW_PATH_KEY = "_flow_path"
[docs]
def _is_hashable(value: Any) -> bool:
"""Return whether `hash(value)` succeeds (tuples can hold unhashable items)."""
try:
hash(value)
except TypeError:
return False
return True
[docs]
def _flow_path(metadata: dict) -> list:
"""Return the names of the nodes a message passed, or an empty list if it carries none."""
path = metadata.get(_FLOW_PATH_KEY)
return list(path) if isinstance(path, (list, tuple)) else []
[docs]
def _source_name(source: HERORef | str) -> str:
"""Return the HERO name of a source."""
return source.name if isinstance(source, HERORef) else source
[docs]
def _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.
ValueError: If `source` is an empty str or contains one of `/ * ? # $`.
"""
if isinstance(source, HERORef):
return
if not isinstance(source, str):
raise TypeError(
f"input source must be a HERORef or a HERO name, got {source!r}. "
f"A BOSS that does not pass HERORef values needs heros-boss >= {_MIN_HEROS_BOSS}"
)
if source.startswith("$"):
raise ValueError(
f"input source {source!r} was not resolved to a HERORef. A $ source needs heros-boss >= {_MIN_HEROS_BOSS}"
)
if not source or _FORBIDDEN_SOURCE_CHARS.intersection(source):
raise ValueError(f"input source {source!r} must be a non-empty HERO name without any of '/*?#$'")
[docs]
def only(inputs: dict[str, Any]) -> Any:
"""Return the single value of a one-entry inputs dict.
Args:
inputs: Inputs keyed by input name.
Returns:
The only value in `inputs`.
Raises:
ValueError: If `inputs` does not contain exactly one entry.
"""
if len(inputs) != 1:
raise ValueError(f"Expected exactly one input, got {len(inputs)}: {sorted(inputs)}")
return next(iter(inputs.values()))
[docs]
class _NodeClosedError(RuntimeError):
"""Raised when a torn-down node is asked for its subscriber."""
[docs]
class Node:
"""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]] = frozenset({"inputs"})
def __init__(
self,
inputs: Mapping[str, HERORef | str] | Sequence[HERORef | str] | None = None,
identifier_key: str = "identifier",
subscriber: FlowSubscriber | None = None,
queue_warn_threshold: int = 100,
):
"""Initialize the node.
Args:
inputs: Sources to subscribe to in `_setup`, as a mapping from input name to source, or as
a list of sources that each name their own input. A source is a `heros.HERORef` or a
HERO name string.
identifier_key: Metadata key holding the identifier of each message.
subscriber: Subscriber to use instead of creating one on first use.
queue_warn_threshold: Queue depth at which the created subscriber logs a warning. Ignored
if `subscriber` is given.
Raises:
ValueError: If `queue_warn_threshold` is not positive, a source is an empty string or
contains one of `/ * ? # $`, `inputs` names a source twice or an input `"manual"`.
TypeError: If `queue_warn_threshold` is not an integer (bool is refused), `inputs` is
not a mapping, list or tuple, or a source is neither a `HERORef` nor a string.
"""
self._input_refs = _normalize_inputs(inputs)
self._inputs = {input_name: _source_name(source) for input_name, source in self._input_refs.items()}
self._input_of_source = {source: input_name for input_name, source in self._inputs.items()}
if not isinstance(queue_warn_threshold, numbers.Integral) or isinstance(queue_warn_threshold, bool):
raise TypeError(f"queue_warn_threshold must be an int, got {queue_warn_threshold!r}")
if queue_warn_threshold <= 0:
raise ValueError(f"queue_warn_threshold must be positive, got {queue_warn_threshold!r}")
self._identifier_key = identifier_key
self._subscriber = subscriber
self._queue_warn_threshold = queue_warn_threshold
self._subscriber_lock = threading.Lock()
self._closed = False
@property
def identifier_key(self) -> str:
"""Metadata key holding the identifier of each message. Read-only."""
return self._identifier_key
@property
def inputs(self) -> dict[str, str]:
"""Copy of the inputs, mapping each input name to its source HERO name. Read-only."""
return dict(self._inputs)
[docs]
def _setup(self) -> 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.
"""
realm = getattr(self, "_realm", None)
if realm is not None:
for ref in self._input_refs.values():
if isinstance(ref, HERORef) and ref.realm != realm:
raise ValueError(
f"{self.name}: source {ref.name!r} is in realm {ref.realm!r}, the node is in realm {realm!r}"
)
for source in self._input_of_source:
if source == self.name:
raise ValueError(f"{self.name}: a node cannot be fed by itself")
self._get_subscriber().add(source)
@property
def name(self) -> str:
"""Name of the node: the HERO name once BOSS created it, otherwise the class name."""
return getattr(self, "_name", self.__class__.__name__)
[docs]
def _get_subscriber(self) -> 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.
"""
with self._subscriber_lock:
if self._closed:
raise _NodeClosedError(f"{self.name} was torn down")
if self._subscriber is None:
session = getattr(self, "_session", None)
if session is None:
raise RuntimeError(f"{self.name} must run as a HERO or be given a subscriber")
self._subscriber = FlowSubscriber(
session,
self._realm,
self._on_flow_data,
queue_warn_threshold=self._queue_warn_threshold,
name=self.name,
)
return self._subscriber
[docs]
def _open_subscriber(self) -> FlowSubscriber | None:
"""Return the subscriber, or `None` (logged at debug level) once the node is torn down."""
try:
return self._get_subscriber()
except _NodeClosedError:
log.debug(f"{self.name}: Node is torn down, ignoring work")
return None
[docs]
def _submit(self, func: Callable[[], None]) -> None:
"""Run a callable on the worker, in order with the messages. Does nothing after teardown."""
subscriber = self._open_subscriber()
if subscriber is not None:
subscriber.submit(func)
[docs]
def _on_flow_data(self, source: str, payload: Any) -> None:
"""Handle a decoded `flow_data` payload on the worker.
Args:
source: Name of the HERO the message was received from.
payload: Tuple of data, metadata and the emitting node's name.
"""
if not (isinstance(payload, (list, tuple)) and len(payload) == 3 and isinstance(payload[1], dict)):
log.error(f"{self.name}: malformed flow_data from {source}: {reprlib.repr(payload)}")
return
data, metadata, _ = payload
self._handle_flow_data(data, metadata, source)
[docs]
def _teardown(self) -> None:
"""Stop the subscriber. Called by BOSS on shutdown."""
with self._subscriber_lock:
self._closed = True
subscriber = self._subscriber
if subscriber is not None:
subscriber.stop()
[docs]
def feed(self, 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.
Args:
data: Input data.
metadata: Input metadata. Must contain `identifier_key`.
Raises:
ValueError: If the identifier is missing or not hashable.
"""
error = self._identifier_error(metadata)
if error is not None:
raise ValueError(f"{self.name}: {error}")
metadata = dict(metadata)
self._submit(lambda: self._feed(data, metadata))
[docs]
def _feed(self, data: Any, metadata: dict) -> None:
"""Validate, process and emit a fed message. Runs on the worker."""
if self._is_valid(_MANUAL_INPUT, data, metadata, _MANUAL_INPUT):
self._fire(({_MANUAL_INPUT: data}, metadata))
[docs]
def _identifier_error(self, metadata: dict) -> str | None:
"""Return why the metadata has no usable identifier, or `None`."""
identifier = metadata.get(self._identifier_key)
if identifier is None:
return f"metadata has no {self._identifier_key!r}"
if not _is_hashable(identifier):
return f"identifier {reprlib.repr(identifier)} is not hashable"
return None
[docs]
def _is_valid(self, input_name: str, data: Any, metadata: dict, source: str) -> bool:
"""Run `validate` for one message, logging failures."""
try:
valid = self.validate({input_name: data}, metadata)
except Exception: # noqa: BLE001 - user-supplied validate() must not crash the pipeline
log.exception(f"Exception during {self.name}.validate")
return False
if not valid:
log.debug(f"{self.name}: Validation for {reprlib.repr(data)}, {metadata} from {source} failed")
return valid
[docs]
@local_only
def validate(self, inputs: dict[str, Any], metadata: dict) -> bool:
"""Validate incoming flow data.
Local only, not exposed over RPC. Subclasses override `_validate`.
Args:
inputs: Input data keyed by input name.
metadata: Input metadata.
Returns:
Whether the data is valid.
"""
log.debug(f"{self.name}: Validating data from {sorted(inputs)}")
return self._validate(inputs, metadata)
[docs]
def _validate(self, 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.
Args:
inputs: One-entry dict with the message's data keyed by its input name.
metadata: Input metadata.
Returns:
Whether the message is valid.
"""
return True
[docs]
@local_only
def process(self, inputs: dict[str, Any], metadata: dict) -> tuple[Any, dict] | None:
"""Process collected flow data.
Local only, not exposed over RPC. Subclasses override `_process`.
Args:
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.
"""
log.debug(f"{self.name}: Processing data from {sorted(inputs)}")
return self._process(inputs, metadata)
[docs]
def _process(self, 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.
Args:
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.
"""
return only(inputs), {}
[docs]
def _handle_flow_data(self, data: Any, metadata: dict, source: str) -> None:
"""Map a message to its input and handle it. Runs on the worker.
Args:
data: Input data.
metadata: Input metadata.
source: Name of the upstream node.
"""
input_name = self._input_of_source.get(source)
if input_name is None:
log.debug(f"{self.name}: Dropping data from unknown source {source}")
return
self._handle_input(input_name, data, metadata, source)
[docs]
def _collect(self, 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.
Args:
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.
"""
return {input_name: data}, dict(metadata)
[docs]
def _fire(self, ready: tuple[dict, dict]) -> None:
"""Process and emit one message, with this node appended to its flow path."""
inputs, metadata = ready
try:
result = self.process(inputs, metadata)
if result is None:
log.debug(f"{self.name}: Nothing to emit")
return
out_data, out_metadata = result
except Exception: # noqa: BLE001 - user-supplied process() must not crash the pipeline
log.exception(f"Exception during {self.name}.process:")
return
emitted_metadata = merge_dicts(metadata, out_metadata)
emitted_metadata[_FLOW_PATH_KEY] = [*_flow_path(metadata), self.name]
self.flow_data(out_data, emitted_metadata)
@event
def flow_data(self, data: Any, metadata: dict) -> tuple[Any, dict, str]:
"""Emit processed flow data as a HEROS event.
Args:
data: Output data.
metadata: Output metadata.
Returns:
Tuple of data, metadata and this node's name.
"""
log.debug(f"{self.name}: Emitting data")
return data, metadata, self.name