From 29d3ec467f23637c700b0b783ab64d02437a5f49 Mon Sep 17 00:00:00 2001 From: Melvin Strobl Date: Fri, 16 Oct 2026 16:40:31 +0200 Subject: [PATCH] Device node: a device's ports are wired in flows - flow/nodes/device.py: `device` node with params `device` and `config`; provides the device's `in` ports, requires its `out` ports without waking anything, never run (Node.passive), down "Offline" from presence - DeviceDef.io and DevicePort go; GET /devices/{id} computes `io` and `bindings` from the device nodes of published flows, in a fixed order, cached by FlowStore.revision; the credential's scope comes from them - issues `unknown_device` and `device_port` (wired twice; against a report the wrong direction or dtype, or unlisted as advisory), rechecked when a device is created or deleted; ValidationIssue.advisory is a field - a device publishing on a bound port is credited to its node by the server; the panel endpoint kind is gone from `_endpoints` and the canvas - a rebuild that moves a device's bindings sends `dashboard_changed` with an empty name, which the firmware refetches on - startup moves each device's `io` rows into device nodes in the flows declaring their messages, draft and published, one commit; rows no flow declares are logged and left out - frontend: Device node settings (device, Publishes, Follows, a port picker from a report), palette shortcuts per reported port, DeviceDetail lists bindings with "Use in flow" in place of the I/O editor - docs: devices, flow editor, node types, issues, API --- backend/fluksio/api/deps.py | 29 +- backend/fluksio/api/routes/devices.py | 93 +++- backend/fluksio/api/routes/flows.py | 34 +- backend/fluksio/api/routes/messages.py | 29 +- backend/fluksio/api/routes/observability.py | 5 +- backend/fluksio/flow/controller.py | 158 ++++++- backend/fluksio/flow/devices.py | 375 +++++++++++++++-- backend/fluksio/flow/nodes/__init__.py | 2 + backend/fluksio/flow/nodes/base.py | 5 + backend/fluksio/flow/nodes/device.py | 87 ++++ backend/fluksio/flow/pipeline.py | 28 +- backend/fluksio/main.py | 3 + backend/tests/api/routes/test_devices.py | 296 +++++++++++-- backend/tests/flow/test_device_node.py | 94 +++++ backend/tests/flow/test_device_store.py | 183 ++++++++ backend/tests/flow/test_endpoints.py | 27 -- backend/tests/flow/test_node_types.py | 7 + docs/code/api.md | 2 +- docs/code/nodes.md | 4 +- docs/concepts/flows.md | 4 +- docs/interface/dashboards.md | 45 +- docs/interface/flow-editor.md | 16 +- docs/reference/node-types.md | 18 + frontend/src/client/schemas.gen.ts | 144 +++++-- frontend/src/client/sdk.gen.ts | 11 +- frontend/src/client/types.gen.ts | 60 ++- .../src/components/Devices/DeviceDetail.tsx | 317 ++++++-------- frontend/src/components/Devices/ports.ts | 79 ++++ frontend/src/components/Devices/queries.ts | 12 +- .../src/components/Flow/CommandPalette.tsx | 39 +- frontend/src/components/Flow/EndpointNode.tsx | 10 +- frontend/src/components/Flow/FlowEditor.tsx | 48 ++- frontend/src/components/Flow/FlowNode.tsx | 6 +- frontend/src/components/Flow/NodePanel.tsx | 397 ++++++++++++++++-- frontend/src/components/Flow/endpoints.ts | 22 +- 35 files changed, 2195 insertions(+), 494 deletions(-) create mode 100644 backend/fluksio/flow/nodes/device.py create mode 100644 backend/tests/flow/test_device_node.py create mode 100644 frontend/src/components/Devices/ports.ts diff --git a/backend/fluksio/api/deps.py b/backend/fluksio/api/deps.py index f549043..796dde4 100644 --- a/backend/fluksio/api/deps.py +++ b/backend/fluksio/api/deps.py @@ -87,9 +87,16 @@ def _device_messages(device: DeviceDef, request: Request) -> set[str]: the answer to give when the answer cannot be worked out. """ store: DashboardStore | None = getattr(request.app.state, "dashboard_store", None) - if store is None: + owned: DeviceStore | None = getattr(request.app.state, "device_store", None) + if store is None or owned is None: return set() - return devices.messages_for(device, store) + return devices.messages_for(device, store, owned) + + +def _device_in(device: DeviceDef, request: Request) -> set[str]: + """The messages this device's own ``in`` ports are wired to publish.""" + owned: DeviceStore | None = getattr(request.app.state, "device_store", None) + return devices.io_in(device, owned) if owned is not None else set() def _locked(defn: DashboardDef, live: dict[str, Any]) -> bool: @@ -128,14 +135,15 @@ def _device_writable(device: DeviceDef, request: Request) -> set[str]: one is what entitles the screen to it. Reads are untouched — read-only is not blind, and a locked dashboard has to keep drawing live data. - The panel's own ``in`` ports are added outside the lock: it covers the - controls on a dashboard, not the sensors on the device. Its ``out`` ports - are read, not written, unless a widget targets them too. + The device's own ``in`` ports, as flows wire them, are added outside the + lock: it covers the controls on a dashboard, not the sensors on the + device. Its ``out`` ports are read, not written, unless a widget targets + them too. """ store: DashboardStore | None = getattr(request.app.state, "dashboard_store", None) if store is None: return set() - writable = devices.io_in(device) + writable = _device_in(device, request) defns = devices.dashboards_for(device, store) bound = sorted( {s.message for d in defns if (s := d.settings.get("locked")) and s.message} @@ -197,11 +205,18 @@ def _device_may(payload: dict[str, Any], request: Request) -> None: a panel to read every message it names and to publish to none of them but the requests its own charts ask by. + Its own ports are the device nodes the published flows hold for it, so + wiring a port in a flow is what lets the device publish on it. + + The device is left on ``request.state`` for the route, which credits a + value published on one of its ports to the node holding that port. + # ponytail: the device file, its nonce and the published dashboards are # re-read per request. Cache them behind the store's revision if this # shows up in a profile. """ device = _device_for(payload, request.app) + request.state.device = device api = settings.API_V1_STR path = request.url.path @@ -252,7 +267,7 @@ def _device_may(payload: dict[str, Any], request: Request) -> None: elif method == "PUT" and path == f"{api}/artifacts": # What the device's own sensor recorded — a microphone's utterance — # uploaded under the ``in`` message it is about to publish it to. - allowed = request.query_params.get("name", "") in devices.io_in(device) + allowed = request.query_params.get("name", "") in _device_in(device, request) if not allowed: raise HTTPException( diff --git a/backend/fluksio/api/routes/devices.py b/backend/fluksio/api/routes/devices.py index 3851775..ba68de1 100644 --- a/backend/fluksio/api/routes/devices.py +++ b/backend/fluksio/api/routes/devices.py @@ -13,7 +13,8 @@ comes. Anyone on the internet can ask the portal for a code, so a request that came that way is adopted by its code only. The credential is scoped: ``fluksio.api.deps`` lets it reach the dashboards -that device was assigned, the messages their widgets bind to, and nothing else. +that device was assigned, the messages their widgets bind to, the messages the +device nodes in published flows wire its own ports to, and nothing else. Deleting the device revokes it, and so does unpairing, which bumps the device's nonce and leaves what it shows as it was. @@ -25,6 +26,7 @@ from __future__ import annotations import secrets import time +from contextlib import suppress from datetime import datetime, timedelta from typing import Any, cast @@ -43,9 +45,12 @@ from fluksio.api.deps import ( from fluksio.cloud import config as cloud_config from fluksio.core import security from fluksio.core.config import settings +from fluksio.flow.controller import RebuildBusy from fluksio.flow.devices import ( + DeviceBinding, DeviceDef, DeviceExists, + DeviceIo, DeviceNotFound, DeviceStore, is_lan, @@ -288,12 +293,27 @@ class DeviceCreate(BaseModel): title: str = "" -class DevicePublic(DeviceDef): +class DeviceDocument(DeviceDef): + """A device as it reads itself: what it shows, and what its ports carry. + + ``io`` is worked out from the device nodes in published flows, in a fixed + order, so an unchanged state reads back as the same bytes. + """ + + #: Each port the flows wire, and the message it publishes or follows. + io: list[DeviceIo] = Field(default_factory=list) + #: The same ports by the flow and node holding them. + bindings: list[DeviceBinding] = Field(default_factory=list) + + +class DevicePublic(DeviceDocument): """A device as the Devices tab lists it, with whether it is there.""" online: bool = False #: The last request it made, since this instance started. last_seen: datetime | None = None + #: What the device says it has, as it last said so. None until it reports. + report: dict[str, Any] | None = None class DevicesPublic(BaseModel): @@ -364,6 +384,26 @@ def _not_found(device_id: str) -> HTTPException: return HTTPException(status_code=404, detail=f"No device named {device_id!r}") +def _document(device: DeviceDef, store: DeviceStore) -> DeviceDocument: + return DeviceDocument( + **device.model_dump(), + io=store.io(device.id), + bindings=store.bindings(device.id), + ) + + +async def _recheck(request: Request) -> None: + """Validate the flows again: a device node may name this device. + + The device is saved either way, so a busy engine is not this request's + failure: the next rebuild validates again. + """ + controller = getattr(request.app.state, "flow_controller", None) + if controller is not None: + with suppress(RebuildBusy): + await controller.recheck() + + def _changed() -> None: """Tell every device socket that what it shows may have moved. @@ -457,15 +497,19 @@ panels_router.add_api_route( @router.get("/", response_model=DevicesPublic, dependencies=[Depends(get_current_user)]) async def read_devices(store: DeviceStoreDep) -> Any: """Every device, what each one shows, and whether it is connected.""" - devices = await run_in_threadpool(store.list) - data = [ - DevicePublic( - **device.model_dump(), - online=presence.online(device.id), - last_seen=presence.last_seen(device.id), - ) - for device in devices - ] + + def gather() -> list[DevicePublic]: + return [ + DevicePublic( + **_document(device, store).model_dump(), + online=presence.online(device.id), + last_seen=presence.last_seen(device.id), + report=store.state(device.id).report, + ) + for device in store.list() + ] + + data = await run_in_threadpool(gather) return DevicesPublic( data=data, count=len(data), @@ -503,7 +547,10 @@ def pending_device(code: str) -> Any: dependencies=[Depends(get_current_active_superuser)], ) async def create_device( - device_id: str, store: DeviceStoreDep, body: DeviceCreate | None = None + device_id: str, + store: DeviceStoreDep, + request: Request, + body: DeviceCreate | None = None, ) -> Any: """Add a device. It shows nothing until it is given dashboards.""" if device_id in RESERVED: @@ -513,35 +560,38 @@ async def create_device( except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) try: - return await run_in_threadpool(store.create, device) + created = await run_in_threadpool(store.create, device) except DeviceExists: raise HTTPException(status_code=409, detail=f"{device_id!r} already exists") + await _recheck(request) + return created @router.get( "/{device_id}", - response_model=DeviceDef, + response_model=DeviceDocument, dependencies=[Depends(get_current_user)], ) async def read_device(device_id: str, store: DeviceStoreDep) -> Any: """One device, read by a person or by the device holding its credential. - The definition alone: a device compares what it reads with what it is - showing, and a field that changes on every request would rebuild its - screen each time. Declared after the pairing routes and ``/pending``, so - neither is mistaken for a device of that name. + The definition and what the flows wire its ports to, nothing about this + moment: a device compares what it reads with what it is showing, and a + field that changes on every request would rebuild its screen each time. + Declared after the pairing routes and ``/pending``, so neither is mistaken + for a device of that name. """ device = await run_in_threadpool(store.find, device_id) if device is None: raise _not_found(device_id) - return device + return await run_in_threadpool(_document, device, store) panels_router.add_api_route( "/{device_id}", read_device, methods=["GET"], - response_model=DeviceDef, + response_model=DeviceDocument, dependencies=[Depends(get_current_user)], ) @@ -580,12 +630,13 @@ async def save_device(device_id: str, body: DeviceDef, store: DeviceStoreDep) -> response_model=Message, dependencies=[Depends(get_current_active_superuser)], ) -async def delete_device(device_id: str, store: DeviceStoreDep) -> Any: +async def delete_device(device_id: str, store: DeviceStoreDep, request: Request) -> Any: """Remove the device and refuse its credential, wherever it was minted here.""" try: await run_in_threadpool(store.delete, device_id) except DeviceNotFound: raise _not_found(device_id) + await _recheck(request) _changed() return Message(message=f"Deleted {device_id}") diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 925c631..76eca9f 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -79,19 +79,18 @@ ws_router = APIRouter(prefix="/flows", tags=["flows"]) class Endpoint(BaseModel): """Something wired into this flow that is not a node in it. - A dashboard control setting one of its messages, a tile showing one, a - port on a panel's own hardware, or a node in another flow on the far side - of a dotted name. The canvas draws these so a value never appears to come + A dashboard control setting one of its messages, a tile showing one, or a + node in another flow on the far side of a dotted name. The canvas draws these so a value never appears to come from nowhere — or worse, appears to come from whichever node happens to be drawn as a producer. """ - #: dashboard, panel or flow. + #: dashboard or flow. kind: str #: Stable within its kind, and used as the canvas node id. id: str label: str - #: What sort of widget, the panel, or the node type in the other flow. + #: What sort of widget, or the node type in the other flow. detail: str = "" #: Messages of this flow it publishes, and ones it reads. provides: list[str] = [] @@ -157,29 +156,6 @@ def _endpoints(controller: FlowController, definition: FlowDef) -> list[Endpoint requires=binding["requires"], ) - # A device's own hardware: an ``in`` port publishes like a control, an - # ``out`` port reads like a tile. The id is the source id the device - # publishes under, so its edges pulse for its values. - prefix = f"{flow}." - devices: DeviceStore | None = getattr(controller, "devices", None) - for device in devices.list() if devices is not None else []: - for port in device.io: - if not port.message.startswith(prefix): - continue - key = f"panel:{device.id}:{port.port}" - endpoint = found.setdefault( - key, - Endpoint( - kind="panel", - id=key, - label=port.port, - detail=device.title or device.id, - ), - ) - side = endpoint.provides if port.direction == "in" else endpoint.requires - if port.message not in side: - side.append(port.message) - def _flow_endpoint(key: str, detail: str) -> Endpoint: """One label per node on the far side, however many names reach it.""" return found.setdefault( @@ -1102,7 +1078,7 @@ def panel_scope(token: str, app: Any) -> set[str] | None: devices: DeviceStore | None = getattr(app.state, "device_store", None) if store is None or devices is None: return set() - return messages_for(devices.find(device_id), store) + return messages_for(devices.find(device_id), store, devices) def event_for_panel(event: dict[str, Any], only: set[str]) -> bool: diff --git a/backend/fluksio/api/routes/messages.py b/backend/fluksio/api/routes/messages.py index 9c2f776..a96e1d3 100644 --- a/backend/fluksio/api/routes/messages.py +++ b/backend/fluksio/api/routes/messages.py @@ -8,11 +8,12 @@ into it without owning a node. from typing import Any -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.concurrency import run_in_threadpool from pydantic import BaseModel from fluksio.api.deps import FlowControllerDep, get_current_user +from fluksio.flow.devices import DeviceDef, DeviceStore from fluksio.flow.messages import flow_of from fluksio.flow.pipeline import ValueSource from fluksio.flow.state import as_number @@ -71,16 +72,38 @@ async def read_messages(controller: FlowControllerDep) -> Any: return MessagesPublic(data=infos, count=len(infos)) +def _port_source(request: Request, name: str) -> ValueSource | None: + """The device node a device's own publish on a bound ``in`` port is from. + + Said by the server rather than by the device: the node is what the canvas + draws the value leaving from, and the device does not know which node in + which flow holds its port. + """ + device: DeviceDef | None = getattr(request.state, "device", None) + devices: DeviceStore | None = getattr(request.app.state, "device_store", None) + if device is None or devices is None: + return None + for bound in devices.bound(device.id): + if bound.direction == "in" and bound.message == name: + return ValueSource( + kind="node", + id=f"{bound.flow}.{bound.node}", + label=device.title or device.id, + detail=bound.port, + ) + return None + + @router.post("/{name}", response_model=MessageValue) async def publish_message( - name: str, body: PublishRequest, controller: FlowControllerDep + name: str, body: PublishRequest, controller: FlowControllerDep, request: Request ) -> Any: """Put a value into the graph, as a dashboard control does. Only a message some flow declares can be published to: flows own the namespace, and a dashboard is a client of it rather than a second author. """ - source = ValueSource( + source = _port_source(request, name) or ValueSource( kind=body.source_kind, id=body.source_id, label=body.source_label or body.source_id or "API", diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index 1374406..5003831 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -28,7 +28,6 @@ from fluksio.api.deps import ( from fluksio.api.routes.runs import elapsed_ms from fluksio.core.config import settings from fluksio.flow.controller import NodeStatus -from fluksio.flow.pipeline import ADVISORY_ISSUES from fluksio.models import EngineEvent, FlowRun, Message, MetricBucket router = APIRouter( @@ -236,9 +235,7 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: { issue.flow for issue in controller.issues - if issue.flow - and issue.code != "node_error" - and issue.code not in ADVISORY_ISSUES + if issue.flow and issue.code != "node_error" and not issue.advisory } ) if invalid: diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 02f1a59..5741e5e 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -22,6 +22,7 @@ import logging import threading import time import traceback +from collections import Counter from collections.abc import Callable from contextlib import suppress from dataclasses import dataclass, field @@ -42,6 +43,7 @@ from fluksio.flow.nodes import ( ChangeNode, DebounceNode, DelayNode, + DeviceNode, ExecNode, FileNode, HttpNode, @@ -58,7 +60,6 @@ from fluksio.flow.nodes import ( WrenNode, ) from fluksio.flow.pipeline import ( - ADVISORY_ISSUES, NodeOutcome, Pipeline, RunCacheLookup, @@ -435,6 +436,12 @@ NODE_TYPES: dict[str, NodeType] = { cls=NtfyNode, params_schema=_schema_of(NtfyNode), ), + "device": NodeType( + title="Device", + description="A paired device's own ports: what it publishes, what it follows.", + cls=DeviceNode, + params_schema=_schema_of(DeviceNode), + ), } @@ -492,9 +499,12 @@ class FlowController: # Set by the lifespan once the store exists; the canvas asks the # controller which dashboards are wired into a flow. self.dashboards: Any = None - #: The device store, for the same reason: a device's own ports are - #: drawn on the canvas of the flow they wire into. + #: The device store, for the same reason: validation checks a device + #: node against the device it names. self.devices: Any = None + #: Every device's bound ports as of the last build, so a rebuild that + #: moves them can tell the devices to read theirs again. + self._device_wiring: Any = None self.pipeline: Pipeline | None = None self.loaded: dict[str, LoadedNode] = {} @@ -696,7 +706,9 @@ class FlowController: self.pipeline.history_limits = self.history_limits if self.execution is not None: self.execution.bind(self.pipeline) - self.issues = _collect_issues(loaded, self.pipeline, flow_inputs) + self.issues = _collect_issues( + loaded, self.pipeline, flow_inputs, self.devices + ) await self._activate() @@ -718,6 +730,7 @@ class FlowController: "paused": self.paused_flows(), } ) + self._rewired() async def reload_flow( self, name: str, wait: float | None = FLOW_REBUILD_WAIT @@ -800,7 +813,7 @@ class FlowController: # Over the whole graph, because it has to be: taking a producer out # of one flow is what leaves another flow's input unconnected. self.issues = _collect_issues( - self.loaded, pipeline, self._all_flow_inputs() + self.loaded, pipeline, self._all_flow_inputs(), self.devices ) await self._activate(name) @@ -821,6 +834,47 @@ class FlowController: "paused": self.paused_flows(), } ) + self._rewired() + + async def recheck(self) -> None: + """Validate the built graph again without rebuilding anything. + + For what validation reads from outside the flows: a device created or + deleted changes whether a device node names one that exists. + """ + if self.pipeline is None: + return + await self._acquire(FLOW_REBUILD_WAIT) + try: + self.issues = _collect_issues( + self.loaded, self.pipeline, self._all_flow_inputs(), self.devices + ) + finally: + self._lock.release() + self._publish( + { + "type": "pipeline_rebuilt", + "issues": [issue.model_dump() for issue in self.issues], + "nodes": [status.model_dump() for status in self.node_statuses()], + "paused": self.paused_flows(), + } + ) + + def _rewired(self) -> None: + """Tell the devices when a rebuild moved what their ports are wired to. + + An empty dashboard name is what a device refetches its own document + on, and what its socket rescopes on. The first build only takes note: + a device connecting later reads the document as it is. + """ + if self.devices is None: + return + wiring = self.devices.wiring() + known, self._device_wiring = self._device_wiring, wiring + if known is not None and known != wiring: + self._publish( + {"type": "dashboard_changed", "dashboard": "", "ts": time.time()} + ) async def reload_failed_flows(self) -> list[str]: """Rebuild only the flows holding a node that would not load. @@ -1375,7 +1429,7 @@ class FlowController: max_workers=self.max_workers, initial_values=initial_values, ) - issues = _collect_issues(loaded, pipeline, flow_inputs) + issues = _collect_issues(loaded, pipeline, flow_inputs, self.devices) return Preview( nodes=[ NodeStatusPublic( @@ -1591,7 +1645,7 @@ class FlowController: # it, so all of them are marked rather than an arbitrary one. troubled: dict[str, str] = {} for issue in self.issues + self._health_issues(): - if issue.code in ADVISORY_ISSUES: + if issue.advisory: continue for member in (*issue.nodes, *filter(None, [issue.node])): troubled.setdefault(member, issue.message) @@ -1983,9 +2037,10 @@ def _collect_issues( loaded: dict[str, LoadedNode], pipeline: Pipeline, flow_inputs: dict[str, bool], + devices: Any = None, ) -> list[ValidationIssue]: """Everything wrong with a built graph, whether it is running or not.""" - issues = pipeline.validate(flow_inputs) + issues = pipeline.validate(flow_inputs) + _device_issues(loaded, devices) issues += [ ValidationIssue( code="node_error", @@ -2046,6 +2101,93 @@ def _collect_issues( return issues +def _device_issues( + loaded: dict[str, LoadedNode], devices: Any +) -> list[ValidationIssue]: + """Device nodes naming no device, and ports that cannot be what they say. + + A port is wired to one message in one place, so one bound twice across + the graph is an error on each. Against a report, a port in the wrong + direction or of another dtype is an error too, and one the device does not + list is a warning: the device may know it all the same. Without a report + a port is the device's business and nothing is said about it. + """ + if devices is None: + return [] + nodes = [ + (entry, entry.node) + for entry in loaded.values() + if isinstance(entry.node, DeviceNode) + ] + if not nodes: + return [] + known = {device.id for device in devices.list()} + bound = Counter( + (node.cfg.device, spec.port) + for _, node in nodes + for spec in (*node.input_ports, *node.output_ports) + if spec.name + ) + issues: list[ValidationIssue] = [] + for entry, node in nodes: + device = node.cfg.device + if device not in known: + issues.append( + ValidationIssue( + code="unknown_device", + message=( + f"'{node.local_id}' names device '{device}', which does " + "not exist." + if device + else f"Pick the device '{node.local_id}' stands for." + ), + flow=entry.flow, + node=entry.id, + ) + ) + continue + reported = devices.reported_ports(device) + for direction, specs in (("in", node.output_ports), ("out", node.input_ports)): + for spec in specs: + if not spec.name: + continue + problem, advisory = "", False + said = reported.get(spec.port) if reported is not None else None + if bound[(device, spec.port)] > 1: + problem = ( + f"Port '{spec.port}' of '{device}' is wired more than " + "once. A port carries one message, in one flow." + ) + elif reported is None: + pass + elif said is None: + problem = f"'{device}' does not report a port '{spec.port}'." + advisory = True + elif said.get("direction") and said["direction"] != direction: + problem = ( + f"'{spec.port}' of '{device}' is an {said['direction']} " + f"port, wired here as {direction}." + ) + elif said.get("dtype") and said["dtype"] != spec.dtype.value: + problem = ( + f"'{spec.port}' of '{device}' carries {said['dtype']}, " + f"wired here to a {spec.dtype.value} message." + ) + if problem: + issues.append( + ValidationIssue( + code="device_port", + message=problem, + flow=entry.flow, + node=entry.id, + port=spec.port, + message_name=spec.name, + advisory=advisory, + ) + ) + return issues + + def _short_error(exc: Exception) -> str: """One line a node author can act on: what broke, and where in their code. diff --git a/backend/fluksio/flow/devices.py b/backend/fluksio/flow/devices.py index f081a4a..874b2ed 100644 --- a/backend/fluksio/flow/devices.py +++ b/backend/fluksio/flow/devices.py @@ -11,6 +11,12 @@ committed like a dashboard, so it is versioned with the dashboards it names. What a revert must never undo lives outside git, in ``devices.json``: the nonce a credential is held against, and the device's own report of what it has. Whether a device is connected right now is kept in memory only. + +A device's own ports, a button on its GPIO or its backlight, are wired in +flows rather than here: a ``device`` node names the device, publishes what its +``in`` ports send and reads what its ``out`` ports follow. What the device +reads back as its ``io``, and what its credential may reach, is worked out +from those nodes in the published flows. """ from __future__ import annotations @@ -22,6 +28,8 @@ import os import re import threading import time +from collections.abc import Callable +from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path from typing import Any, Literal @@ -30,8 +38,8 @@ from pydantic import BaseModel, Field, TypeAdapter, ValidationError, field_valid from fluksio.flow.dashboards import DashboardDef, DashboardNotFound, DashboardStore from fluksio.flow.events import event_bus -from fluksio.flow.messages import validate_qualified -from fluksio.flow.schemas import _validate_name +from fluksio.flow.messages import MessageSpec, flow_of, qualify, validate_qualified +from fluksio.flow.schemas import FlowDef, NodeDef, _validate_name from fluksio.flow.store import FlowStore, InvalidName, StaleVersion, checked logger = logging.getLogger(__name__) @@ -42,32 +50,75 @@ DEVICE_DIR = "_devices" #: Free text the device interprets: ``gpio.4``, ``backlight``, ``mic``. PORT_PATTERN = re.compile(r"^[a-z0-9_.]+$") +#: The node type that stands for a device in a flow. +DEVICE_TYPE = "device" -class DevicePort(BaseModel): - """One piece of the device's own hardware, wired to a message. - A wall panel is a node as well as a screen: a button on its GPIO, a - microphone, a speaker. Which ports exist is the device's business, so the - name is free text and ``config`` is passed through to it unread. - """ +@dataclass(frozen=True) +class Binding: + """One port of a device, wired to a message by a device node.""" + device: str + flow: str + node: str port: str - #: Qualified, ``flow.name``. A flow still has to declare it, as for any - #: widget, before a publish to it lands. + #: Qualified, ``flow.name``. message: str #: ``in`` is the device publishing into fluksio, ``out`` is the device #: following the message. direction: Literal["in", "out"] + dtype: str + #: Passed through to the device unread. + config: dict[str, Any] = field(default_factory=dict) + + +class DeviceIo(BaseModel): + """One port as the device reads it: the message it publishes or follows.""" + + port: str + message: str + direction: Literal["in", "out"] config: dict[str, Any] = Field(default_factory=dict) - @field_validator("port") - @classmethod - def _check_port(cls, value: str) -> str: - if not PORT_PATTERN.fullmatch(value): - raise ValueError("Use lowercase letters, digits, underscores and dots") - return value - _check_message = field_validator("message")(validate_qualified) +class DeviceBinding(BaseModel): + """Where a port is wired: the device node holding it.""" + + flow: str + node: str + port: str + message: str + direction: Literal["in", "out"] + + +def bindings_of(flow: FlowDef) -> list[Binding]: + """The device ports this flow's device nodes wire, in document order.""" + found: list[Binding] = [] + for node in flow.nodes: + if node.type != DEVICE_TYPE: + continue + device = str(node.params.get("device") or "") + config = node.params.get("config") + config = config if isinstance(config, dict) else {} + sides = (("in", node.provides), ("out", node.requires)) + for direction, specs in sides: + for spec in specs: + if not spec.name: + continue + own = config.get(spec.port) + found.append( + Binding( + device=device, + flow=flow.name, + node=node.id, + port=spec.port, + message=qualify(flow.name, spec.name), + direction=direction, # type: ignore[arg-type] + dtype=spec.dtype.value, + config=own if isinstance(own, dict) else {}, + ) + ) + return found class DeviceDef(BaseModel): @@ -83,10 +134,6 @@ class DeviceDef(BaseModel): #: touched rather than pointed at. A fact about the screen, not about any #: dashboard on it. touch: bool = False - #: The device's own inputs and outputs. An ``in`` port may publish whether - #: or not a dashboard is locked, since a lock covers the controls on a - #: dashboard and not the sensors on the wall. An ``out`` port only reads. - io: list[DevicePort] = Field(default_factory=list) #: Bumped on every save. A save based on an older one is refused. version: int = 0 @@ -139,6 +186,12 @@ class DeviceStore: # Read-modify-write of the nonce file. An unpair interleaved with # anything else writing it must not be undone by that write. self._state_lock = threading.Lock() + #: Every device's bindings, at the flow store revision they were read + #: at. A device reads its own on every refetch and its credential is + #: checked against them per request. + self._wiring: dict[str, list[Binding]] = {} + self._wiring_at = -1 + self._wiring_lock = threading.Lock() def _file(self, device_id: str) -> Path: return self.root / f"{checked(device_id)}.json" @@ -148,6 +201,78 @@ class DeviceStore: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(device.model_dump_json(indent=2) + "\n") + # -- what the published flows wire ------------------------------------ + + def wiring(self) -> dict[str, list[Binding]]: + """Every device's bound ports, keyed by device id. + + Each list is in one fixed order, port first, so what a device reads is + the same bytes for as long as the flows are the same: the firmware + compares the whole document and rebuilds its screen on any change. + Read again when the flow store commits, a device save included. + """ + with self._wiring_lock: + revision = self.flows.revision + if self._wiring_at != revision: + found: dict[str, list[Binding]] = {} + for flow in self.flows.read_all(): + for binding in bindings_of(flow): + found.setdefault(binding.device, []).append(binding) + for rows in found.values(): + rows.sort( + key=lambda b: (b.port, b.message, b.direction, b.flow, b.node) + ) + self._wiring, self._wiring_at = found, revision + return self._wiring + + def bound(self, device_id: str) -> list[Binding]: + return self.wiring().get(device_id, []) + + def io(self, device_id: str) -> list[DeviceIo]: + """The ports as the device reads them.""" + return [ + DeviceIo( + port=b.port, message=b.message, direction=b.direction, config=b.config + ) + for b in self.bound(device_id) + ] + + def bindings(self, device_id: str) -> list[DeviceBinding]: + """The same ports by the node holding each, for whoever wires them.""" + return sorted( + ( + DeviceBinding( + flow=b.flow, + node=b.node, + port=b.port, + message=b.message, + direction=b.direction, + ) + for b in self.bound(device_id) + ), + key=lambda b: (b.flow, b.node, b.port, b.message), + ) + + def reported_ports(self, device_id: str) -> dict[str, dict[str, Any]] | None: + """The ports the device says it has, by name, or None without a report. + + Read leniently: the report is the device's own word, and a shape this + version does not know is no report rather than an error. + """ + try: + report = self.state(device_id).report + except (OSError, ValueError): + return None + if not isinstance(report, dict) or not isinstance(report.get("ports"), list): + return None + return { + str(port["port"]): port + for port in report["ports"] + if isinstance(port, dict) and port.get("port") + } + + # -- the definitions --------------------------------------------------- + def list(self) -> list[DeviceDef]: found = [] for path in sorted(self.root.glob("*.json")): @@ -266,6 +391,7 @@ class DeviceStore: return devices: list[tuple[DeviceDef, int]] = [] + legacy: dict[str, Any] = {} for raw in panels: try: device = DeviceDef.model_validate({**raw, "version": 1}) @@ -273,11 +399,23 @@ class DeviceStore: logger.error("Not migrating panel %r: %s", raw, exc) continue devices.append((device, int(raw.get("nonce") or 0))) + if raw.get("io"): + legacy[device.id] = raw["io"] with self._lock, self.flows._write_lock: written = [d.id for d, _ in devices if not self._file(d.id).exists()] for device, _ in devices: - if device.id in written: + if device.id in legacy and device.id in written: + # Kept for `move_io`, which wires the rows into flows next. + self._file(device.id).parent.mkdir(parents=True, exist_ok=True) + self._file(device.id).write_text( + json.dumps( + {**device.model_dump(mode="json"), "io": legacy[device.id]}, + indent=2, + ) + + "\n" + ) + elif device.id in written: self._write(device) if written: self.flows._commit("Migrate panels to devices") @@ -295,6 +433,167 @@ class DeviceStore: ", ".join(written) or "all were there already", ) + def move_io(self) -> None: + """Wire each device's own ``io`` rows into flows as device nodes, once. + + Before device nodes, a device carried its ports itself. Each row goes + to the flow declaring its message, the one the message is named after + if that one does, as a port of one device node per device and flow, + in the published document and the draft alike, with the dtype that + flow declares it with and the row's config under ``params.config``. + Every document changed moves one version on, the device files lose + ``io``, and all of it is one commit. + + A row whose message no flow declares is left out and logged, since + there is nowhere to wire it. Running it again finds no ``io`` left, + and a document already holding a port is not given it twice. + """ + with self._lock, self.flows._write_lock: + legacy: list[tuple[str, dict[str, Any]]] = [] + for path in sorted(self.root.glob("*.json")): + try: + raw = json.loads(path.read_text()) + except (OSError, ValueError): + continue + if isinstance(raw, dict) and "io" in raw: + legacy.append((path.stem, raw)) + if not legacy: + return + + docs = _flow_documents(self.flows.root) + changed: set[Path] = set() + moved: list[str] = [] + for device_id, raw in legacy: + title = str(raw.get("title") or "") + for row in raw.get("io") or []: + where = _place(docs, device_id, title, row) + if where is None: + logger.warning( + "Device '%s': left out port %r, since no flow declares " + "its message. Wire it again in a flow.", + device_id, + row, + ) + continue + changed |= where + moved.append(f"{device_id}.{row.get('port')}") + + for path in sorted(changed): + flow = docs[path] + flow.version += 1 + path.write_text(flow.model_dump_json(indent=2) + "\n") + for device_id, raw in legacy: + try: + self._write(DeviceDef.model_validate(raw)) + except ValidationError as exc: + logger.error("Device '%s' is unreadable: %s", device_id, exc) + self.flows._commit("Move device ports into flows") + logger.warning( + "Moved device ports into flows: %s", + ", ".join(moved) or "none could be placed", + ) + + +def _flow_documents(root: Path) -> dict[Path, FlowDef]: + """Every published and draft flow document, by its file.""" + docs: dict[Path, FlowDef] = {} + for path in sorted([*root.glob("*/flow.json"), *root.glob("*/flow.draft.json")]): + if path.parent.name.startswith("_"): + continue + try: + docs[path] = FlowDef.model_validate_json(path.read_text()) + except (OSError, ValidationError): + logger.warning("Skipping unreadable flow document %s", path) + return docs + + +def _declared(flow: FlowDef, message: str) -> MessageSpec | None: + """How this flow declares a qualified message, if it does.""" + for declared in flow.inputs: + if declared.spec.name and qualify(flow.name, declared.spec.name) == message: + return declared.spec + for node in flow.nodes: + for spec in [*node.provides, *node.requires]: + if spec.name and qualify(flow.name, spec.name) == message: + return spec + return None + + +def _place( + docs: dict[Path, FlowDef], device_id: str, title: str, row: Any +) -> set[Path] | None: + """Put one legacy ``io`` row on its flow's device node. The files changed, + or None when no flow declares its message.""" + try: + port = str(row["port"]) + message = validate_qualified(str(row["message"])) + direction = row["direction"] + if not PORT_PATTERN.fullmatch(port) or direction not in ("in", "out"): + raise ValueError + except (KeyError, TypeError, ValueError): + return None + + owner = flow_of(message) + names = sorted({doc.name for doc in docs.values()}, key=lambda n: (n != owner, n)) + for name in names: + found = [ + (path, doc, spec) + for path, doc in docs.items() + if doc.name == name and (spec := _declared(doc, message)) is not None + ] + if found: + break + else: + return None + # The published declaration first: that is what the engine runs. + _, _, declared = sorted(found, key=lambda one: one[0].name != "flow.json")[0] + rest = message[len(name) + 1 :] if message.startswith(f"{name}.") else "" + spec = MessageSpec( + name=rest if rest and "." not in rest else message, + port=port, + dtype=declared.dtype, + item=declared.item, + trigger=direction == "in", + ) + config = row.get("config") if isinstance(row.get("config"), dict) else {} + + targets = [(path, doc) for path, doc in docs.items() if doc.name == name] + # One node per device and flow, under the same id in both documents. + ours = [ + [ + n + for n in doc.nodes + if n.type == DEVICE_TYPE and n.params.get("device") == device_id + ] + for _, doc in targets + ] + existing = next((nodes[0].id for nodes in ours if nodes), None) + taken = {node.id for _, doc in targets for node in doc.nodes} + node_id, count = existing or device_id, 1 + while existing is None and node_id in taken: + count += 1 + node_id = f"{device_id}_{count}" + + changed: set[Path] = set() + for (path, doc), nodes in zip(targets, ours, strict=True): + node = nodes[0] if nodes else None + if node is None: + node = NodeDef( + id=node_id, + type=DEVICE_TYPE, + title=title, + params={"device": device_id, "config": {}}, + ) + doc.nodes.append(node) + side = node.provides if direction == "in" else node.requires + if any(s.port == spec.port and s.name == spec.name for s in side): + continue + side.append(spec) + if config: + node.params.setdefault("config", {})[port] = config + changed.add(path) + return changed + class Presence: """Which devices are connected now, and when each was last heard from. @@ -308,6 +607,17 @@ class Presence: self._sockets: dict[str, int] = {} self._seen: dict[str, float] = {} self._lock = threading.Lock() + #: Told when a device comes or goes: its device nodes, whose health + #: is whether it is there. + self._watchers: dict[str, set[Callable[[bool], None]]] = {} + + def watch(self, device_id: str, callback: Callable[[bool], None]) -> None: + with self._lock: + self._watchers.setdefault(device_id, set()).add(callback) + + def unwatch(self, device_id: str, callback: Callable[[bool], None]) -> None: + with self._lock: + self._watchers.get(device_id, set()).discard(callback) def seen(self, device_id: str) -> None: self._seen[device_id] = time.time() @@ -345,15 +655,20 @@ class Presence: def _announce(self, device_id: str) -> None: last = self.last_seen(device_id) + online = self.online(device_id) event_bus.publish( { "type": "device_status", "device": device_id, - "online": self.online(device_id), + "online": online, "last_seen": last.isoformat() if last else None, "ts": time.time(), } ) + with self._lock: + watchers = list(self._watchers.get(device_id, ())) + for callback in watchers: + callback(online) presence = Presence() @@ -427,19 +742,21 @@ def targets_of(defn: DashboardDef) -> set[str]: return {w.target for w in defn.widgets if w.target} -def io_in(device: DeviceDef | None) -> set[str]: - """The messages this device's own hardware publishes to.""" +def io_in(device: DeviceDef | None, devices: DeviceStore) -> set[str]: + """The messages this device's own ports publish to.""" if device is None: return set() - return {port.message for port in device.io if port.direction == "in"} + return {b.message for b in devices.bound(device.id) if b.direction == "in"} -def messages_for(device: DeviceDef | None, store: DashboardStore) -> set[str]: - """Every message this device's dashboards and its own I/O read or write. +def messages_for( + device: DeviceDef | None, store: DashboardStore, devices: DeviceStore +) -> set[str]: + """Every message this device's dashboards and its own ports read or write. Empty for a device that is gone, which is the same answer as "nothing". """ - own = {port.message for port in device.io} if device else set() + own = {b.message for b in devices.bound(device.id)} if device else set() return own | { name for defn in dashboards_for(device, store) for name in messages_of(defn) } diff --git a/backend/fluksio/flow/nodes/__init__.py b/backend/fluksio/flow/nodes/__init__.py index 1973a54..13bda89 100644 --- a/backend/fluksio/flow/nodes/__init__.py +++ b/backend/fluksio/flow/nodes/__init__.py @@ -7,6 +7,7 @@ Split by the outside world each one talks to. Importing from from fluksio.flow.nodes.base import RESERVED_SETTINGS, Node from fluksio.flow.nodes.debounce import DebounceNode from fluksio.flow.nodes.delay import DelayNode +from fluksio.flow.nodes.device import DeviceNode from fluksio.flow.nodes.exec import ExecNode from fluksio.flow.nodes.file import FileNode from fluksio.flow.nodes.http import HttpNode @@ -23,6 +24,7 @@ __all__ = [ "ChangeNode", "DebounceNode", "DelayNode", + "DeviceNode", "ExecNode", "FileNode", "HttpNode", diff --git a/backend/fluksio/flow/nodes/base.py b/backend/fluksio/flow/nodes/base.py index d8fb937..12ed281 100644 --- a/backend/fluksio/flow/nodes/base.py +++ b/backend/fluksio/flow/nodes/base.py @@ -182,6 +182,11 @@ class Node: # to the device, which at-least-once delivery must not cause. idempotent: bool = True + # A node standing for something outside the engine, which publishes its + # outputs and reads its inputs itself: a paired device. The pipeline never + # runs one, so its inputs wait for nothing. + passive: bool = False + @classmethod def target_key(cls, params: dict[str, Any]) -> str | None: """Which outside thing these parameters point at, if any. diff --git a/backend/fluksio/flow/nodes/device.py b/backend/fluksio/flow/nodes/device.py new file mode 100644 index 0000000..9db020e --- /dev/null +++ b/backend/fluksio/flow/nodes/device.py @@ -0,0 +1,87 @@ +"""Device: a paired device's own ports, as a node in a flow. + +A wall panel's button, its backlight, its microphone. The device publishes on +its ``in`` ports and follows its ``out`` ports, and this node is where a flow +says which message each one is: its outputs are the ``in`` ports, its inputs +the ``out`` ports, each named by the port. + +The engine never runs it. What it provides is published by the device itself, +through its own credential, and what it requires is read by the device off its +socket. The same wiring is what that credential may reach, and what the device +reads back as its ``io`` (see ``fluksio.flow.devices``). +""" + +from __future__ import annotations + +from collections.abc import Iterable +from typing import TYPE_CHECKING, Any + +from pydantic import BaseModel, Field + +from fluksio.flow.devices import PORT_PATTERN, presence +from fluksio.flow.messages import MessageSpec +from fluksio.flow.nodes.base import Node + +if TYPE_CHECKING: + from fastapi import FastAPI + + +def _never(**_: Any) -> None: + """What the node would compute, were it ever run: nothing.""" + + +class DeviceNode(Node): + """A device's ports, wired to messages. Down while the device is away.""" + + passive = True + + class Params(BaseModel): + device: str = Field(default="", description="The device this node stands for.") + config: dict[str, dict[str, Any]] = Field( + default_factory=dict, + description=( + "Per port, settings handed to the device as they are, such as a " + "pin's pull-up or how long the screen stays lit." + ), + ) + + __slots__ = ("cfg",) + + def __init__( + self, + requires: MessageSpec | Iterable[MessageSpec] = (), + provides: MessageSpec | Iterable[MessageSpec] = (), + params: dict[str, Any] | None = None, + name: str | None = None, + ): + self.cfg = self.Params.model_validate(params or {}) + super().__init__( + f=_never, requires=requires, provides=provides, params=params, name=name + ) + for spec in [*self.input_ports, *self.output_ports]: + if not PORT_PATTERN.fullmatch(spec.port): + raise ValueError( + f"'{spec.port}' is not a port name. Use lowercase letters, " + "digits, underscores and dots, such as gpio.4" + ) + + @classmethod + def target_key(cls, params: dict[str, Any]) -> str | None: + """One neuron per device, however many flows wire its ports.""" + return str(params.get("device") or "") or None + + async def start(self, app: FastAPI | None = None) -> None: + if not self.cfg.device: + return + presence.watch(self.cfg.device, self._presence) + self._presence(presence.online(self.cfg.device)) + + async def stop(self, app: FastAPI | None = None) -> None: + if self.cfg.device: + presence.unwatch(self.cfg.device, self._presence) + + def _presence(self, online: bool) -> None: + if online: + self.report_health("ok") + else: + self.report_health("down", "Offline") diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 0060d3c..2fcfaf9 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -27,7 +27,7 @@ from datetime import UTC, datetime from functools import lru_cache from typing import Any, Literal, Protocol -from pydantic import BaseModel, computed_field +from pydantic import BaseModel, model_validator from fluksio.flow import logs from fluksio.flow.artifacts import is_reference @@ -63,6 +63,8 @@ class ValidationIssue(BaseModel): "missing_source", "node_unhealthy", "ignored_device", + "unknown_device", + "device_port", ] message: str flow: str = "" @@ -70,12 +72,15 @@ class ValidationIssue(BaseModel): node: str | None = None port: str | None = None message_name: str | None = None + #: Worth saying, but not a fault — the UI says so in a softer tone. Set + #: by the code for most, by the issue itself where one code covers both. + advisory: bool = False - @computed_field # type: ignore[prop-decorator] - @property - def advisory(self) -> bool: - """Worth saying, but not a fault — the UI says so in a softer tone.""" - return self.code in ADVISORY_ISSUES + @model_validator(mode="after") + def _advisory_by_code(self) -> ValidationIssue: + if self.code in ADVISORY_ISSUES: + self.advisory = True + return self class ValueSource(BaseModel): @@ -229,7 +234,8 @@ def _derive( for node in nodes: for msg in node.provides: produces.setdefault(msg, []).append(node) - for msg in node.requires: + # A passive node is never run, so nothing published wakes it. + for msg in () if node.passive else node.requires: consumes.setdefault(msg, []).append(node) by_id[node.id] = node @@ -557,7 +563,8 @@ class Pipeline: ) for node in nodes: - for msg_name, spec in node.requires.items(): + # A passive node never runs, so it waits for nothing. + for msg_name, spec in () if node.passive else node.requires.items(): if msg_name in produces: # A message a node both reads and writes carries state # between its runs. If the node is the only one writing it, @@ -1369,6 +1376,11 @@ class Pipeline: # not spend the synchronous claim it needs once it may run. if self._gate_blocks(n) or in_degree[n] != 0: continue + if n.passive: + # Something outside publishes for it; a run is not it. + complete(n) + progressed = True + continue if fresh is not None and untouched(n): # Ahead of the readiness check, so an unchanged node # spends neither a delivery stamp nor a synchronous diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index a6e25b2..1cd7a00 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -354,6 +354,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # Once, for an instance from before devices: its panels file moves # into the repository and the nonce file, and is renamed `.migrated`. await run_in_threadpool(devices.migrate, settings.PANELS_FILE) + # Once, for an instance from before device nodes: each device's own + # ports become a device node in the flow declaring their message. + await run_in_threadpool(devices.move_io) app.state.device_store = devices controller.devices = devices # Charts need a deeper series than the default; tell the engine diff --git a/backend/tests/api/routes/test_devices.py b/backend/tests/api/routes/test_devices.py index 1ab87f6..0ada30a 100644 --- a/backend/tests/api/routes/test_devices.py +++ b/backend/tests/api/routes/test_devices.py @@ -7,6 +7,7 @@ from fastapi.testclient import TestClient from fluksio.api.routes import devices as devices_route from fluksio.cloud import config as cloud_config from fluksio.core.config import settings +from fluksio.flow.devices import DeviceState from fluksio.main import app PREFIX = f"{settings.API_V1_STR}/devices" @@ -522,36 +523,115 @@ def test_a_locked_dashboard_entitles_a_panel_to_read_and_not_to_publish( ) -PANEL_IO = [ - { - "port": "gpio.4", - "message": "demo.doorbell", - "direction": "in", - "config": {"pull": "up", "debounce_ms": 50}, - }, - {"port": "backlight", "message": "demo.backlight", "direction": "out"}, -] +FLOWS = f"{settings.API_V1_STR}/flows" -def test_a_panels_io_is_saved_and_read_back( +def _device_node(device: str, node: str = "porch", **ports: list) -> dict: + """A device node: ``provides`` its ``in`` ports, ``requires`` its ``out`` ones.""" + return { + "id": node, + "type": "device", + "params": {"device": device, "config": ports.pop("config", {})}, + "provides": ports.get("provides", []), + "requires": ports.get("requires", []), + } + + +def _publish_flow( + client: TestClient, headers: dict[str, str], name: str, nodes: list, **more +) -> dict: + """Write a flow and publish it, replacing whatever was there.""" + client.delete(f"{FLOWS}/{name}", headers=headers) + saved = client.put( + f"{FLOWS}/{name}", + headers=headers, + json={"name": name, "nodes": nodes, "version": 0, **more}, + ) + assert saved.status_code == 200, saved.text + published = client.post( + f"{FLOWS}/{name}/publish", + headers=headers, + json={"version": saved.json()["definition"]["version"]}, + ) + assert published.status_code == 200, published.text + return published.json() + + +#: A doorbell button and the backlight, as a device node in flow ``porch``. +PORCH_NODE = _device_node( + "porch_io", + provides=[{"name": "doorbell", "port": "gpio.4", "dtype": "bool"}], + requires=[ + {"name": "backlight", "port": "backlight", "dtype": "int", "trigger": False} + ], + config={"gpio.4": {"pull": "up", "debounce_ms": 50}}, +) + + +def test_a_devices_io_comes_from_its_device_nodes( client: TestClient, superuser_token_headers: dict[str, str] ) -> None: - _panels( - client, - superuser_token_headers, - {"panels": [{"id": "porch_io", "io": PANEL_IO}]}, + """What the device reads as ``io`` is what published flows wire, in port order. + + The same bytes for as long as nothing changes: the firmware compares the + whole document and rebuilds its screen on any difference. + """ + _panels(client, superuser_token_headers, {"panels": [{"id": "porch_io"}]}) + _publish_flow(client, superuser_token_headers, "porch", [PORCH_NODE]) + # A draft wires nothing until it is published. + client.put( + f"{FLOWS}/porch_draft", + headers=superuser_token_headers, + json={ + "name": "porch_draft", + "nodes": [ + _device_node( + "porch_io", + provides=[{"name": "mic", "port": "mic", "dtype": "audio"}], + ) + ], + "version": 0, + }, ) - read = client.get(f"{PREFIX}/porch_io", headers=superuser_token_headers).json() - assert read["io"] == [PANEL_IO[0], {**PANEL_IO[1], "config": {}}] + + first = client.get(f"{PREFIX}/porch_io", headers=superuser_token_headers) + read = first.json() + assert read["io"] == [ + { + "port": "backlight", + "message": "porch.backlight", + "direction": "out", + "config": {}, + }, + { + "port": "gpio.4", + "message": "porch.doorbell", + "direction": "in", + "config": {"pull": "up", "debounce_ms": 50}, + }, + ] + assert [(b["flow"], b["node"], b["port"]) for b in read["bindings"]] == [ + ("porch", "porch", "backlight"), + ("porch", "porch", "gpio.4"), + ] + again = client.get(f"{PANELS}/porch_io", headers=superuser_token_headers) + assert again.content == first.content + # The list carries the same, for the Devices tab. + listed = client.get(f"{PREFIX}/", headers=superuser_token_headers).json() + assert next(d for d in listed["data"] if d["id"] == "porch_io")["io"] == read["io"] + for flow in ("porch", "porch_draft"): + client.delete(f"{FLOWS}/{flow}", headers=superuser_token_headers) -def test_a_panels_own_ports_publish_through_a_lock( +def test_a_devices_own_ports_publish_through_a_lock( client: TestClient, superuser_token_headers: dict[str, str] ) -> None: """An ``in`` port is the device's own sensor, which a dashboard lock is not about. An ``out`` port is the device following a message, so it reads and does - not write — and only an ``in`` port's name opens the artifact upload. + not write, and only an ``in`` port's name opens the artifact upload. A + value published on a bound port is the device node's, whatever the device + says it is from. """ _dashboard_with(client, superuser_token_headers, "panel_io_locked", PANEL_WIDGETS) _publish_settings( @@ -560,36 +640,65 @@ def test_a_panels_own_ports_publish_through_a_lock( _panels( client, superuser_token_headers, - {"panels": [{"id": "door", "dashboards": ["panel_io_locked"], "io": PANEL_IO}]}, + { + "panels": [ + {"id": "porch_io", "title": "Porch", "dashboards": ["panel_io_locked"]} + ] + }, ) - panel_headers = _pair(client, superuser_token_headers, "door") + _publish_flow(client, superuser_token_headers, "porch", [PORCH_NODE]) + panel_headers = _pair(client, superuser_token_headers, "porch_io") messages = f"{settings.API_V1_STR}/messages" - # 404: past the gate, refused by an engine no flow here declares it in. + controller = app.state.flow_controller + sources = [] + original = controller.publish_message + + def recorded(name, value, source=None): # type: ignore[no-untyped-def] + sources.append(source) + return original(name, value, source) + + controller.publish_message = recorded + try: + published = client.post( + f"{messages}/porch.doorbell", + headers=panel_headers, + json={"value": True, "source_kind": "panel", "source_id": "panel:x:y"}, + ) + finally: + del controller.publish_message + assert published.status_code == 200, published.text + assert sources[0].model_dump() == { + "kind": "node", + "id": "porch.porch", + "label": "Porch", + "detail": "gpio.4", + } + assert ( client.post( - f"{messages}/demo.doorbell", headers=panel_headers, json={"value": True} - ).status_code - == 404 - ) - assert ( - client.post( - f"{messages}/demo.backlight", headers=panel_headers, json={"value": 50} + f"{messages}/porch.backlight", headers=panel_headers, json={"value": 50} ).status_code == 403 ) assert ( client.get( - f"{messages}/demo.backlight/history", headers=panel_headers + f"{messages}/porch.backlight/history", headers=panel_headers ).status_code == 200 ) + assert ( + client.post( + f"{messages}/porch.unrelated", headers=panel_headers, json={"value": 1} + ).status_code + == 403 + ) artifacts = f"{settings.API_V1_STR}/artifacts" for name, status in ( - ("demo.doorbell", 200), - ("demo.backlight", 403), - ("demo.unrelated", 403), + ("porch.doorbell", 200), + ("porch.backlight", 403), + ("porch.unrelated", 403), ): uploaded = client.put( artifacts, @@ -599,6 +708,129 @@ def test_a_panels_own_ports_publish_through_a_lock( ) assert uploaded.status_code == status, name + # Unwired, the port is nothing the credential reaches any more. + _publish_flow(client, superuser_token_headers, "porch", []) + assert ( + client.post( + f"{messages}/porch.doorbell", headers=panel_headers, json={"value": True} + ).status_code + == 403 + ) + client.delete(f"{FLOWS}/porch", headers=superuser_token_headers) + + +def test_a_port_wired_twice_or_an_unknown_device_is_an_issue( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + _panels(client, superuser_token_headers, {"panels": [{"id": "porch_io"}]}) + _publish_flow(client, superuser_token_headers, "porch", [PORCH_NODE]) + _publish_flow( + client, + superuser_token_headers, + "porch_too", + [ + _device_node( + "porch_io", + provides=[{"name": "bell", "port": "gpio.4", "dtype": "bool"}], + ), + _device_node("ghost", node="ghost"), + ], + ) + + def codes(flow: str) -> set[tuple[str, str | None, bool]]: + issues = client.get(f"{FLOWS}/{flow}", headers=superuser_token_headers).json()[ + "issues" + ] + return { + (i["code"], i["port"], i["advisory"]) + for i in issues + if i["code"] in ("device_port", "unknown_device") + } + + assert codes("porch") == {("device_port", "gpio.4", False)} + assert codes("porch_too") == { + ("device_port", "gpio.4", False), + ("unknown_device", None, False), + } + # Creating the device is enough, nothing has to be published again. + _panels( + client, + superuser_token_headers, + {"panels": [{"id": "porch_io"}, {"id": "ghost"}]}, + ) + assert codes("porch_too") == {("device_port", "gpio.4", False)} + + # With a report a port it does not list is a warning, and one it lists the + # other way round an error. Seen in a draft, which is validated as it is. + store = app.state.device_store + with store._state_lock: + states = store._read_states() + states.setdefault("ghost", DeviceState()).report = { + "schema": 1, + "ports": [{"port": "touch", "direction": "in", "dtype": "bool"}], + } + store._write_states(states) + client.put( + f"{FLOWS}/porch_too", + headers=superuser_token_headers, + json={ + "name": "porch_too", + "version": client.get( + f"{FLOWS}/porch_too", headers=superuser_token_headers + ).json()["definition"]["version"], + "nodes": [ + _device_node( + "ghost", + node="ghost", + provides=[{"name": "relay", "port": "relay", "dtype": "bool"}], + requires=[ + { + "name": "touch", + "port": "touch", + "dtype": "bool", + "trigger": False, + } + ], + ) + ], + }, + ) + assert codes("porch_too") == { + ("device_port", "relay", True), + ("device_port", "touch", False), + } + for flow in ("porch", "porch_too"): + client.delete(f"{FLOWS}/{flow}", headers=superuser_token_headers) + + +def test_publishing_a_flow_with_device_nodes_tells_the_devices( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """An empty dashboard name is what a device refetches its document on.""" + _panels(client, superuser_token_headers, {"panels": [{"id": "porch_io"}]}) + controller = app.state.flow_controller + sent = [] + original = controller._publish + + def recorded(event): # type: ignore[no-untyped-def] + sent.append(event) + original(event) + + controller._publish = recorded + try: + _publish_flow(client, superuser_token_headers, "porch", [PORCH_NODE]) + told = [e for e in sent if e["type"] == "dashboard_changed"] + assert told and told[-1]["dashboard"] == "" + sent.clear() + # A flow with no device in it moves nothing a device reads. + _publish_flow(client, superuser_token_headers, "porch_plain", []) + assert not [e for e in sent if e["type"] == "dashboard_changed"] + client.delete(f"{FLOWS}/porch", headers=superuser_token_headers) + assert [e for e in sent if e["type"] == "dashboard_changed"] + finally: + del controller._publish + client.delete(f"{FLOWS}/porch_plain", headers=superuser_token_headers) + def test_a_panel_may_read_its_dashboards_own_settings( client: TestClient, superuser_token_headers: dict[str, str] diff --git a/backend/tests/flow/test_device_node.py b/backend/tests/flow/test_device_node.py new file mode 100644 index 0000000..534803a --- /dev/null +++ b/backend/tests/flow/test_device_node.py @@ -0,0 +1,94 @@ +"""A device node: never run by the engine, down while its device is away.""" + +import asyncio +from pathlib import Path + +from fluksio.flow.controller import FlowController +from fluksio.flow.devices import presence +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef +from fluksio.flow.store import FlowStore + + +def _controller(tmp_path: Path) -> FlowController: + store = FlowStore(tmp_path / "flows") + store.write_flow( + FlowDef( + name="hall", + inputs=[ + FlowInput(spec=MessageSpec(name="level", dtype=DType.INT), initial=50) + ], + nodes=[ + NodeDef( + id="panel", + type="device", + params={"device": "wall"}, + provides=[ + MessageSpec(name="bell", port="gpio.4", dtype=DType.BOOL) + ], + requires=[ + MessageSpec( + name="level", + port="backlight", + dtype=DType.INT, + trigger=False, + ) + ], + ), + NodeDef( + id="ring", + type="change", + requires=[MessageSpec(name="bell", dtype=DType.BOOL)], + provides=[MessageSpec(name="rung", dtype=DType.BOOL)], + ), + ], + ) + ) + return FlowController(store) + + +def test_the_engine_never_runs_a_device_node(tmp_path: Path) -> None: + controller = _controller(tmp_path) + asyncio.run(controller.reload()) + assert controller.issues == [] + node = controller.get_node("hall.panel") + assert node is not None and node.passive + ran: list[object] = [] + node.f = lambda **kwargs: ran.append(kwargs) # type: ignore[method-assign] + + # The message its out port follows changes, and a manual run of the flow. + controller.publish_message("hall.level", 80) + controller.run_flow("hall") + assert ran == [] + + # What the device publishes on its in port runs what reads it. + controller.publish_message("hall.bell", True) + assert controller.values("hall")["hall.rung"]["value"] is True + asyncio.run(controller._teardown()) + + +def test_a_device_node_is_down_while_its_device_is_away(tmp_path: Path) -> None: + controller = _controller(tmp_path) + + async def scenario() -> list[tuple[str, str | None]]: + await controller.reload() + entry = controller.loaded["hall.panel"] + seen = [(entry.health, entry.health_detail)] + presence.opened("wall") + seen.append((entry.health, entry.health_detail)) + presence.closed("wall") + seen.append((entry.health, entry.health_detail)) + await controller._teardown() + # Stopped, it no longer follows the device. + presence.opened("wall") + seen.append((entry.health, entry.health_detail)) + presence.closed("wall") + return seen + + assert asyncio.run(scenario()) == [ + ("down", "Offline"), + ("ok", None), + ("down", "Offline"), + ("down", "Offline"), + ] + presence.forget("wall") diff --git a/backend/tests/flow/test_device_store.py b/backend/tests/flow/test_device_store.py index e3d96e3..6c76956 100644 --- a/backend/tests/flow/test_device_store.py +++ b/backend/tests/flow/test_device_store.py @@ -56,3 +56,186 @@ def test_panels_move_into_devices_once(tmp_path: Path) -> None: text=True, ).stdout ) + + +def _flow(flows: FlowStore, name: str, draft: dict | None = None, **published) -> None: + """Write a published flow document, and a draft beside it if given.""" + folder = flows.root / name + folder.mkdir(parents=True, exist_ok=True) + (folder / "flow.json").write_text(json.dumps({"name": name, **published})) + if draft is not None: + (folder / "flow.draft.json").write_text(json.dumps({"name": name, **draft})) + + +def test_device_ports_move_into_flows_once(tmp_path: Path) -> None: + flows = FlowStore(tmp_path / "flows") + _flow( + flows, + "home", + version=3, + inputs=[ + {"spec": {"name": "brightness", "dtype": "int"}, "initial": 80}, + ], + nodes=[ + { + "id": "bell", + "type": "python", + "requires": [{"name": "doorbell", "dtype": "bool"}], + } + ], + draft={ + "version": 5, + "inputs": [{"spec": {"name": "brightness", "dtype": "int"}, "initial": 80}], + "nodes": [ + { + "id": "bell", + "type": "python", + "requires": [{"name": "doorbell", "dtype": "bool"}], + }, + {"id": "hall", "type": "inject"}, + ], + }, + ) + # Declares `home.light` from outside: a row on it goes to the owner first, + # and only here when the owner does not declare it. + _flow( + flows, + "lights", + nodes=[ + { + "id": "dim", + "type": "python", + "provides": [{"name": "home.light", "dtype": "float"}], + } + ], + ) + devices = flows.root / "_devices" + devices.mkdir() + (devices / "hall.json").write_text( + json.dumps( + { + "id": "hall", + "title": "Hall", + "version": 4, + "io": [ + { + "port": "backlight", + "message": "home.brightness", + "direction": "out", + }, + { + "port": "gpio.4", + "message": "home.doorbell", + "direction": "in", + "config": {"pull": "up"}, + }, + {"port": "dimmer", "message": "home.light", "direction": "out"}, + {"port": "mic", "message": "home.nowhere", "direction": "in"}, + ], + } + ) + ) + store = DeviceStore(flows, tmp_path / "devices.json") + store.move_io() + + published = flows.read_flow("home") + draft = flows.read_flow("home", draft=True) + assert (published.version, draft.version) == (4, 6) + for doc in (published, draft): + # `hall` is taken by an inject node in the draft, so the id moves on in + # both documents alike. + node = next(n for n in doc.nodes if n.type == "device") + assert node.id == "hall_2" and node.title == "Hall" + assert node.params == {"device": "hall", "config": {"gpio.4": {"pull": "up"}}} + assert [(s.port, s.name, s.dtype.value) for s in node.provides] == [ + ("gpio.4", "doorbell", "bool") + ] + assert [(s.port, s.name, s.dtype.value, s.trigger) for s in node.requires] == [ + ("backlight", "brightness", "int", False) + ] + lights = flows.read_flow("lights") + node = next(n for n in lights.nodes if n.type == "device") + assert node.id == "hall" and lights.version == 2 + assert [(s.port, s.name) for s in node.requires] == [("dimmer", "home.light")] + + # The device lost `io` and kept its version; the row nothing declares is + # gone. What it reads back is the same rows, in port order. + assert "io" not in json.loads((devices / "hall.json").read_text()) + assert store.find("hall").version == 4 # type: ignore[union-attr] + assert [(r.port, r.message, r.direction, r.config) for r in store.io("hall")] == [ + ("backlight", "home.brightness", "out", {}), + ("dimmer", "home.light", "out", {}), + ("gpio.4", "home.doorbell", "in", {"pull": "up"}), + ] + log = subprocess.run( + ["git", "-C", str(flows.root), "log", "--format=%s"], + capture_output=True, + text=True, + ).stdout.splitlines() + assert log[0] == "Move device ports into flows" + + # Nothing left to move, and nothing written twice. + store.move_io() + assert flows.read_flow("home").version == 4 + assert ( + subprocess.run( + ["git", "-C", str(flows.root), "log", "--format=%s"], + capture_output=True, + text=True, + ).stdout.splitlines() + == log + ) + + +def test_a_crash_after_the_flows_were_written_adds_nothing_twice( + tmp_path: Path, +) -> None: + flows = FlowStore(tmp_path / "flows") + _flow(flows, "home", inputs=[{"spec": {"name": "level", "dtype": "int"}}]) + devices = flows.root / "_devices" + devices.mkdir() + device = { + "id": "hall", + "io": [{"port": "backlight", "message": "home.level", "direction": "out"}], + } + (devices / "hall.json").write_text(json.dumps(device)) + store = DeviceStore(flows, tmp_path / "devices.json") + store.move_io() + # As if the device file had not been rewritten before the process died. + (devices / "hall.json").write_text(json.dumps(device)) + store.move_io() + node = next(n for n in flows.read_flow("home").nodes if n.type == "device") + assert len(node.requires) == 1 + assert flows.read_flow("home").version == 2 + + +def test_panels_with_io_reach_the_flows_in_one_start(tmp_path: Path) -> None: + """An instance from before devices: panels.json, then device nodes.""" + flows = FlowStore(tmp_path / "flows") + _flow(flows, "kiosk", inputs=[{"spec": {"name": "brightness", "dtype": "int"}}]) + panels = tmp_path / "panels.json" + panels.write_text( + json.dumps( + { + "panels": [ + { + "id": "knob", + "io": [ + { + "port": "backlight", + "message": "kiosk.brightness", + "direction": "out", + "config": {"wake_s": 30}, + } + ], + } + ] + } + ) + ) + store = DeviceStore(flows, tmp_path / "devices.json") + store.migrate(panels) + store.move_io() + assert [(r.port, r.message, r.config) for r in store.io("knob")] == [ + ("backlight", "kiosk.brightness", {"wake_s": 30}) + ] diff --git a/backend/tests/flow/test_endpoints.py b/backend/tests/flow/test_endpoints.py index 0a5f7c9..f448e81 100644 --- a/backend/tests/flow/test_endpoints.py +++ b/backend/tests/flow/test_endpoints.py @@ -11,7 +11,6 @@ import pytest from fluksio.api.routes.flows import _endpoints from fluksio.flow.controller import FlowController -from fluksio.flow.devices import DeviceDef, DevicePort, DeviceStore from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.schemas import FlowDef, NodeDef from fluksio.flow.store import FlowStore @@ -110,29 +109,3 @@ def test_the_wiring_index_is_read_once_per_store_revision(controller): controller.store.write_flow(FlowDef(name="shed", nodes=[])) _endpoints(controller, controller.store.read_flow("house")) assert reads == 2 - - -def test_a_device_port_is_drawn_on_the_side_it_wires(controller, tmp_path): - controller.devices = DeviceStore(controller.store, tmp_path / "devices.json") - controller.devices.create( - DeviceDef( - id="hall", - title="Hall", - io=[ - DevicePort(port="gpio.4", message="house.temp", direction="in"), - DevicePort(port="backlight", message="house.temp", direction="out"), - DevicePort(port="mic", message="garage.open", direction="in"), - ], - ) - ) - - endpoints = { - one.id: one - for one in _endpoints(controller, controller.store.read_flow("house")) - } - - # The id is what the device publishes under, so its edges pulse for it. - assert endpoints["panel:hall:gpio.4"].provides == ["house.temp"] - assert endpoints["panel:hall:backlight"].requires == ["house.temp"] - assert endpoints["panel:hall:backlight"].detail == "Hall" - assert "panel:hall:mic" not in endpoints diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py index face945..62e8799 100644 --- a/backend/tests/flow/test_node_types.py +++ b/backend/tests/flow/test_node_types.py @@ -100,6 +100,13 @@ FIXTURES: dict[str, dict] = { "requires": [MessageSpec(name="prompt", dtype=DType.STR)], "provides": [MessageSpec(name="kitchen_light", dtype=DType.BOOL)], }, + "device": { + "params": {"device": "hall", "config": {"gpio.4": {"pull": "up"}}}, + "requires": [ + MessageSpec(name="level", port="backlight", dtype=DType.INT, trigger=False) + ], + "provides": [MessageSpec(name="bell", port="gpio.4", dtype=DType.BOOL)], + }, } diff --git a/docs/code/api.md b/docs/code/api.md index ef23ffe..1a27267 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -191,7 +191,7 @@ on its own, so this is for reclaiming space sooner. | `POST` | `/dashboards/{name}/publish`, `/dashboards/{name}/discard`, `/dashboards/{name}/rename` | publish, discard, or rename | | `POST` | `/dashboards/{name}/duplicate` | copy into an unpublished `{name}_copy` | | `GET` | `/devices/` | list devices, with `online` and `last_seen` | -| `POST`, `GET`, `PUT`, `DELETE` | `/devices/{id}` | create, read, save or delete one; a save sends the `version` it read and gets 409 when someone saved first | +| `POST`, `GET`, `PUT`, `DELETE` | `/devices/{id}` | create, read, save or delete one; a save sends the `version` it read and gets 409 when someone saved first. A read adds `io` and `bindings`, its ports as the device nodes of published flows wire them | | `POST` | `/devices/pair` | start device pairing, optionally with `{name, board, firmware}`; unauthenticated | | `GET` | `/devices/pair/{code}` | poll pairing with its `secret`; answers `device` (and `panel` for older firmware); unauthenticated | | `GET` | `/devices/pending` | devices waiting for a code, as they describe themselves | diff --git a/docs/code/nodes.md b/docs/code/nodes.md index fe151b7..5597234 100644 --- a/docs/code/nodes.md +++ b/docs/code/nodes.md @@ -256,8 +256,8 @@ To wire the loop in a flow `home`: 1. Declare a [flow input](../concepts/flows.md#flow-inputs) `heard` of type `audio` with origin `panel`. -2. Under the panel's [I/O](../interface/dashboards.md#the-devices-own-io), wire - `mic` as `in` to `home.heard` and `speaker` as `out` to `home.reply`. +2. Add a [Device node](../interface/dashboards.md#the-devices-own-ports) for + the panel, publishing `mic` to `heard` and following `reply` on `speaker`. 3. Bind the listening node's `audio` to `heard` and its `text` to `said`. 4. Have your own nodes read `said` and publish an answer as a `str`, such as `answer`. diff --git a/docs/concepts/flows.md b/docs/concepts/flows.md index 34b00e9..e0e43c5 100644 --- a/docs/concepts/flows.md +++ b/docs/concepts/flows.md @@ -138,7 +138,9 @@ The canvas reports issues on the affected nodes: | `node_error` | code or configuration could not load | | `unauthenticated_hook` | a webhook or served websocket has no shared secret; advisory | | `ignored_device` | a node other than a Function node names a device, which nothing reads; advisory | -| `node_unhealthy` | a loaded node reports a runtime problem, such as a lost connection | +| `unknown_device` | a Device node names no device, or one that does not exist | +| `device_port` | a device port is wired in two places, or against the device's report the wrong way or type; a port the report does not list is advisory | +| `node_unhealthy` | a loaded node reports a runtime problem, such as a lost connection, or a device being offline | Build issues prevent the affected flow from running. The last three are runtime or advisory issues: other work continues, and node health clears when the diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md index 120d3b3..6ba487a 100644 --- a/docs/interface/dashboards.md +++ b/docs/interface/dashboards.md @@ -468,9 +468,9 @@ device first, the save is refused and the page shows their version. A device credential reads its assigned dashboards, their bound messages and the artifacts they show. It writes only the targets of their controls and query -charts, and the `in` ports of its own [I/O](#the-devices-own-io). A message its -tiles only draw, or a dashboard setting is bound to, stays read only. On a -locked dashboard only the query charts may still publish. +charts, and the messages of its own `in` [ports](#the-devices-own-ports). A +message its tiles only draw, or a dashboard setting is bound to, stays read +only. On a locked dashboard only the query charts may still publish. **Unpair screen** revokes a locally paired credential and keeps the device and its dashboards. Deleting the device removes the assignment and its access, and @@ -495,30 +495,47 @@ work), and a chart of runs, which a device credential may not read. Media tiles play MJPEG streams, pushed frames, images, and H.264 Baseline or MJPEG clips up to 640×360. Re-encode other clips with the [clip recipe](../code/nodes.md#making-clips-playable-on-an-lvgl-panel). -The device decides what its [I/O](#the-devices-own-io) port names +The device decides what its [port](#the-devices-own-ports) names mean. openpanel serves `gpio.`, `backlight`, `touch`, `mic` and `speaker`, lists each one's `config` in its README and ignores any other name. A browser screen serves none of them. The [speech recipe](../code/nodes.md#speech-to-and-from-a-panel) turns `mic` into text and text into `speaker`. -### The device's own I/O +### The device's own ports -A device can also act as a node. Under **I/O**, wire a port on the device to a -message. A new port saves once it has a port name and a message: +A device can also act as a node, with a button on its GPIO, its backlight or +its microphone. Its ports are wired in a flow with a **Device** node: pick the +device, then add each port under **Publishes** (`in`, the device sends to the +message) or **Follows** (`out`, the device reads it). **Use in flow** on the +device's page puts its node in a flow you pick, and **Ports** lists what the +published flows wire, each linked to its node. | Field | Meaning | |---|---| -| `port` | The device's name for it, such as `gpio.4`, `backlight`, `touch`, `mic` or `speaker`. Lowercase letters, digits, `_` and `.` | -| `direction` | `in`: the device publishes to the message. `out`: the device follows it | -| `message` | A qualified `flow.name`, a lowercase flow name, a dot, then letters, digits and underscores. A flow must declare it, as for a control | -| `config` | A JSON object passed to the device unread, such as `{"pull": "up", "debounce_ms": 50}` | +| port | The device's name for it, such as `gpio.4`, `backlight`, `touch`, `mic` or `speaker`. Lowercase letters, digits, `_` and `.` | +| message | The message the port publishes or follows, named like any port's | +| type | The payload type, checked like any message's | +| config | A JSON object passed to the device unread, such as `{"pull": "up", "debounce_ms": 50}` | +Only published flows count. A port wired in two of them, or a device node +naming a device that does not exist, is an error on the canvas. Until a device +reports what it has, a port name is free text; once it has, a port it does not +list is a warning and one wired the wrong way round an error. + +The device reads its ports as `io` on `GET /devices/{id}`, in a fixed order, +and is told to read them again when publishing or deleting a flow changes them. The credential may publish to its `in` messages even when a dashboard is locked, and may upload an artifact named after one (`PUT /artifacts?name=`), which is how a microphone sends a recording. -`out` messages are read only. The flow canvas draws each port as an endpoint, -an `in` port feeding the nodes that consume its message and an `out` port fed -by the nodes that produce it. +A value it publishes there is credited to its device node, so the node's edge +lights up on the canvas. `out` messages are read only. The node is shown down, +*Offline*, while the device is not connected. + +Before device nodes a device carried its ports itself. On its first start an +instance moves each one into the flow that declares its message, one device +node per device and flow, in the published flow and its draft. A port whose +message no flow declares is left out and named in the startup log, so wire it +again by hand. !!! note "If the local link is wrong" diff --git a/docs/interface/flow-editor.md b/docs/interface/flow-editor.md index 28e4d9c..5c86556 100644 --- a/docs/interface/flow-editor.md +++ b/docs/interface/flow-editor.md @@ -13,14 +13,14 @@ below, so a large flow grows downwards in one column. - **Edges** are routed around the nodes, a loop back to an earlier node included, and pulse when values pass through. Click one to inspect its payload. -- **Endpoints** show flow inputs, dashboard widgets, device I/O ports and nodes - in other flows that exchange messages with this flow. A value passing lights +- **Endpoints** show flow inputs, dashboard widgets and nodes in other flows + that exchange messages with this flow. A value passing lights one up in a sweep: left to right when it sends into the flow, right to left when it reads out of it. - **Clicking an endpoint** shows a flow input's type, starting value and - current value, with a pencil that opens the flow settings to edit it. A device - port opens the device in the Devices tab. A node in another flow opens that - flow on the label leading back here, with its edges highlighted. + current value, with a pencil that opens the flow settings to edit it. A node + in another flow opens that flow on the label leading back here, with its + edges highlighted. Values update over a websocket. A connection banner tells you when the editor cannot reach the engine. @@ -37,6 +37,12 @@ and open its settings. payload type, bound message and delivery interval. Outputs also have a streaming toggle. +A **Device** node stands for a paired device instead: pick the device, then +its ports under **Publishes** and **Follows** (see +[the device's own ports](dashboards.md#the-devices-own-ports)). A device that +reports its ports offers them in a list with their type filled in, and the +command palette offers each one as a shortcut, such as *Hallway camera*. + Message fields suggest existing names. Matching an output's message is how you connect an input. diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index b1fbf8d..40b9627 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -386,6 +386,24 @@ They are reported but cannot receive writes unless wired. The chat's [Control tab](../code/agents.md#control) sends commands by hand. +## Devices + +### Device + +**`device`** — a paired device's own ports: what it publishes, what it follows. + +The engine never runs it. Its outputs are the device's `in` ports, published by +the device itself, and its inputs the `out` ports the device follows, which +never wake anything. Each port is named by the device, such as `gpio.4` or +`backlight`. See [the device's own ports](../interface/dashboards.md#the-devices-own-ports). + +| Setting | Default | Notes | +|---|---|---| +| `device` | | the device's id | +| `config` | `{}` | per port, a JSON object handed to the device unread | + +The node is down, *Offline*, while the device is not connected. + ## Connectors Installed connector packages add types through the `fluksio.node_types` diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 88306bb..69e6d65 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -911,6 +911,36 @@ export const DeadLetterSchema = { title: 'DeadLetter' } as const; +export const DeviceBindingSchema = { + properties: { + flow: { + type: 'string', + title: 'Flow' + }, + node: { + type: 'string', + title: 'Node' + }, + port: { + type: 'string', + title: 'Port' + }, + message: { + type: 'string', + title: 'Message' + }, + direction: { + type: 'string', + enum: ['in', 'out'], + title: 'Direction' + } + }, + type: 'object', + required: ['flow', 'node', 'port', 'message', 'direction'], + title: 'DeviceBinding', + description: 'Where a port is wired: the device node holding it.' +} as const; + export const DeviceCreateSchema = { properties: { title: { @@ -946,13 +976,6 @@ export const DeviceDefSchema = { title: 'Touch', default: false }, - io: { - items: { - '$ref': '#/components/schemas/DevicePort' - }, - type: 'array', - title: 'Io' - }, version: { type: 'integer', title: 'Version', @@ -965,7 +988,59 @@ export const DeviceDefSchema = { description: 'One device, and what it shows.' } as const; -export const DevicePortSchema = { +export const DeviceDocumentSchema = { + properties: { + id: { + type: 'string', + title: 'Id' + }, + title: { + type: 'string', + title: 'Title', + default: '' + }, + dashboards: { + items: { + type: 'string' + }, + type: 'array', + title: 'Dashboards' + }, + touch: { + type: 'boolean', + title: 'Touch', + default: false + }, + version: { + type: 'integer', + title: 'Version', + default: 0 + }, + io: { + items: { + '$ref': '#/components/schemas/DeviceIo' + }, + type: 'array', + title: 'Io' + }, + bindings: { + items: { + '$ref': '#/components/schemas/DeviceBinding' + }, + type: 'array', + title: 'Bindings' + } + }, + type: 'object', + required: ['id'], + title: 'DeviceDocument', + description: `A device as it reads itself: what it shows, and what its ports carry. + +\`\`io\`\` is worked out from the device nodes in published flows, in a fixed +order, so an unchanged state reads back as the same bytes.` +} as const; + +export const DeviceIoSchema = { properties: { port: { type: 'string', @@ -988,12 +1063,8 @@ export const DevicePortSchema = { }, type: 'object', required: ['port', 'message', 'direction'], - title: 'DevicePort', - description: `One piece of the device's own hardware, wired to a message. - -A wall panel is a node as well as a screen: a button on its GPIO, a -microphone, a speaker. Which ports exist is the device's business, so the -name is free text and \`\`config\`\` is passed through to it unread.` + title: 'DeviceIo', + description: 'One port as the device reads it: the message it publishes or follows.' } as const; export const DevicePublicSchema = { @@ -1019,18 +1090,25 @@ export const DevicePublicSchema = { title: 'Touch', default: false }, - io: { - items: { - '$ref': '#/components/schemas/DevicePort' - }, - type: 'array', - title: 'Io' - }, version: { type: 'integer', title: 'Version', default: 0 }, + io: { + items: { + '$ref': '#/components/schemas/DeviceIo' + }, + type: 'array', + title: 'Io' + }, + bindings: { + items: { + '$ref': '#/components/schemas/DeviceBinding' + }, + type: 'array', + title: 'Bindings' + }, online: { type: 'boolean', title: 'Online', @@ -1047,6 +1125,18 @@ export const DevicePublicSchema = { } ], title: 'Last Seen' + }, + report: { + anyOf: [ + { + additionalProperties: true, + type: 'object' + }, + { + type: 'null' + } + ], + title: 'Report' } }, type: 'object', @@ -1125,9 +1215,8 @@ export const EndpointSchema = { title: 'Endpoint', description: `Something wired into this flow that is not a node in it. -A dashboard control setting one of its messages, a tile showing one, a -port on a panel's own hardware, or a node in another flow on the far side -of a dotted name. The canvas draws these so a value never appears to come +A dashboard control setting one of its messages, a tile showing one, or a +node in another flow on the far side of a dotted name. The canvas draws these so a value never appears to come from nowhere — or worse, appears to come from whichever node happens to be drawn as a producer.` } as const; @@ -4638,7 +4727,7 @@ export const ValidationIssueSchema = { properties: { code: { type: 'string', - enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source', 'node_unhealthy', 'ignored_device'], + enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source', 'node_unhealthy', 'ignored_device', 'unknown_device', 'device_port'], title: 'Code' }, message: { @@ -4694,12 +4783,11 @@ export const ValidationIssueSchema = { advisory: { type: 'boolean', title: 'Advisory', - description: 'Worth saying, but not a fault — the UI says so in a softer tone.', - readOnly: true + default: false } }, type: 'object', - required: ['code', 'message', 'advisory'], + required: ['code', 'message'], title: 'ValidationIssue', description: 'Something wrong with a flow — a fault, or merely advisory.' } as const; diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 2193e88..de0ce37 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -935,13 +935,14 @@ export class DevicesService { * Read Device * One device, read by a person or by the device holding its credential. * - * The definition alone: a device compares what it reads with what it is - * showing, and a field that changes on every request would rebuild its - * screen each time. Declared after the pairing routes and ``/pending``, so - * neither is mistaken for a device of that name. + * The definition and what the flows wire its ports to, nothing about this + * moment: a device compares what it reads with what it is showing, and a + * field that changes on every request would rebuild its screen each time. + * Declared after the pairing routes and ``/pending``, so neither is mistaken + * for a device of that name. * @param data The data for the request. * @param data.deviceId - * @returns DeviceDef Successful Response + * @returns DeviceDocument Successful Response * @throws ApiError */ public static readDevice(data: DevicesReadDeviceData): CancelablePromise { diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index e5a9035..22273f2 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -229,6 +229,19 @@ export type DeadLetter = { reason: string; }; +/** + * Where a port is wired: the device node holding it. + */ +export type DeviceBinding = { + flow: string; + node: string; + port: string; + message: string; + direction: 'in' | 'out'; +}; + +export type direction = 'in' | 'out'; + export type DeviceCreate = { title?: string; }; @@ -241,18 +254,29 @@ export type DeviceDef = { title?: string; dashboards?: Array<(string)>; touch?: boolean; - io?: Array; version?: number; }; /** - * One piece of the device's own hardware, wired to a message. + * A device as it reads itself: what it shows, and what its ports carry. * - * A wall panel is a node as well as a screen: a button on its GPIO, a - * microphone, a speaker. Which ports exist is the device's business, so the - * name is free text and ``config`` is passed through to it unread. + * ``io`` is worked out from the device nodes in published flows, in a fixed + * order, so an unchanged state reads back as the same bytes. */ -export type DevicePort = { +export type DeviceDocument = { + id: string; + title?: string; + dashboards?: Array<(string)>; + touch?: boolean; + version?: number; + io?: Array; + bindings?: Array; +}; + +/** + * One port as the device reads it: the message it publishes or follows. + */ +export type DeviceIo = { port: string; message: string; direction: 'in' | 'out'; @@ -261,8 +285,6 @@ export type DevicePort = { }; }; -export type direction = 'in' | 'out'; - /** * A device as the Devices tab lists it, with whether it is there. */ @@ -271,10 +293,14 @@ export type DevicePublic = { title?: string; dashboards?: Array<(string)>; touch?: boolean; - io?: Array; version?: number; + io?: Array; + bindings?: Array; online?: boolean; last_seen?: (string | null); + report?: ({ + [key: string]: unknown; +} | null); }; /** @@ -318,9 +344,8 @@ export type DType = 'float' | 'int' | 'str' | 'bool' | 'json' | 'series' | 'reco /** * Something wired into this flow that is not a node in it. * - * A dashboard control setting one of its messages, a tile showing one, a - * port on a panel's own hardware, or a node in another flow on the far side - * of a dotted name. The canvas draws these so a value never appears to come + * A dashboard control setting one of its messages, a tile showing one, or a + * node in another flow on the far side of a dotted name. The canvas draws these so a value never appears to come * from nowhere — or worse, appears to come from whichever node happens to be * drawn as a producer. */ @@ -1489,20 +1514,17 @@ export type ValidationError = { * Something wrong with a flow — a fault, or merely advisory. */ export type ValidationIssue = { - code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_device'; + code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_device' | 'unknown_device' | 'device_port'; message: string; flow?: string; nodes?: Array<(string)>; node?: (string | null); port?: (string | null); message_name?: (string | null); - /** - * Worth saying, but not a fault — the UI says so in a softer tone. - */ - readonly advisory: boolean; + advisory?: boolean; }; -export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_device'; +export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_device' | 'unknown_device' | 'device_port'; export type ValidationResult = { issues?: Array; @@ -1838,7 +1860,7 @@ export type DevicesReadDeviceData = { deviceId: string; }; -export type DevicesReadDeviceResponse = (DeviceDef); +export type DevicesReadDeviceResponse = (DeviceDocument); export type DevicesSaveDeviceData = { deviceId: string; diff --git a/frontend/src/components/Devices/DeviceDetail.tsx b/frontend/src/components/Devices/DeviceDetail.tsx index d864bf1..a98ea60 100644 --- a/frontend/src/components/Devices/DeviceDetail.tsx +++ b/frontend/src/components/Devices/DeviceDetail.tsx @@ -1,26 +1,23 @@ -import { useQuery } from "@tanstack/react-query" -import { Link } from "@tanstack/react-router" -import { ArrowDown, ArrowUp, Plus, X } from "lucide-react" +import { useMutation, useQuery } from "@tanstack/react-query" +import { Link, useNavigate } from "@tanstack/react-router" +import { ArrowDown, ArrowLeft, ArrowRight, ArrowUp, X } from "lucide-react" import { useState } from "react" import { type ApiError, CloudService, type DeviceDef, - type DevicePort, type DevicePublic, DevicesService, + FlowsService, } from "@/client" import { initials, resolveIcon } from "@/components/Dashboard/icons" -import { MessagePicker } from "@/components/Dashboard/panels" import { dashboardsQueryOptions } from "@/components/Dashboard/queries" -import { qualifiedProblem } from "@/components/Flow/deriveEdges" -import { JsonField } from "@/components/Flow/JsonField" +import { flowsQueryOptions } from "@/components/Flow/queries" import { InfoTip, PanelSection } from "@/components/Flow/SidePanel" import { Badge } from "@/components/ui/badge" import { Button } from "@/components/ui/button" import { Input } from "@/components/ui/input" -import { Segmented } from "@/components/ui/segmented" import { Select, SelectContent, @@ -34,6 +31,13 @@ import useAuth from "@/hooks/useAuth" import useCustomToast from "@/hooks/useCustomToast" import { ago, cn } from "@/lib/utils" import { handleError } from "@/utils" +import { + DEVICE_TYPE, + deviceNodeOf, + freeNodeId, + type ReportedPort, + withPort, +} from "./ports" import { devicesQueryOptions, pendingQueryOptions, @@ -47,28 +51,6 @@ const CARD = "rounded-lg border border-border bg-card p-4 shadow-e1" /** As many characters as a device puts on the wall. */ const CODE_LENGTH = 6 -/** Port names the panel firmware knows; `gpio.` takes a pin number. */ -// ponytail: free-text ports until a device reports what it has (PUT /devices/{id}/report) -const PORT_SUGGESTIONS = ["gpio.", "backlight", "touch", "mic", "speaker"] - -const DIRECTIONS = [ - ["in", "In"], - ["out", "Out"], -] as const - -/** What the server accepts as a port name. */ -const portOf = (value: string) => - value.toLowerCase().replace(/[^a-z0-9_.]/g, "") - -/** A new port the server will take: a name, and a message named with its flow. */ -const ready = (port: DevicePort) => - Boolean(port.port) && qualifiedProblem(port.message) === null - -const objectOnly = (value: unknown) => - value && typeof value === "object" && !Array.isArray(value) - ? null - : "Config is a JSON object" - /** "Online", or when it was last heard from. */ export function presenceText(device: DevicePublic) { if (device.online) return "Online" @@ -159,7 +141,7 @@ export function DeviceDetail({ id }: { id: string }) { />
- +
@@ -564,184 +546,131 @@ function Pairing({ } /** - * The device's own inputs and outputs, each wired to a message. - * - * A new port is held here until it has a name and a qualified message, since - * the server refuses a port without either. + * The device's own ports as the published flows wire them: each one, the + * message it publishes or follows, and a link to the node holding it. + * Wiring is done in a flow, so this only reads. */ -function DeviceIo({ +function DevicePorts({ device, canEdit, - onChange, }: { - device: DeviceDef + device: DevicePublic canEdit: boolean - onChange: (changes: Partial) => void }) { - const [draft, setDraft] = useState(null) - const io = device.io ?? [] - const listId = `ports-${device.id}` - - const write = (io: DevicePort[]) => onChange({ io }) - const keepDraft = (next: DevicePort) => { - if (!ready(next)) return setDraft(next) - setDraft(null) - write([...io, next]) - } - + const bindings = device.bindings ?? [] return ( - - {PORT_SUGGESTIONS.map((port) => ( - - {/* `disabled` on a fieldset reaches every control inside it. */} -
- {io.length === 0 && !draft ? ( -

No ports yet.

- ) : null} - {io.map((row, index) => ( - - write(io.map((one, at) => (at === index ? next : one))) - } - onRemove={() => write(io.filter((_, at) => at !== index))} - /> - ))} - {draft ? ( - setDraft({ ...draft, port })} - onRemove={() => setDraft(null)} - /> - ) : null} - {canEdit && !draft ? ( - - ) : null} -
+ {bindings.length === 0 ? ( +

No ports wired yet.

+ ) : ( +
    + {bindings.map((binding) => { + const Way = binding.direction === "in" ? ArrowRight : ArrowLeft + const way = binding.direction === "in" ? "Publishes" : "Follows" + return ( +
  • + + {binding.port} + + + + {binding.message} + +
    • + ) + })} +
+ )} + {canEdit ? : null} ) } -function IoRow({ - row, - listId, - testId, - onChange, - onType, - onRemove, +/** + * Put this device in a flow: its device node there, made when there is none, + * with `port` on it when one is given, and the flow opened on it. A draft, + * so nothing is wired until the flow is published. + */ +export function UseInFlow({ + device, + port, }: { - row: DevicePort - /** The datalist of port names to suggest. */ - listId: string - testId: string - onChange: (next: DevicePort) => void - /** A new port's name as it is typed. A saved port commits on blur instead. */ - onType?: (port: string) => void - onRemove: () => void + device: DevicePublic + port?: ReportedPort }) { - const name = row.port || "new port" - const problem = row.message ? qualifiedProblem(row.message) : null + const navigate = useNavigate() + const { showErrorToast } = useCustomToast() + const { data } = useQuery(flowsQueryOptions()) + const flows = data?.data ?? [] + const place = useMutation({ + mutationFn: async (flowName: string) => { + const { definition } = await FlowsService.readFlow({ name: flowName }) + const nodes = definition.nodes ?? [] + const existing = deviceNodeOf(nodes, device.id) + const node = existing ?? { + id: freeNodeId(nodes, device.id), + type: DEVICE_TYPE, + title: device.title || device.id, + params: { device: device.id, config: {} }, + requires: [], + provides: [], + } + const placed = port ? withPort(node, port) : node + if (placed !== existing) { + await FlowsService.saveFlow({ + name: flowName, + requestBody: { + ...definition, + nodes: existing + ? nodes.map((one) => (one === existing ? placed : one)) + : [...nodes, placed], + }, + }) + } + return { flowName, node: placed.id } + }, + onSuccess: ({ flowName, node }) => + navigate({ + to: "/flows/$flowName", + params: { flowName }, + search: { node }, + }), + onError: (error) => handleError.call(showErrorToast, error as ApiError), + }) + + if (flows.length === 0) return null return ( -
-
- {onType ? ( - onType(portOf(event.currentTarget.value))} - onBlur={() => onChange(row)} - /> - ) : ( - { - const port = portOf(event.currentTarget.value) - event.currentTarget.value = port || row.port - if (port && port !== row.port) onChange({ ...row, port }) - }} - /> - )} -
- - onChange({ - ...row, - direction: direction as DevicePort["direction"], - }) - } - /> -
- -
-
-
- message.writable !== false - : undefined - } - onPick={(message) => onChange({ ...row, message })} - /> -
- - onChange({ ...row, config: config as DevicePort["config"] }) - } - /> -
- {problem ?

{problem}

: null} -
+ ) } diff --git a/frontend/src/components/Devices/ports.ts b/frontend/src/components/Devices/ports.ts new file mode 100644 index 0000000..71a7743 --- /dev/null +++ b/frontend/src/components/Devices/ports.ts @@ -0,0 +1,79 @@ +import type { DevicePublic, DType, MessageSpec, NodeDef_Input } from "@/client" + +/** The node type a device is in a flow. */ +export const DEVICE_TYPE = "device" + +/** Port names the panel firmware knows; `gpio.` takes a pin number. */ +// ponytail: free-text ports until a device reports what it has (D2, PUT /devices/{id}/report) +export const PORT_SUGGESTIONS = [ + "gpio.", + "backlight", + "touch", + "mic", + "speaker", +] + +/** One port as a device reports it. */ +export type ReportedPort = { + port: string + direction: "in" | "out" + dtype?: DType +} + +/** What the server accepts as a port name. */ +export const portOf = (value: string) => + value.toLowerCase().replace(/[^a-z0-9_.]/g, "") + +/** + * The ports a device says it has, or null when it has not said. + * + * Read leniently: the report is the device's own word, and without one ports + * are free text. + */ +export function reportedPorts(device: DevicePublic | undefined) { + const ports = (device?.report as { ports?: unknown } | null | undefined) + ?.ports + if (!Array.isArray(ports)) return null + return ports.filter( + (port): port is ReportedPort => + typeof port?.port === "string" && + (port.direction === "in" || port.direction === "out"), + ) +} + +/** The device node standing for `device` in a flow, if there is one. */ +export function deviceNodeOf(nodes: NodeDef_Input[], device: string) { + return nodes.find( + (node) => node.type === DEVICE_TYPE && node.params?.device === device, + ) +} + +/** + * The node with one more port: an `in` port publishes, so it is an output, + * and an `out` port follows, so it is an input that never wakes anything. + * The message is named after the port until somebody names it. + */ +export function withPort(node: NodeDef_Input, port: ReportedPort) { + const spec: MessageSpec = { + name: port.port.replace(/\./g, "_"), + port: port.port, + dtype: port.dtype ?? "float", + } + if (port.direction === "in") { + if ((node.provides ?? []).some((one) => one.port === port.port)) return node + return { ...node, provides: [...(node.provides ?? []), spec] } + } + if ((node.requires ?? []).some((one) => one.port === port.port)) return node + return { + ...node, + requires: [...(node.requires ?? []), { ...spec, trigger: false }], + } +} + +/** A node id the flow does not have yet, `base` itself when it is free. */ +export function freeNodeId(nodes: NodeDef_Input[], base: string) { + const taken = new Set(nodes.map((node) => node.id)) + let id = base + for (let index = 2; taken.has(id); index += 1) id = `${base}_${index}` + return id +} diff --git a/frontend/src/components/Devices/queries.ts b/frontend/src/components/Devices/queries.ts index 0275b2d..238e10b 100644 --- a/frontend/src/components/Devices/queries.ts +++ b/frontend/src/components/Devices/queries.ts @@ -69,8 +69,16 @@ export function useSaveDevice(id: string) { mutationFn: (changes: Partial) => { const current = stored() if (!current) throw new Error(`No device named ${id}`) - // Presence is the list's, not part of what is saved. - const { online: _online, last_seen: _seen, ...device } = current + // Presence, the report and the wiring are the list's, not part of what + // is saved: the ports are wired in flows. + const { + online: _online, + last_seen: _seen, + report: _report, + io: _io, + bindings: _bindings, + ...device + } = current return DevicesService.saveDevice({ deviceId: id, requestBody: { ...device, ...changes }, diff --git a/frontend/src/components/Flow/CommandPalette.tsx b/frontend/src/components/Flow/CommandPalette.tsx index 8664997..e4dd9ee 100644 --- a/frontend/src/components/Flow/CommandPalette.tsx +++ b/frontend/src/components/Flow/CommandPalette.tsx @@ -1,7 +1,9 @@ import { useQuery } from "@tanstack/react-query" import { useNavigate } from "@tanstack/react-router" -import type { FlowSummary, NodeTypeInfo } from "@/client" +import type { DevicePublic, FlowSummary, NodeTypeInfo } from "@/client" +import { type ReportedPort, reportedPorts } from "@/components/Devices/ports" +import { devicesQueryOptions } from "@/components/Devices/queries" import { CommandDialog, CommandEmpty, @@ -23,6 +25,7 @@ export function CommandPalette({ flows, onAddNode, onAddSharedNode, + onAddDevicePort, onRun, }: { open: boolean @@ -32,6 +35,8 @@ export function CommandPalette({ onAddNode: (type: string) => void /** Absent for an account that may not add code, which a shared node is. */ onAddSharedNode?: (libName: string) => void + /** One port a device reports, onto this flow's node for that device. */ + onAddDevicePort: (device: DevicePublic, port: ReportedPort) => void onRun: () => void }) { const navigate = useNavigate() @@ -39,6 +44,10 @@ export function CommandPalette({ ...libraryQueryOptions(), enabled: open && Boolean(onAddSharedNode), }) + const { data: devicesData } = useQuery({ + ...devicesQueryOptions(), + enabled: open, + }) // Picking an item re-renders the canvas, which can interrupt the dialog's // exit animation and leave its overlay swallowing clicks. Unmounting the @@ -46,6 +55,11 @@ export function CommandPalette({ if (!open) return null const library = onAddSharedNode ? (libraryData ?? []) : [] + // A shortcut per port a device says it has, "Hallway camera". A device that + // has not said is added as a Device node and its ports typed in. + const capabilities = (devicesData?.data ?? []).flatMap((device) => + (reportedPorts(device) ?? []).map((port) => ({ device, port })), + ) const close = (action: () => void) => { onOpenChange(false) @@ -102,6 +116,29 @@ export function CommandPalette({ ) : null} + {capabilities.length > 0 ? ( + + {capabilities.map(({ device, port }) => { + const title = `${device.title || device.id} ${port.port}` + return ( + close(() => onAddDevicePort(device, port))} + > + + {title} + + {port.direction === "in" ? "Publishes" : "Follows"}{" "} + {port.dtype ?? "a value"} + + + + ) + })} + + ) : null} + {flows.length > 0 ? ( {flows.map((flow) => ( diff --git a/frontend/src/components/Flow/EndpointNode.tsx b/frontend/src/components/Flow/EndpointNode.tsx index 78e8346..e76dbb2 100644 --- a/frontend/src/components/Flow/EndpointNode.tsx +++ b/frontend/src/components/Flow/EndpointNode.tsx @@ -1,11 +1,5 @@ import { Handle, type NodeProps, Position } from "@xyflow/react" -import { - LayoutDashboard, - LogIn, - LogOut, - MonitorSmartphone, - Workflow, -} from "lucide-react" +import { LayoutDashboard, LogIn, LogOut, Workflow } from "lucide-react" import { memo, useEffect, useMemo, useRef, useState } from "react" import { useIsMobile } from "@/hooks/useMobile" @@ -16,8 +10,6 @@ import { useLiveValues } from "./liveStore" const KIND_ICONS = { dashboard: LayoutDashboard, - // The same glyph the dashboards screen opens its panels with. - panel: MonitorSmartphone, flow: Workflow, input: LogIn, output: LogOut, diff --git a/frontend/src/components/Flow/FlowEditor.tsx b/frontend/src/components/Flow/FlowEditor.tsx index f9f8e65..c67c0c4 100644 --- a/frontend/src/components/Flow/FlowEditor.tsx +++ b/frontend/src/components/Flow/FlowEditor.tsx @@ -25,6 +25,7 @@ import { motion } from "motion/react" import { useCallback, useEffect, useMemo, useRef, useState } from "react" import { + type DevicePublic, type FlowDef_Input, type FlowDetail, FlowsService, @@ -33,6 +34,12 @@ import { RunsService, } from "@/client" import { DiscardDraftDialog } from "@/components/Common/Dock" +import { + DEVICE_TYPE, + deviceNodeOf, + type ReportedPort, + withPort, +} from "@/components/Devices/ports" import { useTheme } from "@/components/theme-provider" import { Button } from "@/components/ui/button" import { @@ -511,12 +518,6 @@ function FlowEditorInner({ to: "/dashboards/$name", params: { name: (rest ?? "").split(":")[0] }, }) - } else if (kind === "panel") { - // A device's own port: its page in the Devices tab. - navigate({ - to: "/devices/$id", - params: { id: (rest ?? "").split(":")[0] }, - }) } else if (kind === "flow") { navigate({ to: "/flows/$flowName", @@ -940,6 +941,40 @@ function FlowEditorInner({ [commit, definitions], ) + /** + * A port a device reports, on this flow's node for that device. One node per + * device keeps its ports together, so it is made only when there is none. + */ + const addDevicePort = useCallback( + (device: DevicePublic, port: ReportedPort) => { + const existing = deviceNodeOf(definitions, device.id) + if (existing) { + updateNode(withPort(existing, port)) + setSelectedId(existing.id) + return + } + const id = uniqueNodeId(definitions, device.id) + const node = withPort( + { + id, + type: DEVICE_TYPE, + title: device.title || device.id, + params: { device: device.id, config: {} }, + requires: [], + provides: [], + }, + port, + ) + setCanvasNodes([ + ...canvasNodes, + { id, type: "flow", position: { x: 0, y: 0 }, data: {} }, + ]) + commit([...definitions, node]) + setSelectedId(id) + }, + [canvasNodes, commit, definitions, setCanvasNodes, updateNode], + ) + const deleteNodes = useCallback( (ids: string[]) => { const remaining = definitions.filter((node) => !ids.includes(node.id)) @@ -1510,6 +1545,7 @@ function FlowEditorInner({ onAddSharedNode={ mayEditCode ? (libName) => addNode("python", libName) : undefined } + onAddDevicePort={addDevicePort} onRun={startRun} /> diff --git a/frontend/src/components/Flow/FlowNode.tsx b/frontend/src/components/Flow/FlowNode.tsx index c25acaf..8d414f6 100644 --- a/frontend/src/components/Flow/FlowNode.tsx +++ b/frontend/src/components/Flow/FlowNode.tsx @@ -12,6 +12,7 @@ import { Filter, Globe, Merge, + MonitorSmartphone, Play, Plug, Radio, @@ -60,6 +61,8 @@ const NODE_ICONS = { file: FileText, ntfy: Bell, wren: Bird, + // The glyph the Devices tab goes by. + device: MonitorSmartphone, } as const // One dot says everything about a node's state. Idle nodes carry no dot at all, @@ -159,8 +162,7 @@ function FlowNodeComponent({ data, selected }: NodeProps) { const Icon = NODE_ICONS[definition.type as keyof typeof NODE_ICONS] ?? - // A connector's own type cannot be in the map above, and a device is - // not a piece of code. + // A connector's own type cannot be in the map above. (isPlugin ? Plug : Code2) // Whatever is wrong — it failed to load, it failed to run, or the graph diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index 4ec558a..de6e7c6 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -25,6 +25,14 @@ import { type NodeDef_Input, type NodeTypeInfo, } from "@/client" +import { + DEVICE_TYPE, + PORT_SUGGESTIONS, + portOf, + type ReportedPort, + reportedPorts, +} from "@/components/Devices/ports" +import { devicesQueryOptions } from "@/components/Devices/queries" import { Button } from "@/components/ui/button" import { Command, @@ -485,6 +493,322 @@ function PortList({ ) } +const objectOnly = (value: unknown) => + value && typeof value === "object" && !Array.isArray(value) + ? null + : "Config is a JSON object" + +/** + * A device node: the device it stands for, and the message each of its ports + * carries. A port the device publishes is an output here, one it follows an + * input that wakes nothing. With a report the ports are picked from what the + * device has, without one they are typed in. + */ +function DeviceSettings({ + node, + flow, + suggestions, + onChange, +}: { + node: NodeDef_Input + flow: string + suggestions: PortSuggestions + onChange: (next: NodeDef_Input) => void +}) { + const { data } = useQuery(devicesQueryOptions()) + const devices = data?.data ?? [] + const id = String(node.params?.device ?? "") + const device = devices.find((one) => one.id === id) + const reported = reportedPorts(device) + const config = (node.params?.config ?? {}) as Record + const titleOf = (one?: { id: string; title?: string }) => + one ? one.title || one.id : "" + + /** The ports of one side, with the config following a renamed port. */ + const ports = + (side: "provides" | "requires") => + (specs: MessageSpec[], renamed?: [string, string | null]) => { + const next = { ...config } + if (renamed) { + const [from, to] = renamed + if (to && from in next) next[to] = next[from] + delete next[from] + } + onChange({ + ...node, + params: { ...node.params, config: next }, + [side]: specs, + }) + } + /** One port's config. An empty one is no config, and is not kept. */ + const setConfig = (port: string, value: Record) => { + const next = { ...config, [port]: value } + if (Object.keys(value).length === 0) delete next[port] + onChange({ ...node, params: { ...node.params, config: next } }) + } + + return ( + <> +
+ Device + +
+ + + + ) +} + +function DevicePortList({ + title, + help, + direction, + specs, + flow, + config, + reported, + suggestions, + onChange, + onConfig, +}: { + title: string + help: string + direction: "in" | "out" + specs: MessageSpec[] + flow: string + /** Per port, what the device is handed as it is. */ + config: Record + /** What the device says it has, or null when it has not said. */ + reported: ReportedPort[] | null + suggestions: string[] + /** `renamed` is a port taking another name, or dropped (`to` of null). */ + onChange: (next: MessageSpec[], renamed?: [string, string | null]) => void + onConfig: (port: string, next: Record) => void +}) { + const offered = reported?.filter((port) => port.direction === direction) + const listId = `device-ports-${direction}` + // An out port is read when the device asks for it, never a reason to run. + const fresh = (port: ReportedPort | null): MessageSpec => ({ + name: port ? port.port.replace(/\./g, "_") : "", + port: port?.port ?? "", + dtype: port?.dtype ?? "float", + ...(direction === "out" ? { trigger: false } : {}), + }) + const update = (index: number, patch: Partial) => + onChange( + specs.map((spec, at) => (at === index ? { ...spec, ...patch } : spec)), + ) + + return ( +
+
+ + {title} + {help} + + +
+ + {PORT_SUGGESTIONS.map((port) => ( + + {specs.length === 0 ? ( +

No ports yet.

+ ) : null} + {specs.map((spec, index) => { + const port = spec.port ?? "" + const name = port || "new port" + return ( +
+
+ {offered ? ( + + ) : ( + { + const next = portOf(event.currentTarget.value) + event.currentTarget.value = next || port + if (next && next !== port) + onChange( + specs.map((one, at) => + at === index ? { ...one, port: next } : one, + ), + port ? [port, next] : undefined, + ) + }} + /> + )} + + +
+
+ + update(index, { name: next })} + /> + { + if (port) onConfig(port, next as Record) + }} + /> +
+
+ ) + })} +
+ ) +} + /** The value types a free-form setting can hold, and how to read one back. */ const FREE_TYPES = ["text", "number", "on/off", "json"] as const type FreeType = (typeof FREE_TYPES)[number] @@ -1572,37 +1896,48 @@ function PanelBody({ "md:w-[400px] md:shrink-0 md:overflow-y-auto md:border-l md:border-border", )} > - editNode({ ...node, requires })} - /> - { - const [from, to] = portMove.current ?? ["", null] - portMove.current = null - editNode(repoint({ ...node, provides }, from, to)) - }} - streamable - // Only the publishing side names a message; an input is as often - // re-pointed at a different one as it is renamed. - onRenamed={renamePort} - onPortRenamed={(from, to) => { - portMove.current = [from, to] - }} - onRemoved={(spec) => { - portMove.current = [portName(spec), null] - }} - /> - {settingsAreCode ? ( + {node.type === DEVICE_TYPE ? ( + + ) : ( + <> + editNode({ ...node, requires })} + /> + { + const [from, to] = portMove.current ?? ["", null] + portMove.current = null + editNode(repoint({ ...node, provides }, from, to)) + }} + streamable + // Only the publishing side names a message; an input is as often + // re-pointed at a different one as it is renamed. + onRenamed={renamePort} + onPortRenamed={(from, to) => { + portMove.current = [from, to] + }} + onRemoved={(spec) => { + portMove.current = [portName(spec), null] + }} + /> + + )} + {node.type === DEVICE_TYPE ? null : settingsAreCode ? ( /* An exec node's command is its code, so the whole form is. */
{paramsForm} diff --git a/frontend/src/components/Flow/endpoints.ts b/frontend/src/components/Flow/endpoints.ts index c9e562e..aacb7a2 100644 --- a/frontend/src/components/Flow/endpoints.ts +++ b/frontend/src/components/Flow/endpoints.ts @@ -1,14 +1,15 @@ import type { Edge, Node as FlowCanvasNode } from "@xyflow/react" import type { Endpoint, FlowDef_Input, NodeDef_Input } from "@/client" +import { DEVICE_TYPE } from "@/components/Devices/ports" import type { FlowEdgeData } from "./deriveEdges" import { displayName, portOf, qualify } from "./deriveEdges" /** * Canvas elements for things wired into a flow that are not nodes in it. * - * A dashboard control setting one of its messages, a tile showing one, a port - * on a panel's own hardware, or a node in another flow across a dotted name. Without these the canvas shows a + * A dashboard control setting one of its messages, a tile showing one, or a + * node in another flow across a dotted name. Without these the canvas shows a * value changing with nothing to account for it — and pulses whichever node * happens to be drawn as a producer, which did nothing. * @@ -21,7 +22,7 @@ import { displayName, portOf, qualify } from "./deriveEdges" export const ENDPOINT_TYPE = "endpoint" /** The id prefixes an endpoint may carry, none of which is a node id. */ -const PREFIXES = ["dashboard:", "panel:", "flow:", "input:", "output:"] +const PREFIXES = ["dashboard:", "flow:", "input:", "output:"] export type EndpointNodeData = { label: string @@ -58,8 +59,9 @@ export function publisherOf(endpointId: string): string { * rather than from the server, so declaring one shows up on the canvas before * it has been published. * - * An input a dashboard or another flow already accounts for is left out: the - * label exists so a value has a visible origin, and it already has one. + * An input a dashboard, another flow or a device node already accounts for is + * left out: the label exists so a value has a visible origin, and it already + * has one. */ function boundaryOf( doc: FlowDef_Input, @@ -115,7 +117,15 @@ export function deriveEndpoints( flow: string, doc: FlowDef_Input, ): { nodes: FlowCanvasNode[]; edges: Edge[] } { - const accounted = new Set(endpoints.flatMap((one) => one.provides ?? [])) + const accounted = new Set([ + ...endpoints.flatMap((one) => one.provides ?? []), + // A device publishing on a port is where that value comes from. + ...definitions + .filter((node) => node.type === DEVICE_TYPE) + .flatMap((node) => + (node.provides ?? []).map((spec) => qualify(flow, spec.name ?? "")), + ), + ]) const all = [...boundaryOf(doc, flow, accounted), ...endpoints] if (all.length === 0) return { nodes: [], edges: [] }