stroblme
e9941b770d
POST/GET /flows/{name}/nodes/{id}/train submit wren_train on the node's
targets, examples and tests and read its latest run (run.for_node,
cause "train"); "Use this model" writes `model` into the draft.
1908 lines
72 KiB
Python
1908 lines
72 KiB
Python
"""The flow API. Everything the editor can do is available here first."""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import re
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import orjson
|
|
from fastapi import (
|
|
APIRouter,
|
|
Depends,
|
|
HTTPException,
|
|
Request,
|
|
WebSocket,
|
|
WebSocketDisconnect,
|
|
)
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from jwt.exceptions import InvalidTokenError
|
|
from pydantic import BaseModel
|
|
from sqlmodel import Session, col, select
|
|
|
|
from fluksio.api.deps import (
|
|
CAMERA_IS_SUPERUSERS,
|
|
CODE_IS_SUPERUSERS,
|
|
CurrentUser,
|
|
FlowControllerDep,
|
|
HiddenDep,
|
|
SessionDep,
|
|
TokenDep,
|
|
decode_token,
|
|
get_code_author,
|
|
get_current_user,
|
|
hidden_from,
|
|
user_from_token,
|
|
)
|
|
from fluksio.api.routes.messages import port_source
|
|
from fluksio.api.socket_auth import TICKET_TTL_S, issue_ticket, socket_token
|
|
from fluksio.core.config import settings
|
|
from fluksio.core.db import engine
|
|
from fluksio.flow import live as live_view
|
|
from fluksio.flow.artifacts import is_reference
|
|
from fluksio.flow.controller import NODE_TYPES, FlowController, follow_header
|
|
from fluksio.flow.dashboards import DashboardStore
|
|
from fluksio.flow.demand import demand
|
|
from fluksio.flow.devices import (
|
|
Binding,
|
|
DeviceDef,
|
|
DeviceStore,
|
|
DeviceUnavailable,
|
|
dashboards_for,
|
|
messages_for,
|
|
messages_of,
|
|
presence,
|
|
)
|
|
from fluksio.flow.events import event_bus
|
|
from fluksio.flow.messages import MEDIA_FAMILIES, DType, flow_of, qualify
|
|
from fluksio.flow.nodes.wren import TRAIN_FLOW, WrenNode, pack_meta
|
|
from fluksio.flow.pipeline import ValidationIssue
|
|
from fluksio.flow.runs import RunRejected
|
|
from fluksio.flow.schemas import (
|
|
NAME_PATTERN,
|
|
BrainGraph,
|
|
FlowDef,
|
|
FlowsPublic,
|
|
FlowStatePublic,
|
|
FlowSummary,
|
|
HistoryPoint,
|
|
LibraryNode,
|
|
MessageHistory,
|
|
MessageValue,
|
|
NodeDef,
|
|
NodeSource,
|
|
NodeSourceSaved,
|
|
NodeStatusPublic,
|
|
NodeTypeInfo,
|
|
)
|
|
from fluksio.flow.state import as_number
|
|
from fluksio.flow.store import (
|
|
FlowExists,
|
|
FlowNotFound,
|
|
LibExists,
|
|
LibNotFound,
|
|
StaleVersion,
|
|
)
|
|
from fluksio.models import Flavor, Message, Run, RunArtifact, RunNode
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(
|
|
prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)]
|
|
)
|
|
# The websocket authenticates from its handshake, a header or a ticket, so it
|
|
# stays off the router that enforces the HTTP bearer scheme.
|
|
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, 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 or flow.
|
|
kind: str
|
|
#: Stable within its kind, and used as the canvas node id.
|
|
id: str
|
|
label: str
|
|
#: 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] = []
|
|
requires: list[str] = []
|
|
|
|
|
|
class FlowDetail(BaseModel):
|
|
"""A flow plus how it is currently doing.
|
|
|
|
``definition`` is the working copy — the unpublished draft when there is
|
|
one — because that is what the editor shows. ``nodes`` reports the
|
|
published flow, which is what is actually running.
|
|
"""
|
|
|
|
definition: FlowDef
|
|
nodes: list[NodeStatusPublic] = []
|
|
issues: list[ValidationIssue] = []
|
|
has_draft: bool = False
|
|
enabled: bool = True
|
|
paused: bool = False
|
|
#: Dashboards and other flows wired into this one.
|
|
endpoints: list[Endpoint] = []
|
|
|
|
|
|
class ValidationResult(BaseModel):
|
|
issues: list[ValidationIssue] = []
|
|
|
|
|
|
class RenameRequest(BaseModel):
|
|
new_name: str
|
|
|
|
|
|
class PublishRequest(BaseModel):
|
|
version: int
|
|
|
|
|
|
class ShareRequest(BaseModel):
|
|
lib_name: str
|
|
|
|
|
|
class RunRequest(BaseModel):
|
|
inputs: dict[str, Any] = {}
|
|
|
|
|
|
class TriggerRequest(BaseModel):
|
|
values: dict[str, Any] = {}
|
|
|
|
|
|
class Training(BaseModel):
|
|
"""A wren node's latest training run, as its panel shows it."""
|
|
|
|
id: str
|
|
#: queued, running, ok, error, cancelled or abandoned.
|
|
status: str
|
|
#: What went wrong, when it failed.
|
|
error: str = ""
|
|
#: final_loss, n_params, exact_match, threshold and coverage, once it ended.
|
|
result: dict[str, Any] = {}
|
|
#: The trained model's digest, for the node's ``model`` setting.
|
|
model: str = ""
|
|
#: From the model's own record: the confidence it recommends, and its
|
|
#: score on the node's ``tests`` when there were any.
|
|
threshold: float | None = None
|
|
test_exact_match: float | None = None
|
|
|
|
|
|
def _endpoints(controller: FlowController, definition: FlowDef) -> list[Endpoint]:
|
|
"""Everything wired into this flow from outside it."""
|
|
flow = definition.name
|
|
found: dict[str, Endpoint] = {}
|
|
|
|
dashboards: DashboardStore | None = getattr(controller, "dashboards", None)
|
|
if dashboards is not None:
|
|
for binding in dashboards.bindings_for(flow):
|
|
found[f"dashboard:{binding['dashboard']}:{binding['widget']}"] = Endpoint(
|
|
kind="dashboard",
|
|
id=f"dashboard:{binding['dashboard']}:{binding['widget']}",
|
|
label=binding["title"],
|
|
detail=binding["type"],
|
|
provides=[binding["provides"]] if binding["provides"] else [],
|
|
requires=binding["requires"],
|
|
)
|
|
|
|
def _flow_endpoint(key: str, detail: str) -> Endpoint:
|
|
"""One label per node on the far side, however many names reach it."""
|
|
return found.setdefault(
|
|
f"flow:{key}",
|
|
Endpoint(kind="flow", id=f"flow:{key}", label=key, detail=detail),
|
|
)
|
|
|
|
# An outsider reaching into this flow.
|
|
for other, node_id, provides, requires in controller.cross_flow_nodes(flow):
|
|
endpoint = _flow_endpoint(f"{other}.{node_id}", "flow")
|
|
endpoint.provides += provides
|
|
endpoint.requires += requires
|
|
|
|
# And this flow reaching out: its own ports bound to a message of another
|
|
# flow. Read from the working document rather than from the store, so a
|
|
# name just typed is drawn before it has been published — the same reason
|
|
# the canvas draws its own boundary from the document.
|
|
for node in definition.nodes:
|
|
for spec, ours_publishes in [
|
|
*((spec, False) for spec in node.requires),
|
|
*((spec, True) for spec in node.provides),
|
|
]:
|
|
message = qualify(flow, spec.name or "")
|
|
if not message or flow_of(message) == flow:
|
|
continue
|
|
# The node at the other end, so the label reads like a dashboard's:
|
|
# what it is on the first line, what sort of thing it is on the
|
|
# second. A message no published flow declares yet has no other end
|
|
# to name, so it is drawn as the message it is.
|
|
far = controller.message_node(
|
|
message, published=not ours_publishes, exclude=flow
|
|
)
|
|
endpoint = _flow_endpoint(
|
|
f"{far.flow}.{far.node}" if far else message,
|
|
far.type if far else "flow",
|
|
)
|
|
# An endpoint publishing into this flow is what this flow reads.
|
|
side = endpoint.requires if ours_publishes else endpoint.provides
|
|
if message not in side:
|
|
side.append(message)
|
|
|
|
return list(found.values())
|
|
|
|
|
|
def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail:
|
|
"""The flow and how its nodes are doing. Blocking: call it off the loop.
|
|
|
|
A draft is built to report on it, which asks a worker to compile its
|
|
nodes, and that waits on the socket the loop serves.
|
|
"""
|
|
name = definition.name
|
|
running = {
|
|
"enabled": controller.is_enabled(name),
|
|
"paused": controller.is_paused(name),
|
|
}
|
|
endpoints = _endpoints(controller, definition)
|
|
if controller.store.has_draft(name):
|
|
# Report the draft the editor is showing, not the version running
|
|
# underneath it — otherwise a node the author just broke looks fine.
|
|
preview = controller.preview(name)
|
|
return FlowDetail(
|
|
definition=definition,
|
|
nodes=preview.nodes,
|
|
issues=preview.issues,
|
|
has_draft=True,
|
|
endpoints=endpoints,
|
|
**running,
|
|
)
|
|
return FlowDetail(
|
|
definition=definition,
|
|
nodes=controller.node_statuses(name),
|
|
issues=controller.flow_issues(name),
|
|
endpoints=endpoints,
|
|
**running,
|
|
)
|
|
|
|
|
|
def _read_flow(controller: FlowController, name: str) -> FlowDef:
|
|
"""The working copy: the draft when there is one, else what is published."""
|
|
try:
|
|
return controller.store.read_flow(name, draft=True)
|
|
except FlowNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
|
|
|
|
|
|
def _audit(action: str, flow: str, user: CurrentUser) -> None:
|
|
"""Record who changed what. The collector writes it down; the bus carries it."""
|
|
event_bus.publish(
|
|
{
|
|
"type": "audit",
|
|
"action": action,
|
|
"flow": flow,
|
|
"user": user.email,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
|
|
|
|
def _source_ref(definition: FlowDef, node_id: str) -> str | None:
|
|
"""The library source this node runs, if it is a shared one."""
|
|
node = next((n for n in definition.nodes if n.id == node_id), None)
|
|
return node.source_ref if node else None
|
|
|
|
|
|
def _require_enabled(controller: FlowController, name: str) -> None:
|
|
if not controller.is_enabled(name):
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"Flow '{name}' is stopped — start it before running it",
|
|
)
|
|
|
|
|
|
def _flow_state(
|
|
controller: FlowController, name: str, hidden: frozenset[str]
|
|
) -> FlowStatePublic:
|
|
"""The flow's values, but those of a camera the caller may not see."""
|
|
return FlowStatePublic(
|
|
values={
|
|
key: MessageValue(**value)
|
|
for key, value in controller.values(name).items()
|
|
if key not in hidden
|
|
},
|
|
nodes=controller.node_statuses(name),
|
|
)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Flows
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/", response_model=FlowsPublic)
|
|
def read_flows(controller: FlowControllerDep) -> Any:
|
|
"""List every flow."""
|
|
summaries = []
|
|
for name in controller.store.list_flows():
|
|
try:
|
|
definition = controller.store.read_flow(name, draft=True)
|
|
except FlowNotFound:
|
|
continue
|
|
statuses = controller.node_statuses(name)
|
|
summaries.append(
|
|
FlowSummary(
|
|
name=definition.name,
|
|
title=definition.title,
|
|
mode=definition.mode,
|
|
node_count=len(definition.nodes),
|
|
error_count=sum(1 for s in statuses if s.status == "error"),
|
|
has_draft=controller.store.has_draft(name),
|
|
enabled=controller.is_enabled(name),
|
|
paused=controller.is_paused(name),
|
|
quarantined=controller.is_quarantined(name),
|
|
version=definition.version,
|
|
updated_at=controller.store.updated_at(name),
|
|
)
|
|
)
|
|
return FlowsPublic(data=summaries, count=len(summaries))
|
|
|
|
|
|
@router.get("/node-types", response_model=list[NodeTypeInfo])
|
|
def read_node_types() -> Any:
|
|
"""The node types that can be placed on a canvas."""
|
|
from fluksio.flow.controller import node_type_info
|
|
|
|
return node_type_info()
|
|
|
|
|
|
# Above "/{name}" for the same reason "library" is: a flow called "graph" would
|
|
# otherwise be unreachable.
|
|
@router.get("/graph", response_model=BrainGraph)
|
|
def read_graph(controller: FlowControllerDep) -> Any:
|
|
"""Every flow as one graph, with nodes talking to the same thing merged."""
|
|
return controller.brain_graph()
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Shared nodes
|
|
#
|
|
# Declared above the "/{name}" routes: "library" would otherwise be read as a
|
|
# flow name.
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/library", response_model=list[LibraryNode])
|
|
def read_library(controller: FlowControllerDep) -> Any:
|
|
"""The node sources shared across flows, and which nodes use each."""
|
|
return [
|
|
LibraryNode(name=name, used_by=controller.store.usages(name))
|
|
for name in controller.store.list_lib()
|
|
]
|
|
|
|
|
|
@router.delete("/library/{lib_name}", response_model=Message)
|
|
async def delete_shared_node(
|
|
lib_name: str, controller: FlowControllerDep, detach: bool = False
|
|
) -> Any:
|
|
"""Remove a shared source, as long as no flow still runs it.
|
|
|
|
With ``detach``, every node using it keeps a copy of the code as its own
|
|
first, so the source can be retired while flows still use it.
|
|
"""
|
|
if detach:
|
|
running = await run_in_threadpool(controller.store.usages, lib_name, False)
|
|
try:
|
|
await run_in_threadpool(controller.store.detach_lib, lib_name)
|
|
except LibNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No shared node '{lib_name}'")
|
|
for flow in sorted({usage.split(".", 1)[0] for usage in running}):
|
|
await controller.reload_flow(flow)
|
|
return Message(message=f"Deleted shared node '{lib_name}'")
|
|
# The published flows too: a draft that unshared a node still runs it.
|
|
used_by = sorted(
|
|
{*controller.store.usages(lib_name), *controller.store.usages(lib_name, False)}
|
|
)
|
|
if used_by:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"'{lib_name}' is still used by {', '.join(used_by)}",
|
|
)
|
|
try:
|
|
await run_in_threadpool(controller.store.delete_lib_source, lib_name)
|
|
except LibNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No shared node '{lib_name}'")
|
|
return Message(message=f"Deleted shared node '{lib_name}'")
|
|
|
|
|
|
@router.get("/{name}", response_model=FlowDetail)
|
|
def read_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Read one flow, with the state of its nodes."""
|
|
return _detail(controller, _read_flow(controller, name))
|
|
|
|
|
|
def _check_flavors(definition: FlowDef) -> None:
|
|
"""Refuse a size nobody stored, here rather than when the node is built.
|
|
|
|
Catching it at the save covers the canvas and ``fluksio sync`` in one
|
|
place; a node that only finds out at build time is a red node somebody has
|
|
to go and look at.
|
|
"""
|
|
named = {
|
|
node.resources.flavor
|
|
for node in definition.nodes
|
|
if node.resources is not None and node.resources.flavor
|
|
}
|
|
if not named:
|
|
return
|
|
with Session(engine) as session:
|
|
known = set(session.exec(select(Flavor.name)).all())
|
|
unknown = sorted(named - known)
|
|
if unknown:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"No flavor named '{unknown[0]}' "
|
|
f"(known: {', '.join(sorted(known)) or 'none'})",
|
|
)
|
|
|
|
|
|
def _check_origin(definition: FlowDef) -> None:
|
|
"""Refuse an ``origin.repo`` that is not a project's own directory.
|
|
|
|
Every run of a code-defined flow hashes the python files below it, so
|
|
``/`` or a home directory makes each run walk far more than the project,
|
|
and a relative path names nothing the engine can find.
|
|
"""
|
|
repo = definition.origin.repo if definition.origin is not None else ""
|
|
if not repo:
|
|
return
|
|
if not os.path.isabs(repo):
|
|
problem = "is a relative path"
|
|
elif os.path.realpath(repo) == "/":
|
|
problem = "is the filesystem root"
|
|
elif os.path.realpath(repo) == os.path.realpath(Path.home()):
|
|
problem = "is the home directory"
|
|
else:
|
|
return
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=(
|
|
f"origin.repo '{repo}' {problem}. It has to be the absolute path "
|
|
"of the repository the flow is declared in, since every run hashes "
|
|
"the python files below it. Sync from inside that repository."
|
|
),
|
|
)
|
|
|
|
|
|
def _worker_env(node: NodeDef | None) -> dict[str, str]:
|
|
return node.resources.env if node is not None and node.resources else {}
|
|
|
|
|
|
def _changes_code(node: NodeDef, old: NodeDef | None) -> bool:
|
|
"""Whether saving *node* over *old* changes code the engine would run.
|
|
|
|
A function node's body has its own route, gated there. What the document
|
|
carries is which nodes run code at all, which shared source each runs, the
|
|
command in an exec node's settings, and the environment a node's worker
|
|
starts with — where `PYTHONPATH` alone is enough to import a file of
|
|
somebody's choosing.
|
|
"""
|
|
if _worker_env(old) != _worker_env(node):
|
|
return True
|
|
spec = NODE_TYPES.get(node.type)
|
|
if spec is None or not spec.runs_code:
|
|
return False
|
|
return (
|
|
old is None
|
|
or old.type != node.type
|
|
or old.source_ref != node.source_ref
|
|
or (not spec.has_source and old.params != node.params)
|
|
)
|
|
|
|
|
|
def _refuse_code_changes(definition: FlowDef, before: FlowDef | None) -> None:
|
|
"""Refuse a non-superuser a change to code this flow would run.
|
|
|
|
*before* is the working copy being replaced, None for a new flow. Wiring,
|
|
titles, a function node's settings and every other node stay anyone's.
|
|
"""
|
|
previous = {node.id: node for node in before.nodes} if before else {}
|
|
for node in definition.nodes:
|
|
if _changes_code(node, previous.get(node.id)):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"{CODE_IS_SUPERUSERS} (node '{node.id}')",
|
|
)
|
|
|
|
|
|
@router.put("/{name}", response_model=FlowDetail)
|
|
async def save_flow(
|
|
name: str,
|
|
definition: FlowDef,
|
|
controller: FlowControllerDep,
|
|
user: CurrentUser,
|
|
) -> Any:
|
|
"""Save unpublished changes to a flow.
|
|
|
|
This writes a draft: the running pipeline keeps the published version until
|
|
someone publishes. ``version`` is the one the editor last saw — a mismatch
|
|
means another client saved in between and answers 409 rather than throwing
|
|
their work away.
|
|
"""
|
|
if definition.name != name:
|
|
raise HTTPException(
|
|
status_code=400, detail="The flow name in the body must match the URL"
|
|
)
|
|
|
|
duplicates = {n.id for n in definition.nodes}
|
|
if len(duplicates) != len(definition.nodes):
|
|
raise HTTPException(status_code=400, detail="Node names must be unique")
|
|
|
|
_check_flavors(definition)
|
|
_check_origin(definition)
|
|
if not user.is_superuser:
|
|
before = _read_flow(controller, name) if controller.store.exists(name) else None
|
|
_refuse_code_changes(definition, before)
|
|
|
|
try:
|
|
stored = await run_in_threadpool(
|
|
controller.store.write_draft, definition, definition.version
|
|
)
|
|
except StaleVersion as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={"message": str(exc), "current_version": exc.current},
|
|
)
|
|
return await run_in_threadpool(_detail, controller, stored)
|
|
|
|
|
|
@router.post("/{name}/publish", response_model=FlowDetail)
|
|
async def publish_flow(
|
|
name: str,
|
|
body: PublishRequest,
|
|
controller: FlowControllerDep,
|
|
user: CurrentUser,
|
|
) -> Any:
|
|
"""Deploy the unpublished changes: the engine picks them up from here."""
|
|
_read_flow(controller, name)
|
|
if not controller.store.has_draft(name):
|
|
raise HTTPException(
|
|
status_code=400, detail=f"Flow '{name}' has no unpublished changes"
|
|
)
|
|
try:
|
|
published = await run_in_threadpool(
|
|
controller.store.publish_flow, name, body.version
|
|
)
|
|
except StaleVersion as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={"message": str(exc), "current_version": exc.current},
|
|
)
|
|
_audit("published", name, user)
|
|
await controller.reload_flow(name)
|
|
return await run_in_threadpool(_detail, controller, published)
|
|
|
|
|
|
@router.post("/{name}/discard-draft", response_model=FlowDetail)
|
|
async def discard_draft(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Throw the unpublished changes away and go back to what is running."""
|
|
_read_flow(controller, name)
|
|
if not controller.store.has_draft(name):
|
|
raise HTTPException(
|
|
status_code=400, detail=f"Flow '{name}' has no unpublished changes"
|
|
)
|
|
if not controller.store.is_published(name):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=(
|
|
f"Flow '{name}' has never been published — delete it instead of "
|
|
"discarding it"
|
|
),
|
|
)
|
|
published = await run_in_threadpool(controller.store.discard_draft, name)
|
|
return await run_in_threadpool(_detail, controller, published)
|
|
|
|
|
|
@router.delete("/{name}", response_model=Message)
|
|
async def delete_flow(
|
|
name: str, controller: FlowControllerDep, user: CurrentUser, session: SessionDep
|
|
) -> Any:
|
|
"""Delete a flow and everything in it, except the record of what it ran."""
|
|
live = session.exec(
|
|
select(Run.id).where(
|
|
col(Run.flow) == name, col(Run.status).in_(("running", "queued"))
|
|
)
|
|
).first()
|
|
if live is not None:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=(
|
|
f"Flow '{name}' has a run in progress ({live}). Cancel it, or "
|
|
"wait for it to finish, before deleting the flow."
|
|
),
|
|
)
|
|
try:
|
|
await run_in_threadpool(controller.store.delete_flow, name)
|
|
except FlowNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
|
|
_audit("deleted", name, user)
|
|
# Its files are gone; its values and queued work would otherwise linger.
|
|
# Its runs stay: a run record is a record of what ran, and surviving the
|
|
# flow it belonged to is the point of keeping one. `DELETE /runs?flow=`
|
|
# is what clears them, and with `history=true` its cascades too.
|
|
await run_in_threadpool(controller.forget_flow, name)
|
|
await controller.reload_flow(name)
|
|
return Message(message=f"Deleted flow '{name}'")
|
|
|
|
|
|
@router.post("/{name}/rename", response_model=FlowDetail)
|
|
async def rename_flow(
|
|
name: str,
|
|
body: RenameRequest,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""Rename a flow, along with every reference to its messages."""
|
|
if not NAME_PATTERN.match(body.new_name):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=(
|
|
"Use lowercase letters, digits and underscores, starting with a letter"
|
|
),
|
|
)
|
|
try:
|
|
renamed = await run_in_threadpool(
|
|
controller.store.rename_flow, name, body.new_name
|
|
)
|
|
except FlowNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
|
|
except FlowExists as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc))
|
|
|
|
# The old name is nobody's namespace now; its values would sit there under a
|
|
# flow that no longer exists. They repopulate under the new name on the next
|
|
# run, so this is cleanup rather than a migration.
|
|
await run_in_threadpool(controller.forget_flow, name)
|
|
await controller.reload()
|
|
return await run_in_threadpool(_detail, controller, renamed)
|
|
|
|
|
|
@router.post("/{name}/duplicate", response_model=FlowDetail)
|
|
async def duplicate_flow(
|
|
name: str, controller: FlowControllerDep, user: CurrentUser
|
|
) -> Any:
|
|
"""Copy a flow into a new one that is only a draft, named after it.
|
|
|
|
Nothing to reload: a draft does not run. Its code nodes are new code in
|
|
the copy, so copying them is a superuser's, like writing them.
|
|
"""
|
|
if not user.is_superuser:
|
|
_refuse_code_changes(_read_flow(controller, name), None)
|
|
try:
|
|
copy = await run_in_threadpool(controller.store.duplicate_flow, name)
|
|
except FlowNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
|
|
return await run_in_threadpool(_detail, controller, copy)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Node source
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/{name}/nodes/{node_id}/source", response_model=NodeSource)
|
|
def read_node_source(
|
|
name: str,
|
|
node_id: str,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""Read a node's Python source, including unpublished edits."""
|
|
definition = _read_flow(controller, name)
|
|
ref = _source_ref(definition, node_id)
|
|
if ref:
|
|
try:
|
|
return NodeSource(code=controller.store.read_lib_source(ref))
|
|
except LibNotFound:
|
|
raise HTTPException(status_code=404, detail=f"No shared node '{ref}'")
|
|
return NodeSource(
|
|
code=controller.store.read_node_source(name, node_id, draft=True),
|
|
missing=not controller.store.has_node_source(name, node_id, draft=True),
|
|
)
|
|
|
|
|
|
def _follow_header(
|
|
controller: FlowController, name: str, node_id: str, code: str, drop: bool
|
|
) -> FlowDef | None:
|
|
"""Bring a function node's ports and settings in line with its header.
|
|
|
|
The draft is what the editor is on, so that is what moves, and only for a
|
|
node with its own code: a shared one serves other flows with other ports,
|
|
and a synced flow's ports are declared in its repository. Returns the
|
|
draft when it changed.
|
|
"""
|
|
definition = controller.store.read_flow(name, draft=True)
|
|
node = next((n for n in definition.nodes if n.id == node_id), None)
|
|
spec = NODE_TYPES.get(node.type) if node is not None else None
|
|
if (
|
|
node is None
|
|
or spec is None
|
|
or not spec.free_params
|
|
or node.source_ref
|
|
or definition.origin
|
|
):
|
|
return None
|
|
followed = follow_header(node, code, drop)
|
|
if followed == node:
|
|
return None
|
|
definition.nodes = [followed if n.id == node_id else n for n in definition.nodes]
|
|
try:
|
|
return controller.store.write_draft(definition, definition.version)
|
|
except StaleVersion:
|
|
# Someone saved in between; the next save of this code tries again.
|
|
return None
|
|
|
|
|
|
@router.put(
|
|
"/{name}/nodes/{node_id}/source",
|
|
response_model=NodeSourceSaved,
|
|
dependencies=[Depends(get_code_author)],
|
|
)
|
|
async def save_node_source(
|
|
name: str,
|
|
node_id: str,
|
|
source: NodeSource,
|
|
controller: FlowControllerDep,
|
|
user: CurrentUser,
|
|
drop: bool = True,
|
|
) -> Any:
|
|
"""Save a node's source as an unpublished edit and report whether it loads.
|
|
|
|
The answer comes from compiling the code rather than from the running
|
|
pipeline: a draft is not deployed, and compiling is both faster and more
|
|
precise about what the author just typed.
|
|
|
|
A shared node writes to the library, so the fix reaches every flow using
|
|
it — and that one is live immediately rather than waiting for a publish,
|
|
because the copy is not any single flow's to hold back. Being live, it is
|
|
audited like a publish.
|
|
|
|
A node with its own code takes its input ports and settings from the
|
|
header of ``process`` (see `follow_header`), and the answer carries the
|
|
node when that changed it. ``drop=false`` is the editor's autosave, which
|
|
adds and renames from the header but keeps what it no longer names until
|
|
the editor is left.
|
|
"""
|
|
definition = _read_flow(controller, name)
|
|
ref = _source_ref(definition, node_id)
|
|
if ref:
|
|
changed = await run_in_threadpool(
|
|
controller.store.write_lib_source, ref, source.code
|
|
)
|
|
if changed:
|
|
_audit(f"edited shared node '{ref}'", name, user)
|
|
await controller.reload_lib_users(ref)
|
|
else:
|
|
await run_in_threadpool(
|
|
controller.store.write_node_source, name, node_id, source.code, True
|
|
)
|
|
followed = await run_in_threadpool(
|
|
_follow_header, controller, name, node_id, source.code, drop
|
|
)
|
|
node_def = next((n for n in definition.nodes if n.id == node_id), None)
|
|
worker = (
|
|
node_def.worker
|
|
if node_def is not None and node_def.worker_policy == "require"
|
|
else None
|
|
)
|
|
error = await run_in_threadpool(
|
|
controller.compile_check, name, node_id, source.code, worker
|
|
)
|
|
return NodeSourceSaved(
|
|
id=f"{name}.{node_id}",
|
|
status="error" if error else "active",
|
|
error=error,
|
|
node=next((n for n in followed.nodes if n.id == node_id), None)
|
|
if followed
|
|
else None,
|
|
version=followed.version if followed else None,
|
|
)
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/share", response_model=FlowDetail)
|
|
async def share_node(
|
|
name: str,
|
|
node_id: str,
|
|
body: ShareRequest,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""Move this node's code into the library so other flows can run it too."""
|
|
_read_flow(controller, name)
|
|
if not NAME_PATTERN.match(body.lib_name):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=(
|
|
"Use lowercase letters, digits and underscores, starting with a letter"
|
|
),
|
|
)
|
|
try:
|
|
await run_in_threadpool(
|
|
controller.store.share_node, name, node_id, body.lib_name
|
|
)
|
|
except LibExists as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc))
|
|
return await run_in_threadpool(_detail, controller, _read_flow(controller, name))
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/unshare", response_model=FlowDetail)
|
|
async def unshare_node(
|
|
name: str,
|
|
node_id: str,
|
|
controller: FlowControllerDep,
|
|
) -> Any:
|
|
"""Take a private copy of the shared code back into this flow."""
|
|
_read_flow(controller, name)
|
|
try:
|
|
await run_in_threadpool(controller.store.unshare_node, name, node_id)
|
|
except LibNotFound as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc))
|
|
return await run_in_threadpool(_detail, controller, _read_flow(controller, name))
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Running, stopped, paused
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/{name}/start", response_model=FlowDetail)
|
|
async def start_flow(
|
|
name: str, controller: FlowControllerDep, user: CurrentUser
|
|
) -> Any:
|
|
"""Let the engine run this flow again."""
|
|
_read_flow(controller, name)
|
|
await controller.set_enabled(name, True)
|
|
_audit("started", name, user)
|
|
return await run_in_threadpool(_detail, controller, _read_flow(controller, name))
|
|
|
|
|
|
@router.post("/{name}/stop", response_model=FlowDetail)
|
|
async def stop_flow(name: str, controller: FlowControllerDep, user: CurrentUser) -> Any:
|
|
"""Take this flow off the engine: no subscriptions, schedules or webhooks."""
|
|
_read_flow(controller, name)
|
|
await controller.set_enabled(name, False)
|
|
_audit("stopped", name, user)
|
|
return await run_in_threadpool(_detail, controller, _read_flow(controller, name))
|
|
|
|
|
|
@router.post("/{name}/pause", response_model=Message)
|
|
def pause_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Hold the flow's nodes so its messages can be stepped through."""
|
|
_read_flow(controller, name)
|
|
controller.pause_flow(name)
|
|
return Message(message=f"Paused flow '{name}'")
|
|
|
|
|
|
@router.post("/{name}/resume", response_model=Message)
|
|
async def resume_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Let the flow carry on, running whatever was held back."""
|
|
_read_flow(controller, name)
|
|
await run_in_threadpool(controller.resume_flow, name)
|
|
return Message(message=f"Resumed flow '{name}'")
|
|
|
|
|
|
@router.post("/{name}/step", response_model=Message)
|
|
async def step_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Run one message a pause is holding back, leaving the flow paused."""
|
|
_read_flow(controller, name)
|
|
node = await run_in_threadpool(controller.step_flow, name)
|
|
if node is None:
|
|
return Message(message=f"Nothing held back in flow '{name}'")
|
|
return Message(message=f"Stepped '{node}'")
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Validation and execution
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/{name}/validate", response_model=ValidationResult)
|
|
def validate_flow(name: str, controller: FlowControllerDep) -> Any:
|
|
"""Report what would keep this flow from running."""
|
|
_read_flow(controller, name)
|
|
if controller.store.has_draft(name):
|
|
return ValidationResult(issues=controller.preview(name).issues)
|
|
return ValidationResult(issues=controller.flow_issues(name))
|
|
|
|
|
|
@router.post("/{name}/run", response_model=FlowStatePublic)
|
|
async def run_flow(
|
|
name: str,
|
|
body: RunRequest,
|
|
request: Request,
|
|
controller: FlowControllerDep,
|
|
user: CurrentUser,
|
|
hidden: HiddenDep,
|
|
) -> Any:
|
|
"""Run every node of a flow once.
|
|
|
|
With unpublished changes this runs the draft, so the button matches what is
|
|
on the canvas. Nothing is deployed by running it.
|
|
|
|
A *batch* flow is submitted as a run instead, because that is what running
|
|
one means: it is what keeps the parameters, the series and the result, and
|
|
a button that quietly did something else would be a trap.
|
|
"""
|
|
definition = _read_flow(controller, name)
|
|
_require_enabled(controller, name)
|
|
if definition.mode == "batch":
|
|
service = getattr(request.app.state, "run_service", None)
|
|
if service is None:
|
|
raise HTTPException(status_code=503, detail="Runs are not available")
|
|
try:
|
|
await run_in_threadpool(
|
|
service.submit,
|
|
name,
|
|
params=body.inputs,
|
|
cause="api",
|
|
actor=user.email,
|
|
draft=controller.store.has_draft(name),
|
|
)
|
|
except RunRejected as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
return _flow_state(controller, name, hidden)
|
|
|
|
inputs = {qualify(name, key): value for key, value in body.inputs.items()}
|
|
if controller.store.has_draft(name):
|
|
await run_in_threadpool(controller.run_preview, name, inputs)
|
|
else:
|
|
await run_in_threadpool(controller.run_flow, name, inputs)
|
|
return _flow_state(controller, name, hidden)
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/trigger", response_model=FlowStatePublic)
|
|
async def trigger_node(
|
|
name: str,
|
|
node_id: str,
|
|
body: TriggerRequest,
|
|
controller: FlowControllerDep,
|
|
hidden: HiddenDep,
|
|
) -> Any:
|
|
"""Feed values into a single node."""
|
|
_require_enabled(controller, name)
|
|
try:
|
|
error = await run_in_threadpool(
|
|
controller.trigger_node, f"{name}.{node_id}", body.values
|
|
)
|
|
except KeyError:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
|
|
)
|
|
if error:
|
|
# The canvas already has the failure from the bus; the person who
|
|
# clicked gets to read it too, instead of a stack trace in the log.
|
|
raise HTTPException(status_code=400, detail=error)
|
|
return _flow_state(controller, name, hidden)
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/cancel", response_model=Message)
|
|
def cancel_node(name: str, node_id: str, controller: FlowControllerDep) -> Any:
|
|
"""Stop a node that is running right now, by killing the worker running it.
|
|
|
|
Idempotent on purpose: by the time a click reaches here the node may well
|
|
have finished, and that is the outcome that was asked for.
|
|
"""
|
|
pool = controller.workers
|
|
if pool is not None and pool.cancel(f"{name}.{node_id}"):
|
|
return Message(message=f"Stopped '{node_id}'")
|
|
return Message(message=f"'{node_id}' was not running")
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/acknowledge", response_model=Message)
|
|
def acknowledge_node_error(
|
|
name: str, node_id: str, controller: FlowControllerDep
|
|
) -> Any:
|
|
"""Dismiss what a node last failed with, so the canvas stops marking it.
|
|
|
|
A failure outlives the next good run on purpose — otherwise one that fired
|
|
an alert leaves no trace by the time anyone looks. Reading the traceback is
|
|
what says it has been seen.
|
|
"""
|
|
try:
|
|
controller.acknowledge_error(f"{name}.{node_id}")
|
|
except KeyError:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
|
|
) from None
|
|
# Another browser holds the same marker and has no reason to refetch, so it
|
|
# would keep showing a failure that is gone until its next snapshot.
|
|
event_bus.publish(
|
|
{
|
|
"type": "node_error_acknowledged",
|
|
"node": f"{name}.{node_id}",
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
return Message(message=f"Cleared the failure on '{node_id}'")
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Training a wren node
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
def _node_def(controller: FlowController, name: str, node_id: str) -> NodeDef:
|
|
"""A node as the editor shows it: from the draft when there is one."""
|
|
node = next(
|
|
(n for n in _read_flow(controller, name).nodes if n.id == node_id), None
|
|
)
|
|
if node is None:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
|
|
)
|
|
return node
|
|
|
|
|
|
@router.post("/{name}/nodes/{node_id}/train", response_model=Training, status_code=202)
|
|
async def train_node(
|
|
name: str,
|
|
node_id: str,
|
|
request: Request,
|
|
controller: FlowControllerDep,
|
|
user: CurrentUser,
|
|
) -> Any:
|
|
"""Train a new model for a wren node, as a run of wren's training flow.
|
|
|
|
It learns the node's targets, its ``examples`` and its ``tests``, as the
|
|
editor shows them. The node keeps running its model until the new one is
|
|
set as its ``model``.
|
|
"""
|
|
node = _node_def(controller, name, node_id)
|
|
if node.type != "wren":
|
|
raise HTTPException(status_code=400, detail=f"'{node_id}' is not a wren node")
|
|
try:
|
|
wren = WrenNode(
|
|
requires=[spec for spec in node.requires if spec.name],
|
|
provides=[spec for spec in node.provides if spec.name],
|
|
params=node.params,
|
|
)
|
|
except (ValueError, RuntimeError) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
service = getattr(request.app.state, "run_service", None)
|
|
if service is None:
|
|
raise HTTPException(status_code=503, detail="Runs are not available")
|
|
try:
|
|
run = await run_in_threadpool(
|
|
service.submit,
|
|
TRAIN_FLOW,
|
|
params={"installation": orjson.dumps(wren.installation()).decode()},
|
|
cause="train",
|
|
actor=user.email,
|
|
no_cache=True,
|
|
for_node=f"{name}.{node_id}",
|
|
)
|
|
except FlowNotFound:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=(
|
|
f"wren's training flow '{TRAIN_FLOW}' is not registered. Run "
|
|
"`fluksio sync wren.flow` from an environment with wren[train]."
|
|
),
|
|
) from None
|
|
except RunRejected as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
return Training(id=run.id, status=run.status)
|
|
|
|
|
|
@router.get("/{name}/nodes/{node_id}/train", response_model=Training | None)
|
|
def read_training(
|
|
name: str,
|
|
node_id: str,
|
|
request: Request,
|
|
controller: FlowControllerDep,
|
|
session: SessionDep,
|
|
) -> Any:
|
|
"""A wren node's latest training run, or null when it has none."""
|
|
_node_def(controller, name, node_id)
|
|
run = session.exec(
|
|
select(Run)
|
|
.where(col(Run.flow) == TRAIN_FLOW, col(Run.for_node) == f"{name}.{node_id}")
|
|
.order_by(col(Run.created_at).desc())
|
|
).first()
|
|
if run is None:
|
|
return None
|
|
training = Training(id=run.id, status=run.status, result=run.result)
|
|
if run.status == "error":
|
|
failed = session.exec(
|
|
select(RunNode.error).where(
|
|
col(RunNode.run_id) == run.id, col(RunNode.error) != ""
|
|
)
|
|
).first()
|
|
training.error = failed or run.status_reason
|
|
# The model, once the run made one the store still has.
|
|
pack = session.get(RunArtifact, {"run_id": run.id, "name": f"{TRAIN_FLOW}.pack"})
|
|
store = getattr(request.app.state, "artifact_store", None)
|
|
path = store.path(pack.digest) if pack is not None and store else None
|
|
if pack is not None and path is not None:
|
|
meta = pack_meta(path)
|
|
training.model = pack.digest
|
|
training.threshold = meta.get("threshold")
|
|
training.test_exact_match = meta.get("test_exact_match")
|
|
return training
|
|
|
|
|
|
@router.get("/{name}/state", response_model=FlowStatePublic)
|
|
def read_flow_state(name: str, controller: FlowControllerDep, hidden: HiddenDep) -> Any:
|
|
"""The last value seen on every message of this flow."""
|
|
_read_flow(controller, name)
|
|
return _flow_state(controller, name, hidden)
|
|
|
|
|
|
@router.get("/{name}/history/{message}", response_model=MessageHistory)
|
|
def read_message_history(
|
|
name: str,
|
|
message: str,
|
|
controller: FlowControllerDep,
|
|
hidden: HiddenDep,
|
|
) -> Any:
|
|
"""The recent values of one message, for plotting.
|
|
|
|
``message`` may be given bare or qualified; a message that never carried a
|
|
number or a flag comes back with an empty series.
|
|
"""
|
|
_read_flow(controller, name)
|
|
key = qualify(name, message)
|
|
if key in hidden:
|
|
raise HTTPException(status_code=403, detail=CAMERA_IS_SUPERUSERS)
|
|
return MessageHistory(
|
|
message=key,
|
|
numeric=as_number(controller.state.get(key)) is not None,
|
|
points=[
|
|
HistoryPoint(ts=ts, value=value)
|
|
for ts, value in controller.state.history(key)
|
|
],
|
|
)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Live updates
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
def snapshot_payload(
|
|
controller: FlowController,
|
|
only: set[str] | None = None,
|
|
device: str = "",
|
|
uplink: str = "",
|
|
hidden: frozenset[str] = frozenset(),
|
|
) -> dict[str, Any]:
|
|
"""Everything a client needs to catch up, sent the moment it connects.
|
|
|
|
Also sent by the tunnel connector, which serves this websocket inline: the
|
|
two have to agree, so they build the message here rather than each their
|
|
own. ``emits`` is what the bus counted while nobody was listening — a
|
|
client that reconnects between two pages would otherwise start from zero.
|
|
``queued`` is the nodes waiting for a machine, for the same reason: the bus
|
|
announces a wait once, and a client that missed it would draw the node idle.
|
|
|
|
``only`` bounds it to a set of message names, which is what a wall panel
|
|
gets: the values its own dashboards draw, and none of the rest — no node
|
|
status, no logs, no shape of the graph. A panel renders none of that, and
|
|
a screen may be hanging somewhere nobody here can see.
|
|
|
|
A device's socket also says which of its ports have ``demand`` now, and
|
|
which ``uplink`` its frames take: ``binary`` on its own socket, ``http``
|
|
through the portal, whose tunnel carries text only.
|
|
|
|
``hidden`` are the camera messages a person may not see, left out.
|
|
"""
|
|
if only is not None:
|
|
# The same keys, emptied rather than dropped: a screen already hanging
|
|
# runs whatever bundle it was paired with, and the shape of this
|
|
# message is what that bundle reads.
|
|
payload: dict[str, Any] = {
|
|
"type": "snapshot",
|
|
"values": {k: v for k, v in controller.values().items() if k in only},
|
|
"nodes": [],
|
|
"issues": [],
|
|
"paused": [],
|
|
"logs": [],
|
|
"emits": {},
|
|
"queued": [],
|
|
}
|
|
if device:
|
|
payload["demand"] = demand.ports(device)
|
|
payload["uplink"] = uplink
|
|
return payload
|
|
values = controller.values()
|
|
return {
|
|
"type": "snapshot",
|
|
"values": (
|
|
{k: v for k, v in values.items() if k not in hidden} if hidden else values
|
|
),
|
|
"nodes": [s.model_dump() for s in controller.node_statuses()],
|
|
"issues": [i.model_dump() for i in controller.issues],
|
|
"paused": controller.paused_flows(),
|
|
"logs": list(event_bus.recent_logs),
|
|
"emits": dict(event_bus.emits),
|
|
"queued": controller.placer.snapshot()["waiting"] if controller.placer else [],
|
|
}
|
|
|
|
|
|
def demand_frame(ports: list[str]) -> str:
|
|
"""What a device is told when its ports with demand change."""
|
|
return orjson.dumps({"type": "demand", "ports": ports}).decode()
|
|
|
|
|
|
def device_of(token: str, app: Any) -> str:
|
|
"""The device this credential stands for, or "" for a person or none."""
|
|
try:
|
|
return str(decode_token(token, app=app).get("panel") or "")
|
|
except InvalidTokenError:
|
|
return ""
|
|
|
|
|
|
class Credential:
|
|
"""Whether a device socket's credential still stands, asked per frame.
|
|
|
|
Checked in full only once what it is checked against has moved: the
|
|
device files, which a device's save, deletion or a revert commits to the
|
|
flow store, and the nonces in ``devices.json``, which an unpair bumps.
|
|
Otherwise a frame costs a stat of that file.
|
|
"""
|
|
|
|
def __init__(self, token: str, app: Any, device_id: str) -> None:
|
|
self.token = token
|
|
self.app = app
|
|
self.device = device_id
|
|
self._seen: tuple[int, int] | None = None
|
|
|
|
def _key(self) -> tuple[int, int] | None:
|
|
devices: DeviceStore | None = getattr(self.app.state, "device_store", None)
|
|
if devices is None:
|
|
return None
|
|
try:
|
|
return devices.flows.revision, devices.nonce(self.device)
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
async def revoked(self) -> bool:
|
|
key = self._key()
|
|
if key is not None and key == self._seen:
|
|
return False
|
|
if await run_in_threadpool(revoked, self.token, self.app):
|
|
return True
|
|
# The key read before the check, so a change meanwhile is checked.
|
|
self._seen = key
|
|
return False
|
|
|
|
|
|
def revoked(token: str, app: Any) -> bool:
|
|
"""Whether a device's credential has stopped standing since its handshake.
|
|
|
|
Unpaired or deleted. One that cannot be checked right now has not: a
|
|
device closed with 1008 checks whether it is still paired.
|
|
"""
|
|
try:
|
|
decode_token(token, app=app)
|
|
except InvalidTokenError:
|
|
return True
|
|
except DeviceUnavailable:
|
|
return False
|
|
return False
|
|
|
|
|
|
def panel_scope(token: str, app: Any) -> set[str] | None:
|
|
"""The messages this credential is bounded to, or None if it is a person's.
|
|
|
|
The socket is the one authenticated surface the route check cannot reach,
|
|
a handshake has no route to judge, so a device is bounded by what it is
|
|
sent instead of by what it asks for.
|
|
|
|
Its own camera's frames are left out unless a dashboard it shows draws
|
|
them: it sent them, and a camera's values come several a second.
|
|
"""
|
|
device_id = device_of(token, app)
|
|
if not device_id:
|
|
return None
|
|
store: DashboardStore | None = getattr(app.state, "dashboard_store", None)
|
|
devices: DeviceStore | None = getattr(app.state, "device_store", None)
|
|
if store is None or devices is None:
|
|
return set()
|
|
device = devices.find(device_id)
|
|
found = messages_for(device, store, devices)
|
|
if device is not None:
|
|
shown = {
|
|
name for defn in dashboards_for(device, store) for name in messages_of(defn)
|
|
}
|
|
found -= {
|
|
b.message
|
|
for b in devices.bound(device.id)
|
|
if b.direction == "in" and b.dtype == "image"
|
|
} - shown
|
|
return found
|
|
|
|
|
|
def event_for_panel(event: dict[str, Any], only: set[str]) -> bool:
|
|
"""Whether a panel's socket should carry this event.
|
|
|
|
The same bound as the snapshot above, applied to the stream that follows
|
|
it: a value the panel draws, and nothing else on the bus — save for a
|
|
dashboard being published, which is how a screen hears that the document
|
|
it is drawing, or the set of them it was given, has moved.
|
|
"""
|
|
if event.get("type") == "dashboard_changed":
|
|
return True
|
|
return event.get("type") == "message_value" and str(event.get("name") or "") in only
|
|
|
|
|
|
def carried(
|
|
event: dict[str, Any], only: set[str] | None, hidden: frozenset[str]
|
|
) -> bool:
|
|
"""Whether a socket carries this event on the bus.
|
|
|
|
A panel's what it draws (``event_for_panel``), a person's everything but
|
|
the values of the cameras ``hidden`` from them.
|
|
"""
|
|
if only is not None:
|
|
return event_for_panel(event, only)
|
|
return not (
|
|
hidden and event.get("type") == "message_value" and event.get("name") in hidden
|
|
)
|
|
|
|
|
|
# How many events one frame may carry. A ceiling rather than a target: the
|
|
# batch is whatever the bus happens to hold, and a client that has been away
|
|
# should not be handed the whole queue in one message.
|
|
MAX_FRAME_EVENTS = 64
|
|
|
|
#: How many message names one socket may ask for bytes on. A screen draws a
|
|
#: handful of tiles, and every live one keeps a camera sending, so a phone
|
|
#: scrolled past several cameras does not pull them all. The rest are refused.
|
|
MAX_MEDIA_NAMES = 8
|
|
|
|
|
|
def wanted_names(
|
|
frame: str, only: set[str] | None, hidden: frozenset[str] = frozenset()
|
|
) -> set[str] | None:
|
|
"""What this client is asking to be sent bytes for, or None if it said
|
|
something else.
|
|
|
|
A client asks by name — the tiles it is drawing — and is answered only for
|
|
the names a panel credential would have been allowed anyway, and none of
|
|
the ``hidden`` cameras. Bytes are the expensive thing on this socket, so
|
|
nothing is pushed until something says it is looking at it.
|
|
|
|
Past the cap the names it gave last are kept: a client lists the tile that
|
|
came on screen most recently last, so one scrolled back to keeps working.
|
|
"""
|
|
try:
|
|
payload = orjson.loads(frame)
|
|
except orjson.JSONDecodeError:
|
|
return None
|
|
if not isinstance(payload, dict) or payload.get("type") != "media":
|
|
return None
|
|
names = payload.get("names")
|
|
if not isinstance(names, list):
|
|
return set()
|
|
asked = {str(name) for name in names}
|
|
if len(asked) > MAX_MEDIA_NAMES:
|
|
kept = list(dict.fromkeys(str(name) for name in names))[-MAX_MEDIA_NAMES:]
|
|
logger.warning(
|
|
"A socket asked for live media on %d messages; only %d are served, "
|
|
"the last it named: %s",
|
|
len(asked),
|
|
MAX_MEDIA_NAMES,
|
|
", ".join(kept),
|
|
)
|
|
asked = set(kept)
|
|
asked -= hidden
|
|
return asked if only is None else asked & only
|
|
|
|
|
|
def media_frames(
|
|
events: list[dict[str, Any]], wanted: set[str], store: Any
|
|
) -> list[bytes]:
|
|
"""The bytes behind the media values in this batch, one frame each.
|
|
|
|
Referenced-then-fetched costs a round trip per frame, which is what keeps
|
|
the message plane at a glance rather than a view. Pushing the bytes down
|
|
the socket that already carries the event closes that, and only for the
|
|
frames a client said it was drawing.
|
|
|
|
Only what the ring holds, out of memory: the durable store is what a fetch
|
|
is for, and a checkpoint has no business being pushed at anyone. The
|
|
newest value per name wins, so a client that fell behind is not handed a
|
|
backlog of frames it would only draw over.
|
|
"""
|
|
if not wanted or store is None or getattr(store, "volatile", None) is None:
|
|
return []
|
|
|
|
newest: dict[str, dict[str, Any]] = {}
|
|
for event in events:
|
|
if event.get("type") != "message_value":
|
|
continue
|
|
name = str(event.get("name") or "")
|
|
if name not in wanted or not is_reference(event.get("value")):
|
|
continue
|
|
newest[name] = event
|
|
|
|
frames: list[bytes] = []
|
|
for name, event in newest.items():
|
|
value = event["value"]
|
|
digest = str(value.get("digest") or "")
|
|
payload = store.volatile.get(digest)
|
|
if payload is None:
|
|
continue
|
|
header = orjson.dumps(
|
|
{
|
|
"type": "media",
|
|
"name": name,
|
|
"digest": digest,
|
|
"media_type": str(value.get("media_type") or ""),
|
|
"ts": event.get("ts"),
|
|
}
|
|
)
|
|
# Length-prefixed, so one frame carries both halves and the reader
|
|
# never has to guess where the JSON stops.
|
|
frames.append(len(header).to_bytes(4, "big") + header + payload)
|
|
return frames
|
|
|
|
|
|
#: What a device may say a frame is: a media type, parameters allowed
|
|
#: (``audio/L16;rate=16000``).
|
|
_MEDIA_TYPE = re.compile(r"^[\w.+-]+/[\w.+-]+(;[\w.+-]+=[\w.+-]+)*$")
|
|
|
|
|
|
def parse_frame(data: bytes) -> tuple[dict[str, Any], bytes] | None:
|
|
"""Split a binary frame into its JSON header and the bytes after it.
|
|
|
|
The layout media frames go out in: four bytes of header length, big
|
|
endian, the header, then the bytes. None for anything else.
|
|
"""
|
|
if len(data) < 4:
|
|
return None
|
|
length = int.from_bytes(data[:4], "big")
|
|
if length == 0 or 4 + length > len(data):
|
|
return None
|
|
try:
|
|
header = orjson.loads(data[4 : 4 + length])
|
|
except orjson.JSONDecodeError:
|
|
return None
|
|
if not isinstance(header, dict):
|
|
return None
|
|
return header, data[4 + length :]
|
|
|
|
|
|
#: Why a device's frame was dropped, by the key it is logged under once per
|
|
#: socket. Fixed, so nothing the device chose reaches the log.
|
|
FRAME_REFUSALS = {
|
|
"not_a_frame": "not a frame",
|
|
"not_a_device": "only a device sends frames",
|
|
"no_engine": "the engine is not running",
|
|
"not_wired": "it names no message one of the device's ports publishes",
|
|
"not_media": "the port it names is not wired as an image or audio",
|
|
"media_type": "its media type does not match its port",
|
|
"too_large": "it is larger than MAX_FRAME_BYTES",
|
|
"not_scalar": "the port it names is not wired as a bool, int, float or str",
|
|
"undeclared": "no flow declares its message",
|
|
"refused": "its message refused the value",
|
|
}
|
|
|
|
|
|
def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str:
|
|
"""Store a frame a device sent on its socket and publish it.
|
|
|
|
Returns the key in ``FRAME_REFUSALS`` it was refused for, or "" once it
|
|
is published.
|
|
|
|
The same frame over HTTP is two requests, each authenticated: the upload,
|
|
then the publish. Here the socket was authenticated once, at its
|
|
handshake, and the frame is held and published in one step. Only a
|
|
message one of the device's own ``in`` ports is wired to, as an ``image``
|
|
or an ``audio`` chunk, and the value is credited to the device node holding that port, as an
|
|
HTTP publish is. The frame is named after that message.
|
|
"""
|
|
cap = settings.MAX_FRAME_BYTES
|
|
if cap and len(data) > cap:
|
|
return "too_large"
|
|
parsed = parse_frame(data)
|
|
if parsed is None or parsed[0].get("type") != "frame":
|
|
return "not_a_frame"
|
|
header, payload = parsed
|
|
name = str(header.get("name") or "")
|
|
devices: DeviceStore | None = getattr(app.state, "device_store", None)
|
|
store: Any = getattr(app.state, "artifact_store", None)
|
|
controller: FlowController | None = getattr(app.state, "flow_controller", None)
|
|
if devices is None or store is None or controller is None:
|
|
return "no_engine"
|
|
binding = _bound_in(devices, device, name)
|
|
if binding is None:
|
|
return "not_wired"
|
|
if binding.dtype not in (DType.IMAGE, DType.AUDIO):
|
|
return "not_media"
|
|
media_type = str(header.get("media_type") or "")
|
|
if not _MEDIA_TYPE.match(media_type) or not media_type.startswith(
|
|
MEDIA_FAMILIES[DType(binding.dtype)]
|
|
):
|
|
return "media_type"
|
|
reference = store.put([payload], name=name, media_type=media_type, volatile=True)
|
|
try:
|
|
controller.publish_message(name, reference, port_source(devices, device, name))
|
|
except KeyError:
|
|
return "undeclared"
|
|
except TypeError:
|
|
return "refused"
|
|
return ""
|
|
|
|
|
|
def _bound_in(devices: DeviceStore, device: DeviceDef, name: str) -> Binding | None:
|
|
"""The device's own ``in`` port wired to ``name``, if one is."""
|
|
return next(
|
|
(
|
|
b
|
|
for b in devices.bound(device.id)
|
|
if b.direction == "in" and b.message == name
|
|
),
|
|
None,
|
|
)
|
|
|
|
|
|
def published(frame: str) -> dict[str, Any] | None:
|
|
"""A value a device publishes on its socket, or None for anything else."""
|
|
try:
|
|
payload = orjson.loads(frame)
|
|
except orjson.JSONDecodeError:
|
|
return None
|
|
if isinstance(payload, dict) and payload.get("type") == "publish":
|
|
return payload
|
|
return None
|
|
|
|
|
|
def ingest_publish(app: Any, device: DeviceDef, payload: dict[str, Any]) -> str:
|
|
"""Publish a value a device sent on its socket, as ``POST /messages`` would.
|
|
|
|
Returns the key in ``FRAME_REFUSALS`` it was refused for, or "" once it
|
|
is published. Only a message one of the device's own ``in`` ports is wired
|
|
to as a scalar, credited to the device node holding that port. Taken in
|
|
turn with the device's binary frames, so a ``listening`` false lands after
|
|
the last chunk of what was heard.
|
|
"""
|
|
name = str(payload.get("name") or "")
|
|
devices: DeviceStore | None = getattr(app.state, "device_store", None)
|
|
controller: FlowController | None = getattr(app.state, "flow_controller", None)
|
|
if devices is None or controller is None:
|
|
return "no_engine"
|
|
binding = _bound_in(devices, device, name)
|
|
if binding is None:
|
|
return "not_wired"
|
|
if binding.dtype not in (DType.BOOL, DType.INT, DType.FLOAT, DType.STR):
|
|
return "not_scalar"
|
|
try:
|
|
controller.publish_message(
|
|
name, payload.get("value"), port_source(devices, device, name)
|
|
)
|
|
except KeyError:
|
|
return "undeclared"
|
|
except TypeError:
|
|
return "refused"
|
|
return ""
|
|
|
|
|
|
async def _receive(websocket: WebSocket) -> str | bytes:
|
|
"""The client's next frame, text or binary."""
|
|
message = await websocket.receive()
|
|
if message["type"] == "websocket.disconnect":
|
|
raise WebSocketDisconnect(message.get("code", 1000), message.get("reason"))
|
|
data = message.get("bytes")
|
|
return data if data is not None else str(message.get("text") or "")
|
|
|
|
|
|
async def _send(websocket: WebSocket, events: list[dict[str, Any]]) -> None:
|
|
"""One event, or a batch of them under `events`.
|
|
|
|
Serialised once with orjson rather than per client with the stdlib, which
|
|
is what `send_json` does.
|
|
"""
|
|
payload = events[0] if len(events) == 1 else {"type": "batch", "events": events}
|
|
await websocket.send_text(orjson.dumps(payload).decode())
|
|
|
|
|
|
class SocketTicket(BaseModel):
|
|
ticket: str
|
|
expires_in: float
|
|
|
|
|
|
@router.post("/ws-ticket", response_model=SocketTicket)
|
|
def issue_socket_ticket(token: TokenDep) -> Any:
|
|
"""A one-time pass for opening `/flows/ws`, good for thirty seconds.
|
|
|
|
A browser cannot set a header on a websocket handshake, and a token in the
|
|
URL ends up in every access log on the way. The ticket goes there instead
|
|
and stands for the bearer token of this call, panel credentials included.
|
|
"""
|
|
return SocketTicket(ticket=issue_ticket(token), expires_in=TICKET_TTL_S)
|
|
|
|
|
|
@ws_router.websocket("/ws")
|
|
async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -> None:
|
|
"""Stream values, node status and execution events as they happen.
|
|
|
|
A browser brings a ticket from `POST /flows/ws-ticket`; anything that can
|
|
set headers sends `Authorization: Bearer`. `?token=` is still read for one
|
|
release, for clients from before either.
|
|
|
|
A client says which media it is drawing (``{"type": "media", "names":
|
|
[...]}``) and is pushed those frames. A device also sends its camera's
|
|
frames here, as binary frames in the layout it is pushed them in, and
|
|
the values of its scalar ports as ``publish`` messages, and is told when
|
|
its on-demand ports have demand.
|
|
|
|
A live view's signalling rides here too (``fluksio.flow.live``): ``live``
|
|
messages on a person's socket, ``signal`` messages on a device's.
|
|
"""
|
|
token = socket_token(websocket.headers, ticket, token)
|
|
|
|
# On a thread: authenticating is a database round trip, and the snapshot
|
|
# below reads the whole of state. Neither belongs on the event loop, which
|
|
# every other socket and every request is sharing.
|
|
def _authenticate() -> Any:
|
|
with Session(engine) as session:
|
|
user = user_from_token(session, token, app=websocket.app)
|
|
only = panel_scope(token, websocket.app) if user is not None else None
|
|
# A device is online for as long as this socket is open.
|
|
device = device_of(token, websocket.app) if only is not None else ""
|
|
return user, only, device, _device_def(websocket.app, device)
|
|
|
|
try:
|
|
user, only, device, device_def = await run_in_threadpool(_authenticate)
|
|
except Exception as exc:
|
|
# Not a refusal: the database or the device store could not say. A
|
|
# device takes 1008, or a 403 before accepting, as its credential
|
|
# being gone and checks it, so this one is told to try again later.
|
|
logger.warning("Could not check a socket's credential: %s", exc)
|
|
await websocket.accept()
|
|
await websocket.close(code=1013)
|
|
return
|
|
if user is None:
|
|
await websocket.close(code=1008)
|
|
return
|
|
# Nothing until a client says it is drawing something: bytes are what this
|
|
# socket cannot afford to send speculatively.
|
|
wanted: set[str] = set()
|
|
# This socket's entry in the registry every socket's wants are kept in.
|
|
viewer = object()
|
|
# Each reason a frame was refused for, logged once per socket.
|
|
refused: set[str] = set()
|
|
# The cameras a person may not see. Looked at again with each batch, so
|
|
# a device switched to members, or back, takes effect on open sockets.
|
|
hidden = hidden_from(user, websocket.app, device=only is not None)
|
|
# A device's credential, checked again per frame once it may have moved.
|
|
credential = Credential(token, websocket.app, device) if device else None
|
|
|
|
await websocket.accept()
|
|
|
|
controller: FlowController | None = getattr(
|
|
websocket.app.state, "flow_controller", None
|
|
)
|
|
store = getattr(websocket.app.state, "artifact_store", None)
|
|
|
|
async def send_snapshot() -> list[str]:
|
|
"""Send the snapshot, and return the demand it told a device of."""
|
|
if controller is None:
|
|
return []
|
|
payload = await run_in_threadpool(
|
|
snapshot_payload, controller, only, device, "binary", hidden
|
|
)
|
|
await websocket.send_text(orjson.dumps(payload).decode())
|
|
return list(payload.get("demand", []))
|
|
|
|
async def catch_up() -> bool:
|
|
"""Rescope a panel and send the snapshot again, demand included.
|
|
|
|
False once a device whose credential stopped standing is closed with
|
|
1008, which sends it to check: otherwise its frames would still be
|
|
taken here.
|
|
"""
|
|
nonlocal only, device_def, wanted
|
|
if only is not None:
|
|
if device and await run_in_threadpool(revoked, token, websocket.app):
|
|
await websocket.close(code=1008)
|
|
return False
|
|
# The scope was resolved once, at the handshake. A panel pointed
|
|
# at another dashboard would otherwise fetch the new document and
|
|
# then draw tiles nothing ever updates. ``or set()`` because a
|
|
# panel that was deleted resolves to None, the same as a person's
|
|
# token — and that would widen this socket to everything on the
|
|
# bus.
|
|
only = panel_scope(token, websocket.app) or set()
|
|
device_def = _device_def(websocket.app, device)
|
|
wanted &= only
|
|
demand.want(viewer, wanted)
|
|
await send_snapshot()
|
|
return True
|
|
|
|
told = await send_snapshot()
|
|
|
|
async with event_bus.subscribe() as queue:
|
|
receiver = asyncio.create_task(_receive(websocket))
|
|
# Where a live view this socket is one end of reaches it. Its own
|
|
# outbox rather than the bus's queue, which drops when full.
|
|
endpoint = live_view.Endpoint()
|
|
if device:
|
|
presence.opened(device)
|
|
# Demand that moved between the snapshot and the subscription
|
|
# would otherwise wait for the next change.
|
|
ports = demand.ports(device)
|
|
if ports != told:
|
|
await websocket.send_text(demand_frame(ports))
|
|
live_view.sessions.attach(device, endpoint)
|
|
outbox = asyncio.create_task(endpoint.next())
|
|
try:
|
|
while True:
|
|
sender = asyncio.create_task(queue.get())
|
|
done, _ = await asyncio.wait(
|
|
{sender, receiver, outbox}, return_when=asyncio.FIRST_COMPLETED
|
|
)
|
|
if outbox in done:
|
|
# Signalling, for this socket alone.
|
|
if sender not in done:
|
|
sender.cancel()
|
|
text = outbox.result()
|
|
outbox = asyncio.create_task(endpoint.next())
|
|
await websocket.send_text(text)
|
|
if receiver not in done and sender not in done:
|
|
continue
|
|
if receiver in done:
|
|
if sender not in done:
|
|
# Nobody will ever read this one: the next turn makes
|
|
# its own. Left behind it stays parked on the queue's
|
|
# waiter list, takes an event the live reader wanted,
|
|
# and is reported as a task destroyed while pending.
|
|
sender.cancel()
|
|
# A frame from the client. Only a disconnect ends the
|
|
# stream — a keepalive, or anything else it decides to
|
|
# say, used to be read as the client going away and cost
|
|
# it every live update from then on.
|
|
failure = receiver.exception()
|
|
if failure is not None:
|
|
# The client is gone, and `receive` raises for good
|
|
# from here on. Reading it again would spin this loop
|
|
# at full tilt against a dead socket.
|
|
raise failure
|
|
frame = receiver.result()
|
|
if isinstance(frame, bytes):
|
|
# A camera's frame. Refused ones are dropped and the
|
|
# socket stays open: one bad frame is not a reason to
|
|
# take the device's live values away. A device that
|
|
# was unpaired meanwhile is closed instead.
|
|
if credential is not None and await credential.revoked():
|
|
await websocket.close(code=1008)
|
|
return
|
|
reason = (
|
|
await run_in_threadpool(
|
|
ingest_frame, websocket.app, device_def, frame
|
|
)
|
|
if device_def is not None
|
|
else "not_a_device"
|
|
)
|
|
if reason and reason not in refused:
|
|
refused.add(reason)
|
|
logger.warning(
|
|
"Dropped a frame from %s: %s",
|
|
device or "a person's socket",
|
|
FRAME_REFUSALS[reason],
|
|
)
|
|
else:
|
|
asked = wanted_names(frame, only, hidden)
|
|
if asked is not None:
|
|
wanted = asked
|
|
demand.want(viewer, wanted)
|
|
elif (
|
|
device_def is not None
|
|
and (value := published(frame)) is not None
|
|
):
|
|
if credential is not None and await credential.revoked():
|
|
await websocket.close(code=1008)
|
|
return
|
|
reason = await run_in_threadpool(
|
|
ingest_publish, websocket.app, device_def, value
|
|
)
|
|
if reason and reason not in refused:
|
|
refused.add(reason)
|
|
logger.warning(
|
|
"Dropped a value from %s: %s",
|
|
device,
|
|
FRAME_REFUSALS[reason],
|
|
)
|
|
elif (signal := live_view.parse(frame)) is not None:
|
|
live_view.sessions.heard(
|
|
signal,
|
|
endpoint,
|
|
person=str(user.id) if only is None else "",
|
|
device=device,
|
|
devices=getattr(
|
|
websocket.app.state, "device_store", None
|
|
),
|
|
superuser=user.is_superuser,
|
|
)
|
|
receiver = asyncio.create_task(_receive(websocket))
|
|
if sender not in done:
|
|
continue
|
|
# `sender` is done from here: the turns where it is not have
|
|
# gone round again above.
|
|
|
|
# Everything the bus has right now, not just the one event
|
|
# that woke this: a cascade puts a dozen in at once, and one
|
|
# frame carrying them costs one wakeup rather than a dozen.
|
|
batch = [sender.result()]
|
|
while len(batch) < MAX_FRAME_EVENTS:
|
|
try:
|
|
batch.append(queue.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
|
|
# A panel this far behind lost events the bus dropped for it:
|
|
# its demand, a dashboard published or its own unpairing. It
|
|
# catches up once this batch is out, as after a dashboard.
|
|
resync = only is not None and event_bus.dropped(queue)
|
|
if only is None:
|
|
fresh = hidden_from(user, websocket.app)
|
|
if fresh != hidden:
|
|
# A camera was opened or closed to members: the
|
|
# snapshot gives back what may now be seen.
|
|
hidden = fresh
|
|
wanted -= hidden
|
|
demand.want(viewer, wanted)
|
|
resync = True
|
|
|
|
out: list[dict[str, Any]] = []
|
|
for event in batch:
|
|
kind = event.get("type")
|
|
if kind == "device_demand":
|
|
# For the device it names, and for nobody else.
|
|
if device and event.get("device") == device:
|
|
await websocket.send_text(
|
|
demand_frame(list(event.get("ports") or []))
|
|
)
|
|
continue
|
|
if only is not None and kind == "dashboard_changed":
|
|
if out:
|
|
await _send(websocket, out)
|
|
out = []
|
|
if not await catch_up():
|
|
return
|
|
if carried(event, only, hidden):
|
|
out.append(event)
|
|
if out:
|
|
# The bytes first: a tile that has the frame when the
|
|
# value lands draws it in one pass rather than two.
|
|
for frame in media_frames(out, wanted, store):
|
|
await websocket.send_bytes(frame)
|
|
await _send(websocket, out)
|
|
if resync and not await catch_up():
|
|
return
|
|
except (WebSocketDisconnect, RuntimeError):
|
|
# A peer that goes away mid-send takes the RuntimeError route
|
|
# ("websocket.send after websocket.close") rather than the clean
|
|
# disconnect. Either way the socket is gone and the loop is over.
|
|
pass
|
|
finally:
|
|
receiver.cancel()
|
|
outbox.cancel()
|
|
demand.drop(viewer)
|
|
live_view.sessions.detach(endpoint)
|
|
if device:
|
|
presence.closed(device)
|
|
|
|
|
|
def _device_def(app: Any, device_id: str) -> DeviceDef | None:
|
|
"""The device a socket stands for, as its frames are credited to it."""
|
|
devices: DeviceStore | None = getattr(app.state, "device_store", None)
|
|
if not device_id or devices is None:
|
|
return None
|
|
return devices.find(device_id)
|