#!/usr/bin/env python3 """Serves the web viewer and mints LiveKit dev tokens. GET /viewer.html -> the live viewer page GET /recordings.html -> the recordings browser page GET /token?room=demo&identity=alice -> subscribe+mic-only access token (plain text) GET /api/recordings -> JSON list of recordings + CV metadata GET /recordings/.mp4 -> recording playback (supports Range) Dev use only: the API secret lives here so browsers on insecure origins (plain http on a non-localhost host) don't need a pasted token. Configure via env: LIVEKIT_API_KEY, LIVEKIT_API_SECRET, PORT, RECORDINGS_DIR, CV_DIR. """ import base64 import hashlib import hmac import json import os import time from http.server import HTTPServer, SimpleHTTPRequestHandler from pathlib import Path from urllib.parse import parse_qs, urlparse BASE_DIR = Path(__file__).resolve().parent API_KEY = os.environ.get("LIVEKIT_API_KEY", "devkey") API_SECRET = os.environ.get("LIVEKIT_API_SECRET", "secret") PORT = int(os.environ.get("PORT", "8088")) RECORDINGS_DIR = Path(os.environ.get( "RECORDINGS_DIR", BASE_DIR / "../agent/recordings")).resolve() CV_DIR = Path(os.environ.get( "CV_DIR", BASE_DIR / "../agent/computerVision")).resolve() def b64url(raw: bytes) -> bytes: return base64.urlsafe_b64encode(raw).rstrip(b"=") def mint_token(room: str, identity: str) -> str: header = b64url(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()) now = int(time.time()) payload = b64url(json.dumps({ "exp": now + 24 * 3600, "nbf": now - 60, "iss": API_KEY, "sub": identity, "name": identity, "video": { "room": room, "roomJoin": True, "canPublish": True, "canSubscribe": True, "canPublishSources": ["microphone"], }, }).encode()) signing_input = header + b"." + payload sig = b64url(hmac.new(API_SECRET.encode(), signing_input, hashlib.sha256).digest()) return (signing_input + b"." + sig).decode() def list_recordings() -> list[dict]: entries = [] if not RECORDINGS_DIR.is_dir(): return entries for mp4 in sorted(RECORDINGS_DIR.glob("*.mp4"), key=lambda p: p.stat().st_mtime, reverse=True): stem = mp4.stem # - room, _, epoch = stem.rpartition("-") meta = {} meta_file = CV_DIR / f"{stem}.meta" if meta_file.is_file(): try: meta = json.loads(meta_file.read_text()) except (json.JSONDecodeError, OSError): pass entries.append({ "file": mp4.name, "room": room or stem, "epoch": int(epoch) if epoch.isdigit() else int(mp4.stat().st_mtime), "size_bytes": mp4.stat().st_size, "duration_seconds": meta.get("duration_seconds"), "device": meta.get("device"), "categories": meta.get("categories_present", []), "detections": meta.get("detections", {}), "analyzed_seconds": meta.get("analyzed_seconds"), "model": meta.get("model"), }) return entries class Handler(SimpleHTTPRequestHandler): def do_GET(self): url = urlparse(self.path) if url.path == "/token": return self.serve_token(url) if url.path == "/api/recordings": return self.send_json(list_recordings()) if url.path.startswith("/recordings/"): return self.serve_video(url.path.removeprefix("/recordings/")) return super().do_GET() def send_json(self, payload): body = json.dumps(payload).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def serve_token(self, url): query = parse_qs(url.query) room = query.get("room", ["demo3"])[0] identity = query.get("identity", [f"web-viewer-{int(time.time()) % 10000}"])[0] body = mint_token(room, identity).encode() self.send_response(200) self.send_header("Content-Type", "text/plain") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def serve_video(self, name: str): """Serves a recording with HTTP Range support (required for seeking and for Safari to play video at all).""" path = (RECORDINGS_DIR / name).resolve() if ("/" in name or not name.endswith(".mp4") or path.parent != RECORDINGS_DIR or not path.is_file()): self.send_error(404) return size = path.stat().st_size start, end = 0, size - 1 range_header = self.headers.get("Range", "") is_partial = range_header.startswith("bytes=") if is_partial: spec = range_header[len("bytes="):].split(",")[0] start_s, _, end_s = spec.partition("-") try: start = int(start_s) if start_s else 0 end = min(int(end_s), size - 1) if end_s else size - 1 except ValueError: start, end = 0, size - 1 if start > end or start >= size: self.send_response(416) self.send_header("Content-Range", f"bytes */{size}") self.end_headers() return self.send_response(206 if is_partial else 200) self.send_header("Content-Type", "video/mp4") self.send_header("Accept-Ranges", "bytes") if is_partial: self.send_header("Content-Range", f"bytes {start}-{end}/{size}") self.send_header("Content-Length", str(end - start + 1)) self.end_headers() try: with path.open("rb") as f: f.seek(start) remaining = end - start + 1 while remaining > 0: chunk = f.read(min(65536, remaining)) if not chunk: break self.wfile.write(chunk) remaining -= len(chunk) except (BrokenPipeError, ConnectionResetError): pass # player closed the connection (normal during seeks) if __name__ == "__main__": os.chdir(os.path.dirname(os.path.abspath(__file__))) print(f"Serving web viewer + /token on 0.0.0.0:{PORT}") HTTPServer(("0.0.0.0", PORT), Handler).serve_forever()