==============
HEROS Flow API
==============

``herostools.flow`` wires HEROS objects into event-driven processing flows.
A flow is a directed graph of :py:class:`~herostools.flow.Node` objects. Each node names its sources in its
``inputs``. Nodes are HEROs, started by BOSS from a JSON configuration.
See the :doc:`API reference <autoapi/index>`.

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
====

:py:class:`~herostools.flow.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
:py:meth:`~herostools.flow.Node.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.
:py:class:`~herostools.flow.Join` is the exception for ``_process``.

.. list-table::
   :header-rows: 1
   :widths: 30 70

   * - 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.
:py:func:`~herostools.flow.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
-----------

.. code-block:: python

   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"}

.. code-block:: json

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

Manual input
------------

:py:meth:`~herostools.flow.Node.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.

.. code-block:: python

   import time

   from heros import RemoteHERO

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

A call through :py:class:`~heros.RemoteHERO` does not raise on the node's errors. It logs them in the node and
returns ``None``.

Join
====

:py:class:`~herostools.flow.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:

.. code-block:: python

   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"], {}

.. code-block:: json

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

EventInputNode
==============

:py:class:`~herostools.flow.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``.

.. code-block:: json

   {"_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``.

.. literalinclude:: ../../examples/flow/testflow.json
   :language: json
   :caption: examples/flow/testflow.json

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

.. code-block:: python

   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.