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

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Melvin Strobl
2026-10-16 16:40:31 +02:00
co-authored by Claude Opus 5.5
parent 3ba407eec6
commit f8011d2a8a
35 changed files with 2195 additions and 494 deletions
+22 -7
View File
@@ -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(
+72 -21
View File
@@ -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}")
+5 -29
View File
@@ -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:
+26 -3
View File
@@ -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",
+1 -4
View File
@@ -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:
+150 -8
View File
@@ -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.
+346 -29
View File
@@ -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)
}
+2
View File
@@ -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",
+5
View File
@@ -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.
+87
View File
@@ -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")
+20 -8
View File
@@ -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
+3
View File
@@ -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
+264 -32
View File
@@ -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]
+94
View File
@@ -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")
+183
View File
@@ -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})
]
-27
View File
@@ -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
+7
View File
@@ -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)],
},
}
+1 -1
View File
@@ -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 |
+2 -2
View File
@@ -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`.
+3 -1
View File
@@ -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
+31 -14
View File
@@ -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.<n>`, `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=<message>`), 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"
+11 -5
View File
@@ -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.
+18
View File
@@ -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`
+116 -28
View File
@@ -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;
+6 -5
View File
@@ -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<DevicesReadDeviceResponse> {
+41 -19
View File
@@ -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<DevicePort>;
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<DeviceIo>;
bindings?: Array<DeviceBinding>;
};
/**
* 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<DevicePort>;
version?: number;
io?: Array<DeviceIo>;
bindings?: Array<DeviceBinding>;
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<ValidationIssue>;
@@ -1838,7 +1860,7 @@ export type DevicesReadDeviceData = {
deviceId: string;
};
export type DevicesReadDeviceResponse = (DeviceDef);
export type DevicesReadDeviceResponse = (DeviceDocument);
export type DevicesSaveDeviceData = {
deviceId: string;
+123 -194
View File
@@ -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 }) {
/>
</section>
<section className={CARD}>
<DeviceIo device={device} canEdit={canEdit} onChange={write} />
<DevicePorts device={device} canEdit={canEdit} />
</section>
</div>
</div>
@@ -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<DeviceDef>) => void
}) {
const [draft, setDraft] = useState<DevicePort | null>(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 (
<PanelSection
title="I/O"
help="Hardware on the device itself, such as a GPIO pin or its microphone. In publishes to the message, Out follows it, and Config goes to the device as it is."
title="Ports"
help="Hardware on the device itself, such as a GPIO pin or its backlight. A device node in a flow wires each port to a message."
>
<datalist id={listId}>
{PORT_SUGGESTIONS.map((port) => (
<option key={port} value={port} />
))}
</datalist>
{/* `disabled` on a fieldset reaches every control inside it. */}
<fieldset disabled={!canEdit} className="grid min-w-0 gap-4">
{io.length === 0 && !draft ? (
<p className="text-sm text-muted-foreground">No ports yet.</p>
) : null}
{io.map((row, index) => (
<IoRow
// Ports have no id, and two may share a name until one is renamed.
key={index}
row={row}
listId={listId}
testId={`io-${device.id}-${index}`}
onChange={(next) =>
write(io.map((one, at) => (at === index ? next : one)))
}
onRemove={() => write(io.filter((_, at) => at !== index))}
/>
))}
{draft ? (
<IoRow
row={draft}
listId={listId}
testId={`io-${device.id}-new`}
onChange={keepDraft}
onType={(port) => setDraft({ ...draft, port })}
onRemove={() => setDraft(null)}
/>
) : null}
{canEdit && !draft ? (
<Button
variant="outline"
size="sm-touch"
className="justify-self-start"
data-testid={`add-port-${device.id}`}
onClick={() => setDraft({ port: "", message: "", direction: "in" })}
>
<Plus />
Add port
</Button>
) : null}
</fieldset>
{bindings.length === 0 ? (
<p className="text-sm text-muted-foreground">No ports wired yet.</p>
) : (
<ul className="grid gap-1" data-testid="device-bindings">
{bindings.map((binding) => {
const Way = binding.direction === "in" ? ArrowRight : ArrowLeft
const way = binding.direction === "in" ? "Publishes" : "Follows"
return (
<li
key={`${binding.flow}.${binding.node}.${binding.port}.${binding.message}`}
className="flex min-w-0 items-center gap-2 text-sm"
>
<span className="w-24 shrink-0 truncate font-mono">
{binding.port}
</span>
<Way
role="img"
aria-label={way}
className="size-4 shrink-0 text-muted-foreground"
/>
<Link
to="/flows/$flowName"
params={{ flowName: binding.flow }}
search={{ node: binding.node }}
className="min-w-0 flex-1 truncate font-mono hover:underline"
>
{binding.message}
</Link>
</li>
)
})}
</ul>
)}
{canEdit ? <UseInFlow device={device} /> : null}
</PanelSection>
)
}
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 (
<div className="grid gap-2" data-testid={testId}>
<div className="flex items-center gap-2">
{onType ? (
<Input
value={row.port}
placeholder="Port, e.g. gpio.4"
aria-label={`Port of ${name}`}
autoComplete="off"
list={listId}
onChange={(event) => onType(portOf(event.currentTarget.value))}
onBlur={() => onChange(row)}
/>
) : (
<Input
// Remounted with whatever was saved, so a refused name snaps back.
key={row.port}
defaultValue={row.port}
placeholder="Port, e.g. gpio.4"
aria-label={`Port of ${name}`}
autoComplete="off"
list={listId}
onBlur={(event) => {
const port = portOf(event.currentTarget.value)
event.currentTarget.value = port || row.port
if (port && port !== row.port) onChange({ ...row, port })
}}
/>
)}
<div className="shrink-0">
<Segmented
value={row.direction}
options={DIRECTIONS}
label={`Direction of ${name}`}
onChange={(direction) =>
onChange({
...row,
direction: direction as DevicePort["direction"],
})
}
/>
</div>
<Button
variant="ghost"
size="icon-touch"
className="shrink-0 text-muted-foreground"
aria-label={`Remove ${name}`}
title={`Remove ${name}`}
onClick={onRemove}
>
<X />
</Button>
</div>
<div className="flex items-center gap-2">
<div className="min-w-0 flex-1">
<MessagePicker
value={row.message}
label={`Message of ${name}`}
hideLabel
// A port the device publishes from needs a message a flow takes in.
filter={
row.direction === "in"
? (message) => message.writable !== false
: undefined
}
onPick={(message) => onChange({ ...row, message })}
/>
</div>
<JsonField
value={row.config ?? {}}
label={`Config of ${name}`}
check={objectOnly}
className="h-9 w-28 shrink-0"
onChange={(config) =>
onChange({ ...row, config: config as DevicePort["config"] })
}
/>
</div>
{problem ? <p className="text-xs text-destructive">{problem}</p> : null}
</div>
<Select value="" onValueChange={(flowName) => place.mutate(flowName)}>
<SelectTrigger
className="justify-self-start"
aria-label={`Use ${device.title || device.id} in a flow`}
disabled={place.isPending}
data-testid="use-in-flow"
>
<SelectValue placeholder="Use in flow" />
</SelectTrigger>
<SelectContent>
{flows.map((flow) => (
<SelectItem key={flow.name} value={flow.name}>
{flow.title || flow.name}
</SelectItem>
))}
</SelectContent>
</Select>
)
}
+79
View File
@@ -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
}
+10 -2
View File
@@ -69,8 +69,16 @@ export function useSaveDevice(id: string) {
mutationFn: (changes: Partial<DeviceDef>) => {
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 },
@@ -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({
</CommandGroup>
) : null}
{capabilities.length > 0 ? (
<CommandGroup heading="Devices">
{capabilities.map(({ device, port }) => {
const title = `${device.title || device.id} ${port.port}`
return (
<CommandItem
key={`${device.id}:${port.port}`}
value={`device ${title}`}
onSelect={() => close(() => onAddDevicePort(device, port))}
>
<span className="flex flex-col">
<span>{title}</span>
<span className="text-xs text-muted-foreground">
{port.direction === "in" ? "Publishes" : "Follows"}{" "}
{port.dtype ?? "a value"}
</span>
</span>
</CommandItem>
)
})}
</CommandGroup>
) : null}
{flows.length > 0 ? (
<CommandGroup heading="Flows">
{flows.map((flow) => (
@@ -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,
+42 -6
View File
@@ -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}
/>
+4 -2
View File
@@ -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
+366 -31
View File
@@ -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<string, unknown>
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<string, unknown>) => {
const next = { ...config, [port]: value }
if (Object.keys(value).length === 0) delete next[port]
onChange({ ...node, params: { ...node.params, config: next } })
}
return (
<>
<div className="grid gap-1.5">
<FieldLabel htmlFor="device-node-device">Device</FieldLabel>
<Select
value={id || undefined}
onValueChange={(next) => {
const picked = devices.find((one) => one.id === next)
// A title that only ever named the device follows it.
const title =
!node.title || node.title === (titleOf(device) || id)
? titleOf(picked)
: node.title
onChange({
...node,
title,
params: { ...node.params, device: next },
})
}}
>
<SelectTrigger
id="device-node-device"
className="!h-8 w-full text-sm"
data-testid="device-node-device"
>
<SelectValue placeholder="Pick a device" />
</SelectTrigger>
<SelectContent>
{devices.map((one) => (
<SelectItem key={one.id} value={one.id}>
{titleOf(one)}
</SelectItem>
))}
</SelectContent>
</Select>
</div>
<DevicePortList
title="Publishes"
help="Ports the device sends from, such as a button. Each publishes the message beside it."
direction="in"
specs={node.provides ?? []}
flow={flow}
config={config}
reported={reported}
suggestions={suggestions.provides}
onChange={ports("provides")}
onConfig={setConfig}
/>
<DevicePortList
title="Follows"
help="Ports the device drives, such as its backlight. Each follows the message beside it."
direction="out"
specs={node.requires ?? []}
flow={flow}
config={config}
reported={reported}
suggestions={suggestions.consumes}
onChange={ports("requires")}
onConfig={setConfig}
/>
</>
)
}
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<string, unknown>
/** 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<string, unknown>) => 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<MessageSpec>) =>
onChange(
specs.map((spec, at) => (at === index ? { ...spec, ...patch } : spec)),
)
return (
<div
className="grid min-w-0 gap-3"
data-testid={`device-ports-${direction}`}
>
<div className="flex items-center justify-between">
<span className="flex items-center gap-1">
<span className={SECTION}>{title}</span>
<InfoTip label={title}>{help}</InfoTip>
</span>
<Button
variant="ghost"
size="sm"
className="h-7 text-xs text-muted-foreground"
onClick={() => onChange([...specs, fresh(null)])}
>
Add
</Button>
</div>
<datalist id={listId}>
{PORT_SUGGESTIONS.map((port) => (
<option key={port} value={port} />
))}
</datalist>
{specs.length === 0 ? (
<p className="text-sm text-muted-foreground">No ports yet.</p>
) : null}
{specs.map((spec, index) => {
const port = spec.port ?? ""
const name = port || "new port"
return (
<div key={`port-${index}`} className="grid min-w-0 gap-1.5">
<div className="flex min-w-0 items-center gap-1.5">
{offered ? (
<Select
value={port || undefined}
onValueChange={(next) => {
// The device says what it carries, so the type follows.
const picked = offered.find((one) => one.port === next)
onChange(
specs.map((one, at) =>
at === index
? {
...one,
port: next,
dtype: picked?.dtype ?? one.dtype,
}
: one,
),
port ? [port, next] : undefined,
)
}}
>
<SelectTrigger
className="!h-8 min-w-0 flex-1 font-mono text-sm"
aria-label={`Port of ${name}`}
>
<SelectValue placeholder="Pick a port" />
</SelectTrigger>
<SelectContent>
{/* One the report has dropped stays pickable as it is. */}
{port && !offered.some((one) => one.port === port) ? (
<SelectItem value={port}>{port}</SelectItem>
) : null}
{offered.map((one) => (
<SelectItem key={one.port} value={one.port}>
{one.port}
</SelectItem>
))}
</SelectContent>
</Select>
) : (
<Input
// Remounted with whatever was saved, so a refused name snaps back.
key={port}
defaultValue={port}
placeholder="Port, e.g. gpio.4"
aria-label={`Port of ${name}`}
autoComplete="off"
list={listId}
className="h-8 min-w-0 flex-1 font-mono text-sm"
onBlur={(event) => {
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,
)
}}
/>
)}
<Select
value={spec.dtype ?? "float"}
onValueChange={(value) =>
update(index, { dtype: value as DType })
}
>
<SelectTrigger
className="!h-8 w-[104px] text-sm"
aria-label={`Type of ${name}`}
>
<SelectValue />
</SelectTrigger>
<SelectContent>
{DTYPES.map((dtype) => (
<SelectItem key={dtype} value={dtype}>
{dtype}
</SelectItem>
))}
</SelectContent>
</Select>
<Button
variant="ghost"
size="icon-sm"
className="text-muted-foreground"
aria-label={`Remove ${name}`}
onClick={() =>
onChange(
specs.filter((_, at) => at !== index),
port ? [port, null] : undefined,
)
}
>
<X />
</Button>
</div>
<div className="flex min-w-0 items-center gap-1.5">
<CornerDownRight
aria-hidden
className="size-4 shrink-0 text-muted-foreground"
/>
<MessageNameInput
value={spec.name ?? ""}
suggestions={suggestions}
placeholder={`message in ${flow}`}
autoFocus={false}
onChange={(next) => update(index, { name: next })}
/>
<JsonField
value={config[port] ?? {}}
label={`Config of ${name}`}
check={objectOnly}
className="h-8 w-24 shrink-0"
onChange={(next) => {
if (port) onConfig(port, next as Record<string, unknown>)
}}
/>
</div>
</div>
)
})}
</div>
)
}
/** 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",
)}
>
<PortList
title="Consumes"
specs={node.requires ?? []}
flow={flow}
emptyHint="Nothing yet. Add a message this node reads."
suggestions={suggestions.consumes}
onChange={(requires) => editNode({ ...node, requires })}
/>
<PortList
title="Provides"
specs={node.provides ?? []}
flow={flow}
emptyHint="Nothing yet. Add a message this node publishes."
suggestions={suggestions.provides}
onChange={(provides) => {
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 ? (
<DeviceSettings
node={node}
flow={flow}
suggestions={suggestions}
onChange={editNode}
/>
) : (
<>
<PortList
title="Consumes"
specs={node.requires ?? []}
flow={flow}
emptyHint="Nothing yet. Add a message this node reads."
suggestions={suggestions.consumes}
onChange={(requires) => editNode({ ...node, requires })}
/>
<PortList
title="Provides"
specs={node.provides ?? []}
flow={flow}
emptyHint="Nothing yet. Add a message this node publishes."
suggestions={suggestions.provides}
onChange={(provides) => {
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. */
<fieldset disabled className="grid min-w-0 gap-2">
{paramsForm}
+16 -6
View File
@@ -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: [] }