From 413501c6ced6ebeb3c41d913db987daf0297e483 Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 17 Aug 2026 15:06:45 +0200 Subject: [PATCH] flow: structured dtypes, and the widgets that read them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A series, record or list message declares its shape instead of riding DType.JSON, so a widget binds a shape rather than some JSON and a wrong binding is refused before anything runs. A list declares its item type, which is what keeps list[float] expressible for a pipeline. On top of that: an agenda over a list, a notification over a record, and a dashboard alert channel that publishes engine faults as one — so a panel can show what went wrong without a flow wiring it by hand. Also: only None means a node published nothing, a falsy value of the wrong shape is now the named error it always should have been; and the gauge's readout says its size is viewBox geometry rather than type scale. Co-Authored-By: Claude Opus 5 (1M context) --- backend/app/flow/alerts.py | 25 ++- backend/app/flow/dashboards.py | 46 +++++- backend/app/flow/messages.py | 117 +++++++++++-- backend/app/flow/nodes/base.py | 12 +- backend/app/flow/nodes/influx.py | 87 +++++++++- backend/app/main.py | 6 + backend/tests/flow/test_alerts.py | 33 ++++ backend/tests/flow/test_dashboards.py | 61 +++++++ backend/tests/flow/test_messages.py | 61 +++++++ backend/tests/flow/test_node_types.py | 57 +++++++ frontend/src/client/schemas.gen.ts | 34 +++- frontend/src/client/types.gen.ts | 27 ++- frontend/src/components/Common/UplotChart.tsx | 27 ++- frontend/src/components/Dashboard/widgets.tsx | 156 +++++++++++++++++- frontend/src/components/Flow/NodePanel.tsx | 40 ++++- frontend/src/routes/_layout/alerts.tsx | 5 +- 16 files changed, 751 insertions(+), 43 deletions(-) diff --git a/backend/app/flow/alerts.py b/backend/app/flow/alerts.py index 3390168..137f228 100644 --- a/backend/app/flow/alerts.py +++ b/backend/app/flow/alerts.py @@ -16,6 +16,7 @@ import asyncio import logging import time from collections import deque +from collections.abc import Callable from typing import Any, Literal import httpx @@ -51,9 +52,10 @@ class Channel(BaseModel): """Somewhere to send an alert.""" name: str - kind: Literal["ntfy", "smtp", "webhook"] + kind: Literal["ntfy", "smtp", "webhook", "dashboard"] enabled: bool = True # ntfy: server + topic + optional token. smtp: to. webhook: url. + # dashboard: message — the record a notification widget reads. config: dict[str, Any] = Field(default_factory=dict) @@ -160,6 +162,10 @@ class AlertManager: self._events = events self.config = config or AlertsConfig() self._now = now + #: How a "dashboard" channel puts its alert into the graph. Bound after + #: construction, because the controller that publishes does not exist + #: yet when the manager is built. + self.publish: Callable[[str, Any], None] | None = None self._last_sent: dict[str, float] = {} self._suppressed: dict[str, int] = {} self._health_flips: dict[str, deque[float]] = {} @@ -294,6 +300,8 @@ class AlertManager: await self._send_ntfy(config, alert) elif channel.kind == "webhook": await self._send_webhook(config, alert) + elif channel.kind == "dashboard": + await self._send_dashboard(config, alert) else: await self._send_email(config, alert) except Exception as exc: @@ -319,6 +327,21 @@ class AlertManager: ) response.raise_for_status() + async def _send_dashboard(self, config: dict[str, Any], alert: Alert) -> None: + """Put the alert into the graph, where a notification widget shows it. + + The message has to be one a flow declares, like any other a dashboard + writes to — so a panel that shows engine faults says so in a flow + rather than appearing from nowhere. + """ + message = str(config.get("message") or "") + if not message: + raise ValueError("a dashboard channel needs a message name") + if self.publish is None: + raise RuntimeError("nothing is wired up to publish this") + # Blocking: it reads the store to find the declared message. + await asyncio.to_thread(self.publish, message, alert.model_dump()) + async def _send_webhook(self, config: dict[str, Any], alert: Alert) -> None: url = config.get("url") if not url: diff --git a/backend/app/flow/dashboards.py b/backend/app/flow/dashboards.py index fc6eb84..3d68a61 100644 --- a/backend/app/flow/dashboards.py +++ b/backend/app/flow/dashboards.py @@ -40,6 +40,8 @@ WidgetType = Literal[ "gauge", "chart", "markdown", + "agenda", + "notification", # Input "button", "switch", @@ -57,9 +59,13 @@ INPUT_WIDGETS = {"button", "switch", "slider", "input", "dropdown"} #: (``frontend/src/components/Dashboard/widgets.tsx``). WIDGET_DTYPES: dict[str, set[str]] = { "gauge": {"float", "int"}, + # A chart reading the engine's ring. One that queries binds a `series` + # answer and a `record` request instead, checked separately below. "chart": {"float", "int"}, "slider": {"float", "int"}, "switch": {"bool"}, + "agenda": {"list"}, + "notification": {"record"}, } @@ -78,6 +84,13 @@ class WidgetDef(BaseModel): ``config`` is per type — a chart names its series, a button names the message it publishes — and is validated against the type below rather than by a schema per class, because the whole set is small and closed. + + A chart comes in two kinds. The default reads what the engine kept for a + message. One with ``source: "query"`` asks instead, and its config is + ``{source, request, request_dtype: "record", message, dtype: "series", + refresh_s, range_s}``: it publishes ``{range_s, interval_s}`` to + ``request`` exactly as a slider publishes a value, and draws the ``series`` + a flow answers with on ``message``. """ id: str @@ -93,9 +106,17 @@ class WidgetDef(BaseModel): def _check_id(cls, value: str) -> str: return _validate_name(value) + @property + def _query_chart(self) -> bool: + """A chart that asks a flow for its series instead of reading the ring.""" + return self.type == "chart" and self.config.get("source") == "query" + @property def messages(self) -> list[str]: """Every message name this widget reads.""" + if self._query_chart: + name = self.config.get("message") + return [str(name)] if name else [] if self.type == "chart": return [ str(series.get("message")) @@ -107,13 +128,23 @@ class WidgetDef(BaseModel): @property def target(self) -> str: - """The message this widget publishes, if it is an input.""" + """The message this widget publishes, if it is an input. + + A querying chart is one too: its request is a value it puts into the + graph, so the canvas draws it as an endpoint like any other control. + """ + if self._query_chart: + return str(self.config.get("request") or "") return str(self.config.get("target") or "") @property def history_points(self) -> int: - """How much past this widget needs kept for it.""" - if self.type != "chart": + """How much past this widget needs kept for it. + + Nothing, for a chart that queries: the answer carries its own past, so + asking the engine to keep a ring as well would store it twice. + """ + if self.type != "chart" or self._query_chart: return 0 points = int((self.config.get("history") or {}).get("points") or 0) return min(points, HISTORY_CAP) @@ -135,6 +166,15 @@ class WidgetDef(BaseModel): @model_validator(mode="after") def _check_binding(self) -> WidgetDef: """Refuse a widget wired to a message it cannot carry.""" + if self._query_chart: + for key, want in (("dtype", "series"), ("request_dtype", "record")): + bound = str(self.config.get(key) or "") + if bound and bound != want: + raise ValueError( + f"a querying chart's {key} must be '{want}', not '{bound}'" + ) + return self + allowed = WIDGET_DTYPES.get(self.type) if not allowed: return self diff --git a/backend/app/flow/messages.py b/backend/app/flow/messages.py index f832e8d..7deb0c3 100644 --- a/backend/app/flow/messages.py +++ b/backend/app/flow/messages.py @@ -20,6 +20,11 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator class DType(str, Enum): """Serializable payload types. + The scalars carry what a single reading can say. The three structured ones + are declared shapes rather than "some JSON": a widget or a downstream node + knows what it is getting before anything runs, which is what lets the + dashboard picker offer a message and refuse a wrong binding. + Binary payloads (tensors, images) will arrive later as explicitly declared codec fields; until then everything on the wire is JSON. """ @@ -29,10 +34,78 @@ class DType(str, Enum): STR = "str" BOOL = "bool" JSON = "json" + #: ``{"lines": [{"label": str, "points": [[ts, value], ...]}], ...}``. + #: Keys beside ``lines`` are carried through untouched — a chart's query + #: puts the range and interval it asked for there and reads them back. + SERIES = "series" + #: Flat named scalars: ``{"title": "Boiler", "severity": "error"}``. + RECORD = "record" + #: Ordered items of one declared shape; see :attr:`MessageSpec.item`. + LIST = "list" _JSON_TYPES = (dict, list, str, int, float, bool, type(None)) +#: What a record may hold. Nesting is deliberately out: a record that can +#: contain a record is a schema language, and the shape stops being readable +#: from the declaration alone. +_SCALARS = (str, int, float, bool) + +#: Item types a list may declare. Recursion is refused for the same reason. +_ITEM_TYPES = frozenset( + {DType.FLOAT, DType.INT, DType.STR, DType.BOOL, DType.JSON, DType.RECORD} +) + + +def _is_number(value: Any) -> bool: + """A measurement. bool is an int subclass; a flag is not a number here.""" + return isinstance(value, (int, float)) and not isinstance(value, bool) + + +def _is_record(value: Any) -> bool: + return isinstance(value, dict) and all( + isinstance(key, str) and (item is None or isinstance(item, _SCALARS)) + for key, item in value.items() + ) + + +def _is_series(value: Any) -> bool: + """Labelled lines of ``(timestamp, value)`` pairs. + + Checked all the way down. That is O(n) in the number of points, but so is + the JSON encoding every message already pays for. + """ + if not isinstance(value, dict) or not isinstance(value.get("lines"), list): + return False + return all( + isinstance(line, dict) + and isinstance(line.get("label"), str) + and isinstance(line.get("points"), list) + and all( + isinstance(point, (list, tuple)) + and len(point) == 2 + and _is_number(point[0]) + and _is_number(point[1]) + for point in line["points"] + ) + for line in value["lines"] + ) + + +def _matches(dtype: DType, value: Any) -> bool: + """Whether one value satisfies a scalar or record type.""" + if dtype is DType.BOOL: + return isinstance(value, bool) + if dtype is DType.INT: + return isinstance(value, int) and not isinstance(value, bool) + if dtype is DType.FLOAT: + return _is_number(value) + if dtype is DType.STR: + return isinstance(value, str) + if dtype is DType.RECORD: + return _is_record(value) + return isinstance(value, _JSON_TYPES) + class MessageSpec(BaseModel): """A single port of a node, and the message it is bound to. @@ -42,6 +115,10 @@ class MessageSpec(BaseModel): :param port: The identifier the node function sees. Defaults to the last segment of ``name``, so unqualified flows read naturally. :param dtype: Payload type, validated on every message that passes through. + :param item: The type of each item of a ``list`` port, ignored otherwise. + Unset means ``record``, which is what the agenda and forecast widgets + read; ``float`` is the numeric list a pipeline passes around. A list of + lists, or of series, is refused — one declared level is the point. :param interval: Deliver at most every this many seconds; 0 is every time. On an output it holds back publishing, on an input it holds back waking the node. The value is never lost — state keeps the latest — only the @@ -57,6 +134,7 @@ class MessageSpec(BaseModel): name: str = "" port: str = "" dtype: DType = DType.FLOAT + item: DType | None = None interval: float = Field(default=0, ge=0) trigger: bool = True @@ -64,25 +142,36 @@ class MessageSpec(BaseModel): def _default_port(self) -> MessageSpec: if not self.port and self.name: object.__setattr__(self, "port", self.name.rsplit(".", 1)[-1]) + if self.item is not None and self.item not in _ITEM_TYPES: + raise ValueError(f"a list cannot hold '{self.item.value}' items") return self + @property + def item_dtype(self) -> DType: + """What each item of a ``list`` port is, declared or defaulted.""" + return self.item or DType.RECORD + def check(self, value: Any) -> None: """Raise if ``value`` does not match this port's declared type.""" - if self.dtype is DType.BOOL: - ok = isinstance(value, bool) - elif self.dtype is DType.INT: - # bool is an int subclass; a flag is not a number here. - ok = isinstance(value, int) and not isinstance(value, bool) - elif self.dtype is DType.FLOAT: - ok = isinstance(value, (int, float)) and not isinstance(value, bool) - elif self.dtype is DType.STR: - ok = isinstance(value, str) + where = self.name or self.port + if self.dtype is DType.SERIES: + ok = _is_series(value) + elif self.dtype is DType.LIST: + if not isinstance(value, list): + raise TypeError(f"{where}: expected list, got {type(value).__name__}") + item = self.item_dtype + for index, element in enumerate(value): + if not _matches(item, element): + raise TypeError( + f"{where}: expected list[{item.value}], got " + f"{type(element).__name__} at index {index}" + ) + return else: - ok = isinstance(value, _JSON_TYPES) + ok = _matches(self.dtype, value) if not ok: raise TypeError( - f"{self.name or self.port}: expected {self.dtype.value}, " - f"got {type(value).__name__}" + f"{where}: expected {self.dtype.value}, got {type(value).__name__}" ) def coerce(self, value: Any) -> Any: @@ -101,6 +190,10 @@ class MessageSpec(BaseModel): return str(value).lower() in ("true", "1", "yes", "on") if self.dtype is DType.STR: return value if isinstance(value, str) else json.dumps(value) + if self.dtype in (DType.SERIES, DType.RECORD, DType.LIST): + # A structured payload arriving as text is the same hint a numeric + # one is; the shape itself is still checked afterwards. + return json.loads(value) if isinstance(value, str) else value return value def __repr__(self) -> str: diff --git a/backend/app/flow/nodes/base.py b/backend/app/flow/nodes/base.py index efc236c..cf5f343 100644 --- a/backend/app/flow/nodes/base.py +++ b/backend/app/flow/nodes/base.py @@ -271,13 +271,19 @@ class Node: return kwargs def _to_messages(self, retval: Any) -> dict[str, Any] | None: - """Map a function's port-keyed return value onto message names.""" - if not retval: + """Map a function's port-keyed return value onto message names. + + Only ``None`` means "nothing to publish". A falsy value of the wrong + shape — ``0``, ``""``, an empty list — is a mistake worth naming rather + than silence someone has to debug from an empty canvas. + """ + if retval is None: return None if not isinstance(retval, dict): raise NodeOutputError( f"'{self.local_id}' returned {type(retval).__name__}. Outputs are " - "keyed by port, so return a dict like {'out': value}, or None." + "keyed by port, so return a dict like {'out': value}, or None to " + "publish nothing." ) by_port = {s.port: s for s in self.output_ports if s.name} outputs = {} diff --git a/backend/app/flow/nodes/influx.py b/backend/app/flow/nodes/influx.py index 7ce3a2b..cbb6a44 100644 --- a/backend/app/flow/nodes/influx.py +++ b/backend/app/flow/nodes/influx.py @@ -29,6 +29,22 @@ class InfluxDbNode(Node): Both operations use a similar configuration pattern in params, making the API consistent and the input/output data simple (just values). + - **Query passthrough**: an incoming message holding a ``flux`` key is run + as written, and every row comes back on the first output port as + ``{"rows": [{ts, value, field, measurement, tags}], **echo}``. + + The passthrough is what keeps a database node a database node. It holds the + credentials and the connection and nothing else: building a query and + shaping its rows are ordinary Python nodes on either side, so a dashboard + widget never learns which database answered it. A chart drawing an + InfluxDB series is:: + + [chart] -record-> [py: build] -record{flux, ...}-> [influx] + [chart] <-series- [py: parse] <-json{rows, ...}---' + + Nothing rate-limits the requests here; an ``interval`` on the request port + is what holds back a caller asking too often. + :param requires: Messages to write to InfluxDB. The actual value from each message is written according to the corresponding config in ``writes``. :type requires: MessageSpec | list[MessageSpec] | None @@ -208,10 +224,13 @@ class InfluxDbNode(Node): def _handler(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: """ - Handle incoming data - write to InfluxDB and optionally query. + Handle incoming data - write to InfluxDB, run a query, or both. This method is called when upstream dependencies (requires) are satisfied. - It writes the incoming data to InfluxDB and can also perform reads. + + An incoming dict carrying a ``flux`` key is a query request rather than + something to store: it is executed as written and answered on the first + output port. Everything else on the inputs is written as points. :param params: Node parameters. :type params: dict @@ -220,9 +239,24 @@ class InfluxDbNode(Node): :returns: Query results if provides is configured, None otherwise. :rtype: dict | None """ - # Write incoming data - if kwargs: - self._write_points(kwargs) + # ponytail: one request/answer pair per node — the first input holding + # a "flux" key is the request and the answer goes out on the first + # output port. A second query stream means a second node. + request = next( + (v for v in kwargs.values() if isinstance(v, dict) and "flux" in v), None + ) + writes = {k: v for k, v in kwargs.items() if v is not request} + + if writes: + self._write_points(writes) + + if request is not None: + if not self.output_ports: + raise ValueError( + f"Node '{self.name}' was sent a query but has no output to " + "answer on" + ) + return {self.output_ports[0].port: self._run_flux(request)} # If we have provides, perform queries if self.provides: @@ -230,6 +264,49 @@ class InfluxDbNode(Node): return None + def _run_flux(self, request: dict[str, Any]) -> dict[str, Any]: + """ + Execute a Flux query exactly as it was handed over, and answer its rows. + + The node stays out of the query's business: building it is the job of + whatever produced the request, and shaping the rows into something a + widget draws is the job of whatever reads the answer. Every field of + the request except ``flux`` is echoed back untouched, which is how a + caller tells its own answer from someone else's. + + :param request: The query and whatever the caller wants echoed. + :type request: dict + :returns: ``{"rows": [...], **echo}``. + :rtype: dict + """ + from influxdb_client import InfluxDBClient + + flux = str(request["flux"]) + echo = {key: value for key, value in request.items() if key != "flux"} + + logger.info("Running Flux for node '%s': %s", self.name, flux) + with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client: + tables = client.query_api().query(flux, org=self.org) + + rows = [ + { + "ts": record.get_time().timestamp() if record.get_time() else None, + "value": record.get_value(), + # Read off `values` rather than the getters: a query that keeps + # only what it needs drops these columns, and the getters raise. + "field": record.values.get("_field"), + "measurement": record.values.get("_measurement"), + "tags": { + key: value + for key, value in record.values.items() + if not key.startswith("_") and key not in ("result", "table") + }, + } + for table in tables + for record in table.records + ] + return {"rows": rows, **echo} + def _write_points(self, data: dict[str, Any]) -> None: """ Write data points to InfluxDB using configuration from ``writes``. diff --git a/backend/app/main.py b/backend/app/main.py index 7c1abd0..85fde6d 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -22,6 +22,7 @@ from app.flow.events import event_bus from app.flow.executor import ExecutionService from app.flow.metrics import MetricsCollector from app.flow.nodes.http import close_shared_client +from app.flow.pipeline import ValueSource from app.flow.plugins import load_plugins from app.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue from app.flow.secrets import init_secrets @@ -98,6 +99,11 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: workers=pool, ) app.state.flow_controller = controller + # A "dashboard" alert channel puts its alert into the graph. Bound here + # rather than passed in: the manager is built before the controller is. + alerts.publish = lambda name, value: controller.publish_message( + name, value, ValueSource(kind="api", id="alerts", label="Alerts") + ) dashboards = DashboardStore(controller.store) app.state.dashboard_store = dashboards controller.dashboards = dashboards diff --git a/backend/tests/flow/test_alerts.py b/backend/tests/flow/test_alerts.py index cda1938..20d03bf 100644 --- a/backend/tests/flow/test_alerts.py +++ b/backend/tests/flow/test_alerts.py @@ -197,3 +197,36 @@ def test_alerting_can_be_switched_off(): def test_every_alerting_event_reads_as_a_sentence(event, expected): alert = describe(event) assert (alert.title if alert else None) == expected + + +def test_a_dashboard_channel_publishes_the_alert_as_a_record(): + published: list[tuple[str, dict]] = [] + config = AlertsConfig( + channels=[ + Channel(name="panel", kind="dashboard", config={"message": "house.notice"}) + ], + rules=[Rule(events=[], channels=["panel"])], + ) + alerts = AlertManager(EventBus(), config=config, now=Clock()) + alerts.publish = lambda name, value: published.append((name, value)) + + asyncio.run(alerts.handle(error_event())) + + assert len(published) == 1 + name, record = published[0] + assert name == "house.notice" + # Flat named scalars, which is what a notification widget binds. + assert record["title"] == "heating.pump failed" + assert record["severity"] == "error" + assert all(isinstance(v, str) for v in record.values()) + + +def test_a_dashboard_channel_without_a_message_says_so(): + alerts = AlertManager(EventBus(), now=Clock()) + alerts.publish = lambda name, value: None + channel = Channel(name="panel", kind="dashboard") + + with pytest.raises(ValueError): + asyncio.run( + alerts.send(channel, Alert(title="t", body="b"), raise_on_error=True) + ) diff --git a/backend/tests/flow/test_dashboards.py b/backend/tests/flow/test_dashboards.py index 6cc7cd8..1e0aa07 100644 --- a/backend/tests/flow/test_dashboards.py +++ b/backend/tests/flow/test_dashboards.py @@ -187,3 +187,64 @@ def test_publishing_nothing_does_nothing(): pipeline.publish({}) assert pipeline.values() == {} + + +def query_chart(**config) -> dict: + return { + "source": "query", + "request": "heating.chart_req", + "request_dtype": "record", + "message": "heating.chart_series", + "dtype": "series", + "refresh_s": 30, + **config, + } + + +def test_a_widget_refuses_a_dtype_it_cannot_carry(): + WidgetDef(id="s", type="switch", config={"message": "a.b", "dtype": "bool"}) + # A document written before the editor recorded types binds anything. + WidgetDef(id="s", type="switch", config={"message": "a.b"}) + + with pytest.raises(ValueError): + WidgetDef(id="s", type="switch", config={"message": "a.b", "dtype": "float"}) + + +def test_the_structured_widgets_bind_their_shapes(): + WidgetDef(id="a", type="agenda", config={"message": "a.b", "dtype": "list"}) + WidgetDef(id="n", type="notification", config={"message": "a.b", "dtype": "record"}) + + with pytest.raises(ValueError): + WidgetDef(id="a", type="agenda", config={"message": "a.b", "dtype": "json"}) + with pytest.raises(ValueError): + WidgetDef( + id="n", type="notification", config={"message": "a.b", "dtype": "list"} + ) + + +def test_a_querying_chart_asks_with_a_record_and_draws_a_series(): + widget = WidgetDef(id="c", type="chart", config=query_chart()) + + # It publishes its request and reads the answer, so the canvas draws both. + assert widget.target == "heating.chart_req" + assert widget.messages == ["heating.chart_series"] + + with pytest.raises(ValueError): + WidgetDef(id="c", type="chart", config=query_chart(dtype="float")) + with pytest.raises(ValueError): + WidgetDef(id="c", type="chart", config=query_chart(request_dtype="json")) + + +def test_a_querying_chart_keeps_no_ring(): + """The answer carries its own past; a ring would store it twice.""" + widget = WidgetDef( + id="c", type="chart", config=query_chart(history={"points": 900}) + ) + + assert widget.history_points == 0 + + +def test_a_dashboard_is_cut_into_a_sane_number_of_columns(): + for columns in (0, 49): + with pytest.raises(ValueError): + DashboardDef(name="house", columns=columns) diff --git a/backend/tests/flow/test_messages.py b/backend/tests/flow/test_messages.py index e202dcb..d3aab24 100644 --- a/backend/tests/flow/test_messages.py +++ b/backend/tests/flow/test_messages.py @@ -36,6 +36,67 @@ def test_json_dtype_round_trips(): assert json.loads(json.dumps(value)) == value +def test_record_takes_flat_scalars_only(): + spec = MessageSpec(name="notice", dtype=DType.RECORD) + spec.check({"title": "Boiler", "count": 3, "hot": True, "detail": None}) + for wrong in ({"a": {"nested": 1}}, {"a": [1]}, [], "x"): + with pytest.raises(TypeError): + spec.check(wrong) + + +def test_series_carries_lines_and_whatever_else_the_answer_echoes(): + spec = MessageSpec(name="temps", dtype=DType.SERIES) + spec.check( + { + "range_s": 3600, + "interval_s": 60, + "lines": [{"label": "living", "points": [[1.0, 21.5], [2.0, 21.6]]}], + } + ) + spec.check({"lines": []}) + for wrong in ( + {"lines": [{"label": "a", "points": [[1.0, 2.0, 3.0]]}]}, + {"lines": [{"label": "a", "points": [["1", 2.0]]}]}, + {"lines": [{"label": 1, "points": []}]}, + {"lines": {}}, + {}, + ): + with pytest.raises(TypeError): + spec.check(wrong) + + +def test_list_items_follow_the_declared_shape(): + records = MessageSpec(name="agenda", dtype=DType.LIST) + records.check([{"title": "Dentist", "ts": 1.0}]) + records.check([]) + with pytest.raises(TypeError): + records.check([1.0]) + + numbers = MessageSpec(name="window", dtype=DType.LIST, item=DType.FLOAT) + numbers.check([21.4, 2]) + for wrong in ([True], ["1"], 1.0): + with pytest.raises(TypeError): + numbers.check(wrong) + + +def test_the_failing_item_is_named(): + spec = MessageSpec(name="window", dtype=DType.LIST, item=DType.FLOAT) + with pytest.raises(TypeError, match="index 1"): + spec.check([1.0, "x"]) + + +def test_a_list_holds_one_declared_level(): + for item in (DType.LIST, DType.SERIES): + with pytest.raises(ValueError): + MessageSpec(name="x", dtype=DType.LIST, item=item) + + +def test_coerce_parses_structured_text(): + spec = MessageSpec(name="notice", dtype=DType.RECORD) + assert spec.coerce('{"title": "Boiler"}') == {"title": "Boiler"} + assert spec.coerce({"title": "Boiler"}) == {"title": "Boiler"} + + def test_coerce_from_text(): assert MessageSpec(name="a", dtype=DType.FLOAT).coerce("2.5") == 2.5 assert MessageSpec(name="a", dtype=DType.INT).coerce("7") == 7 diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py index 771ecf0..c04c81d 100644 --- a/backend/tests/flow/test_node_types.py +++ b/backend/tests/flow/test_node_types.py @@ -133,3 +133,60 @@ def test_every_offered_type_has_a_fixture(): if info.type != "python" and not info.plugin } assert offered == set(FIXTURES) + + +def test_a_flux_request_is_run_rather_than_written(monkeypatch): + """A query passing through is not a point to store.""" + from app.flow.nodes import InfluxDbNode + + node = InfluxDbNode( + requires=[ + MessageSpec(name="query", dtype=DType.RECORD), + MessageSpec(name="temp", dtype=DType.FLOAT), + ], + provides=[MessageSpec(name="answer", dtype=DType.JSON)], + params={"url": "http://influx", "token": "t", "org": "o", "bucket": "b"}, + ) + # The class, not the instance: the node types use __slots__. + written: list[dict] = [] + monkeypatch.setattr( + InfluxDbNode, "_write_points", lambda self, data: written.append(data) + ) + monkeypatch.setattr( + InfluxDbNode, + "_run_flux", + lambda self, request: { + "rows": [], + **{key: value for key, value in request.items() if key != "flux"}, + }, + ) + + out = node.execute( + { + "query": {"flux": 'from(bucket: "b")', "range_s": 3600}, + "temp": 21.5, + } + ) + + # The reading is stored, the query is not. + assert written == [{"temp": 21.5}] + # The answer carries back what the caller asked for. + assert out == {"answer": {"rows": [], "range_s": 3600}} + + +def test_a_falsy_return_is_a_mistake_not_silence(): + """Only None means "nothing to publish".""" + from app.flow.nodes import Node + from app.flow.nodes.base import NodeOutputError + + def make(retval): + return Node( + f=lambda params: retval, + provides=[MessageSpec(name="out", dtype=DType.FLOAT)], + ) + + assert make(None).execute({}) is None + assert make({}).execute({}) is None + for wrong in ([], 0, ""): + with pytest.raises(NodeOutputError): + make(wrong).execute({}) diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index ee1f1a8..04c9633 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -298,7 +298,7 @@ export const ChannelSchema = { }, kind: { type: 'string', - enum: ['ntfy', 'smtp', 'webhook'], + enum: ['ntfy', 'smtp', 'webhook', 'dashboard'], title: 'Kind' }, enabled: { @@ -320,10 +320,15 @@ export const ChannelSchema = { export const DTypeSchema = { type: 'string', - enum: ['float', 'int', 'str', 'bool', 'json'], + enum: ['float', 'int', 'str', 'bool', 'json', 'series', 'record', 'list'], title: 'DType', description: `Serializable payload types. +The scalars carry what a single reading can say. The three structured ones +are declared shapes rather than "some JSON": a widget or a downstream node +knows what it is getting before anything runs, which is what lets the +dashboard picker offer a message and refuse a wrong binding. + Binary payloads (tensors, images) will arrive later as explicitly declared codec fields; until then everything on the wire is JSON.` } as const; @@ -1151,6 +1156,16 @@ export const MessageSpecSchema = { '$ref': '#/components/schemas/DType', default: 'float' }, + item: { + anyOf: [ + { + '$ref': '#/components/schemas/DType' + }, + { + type: 'null' + } + ] + }, interval: { type: 'number', minimum: 0, @@ -1172,6 +1187,10 @@ export const MessageSpecSchema = { :param port: The identifier the node function sees. Defaults to the last segment of \`\`name\`\`, so unqualified flows read naturally. :param dtype: Payload type, validated on every message that passes through. +:param item: The type of each item of a \`\`list\`\` port, ignored otherwise. + Unset means \`\`record\`\`, which is what the agenda and forecast widgets + read; \`\`float\`\` is the numeric list a pipeline passes around. A list of + lists, or of series, is refused — one declared level is the point. :param interval: Deliver at most every this many seconds; 0 is every time. On an output it holds back publishing, on an input it holds back waking the node. The value is never lost — state keeps the latest — only the @@ -2497,7 +2516,7 @@ export const WidgetDefSchema = { }, type: { type: 'string', - enum: ['stat', 'gauge', 'chart', 'markdown', 'button', 'switch', 'slider', 'input', 'dropdown'], + enum: ['stat', 'gauge', 'chart', 'markdown', 'agenda', 'notification', 'button', 'switch', 'slider', 'input', 'dropdown'], title: 'Type' }, title: { @@ -2525,7 +2544,14 @@ export const WidgetDefSchema = { \`\`config\`\` is per type — a chart names its series, a button names the message it publishes — and is validated against the type below rather than -by a schema per class, because the whole set is small and closed.` +by a schema per class, because the whole set is small and closed. + +A chart comes in two kinds. The default reads what the engine kept for a +message. One with \`\`source: "query"\`\` asks instead, and its config is +\`\`{source, request, request_dtype: "record", message, dtype: "series", +refresh_s, range_s}\`\`: it publishes \`\`{range_s, interval_s}\`\` to +\`\`request\`\` exactly as a slider publishes a value, and draws the \`\`series\`\` +a flow answers with on \`\`message\`\`.` } as const; export const app__api__routes__dashboards__PublishRequestSchema = { diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 6996b11..1a6b205 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -120,14 +120,14 @@ export type BrainNode = { */ export type Channel = { name: string; - kind: 'ntfy' | 'smtp' | 'webhook'; + kind: 'ntfy' | 'smtp' | 'webhook' | 'dashboard'; enabled?: boolean; config?: { [key: string]: unknown; }; }; -export type kind = 'ntfy' | 'smtp' | 'webhook'; +export type kind = 'ntfy' | 'smtp' | 'webhook' | 'dashboard'; /** * A dashboard as stored, and as the API hands it over. @@ -185,10 +185,15 @@ export type DeadLetter = { /** * Serializable payload types. * + * The scalars carry what a single reading can say. The three structured ones + * are declared shapes rather than "some JSON": a widget or a downstream node + * knows what it is getting before anything runs, which is what lets the + * dashboard picker offer a message and refuse a wrong binding. + * * Binary payloads (tensors, images) will arrive later as explicitly declared * codec fields; until then everything on the wire is JSON. */ -export type DType = 'float' | 'int' | 'str' | 'bool' | 'json'; +export type DType = 'float' | 'int' | 'str' | 'bool' | 'json' | 'series' | 'record' | 'list'; /** * Something wired into this flow that is not a node in it. @@ -390,6 +395,10 @@ export type MessagePoints = { * :param port: The identifier the node function sees. Defaults to the last * segment of ``name``, so unqualified flows read naturally. * :param dtype: Payload type, validated on every message that passes through. + * :param item: The type of each item of a ``list`` port, ignored otherwise. + * Unset means ``record``, which is what the agenda and forecast widgets + * read; ``float`` is the numeric list a pipeline passes around. A list of + * lists, or of series, is refused — one declared level is the point. * :param interval: Deliver at most every this many seconds; 0 is every time. * On an output it holds back publishing, on an input it holds back waking * the node. The value is never lost — state keeps the latest — only the @@ -403,6 +412,7 @@ export type MessageSpec = { name?: string; port?: string; dtype?: DType; + item?: (DType | null); interval?: number; trigger?: boolean; }; @@ -764,10 +774,17 @@ export type ValidationResult = { * ``config`` is per type — a chart names its series, a button names the * message it publishes — and is validated against the type below rather than * by a schema per class, because the whole set is small and closed. + * + * A chart comes in two kinds. The default reads what the engine kept for a + * message. One with ``source: "query"`` asks instead, and its config is + * ``{source, request, request_dtype: "record", message, dtype: "series", + * refresh_s, range_s}``: it publishes ``{range_s, interval_s}`` to + * ``request`` exactly as a slider publishes a value, and draws the ``series`` + * a flow answers with on ``message``. */ export type WidgetDef = { id: string; - type: 'stat' | 'gauge' | 'chart' | 'markdown' | 'button' | 'switch' | 'slider' | 'input' | 'dropdown'; + type: 'stat' | 'gauge' | 'chart' | 'markdown' | 'agenda' | 'notification' | 'button' | 'switch' | 'slider' | 'input' | 'dropdown'; title?: string; layout?: { [key: string]: Placement; @@ -777,7 +794,7 @@ export type WidgetDef = { }; }; -export type type = 'stat' | 'gauge' | 'chart' | 'markdown' | 'button' | 'switch' | 'slider' | 'input' | 'dropdown'; +export type type = 'stat' | 'gauge' | 'chart' | 'markdown' | 'agenda' | 'notification' | 'button' | 'switch' | 'slider' | 'input' | 'dropdown'; export type AlertsReadAlertsConfigResponse = (AlertsConfig); diff --git a/frontend/src/components/Common/UplotChart.tsx b/frontend/src/components/Common/UplotChart.tsx index 77c0071..1c30105 100644 --- a/frontend/src/components/Common/UplotChart.tsx +++ b/frontend/src/components/Common/UplotChart.tsx @@ -58,6 +58,8 @@ export function UplotChart({ labels, plots, empty = "Nothing has come through yet.", + unit, + yRange, onCursor, onSelect, }: { @@ -66,6 +68,10 @@ export function UplotChart({ /** The points of each series, in the same order as `labels`. */ plots: HistoryPoint[][] empty?: string + /** Written after every reading, on the axis and in the legend. */ + unit?: string + /** A y axis fixed to these bounds; unset lets it follow the data. */ + yRange?: [number, number] /** The x value under the pointer, and null once it leaves the plot. */ onCursor?: (ts: number | null) => void /** The x value clicked, or null for a click that landed on no point. */ @@ -85,8 +91,9 @@ export function UplotChart({ const points = plots.reduce((total, plot) => total + plot.length, 0) // The identity of the series set: the chart is rebuilt when it changes, - // while a new reading only sets its data. - const key = labels.join(" ") + // while a new reading only sets its data. The unit and the fixed range are + // part of it — both are baked into the axes when the chart is built. + const key = `${labels.join(" ")}|${unit ?? ""}|${yRange?.join(",") ?? ""}` // uPlot leaves its axes half-initialised while the scales have no range, and // a resize in that window (a card still settling, say) draws them anyway and // throws. Waiting for the first reading avoids the state altogether. @@ -139,16 +146,21 @@ export function UplotChart({ }, ], }, - scales: { x: { time: true } }, + scales: { + x: { time: true }, + ...(yRange ? { y: { range: yRange } } : {}), + }, axes: [ { ...axis, size: 28 }, { ...axis, - size: 46, + // The unit is written after every tick, so the gutter widens to + // hold it rather than clipping the number in front of it. + size: unit ? 62 : 46, // The gutter has a fixed width, so a grouped "15,000" would be // clipped to something that reads as a different number entirely. values: (_self: uPlot, ticks: number[]) => - ticks.map((value) => si(value)), + ticks.map((value) => (unit ? `${si(value)} ${unit}` : si(value))), }, ], series: [ @@ -160,8 +172,9 @@ export function UplotChart({ // rebuilt chart. stroke: () => seriesColor(index), // The cursor readout is what decides how wide the legend gets, so - // it is shortened here and the unit named in the card's title. - value: (_self: uPlot, raw: number) => si(raw), + // it is shortened here; a named unit is short enough to keep. + value: (_self: uPlot, raw: number) => + unit ? `${si(raw)} ${unit}` : si(raw), // Series arrive on their own clocks; a joined table is mostly // holes, and a line with a hole per point is not a line. spanGaps: true, diff --git a/frontend/src/components/Dashboard/widgets.tsx b/frontend/src/components/Dashboard/widgets.tsx index 8b59d25..02c5bce 100644 --- a/frontend/src/components/Dashboard/widgets.tsx +++ b/frontend/src/components/Dashboard/widgets.tsx @@ -43,9 +43,13 @@ export type WidgetKind = WidgetDef["type"] */ export const WIDGET_DTYPES: Partial> = { gauge: ["float", "int"], + // A chart reading the engine's ring. One that queries binds a `series` + // answer and a `record` request instead, checked in `widgetIssue`. chart: ["float", "int"], slider: ["float", "int"], switch: ["bool"], + agenda: ["list"], + notification: ["record"], } /** Whether a message of this payload type may drive this kind of widget. */ @@ -59,6 +63,8 @@ export const WIDGET_LABELS: Record = { gauge: "Gauge", chart: "Chart", markdown: "Text", + agenda: "Agenda", + notification: "Notification", button: "Button", switch: "Switch", slider: "Slider", @@ -72,6 +78,8 @@ export const WIDGET_SIZES: Record = { gauge: { w: 3, h: 3 }, chart: { w: 6, h: 4 }, markdown: { w: 6, h: 2 }, + agenda: { w: 4, h: 4 }, + notification: { w: 4, h: 2 }, button: { w: 3, h: 2 }, switch: { w: 3, h: 2 }, slider: { w: 4, h: 2 }, @@ -120,6 +128,20 @@ export function widgetIssue(widget: WidgetDef): string | null { if (widget.type === "markdown") return null const cfg = config(widget) + if (widget.type === "chart" && cfg.source === "query") { + if (!text(cfg.request)) return "This chart does not ask for anything yet." + if (!text(cfg.message)) return "This chart has no answer to draw yet." + const answer = text(cfg.dtype) + if (answer && answer !== "series") { + return `${text(cfg.message)} is a ${answer}; a chart that queries draws a series.` + } + const asked = text(cfg.request_dtype) + if (asked && asked !== "record") { + return `${text(cfg.request)} is a ${asked}; a request is a record.` + } + return null + } + if (widget.type === "chart") { const series = seriesOf(widget) if (series.length === 0) return "This chart has no series yet." @@ -317,8 +339,11 @@ function GaugeWidget({ widget }: WidgetProps) { {format( value, @@ -365,6 +390,119 @@ function MarkdownWidget({ widget }: WidgetProps) { ) } +/** One item of an agenda, as the `list` message declares it. */ +type AgendaItem = { title: string; ts: number; all_day?: boolean } + +const DAY_MS = 86_400_000 + +/** + * Which day something falls on, said the way a person would. + * + * Today and tomorrow by name, the rest of the week by weekday, and anything + * further out by date — past a week "Thursday" stops telling you which one. + */ +function dayLabel(when: Date, now: Date): string { + const midnight = new Date(now).setHours(0, 0, 0, 0) + const days = Math.floor( + (new Date(when).setHours(0, 0, 0, 0) - midnight) / DAY_MS, + ) + if (days === 0) return "Today" + if (days === 1) return "Tomorrow" + if (days < 7) return when.toLocaleDateString(undefined, { weekday: "long" }) + return when.toLocaleDateString() +} + +/** + * What is coming up, from a `list` of items the message declares. + * + * The shape is the widget's contract rather than a path per binding: every + * item is `{title, ts}` with an optional `all_day`, so a flow answering with + * a calendar decides what an entry is called and this only has to draw it. + */ +function AgendaWidget({ widget }: WidgetProps) { + const cfg = config(widget) + const message = text(cfg.message) + const live = useLiveValue(message || undefined) + if (!message) return + + const now = new Date() + const today = new Date(now).setHours(0, 0, 0, 0) / 1000 + const items = (Array.isArray(live?.value) ? live.value : []) + .filter( + (item): item is AgendaItem => + typeof item?.title === "string" && Number.isFinite(item?.ts), + ) + .filter((item) => item.ts >= today) + .sort((a, b) => a.ts - b.ts) + .slice(0, num(cfg.count, 5)) + + if (items.length === 0) { + return

