1087 lines
41 KiB
Python
1087 lines
41 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,
|
|
ValidationInfo,
|
|
ValidatorFunctionWrapHandler,
|
|
field_validator,
|
|
model_validator,
|
|
)
|
|
|
|
from fluksio.flow.dashboards import DashboardDef, 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"
|
|
|
|
#: The dtypes a device's ``viewers`` setting covers: what a camera sends.
|
|
CAMERA_DTYPES = frozenset({"image", "video"})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Binding:
|
|
"""One port of a device, wired to a message by a device node."""
|
|
|
|
device: str
|
|
#: The flow's key, which ``message`` is qualified with.
|
|
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)
|
|
#: The flow's id, which is where wiring the port is edited.
|
|
flow_id: str = ""
|
|
|
|
|
|
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."""
|
|
|
|
#: The flow's key, and its id to open it by.
|
|
flow: str
|
|
flow_id: str = ""
|
|
node: str
|
|
port: str
|
|
message: str
|
|
direction: Literal["in", "out"]
|
|
dtype: str = ""
|
|
|
|
|
|
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,
|
|
flow_id=flow.id,
|
|
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 = ""
|
|
#: Dashboard ids, ordered. The first one is what the device opens after
|
|
#: pairing, and the rail follows this order. An id that no longer resolves
|
|
#: is a dashboard someone deleted, and the device skips it. A dashboard from
|
|
#: before ids has its key as its id, so a list from then still holds.
|
|
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
|
|
#: Who may see what the device's camera sends, its ``image`` and ``video``
|
|
#: ``in`` messages, their bytes and its live view: superusers only, or
|
|
#: every member too. A device credential is bounded by its dashboards
|
|
#: instead, which a superuser assigned.
|
|
viewers: Literal["superusers", "members"] = "superusers"
|
|
#: 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
|
|
#: When a pairing here last minted the device a credential. None for one
|
|
#: paired before this was kept, whose settings were all chosen by hand.
|
|
paired_at: datetime | None = None
|
|
|
|
|
|
_STATES = TypeAdapter(dict[str, DeviceState])
|
|
|
|
|
|
#: Read in place of a value this version does not know, such as a newer
|
|
#: firmware's display shape or port direction: no reason to refuse the rest of
|
|
#: what the device says.
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
def _known(value: Any, known: tuple[str, ...]) -> Any:
|
|
return value if value in known else UNKNOWN
|
|
|
|
|
|
def _left_out(info: ValidationInfo, what: str, why: str) -> None:
|
|
"""Note a part of a report that was dropped, for whoever reads it to log."""
|
|
if isinstance(info.context, dict):
|
|
info.context.setdefault("left_out", []).append(f"{what} ({why})")
|
|
|
|
|
|
def _why(exc: ValidationError) -> str:
|
|
error = exc.errors(include_url=False)[0]
|
|
where = ".".join(str(part) for part in error["loc"])
|
|
return f"{where}: {error['msg']}" if where else str(error["msg"])
|
|
|
|
|
|
class ReportDisplay(BaseModel):
|
|
"""The device's screen, in pixels."""
|
|
|
|
w: int = Field(ge=0)
|
|
h: int = Field(ge=0)
|
|
shape: Literal["rect", "round", "unknown"] = "rect"
|
|
touch: bool = False
|
|
|
|
@field_validator("shape", mode="before")
|
|
@classmethod
|
|
def _shape(cls, value: Any) -> Any:
|
|
return _known(value, ("rect", "round"))
|
|
|
|
|
|
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", "unknown"]
|
|
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
|
|
|
|
@field_validator("direction", mode="before")
|
|
@classmethod
|
|
def _direction(cls, value: Any) -> Any:
|
|
return _known(value, ("in", "out"))
|
|
|
|
@field_validator("modes", mode="before")
|
|
@classmethod
|
|
def _modes(cls, value: Any) -> Any:
|
|
# The ones this version wires. Another is not said, rather than wrong.
|
|
if isinstance(value, list):
|
|
return [mode for mode in value if mode in ("in", "out")]
|
|
return value
|
|
|
|
|
|
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.
|
|
|
|
Read leniently, so a newer firmware still reports to an older instance.
|
|
Keys this version does not know are dropped, a value of a closed set it
|
|
does not know reads as ``unknown``, and a port or a section it cannot read
|
|
is left out on its own. Only a report that is not version 1, or not an
|
|
object at all, is refused. :func:`read_report` says what was left out.
|
|
|
|
A port's ``config`` may name an entry of ``config_schemas`` instead of
|
|
spelling the schema out, so a board with thirty pins sends the pin schema
|
|
once. It is put in place when the report is read. 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)
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def _named_configs(cls, data: Any, info: ValidationInfo) -> Any:
|
|
if not isinstance(data, dict) or not isinstance(data.get("ports"), list):
|
|
return data
|
|
schemas = data.get("config_schemas")
|
|
schemas = schemas if isinstance(schemas, dict) else {}
|
|
ports = []
|
|
for port in data["ports"]:
|
|
if isinstance(port, dict) and isinstance(port.get("config"), str):
|
|
schema = schemas.get(port["config"])
|
|
if not isinstance(schema, dict):
|
|
_left_out(
|
|
info,
|
|
f"the config of port {port.get('port')!r}",
|
|
f"no schema named {port['config']!r}",
|
|
)
|
|
schema = None
|
|
port = {**port, "config": schema}
|
|
ports.append(port)
|
|
return {**data, "ports": ports}
|
|
|
|
@field_validator("ports", mode="before")
|
|
@classmethod
|
|
def _each_port(cls, value: Any, info: ValidationInfo) -> Any:
|
|
if not isinstance(value, list):
|
|
_left_out(info, "ports", "not a list")
|
|
return []
|
|
kept = []
|
|
for raw in value:
|
|
try:
|
|
kept.append(ReportPort.model_validate(raw))
|
|
except ValidationError as exc:
|
|
name = raw.get("port") if isinstance(raw, dict) else raw
|
|
_left_out(info, f"port {name!r}", _why(exc))
|
|
return kept
|
|
|
|
@field_validator("features", mode="before")
|
|
@classmethod
|
|
def _flags(cls, value: Any, info: ValidationInfo) -> Any:
|
|
if not isinstance(value, dict):
|
|
_left_out(info, "features", "not an object")
|
|
return {}
|
|
flags = {}
|
|
for name, flag in value.items():
|
|
if isinstance(flag, bool):
|
|
flags[str(name)] = flag
|
|
else:
|
|
_left_out(info, f"feature {name!r}", f"{flag!r} is not true or false")
|
|
return flags
|
|
|
|
@field_validator("platform", "board", "firmware", "display", "camera", mode="wrap")
|
|
@classmethod
|
|
def _or_left_out(
|
|
cls, value: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo
|
|
) -> Any:
|
|
try:
|
|
return handler(value)
|
|
except ValidationError as exc:
|
|
name = info.field_name or ""
|
|
_left_out(info, name, _why(exc))
|
|
return cls.model_fields[name].get_default()
|
|
|
|
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)
|
|
|
|
|
|
def read_report(raw: bytes) -> tuple[DeviceReport, list[str]]:
|
|
"""A device's report, and each part of it that was left out.
|
|
|
|
Raises ``ValidationError`` only for a report that is not version 1 or
|
|
not an object.
|
|
"""
|
|
context: dict[str, list[str]] = {"left_out": []}
|
|
report = DeviceReport.model_validate_json(raw, context=context)
|
|
return report, context["left_out"]
|
|
|
|
|
|
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 DeviceUnavailable(RuntimeError):
|
|
"""Whether a device credential stands cannot be told right now.
|
|
|
|
Not a refusal: a device told its credential is refused pairs again, so
|
|
a store that is not running or a nonce file that does not read answers
|
|
"try again later" instead.
|
|
"""
|
|
|
|
|
|
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()
|
|
#: ``devices.json`` as last parsed, by the file's stat. Every device
|
|
#: request holds its nonce against it.
|
|
self._states_seen: (
|
|
tuple[tuple[int, int, int], dict[str, DeviceState]] | None
|
|
) = None
|
|
#: The camera messages only superusers may see, at the flow store
|
|
#: revision they were worked out at. A device save commits there too.
|
|
self._restricted: tuple[int, frozenset[str]] | None = None
|
|
|
|
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)
|
|
# Through a temporary file: a credential checked against a half
|
|
# written one would read as a deleted device and be refused.
|
|
tmp = path.with_name(path.name + ".tmp")
|
|
tmp.write_text(device.model_dump_json(indent=2) + "\n")
|
|
os.replace(tmp, path)
|
|
|
|
# -- what the published flows wire ------------------------------------
|
|
|
|
def wiring(self) -> dict[str, list[Binding]]:
|
|
"""Every device's bound ports in enabled published flows, by device id.
|
|
|
|
A stopped flow wires nothing: its ports go out of the device's ``io``
|
|
and out of its credential's reach until it is started again.
|
|
|
|
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, starting and
|
|
stopping a flow 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():
|
|
if not self.flows.read_enabled(flow.id):
|
|
continue
|
|
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]:
|
|
"""What the device's ports are wired to, one binding per port.
|
|
|
|
A port wired twice is an issue on both nodes. Until it is fixed the
|
|
device follows the first of the two in the fixed order, for its
|
|
``io`` and its credential's reach alike, so neither flips between them.
|
|
"""
|
|
found: list[Binding] = []
|
|
seen: set[str] = set()
|
|
for binding in self.wiring().get(device_id, []):
|
|
if binding.port not in seen:
|
|
seen.add(binding.port)
|
|
found.append(binding)
|
|
return found
|
|
|
|
def restricted(self) -> frozenset[str]:
|
|
"""The messages only a superuser may see: what the cameras of devices
|
|
whose ``viewers`` are superusers publish, their ``image`` and ``video``
|
|
``in`` ports.
|
|
|
|
Asked on every value a person's socket carries, so worked out again
|
|
only once the flow store commits, which every device save does.
|
|
"""
|
|
revision = self.flows.revision
|
|
held = self._restricted
|
|
if held is not None and held[0] == revision:
|
|
return held[1]
|
|
wiring = self.wiring()
|
|
found = frozenset(
|
|
binding.message
|
|
for device in self.list()
|
|
if device.viewers == "superusers"
|
|
for binding in wiring.get(device.id, ())
|
|
if binding.direction == "in" and binding.dtype in CAMERA_DTYPES
|
|
)
|
|
# Keyed by the revision read before, so a commit meanwhile is seen.
|
|
self._restricted = (revision, found)
|
|
return found
|
|
|
|
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.
|
|
|
|
Every binding, a port wired twice twice, so the one to remove shows.
|
|
"""
|
|
return sorted(
|
|
(
|
|
DeviceBinding(
|
|
flow=b.flow,
|
|
flow_id=b.flow_id,
|
|
node=b.node,
|
|
port=b.port,
|
|
message=b.message,
|
|
direction=b.direction,
|
|
dtype=b.dtype,
|
|
)
|
|
for b in self.wiring().get(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. Its report goes, since a device
|
|
created again is not the board that made 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, forget=True)
|
|
presence.forget(device_id)
|
|
|
|
# -- outside git -------------------------------------------------------
|
|
|
|
def _read_states(self) -> dict[str, DeviceState]:
|
|
"""The file as it is now, for a writer to change and write back.
|
|
|
|
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 _states(self) -> dict[str, DeviceState]:
|
|
"""The same for reading only, parsed again only once the file changed.
|
|
|
|
Every device request reads its nonce here. A write from this process
|
|
replaces what is held at once, and one from anywhere else changes the
|
|
file's stat. Shared between callers, so never changed in place.
|
|
"""
|
|
try:
|
|
seen = self.state_file.stat()
|
|
except FileNotFoundError:
|
|
return {}
|
|
key = (seen.st_mtime_ns, seen.st_size, seen.st_ino)
|
|
held = self._states_seen
|
|
if held is not None and held[0] == key:
|
|
return held[1]
|
|
# Stat before read, so what is held is never older than its key.
|
|
states = _STATES.validate_json(self.state_file.read_bytes())
|
|
self._states_seen = (key, states)
|
|
return states
|
|
|
|
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)
|
|
seen = self.state_file.stat()
|
|
self._states_seen = ((seen.st_mtime_ns, seen.st_size, seen.st_ino), states)
|
|
|
|
def state(self, device_id: str) -> DeviceState:
|
|
return self._states().get(device_id, DeviceState())
|
|
|
|
def nonce(self, device_id: str) -> int:
|
|
return self.state(device_id).nonce
|
|
|
|
def bump(self, device_id: str, *, forget: bool = False) -> int:
|
|
"""Refuse the credential this device holds now. Returns the new nonce.
|
|
|
|
``forget`` drops everything else kept about the device as well.
|
|
"""
|
|
with self._state_lock:
|
|
states = self._read_states()
|
|
state = states.get(device_id) or DeviceState()
|
|
if forget:
|
|
state = DeviceState(nonce=state.nonce)
|
|
state.nonce += 1
|
|
states[device_id] = state
|
|
self._write_states(states)
|
|
return state.nonce
|
|
|
|
def paired(self, device_id: str) -> None:
|
|
"""Note that a pairing here just minted this device a credential."""
|
|
with self._state_lock:
|
|
states = self._read_states()
|
|
states.setdefault(device_id, DeviceState()).paired_at = datetime.now(UTC)
|
|
self._write_states(states)
|
|
|
|
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. An id 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 []
|
|
return [defn for dash_id in device.dashboards if (defn := store.published(dash_id))]
|
|
|
|
|
|
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)
|
|
}
|