""" RoboPark Session Scheduler - Backend API Manages robot fleet, LiveKit servers, and session orchestration """ import os import asyncio import logging import re import secrets import hashlib import hmac from datetime import datetime, timedelta from typing import Optional, List from contextlib import asynccontextmanager from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, Query, Header from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.responses import HTMLResponse from pydantic import BaseModel import httpx import aiosqlite # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Database path DB_PATH = os.path.join(os.getenv("SCHEDULER_DATA_DIR", "/app/data"), "scheduler.db") # Path to a ROBOVOICE-style settings file the scheduler reads for characters. # In production this should be a bind-mount to ROBOVOICE-main/settings.json (or .default.json). ROBOVOICE_SETTINGS_PATH = os.path.join(os.getenv("SCHEDULER_DATA_DIR", "/app/data"), "robovoice_settings.json") # Session end_reasons considered "abnormal" for drop-rate telemetry. A session # that ended with one of these reasons is counted as a dropped session; anything # that ended cleanly (e.g. "silence", "completed", "user_ended") is not. ABNORMAL_END_REASONS = { "error", "timeout", "crash", "server_error", "connection_lost", "lost", "failed", "disconnect", "unknown", } # ============================================================================= # MODELS # ============================================================================= class Robot(BaseModel): id: str name: str character_id: Optional[str] = None status: str = "idle" # idle, detecting, connecting, running current_session_id: Optional[str] = None connected_server_id: Optional[str] = None ip_address: Optional[str] = None last_heartbeat: Optional[datetime] = None total_sessions: int = 0 total_runtime_seconds: int = 0 class LiveKitServer(BaseModel): id: str name: str url: str webhook_url: str # Internal URL for health/metrics api_key: Optional[str] = None api_secret: Optional[str] = None gpu_name: Optional[str] = None gpu_vram_mb: int = 0 max_sessions: int = 8 status: str = "unknown" # online, offline, maintenance class Session(BaseModel): id: str robot_id: str server_id: str room_name: str started_at: datetime ended_at: Optional[datetime] = None end_reason: Optional[str] = None duration_seconds: Optional[int] = None class GpuModel(BaseModel): name: str size_gb: float is_loaded: bool = False vram_used_mb: int = 0 class ServerMetrics(BaseModel): server_id: str gpu_utilization: int = 0 vram_used_mb: int = 0 vram_total_mb: int = 0 active_sessions: int = 0 models: List[GpuModel] = [] class SessionRequest(BaseModel): robot_id: str class WebhookEvent(BaseModel): event: str data: dict class Device(BaseModel): id: str name: str tailscale_ip: Optional[str] = None lan_ip: Optional[str] = None motor_server_url: Optional[str] = None character_id: Optional[str] = None livekit_url: Optional[str] = None status: str = "enrolled" # enrolled, online, offline, disabled last_heartbeat: Optional[datetime] = None enrolled_at: Optional[datetime] = None last_seen_ip: Optional[str] = None notes: Optional[str] = None created_at: Optional[datetime] = None class DeviceCreate(BaseModel): name: str tailscale_ip: Optional[str] = None lan_ip: Optional[str] = None motor_server_url: Optional[str] = None character_id: Optional[str] = None livekit_url: Optional[str] = None notes: Optional[str] = None enrollment_token: Optional[str] = None # if omitted, one is auto-generated class DeviceEnrollRequest(BaseModel): enrollment_token: str name: Optional[str] = None tailscale_ip: Optional[str] = None lan_ip: Optional[str] = None motor_server_url: Optional[str] = None livekit_url: Optional[str] = None character_id: Optional[str] = None class DeviceEnrollResponse(BaseModel): device_id: str device_token: str scheduler_url: str class DeviceHeartbeat(BaseModel): status: Optional[str] = None ip: Optional[str] = None uptime_seconds: Optional[int] = None production_mode: Optional[bool] = None class SettingsPayload(BaseModel): production_mode: bool class LiveKitConfig(BaseModel): url: Optional[str] = None api_key: Optional[str] = None has_secret: bool = False class LiveKitTokenRequest(BaseModel): room: str identity: str name: Optional[str] = None can_publish: bool = True can_subscribe: bool = True ttl_seconds: int = 3600 class LiveKitTokenResponse(BaseModel): url: str token: str room: str identity: str expires_at: datetime # ============================================================================= # DATABASE # ============================================================================= async def init_db(): """Initialize SQLite database""" os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) async with aiosqlite.connect(DB_PATH) as db: await db.executescript(""" CREATE TABLE IF NOT EXISTS robots ( id TEXT PRIMARY KEY, name TEXT NOT NULL, character_id TEXT, status TEXT DEFAULT 'idle', current_session_id TEXT, connected_server_id TEXT, ip_address TEXT, last_heartbeat TEXT, total_sessions INTEGER DEFAULT 0, total_runtime_seconds INTEGER DEFAULT 0, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS livekit_servers ( id TEXT PRIMARY KEY, name TEXT NOT NULL, url TEXT NOT NULL, webhook_url TEXT NOT NULL, api_key TEXT, api_secret TEXT, gpu_name TEXT, gpu_vram_mb INTEGER DEFAULT 0, max_sessions INTEGER DEFAULT 8, status TEXT DEFAULT 'unknown', created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, robot_id TEXT, server_id TEXT, room_name TEXT, started_at TEXT, ended_at TEXT, end_reason TEXT, duration_seconds INTEGER, FOREIGN KEY (robot_id) REFERENCES robots(id), FOREIGN KEY (server_id) REFERENCES livekit_servers(id) ); CREATE TABLE IF NOT EXISTS metrics_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, server_id TEXT, gpu_utilization INTEGER, vram_used_mb INTEGER, active_sessions INTEGER, recorded_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS devices ( id TEXT PRIMARY KEY, name TEXT NOT NULL, tailscale_ip TEXT, lan_ip TEXT, motor_server_url TEXT, character_id TEXT, livekit_url TEXT, token_hash TEXT, enrollment_token_hash TEXT, status TEXT DEFAULT 'enrolled', last_heartbeat TEXT, enrolled_at TEXT, last_seen_ip TEXT, notes TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS previews ( robot_id TEXT PRIMARY KEY, room_name TEXT NOT NULL, server_id TEXT, expires_at TEXT NOT NULL, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); """) await db.commit() # ---- Additive telemetry migrations (safe on existing DBs) ---- # trigger_count on robots (camera-trigger counter) and joined_at on # sessions (room-join timestamp for latency). SQLite's # ALTER TABLE ... ADD COLUMN raises if the column already exists, so # each one is guarded and idempotent. for _table, _column, _coldef in ( ("robots", "trigger_count", "INTEGER DEFAULT 0"), ("sessions", "joined_at", "TEXT"), ): try: await db.execute(f"ALTER TABLE {_table} ADD COLUMN {_column} {_coldef}") except Exception: pass # column already present await db.commit() # Insert default robots if not exist default_robots = [ ("robopanda", "RoboPanda", "panda-character"), ("robobear", "RoboBear", "bear-character"), ("robocar", "RoboCar", "car-character"), ("robodragon", "RoboDragon", "dragon-character"), ("robofrog", "RoboFrog", "frog-character"), ] for robot_id, name, char_id in default_robots: await db.execute( "INSERT OR IGNORE INTO robots (id, name, character_id) VALUES (?, ?, ?)", (robot_id, name, char_id) ) await db.commit() # Seed default enrollment token + production_mode flag (only on first run) async with db.execute("SELECT value FROM settings WHERE key = 'default_enrollment_token'") as c: row = await c.fetchone() if not row: token = secrets.token_urlsafe(24) token_hash = hashlib.sha256(token.encode()).hexdigest() await db.execute( "INSERT INTO settings (key, value) VALUES (?, ?)", ("default_enrollment_token", token_hash) ) logger.info("=" * 70) logger.info(f"DEFAULT ENROLLMENT TOKEN (Pi first-boot): {token}") logger.info("Use this on the Pi: --enrollment-token ") logger.info("Rotate or replace it in the UI under Settings.") logger.info("=" * 70) async with db.execute("SELECT value FROM settings WHERE key = 'production_mode'") as c: row = await c.fetchone() if not row: await db.execute( "INSERT INTO settings (key, value) VALUES (?, ?)", ("production_mode", "false") ) await db.commit() logger.info("Database initialized") # ============================================================================= # APP SETUP # ============================================================================= @asynccontextmanager async def lifespan(app: FastAPI): await init_db() asyncio.create_task(metrics_collector()) asyncio.create_task(health_checker()) yield app = FastAPI( title="RoboPark Session Scheduler", description="Manages robot fleet and LiveKit server orchestration", version="1.0.0", lifespan=lifespan ) # CORS: env-driven allowlist. Set ROBOPARK_CORS_ORIGINS to a comma-separated # list of allowed origins; defaults to localhost only. A wildcard origin is # incompatible with credentialed requests, so credentials are only enabled # when an explicit (non-wildcard) allowlist is configured. _cors_env = os.getenv("ROBOPARK_CORS_ORIGINS", "http://localhost:3000,http://127.0.0.1:3000") _cors_origins = [o.strip() for o in _cors_env.split(",") if o.strip()] _allow_credentials = "*" not in _cors_origins app.add_middleware( CORSMiddleware, allow_origins=_cors_origins, allow_credentials=_allow_credentials, allow_methods=["*"], allow_headers=["*"], ) # WebSocket clients for real-time updates ws_clients: List[WebSocket] = [] # ============================================================================= # SETTINGS HELPERS # ============================================================================= def _hash(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() async def get_setting(key: str, default: Optional[str] = None) -> Optional[str]: async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT value FROM settings WHERE key = ?", (key,)) as c: row = await c.fetchone() return row[0] if row else default async def set_setting(key: str, value: str) -> None: async with aiosqlite.connect(DB_PATH) as db: await db.execute( "INSERT INTO settings (key, value, updated_at) VALUES (?, ?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at", (key, value, datetime.utcnow().isoformat()), ) await db.commit() async def get_production_mode() -> bool: v = await get_setting("production_mode", "false") return (v or "false").lower() == "true" # ============================================================================= # WEBSOCKET # ============================================================================= @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() ws_clients.append(websocket) logger.info(f"WebSocket client connected. Total: {len(ws_clients)}") try: while True: await websocket.receive_text() except WebSocketDisconnect: ws_clients.remove(websocket) logger.info(f"WebSocket client disconnected. Total: {len(ws_clients)}") async def broadcast(event_type: str, data: dict): """Broadcast update to all connected WebSocket clients""" if not ws_clients: return message = {"type": event_type, "data": data, "timestamp": datetime.utcnow().isoformat()} for client in ws_clients.copy(): try: await client.send_json(message) except: ws_clients.remove(client) # ============================================================================= # ROBOT ENDPOINTS # ============================================================================= @app.get("/api/robots", response_model=List[Robot]) async def list_robots(): """List all robots""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM robots ORDER BY name") as cursor: rows = await cursor.fetchall() return [dict(row) for row in rows] @app.get("/api/robots/{robot_id}", response_model=Robot) async def get_robot(robot_id: str): """Get single robot""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM robots WHERE id = ?", (robot_id,)) as cursor: row = await cursor.fetchone() if not row: raise HTTPException(404, "Robot not found") return dict(row) @app.post("/api/robots/{robot_id}/heartbeat") async def robot_heartbeat(robot_id: str, ip_address: Optional[str] = None): """Robot heartbeat - call every 5 seconds""" async with aiosqlite.connect(DB_PATH) as db: await db.execute( "UPDATE robots SET last_heartbeat = ?, ip_address = COALESCE(?, ip_address) WHERE id = ?", (datetime.utcnow().isoformat(), ip_address, robot_id) ) await db.commit() await broadcast("robot_heartbeat", {"robot_id": robot_id}) return {"status": "ok"} @app.post("/api/robots/{robot_id}/request-session") async def request_session(robot_id: str, authorization: Optional[str] = Header(default=None)): """Robot requests a session after scene detection. Requires a valid enrolled-device Bearer token. The response never contains the LiveKit api_key/api_secret; instead the scheduler mints a short-lived JWT the robot uses to join the room.""" if not await _authorize_fleet(authorization): raise HTTPException(401, "Invalid or missing device token") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row # Check robot exists and is idle async with db.execute("SELECT * FROM robots WHERE id = ?", (robot_id,)) as cursor: robot = await cursor.fetchone() if not robot: raise HTTPException(404, "Robot not found") if robot["status"] != "idle": raise HTTPException(400, f"Robot is {robot['status']}, not idle") # Find least-loaded online server async with db.execute(""" SELECT s.*, (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) as active FROM livekit_servers s WHERE s.status = 'online' ORDER BY active ASC LIMIT 1 """) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(503, "No available servers") if server["active"] >= server["max_sessions"]: raise HTTPException(503, "All servers at capacity") # Create session session_id = f"session_{robot_id}_{int(datetime.utcnow().timestamp())}" room_name = f"robopark_{robot_id}_{int(datetime.utcnow().timestamp())}" await db.execute(""" INSERT INTO sessions (id, robot_id, server_id, room_name, started_at) VALUES (?, ?, ?, ?, ?) """, (session_id, robot_id, server["id"], room_name, datetime.utcnow().isoformat())) # Update robot status + count this as a camera trigger. request-session # is the natural trigger point ("robot requests a session after scene # detection"), so the trigger_count populates here with no client change. # A client that wants to report a camera-open that does NOT reach # request-session can call POST /api/robots/{robot_id}/trigger instead # (it should not call both for the same detection, to avoid double count). await db.execute(""" UPDATE robots SET status = 'connecting', current_session_id = ?, connected_server_id = ?, trigger_count = COALESCE(trigger_count, 0) + 1 WHERE id = ? """, (session_id, server["id"], robot_id)) await db.commit() await broadcast("session_requested", { "robot_id": robot_id, "server_id": server["id"], "session_id": session_id }) # Mint a short-lived LiveKit token so the robot can join the room without # ever receiving the raw api_key/api_secret. try: from livekit.api import AccessToken, VideoGrants except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") lk_key = server["api_key"] or "devkey" lk_secret = server["api_secret"] or "secret" grants = VideoGrants( room=room_name, room_join=True, can_publish=True, can_subscribe=True, can_publish_data=True, ) token = ( AccessToken(lk_key, lk_secret) .with_identity(robot_id) .with_name(robot_id) .with_ttl(timedelta(hours=1)) .with_grants(grants) .to_jwt() ) return { "session_id": session_id, "server_id": server["id"], "server_url": server["url"], "room_name": room_name, "token": token, } @app.post("/api/robots/{robot_id}/end-session") async def end_session(robot_id: str, reason: str = "silence"): """Robot ends session""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM robots WHERE id = ?", (robot_id,)) as cursor: robot = await cursor.fetchone() if not robot or not robot["current_session_id"]: raise HTTPException(400, "No active session") # Update session async with db.execute("SELECT * FROM sessions WHERE id = ?", (robot["current_session_id"],)) as cursor: session = await cursor.fetchone() if session: started = datetime.fromisoformat(session["started_at"]) duration = int((datetime.utcnow() - started).total_seconds()) await db.execute(""" UPDATE sessions SET ended_at = ?, end_reason = ?, duration_seconds = ? WHERE id = ? """, (datetime.utcnow().isoformat(), reason, duration, session["id"])) # Update robot stats await db.execute(""" UPDATE robots SET status = 'idle', current_session_id = NULL, connected_server_id = NULL, total_sessions = total_sessions + 1, total_runtime_seconds = total_runtime_seconds + ? WHERE id = ? """, (duration, robot_id)) await db.commit() await broadcast("session_ended", {"robot_id": robot_id, "reason": reason}) return {"status": "ok"} @app.post("/api/robots/{robot_id}/trigger") async def robot_trigger(robot_id: str, authorization: Optional[str] = Header(default=None)): """Record a camera trigger for a robot. Called when the robot's camera detects a scene / opens a session, to keep a per-robot trigger counter. Requires a valid enrolled-device Bearer token (same fleet auth as /request-session). NOTE: /request-session already increments trigger_count as the natural trigger point. Use this endpoint only to report a camera-open that does NOT proceed to /request-session; do not call both for the same detection or the trigger will be counted twice.""" if not await _authorize_fleet(authorization): raise HTTPException(401, "Invalid or missing device token") async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute( "UPDATE robots SET trigger_count = COALESCE(trigger_count, 0) + 1 WHERE id = ?", (robot_id,), ) await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Robot not found") async with db.execute("SELECT trigger_count FROM robots WHERE id = ?", (robot_id,)) as c: row = await c.fetchone() await broadcast("robot_triggered", {"robot_id": robot_id}) return {"status": "ok", "trigger_count": row[0] if row else None} # ============================================================================= # SERVER ENDPOINTS # ============================================================================= @app.get("/api/servers") async def list_servers(): """List all LiveKit servers with current load""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute(""" SELECT s.*, (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) as active_sessions FROM livekit_servers s ORDER BY s.name """) as cursor: return [dict(row) for row in await cursor.fetchall()] @app.post("/api/servers") async def add_server(server: LiveKitServer): """Add a new LiveKit server""" async with aiosqlite.connect(DB_PATH) as db: await db.execute(""" INSERT OR REPLACE INTO livekit_servers (id, name, url, webhook_url, api_key, api_secret, gpu_name, gpu_vram_mb, max_sessions, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (server.id, server.name, server.url, server.webhook_url, server.api_key, server.api_secret, server.gpu_name, server.gpu_vram_mb, server.max_sessions, server.status)) await db.commit() await broadcast("server_added", {"server_id": server.id}) return {"status": "ok"} @app.delete("/api/servers/{server_id}") async def remove_server(server_id: str): """Remove a LiveKit server""" async with aiosqlite.connect(DB_PATH) as db: await db.execute("DELETE FROM livekit_servers WHERE id = ?", (server_id,)) await db.commit() await broadcast("server_removed", {"server_id": server_id}) return {"status": "ok"} @app.get("/api/servers/{server_id}/metrics") async def get_server_metrics(server_id: str) -> ServerMetrics: """Get real-time metrics from a LiveKit server""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM livekit_servers WHERE id = ?", (server_id,)) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(404, "Server not found") # Fetch metrics from server try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.get(f"{server['webhook_url']}/metrics") data = resp.json() # Get models models_resp = await client.get(f"{server['webhook_url']}/models") models_data = models_resp.json() models = [] for m in models_data.get("models", []): models.append(GpuModel( name=m["name"], size_gb=m.get("size", 0) / (1024**3), is_loaded=True, vram_used_mb=int(m.get("size", 0) / (1024**2)) )) return ServerMetrics( server_id=server_id, gpu_utilization=data.get("gpu_utilization", 0), vram_used_mb=data.get("vram_used_mb", 0), vram_total_mb=data.get("vram_total_mb", 0), active_sessions=data.get("active_sessions", 0), models=models ) except Exception as e: logger.error(f"Failed to get metrics from {server_id}: {e}") return ServerMetrics(server_id=server_id) @app.post("/api/servers/{server_id}/models/{model_name}/load") async def load_model(server_id: str, model_name: str): """Load a model on a server""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT webhook_url FROM livekit_servers WHERE id = ?", (server_id,)) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(404, "Server not found") try: async with httpx.AsyncClient(timeout=120.0) as client: resp = await client.post(f"{server['webhook_url']}/models/{model_name}/load") await broadcast("model_loaded", {"server_id": server_id, "model": model_name}) return resp.json() except Exception as e: raise HTTPException(500, f"Failed to load model: {e}") @app.post("/api/servers/{server_id}/models/{model_name}/unload") async def unload_model(server_id: str, model_name: str): """Unload a model from a server""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT webhook_url FROM livekit_servers WHERE id = ?", (server_id,)) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(404, "Server not found") try: async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post(f"{server['webhook_url']}/models/{model_name}/unload") await broadcast("model_unloaded", {"server_id": server_id, "model": model_name}) return resp.json() except Exception as e: raise HTTPException(500, f"Failed to unload model: {e}") # ============================================================================= # SESSION ENDPOINTS # ============================================================================= class WebClientSessionRequest(BaseModel): """Request body for web client session""" client_name: Optional[str] = "Web User" class WebClientSessionResponse(BaseModel): """Response for web client session""" session_id: str server_id: str server_url: str room_name: str api_key: str api_secret: str @app.post("/api/sessions/web-client", response_model=WebClientSessionResponse) async def create_web_client_session(req: Optional[WebClientSessionRequest] = None): """Create a session for a web client (not tied to a robot)""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row # Find least-loaded online server async with db.execute(""" SELECT s.*, (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) as active FROM livekit_servers s WHERE s.status = 'online' ORDER BY active ASC LIMIT 1 """) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(503, "No available servers") if server["active"] >= server["max_sessions"]: raise HTTPException(503, "All servers at capacity") # Create session session_id = f"web_{int(datetime.utcnow().timestamp())}" room_name = f"robopark_web_{int(datetime.utcnow().timestamp())}" await db.execute(""" INSERT INTO sessions (id, robot_id, server_id, room_name, started_at) VALUES (?, ?, ?, ?, ?) """, (session_id, "web-client", server["id"], room_name, datetime.utcnow().isoformat())) await db.commit() await broadcast("web_session_started", { "session_id": session_id, "server_id": server["id"], }) return WebClientSessionResponse( session_id=session_id, server_id=server["id"], server_url=server["url"], room_name=room_name, api_key=server["api_key"] or "devkey", api_secret=server["api_secret"] or "secret", ) @app.post("/api/sessions/{session_id}/end") async def end_web_session(session_id: str, reason: str = "disconnect"): """End a web client session""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)) as cursor: session = await cursor.fetchone() if not session: return {"status": "not_found"} if session["ended_at"]: return {"status": "already_ended"} started = datetime.fromisoformat(session["started_at"]) duration = int((datetime.utcnow() - started).total_seconds()) await db.execute(""" UPDATE sessions SET ended_at = ?, end_reason = ?, duration_seconds = ? WHERE id = ? """, (datetime.utcnow().isoformat(), reason, duration, session_id)) await db.commit() await broadcast("session_ended", {"session_id": session_id, "reason": reason}) return {"status": "ok"} def _session_latency_ms(started_at: Optional[str], joined_at: Optional[str]) -> Optional[int]: """Milliseconds between session request (started_at) and room join (joined_at).""" if not started_at or not joined_at: return None try: s = datetime.fromisoformat(started_at) j = datetime.fromisoformat(joined_at) return int((j - s).total_seconds() * 1000) except Exception: return None @app.post("/api/sessions/{session_id}/joined") async def session_joined(session_id: str, authorization: Optional[str] = Header(default=None)): """Mark the moment a robot successfully joined the LiveKit room for a session. Used to compute session latency = joined_at - started_at (the session row's started_at is set when the session is requested). Requires a valid enrolled-device Bearer token. Idempotent: the first join wins; later calls leave the original join time in place.""" if not await _authorize_fleet(authorization): raise HTTPException(401, "Invalid or missing device token") now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)) as c: session = await c.fetchone() if not session: raise HTTPException(404, "Session not found") if session["joined_at"]: return { "status": "already_joined", "joined_at": session["joined_at"], "latency_ms": _session_latency_ms(session["started_at"], session["joined_at"]), } await db.execute("UPDATE sessions SET joined_at = ? WHERE id = ?", (now, session_id)) await db.commit() latency_ms = _session_latency_ms(session["started_at"], now) await broadcast("session_joined", {"session_id": session_id, "latency_ms": latency_ms}) return {"status": "ok", "joined_at": now, "latency_ms": latency_ms} # ── Back-compat: legacy web-client session token (ported from the deployed OLD # scheduler so voice_client.html / the old frontend keep working after the # fork reconcile). Raw-PyJWT HS256, publish-capable "Web User" token. ── @app.get("/api/sessions/{session_id}/token") async def get_session_token(session_id: str): """Generate a LiveKit access token for a session""" import jwt async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute(""" SELECT s.*, srv.api_key, srv.api_secret FROM sessions s JOIN livekit_servers srv ON s.server_id = srv.id WHERE s.id = ? """, (session_id,)) as cursor: session = await cursor.fetchone() if not session: raise HTTPException(404, "Session not found") if session["ended_at"]: raise HTTPException(400, "Session already ended") api_key = session["api_key"] or "devkey" api_secret = session["api_secret"] or "secret" room_name = session["room_name"] # Generate token token = jwt.encode( { "name": "Web User", "video": { "room": room_name, "roomJoin": True, "canPublish": True, "canPublishData": True, "canSubscribe": True }, "iss": api_key, "nbf": 0, "exp": int(datetime.utcnow().timestamp()) + 3600, # 1 hour "sub": f"web_user_{int(datetime.utcnow().timestamp())}" }, api_secret, algorithm="HS256" ) return {"token": token, "room_name": room_name} @app.get("/api/sessions") async def list_sessions(active_only: bool = False, limit: int = 50): """List sessions""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row if active_only: query = "SELECT * FROM sessions WHERE ended_at IS NULL ORDER BY started_at DESC" else: query = f"SELECT * FROM sessions ORDER BY started_at DESC LIMIT {limit}" async with db.execute(query) as cursor: return [dict(row) for row in await cursor.fetchall()] @app.get("/api/sessions/stats") async def session_stats(): """Get session statistics""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row today = datetime.utcnow().date().isoformat() stats = {} # Today's sessions async with db.execute( "SELECT COUNT(*) as count FROM sessions WHERE started_at LIKE ?", (f"{today}%",) ) as cursor: row = await cursor.fetchone() stats["today_sessions"] = row["count"] # Active sessions async with db.execute( "SELECT COUNT(*) as count FROM sessions WHERE ended_at IS NULL" ) as cursor: row = await cursor.fetchone() stats["active_sessions"] = row["count"] # Average duration async with db.execute( "SELECT AVG(duration_seconds) as avg FROM sessions WHERE duration_seconds IS NOT NULL" ) as cursor: row = await cursor.fetchone() stats["avg_duration_seconds"] = int(row["avg"] or 0) # Total sessions async with db.execute("SELECT COUNT(*) as count FROM sessions") as cursor: row = await cursor.fetchone() stats["total_sessions"] = row["count"] return stats # ============================================================================= # WEBHOOK RECEIVER # ============================================================================= @app.post("/api/webhooks/livekit") async def livekit_webhook(event: WebhookEvent): """Receive events from LiveKit servers""" logger.info(f"Webhook received: {event.event} - {event.data}") if event.event == "session_started": robot_id = event.data.get("robot_id") if robot_id: async with aiosqlite.connect(DB_PATH) as db: await db.execute( "UPDATE robots SET status = 'running' WHERE id = ?", (robot_id,) ) await db.commit() await broadcast("session_started", event.data) elif event.event == "session_ended": robot_id = event.data.get("robot_id") if robot_id: # Use the end_session logic try: await end_session(robot_id, event.data.get("reason", "unknown")) except: pass return {"status": "ok"} # ============================================================================= # BACKGROUND TASKS # ============================================================================= async def metrics_collector(): """Collect metrics from all servers periodically""" while True: try: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM livekit_servers WHERE status = 'online'") as cursor: servers = await cursor.fetchall() all_metrics = [] for server in servers: try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.get(f"{server['webhook_url']}/metrics") data = resp.json() # Store in history await db.execute(""" INSERT INTO metrics_history (server_id, gpu_utilization, vram_used_mb, active_sessions) VALUES (?, ?, ?, ?) """, (server["id"], data.get("gpu_utilization", 0), data.get("vram_used_mb", 0), data.get("active_sessions", 0))) all_metrics.append({ "server_id": server["id"], **data }) except Exception as e: logger.warning(f"Failed to collect metrics from {server['id']}: {e}") await db.commit() if all_metrics: await broadcast("metrics_update", {"servers": all_metrics}) except Exception as e: logger.error(f"Metrics collector error: {e}") await asyncio.sleep(5) # Collect every 5 seconds async def health_checker(): """Check server health periodically""" while True: try: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM livekit_servers") as cursor: servers = await cursor.fetchall() for server in servers: try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.get(f"{server['webhook_url']}/health") if resp.status_code == 200: new_status = "online" else: new_status = "offline" except: new_status = "offline" if new_status != server["status"]: await db.execute( "UPDATE livekit_servers SET status = ? WHERE id = ?", (new_status, server["id"]) ) await broadcast("server_status_changed", { "server_id": server["id"], "status": new_status }) # Mark devices offline if heartbeat is stale (> 30s) cutoff = (datetime.utcnow() - timedelta(seconds=30)).isoformat() async with db.execute( "SELECT id, status FROM devices WHERE status = 'online' AND (last_heartbeat IS NULL OR last_heartbeat < ?)", (cutoff,), ) as c: stale = await c.fetchall() for (did, _st) in stale: await db.execute("UPDATE devices SET status = 'offline' WHERE id = ?", (did,)) await broadcast("device_status_changed", {"device_id": did, "status": "offline"}) await db.commit() except Exception as e: logger.error(f"Health checker error: {e}") await asyncio.sleep(10) # Check every 10 seconds # ============================================================================= # SETTINGS ENDPOINTS # ============================================================================= @app.get("/api/settings") async def get_settings(): """Public settings (production mode + whether default enrollment token exists).""" pm = await get_production_mode() has_default_token = bool(await get_setting("default_enrollment_token")) return { "production_mode": pm, "default_enrollment_token_set": has_default_token, } @app.put("/api/settings") async def update_settings(payload: SettingsPayload): """Update scheduler settings (production mode, etc.).""" await set_setting("production_mode", "true" if payload.production_mode else "false") await broadcast("settings_changed", {"production_mode": payload.production_mode}) return {"status": "ok", "production_mode": payload.production_mode} @app.post("/api/settings/enrollment-token/rotate") async def rotate_enrollment_token(): """Generate a fresh default enrollment token (rotates the secret).""" token = secrets.token_urlsafe(24) await set_setting("default_enrollment_token", _hash(token)) logger.info("Default enrollment token rotated") return {"enrollment_token": token} # ============================================================================= # LIVEKIT CONFIG + TOKEN ISSUANCE # ============================================================================= # # The scheduler is the single source of truth for LiveKit credentials. # The Pi UI (or any client) calls /api/livekit/token to receive a short-lived # JWT for a given room + identity, signed with the LiveKit API key/secret # that the scheduler was started with. # # Required env vars (set in docker-compose or the host): # LIVEKIT_URL e.g. ws://localhost:7880 # LIVEKIT_API_KEY e.g. devkey # LIVEKIT_API_SECRET e.g. # # These can also be set via the API (PUT /api/livekit/config) and are stored # in the settings table; env vars take precedence at startup. def _lk_settings() -> tuple[Optional[str], Optional[str], Optional[str]]: """Return (url, api_key, api_secret). Env overrides stored settings.""" url = os.getenv("LIVEKIT_URL") or None key = os.getenv("LIVEKIT_API_KEY") or None sec = os.getenv("LIVEKIT_API_SECRET") or None return url, key, sec async def _lk_config_async() -> tuple[str, str, str]: """Async lookup that falls back to settings table for any missing env value.""" url, key, sec = _lk_settings() if not url: url = await get_setting("livekit_url") if not key: key = await get_setting("livekit_api_key") if not sec: sec = await get_setting("livekit_api_secret") if not (url and key and sec): raise HTTPException( 503, "LiveKit is not configured. Set LIVEKIT_URL / LIVEKIT_API_KEY / " "LIVEKIT_API_SECRET env vars, or PUT /api/livekit/config.", ) return url, key, sec @app.get("/api/livekit/config", response_model=LiveKitConfig) async def get_livekit_config(): """Show the current LiveKit configuration. Secret is never returned.""" url, key, sec = _lk_settings() has_secret = bool(sec) if not url: url = await get_setting("livekit_url") if not key: key = await get_setting("livekit_api_key") if not has_secret: has_secret = bool(await get_setting("livekit_api_secret")) return LiveKitConfig(url=url, api_key=key, has_secret=has_secret) @app.put("/api/livekit/config") async def update_livekit_config(payload: LiveKitConfig): """Persist LiveKit config. If api_key is omitted, the existing key is kept. If api_secret is None, it is left as-is; if it is the empty string it is cleared.""" if payload.url is not None: await set_setting("livekit_url", payload.url) if payload.api_key is not None: await set_setting("livekit_api_key", payload.api_key) # Note: there's no separate endpoint to set the secret via JSON to avoid # accidental overwrites; use POST /api/livekit/config/secret for that. await broadcast("livekit_config_changed", {}) return {"status": "ok"} @app.post("/api/livekit/config/secret") async def set_livekit_secret(payload: dict): secret = payload.get("api_secret") if not isinstance(secret, str) or len(secret) < 8: raise HTTPException(400, "api_secret must be a string of length >= 8") await set_setting("livekit_api_secret", secret) await broadcast("livekit_config_changed", {}) return {"status": "ok"} @app.post("/api/livekit/token", response_model=LiveKitTokenResponse) async def issue_livekit_token(payload: LiveKitTokenRequest, authorization: Optional[str] = Header(default=None)): """Sign a short-lived LiveKit access token for a given room + identity. Pi clients (or the Pi UI) use this to join a room in the browser. Requires a valid enrolled-device Bearer token and production_mode to be ON.""" if not await _authorize_fleet(authorization): raise HTTPException(401, "Invalid or missing device token") if not await get_production_mode(): raise HTTPException(403, "Production mode is disabled") if not payload.room or not payload.identity: raise HTTPException(400, "room and identity are required") if not re.match(r"^[A-Za-z0-9_\-=]{1,128}$", payload.room): raise HTTPException(400, "invalid room name") if not re.match(r"^[A-Za-z0-9_\-:.@]{1,128}$", payload.identity): raise HTTPException(400, "invalid identity") url, key, sec = await _lk_config_async() ttl = max(60, min(int(payload.ttl_seconds), 6 * 3600)) try: from livekit.api import AccessToken, VideoGrants except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") grants = VideoGrants( room=payload.room, room_join=True, can_publish=payload.can_publish, can_subscribe=payload.can_subscribe, can_publish_data=True, ) at = AccessToken(key, sec) \ .with_identity(payload.identity) \ .with_name(payload.name or payload.identity) \ .with_ttl(timedelta(seconds=ttl)) \ .with_grants(grants) token = at.to_jwt() expires = datetime.utcnow() + timedelta(seconds=ttl) return LiveKitTokenResponse( url=url, token=token, room=payload.room, identity=payload.identity, expires_at=expires, ) @app.get("/api/robots/{robot_id}/stream") async def robot_stream(robot_id: str, authorization: Optional[str] = Header(default=None)): """Mint a SUBSCRIBE-ONLY LiveKit token so an operator dashboard can watch a robot's live camera + audio for its CURRENT session. The Pi already publishes a 'cam' (video) and 'mic' (audio) track into its session room (pi-client livekit_bridge _publish_cam/_publish_mic), so a viewer just needs a subscribe-only token for that room. Returns {"active": false} when the robot has no live session (nothing to view yet). Token is signed with the robot's ASSIGNED server key (the same key that signed the publisher) so it validates against the room the robot publishes into. Subscribe-only, short TTL, and only while production_mode is ON. This sits at the same (open) tier as the other /api/robots read routes; to tighten, add `if not await _authorize_fleet(authorization): raise HTTPException(401, ...)` like /api/livekit/token.""" if not await get_production_mode(): return {"active": False, "reason": "production_mode_off"} async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM robots WHERE id = ?", (robot_id,)) as c: robot = await c.fetchone() if not robot: raise HTTPException(404, "Robot not found") if not robot["current_session_id"]: return {"active": False, "reason": "idle"} async with db.execute("SELECT * FROM sessions WHERE id = ?", (robot["current_session_id"],)) as c: session = await c.fetchone() if not session or session["ended_at"]: return {"active": False, "reason": "idle"} server = None if session["server_id"]: async with db.execute("SELECT * FROM livekit_servers WHERE id = ?", (session["server_id"],)) as c: server = await c.fetchone() room_name = session["room_name"] if server: url, lk_key, lk_secret = server["url"], (server["api_key"] or "devkey"), (server["api_secret"] or "secret") else: url, lk_key, lk_secret = await _lk_config_async() try: from livekit.api import AccessToken, VideoGrants except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") identity = f"viewer:{robot_id}:{int(datetime.utcnow().timestamp())}" grants = VideoGrants( room=room_name, room_join=True, can_publish=False, can_publish_data=False, can_subscribe=True, ) token = ( AccessToken(lk_key, lk_secret) .with_identity(identity) .with_name("operator-dashboard") .with_ttl(timedelta(minutes=15)) .with_grants(grants) .to_jwt() ) return { "active": True, "url": url, "token": token, "room": room_name, "identity": identity, } # ============================================================================= # ON-DEMAND CAMERA PREVIEW (operator-initiated, independent of scene detection) # ============================================================================= # Flow: dashboard POSTs /preview/start -> a preview room is assigned with a short # TTL and the operator gets a SUBSCRIBE-only viewer token. The robot polls # GET /preview/agent; while a preview is active it receives a PUBLISH token and # opens its cam into the same room. The dashboard POSTs /preview/keepalive while # watching; closing it POSTs /preview/stop. If keepalives stop, the preview # auto-expires and the robot's next poll returns {active:false} so it stops the # cam. Cam therefore runs ONLY while an operator is watching. PREVIEW_TTL_SECONDS = 45 # preview auto-expires this long after the last keepalive async def _pick_preview_server(db, prefer_id: Optional[str] = None): """Return a LiveKit server row for a preview: a preferred one if still present, else the least-loaded online server, else any configured server.""" if prefer_id: async with db.execute("SELECT * FROM livekit_servers WHERE id = ?", (prefer_id,)) as c: row = await c.fetchone() if row: return row async with db.execute(""" SELECT s.* FROM livekit_servers s WHERE s.status = 'online' OR s.status IS NULL OR s.status = 'unknown' ORDER BY (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) ASC LIMIT 1 """) as c: row = await c.fetchone() if row: return row async with db.execute("SELECT * FROM livekit_servers LIMIT 1") as c: return await c.fetchone() def _preview_active(row) -> bool: if not row: return False try: return datetime.fromisoformat(row["expires_at"]) > datetime.utcnow() except Exception: return False def _mint_lk(server, room, identity, name, publish: bool, ttl_min: int): """Mint a LiveKit token via livekit-api, matching the other routes' pattern.""" from livekit.api import AccessToken, VideoGrants grants = VideoGrants( room=room, room_join=True, can_publish=publish, can_publish_data=publish, can_subscribe=not publish, ) return ( AccessToken(server["api_key"] or "devkey", server["api_secret"] or "secret") .with_identity(identity).with_name(name) .with_ttl(timedelta(minutes=ttl_min)).with_grants(grants).to_jwt() ) @app.post("/api/robots/{robot_id}/preview/start") async def preview_start(robot_id: str): """Operator dashboard: begin an on-demand live camera preview for a robot. Returns a SUBSCRIBE-only viewer token + the preview room.""" if not await get_production_mode(): return {"active": False, "reason": "production_mode_off"} async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM robots WHERE id = ?", (robot_id,)) as c: if not await c.fetchone(): raise HTTPException(404, "Robot not found") async with db.execute("SELECT * FROM previews WHERE robot_id = ?", (robot_id,)) as c: existing = await c.fetchone() server = await _pick_preview_server(db, existing["server_id"] if existing else None) if not server: raise HTTPException(503, "No LiveKit server configured") room_name = (existing["room_name"] if existing else None) or f"preview_{robot_id}" expires = (datetime.utcnow() + timedelta(seconds=PREVIEW_TTL_SECONDS)).isoformat() await db.execute(""" INSERT INTO previews (robot_id, room_name, server_id, expires_at) VALUES (?, ?, ?, ?) ON CONFLICT(robot_id) DO UPDATE SET room_name=excluded.room_name, server_id=excluded.server_id, expires_at=excluded.expires_at """, (robot_id, room_name, server["id"], expires)) await db.commit() try: identity = f"viewer:{robot_id}:{int(datetime.utcnow().timestamp())}" token = _mint_lk(server, room_name, identity, "operator-preview", publish=False, ttl_min=5) except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") return {"active": True, "url": server["url"], "token": token, "room": room_name, "identity": identity, "expires_in": PREVIEW_TTL_SECONDS} @app.post("/api/robots/{robot_id}/preview/keepalive") async def preview_keepalive(robot_id: str): """Dashboard pings while the operator is watching, to keep the preview alive.""" expires = (datetime.utcnow() + timedelta(seconds=PREVIEW_TTL_SECONDS)).isoformat() async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute("UPDATE previews SET expires_at = ? WHERE robot_id = ?", (expires, robot_id)) await db.commit() if cur.rowcount == 0: return {"active": False} return {"active": True, "expires_in": PREVIEW_TTL_SECONDS} @app.post("/api/robots/{robot_id}/preview/stop") async def preview_stop(robot_id: str): """Dashboard closed / operator stopped watching -> end the preview now.""" async with aiosqlite.connect(DB_PATH) as db: await db.execute("DELETE FROM previews WHERE robot_id = ?", (robot_id,)) await db.commit() return {"stopped": True} @app.get("/api/robots/{robot_id}/preview/agent") async def preview_agent(robot_id: str, authorization: Optional[str] = Header(default=None)): """The ROBOT polls this. While an operator preview is active it returns a PUBLISH token + room so the Pi opens its cam and joins; otherwise {active:false} so the Pi stops publishing. Enrolled-device auth.""" if not await _authorize_fleet(authorization): raise HTTPException(401, "Invalid or missing device token") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM previews WHERE robot_id = ?", (robot_id,)) as c: row = await c.fetchone() if not _preview_active(row): return {"active": False} # If the robot is in a real scene-detection session, ROBOVOICE owns the # camera device — don't have the preview agent fight it for /dev/video0. # The operator sees the live session cam via /stream instead. async with db.execute("SELECT current_session_id FROM robots WHERE id = ?", (robot_id,)) as c: rob = await c.fetchone() if rob and rob["current_session_id"]: return {"active": False, "reason": "in_session"} server = await _pick_preview_server(db, row["server_id"]) if not server: return {"active": False} try: token = _mint_lk(server, row["room_name"], robot_id, robot_id, publish=True, ttl_min=2) except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") return {"active": True, "url": server["url"], "token": token, "room": row["room_name"]} # ============================================================================= # DEVICE ENDPOINTS (Pi fleet) # ============================================================================= def _device_row_to_model(row) -> Device: """Convert a sqlite Row (or dict) to a Device, normalizing fields.""" d = dict(row) d.pop("token_hash", None) d.pop("enrollment_token_hash", None) return Device(**d) @app.get("/api/devices", response_model=List[Device]) async def list_devices(): """List all enrolled Pi devices.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM devices ORDER BY name") as cursor: return [_device_row_to_model(r) for r in await cursor.fetchall()] @app.get("/api/devices/by-room/{room_name}") async def get_device_by_room(room_name: str): """Look up the device bound to a LiveKit room. Convention: room = 'robopark-'.""" if not room_name.startswith("robopark-"): raise HTTPException(404, "Room is not a robopark room") device_id = room_name[len("robopark-"):] async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM devices WHERE id = ?", (device_id,)) as c: row = await c.fetchone() if not row: raise HTTPException(404, "No device bound to this room") return _device_row_to_model(row) @app.get("/api/devices/{device_id}", response_model=Device) async def get_device(device_id: str): async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM devices WHERE id = ?", (device_id,)) as c: row = await c.fetchone() if not row: raise HTTPException(404, "Device not found") return _device_row_to_model(row) class DeviceUpdate(BaseModel): name: Optional[str] = None character_id: Optional[str] = None motor_server_url: Optional[str] = None livekit_url: Optional[str] = None tailscale_ip: Optional[str] = None lan_ip: Optional[str] = None notes: Optional[str] = None status: Optional[str] = None @app.patch("/api/devices/{device_id}", response_model=Device) async def update_device(device_id: str, payload: DeviceUpdate): """Partial update for a device (character binding, addresses, notes, status).""" fields = {k: v for k, v in payload.model_dump(exclude_none=True).items()} if not fields: return await get_device(device_id) set_clause = ", ".join(f"{k} = ?" for k in fields) values = list(fields.values()) + [device_id] async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute(f"UPDATE devices SET {set_clause} WHERE id = ?", values) await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Device not found") await broadcast("device_updated", {"device_id": device_id, "fields": list(fields.keys())}) return await get_device(device_id) @app.post("/api/devices", response_model=Device) async def create_device(payload: DeviceCreate): """Manually register a device from the UI. Returns the device record and a one-time enrollment token (if one wasn't provided).""" if not payload.name or not payload.name.strip(): raise HTTPException(400, "name is required") device_id = f"dev_{secrets.token_hex(4)}" # Resolve enrollment token: use provided one, else generate a fresh one. if payload.enrollment_token: enrollment_token = payload.enrollment_token else: enrollment_token = secrets.token_urlsafe(24) enrollment_hash = _hash(enrollment_token) now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO devices (id, name, tailscale_ip, lan_ip, motor_server_url, character_id, livekit_url, enrollment_token_hash, status, enrolled_at, notes, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'enrolled', ?, ?, ?)""", ( device_id, payload.name.strip(), payload.tailscale_ip, payload.lan_ip, payload.motor_server_url, payload.character_id, payload.livekit_url, enrollment_hash, now, payload.notes, now, ), ) await db.commit() logger.info(f"Device {device_id} ({payload.name}) created") await broadcast("device_added", {"device_id": device_id, "name": payload.name}) device = await get_device(device_id) return { **device.model_dump(), "enrollment_token": enrollment_token, # returned ONCE, not stored plaintext } @app.delete("/api/devices/{device_id}") async def delete_device(device_id: str): async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT id FROM devices WHERE id = ?", (device_id,)) as c: row = await c.fetchone() if not row: raise HTTPException(404, "Device not found") await db.execute("DELETE FROM devices WHERE id = ?", (device_id,)) await db.commit() await broadcast("device_removed", {"device_id": device_id}) return {"status": "deleted", "device_id": device_id} @app.post("/api/devices/{device_id}/token/rotate") async def rotate_device_token(device_id: str): """Issue a new device token; the old one is invalidated.""" new_token = secrets.token_urlsafe(32) new_hash = _hash(new_token) async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute("UPDATE devices SET token_hash = ? WHERE id = ?", (new_hash, device_id)) await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Device not found") await broadcast("device_token_rotated", {"device_id": device_id}) return {"device_id": device_id, "device_token": new_token} async def _authorize_device(device_id: str, authorization: Optional[str]) -> bool: """Verify Bearer token matches stored hash for this device.""" if not authorization or not authorization.lower().startswith("bearer "): return False token = authorization.split(" ", 1)[1].strip() async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT token_hash FROM devices WHERE id = ?", (device_id,)) as c: row = await c.fetchone() if not row or not row[0]: return False return hmac.compare_digest(row[0], _hash(token)) async def _authorize_fleet(authorization: Optional[str]) -> bool: """Verify a Bearer token matches ANY enrolled device token (fleet auth). Used by endpoints that hand out LiveKit access but are keyed by robot_id (not device_id), so we can't scope to a single device row. Mirrors the hash + constant-time compare used by _authorize_device.""" if not authorization or not authorization.lower().startswith("bearer "): return False token_hash = _hash(authorization.split(" ", 1)[1].strip()) async with aiosqlite.connect(DB_PATH) as db: async with db.execute( "SELECT token_hash FROM devices WHERE token_hash IS NOT NULL" ) as c: rows = await c.fetchall() return any(r[0] and hmac.compare_digest(r[0], token_hash) for r in rows) @app.post("/api/devices/enroll", response_model=DeviceEnrollResponse) async def enroll_device(payload: DeviceEnrollRequest): """First-boot enrollment for a Pi. Requires production_mode = true and a valid enrollment token. Returns a long-lived device_token the Pi will use for heartbeats.""" if not await get_production_mode(): raise HTTPException(403, "Production mode is disabled. Enable it in Settings before enrolling devices.") if not payload.enrollment_token: raise HTTPException(400, "enrollment_token is required") token_hash = _hash(payload.enrollment_token) # Two paths: # 1. Token matches a pre-registered device row -> reuse that device, rotate its token. # 2. Token matches the GLOBAL default enrollment token -> mint a NEW device on the fly. async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT * FROM devices WHERE enrollment_token_hash = ?", (token_hash,) ) as c: existing = await c.fetchone() default_token_hash = await get_setting("default_enrollment_token") matches_default = default_token_hash and hmac.compare_digest(default_token_hash, token_hash) if not existing and not matches_default: raise HTTPException(401, "Invalid enrollment token") if existing: device_id = existing["id"] new_token = secrets.token_urlsafe(32) new_hash = _hash(new_token) now = datetime.utcnow().isoformat() await db.execute( """UPDATE devices SET token_hash = ?, status = 'enrolled', enrolled_at = COALESCE(enrolled_at, ?), tailscale_ip = COALESCE(?, tailscale_ip), lan_ip = COALESCE(?, lan_ip), motor_server_url = COALESCE(?, motor_server_url), livekit_url = COALESCE(?, livekit_url), character_id = COALESCE(?, character_id), name = COALESCE(?, name) WHERE id = ?""", ( new_hash, now, payload.tailscale_ip, payload.lan_ip, payload.motor_server_url, payload.livekit_url, payload.character_id, payload.name, device_id, ), ) else: # First-boot enroll via the global default token -> create a new device. device_id = f"dev_{secrets.token_hex(4)}" new_token = secrets.token_urlsafe(32) new_hash = _hash(new_token) now = datetime.utcnow().isoformat() await db.execute( """INSERT INTO devices (id, name, tailscale_ip, lan_ip, motor_server_url, character_id, livekit_url, token_hash, enrollment_token_hash, status, enrolled_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'enrolled', ?, ?)""", ( device_id, (payload.name or f"Pi-{device_id[-4:]}").strip(), payload.tailscale_ip, payload.lan_ip, payload.motor_server_url, payload.character_id, payload.livekit_url, new_hash, token_hash, now, now, ), ) await db.commit() await broadcast("device_enrolled", {"device_id": device_id}) logger.info(f"Device {device_id} enrolled") # Discover our own scheduler URL (best effort) scheduler_url = os.getenv("SCHEDULER_PUBLIC_URL", f"http://localhost:{os.getenv('SCHEDULER_PORT', '8080')}") return DeviceEnrollResponse(device_id=device_id, device_token=new_token, scheduler_url=scheduler_url) @app.post("/api/devices/{device_id}/heartbeat") async def device_heartbeat(device_id: str, payload: DeviceHeartbeat, authorization: Optional[str] = Header(default=None)): if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") async with aiosqlite.connect(DB_PATH) as db: async with db.execute( """UPDATE devices SET status = ?, last_heartbeat = ?, last_seen_ip = COALESCE(?, last_seen_ip) WHERE id = ?""", ( payload.status or "online", datetime.utcnow().isoformat(), payload.ip, device_id, ), ) as cur: await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Device not found") await broadcast("device_heartbeat", {"device_id": device_id, "status": payload.status}) # Tell the Pi whether production mode is on (so it can decide to join LiveKit or stand by) return { "status": "ok", "production_mode": await get_production_mode(), "server_time": datetime.utcnow().isoformat(), } @app.get("/api/devices/{device_id}/config") async def device_config(device_id: str, authorization: Optional[str] = Header(default=None)): """Pi polls this to learn current production_mode + assigned character, etc.""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT id, name, character_id, motor_server_url, livekit_url FROM devices WHERE id = ?", (device_id,), ) as c: row = await c.fetchone() if not row: raise HTTPException(404, "Device not found") return { "device_id": row["id"], "name": row["name"], "character_id": row["character_id"], "motor_server_url": row["motor_server_url"], "livekit_url": row["livekit_url"], "production_mode": await get_production_mode(), } @app.post("/api/devices/{device_id}/request-session") async def device_request_session(device_id: str, authorization: Optional[str] = Header(default=None)): """Enrolled DEVICE requests a session (presence-triggered LiveKit join). This is the device-keyed sibling of POST /api/robots/{robot_id}/request-session. The pi-client is an enrolled *device*, but sessions + assignment are keyed by *robot*. Rather than link the two fleet models with a new schema, we MIRROR the device onto a robot row whose id == device_id (additive + idempotent via INSERT OR IGNORE). Everything downstream (least-loaded server selection, session creation, telemetry) then reuses the existing robot machinery unchanged. Auth: the specific enrolled-device Bearer token (_authorize_device), so a device can only request a session for itself. Response mirrors the robot path plus is safe to feed straight into the Pi's join path: {session_id, server_id, server_url, room_name, token}. A session row is created with started_at so latency (joined_at - started_at) computes once POST /api/sessions/{session_id}/joined is called. NOTE: unlike the robot path, this does NOT hard-fail when the mirrored robot is not 'idle'. Presence can re-trigger while a prior session's end-session was never delivered; blocking would strand the device. Each call creates a fresh session and bumps trigger_count (the natural camera-trigger point).""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid or missing device token") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row # Resolve the enrolled device. async with db.execute("SELECT * FROM devices WHERE id = ?", (device_id,)) as c: device = await c.fetchone() if not device: raise HTTPException(404, "Device not found") # Mirror device -> robot identity (id == device_id). Additive + idempotent: # if the robot row already exists it is left untouched. await db.execute( "INSERT OR IGNORE INTO robots (id, name, character_id) VALUES (?, ?, ?)", (device_id, device["name"] or device_id, device["character_id"]), ) # Find least-loaded online server (identical selection to the robot path). async with db.execute(""" SELECT s.*, (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) as active FROM livekit_servers s WHERE s.status = 'online' ORDER BY active ASC LIMIT 1 """) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(503, "No available servers") if server["active"] >= server["max_sessions"]: raise HTTPException(503, "All servers at capacity") ts = int(datetime.utcnow().timestamp()) session_id = f"session_{device_id}_{ts}" room_name = f"robopark_{device_id}_{ts}" await db.execute(""" INSERT INTO sessions (id, robot_id, server_id, room_name, started_at) VALUES (?, ?, ?, ?, ?) """, (session_id, device_id, server["id"], room_name, datetime.utcnow().isoformat())) # Mark the mirrored robot connecting + count the camera trigger, exactly # as the robot path does (request-session is the natural trigger point). await db.execute(""" UPDATE robots SET status = 'connecting', current_session_id = ?, connected_server_id = ?, trigger_count = COALESCE(trigger_count, 0) + 1 WHERE id = ? """, (session_id, server["id"], device_id)) await db.commit() await broadcast("session_requested", { "robot_id": device_id, "device_id": device_id, "server_id": server["id"], "session_id": session_id, }) # Mint a short-lived room-scoped JWT so the device joins the assigned room # without ever receiving the raw api_key/api_secret. Identity keeps the # existing pi: convention so the agent side sees the same participant. try: from livekit.api import AccessToken, VideoGrants except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") lk_key = server["api_key"] or "devkey" lk_secret = server["api_secret"] or "secret" identity = f"pi:{device_id}" grants = VideoGrants( room=room_name, room_join=True, can_publish=True, can_subscribe=True, can_publish_data=True, ) token = ( AccessToken(lk_key, lk_secret) .with_identity(identity) .with_name(device["name"] or device_id) .with_ttl(timedelta(hours=1)) .with_grants(grants) .to_jwt() ) return { "session_id": session_id, "server_id": server["id"], "server_url": server["url"], "room_name": room_name, "token": token, } # ============================================================================= # CHARACTER ENDPOINTS (read from ROBOVOICE settings.json) # ============================================================================= import json as _json class CharacterSummary(BaseModel): id: str name: Optional[str] = None description: Optional[str] = None tts_voice: Optional[str] = None motors: List[str] = [] class Character(CharacterSummary): system_prompt: Optional[str] = None raw: dict = {} def _load_robovoice_settings() -> dict: """Read the ROBOVOICE settings file. Returns {} if missing/invalid.""" try: if not os.path.exists(ROBOVOICE_SETTINGS_PATH): return {} with open(ROBOVOICE_SETTINGS_PATH, "r", encoding="utf-8") as f: return _json.load(f) except Exception as e: log.warning(f"Could not read ROBOVOICE settings at {ROBOVOICE_SETTINGS_PATH}: {e}") return {} def _character_summary(c: dict) -> CharacterSummary: motors = c.get("motors") or [] if isinstance(motors, list) and motors and isinstance(motors[0], dict): motor_names = [m.get("name") for m in motors if m.get("name")] else: motor_names = [m for m in motors if isinstance(m, str)] voice = ( c.get("tts_voice") or c.get("elevenlabs_voice_id") or c.get("tts_voice_kokoro") or c.get("tts_voice_piper") ) return CharacterSummary( id=c.get("id") or c.get("name") or "unknown", name=c.get("name"), description=c.get("description"), tts_voice=voice, motors=motor_names, ) @app.get("/api/characters", response_model=List[CharacterSummary]) async def list_characters(): """List all characters available in the bound ROBOVOICE settings file.""" s = _load_robovoice_settings() chars = s.get("characters") or [] return [_character_summary(c) for c in chars] @app.get("/api/characters/{character_id}", response_model=Character) async def get_character(character_id: str): """Get a single character (incl. system_prompt and motor config) by id.""" s = _load_robovoice_settings() for c in s.get("characters") or []: if c.get("id") == character_id: summary = _character_summary(c) return Character( **summary.model_dump(), system_prompt=c.get("system_prompt"), raw=c, ) raise HTTPException(404, f"Character '{character_id}' not found in {ROBOVOICE_SETTINGS_PATH}") # ============================================================================= # TELEMETRY (close-the-loop: trigger counts, latency, drop rate, error reasons) # ============================================================================= async def _robot_telemetry(db, robot_id: str) -> Optional[dict]: """Per-robot telemetry. `db` must have row_factory = aiosqlite.Row set. Returns trigger_count, sessions_total, avg_latency_ms, drop_rate, drops and an end_reason breakdown; None if the robot does not exist.""" async with db.execute( "SELECT trigger_count FROM robots WHERE id = ?", (robot_id,) ) as c: robot = await c.fetchone() if not robot: return None trigger_count = robot["trigger_count"] or 0 # Average latency over sessions that recorded a room join. async with db.execute( "SELECT started_at, joined_at FROM sessions WHERE robot_id = ? AND joined_at IS NOT NULL", (robot_id,), ) as c: latency_rows = await c.fetchall() latencies = [ ms for ms in ( _session_latency_ms(r["started_at"], r["joined_at"]) for r in latency_rows ) if ms is not None ] avg_latency_ms = int(sum(latencies) / len(latencies)) if latencies else None async with db.execute( "SELECT COUNT(*) AS n FROM sessions WHERE robot_id = ?", (robot_id,) ) as c: sessions_total = (await c.fetchone())["n"] # End-reason breakdown (surfaces "per-robot error reasons"); drops = the # subset whose reason is abnormal. async with db.execute( """SELECT end_reason, COUNT(*) AS n FROM sessions WHERE robot_id = ? AND end_reason IS NOT NULL GROUP BY end_reason""", (robot_id,), ) as c: end_reasons = {r["end_reason"]: r["n"] for r in await c.fetchall()} drops = sum(n for reason, n in end_reasons.items() if reason in ABNORMAL_END_REASONS) drop_rate = round(drops / sessions_total, 4) if sessions_total else 0.0 return { "robot_id": robot_id, "trigger_count": trigger_count, "sessions_total": sessions_total, "avg_latency_ms": avg_latency_ms, "drop_rate": drop_rate, "drops": drops, "end_reasons": end_reasons, } @app.get("/api/robots/{robot_id}/telemetry") async def robot_telemetry(robot_id: str): """Per-robot telemetry: camera-trigger count, average session latency (ms), drop rate, total sessions, and an end-reason breakdown.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row data = await _robot_telemetry(db, robot_id) if data is None: raise HTTPException(404, "Robot not found") return data @app.get("/api/telemetry") async def telemetry(): """Fleet-wide telemetry: per-robot trigger counts, latency, drop rates and session totals, plus fleet totals and the latest GPU metrics recorded per LiveKit server (surfaced from the previously write-only metrics_history).""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT id FROM robots ORDER BY name") as c: robot_ids = [r["id"] for r in await c.fetchall()] robots = [] for rid in robot_ids: t = await _robot_telemetry(db, rid) if t is not None: robots.append(t) # Surface metrics_history (written by metrics_collector, never read # before): the latest sample per server. async with db.execute( """SELECT m.server_id, m.gpu_utilization, m.vram_used_mb, m.active_sessions, m.recorded_at FROM metrics_history m JOIN ( SELECT server_id, MAX(recorded_at) AS latest FROM metrics_history GROUP BY server_id ) latest ON m.server_id = latest.server_id AND m.recorded_at = latest.latest""" ) as c: server_metrics = [dict(r) for r in await c.fetchall()] totals = { "trigger_count": sum(r["trigger_count"] for r in robots), "sessions_total": sum(r["sessions_total"] for r in robots), "drops": sum(r["drops"] for r in robots), } totals["drop_rate"] = ( round(totals["drops"] / totals["sessions_total"], 4) if totals["sessions_total"] else 0.0 ) return { "robots": robots, "servers": server_metrics, "totals": totals, "abnormal_end_reasons": sorted(ABNORMAL_END_REASONS), } # ============================================================================= # DASHBOARD (Embedded) # ============================================================================= # ── Back-compat: legacy /voice page (ported verbatim from the deployed OLD # scheduler). Serves voice_client.html if present next to main.py, else an # inline fallback — matches the deployed container exactly (its Dockerfile # copies only main.py, so the fallback is what actually renders). ── @app.get("/voice", response_class=HTMLResponse) async def voice_client(): """Serve voice client page for remote access""" import os html_path = os.path.join(os.path.dirname(__file__), "voice_client.html") if os.path.exists(html_path): with open(html_path, "r") as f: return f.read() return """ Voice Client

Voice client HTML not found

""" @app.get("/", response_class=HTMLResponse) async def dashboard(): """Serve embedded dashboard""" return """ RoboPark Control Center
""" if __name__ == "__main__": import uvicorn host = os.getenv("SCHEDULER_HOST", "0.0.0.0") port = int(os.getenv("SCHEDULER_PORT", "8080")) uvicorn.run(app, host=host, port=port)