Nothing coming up.

+ } + + return ( +
    + {items.map((item, index) => { + const when = new Date(item.ts * 1000) + return ( +
  • + + {dayLabel(when, now)} + {item.all_day + ? "" + : ` ${when.toLocaleTimeString(undefined, { + hour: "2-digit", + minute: "2-digit", + })}`} + + {item.title} +
    • + ) + })} +
+ ) +} + +/** + * The last thing worth saying, held until something replaces it. + * + * No state of its own: the live store already keeps the latest value of a + * message, so what was published stays on the panel until the next one lands. + */ +function NotificationWidget({ widget }: WidgetProps) { + const cfg = config(widget) + const message = text(cfg.message) + const live = useLiveValue(message || undefined) + if (!message) return + + const record = (live?.value ?? null) as Record | null + const title = text(record?.title) + const body = text(record?.body) + if (!title && !body) { + return

Nothing to report.

+ } + + return ( +
+ {title ? ( +

+ {title} +

+ ) : null} + {body ?

{body}

: null} +
+ ) +} + // --------------------------------------------------------------------------- // Input // --------------------------------------------------------------------------- @@ -437,6 +575,12 @@ function SliderWidget({ widget, dashboard }: WidgetProps) { const current = dragging ?? (typeof live?.value === "number" ? live.value : min) + // A 20–22 °C setpoint at 0.1 is unusable without marks to aim at. Past + // fifty of them the ticks are a smear, so the browser gets none. + const steps = step > 0 ? (max - min) / step : 0 + const ticks = Number.isFinite(steps) && steps > 0 && steps <= 50 ? steps : 0 + const ticksId = `ticks-${widget.id}` + return (
@@ -447,12 +591,20 @@ function SliderWidget({ widget, dashboard }: WidgetProps) { ) : null}
+ {ticks ? ( + + {Array.from({ length: Math.floor(ticks) + 1 }, (_, index) => ( + + ) : null} setDragging(Number(event.target.value))} @@ -543,6 +695,8 @@ const RENDERERS: Partial< gauge: GaugeWidget, chart: ChartWidget, markdown: MarkdownWidget, + agenda: AgendaWidget, + notification: NotificationWidget, button: ButtonWidget, switch: SwitchWidget, slider: SliderWidget, diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index ee424bb..dcbfeb4 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -50,7 +50,19 @@ import { PANEL_SECTION, PanelTitle, SidePanel } from "./SidePanel" const NodeEditor = lazy(() => import("./NodeEditor")) -const DTYPES: DType[] = ["float", "int", "str", "bool", "json"] +const DTYPES: DType[] = [ + "float", + "int", + "str", + "bool", + "json", + "series", + "record", + "list", +] + +/** What a list may hold. One declared level: no list of lists. */ +const ITEM_DTYPES: DType[] = ["record", "float", "int", "str", "bool", "json"] /** Radix selects cannot hold an empty value, so "no secret" needs a name. */ const NO_SECRET = "__none__" @@ -274,6 +286,29 @@ function PortList({ ))} + {spec.dtype === "list" ? ( + + ) : null} = { bool: "False", str: '""', json: "{}", + series: '{"lines": []}', + record: "{}", + list: "[]", } const SCAFFOLD_DOC = diff --git a/frontend/src/routes/_layout/alerts.tsx b/frontend/src/routes/_layout/alerts.tsx index 21c5b36..1c00162 100644 --- a/frontend/src/routes/_layout/alerts.tsx +++ b/frontend/src/routes/_layout/alerts.tsx @@ -57,9 +57,12 @@ const FIELDS: Record = { ], smtp: [["to", "Send to", "someone@example.com"]], webhook: [["url", "URL", "https://example.com/hook"]], + // The message has to be one a flow declares, like anything a dashboard + // writes to. A notification widget bound to it is what shows the alert. + dashboard: [["message", "Message", "house.notice"]], } -const KINDS: Channel["kind"][] = ["ntfy", "smtp", "webhook"] +const KINDS: Channel["kind"][] = ["ntfy", "smtp", "webhook", "dashboard"] /** A setting may hold a `{"$secret": "name"}` reference rather than a literal, * so text that parses as JSON is stored as JSON and survives a round trip. */