"""Live status website for the voice loop.
Serves a single self-contained page plus a Server-Sent-Events stream so you can
open a browser and watch, step by step: is it listening, what it heard, what it
thought/answered, how long each stage took, and whether anything errored.
Pure stdlib (``http.server``). Runs in a background thread so it never blocks
the asyncio pipeline.
Endpoints:
GET / -> the dashboard HTML
GET /api/state -> JSON snapshot (initial load / fallback polling)
GET /events -> text/event-stream live push
"""
from __future__ import annotations
import json
import logging
import queue
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from .monitor import Monitor
log = logging.getLogger("wsai.dashboard")
def _make_handler(monitor: Monitor):
class Handler(BaseHTTPRequestHandler):
# Quiet: don't spam the console with one line per request.
def log_message(self, *args) -> None: # noqa: D401
return
def _send(self, code: int, body: bytes, ctype: str) -> None:
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None: # noqa: N802
path = self.path.split("?", 1)[0]
if path == "/" or path == "/index.html":
self._send(200, PAGE.encode("utf-8"), "text/html; charset=utf-8")
elif path == "/api/state":
body = json.dumps(monitor.snapshot(), ensure_ascii=False).encode("utf-8")
self._send(200, body, "application/json; charset=utf-8")
elif path == "/events":
self._stream_events()
else:
self._send(404, b"not found", "text/plain; charset=utf-8")
def _stream_events(self) -> None:
self.send_response(200)
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.send_header("Cache-Control", "no-store")
self.send_header("Connection", "keep-alive")
self.end_headers()
q = monitor.subscribe()
try:
# Prime the client with a full snapshot so it renders instantly.
first = json.dumps(
{"type": "snapshot", "snapshot": monitor.snapshot()},
ensure_ascii=False,
)
self.wfile.write(f"data: {first}\n\n".encode("utf-8"))
self.wfile.flush()
while True:
try:
data = q.get(timeout=15)
except queue.Empty:
# Heartbeat keeps proxies / the browser from timing out.
self.wfile.write(b": ping\n\n")
self.wfile.flush()
continue
self.wfile.write(f"data: {data}\n\n".encode("utf-8"))
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError):
pass
finally:
monitor.unsubscribe(q)
return Handler
class Dashboard:
"""Owns the HTTP server thread."""
def __init__(self, monitor: Monitor, host: str = "0.0.0.0", port: int = 8787) -> None:
self.monitor = monitor
self.host = host
self.port = port
self._server: ThreadingHTTPServer | None = None
self._thread: threading.Thread | None = None
def start(self) -> None:
handler = _make_handler(self.monitor)
self._server = ThreadingHTTPServer((self.host, self.port), handler)
self._server.daemon_threads = True
self._thread = threading.Thread(
target=self._server.serve_forever, name="wsai-dashboard", daemon=True
)
self._thread.start()
log.info("dashboard on http://%s:%d", self.host, self.port)
def stop(self) -> None:
if self._server is not None:
self._server.shutdown()
self._server.server_close()
self._server = None
# --------------------------------------------------------------------------- #
# The page. One file, no external assets, so it works offline / behind a LAN.
# --------------------------------------------------------------------------- #
PAGE = r"""
watch_sceen_ai · 실시간 상태