Spaces:
Running
Running
File size: 3,717 Bytes
d65a9ae | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """nova-connect: the web page and the room-ticket desk for a Nova Pi.
Runs as a Hugging Face Docker Space (port 7860) or any container host. It serves
the static UI and one endpoint: POST /api/session with the passcode returns a
LiveKit URL and a 10-minute token. It holds the LiveKit API secret; the browser
never sees it. Audio never passes through here -- browser and Pi meet at the SFU.
Configuration (Space secrets / environment):
LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET -- the SFU and its key pair
NOVA_AGENT_NAME -- the Pi's agent name, printed by the Pi at startup
("LiveKit: registered as nova-<hostname>")
NOVA_PASSCODE_HASH -- from scripts/make_passcode.py
"""
from __future__ import annotations
import logging
import os
from dataclasses import dataclass
from pathlib import Path
from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import token_service as ts
STATIC = Path(__file__).resolve().parent / "static"
logger = logging.getLogger("nova-connect")
@dataclass
class Settings:
livekit_url: str
api_key: str
api_secret: str
agent_name: str
passcode_hash: str
@classmethod
def from_env(cls) -> Settings:
e = os.environ.get
return cls(
livekit_url=e("LIVEKIT_URL", ""),
api_key=e("LIVEKIT_API_KEY", ""),
api_secret=e("LIVEKIT_API_SECRET", ""),
agent_name=e("NOVA_AGENT_NAME", ""),
passcode_hash=e("NOVA_PASSCODE_HASH", ""),
)
@property
def configured(self) -> bool:
return all(
(self.livekit_url, self.api_key, self.api_secret, self.agent_name, self.passcode_hash)
)
class SessionRequest(BaseModel):
passcode: str
def _client_ip(request: Request) -> str:
# Behind the HF / reverse proxy the socket peer is the proxy; the first
# X-Forwarded-For hop is the caller.
fwd = request.headers.get("x-forwarded-for", "")
if fwd:
return fwd.split(",")[0].strip()
return request.client.host if request.client else "unknown"
def create_app(settings: Settings | None = None, *, limiter: ts.RateLimiter | None = None) -> FastAPI:
settings = settings or Settings.from_env()
limiter = limiter or ts.RateLimiter()
app = FastAPI(title="nova-connect", docs_url=None, redoc_url=None)
@app.get("/api/health")
def health():
return {"configured": settings.configured, "agent": settings.agent_name or None}
@app.post("/api/session")
def session(req: SessionRequest, request: Request):
if not settings.configured:
return JSONResponse({"error": "not_configured"}, status_code=503)
if not limiter.allow(_client_ip(request)):
return JSONResponse({"error": "rate_limited"}, status_code=429)
if not ts.verify_passcode(req.passcode, settings.passcode_hash):
return JSONResponse({"error": "bad_passcode"}, status_code=401)
s = ts.mint_session(
api_key=settings.api_key, api_secret=settings.api_secret, agent_name=settings.agent_name
)
logger.info("session issued room=%s", s.room)
return {"url": settings.livekit_url, "token": s.token, "room": s.room, "expires_in": s.expires_in}
@app.get("/")
def index():
return FileResponse(STATIC / "index.html")
app.mount("/static", StaticFiles(directory=STATIC), name="static")
return app
if __name__ == "__main__":
import uvicorn
logging.basicConfig(level=logging.INFO)
uvicorn.run(create_app(), host="0.0.0.0", port=int(os.environ.get("PORT", "7860")))
|