"""Tiny stdlib MJPEG + control dashboard over HTTP.
Serves a live JPEG stream (the Go2 camera with GoWelcome's detection/state
overlay) plus an optional control dashboard -- ideal for headless operation on
the dog. Stdlib only; the caller supplies a ``jpeg_provider`` that returns
already-encoded JPEG bytes (so this module needs no cv2), and optionally a
``status_provider`` (-> dict) and ``control_handler`` (dict -> dict) to power
the dashboard panel + buttons.
Routes:
GET / -> dashboard page (or bare viewer if no controls)
GET /stream.mjpg -> multipart/x-mixed-replace MJPEG stream
GET /snapshot.jpg -> a single current JPEG frame
GET /status.json -> current status dict (if status_provider given)
POST /control -> apply a control command (if control_handler given)
GET /healthz -> "ok" (liveness)
"""
from __future__ import annotations
import json
import logging
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Callable, Optional
logger = logging.getLogger(__name__)
# A callable returning the current frame as encoded JPEG bytes, or None if no
# frame is available yet.
JpegProvider = Callable[[], Optional[bytes]]
StatusProvider = Callable[[], dict]
ControlHandler = Callable[[dict], dict]
_BOUNDARY = "gowelcomeframe"
def _index_html(title: str, controls: bool) -> bytes:
head = (
""
""
f"{title}"
"
"
)
if not controls:
body = (
f"
{title} — live camera
"
""
)
else:
body = (
f"
{title} — control dashboard
"
""
"
Play:"
""
""
""
"
"
"
"
""
""
""
"
"
"
"
""
""
"
"
"
loading...
"
""
)
return (head + body + "
").encode("utf-8")
class MjpegServer:
"""Serve a JPEG stream over HTTP on a background daemon thread.
Args:
jpeg_provider: Returns the current frame as JPEG bytes (or ``None``).
host: Bind address (``0.0.0.0`` to allow remote viewers).
port: TCP port.
fps: Maximum stream frame rate (paces the multipart writer).
title: Page/title shown in the browser.
"""
def __init__(
self,
jpeg_provider: JpegProvider,
host: str = "0.0.0.0",
port: int = 8080,
fps: float = 10.0,
title: str = "GoWelcome",
status_provider: Optional[StatusProvider] = None,
control_handler: Optional[ControlHandler] = None,
) -> None:
self._provider = jpeg_provider
self._status_provider = status_provider
self._control_handler = control_handler
self._host = host
self._port = port
self._period = 1.0 / max(1e-3, fps)
self._title = title
self._httpd: Optional[ThreadingHTTPServer] = None
self._thread: Optional[threading.Thread] = None
self._stop = threading.Event()
@property
def has_controls(self) -> bool:
return self._control_handler is not None or self._status_provider is not None
@property
def port(self) -> int:
"""The bound TCP port (resolved after :meth:`start`)."""
return self._port
# ------------------------------------------------------------------ #
def start(self) -> None:
"""Bind the socket and start serving on a daemon thread.
Raises:
OSError: if the port cannot be bound (e.g. already in use).
"""
server = self
class Handler(BaseHTTPRequestHandler):
# Per-connection socket timeout so a stalled/non-reading client
# cannot pin a handler thread forever inside wfile.write once the
# TCP send buffer fills: a timed-out write raises socket.timeout,
# caught by _send_stream's except -> the handler exits cleanly.
# (StreamRequestHandler.setup applies this via socket.settimeout.)
timeout = max(10.0, server._period * 4)
# Silence the default noisy stderr logging.
def log_message(self, fmt, *args): # noqa: N802, D401
logger.debug("web: " + fmt, *args)
def do_GET(self): # noqa: N802
if self.path in ("/", "/index.html"):
self._send_index()
elif self.path.startswith("/stream.mjpg"):
self._send_stream()
elif self.path.startswith("/snapshot.jpg"):
self._send_snapshot()
elif self.path.startswith("/status.json"):
self._send_json(server._status())
elif self.path.startswith("/healthz"):
self._send_text("ok")
else:
self.send_error(404)
def do_POST(self): # noqa: N802
if not self.path.startswith("/control"):
self.send_error(404)
return
try:
n = int(self.headers.get("Content-Length", 0) or 0)
if n > 64 * 1024: # control commands are tiny; reject floods
self.send_error(413, "body too large")
return
raw = self.rfile.read(n) if n > 0 else b"{}"
cmd = json.loads(raw.decode("utf-8") or "{}")
except Exception:
self.send_error(400, "bad JSON")
return
self._send_json(server._control(cmd))
def _send_index(self):
body = _index_html(server._title, server.has_controls)
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _send_text(self, text):
body = text.encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/plain")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _send_json(self, obj):
body = json.dumps(obj).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _send_snapshot(self):
jpeg = server._safe_provide()
if jpeg is None:
self.send_error(503, "no frame yet")
return
self.send_response(200)
self.send_header("Content-Type", "image/jpeg")
self.send_header("Content-Length", str(len(jpeg)))
self.end_headers()
self.wfile.write(jpeg)
def _send_stream(self):
self.send_response(200)
self.send_header("Age", "0")
self.send_header("Cache-Control", "no-cache, private")
self.send_header("Pragma", "no-cache")
self.send_header(
"Content-Type",
f"multipart/x-mixed-replace; boundary={_BOUNDARY}",
)
self.end_headers()
try:
while not server._stop.is_set():
jpeg = server._safe_provide()
if jpeg:
self.wfile.write(f"--{_BOUNDARY}\r\n".encode())
self.send_header("Content-Type", "image/jpeg")
self.send_header("Content-Length", str(len(jpeg)))
self.end_headers()
self.wfile.write(jpeg)
self.wfile.write(b"\r\n")
server._stop.wait(server._period)
except (BrokenPipeError, ConnectionResetError):
logger.debug("web: client disconnected from stream")
except Exception: # noqa: BLE001 - a viewer error must not crash us
logger.debug("web: stream handler error", exc_info=True)
self._httpd = ThreadingHTTPServer((self._host, self._port), Handler)
self._httpd.daemon_threads = True
# Reflect the actually-bound port (supports port=0 -> ephemeral).
self._port = self._httpd.server_address[1]
self._thread = threading.Thread(
target=self._httpd.serve_forever, name="MjpegServer", daemon=True
)
self._thread.start()
logger.info(
"MJPEG viewer at http://%s:%d/ (stream: /stream.mjpg)",
"" if self._host in ("0.0.0.0", "") else self._host,
self._port,
)
def _safe_provide(self) -> Optional[bytes]:
try:
return self._provider()
except Exception: # noqa: BLE001
logger.debug("web: jpeg_provider raised", exc_info=True)
return None
def _status(self) -> dict:
if self._status_provider is None:
return {}
try:
return self._status_provider()
except Exception: # noqa: BLE001
logger.debug("web: status_provider raised", exc_info=True)
return {"error": "status unavailable"}
def _control(self, cmd: dict) -> dict:
if self._control_handler is None:
return {"ok": False, "error": "controls disabled"}
try:
return self._control_handler(cmd) or {"ok": True}
except Exception as exc: # noqa: BLE001
logger.warning("web: control_handler error: %s", exc)
return {"ok": False, "error": str(exc)}
def stop(self) -> None:
"""Stop serving and release the socket. Idempotent."""
self._stop.set()
if self._httpd is not None:
try:
self._httpd.shutdown()
self._httpd.server_close()
except Exception: # noqa: BLE001
logger.debug("web: shutdown error", exc_info=True)
self._httpd = None
logger.info("MJPEG viewer stopped")