HEROS Flow API¶

herostools.flow wires HEROS objects into event-driven processing flows. A flow is a directed graph of Node objects. Each node names its sources in its inputs. Nodes are HEROs, started by BOSS from a JSON configuration. See the API reference.

Model¶

event source -> EventInputNode -> Node -> Node -> consumer

A node receives data on named inputs, processes it and emits the result as its flow_data event. Data moves by pub/sub. Each node subscribes to @object/<realm>/<source>/flow_data for every source in its inputs and puts the messages on one queue. A single worker thread per node takes them off the queue, so a slow node does not block the nodes upstream of it. The flow_data payload is the tuple (data, metadata, source), where source is the name of the emitting HERO. A payload that is not a list or tuple of three items with a dict as second item is logged as an error and not processed.

Node¶

Node is the processing unit. The default implementation passes its single input through unchanged.

Inputs¶

inputs is a mapping from input name to source, or a list of sources that each name their own input. In a BOSS config a source is written "$name". BOSS passes it as a heros.HERORef without looking the HERO up. In Python a source is a heros.HERORef or a HERO name string. The HERO name of a HERORef names the input in a list. The name is the key in the inputs dict that _validate and _process receive. The inputs are fixed at construction. One source can feed only one input, and the name "manual" is reserved for feed(). A violation raises ValueError at construction. Sources are compared by HERO name. A string that is empty or contains one of * ? # $ / raises ValueError. A literal leading $ in a string argument of a BOSS config is written $$. Wildcards belong in object_selector, not in inputs. A source that is neither a HERORef nor a string, and an inputs value that is not a mapping, list or tuple, raise TypeError. BOSS calls _setup after it created the node, and _setup subscribes to every source. A node that names itself as a source raises ValueError there. A HERORef from another realm than the node’s raises ValueError there before anything is subscribed. A node without a HERO realm, such as one that was given a subscriber and is no HERO, skips this check. A message from a source that is not an input is dropped and logged at DEBUG level.

Identifier¶

Every message must carry metadata[identifier_key]. The default key is "identifier". The value must be hashable and unique in space and time. Each new message that enters a flow gets an identifier that no other source uses and that is never reused, also not after a restart. Messages derived from it keep that identifier. A Join ignores messages with an identifier it has recently finished. A message without a usable identifier is rejected with an error log.

Flow path¶

Every emitted message lists the names of the nodes it passed in metadata["_flow_path"], in order. A node appends its own name when it emits. The key is reserved: a value returned by _process under it is replaced. A message that already passed the receiving node is dropped with a warning, so a loop in the flow is cut after one pass. A callable run through _submit that emits with flow_data does not get this node appended.

Hooks¶

Subclasses override _validate and _process. The public validate and process wrappers are local only and are not exposed over RPC. A node calls both once per message, and inputs holds one entry, keyed by the name of the input that delivered it. Join is the exception for _process.

Hook	Behaviour
`_validate(inputs, metadata) -> bool`	`False` drops the message and is logged at DEBUG level. An exception is logged with its traceback and drops the message.
`_process(inputs, metadata) -> (data, metadata) \| None`	Returns the output. The values can be of any type, nothing is converted. `None` emits nothing for this call. An exception is logged and nothing is emitted.

In all cases the node keeps running. only() returns the value of a one-entry inputs dict and raises ValueError otherwise. The metadata of the emitted event is the metadata of the message merged with the metadata returned by _process. On a key collision the node’s output wins. A node with its 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 on a timeout. It does nothing after teardown. The callable emits by calling self.flow_data(data, metadata) with the complete metadata. It bypasses process and the metadata merge, so the callable merges metadata itself. An exception from the callable is logged by the worker, which keeps running.

Custom node¶

from typing import Any

from herostools.flow import Node, only


class Scale(Node):
    """Emit the input multiplied by 2."""

    def _validate(self, inputs: dict[str, Any], metadata: dict) -> bool:
        return isinstance(only(inputs), (int, float))

    def _process(self, inputs: dict[str, Any], metadata: dict) -> tuple[Any, dict]:
        return only(inputs) * 2, {"unit": "V"}

{"_id": "scaled", "classname": "mypackage.nodes.Scale", "arguments": {"inputs": {"voltage": "$sensor_a"}}}

Manual input¶

feed() injects one message into a node, for example to test a flow. The message goes through the input "manual" and needs no source. feed returns at once. The message is validated, processed and emitted on the worker, in order with the subscribed messages. The metadata must contain a hashable identifier, otherwise feed raises ValueError on the caller.

import time

from heros import RemoteHERO

RemoteHERO("flow_test_input").feed(10, {"identifier": time.time()})

A call through RemoteHERO does not raise on the node’s errors. It logs them in the node and returns None.

Join¶

Join collects one message per input and then calls _process once. Its inputs must not be empty. _process receives the data of every input keyed by input name, in config order. The default _process emits that dict.

