137 lines
5.6 KiB
Python
137 lines
5.6 KiB
Python
"""Fan-out of pipeline result frames and status to connected web clients.
|
|
|
|
A single broadcaster task reads cached :class:`WebController` snapshots and
|
|
pushes the freshest frame (latest-wins) plus a slower status heartbeat to every
|
|
registered client. Each client keeps one latest message per type, so a slow
|
|
socket skips intermediate frames without losing status or settings updates.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
|
|
from python_app.webui.controller import WebController
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Poll the controller this often; the C++ pipeline publishes well below this rate,
|
|
# so this is a comfortable latest-wins cadence without busy-spinning the loop.
|
|
_FRAME_INTERVAL_S = 0.05
|
|
# Status is cheap but rarely changes; emit it about once a second.
|
|
_STATUS_INTERVAL_S = 1.0
|
|
|
|
|
|
class ClientUpdates:
|
|
"""At most one pending frame, status and settings update for one socket."""
|
|
|
|
def __init__(self) -> None:
|
|
self._pending: dict[str, dict] = {}
|
|
self._ready = asyncio.Event()
|
|
|
|
def put(self, message: dict) -> None:
|
|
self._pending[message["type"]] = message
|
|
self._ready.set()
|
|
|
|
async def get(self) -> dict:
|
|
await self._ready.wait()
|
|
message = self._pending.pop(next(iter(self._pending)))
|
|
if not self._pending:
|
|
self._ready.clear()
|
|
return message
|
|
|
|
|
|
class RingBroadcaster:
|
|
"""Polls the controller and fans frames/status out to all WebSocket clients."""
|
|
|
|
def __init__(self, controller: WebController) -> None:
|
|
self._controller = controller
|
|
self._clients: set[ClientUpdates] = set()
|
|
self._task: asyncio.Task[None] | None = None
|
|
self._last_frame_seq: int | None = None
|
|
self._last_settings: list | None = None
|
|
|
|
def register(self) -> ClientUpdates:
|
|
"""Seed every connection with current snapshots, then stream updates."""
|
|
client = ClientUpdates()
|
|
frame = self._controller.peek_frame()
|
|
if frame is not None:
|
|
client.put(frame)
|
|
client.put(self._status_message())
|
|
client.put({"type": "settings", "schema": self._controller.current_live_settings()})
|
|
if not self._clients:
|
|
self._controller.set_frame_streaming(True)
|
|
self._clients.add(client)
|
|
logger.debug("Registered web client queue (clients=%d)", len(self._clients))
|
|
return client
|
|
|
|
def unregister(self, client: ClientUpdates) -> None:
|
|
"""Remove a client queue; safe to call more than once."""
|
|
was_connected = bool(self._clients)
|
|
self._clients.discard(client)
|
|
if was_connected and not self._clients:
|
|
self._controller.set_frame_streaming(False)
|
|
logger.debug("Unregistered web client queue (clients=%d)", len(self._clients))
|
|
|
|
def start(self) -> None:
|
|
"""Launch the single polling task (idempotent)."""
|
|
if self._task is None or self._task.done():
|
|
self._task = asyncio.create_task(self._run(), name="ring-broadcaster")
|
|
self._task.add_done_callback(self._on_task_done)
|
|
logger.info("Ring broadcaster started")
|
|
|
|
@staticmethod
|
|
def _on_task_done(task: "asyncio.Task[None]") -> None:
|
|
"""Surface an unexpected broadcaster death (the loop should never exit)."""
|
|
if not task.cancelled() and task.exception() is not None:
|
|
logger.error("ring broadcaster task exited unexpectedly: %r", task.exception())
|
|
|
|
async def stop(self) -> None:
|
|
"""Cancel the polling task and wait for it to unwind."""
|
|
self._controller.set_frame_streaming(False)
|
|
self._clients.clear()
|
|
if self._task is None:
|
|
return
|
|
self._task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._task
|
|
self._task = None
|
|
logger.info("Ring broadcaster stopped")
|
|
|
|
def _publish(self, message: dict) -> None:
|
|
"""Replace only older messages of the same type."""
|
|
for client in self._clients:
|
|
client.put(message)
|
|
|
|
def _status_message(self) -> dict:
|
|
"""Build a status broadcast from the controller's current state."""
|
|
return {"type": "status", **self._controller.status()}
|
|
|
|
async def _run(self) -> None:
|
|
"""Poll on a fixed cadence; never block the loop or die on a bad frame."""
|
|
loop = asyncio.get_running_loop()
|
|
next_status = loop.time()
|
|
while True:
|
|
try:
|
|
now = loop.time()
|
|
if self._clients and now >= next_status:
|
|
self._publish(self._status_message())
|
|
# Push live settings (Qt -> web) only when they change, so the
|
|
# browser form mirrors desktop edits in real time without churn.
|
|
settings = self._controller.current_live_settings()
|
|
if settings != self._last_settings:
|
|
self._publish({"type": "settings", "schema": settings})
|
|
self._last_settings = settings
|
|
next_status = now + _STATUS_INTERVAL_S
|
|
if self._clients:
|
|
frame = self._controller.peek_frame()
|
|
if frame is not None and frame["seq"] != self._last_frame_seq:
|
|
self._publish(frame)
|
|
self._last_frame_seq = frame["seq"]
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception: # noqa: BLE001 - one bad frame must not stop streaming
|
|
logger.warning("ring broadcaster iteration failed; continuing", exc_info=True)
|
|
await asyncio.sleep(_FRAME_INTERVAL_S)
|