From 02b736c9e8e831015aac1bc4bcbb282b22853389 Mon Sep 17 00:00:00 2001 From: Melvin Strobl Date: Sat, 17 Oct 2026 08:37:39 +0200 Subject: [PATCH] Live view over WebRTC (openpanel L1, server and browser) A device reporting a `live` port streams its camera to the browser's own RTCPeerConnection. The flows socket carries the signalling as text, so it crosses the portal: `live` messages on a person's socket, `signal` on a device's (start, sdp, candidate, bye, busy). The server routes by session, holds a device to two viewers, ends a session when either socket closes or no answer comes within 10 s, and gives each socket an outbox of its own that ends a session rather than drop a message. TURN credentials are minted per session in coturn's REST scheme (STUN_URLS, TURN_URLS, TURN_SECRET), the viewer's with the device's offer; GET /devices/{id}/ice hands them out too. The tunnel keeps a stream's early frames instead of dropping them past eight and forgets them when the stream or link goes. A media tile on a device's camera, and the Devices tab's preview, play live when tapped, pause frames meanwhile, and fall back to frames with a short note on busy, failure or no picture within 10 s. coturn is an optional compose service behind the `turn` profile. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0138N7k5PT4dcR1iEqXaaQ9k --- .env.example | 7 + backend/fluksio/api/routes/devices.py | 32 ++ backend/fluksio/api/routes/flows.py | 36 +- backend/fluksio/cloud/connector.py | 72 ++- backend/fluksio/core/config.py | 12 + backend/fluksio/flow/live.py | 365 +++++++++++++++ backend/tests/api/routes/test_live.py | 441 ++++++++++++++++++ docker/compose.yml | 51 ++ docs/code/api.md | 32 ++ docs/interface/dashboards.md | 24 +- docs/reference/configuration.md | 18 + frontend/src/client/schemas.gen.ts | 57 +++ frontend/src/client/sdk.gen.ts | 30 +- frontend/src/client/types.gen.ts | 20 + .../src/components/Dashboard/MediaWidget.tsx | 123 ++++- .../src/components/Dashboard/useLiveView.ts | 153 ++++++ .../src/components/Devices/DeviceDetail.tsx | 9 +- frontend/src/components/Devices/ports.ts | 9 +- .../src/components/Flow/CommandPalette.tsx | 10 +- frontend/src/components/Flow/useFlowSocket.ts | 58 +++ 20 files changed, 1535 insertions(+), 24 deletions(-) create mode 100644 backend/fluksio/flow/live.py create mode 100644 backend/tests/api/routes/test_live.py create mode 100644 frontend/src/components/Dashboard/useLiveView.ts diff --git a/.env.example b/.env.example index 0ed4641..ac40181 100644 --- a/.env.example +++ b/.env.example @@ -67,3 +67,10 @@ SIGNUP_ENABLED=false # Local time for schedules; set from the root .env by scripts/setup.sh. TZ=UTC + +# A camera's live view between networks: STUN and TURN servers, comma +# separated, and the secret shared with the TURN server (coturn's +# static-auth-secret). Empty is direct connections only. +STUN_URLS= +TURN_URLS= +TURN_SECRET= diff --git a/backend/fluksio/api/routes/devices.py b/backend/fluksio/api/routes/devices.py index 1815f80..be0548b 100644 --- a/backend/fluksio/api/routes/devices.py +++ b/backend/fluksio/api/routes/devices.py @@ -71,6 +71,7 @@ from fluksio.flow.devices import ( read_report, ) from fluksio.flow.events import event_bus +from fluksio.flow.live import ICE_TTL_S, ice_servers from fluksio.flow.state import CONNECT_TIMEOUT_S, SOCKET_TIMEOUT_S from fluksio.flow.store import StaleVersion from fluksio.models import Message, User @@ -387,6 +388,20 @@ class DevicesPublic(BaseModel): frontend_host: str = "" +class IceServer(BaseModel): + """One STUN or TURN server, as `RTCPeerConnection` takes it.""" + + urls: list[str] + username: str | None = None + credential: str | None = None + + +class IceServers(BaseModel): + ice_servers: list[IceServer] = Field(default_factory=list) + #: Seconds the TURN credentials stay good. + ttl: int = 0 + + def _describe(request: Request) -> str: """A line naming the device behind a pairing request. @@ -665,6 +680,23 @@ async def read_device(device_id: str, store: DeviceStoreDep) -> Any: return await run_in_threadpool(_document, device, store) +@router.get( + "/{device_id}/ice", + response_model=IceServers, + response_model_exclude_none=True, +) +async def read_ice(device_id: str, store: DeviceStoreDep, user: CurrentUser) -> Any: + """Where a browser looks for a way to this device's live view. + + STUN and TURN servers, the TURN ones with credentials of the caller's own + that expire. Empty when none are set, which is enough on one network. + """ + if await run_in_threadpool(store.find, device_id) is None: + raise _not_found(device_id) + servers = [IceServer(**server) for server in ice_servers(str(user.id))] + return IceServers(ice_servers=servers, ttl=ICE_TTL_S) + + panels_router.add_api_route( "/{device_id}", read_device, diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 82a5e43..7cf7571 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -37,6 +37,7 @@ 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 @@ -1369,6 +1370,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - [...]}``) 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 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) @@ -1425,6 +1429,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - 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 @@ -1432,12 +1439,23 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - 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}, return_when=asyncio.FIRST_COMPLETED + {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 @@ -1479,11 +1497,21 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - if asked is not None: wanted = asked demand.want(viewer, wanted) + 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 + ), + ) receiver = asyncio.create_task(_receive(websocket)) if sender not in done: continue - # `sender` is done from here: `FIRST_COMPLETED` returns with at - # least one of the two, so `receiver` missing means this one is. + # `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 @@ -1544,7 +1572,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - pass finally: receiver.cancel() + outbox.cancel() demand.drop(viewer) + live_view.sessions.detach(endpoint) if device: presence.closed(device) diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index b7b8ac6..42ead1e 100644 --- a/backend/fluksio/cloud/connector.py +++ b/backend/fluksio/cloud/connector.py @@ -73,7 +73,9 @@ ALLOWED_PATHS = frozenset( #: reference to a task that was still running. MAX_IN_FLIGHT = 256 #: How many of a client's frames are kept for a stream not listening yet. -MAX_EARLY = 8 +#: One past it closes the stream rather than losing a frame: a live view's +#: candidates are worth nothing with one missing. +MAX_EARLY = 64 #: How often to look for a config that appeared while the engine was up. #: Enrolment is something a person does and then waits on, so this is the #: delay they sit through — short enough not to be worth a restart. @@ -299,7 +301,20 @@ class CloudConnector: # A browser speaks as soon as the hub accepts it, # before this end has checked its credential. early = self._early.setdefault(stream, []) - early[:] = [*early[-(MAX_EARLY - 1) :], text] + if len(early) < MAX_EARLY: + early.append(text) + else: + logger.warning( + "Closing stream %s: %d frames before it listened", + stream, + MAX_EARLY, + ) + self._cancel_stream(stream) + await socket.send( + _dump( + {"op": "ws_close", "id": stream, "code": 1011} + ) + ) elif op == "ws_close": self._cancel_stream(str(frame.get("id") or "")) elif op == "member_removed": @@ -321,6 +336,7 @@ class CloudConnector: task.cancel() self._calls.clear() self._streams.clear() + self._early.clear() return attached @staticmethod @@ -631,12 +647,16 @@ class CloudConnector: "Refusing stream %s: %d already open", stream_id, len(self._streams) ) return - existing = self._streams.pop(stream_id, None) - if existing is not None: - existing.cancel() + self._cancel_stream(stream_id) task = asyncio.create_task(self._stream(socket, frame)) self._streams[stream_id] = task - task.add_done_callback(lambda _t: self._streams.pop(stream_id, None)) + task.add_done_callback(lambda done: self._stream_done(stream_id, done)) + + def _stream_done(self, stream_id: str, task: asyncio.Task[None]) -> None: + # Its own entry only: an id the hub reused belongs to the newer task. + if self._streams.get(stream_id) is task: + del self._streams[stream_id] + self._early.pop(stream_id, None) async def _stream(self, socket: Any, frame: dict[str, Any]) -> None: """Feed the engine's live events down the tunnel. @@ -661,6 +681,7 @@ class CloudConnector: ) from fluksio.api.socket_auth import redeem_ticket from fluksio.core.db import engine + from fluksio.flow import live as live_view from fluksio.flow.demand import demand from fluksio.flow.devices import presence from fluksio.flow.events import event_bus @@ -714,6 +735,11 @@ class CloudConnector: # are fetched one by one through the portal. viewer = ("tunnel", stream_id) wanted: set[str] = set() + # Where a live view this stream is one end of reaches it. Its + # signalling is text, so it crosses, on an outbox of its own. + endpoint = live_view.Endpoint() + outbox: asyncio.Task[str] | None = None + sender: asyncio.Task[dict[str, Any]] | None = None def heard(text: str) -> None: nonlocal wanted @@ -721,6 +747,14 @@ class CloudConnector: if asked is not None: wanted = asked demand.want(viewer, asked) + elif (signal := live_view.parse(text)) is not None: + live_view.sessions.heard( + signal, + endpoint, + person=str(user.id) if only is None else "", + device=device, + devices=getattr(self._app.state, "device_store", None), + ) self._listeners[stream_id] = heard for text in self._early.pop(stream_id, []): @@ -728,8 +762,10 @@ class CloudConnector: if device: presence.opened(device) + live_view.sessions.attach(device, endpoint) try: async with event_bus.subscribe() as queue: + outbox = asyncio.create_task(endpoint.next()) if device and (ports := demand.ports(device)) != told: await socket.send( _dump( @@ -741,7 +777,22 @@ class CloudConnector: ) ) while True: - event = await queue.get() + sender = asyncio.create_task(queue.get()) + done, _ = await asyncio.wait( + {sender, outbox}, return_when=asyncio.FIRST_COMPLETED + ) + if outbox in done: + # Signalling, for this stream alone. + if sender not in done: + sender.cancel() + text = outbox.result() + outbox = asyncio.create_task(endpoint.next()) + await socket.send( + _dump({"op": "ws_msg", "id": stream_id, "text": text}) + ) + if sender not in done: + continue + event = sender.result() if event.get("type") == "device_demand": # For the device it names, and for nobody else. if device and event.get("device") == device: @@ -793,9 +844,13 @@ class CloudConnector: _dump({"op": "ws_close", "id": stream_id, "code": 1011}) ) finally: + for task in (outbox, sender): + if task is not None: + task.cancel() self._listeners.pop(stream_id, None) self._early.pop(stream_id, None) demand.drop(viewer) + live_view.sessions.detach(endpoint) if device: presence.closed(device) @@ -867,6 +922,9 @@ class CloudConnector: ) def _cancel_stream(self, stream_id: str) -> None: + # The frames it was holding go with it, including those of a stream + # cancelled before it ever ran its own cleanup. + self._early.pop(stream_id, None) task = self._streams.pop(stream_id, None) if task is not None: task.cancel() diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index 2003cc2..205145e 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -229,6 +229,18 @@ class Settings(BaseSettings): list[AnyUrl] | str, BeforeValidator(parse_cors) ] = [] + #: Where a live view's two ends look for a way to each other, comma + #: separated. None by default: on one network the peers find each other + #: without, and a third-party server would learn who watches what. coturn + #: serves STUN as well, so its `stun:` address goes here. + STUN_URLS: Annotated[list[str] | str, BeforeValidator(parse_cors)] = [] + #: The relay a live view falls back to when the two ends cannot reach each + #: other, as `turn:` and `turns:` URLs. Used only with `TURN_SECRET`, the + #: secret the TURN server shares (coturn's `static-auth-secret`), from + #: which each viewer and device gets credentials of its own. + TURN_URLS: Annotated[list[str] | str, BeforeValidator(parse_cors)] = [] + TURN_SECRET: str = "" + @model_validator(mode="before") @classmethod def _derive_data_paths(cls, data: Any) -> Any: diff --git a/backend/fluksio/flow/live.py b/backend/fluksio/flow/live.py new file mode 100644 index 0000000..396356f --- /dev/null +++ b/backend/fluksio/flow/live.py @@ -0,0 +1,365 @@ +"""Live view: WebRTC signalling between a viewer's socket and a device's. + +A dashboard tile asks a device's camera for H.264 over WebRTC. The video goes +from the device to the browser directly, or through a TURN relay, never +through here: this only carries the few messages that set a connection up. +Both ends already hold a ``/flows/ws`` socket, which also crosses the portal +tunnel because it carries text, so signalling rides on it. + +A person's socket speaks ``{"type": "live", "device", "session", "action"}`` +and a device's ``{"type": "signal", "session", "action"}``. ``start`` from +the viewer is handed to the device with ICE servers of its own, the device +offers (``sdp``), which reaches the viewer with ICE servers of the viewer's +own, the viewer answers (``sdp``), both trickle ``candidate``, and ``bye`` +from either side ends it. A device past its cap answers ``busy``, and so does +this end once a device has ``MAX_SESSIONS``. + +Signalling never rides the event bus, which drops a slow socket's oldest +events: one lost candidate is a connection that never comes up. Each socket +is an :class:`Endpoint` with an outbox of its own that its loop drains, so +only that socket writes to its connection, and a full outbox ends the session +instead of losing a message. One process holds every socket, so one registry +is the whole picture. +""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import hmac +import logging +import re +import time +from dataclasses import dataclass, field +from typing import Any + +import orjson + +from fluksio.core.config import settings + +logger = logging.getLogger(__name__) + +#: The port a device reports when it can stream live. Signalled, never wired. +LIVE_PORT = "live" +#: Live viewers per device. The NANO encodes once for all of them, and S3 +#: measured two at 720p; past that a tile stays on frames. +MAX_SESSIONS = 2 +#: How long a session has to get from ``start`` to the viewer's answer. +ANSWER_TIMEOUT_S = 10.0 +#: How long TURN credentials stay good. Minted per session, for whoever is +#: one end of it. coturn checks them on every refresh, so a view relayed +#: through it ends this long after it started. +ICE_TTL_S = 3600 +#: Signalling messages one socket may have waiting. A session trickles a +#: dozen candidates at most, so a socket this far behind is stuck. +OUTBOX = 64 + +_SESSION = re.compile(r"^[A-Za-z0-9_-]{8,64}$") +#: An offer is one to three kilobytes, a candidate a line. +MAX_SDP = 16 * 1024 +MAX_CANDIDATE = 1024 +#: Which actions are passed on once a session stands. +_RELAYED = ("sdp", "candidate") + + +def ice_servers(who: str) -> list[dict[str, Any]]: + """The STUN and TURN servers for one end, with TURN credentials of its own. + + coturn's REST scheme (``use-auth-secret``): the username is + ``:`` and the password the base64 HMAC-SHA1 of it under the + shared secret, so coturn checks it without asking anyone. + """ + servers: list[dict[str, Any]] = [] + if settings.STUN_URLS: + servers.append({"urls": list(settings.STUN_URLS)}) + if settings.TURN_URLS and settings.TURN_SECRET: + username = f"{int(time.time()) + ICE_TTL_S}:{who}" + digest = hmac.new( + settings.TURN_SECRET.encode(), username.encode(), hashlib.sha1 + ).digest() + servers.append( + { + "urls": list(settings.TURN_URLS), + "username": username, + "credential": base64.b64encode(digest).decode(), + } + ) + return servers + + +def parse(frame: str) -> dict[str, Any] | None: + """A signalling message a socket sent, or None for anything else.""" + try: + payload = orjson.loads(frame) + except orjson.JSONDecodeError: + return None + if isinstance(payload, dict) and payload.get("type") in ("live", "signal"): + return payload + return None + + +class Endpoint: + """One socket signalling can reach, and what is waiting to go out on it.""" + + def __init__(self) -> None: + self._outbox: asyncio.Queue[str] = asyncio.Queue(maxsize=OUTBOX) + + def deliver(self, message: dict[str, Any]) -> bool: + """Put a message in the outbox. False when it is full.""" + try: + self._outbox.put_nowait(orjson.dumps(message).decode()) + except asyncio.QueueFull: + return False + return True + + async def next(self) -> str: + """The next message for this socket to send, as text.""" + return await self._outbox.get() + + +@dataclass(eq=False) +class Session: + id: str + device: str + viewer: Endpoint + camera: Endpoint + #: Who is watching, whose TURN credentials the viewer is handed. + who: str + offered: bool = False + answered: bool = False + started: float = field(default_factory=time.monotonic) + timer: asyncio.TimerHandle | None = None + + +class LiveSessions: + """Every live view being set up or running, and the devices that serve them. + + Called only from the event loop, by the sockets themselves. + """ + + def __init__(self) -> None: + self._sessions: dict[str, Session] = {} + #: Each device's open sockets, newest last. A device reconnecting can + #: hold two for a moment; a new session goes to the newer one. + self._devices: dict[str, list[Endpoint]] = {} + + def attach(self, device_id: str, endpoint: Endpoint) -> None: + """A device's socket opened.""" + self._devices.setdefault(device_id, []).append(endpoint) + + def detach(self, endpoint: Endpoint) -> None: + """A socket closed: every session it was one end of is over.""" + for device_id, endpoints in list(self._devices.items()): + if endpoint in endpoints: + endpoints.remove(endpoint) + if not endpoints: + del self._devices[device_id] + for session in list(self._sessions.values()): + if session.viewer is endpoint: + self._end(session, "closed", tell_viewer=False) + elif session.camera is endpoint: + self._end(session, "offline", tell_camera=False) + + def count(self, device_id: str) -> int: + return sum(1 for s in self._sessions.values() if s.device == device_id) + + def heard( + self, + message: dict[str, Any], + endpoint: Endpoint, + *, + person: str, + device: str, + devices: Any, + ) -> None: + """One signalling message from a socket. + + ``person`` is the id of the account holding the socket and ``device`` + the device whose credential does, one of them "". Any person may watch + any device, camera privacy being deferred. A device speaks only for + the sessions it was asked to serve on this socket. + """ + session_id = str(message.get("session") or "") + if not _SESSION.match(session_id): + return + if message.get("type") == "live" and person: + self._from_viewer(message, session_id, endpoint, person, devices) + elif message.get("type") == "signal" and device: + self._from_device(message, session_id, endpoint, device) + + # -- the two directions --------------------------------------------------- + + def _from_viewer( + self, + message: dict[str, Any], + session_id: str, + endpoint: Endpoint, + who: str, + devices: Any, + ) -> None: + device_id = str(message.get("device") or "") + action = message.get("action") + session = self._sessions.get(session_id) + if action == "start": + if session is not None: + # Said twice, or somebody else's id: neither gets a second one. + return + reason = self._refusal(device_id, devices) + if reason: + endpoint.deliver( + _viewer_message(device_id, session_id, "bye", reason=reason) + ) + return + if self.count(device_id) >= MAX_SESSIONS: + endpoint.deliver(_viewer_message(device_id, session_id, "busy")) + return + session = Session( + session_id, device_id, endpoint, self._devices[device_id][-1], who + ) + session.timer = asyncio.get_running_loop().call_later( + ANSWER_TIMEOUT_S, self._expire, session + ) + self._sessions[session_id] = session + logger.info("Live view %s on %s asked for", session_id, device_id) + self._to_camera( + session, + { + "action": "start", + "ice_servers": ice_servers(f"device-{device_id}"), + }, + ) + return + if ( + session is None + or session.viewer is not endpoint + or session.device != device_id + ): + return + if action == "bye": + self._end(session, "closed", tell_viewer=False) + return + relayed = _relayed(message, action) + if relayed is None: + return + if action == "sdp" and session.offered and not session.answered: + session.answered = True + if session.timer is not None: + session.timer.cancel() + self._to_camera(session, relayed) + + def _from_device( + self, + message: dict[str, Any], + session_id: str, + endpoint: Endpoint, + device_id: str, + ) -> None: + session = self._sessions.get(session_id) + if ( + session is None + or session.device != device_id + or session.camera is not endpoint + ): + return + action = message.get("action") + if action in ("bye", "busy"): + reason = str(message.get("reason") or "")[:64] or "closed" + self._end(session, reason, tell_camera=False, action=str(action)) + return + relayed = _relayed(message, action) + if relayed is None: + return + if action == "sdp" and not session.offered: + # The viewer builds its connection on the offer, so its ICE + # servers come with it: minted only for a session that got a slot. + session.offered = True + relayed["ice_servers"] = ice_servers(session.who) + self._to_viewer(session, relayed) + + # -- delivering and ending ------------------------------------------------ + + def _to_camera(self, session: Session, message: dict[str, Any]) -> None: + sent = {"type": "signal", "session": session.id, **message} + if not session.camera.deliver(sent): + self._end(session, "overflow", tell_camera=False) + + def _to_viewer(self, session: Session, message: dict[str, Any]) -> None: + sent = _viewer_message(session.device, session.id, **message) + if not session.viewer.deliver(sent): + self._end(session, "overflow", tell_viewer=False) + + def _refusal(self, device_id: str, devices: Any) -> str: + """Why this device cannot serve a live view now, or "".""" + if not self._devices.get(device_id): + return "offline" + reported = devices.reported_ports(device_id) if devices is not None else None + if LIVE_PORT not in (reported or {}): + return "unsupported" + return "" + + def _expire(self, session: Session) -> None: + if self._sessions.get(session.id) is session and not session.answered: + self._end(session, "timeout") + + def _end( + self, + session: Session, + reason: str, + *, + tell_viewer: bool = True, + tell_camera: bool = True, + action: str = "bye", + ) -> None: + if self._sessions.pop(session.id, None) is not session: + return + if session.timer is not None: + session.timer.cancel() + # Told once, without a retry: a side whose outbox is full is stuck, + # and its own timeout or a closing socket ends it there. + if tell_viewer: + session.viewer.deliver( + _viewer_message(session.device, session.id, action, reason=reason) + ) + if tell_camera: + session.camera.deliver( + { + "type": "signal", + "session": session.id, + "action": "bye", + "reason": reason, + } + ) + logger.info( + "Live view %s on %s ended after %.1f s: %s", + session.id, + session.device, + time.monotonic() - session.started, + reason, + ) + + +def _relayed(message: dict[str, Any], action: Any) -> dict[str, Any] | None: + """What of an ``sdp`` or ``candidate`` message is passed on, or None.""" + if action not in _RELAYED: + return None + value = message.get(action) + limit = MAX_SDP if action == "sdp" else MAX_CANDIDATE + if not isinstance(value, str) or not value or len(value) > limit: + return None + return {"action": action, action: value} + + +def _viewer_message( + device: str, session: str, action: str, **rest: Any +) -> dict[str, Any]: + return { + "type": "live", + "device": device, + "session": session, + "action": action, + **rest, + } + + +sessions = LiveSessions() diff --git a/backend/tests/api/routes/test_live.py b/backend/tests/api/routes/test_live.py new file mode 100644 index 0000000..874bf03 --- /dev/null +++ b/backend/tests/api/routes/test_live.py @@ -0,0 +1,441 @@ +"""Live view: WebRTC signalling relayed between a viewer's socket and a device's. + +openpanel ROADMAP L1. The server routes by session, holds a device to two +viewers, ends a session when either socket goes or no answer comes, and hands +out TURN credentials in coturn's REST scheme. +""" + +import asyncio +import base64 +import hashlib +import hmac +import json +import time +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +from fluksio.api.routes import oauth +from fluksio.cloud.connector import CloudConnector +from fluksio.core.config import settings +from fluksio.flow import live as live_view +from tests.api.routes.test_cameras import REPORT, wait_for +from tests.api.routes.test_devices import _pair, _panels + +PREFIX = f"{settings.API_V1_STR}/devices" +SOCKET = f"{settings.API_V1_STR}/flows/ws" +LIVE_REPORT = { + **REPORT, + "ports": [ + *REPORT["ports"], + {"port": "live", "direction": "in", "dtype": "video", "on_demand": True}, + ], +} + + +@pytest.fixture(scope="module") +def cams(client: TestClient, superuser_token_headers: dict[str, str]): + """``cam1`` reports a live port, ``cam2`` does not. Yields their headers.""" + for bucket in [b for b in oauth._hits if b.startswith("pair:")]: + del oauth._hits[bucket] + _panels( + client, superuser_token_headers, {"panels": [{"id": "cam1"}, {"id": "cam2"}]} + ) + cam1 = _pair(client, superuser_token_headers, "cam1") + cam2 = _pair(client, superuser_token_headers, "cam2") + for headers, report in ((cam1, LIVE_REPORT), (cam2, REPORT)): + device = "cam1" if headers is cam1 else "cam2" + put = client.put(f"{PREFIX}/{device}/report", headers=headers, json=report) + assert put.status_code == 200, put.text + yield cam1, cam2 + for device in ("cam1", "cam2"): + client.delete(f"{PREFIX}/{device}", headers=superuser_token_headers) + + +def _viewer(client: TestClient, headers: dict[str, str]) -> Any: + ticket = client.post(f"{settings.API_V1_STR}/flows/ws-ticket", headers=headers) + return client.websocket_connect(f"{SOCKET}?ticket={ticket.json()['ticket']}") + + +def _next(ws: Any, kind: str) -> dict[str, Any]: + """The next message of this type on a socket, skipping the rest.""" + while True: + message = ws.receive_json() + if message.get("type") == kind: + return message + + +def _start(ws: Any, session: str, device: str = "cam1") -> None: + ws.send_json( + {"type": "live", "device": device, "session": session, "action": "start"} + ) + + +def test_signalling_is_relayed_both_ways( + client: TestClient, superuser_token_headers: dict[str, str], cams +) -> None: + """start, the device's offer, the answer, candidates both ways, bye.""" + cam1, _ = cams + with ( + client.websocket_connect(SOCKET, headers=cam1) as device, + _viewer(client, superuser_token_headers) as viewer, + ): + device.receive_json() + _start(viewer, "session-one") + start = _next(device, "signal") + assert start["action"] == "start" + assert start["session"] == "session-one" + assert start["ice_servers"] == [] + + device.send_json( + { + "type": "signal", + "session": "session-one", + "action": "sdp", + "sdp": "v=0 offer", + } + ) + assert _next(viewer, "live") == { + "type": "live", + "device": "cam1", + "session": "session-one", + "action": "sdp", + "sdp": "v=0 offer", + "ice_servers": [], + } + viewer.send_json( + { + "type": "live", + "device": "cam1", + "session": "session-one", + "action": "sdp", + "sdp": "v=0 answer", + } + ) + assert _next(device, "signal")["sdp"] == "v=0 answer" + viewer.send_json( + { + "type": "live", + "device": "cam1", + "session": "session-one", + "action": "candidate", + "candidate": "candidate:1 1 udp 1 10.0.0.2 5000 typ host", + } + ) + assert _next(device, "signal")["candidate"].startswith("candidate:1 ") + device.send_json( + { + "type": "signal", + "session": "session-one", + "action": "candidate", + "candidate": "candidate:2 1 udp 1 10.0.0.3 5000 typ host", + } + ) + assert _next(viewer, "live")["candidate"].startswith("candidate:2 ") + + viewer.send_json( + { + "type": "live", + "device": "cam1", + "session": "session-one", + "action": "bye", + } + ) + bye = _next(device, "signal") + assert (bye["action"], bye["session"]) == ("bye", "session-one") + assert live_view.sessions.count("cam1") == 0 + + +def test_a_third_viewer_is_busy_and_closing_ends_the_sessions( + client: TestClient, superuser_token_headers: dict[str, str], cams +) -> None: + cam1, _ = cams + with client.websocket_connect(SOCKET, headers=cam1) as device: + device.receive_json() + with _viewer(client, superuser_token_headers) as viewer: + _start(viewer, "session-a") + _start(viewer, "session-b") + _start(viewer, "session-c") + busy = _next(viewer, "live") + assert (busy["session"], busy["action"]) == ("session-c", "busy") + asked = {_next(device, "signal")["session"] for _ in range(2)} + assert asked == {"session-a", "session-b"} + # The viewer's socket closed: the device hears bye for both. + ended = {_next(device, "signal")["session"] for _ in range(2)} + assert ended == {"session-a", "session-b"} + wait_for(lambda: live_view.sessions.count("cam1") == 0) + + with _viewer(client, superuser_token_headers) as viewer: + with client.websocket_connect(SOCKET, headers=cam1) as device: + device.receive_json() + _start(viewer, "session-d") + _next(device, "signal") + # The device answers busy itself, past its own cap. + device.send_json( + {"type": "signal", "session": "session-d", "action": "busy"} + ) + assert _next(viewer, "live")["action"] == "busy" + + _start(viewer, "session-e") + _next(device, "signal") + # The device's socket closed. + bye = _next(viewer, "live") + assert (bye["session"], bye["action"], bye["reason"]) == ( + "session-e", + "bye", + "offline", + ) + + +def test_a_session_without_an_answer_times_out( + client: TestClient, + superuser_token_headers: dict[str, str], + cams, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(live_view, "ANSWER_TIMEOUT_S", 0.2) + cam1, _ = cams + with ( + client.websocket_connect(SOCKET, headers=cam1) as device, + _viewer(client, superuser_token_headers) as viewer, + ): + device.receive_json() + _start(viewer, "session-slow") + _next(device, "signal") + bye = _next(viewer, "live") + assert (bye["action"], bye["reason"]) == ("bye", "timeout") + told = _next(device, "signal") + assert (told["action"], told["reason"]) == ("bye", "timeout") + + +def test_only_a_person_starts_and_only_the_device_speaks_for_its_session( + client: TestClient, superuser_token_headers: dict[str, str], cams +) -> None: + cam1, cam2 = cams + with _viewer(client, superuser_token_headers) as viewer: + # Not connected, then connected without a live port. + _start(viewer, "session-off") + assert _next(viewer, "live")["reason"] == "offline" + with client.websocket_connect(SOCKET, headers=cam2) as other: + other.receive_json() + _start(viewer, "session-old", device="cam2") + assert _next(viewer, "live")["reason"] == "unsupported" + + with client.websocket_connect(SOCKET, headers=cam1) as device: + device.receive_json() + # A device's credential cannot start one. + _start(other, "session-dev") + _start(viewer, "session-ok") + assert _next(device, "signal")["session"] == "session-ok" + assert live_view.sessions.count("cam1") == 1 + + # Another device cannot speak for it. + other.send_json( + { + "type": "signal", + "session": "session-ok", + "action": "sdp", + "sdp": "x", + } + ) + time.sleep(0.1) + device.send_json( + { + "type": "signal", + "session": "session-ok", + "action": "sdp", + "sdp": "y", + } + ) + assert _next(viewer, "live")["sdp"] == "y" + + # Nor can another viewer, for a session it does not hold. + with _viewer(client, superuser_token_headers) as stranger: + stranger.send_json( + { + "type": "live", + "device": "cam1", + "session": "session-ok", + "action": "bye", + } + ) + time.sleep(0.1) + assert live_view.sessions.count("cam1") == 1 + + +def test_ice_servers_carry_coturn_credentials( + client: TestClient, + superuser_token_headers: dict[str, str], + cams, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """``:``, and the base64 HMAC-SHA1 of it under the secret.""" + monkeypatch.setattr(settings, "STUN_URLS", ["stun:turn.example.com:3478"]) + monkeypatch.setattr( + settings, + "TURN_URLS", + ["turn:turn.example.com:3478", "turns:turn.example.com:5349?transport=tcp"], + ) + monkeypatch.setattr(settings, "TURN_SECRET", "s3cret") + cam1, _ = cams + + def check(server: dict[str, Any], who: str) -> None: + expiry, _, name = server["username"].partition(":") + assert name == who + assert 0 < int(expiry) - time.time() <= live_view.ICE_TTL_S + mac = hmac.new(b"s3cret", server["username"].encode(), hashlib.sha1) + assert server["credential"] == base64.b64encode(mac.digest()).decode() + + answer = client.get(f"{PREFIX}/cam1/ice", headers=superuser_token_headers) + assert answer.status_code == 200, answer.text + stun, turn = answer.json()["ice_servers"] + assert stun == {"urls": ["stun:turn.example.com:3478"]} + me = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers) + check(turn, me.json()["id"]) + + assert ( + client.get(f"{PREFIX}/nope/ice", headers=superuser_token_headers).status_code + == 404 + ) + # A device gets its servers in `start`, not here. + assert client.get(f"{PREFIX}/cam1/ice", headers=cam1).status_code == 403 + + with ( + client.websocket_connect(SOCKET, headers=cam1) as device, + _viewer(client, superuser_token_headers) as viewer, + ): + device.receive_json() + _start(viewer, "session-ice") + check(_next(device, "signal")["ice_servers"][1], "device-cam1") + # The viewer's come with the offer, minted for this session's user. + device.send_json( + {"type": "signal", "session": "session-ice", "action": "sdp", "sdp": "v=0"} + ) + check(_next(viewer, "live")["ice_servers"][1], me.json()["id"]) + + +def test_a_full_outbox_ends_the_session_instead_of_dropping() -> None: + """A socket that stops reading loses its session, never a message.""" + + class Devices: + def reported_ports(self, _device: str) -> dict: + return {"live": {}} + + async def run() -> None: + registry = live_view.LiveSessions() + camera, viewer = live_view.Endpoint(), live_view.Endpoint() + registry.attach("cam9", camera) + registry.heard( + { + "type": "live", + "device": "cam9", + "session": "s-overflow", + "action": "start", + }, + viewer, + person="someone", + device="", + devices=Devices(), + ) + assert registry.count("cam9") == 1 + # The viewer never reads, and the device trickles on. + for index in range(live_view.OUTBOX + 1): + registry.heard( + { + "type": "signal", + "session": "s-overflow", + "action": "candidate", + "candidate": f"candidate:{index} 1 udp 1 10.0.0.9 5000 typ host", + }, + camera, + person="", + device="cam9", + devices=None, + ) + assert registry.count("cam9") == 0 + start, bye = [json.loads(await camera.next()) for _ in range(2)] + assert start["action"] == "start" + assert (bye["action"], bye["reason"]) == ("bye", "overflow") + + asyncio.run(run()) + + +def test_the_tunnel_carries_signalling( + client: TestClient, superuser_token_headers: dict[str, str], cams +) -> None: + """A viewer through the portal, a device on this network, and back.""" + cam1, _ = cams + ticket = client.post( + f"{settings.API_V1_STR}/flows/ws-ticket", headers=superuser_token_headers + ).json()["ticket"] + connector = CloudConnector(client.app) + sent: list[dict[str, Any]] = [] + + class Socket: + async def send(self, text: str) -> None: + sent.append(json.loads(text)) + + def said() -> list[dict[str, Any]]: + return [ + json.loads(frame["text"]) + for frame in sent + if frame["op"] == "ws_msg" and '"live"' in frame["text"] + ] + + frame = {"id": "t1", "path": "/api/v1/flows/ws", "query": f"ticket={ticket}"} + assert client.portal is not None + with client.websocket_connect(SOCKET, headers=cam1) as device: + device.receive_json() + stream = client.portal.start_task_soon(connector._stream, Socket(), frame) + try: + wait_for(lambda: "t1" in connector._listeners) + hear = connector._listeners["t1"] + # The stream subscribes to the bus once it has said its snapshot. + wait_for(lambda: any(f["op"] == "ws_msg" for f in sent)) + time.sleep(0.1) + client.portal.call( + hear, + json.dumps( + { + "type": "live", + "device": "cam1", + "session": "tunnel-1", + "action": "start", + } + ), + ) + assert _next(device, "signal")["session"] == "tunnel-1" + device.send_json( + {"type": "signal", "session": "tunnel-1", "action": "sdp", "sdp": "v=0"} + ) + offer = wait_for(lambda: said())[0] + assert (offer["session"], offer["sdp"]) == ("tunnel-1", "v=0") + finally: + stream.cancel() + wait_for(stream.done) + # The stream went: the device hears bye. + assert _next(device, "signal")["action"] == "bye" + + +def test_frames_held_for_a_stream_go_with_it(client: TestClient) -> None: + """A stream refused before it listened leaves nothing behind.""" + connector = CloudConnector(client.app) + sent: list[str] = [] + + class Socket: + async def send(self, text: str) -> None: + sent.append(text) + + frame = {"id": "t2", "path": "/api/v1/flows/ws", "query": "ticket=spent"} + assert client.portal is not None + + def open_and_speak() -> None: + connector._start_stream(Socket(), frame) + # What the link does with a frame for a stream not listening yet, + # here before the stream has even started to run. + connector._early["t2"] = ['{"type": "media", "names": []}'] + + client.portal.call(open_and_speak) + wait_for(lambda: "t2" not in connector._streams) + assert "t2" not in connector._early diff --git a/docker/compose.yml b/docker/compose.yml index fed45a1..4269833 100644 --- a/docker/compose.yml +++ b/docker/compose.yml @@ -226,6 +226,57 @@ services: volumes: - /var/run/docker.sock:/var/run/docker.sock:ro + # ── TURN relay for live view (optional) ──────────────────────────── + # coturn, for a camera's live view between a viewer and a device that + # cannot reach each other directly. It serves STUN too. Behind the `turn` + # profile, so a plain `make up` never starts it. On the host's network, + # because a relay port range through Docker's port mapping is a proxy + # process per port. Credentials are minted by the backend from the same + # TURN_SECRET (coturn's REST scheme), so nothing is stored here. The secret + # goes in through a config file rather than the command line, which `ps` + # shows. See DEPLOY.md for ports, DNS and certificates. + coturn: + image: coturn/coturn:4.6.3 + container_name: fluksio-coturn + profiles: ["turn"] + restart: always + network_mode: host + security_opt: + - no-new-privileges:true + environment: + # Not `?`: compose reads every service, so a stack without the + # profile would refuse to start. Checked when coturn starts instead. + - TURN_SECRET=${TURN_SECRET:-} + - TURN_REALM=turn.${DOMAIN?Variable not set} + - TURN_MIN_PORT=${TURN_MIN_PORT:-49160} + - TURN_MAX_PORT=${TURN_MAX_PORT:-49200} + volumes: + # fullchain.pem and privkey.pem for turn.${DOMAIN}. Without them coturn + # runs without TLS on 5349. + - ${TURN_CERT_DIR:-./turn-certs}:/certs:ro + entrypoint: ["sh", "-c"] + command: + - | + [ -n "$$TURN_SECRET" ] || { echo "TURN_SECRET is empty" >&2; exit 1; } + cat > /tmp/turnserver.conf <`. A device that is unpaired or deleted while connected is closed with 1008. +### Live view + +A device that reports a `live` port streams its camera over WebRTC. The +video goes peer to peer, or through a TURN relay, and the socket carries only +the signalling, as text, so it crosses the portal too. A person's socket +sends and hears `{"type": "live", "device": "", "session": "", +"action": ...}`, a device's socket `{"type": "signal", "session": "", +"action": ...}`. A session id is 8 to 64 letters, digits, `-` or `_`, chosen by +the viewer. In order: + +| Step | Viewer's socket | Device's socket | +|---|---|---| +| 1 | sends `start` | hears `start` with `ice_servers` minted for the device | +| 2 | hears `sdp` (the offer) with `ice_servers` minted for the viewer | sends `sdp` with its offer | +| 3 | sends `sdp` with its answer | hears `sdp` | +| 4 | sends and hears `candidate`, the bare `candidate:...` line | hears and sends `candidate` | +| 5 | sends `bye`, or hears `bye` or `busy` with a `reason` | hears `bye` with a `reason`, or sends `bye` or `busy` | + +Any signed-in person may watch any device. A device speaks only for the +sessions it was handed on that socket. The server answers `busy` itself once a +device has 2 sessions, and `bye` with `offline` or `unsupported` when the +device is not connected or reports no `live` port. A session the viewer has not +answered within 10 seconds ends with `timeout`, and either socket closing ends +its sessions with `closed` or `offline`. A socket that falls 64 signalling +messages behind ends the session with `overflow` instead of losing one. +`ice_servers` follow `RTCIceServer`: `{urls: [...]}` for STUN, +`{urls, username, credential}` for TURN, in coturn's REST scheme with a +username of `:`, good for an hour. A device that sends its camera's +frames on the same socket pauses them while a session negotiates, so its +offer and candidates are not queued behind frames. + ## Errors and client generation | Code | Meaning | diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md index 986ff9f..06f6356 100644 --- a/docs/interface/dashboards.md +++ b/docs/interface/dashboards.md @@ -208,7 +208,8 @@ at, and stops when it scrolls away, the tab is hidden or the tile closes. A clip keeps playing either way. One page is pushed frames for at most 8 messages at once. A ninth tile fetches its frames one by one and does not turn a camera on. A [device's camera](#a-devices-camera) sends only while a tile like -this wants it. +this wants it. A tile on a device's camera that streams live shows **Live**: +tap it for [live video](#live-view). For continuous video, set **Live stream** to a browser-compatible camera URL. The browser must be able to reach that URL; enrollment does not expose the @@ -600,7 +601,7 @@ keeps the switch as it was set. ### A device's camera -A camera device (openpanel on a NANO) reports four ports: +A camera device (openpanel on a NANO) reports these ports: | Port | Type | Settings | What it sends | |---|---|---|---| @@ -608,6 +609,7 @@ A camera device (openpanel on a NANO) reports four ports: | `person` | `bool` | `hold_s` (10) | whether someone is in view, held for `hold_s` after they leave | | `detections` | `json` | | `{count, boxes: [{score, x0, y0, x1, y1}]}`, boxes scaled 0 to 1, on change and at most once a second | | `snapshot` | `image` | | one stored frame each time `person` turns true | +| `live` | `video`, on demand | | H.264 over WebRTC while a [live view](#live-view) runs; never wired in a flow | `camera` sends only while it has demand: while a page wants its frames (a dashboard tile, the preview on the device's page, a screen through the @@ -633,6 +635,24 @@ A device may fetch the last 8 artifacts each of its messages pointed at, not only the current one, so a screen fetching a camera frame by frame through the portal is not refused a frame the camera has just replaced. +#### Live view + +A media tile showing a device's `camera`, and the preview on the device's +page, offer **Live** while the device is online and reports `live`. Tap the +tile and the browser connects to the device with WebRTC, without a plugin, +and plays its video in the tile under a **Live** badge. Tap again to stop. It +stops by itself when the tile scrolls away, the tab is hidden or the page +closes. While it plays the tile asks for no frames, so the camera encodes one +picture rather than two, and frames come back when it ends. + +A device serves 2 live viewers. A third tile, a failed connection or no +picture within 10 seconds stays on frames and says so briefly. On one network +the two ends find each other directly. Between networks they need STUN and, +behind strict NATs, a TURN relay: set `STUN_URLS`, `TURN_URLS` and +`TURN_SECRET` (see [Configuration](../reference/configuration.md#live-view)). +The server only passes on the few messages that set a connection up, the +video never crosses it. A wall panel's own screen does not offer live view. + ### Before device nodes Before device nodes a device carried its ports itself. On its first start an diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 22e1940..777cc25 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -189,6 +189,24 @@ under Docker, raise `shm_size` with the ring size. See [Agents over MCP](../code/agents.md). +## Live view + +A camera's [live view](../interface/dashboards.md#live-view) connects the +browser and the device directly. These name the servers that help when the two +are on different networks. Unset, only direct connections are tried, and no +third party learns who watches what. + +| Variable | Default | Purpose | +|---|---|---| +| `STUN_URLS` | unset | STUN servers, comma separated, such as `stun:turn.example.com:3478` | +| `TURN_URLS` | unset | TURN relays, comma separated, such as `turn:turn.example.com:3478,turns:turn.example.com:5349?transport=tcp` | +| `TURN_SECRET` | unset | the secret shared with the TURN server (coturn's `static-auth-secret`); TURN is used only with it | + +Each viewer and device gets TURN credentials of its own per live view, good for +an hour, which coturn checks against the shared secret. A view relayed through +TURN therefore ends after an hour. coturn also answers STUN, so one server +covers both. + ## Mail and monitoring Password-reset mail requires `SMTP_HOST` and `EMAILS_FROM_EMAIL`. diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 11f8dd1..9b2907c 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -2006,6 +2006,63 @@ export const HistoryPointSchema = { description: 'One numeric value a message carried, and when.' } as const; +export const IceServerSchema = { + properties: { + urls: { + items: { + type: 'string' + }, + type: 'array', + title: 'Urls' + }, + username: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Username' + }, + credential: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Credential' + } + }, + type: 'object', + required: ['urls'], + title: 'IceServer', + description: 'One STUN or TURN server, as `RTCPeerConnection` takes it.' +} as const; + +export const IceServersSchema = { + properties: { + ice_servers: { + items: { + '$ref': '#/components/schemas/IceServer' + }, + type: 'array', + title: 'Ice Servers' + }, + ttl: { + type: 'integer', + title: 'Ttl', + default: 0 + } + }, + type: 'object', + title: 'IceServers' +} as const; + export const ImportArtifactSchema = { properties: { name: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 526d073..5f6261a 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesReadIceData, DevicesReadIceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AiService { /** @@ -269,7 +269,9 @@ export class ArtifactsService { * * ``volatile`` puts it in the ring instead of the store — a frame a screen is * watching now, which the oldest of falls out of memory rather than being - * kept. A worker on another host publishing a camera sends this. + * kept. A worker on another host or a device publishing a camera sends this. + * It is read into memory, where the ring holds it, and never touches the + * disk. * @param data The data for the request. * @param data.name * @param data.mediaType @@ -1007,6 +1009,30 @@ export class DevicesService { }); } + /** + * Read Ice + * Where a browser looks for a way to this device's live view. + * + * STUN and TURN servers, the TURN ones with credentials of the caller's own + * that expire. Empty when none are set, which is enough on one network. + * @param data The data for the request. + * @param data.deviceId + * @returns IceServers Successful Response + * @throws ApiError + */ + public static readIce(data: DevicesReadIceData): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/devices/{device_id}/ice', + path: { + device_id: data.deviceId + }, + errors: { + 422: 'Validation Error' + } + }); + } + /** * Put Report * A device says what it has: its board, its screen, its ports. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 7df2e73..48e5a11 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -686,6 +686,20 @@ export type HTTPValidationError = { detail?: Array; }; +/** + * One STUN or TURN server, as `RTCPeerConnection` takes it. + */ +export type IceServer = { + urls: Array<(string)>; + username?: (string | null); + credential?: (string | null); +}; + +export type IceServers = { + ice_servers?: Array; + ttl?: number; +}; + export type ImportArtifact = { name: string; filename?: (string | null); @@ -1879,6 +1893,12 @@ export type DevicesDeleteDeviceData = { export type DevicesDeleteDeviceResponse = (Message); +export type DevicesReadIceData = { + deviceId: string; +}; + +export type DevicesReadIceResponse = (IceServers); + export type DevicesPutReportData = { deviceId: string; }; diff --git a/frontend/src/components/Dashboard/MediaWidget.tsx b/frontend/src/components/Dashboard/MediaWidget.tsx index 77f36cf..0f47e4a 100644 --- a/frontend/src/components/Dashboard/MediaWidget.tsx +++ b/frontend/src/components/Dashboard/MediaWidget.tsx @@ -1,10 +1,12 @@ -import { useRef } from "react" +import { LoaderCircle, Radio, X } from "lucide-react" +import { type ReactNode, useEffect, useRef } from "react" import { useOnScreen } from "@/hooks/useOnScreen" import { isRef, useArtifactUrl } from "@/lib/media" import { cn } from "@/lib/utils" import { useBoundValue } from "./dataContext" import { config, text } from "./ui/core/config" +import { useLiveDevice, useLiveView } from "./useLiveView" import type { WidgetProps } from "./widgets" /** @@ -37,6 +39,10 @@ export function MediaWidget({ widget }: WidgetProps) { * looked at: a camera is only sent while somebody watches it, and a tile * scrolled away or behind another tab is nobody. A clip keeps playing either * way. + * + * A camera whose device streams live is tapped for it. Frames pause while + * the live view plays, so the device encodes one picture rather than two, + * and come back when it ends. */ export function MediaView({ message, @@ -55,12 +61,15 @@ export function MediaView({ }) { const box = useRef(null) const shown = useOnScreen(box) - const live = useBoundValue(message || undefined) - const value = live?.value + const bound = useBoundValue(message || undefined) + const value = bound?.value const ref = isRef(value) ? value : null const kind = (ref?.media_type ?? dtype).split("/")[0] const playing = kind === "audio" || kind === "video" - const active = shown || playing + const camera = useLiveDevice(stream ? "" : message) + const live = useLiveView(camera) + const onAir = live.phase === "playing" + const active = (shown || playing) && !onAir const url = useArtifactUrl( active ? ref : null, active ? message || undefined : undefined, @@ -69,12 +78,24 @@ export function MediaView({ const objectFit = fit === "contain" ? "object-contain" : "object-cover" const alt = label || ref?.name || message + // Scrolled away, or the tab hidden: nobody is watching. + const { stop } = live + useEffect(() => { + if (!shown) stop() + }, [shown, stop]) + return (
- {body()} + {camera ? ( + + {body()} + + ) : ( + body() + )}
) @@ -141,3 +162,95 @@ export function MediaView({ ) } } + +/** + * A camera tile that streams live when tapped, and stops when tapped again. + * + * The whole tile is the control, so it is as easy to hit on a wall panel as + * on a desk. The video waits under the frames until its first picture, then + * takes their place. + */ +function LiveFrame({ + live, + label, + fit, + children, +}: { + live: ReturnType + label: string + fit: string + children: ReactNode +}) { + const { phase, stream, note } = live + const onAir = phase === "playing" + const chip = + "pointer-events-none absolute z-20 flex items-center gap-1 rounded-full border border-border px-2.5 py-1 text-xs font-medium shadow-e2" + const frosted = "bg-card/80 text-foreground backdrop-blur-md" + + return ( + + ) +} diff --git a/frontend/src/components/Dashboard/useLiveView.ts b/frontend/src/components/Dashboard/useLiveView.ts new file mode 100644 index 0000000..c45e97a --- /dev/null +++ b/frontend/src/components/Dashboard/useLiveView.ts @@ -0,0 +1,153 @@ +import { useQuery } from "@tanstack/react-query" +import { useCallback, useEffect, useRef, useState } from "react" + +import { LIVE_PORT, reportedPorts } from "@/components/Devices/ports" +import { devicesQueryOptions } from "@/components/Devices/queries" +import { + type LiveSignal, + listenLive, + sendLive, +} from "@/components/Flow/useFlowSocket" +import { appRoute } from "@/lib/portal" + +/** How long a live view gets from the tap to its first frame. */ +const FIRST_FRAME_MS = 10_000 +/** How long the note after a fallback stays. */ +const NOTE_MS = 4000 + +const BUSY = "Live view busy, showing frames" +const FAILED = "Live view unavailable, showing frames" + +export type LivePhase = "off" | "connecting" | "playing" + +/** + * The device whose camera publishes `message` and can stream it live, or "". + * + * A device offers it by reporting a `live` port, and only while it is online. + * Read off the device list, which a wall panel's credential cannot read, so a + * panel never offers it. + */ +export function useLiveDevice(message: string): string { + const enabled = Boolean(message) && !appRoute().startsWith("/panel") + const { data } = useQuery({ ...devicesQueryOptions(), enabled }) + if (!enabled) return "" + const device = data?.data?.find( + (one) => + one.online && + (one.bindings ?? []).some( + (binding) => + binding.port === "camera" && + binding.direction === "in" && + binding.message === message, + ) && + (reportedPorts(one) ?? []).some((port) => port.port === LIVE_PORT), + ) + return device?.id ?? "" +} + +/** + * A live view of one device's camera: the browser's own `RTCPeerConnection` + * answering the device's offer, signalled over the page's socket. + * + * Busy, a failure or no first frame within ten seconds ends it with a short + * note, and the tile goes on showing frames. Leaving says `bye`. + */ +export function useLiveView(device: string) { + const [phase, setPhase] = useState("off") + const [stream, setStream] = useState(null) + const [note, setNote] = useState("") + const running = useRef<{ stop: () => void; played: () => void } | null>(null) + + const stop = useCallback(() => running.current?.stop(), []) + const played = useCallback(() => running.current?.played(), []) + + const start = useCallback(() => { + if (!device || running.current) return + // Not `randomUUID`, which a page on plain HTTP on a LAN does not have. + const session = Array.from( + crypto.getRandomValues(new Uint8Array(12)), + (byte) => byte.toString(16).padStart(2, "0"), + ).join("") + let pc: RTCPeerConnection | null = null + let over = false + // The device's candidates that came before its offer was applied. + const early: string[] = [] + + const end = (why: string, tell: boolean) => { + if (over) return + over = true + clearTimeout(timer) + unlisten() + if (tell) sendLive({ device, session, action: "bye" }) + pc?.close() + running.current = null + setStream(null) + setPhase("off") + setNote(why) + } + const add = (candidate: string) => + pc + ?.addIceCandidate({ candidate, sdpMLineIndex: 0 }) + .catch(() => undefined) + + const answer = async (offer: LiveSignal) => { + const peer = new RTCPeerConnection({ + iceServers: offer.ice_servers ?? [], + }) + pc = peer + peer.ontrack = (event) => + setStream(event.streams[0] ?? new MediaStream([event.track])) + peer.onicecandidate = (event) => { + // The bare `candidate:` line, which is what the device reads. + const candidate = event.candidate?.candidate + if (candidate) + sendLive({ device, session, action: "candidate", candidate }) + } + peer.onconnectionstatechange = () => { + if (peer.connectionState === "failed") end(FAILED, true) + } + await peer.setRemoteDescription({ type: "offer", sdp: offer.sdp }) + const reply = await peer.createAnswer() + await peer.setLocalDescription(reply) + sendLive({ device, session, action: "sdp", sdp: reply.sdp }) + for (const candidate of early.splice(0)) await add(candidate) + } + + const unlisten = listenLive(session, (message) => { + if (over) return + if (message.action === "busy") end(BUSY, false) + else if (message.action === "bye") end(FAILED, false) + else if (message.action === "sdp" && message.sdp && !pc) + answer(message).catch(() => end(FAILED, true)) + else if (message.action === "candidate" && message.candidate) { + if (pc?.remoteDescription) add(message.candidate) + else early.push(message.candidate) + } + }) + const timer = setTimeout(() => end(FAILED, true), FIRST_FRAME_MS) + running.current = { + stop: () => end("", true), + played: () => { + clearTimeout(timer) + setPhase("playing") + }, + } + setNote("") + setPhase("connecting") + if (!sendLive({ device, session, action: "start" })) end(FAILED, false) + }, [device]) + + // A tile that goes away, or is pointed at another device, ends its view. + useEffect(() => { + if (!device) return + return () => running.current?.stop() + }, [device]) + + useEffect(() => { + if (!note) return + const timer = setTimeout(() => setNote(""), NOTE_MS) + return () => clearTimeout(timer) + }, [note]) + + return { phase, stream, note, start, stop, played } +} diff --git a/frontend/src/components/Devices/DeviceDetail.tsx b/frontend/src/components/Devices/DeviceDetail.tsx index 02048f3..01085c5 100644 --- a/frontend/src/components/Devices/DeviceDetail.tsx +++ b/frontend/src/components/Devices/DeviceDetail.tsx @@ -44,6 +44,7 @@ import { type DeviceReport, deviceNodeOf, freeNodeId, + LIVE_PORT, type ReportedPort, reportedPorts, reportOf, @@ -233,7 +234,7 @@ function Camera({ device }: { device: DevicePublic }) {
) : null} - {canEdit && said && wired.length === 0 ? ( + {said?.port === LIVE_PORT ? ( + + Live view + + ) : canEdit && said && wired.length === 0 ? ( ) : null} diff --git a/frontend/src/components/Devices/ports.ts b/frontend/src/components/Devices/ports.ts index d3b7460..cc58499 100644 --- a/frontend/src/components/Devices/ports.ts +++ b/frontend/src/components/Devices/ports.ts @@ -3,6 +3,12 @@ import type { DevicePublic, DType, MessageSpec, NodeDef_Input } from "@/client" /** The node type a device is in a flow. */ export const DEVICE_TYPE = "device" +/** + * The port a device reports when its camera can stream live over WebRTC. + * Signalled on the socket, never wired in a flow. + */ +export const LIVE_PORT = "live" + /** One port as a device reports it. */ export type ReportedPort = { port: string @@ -57,8 +63,9 @@ export function screensOf(devices: DevicePublic[] = []) { }) } -/** Whether a reported port may be wired in `direction`. */ +/** Whether a reported port may be wired in `direction`. `live` never is. */ export const wiresAs = (port: ReportedPort, direction: "in" | "out") => + port.port !== LIVE_PORT && (port.modes?.length ? port.modes : [port.direction]).includes(direction) /** What the server accepts as a port name. */ diff --git a/frontend/src/components/Flow/CommandPalette.tsx b/frontend/src/components/Flow/CommandPalette.tsx index e4dd9ee..f685266 100644 --- a/frontend/src/components/Flow/CommandPalette.tsx +++ b/frontend/src/components/Flow/CommandPalette.tsx @@ -2,7 +2,11 @@ import { useQuery } from "@tanstack/react-query" import { useNavigate } from "@tanstack/react-router" import type { DevicePublic, FlowSummary, NodeTypeInfo } from "@/client" -import { type ReportedPort, reportedPorts } from "@/components/Devices/ports" +import { + LIVE_PORT, + type ReportedPort, + reportedPorts, +} from "@/components/Devices/ports" import { devicesQueryOptions } from "@/components/Devices/queries" import { CommandDialog, @@ -58,7 +62,9 @@ export function CommandPalette({ // A shortcut per port a device says it has, "Hallway camera". A device that // has not said is added as a Device node and its ports typed in. const capabilities = (devicesData?.data ?? []).flatMap((device) => - (reportedPorts(device) ?? []).map((port) => ({ device, port })), + (reportedPorts(device) ?? []) + .filter((port) => port.port !== LIVE_PORT) + .map((port) => ({ device, port })), ) const close = (action: () => void) => { diff --git a/frontend/src/components/Flow/useFlowSocket.ts b/frontend/src/components/Flow/useFlowSocket.ts index bf4824d..187bf99 100644 --- a/frontend/src/components/Flow/useFlowSocket.ts +++ b/frontend/src/components/Flow/useFlowSocket.ts @@ -134,6 +134,23 @@ type FlowEvent = names: string[] ts?: number } + | LiveSignal + +/** + * A live view's signalling, both ways on this socket (`fluksio.flow.live`). + * The page sends `start`, `sdp`, `candidate` and `bye`, and hears `sdp` (the + * device's offer, with the ICE servers to use), `candidate`, `bye` and `busy`. + */ +export type LiveSignal = { + type: "live" + device: string + session: string + action: "start" | "sdp" | "candidate" | "bye" | "busy" + sdp?: string + candidate?: string + reason?: string + ice_servers?: RTCIceServer[] +} async function socketUrl(): Promise { const base = String(OpenAPI.BASE || window.location.origin) @@ -176,6 +193,42 @@ let client: QueryClient | null = null const authHandlers = new Set<() => void>() /** Floor re-reads still owed, per node, so a recovery can cancel its own. */ const healthTimers = new Map>() +/** What each live session on this page does with what it hears. */ +const liveListeners = new Map void>() + +/** Send a live view's signalling. False while the socket is not open. */ +export function sendLive(message: Omit): boolean { + if (socket?.readyState !== WebSocket.OPEN) return false + socket.send(JSON.stringify({ type: "live", ...message })) + return true +} + +/** + * Hear one live session's signalling, until the returned function is called. + * A socket that drops ends every session on it, so each hears `bye` then. + */ +export function listenLive( + session: string, + listener: (message: LiveSignal) => void, +): () => void { + liveListeners.set(session, listener) + return () => { + if (liveListeners.get(session) === listener) liveListeners.delete(session) + } +} + +/** The socket went: so did every live session riding on it. */ +function endLive() { + for (const [session, listener] of [...liveListeners]) { + listener({ + type: "live", + device: "", + session, + action: "bye", + reason: "socket", + }) + } +} /** Everything a node's health shows up in, which is two unrelated keys. */ function invalidateHealth(flow: string | undefined) { @@ -436,6 +489,9 @@ function open(url: string) { // live value, so nothing here goes through the live store. client?.invalidateQueries({ queryKey: runKeys.all }) break + case "live": + liveListeners.get(message.session)?.(message) + break case "run_metric": // A batch of readings was written. Only what draws them is refetched — // the run's own detail (its metric names and curve hang below that @@ -480,6 +536,7 @@ function open(url: string) { socket = null liveStore.setConnected(false) connectionStore.setSocketOpen(false) + endLive() if (watchers === 0) return if (event.code === 1008 && refused()) return // 1013 is the portal saying the instance is not attached — the one @@ -504,6 +561,7 @@ function release(onAuthFailure?: () => void) { const ws = socket socket = null ws?.close() + endLive() liveStore.setConnected(false) // Its `onclose` is dropped by the guard above, so say so here instead. connectionStore.setSocketOpen(false)