Messages are grouped into rounds by their identifier. A round completes when every 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 (default 10). It must be a positive integer. A message with a new identifier at the limit drops the oldest incomplete round and logs a warning with 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 it.
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 _validate does not count.
feed is processed alone as input "manual" and never joins a round.

A subclass overrides _process to combine the inputs:

from typing import Any

from herostools.flow import Join

class Subtract(Join):
    """Emit left minus right."""

    _hero_implements = [*Join._hero_implements, "mypackage.Subtract"]

    def _process(self, inputs: dict[str, Any], metadata: dict) -> tuple[Any, dict]:
        return inputs["left"] - inputs["right"], {}

{"_id": "diff", "classname": "mypackage.nodes.Subtract", "arguments": {"inputs": {"left": "$sensor_a", "right": "$sensor_b"}, "max_open_rounds": 20}}

EventInputNode¶

EventInputNode is the entry point for device events. It turns any HERO event into flow data. It takes an object_selector and an event_name like a HEROS event observer. The name of the emitting HERO is the input name of its events. No inputs entry is needed. Events emitted by the node itself are ignored, so a selector such as * that matches the node does not loop. Every event is validated, processed and emitted on its own.

A payload that is a list or tuple of at least two items with a dict as second item is split into data (first item) and a copy of that dict as metadata. Any other payload becomes the data with empty metadata. The emitting HERO must put the identifier into the metadata. An event without a usable identifier is rejected with an error log, as for any node. The handle_event method that receives the events is local only.

Restart behaviour¶

Sources can restart at any time. The subscription of a node picks up the new publisher, so sources need nothing. A restarted node runs _setup again and subscribes again from its config. On shutdown BOSS calls _teardown, which stops the subscriber. Work submitted afterwards is ignored. Queued messages are discarded and only the item that is running finishes. A warning is logged if it takes longer than 5 s.

Queue¶

The queue of a node is unbounded. The node logs a WARNING each time the queue depth crosses queue_warn_threshold. It warns again only after the depth has dropped below the threshold and crossed it again. The default is 100. A value that is not an integer, or a bool, raises TypeError at construction. An integer below 1 raises ValueError.

{"_id": "slow_node", "classname": "mypackage.nodes.Scale", "arguments": {"inputs": ["$sensor_a"], "queue_warn_threshold": 20}}

A message that cannot be decoded is dropped with an error log, and later messages still arrive. A payload that references a HERO that no longer exists stalls the worker for about 2 s while it is decoded. Messages behind it wait.

Monitoring¶

Nodes announce themselves through _hero_implements: "herostools.flow.Node", and "herostools.flow.Join" for a join. A view finds them by this implements entry. A subclass extends the list and keeps the base entry. identifier_key and inputs (input_name -> source HERO name, a copy) are readable remotely and read-only. A remote write to one of them hangs the caller for about 10 s and logs an error on the node. This is a limitation of HEROS.

Configuration example¶

examples/flow/testflow.json defines the chain flow_test_input -> flow_test_add_3 -> flow_test_square with one row per node. flow_test_join collects the outputs of flow_test_add_3 and flow_test_square per identifier and feeds flow_test_logger.

examples/flow/testflow.json¶

{
  "rows": [
    {
      "_id": "flow_test_input",
      "tags": ["flow:test:nodes"],
      "classname": "herostools.flow.Node",
      "arguments": {}
    },
    {
      "_id": "flow_test_square",
      "tags": ["flow:test:nodes"],
      "classname": "herostools.flow.dummy.ToPower",
      "arguments": {
        "exponent": 2,
        "inputs": ["$flow_test_add_3"]
      }
    },
    {
      "_id": "flow_test_add_3",
      "tags": ["flow:test:nodes"],
      "classname": "herostools.flow.dummy.AddConstant",
      "arguments": {
        "constant": 3,
        "inputs": ["$flow_test_input"]
      }
    },
    {
      "_id": "flow_test_join",
      "tags": ["flow:test:nodes"],
      "classname": "herostools.flow.Join",
      "arguments": {
        "inputs": {"added": "$flow_test_add_3", "squared": "$flow_test_square"}
      }
    },
    {
      "_id": "flow_test_logger",
      "tags": ["flow:test:nodes"],
      "classname": "herostools.flow.dummy.Logger",
      "arguments": {
        "inputs": ["$flow_test_join"]
      }
    }
  ]
}

Start it with BOSS and send a value into the first node:

import time

from heros import RemoteHERO

RemoteHERO("flow_test_input").feed(10, {"identifier": time.time()})

The logger prints {'added': 13, 'squared': 169} from flow_test_join, with the timestamp as identifier and the flow path ['flow_test_input', 'flow_test_add_3', 'flow_test_square', 'flow_test_join'] in the metadata.