Melvin StroblandClaude Opus 5.5
ad76200d3d
A device reports its board, screen, camera and ports from its own credential (16 KB, schema 1, replacing the last report), and the flows are validated against it again. The node panel turns a reported port's config schema into a form, a reported screen is a canvas preset and a touch screen's first report turns Touch friendly on. An offline device node is degraded: amber on the canvas, a warning on Home, never a fault. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0138N7k5PT4dcR1iEqXaaQ9k
838 lines
31 KiB
Python
838 lines
31 KiB
Python
"""Devices: the screens and boards paired with this instance.
|
|
|
|
A device is a wall panel, a knob, a camera board: something without a keyboard
|
|
that pairs with a code, holds a scoped credential and shows an ordered set of
|
|
whole dashboards. A hallway tablet and a workshop one carry different sets, the
|
|
same dashboard may hang on both, and each dashboard keeps its own canvas, draft
|
|
and version.
|
|
|
|
The definition is stored in the flow repository as ``_devices/<id>.json`` and
|
|
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
|
|
|
|
import ipaddress
|
|
import json
|
|
import logging
|
|
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
|
|
|
|
from pydantic import BaseModel, Field, TypeAdapter, ValidationError, field_validator
|
|
|
|
from fluksio.flow.dashboards import DashboardDef, DashboardNotFound, DashboardStore
|
|
from fluksio.flow.events import event_bus
|
|
from fluksio.flow.messages import (
|
|
DType,
|
|
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__)
|
|
|
|
#: Sibling of the dashboards, and likewise not a flow.
|
|
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"
|
|
|
|
|
|
@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``.
|
|
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)
|
|
|
|
|
|
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):
|
|
"""One device, and what it shows."""
|
|
|
|
id: str
|
|
title: str = ""
|
|
#: Ordered. The first one is what the device opens after pairing, and the
|
|
#: rail follows this order. A name that no longer resolves is a dashboard
|
|
#: someone deleted, and the device skips it.
|
|
dashboards: list[str] = Field(default_factory=list)
|
|
#: Bigger controls, a bigger rail and no hover states, for a screen that is
|
|
#: touched rather than pointed at. A fact about the screen, not about any
|
|
#: dashboard on it.
|
|
touch: bool = False
|
|
#: Bumped on every save. A save based on an older one is refused.
|
|
version: int = 0
|
|
|
|
@field_validator("id")
|
|
@classmethod
|
|
def _check_id(cls, value: str) -> str:
|
|
return _validate_name(value)
|
|
|
|
|
|
class DeviceState(BaseModel):
|
|
"""What this instance keeps about a device outside the flow repository."""
|
|
|
|
#: Which generation of credential the device honours. A token names the
|
|
#: nonce it was minted at, so bumping this refuses the screen hanging there
|
|
#: and leaves the device and its dashboards as they are.
|
|
nonce: int = 0
|
|
#: What the device says it has, written by the device itself.
|
|
report: dict[str, Any] | None = None
|
|
reported_at: datetime | None = None
|
|
|
|
|
|
_STATES = TypeAdapter(dict[str, DeviceState])
|
|
|
|
|
|
class ReportDisplay(BaseModel):
|
|
"""The device's screen, in pixels."""
|
|
|
|
w: int = Field(ge=0)
|
|
h: int = Field(ge=0)
|
|
shape: Literal["rect", "round"] = "rect"
|
|
touch: bool = False
|
|
|
|
|
|
class ReportPort(BaseModel):
|
|
"""One port the device has, and how it may be wired."""
|
|
|
|
port: str = Field(pattern=PORT_PATTERN.pattern)
|
|
#: Its default direction: ``in`` publishes, ``out`` follows.
|
|
direction: Literal["in", "out"]
|
|
dtype: DType
|
|
#: A JSON schema of the ``config`` the device reads for this port.
|
|
config: dict[str, Any] | None = None
|
|
#: Sends only while somebody wants it, such as a camera.
|
|
on_demand: bool | None = None
|
|
#: The directions it may be wired in, when it may be either: a GPIO pin.
|
|
modes: list[Literal["in", "out"]] | None = None
|
|
|
|
|
|
class ReportCamera(BaseModel):
|
|
"""The camera sensor the device found, and what it delivers."""
|
|
|
|
sensor: str = ""
|
|
w: int = 0
|
|
h: int = 0
|
|
fps: int = 0
|
|
|
|
|
|
class DeviceReport(BaseModel):
|
|
"""What a device says it has, version 1.
|
|
|
|
Keys this version does not know are dropped, so a newer firmware still
|
|
reports to an older instance. The whole report replaces the one before.
|
|
"""
|
|
|
|
version: Literal[1] = Field(alias="schema")
|
|
platform: str = ""
|
|
board: str = ""
|
|
firmware: str = ""
|
|
display: ReportDisplay | None = None
|
|
ports: list[ReportPort] = Field(default_factory=list)
|
|
camera: ReportCamera | None = None
|
|
#: The build's features, true when there: ``{"camera": true}``.
|
|
features: dict[str, bool] = Field(default_factory=dict)
|
|
|
|
def stored(self) -> dict[str, Any]:
|
|
"""As ``devices.json`` keeps it: the device's own keys, nothing unset."""
|
|
return self.model_dump(mode="json", by_alias=True, exclude_none=True)
|
|
|
|
|
|
class DeviceNotFound(KeyError):
|
|
def __init__(self, device_id: str) -> None:
|
|
super().__init__(device_id)
|
|
self.id = device_id
|
|
|
|
|
|
class DeviceExists(ValueError):
|
|
def __init__(self, device_id: str) -> None:
|
|
super().__init__(device_id)
|
|
self.id = device_id
|
|
|
|
|
|
class DeviceStore:
|
|
"""Device definitions in the flow repository, their nonces beside it.
|
|
|
|
Shares the flow store's write lock and commit, as the dashboards do, so a
|
|
device save cannot interleave with a flow or dashboard save.
|
|
"""
|
|
|
|
def __init__(self, flows: FlowStore, state_file: Path) -> None:
|
|
self.flows = flows
|
|
self.root = flows.root / DEVICE_DIR
|
|
self.state_file = state_file
|
|
# Read-modify-write of a device's version.
|
|
self._lock = threading.Lock()
|
|
# 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"
|
|
|
|
def _write(self, device: DeviceDef) -> None:
|
|
path = self._file(device.id)
|
|
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")):
|
|
try:
|
|
found.append(DeviceDef.model_validate_json(path.read_text()))
|
|
except (OSError, ValidationError):
|
|
logger.warning("Skipping unreadable device %s", path.name)
|
|
return found
|
|
|
|
def find(self, device_id: str) -> DeviceDef | None:
|
|
"""The device by that id, or None if it is gone.
|
|
|
|
Read from disk on every call: this is what makes deleting a device
|
|
revoke its credential, so it has to see the current file.
|
|
"""
|
|
try:
|
|
path = self._file(device_id)
|
|
except InvalidName:
|
|
return None
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return DeviceDef.model_validate_json(path.read_text())
|
|
except (OSError, ValidationError):
|
|
return None
|
|
|
|
def create(self, device: DeviceDef) -> DeviceDef:
|
|
with self._lock, self.flows._write_lock:
|
|
if self._file(device.id).exists():
|
|
raise DeviceExists(device.id)
|
|
saved = device.model_copy(update={"version": 1})
|
|
self._write(saved)
|
|
self.flows._commit(f"Add device '{device.id}'")
|
|
return saved
|
|
|
|
def write(self, device: DeviceDef, base_version: int) -> DeviceDef:
|
|
"""Save a device, refusing a save someone else has moved past."""
|
|
with self._lock, self.flows._write_lock:
|
|
current = self.find(device.id)
|
|
if current is None:
|
|
raise DeviceNotFound(device.id)
|
|
if base_version != current.version:
|
|
raise StaleVersion(device.id, current.version)
|
|
saved = device.model_copy(update={"version": current.version + 1})
|
|
self._write(saved)
|
|
self.flows._commit(f"Save device '{device.id}'")
|
|
return saved
|
|
|
|
def delete(self, device_id: str) -> None:
|
|
"""Remove the device, and refuse every credential minted for it.
|
|
|
|
The nonce is bumped rather than dropped: a device created again under
|
|
the same id, or brought back by a revert, must not take the old
|
|
screen's credential back with it.
|
|
"""
|
|
with self.flows._write_lock:
|
|
path = self._file(device_id)
|
|
if not path.exists():
|
|
raise DeviceNotFound(device_id)
|
|
path.unlink()
|
|
self.flows._commit(f"Delete device '{device_id}'")
|
|
self.bump(device_id)
|
|
presence.forget(device_id)
|
|
|
|
# -- outside git -------------------------------------------------------
|
|
|
|
def _read_states(self) -> dict[str, DeviceState]:
|
|
"""Raises on a file that does not parse, rather than reading it as empty.
|
|
|
|
Empty would put every nonce back to 0, which honours again every
|
|
credential an unpair refused.
|
|
"""
|
|
if not self.state_file.exists():
|
|
return {}
|
|
return _STATES.validate_json(self.state_file.read_bytes())
|
|
|
|
def _write_states(self, states: dict[str, DeviceState]) -> None:
|
|
# Through a temporary file, so a crash mid-write cannot leave a torn
|
|
# file behind.
|
|
self.state_file.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = self.state_file.with_name(self.state_file.name + ".tmp")
|
|
tmp.write_bytes(_STATES.dump_json(states, indent=2))
|
|
os.replace(tmp, self.state_file)
|
|
|
|
def state(self, device_id: str) -> DeviceState:
|
|
return self._read_states().get(device_id, DeviceState())
|
|
|
|
def nonce(self, device_id: str) -> int:
|
|
return self.state(device_id).nonce
|
|
|
|
def bump(self, device_id: str) -> int:
|
|
"""Refuse the credential this device holds now. Returns the new nonce."""
|
|
with self._state_lock:
|
|
states = self._read_states()
|
|
state = states.setdefault(device_id, DeviceState())
|
|
state.nonce += 1
|
|
self._write_states(states)
|
|
return state.nonce
|
|
|
|
def report(self, device_id: str, report: DeviceReport) -> bool:
|
|
"""Keep what the device says it has, in place of what it said before.
|
|
|
|
Returns whether it is the first report this device made.
|
|
"""
|
|
with self._state_lock:
|
|
states = self._read_states()
|
|
state = states.setdefault(device_id, DeviceState())
|
|
first = state.report is None
|
|
state.report = report.stored()
|
|
state.reported_at = datetime.now(UTC)
|
|
self._write_states(states)
|
|
return first
|
|
|
|
# -- the panels file of an older instance -------------------------------
|
|
|
|
def migrate(self, panels_file: Path) -> None:
|
|
"""Move ``panels.json`` into the store and the nonce file, once.
|
|
|
|
Each panel becomes ``_devices/<id>.json`` at version 1, in one commit,
|
|
and its nonce goes to ``devices.json`` (the higher one wins, should
|
|
both have it). The file is then renamed ``.migrated``. Running it again
|
|
after a crash part-way skips the devices already written.
|
|
"""
|
|
if not panels_file.exists():
|
|
return
|
|
try:
|
|
panels = json.loads(panels_file.read_text()).get("panels", [])
|
|
except (OSError, ValueError, AttributeError) as exc:
|
|
logger.error("Could not read %s to migrate it: %s", panels_file, exc)
|
|
return
|
|
|
|
devices: list[tuple[DeviceDef, int]] = []
|
|
legacy: dict[str, Any] = {}
|
|
for raw in panels:
|
|
try:
|
|
device = DeviceDef.model_validate({**raw, "version": 1})
|
|
except (ValidationError, TypeError) as exc:
|
|
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 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")
|
|
with self._state_lock:
|
|
states = self._read_states()
|
|
for device, nonce in devices:
|
|
state = states.setdefault(device.id, DeviceState())
|
|
state.nonce = max(state.nonce, nonce)
|
|
self._write_states(states)
|
|
panels_file.replace(panels_file.with_name(panels_file.name + ".migrated"))
|
|
logger.warning(
|
|
"Moved %d panels from %s into devices (%s)",
|
|
len(devices),
|
|
panels_file,
|
|
", ".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.
|
|
|
|
In memory: a restart forgets it, and a device that is still there
|
|
reconnects within seconds. One process serves the API, so one dictionary
|
|
is the whole answer.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
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()
|
|
|
|
def opened(self, device_id: str) -> None:
|
|
with self._lock:
|
|
count = self._sockets.get(device_id, 0) + 1
|
|
self._sockets[device_id] = count
|
|
self._seen[device_id] = time.time()
|
|
if count == 1:
|
|
self._announce(device_id)
|
|
|
|
def closed(self, device_id: str) -> None:
|
|
with self._lock:
|
|
count = max(0, self._sockets.get(device_id, 0) - 1)
|
|
if count:
|
|
self._sockets[device_id] = count
|
|
else:
|
|
self._sockets.pop(device_id, None)
|
|
self._seen[device_id] = time.time()
|
|
if not count:
|
|
self._announce(device_id)
|
|
|
|
def forget(self, device_id: str) -> None:
|
|
with self._lock:
|
|
self._sockets.pop(device_id, None)
|
|
self._seen.pop(device_id, None)
|
|
|
|
def online(self, device_id: str) -> bool:
|
|
return self._sockets.get(device_id, 0) > 0
|
|
|
|
def last_seen(self, device_id: str) -> datetime | None:
|
|
seen = self._seen.get(device_id)
|
|
return datetime.fromtimestamp(seen, UTC) if seen else None
|
|
|
|
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": 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()
|
|
|
|
|
|
def is_lan(address: str) -> bool:
|
|
"""Whether a request came from a private network rather than the internet."""
|
|
try:
|
|
ip = ipaddress.ip_address(address)
|
|
except ValueError:
|
|
return False
|
|
return ip.is_private or ip.is_loopback
|
|
|
|
|
|
# -- what a device's credential reaches --------------------------------------
|
|
|
|
|
|
def dashboards_for(
|
|
device: DeviceDef | None, store: DashboardStore
|
|
) -> list[DashboardDef]:
|
|
"""The published documents this device shows, in rail order.
|
|
|
|
Published, since that is what a device draws. A name that no longer
|
|
resolves is skipped, and a device that is gone shows nothing.
|
|
|
|
Handed back whole rather than walked here, because what a device may do
|
|
with a message depends on the document it came from: a dashboard that says
|
|
it is locked entitles a screen to read it and not to touch it.
|
|
"""
|
|
if device is None:
|
|
return []
|
|
found: list[DashboardDef] = []
|
|
for name in device.dashboards:
|
|
try:
|
|
found.append(store.read(name))
|
|
except DashboardNotFound:
|
|
continue
|
|
return found
|
|
|
|
|
|
def messages_of(defn: DashboardDef) -> set[str]:
|
|
"""Every message one dashboard reads or writes.
|
|
|
|
A dashboard's own bound settings count, not only its widgets': the theme a
|
|
panel is driven to is a message no tile on it draws.
|
|
"""
|
|
names = set(defn.setting_messages)
|
|
for widget in defn.widgets:
|
|
names.update(widget.messages)
|
|
if widget.target:
|
|
names.add(widget.target)
|
|
return names
|
|
|
|
|
|
def requests_of(defn: DashboardDef) -> set[str]:
|
|
"""The publishes this dashboard makes in order to read.
|
|
|
|
A querying chart asks a flow for the series it draws by publishing a
|
|
request, so that publish is how the tile reads. Every other message a
|
|
dashboard sends comes from a control, which is what locking it turns off,
|
|
so this is what a locked dashboard may still send.
|
|
"""
|
|
return {w.target for w in defn.widgets if w.type == "chart" and w.target}
|
|
|
|
|
|
def targets_of(defn: DashboardDef) -> set[str]:
|
|
"""The messages this dashboard's inputs publish to.
|
|
|
|
What a screen showing it may write, a narrower set than what it may read.
|
|
"""
|
|
return {w.target for w in defn.widgets if w.target}
|
|
|
|
|
|
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 {b.message for b in devices.bound(device.id) if b.direction == "in"}
|
|
|
|
|
|
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 = {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)
|
|
}
|