diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index f3d3ad1..c4a15d3 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -26,11 +26,14 @@ from app.flow.schemas import ( FlowsPublic, FlowStatePublic, FlowSummary, + HistoryPoint, + MessageHistory, MessageValue, NodeSource, NodeStatusPublic, NodeTypeInfo, ) +from app.flow.state import as_number from app.flow.store import FlowExists, FlowNotFound from app.models import Message @@ -275,6 +278,29 @@ def read_flow_state(name: str, controller: FlowControllerDep) -> Any: return _flow_state(controller, name) +@router.get("/{name}/history/{message}", response_model=MessageHistory) +def read_message_history( + name: str, + message: str, + controller: FlowControllerDep, +) -> Any: + """The recent values of one message, for plotting. + + ``message`` may be given bare or qualified; a message that never carried a + number comes back with an empty series. + """ + _read_flow(controller, name) + key = qualify(name, message) + return MessageHistory( + message=key, + numeric=as_number(controller.state.get(key)) is not None, + points=[ + HistoryPoint(ts=ts, value=value) + for ts, value in controller.state.history(key) + ], + ) + + # ----------------------------------------------------------------------------- # Live updates # ----------------------------------------------------------------------------- diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index 2970073..92498db 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -322,6 +322,8 @@ class Pipeline: state.update( {self._timestamp_key(name): ts for name in result}, ) + # Append-only, so it needs no lock of its own. + state.append_history(result, ts) self._increment_message_versions(result) for name, value in result.items(): self._publish( @@ -436,6 +438,7 @@ class Pipeline: with state.lock(): state.update(outputs) state.update({self._timestamp_key(name): ts for name in outputs}) + state.append_history(outputs, ts) self._increment_message_versions(outputs) for name, value in outputs.items(): self._publish( diff --git a/backend/app/flow/schemas.py b/backend/app/flow/schemas.py index 2b140d9..20abcfd 100644 --- a/backend/app/flow/schemas.py +++ b/backend/app/flow/schemas.py @@ -91,6 +91,25 @@ class MessageValue(BaseModel): ts: float | None = None +class HistoryPoint(BaseModel): + """One numeric value a message carried, and when.""" + + ts: float + value: float + + +class MessageHistory(BaseModel): + """A message's recent numeric values, oldest first. + + Only numbers are recorded, so ``numeric`` tells the panel whether an empty + series means "nothing plottable here" or "nothing has arrived yet". + """ + + message: str + numeric: bool = False + points: list[HistoryPoint] = Field(default_factory=list) + + class FlowSummary(BaseModel): name: str title: str = "" diff --git a/backend/app/flow/state.py b/backend/app/flow/state.py index f0589df..ac49cc5 100644 --- a/backend/app/flow/state.py +++ b/backend/app/flow/state.py @@ -9,6 +9,7 @@ from __future__ import annotations import json from abc import ABC, abstractmethod +from collections import deque from collections.abc import Iterator from contextlib import contextmanager from threading import RLock @@ -16,6 +17,19 @@ from typing import Any, cast import redis +# A sparkline only means something for numbers, so the history keeps the values +# it can plot and nothing else. 120 points fill a panel-wide chart while leaving +# Redis a cache rather than a time-series database. +HISTORY_LIMIT = 120 + + +def as_number(value: Any) -> float | None: + """The plottable form of a value, or None if it is not a number.""" + # bool is an int subclass; a flag is not a measurement. + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) + class StateBackend(ABC): """ @@ -149,6 +163,36 @@ class StateBackend(ABC): """ ... + # ------------------------------------------------------------------------- + # Message history + # ------------------------------------------------------------------------- + + @abstractmethod + def append_history(self, values: dict[str, Any], ts: float) -> None: + """ + Record freshly published values in each message's capped history. + + Non-numeric payloads are skipped — the history exists to be plotted. + + :param values: Message names mapped to the value just published. + :type values: dict[str, Any] + :param ts: When they were published. + :type ts: float + """ + ... + + @abstractmethod + def history(self, key: str) -> list[tuple[float, float]]: + """ + The recorded ``(timestamp, value)`` pairs of one message, oldest first. + + :param key: The message name. + :type key: str + :returns: At most ``HISTORY_LIMIT`` points; empty if nothing was recorded. + :rtype: list[tuple[float, float]] + """ + ... + def __contains__(self, key: str) -> bool: return self.exists(key) @@ -180,10 +224,11 @@ class MemoryState(StateBackend): 'value' """ - __slots__ = ("_data", "_lock") + __slots__ = ("_data", "_history", "_lock") def __init__(self) -> None: self._data: dict[str, Any] = {} + self._history: dict[str, deque[tuple[float, float]]] = {} self._lock = RLock() # Reentrant lock for nested access def get(self, key: str, default: Any = None) -> Any: @@ -201,6 +246,7 @@ class MemoryState(StateBackend): def clear(self) -> None: with self._lock: self._data.clear() + self._history.clear() def keys(self) -> list[str]: with self._lock: @@ -251,6 +297,21 @@ class MemoryState(StateBackend): self._data.update(updates) return True + def append_history(self, values: dict[str, Any], ts: float) -> None: + """Append the numeric values to their message's bounded series.""" + with self._lock: + for key, value in values.items(): + number = as_number(value) + if number is None: + continue + series = self._history.setdefault(key, deque(maxlen=HISTORY_LIMIT)) + series.append((ts, number)) + + def history(self, key: str) -> list[tuple[float, float]]: + """The recorded points of one message, oldest first.""" + with self._lock: + return list(self._history.get(key, ())) + class RedisState(StateBackend): """ @@ -485,3 +546,33 @@ class RedisState(StateBackend): except redis.WatchError: # Another client modified one of the watched keys return False + + def _history_key(self, key: str) -> str: + return self._key(f"__history__:{key}") + + def append_history(self, values: dict[str, Any], ts: float) -> None: + """Push the numeric values onto their capped list, in one round-trip.""" + pipe = self._client.pipeline() + queued = False + for key, value in values.items(): + number = as_number(value) + if number is None: + continue + history_key = self._history_key(key) + pipe.lpush(history_key, json.dumps([ts, number])) + pipe.ltrim(history_key, 0, HISTORY_LIMIT - 1) + if self._ttl: + pipe.expire(history_key, self._ttl) + queued = True + if queued: + pipe.execute() + + def history(self, key: str) -> list[tuple[float, float]]: + """The recorded points of one message, oldest first.""" + entries = cast(list[bytes], self._client.lrange(self._history_key(key), 0, -1)) + # LPUSH puts the newest first, a chart reads the other way round. + points: list[tuple[float, float]] = [] + for entry in reversed(entries): + ts, value = json.loads(entry) + points.append((float(ts), float(value))) + return points diff --git a/backend/tests/flow/test_history.py b/backend/tests/flow/test_history.py new file mode 100644 index 0000000..8650dbb --- /dev/null +++ b/backend/tests/flow/test_history.py @@ -0,0 +1,58 @@ +"""Message history: the series a node panel draws as a sparkline.""" + +from app.flow.messages import MessageSpec +from app.flow.nodes import Node +from app.flow.pipeline import Pipeline +from app.flow.state import HISTORY_LIMIT, MemoryState + + +def emitter(values) -> Node: + """A node publishing the given values, one per execution.""" + remaining = list(values) + + def process(params): + return {"out": remaining.pop(0)} + + node = Node(f=process, requires=[], provides=[MessageSpec(name="out")], name="src") + node.assign_flow("f", "src") + return node + + +def test_the_series_follows_the_order_values_were_published(): + node = emitter([1.0, 2.0, 3.0]) + pipeline = Pipeline(nodes=[node]) + + # An injecting node publishes directly, an executed one goes through the + # executor; both paths land in the same series. + node.inject() + pipeline.run() + pipeline.run() + + points = pipeline.state.history("f.out") + assert [value for _, value in points] == [1.0, 2.0, 3.0] + assert [ts for ts, _ in points] == sorted(ts for ts, _ in points) + + +def test_the_series_is_capped(): + state = MemoryState() + for i in range(HISTORY_LIMIT + 20): + state.append_history({"f.out": float(i)}, ts=float(i)) + + points = state.history("f.out") + assert len(points) == HISTORY_LIMIT + # The oldest twenty fell off the end. + assert points[0] == (20.0, 20.0) + assert points[-1] == (139.0, 139.0) + + +def test_only_numbers_are_recorded(): + state = MemoryState() + state.append_history({"f.text": "warm", "f.flag": True, "f.temp": 21}, ts=1.0) + + assert state.history("f.text") == [] + assert state.history("f.flag") == [] + assert state.history("f.temp") == [(1.0, 21.0)] + + +def test_a_message_without_history_comes_back_empty(): + assert MemoryState().history("f.never") == []