Jimmy
5094c7bcee
Bring the five-panel hex portal online with a browser 3D/schematic preview, Pi SPI backends, and renamed multi-panel Pico UDP firmware. Co-authored-by: Cursor <cursoragent@cursor.com>
146 lines
4.3 KiB
Python
146 lines
4.3 KiB
Python
"""FastAPI app for the portal web simulator."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException, Query
|
|
from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from leds.animations import DEFAULT_FPS
|
|
from leds.portal_sim import portal_config, render_portal_frame
|
|
|
|
WEB_DIR = Path(__file__).resolve().parent.parent / "web"
|
|
|
|
_reload_queues: list[asyncio.Queue[str]] = []
|
|
|
|
|
|
def _dev_reload_enabled() -> bool:
|
|
return os.environ.get("PORTAL_DEV_RELOAD", "0") == "1"
|
|
|
|
|
|
async def _notify_reload_clients() -> None:
|
|
for queue in list(_reload_queues):
|
|
await queue.put("reload")
|
|
|
|
|
|
async def _watch_web_files() -> None:
|
|
from watchfiles import awatch
|
|
|
|
async for _changes in awatch(WEB_DIR):
|
|
await _notify_reload_clients()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _lifespan(_app: FastAPI):
|
|
watcher: asyncio.Task | None = None
|
|
if _dev_reload_enabled():
|
|
watcher = asyncio.create_task(_watch_web_files())
|
|
yield
|
|
if watcher is not None:
|
|
watcher.cancel()
|
|
try:
|
|
await watcher
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
app = FastAPI(title="Portal Simulator", lifespan=_lifespan)
|
|
|
|
@app.get("/")
|
|
async def index() -> HTMLResponse:
|
|
html = (WEB_DIR / "index.html").read_text(encoding="utf-8")
|
|
if _dev_reload_enabled():
|
|
snippet = '<script type="module" src="/static/js/dev-reload.js"></script>'
|
|
html = html.replace("</body>", f" {snippet}\n </body>")
|
|
return HTMLResponse(html)
|
|
|
|
@app.get("/api/config")
|
|
async def api_config() -> dict:
|
|
return portal_config()
|
|
|
|
@app.get("/api/frame")
|
|
async def api_frame(
|
|
animation: str = Query("rainbow"),
|
|
frame: int = Query(0),
|
|
brightness: float = Query(0.35),
|
|
) -> dict:
|
|
try:
|
|
return await asyncio.to_thread(
|
|
render_portal_frame, animation, frame, brightness=brightness
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
@app.get("/api/stream")
|
|
async def api_stream(
|
|
animation: str = Query("rainbow"),
|
|
brightness: float = Query(0.35),
|
|
fps: float = Query(None),
|
|
) -> StreamingResponse:
|
|
rate = fps if fps is not None else DEFAULT_FPS.get(animation, 25)
|
|
rate = max(1.0, min(rate, 60.0))
|
|
delay = 1.0 / rate
|
|
|
|
async def generate() -> object:
|
|
frame = 0
|
|
while True:
|
|
payload = await asyncio.to_thread(
|
|
render_portal_frame,
|
|
animation,
|
|
frame,
|
|
brightness=brightness,
|
|
)
|
|
yield f"data: {json.dumps(payload)}\n\n"
|
|
frame += 1
|
|
await asyncio.sleep(delay)
|
|
|
|
return StreamingResponse(
|
|
generate(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
|
|
)
|
|
|
|
@app.get("/api/dev/reload")
|
|
async def api_dev_reload() -> StreamingResponse:
|
|
if not _dev_reload_enabled():
|
|
raise HTTPException(status_code=404)
|
|
|
|
queue: asyncio.Queue[str] = asyncio.Queue()
|
|
_reload_queues.append(queue)
|
|
|
|
async def generate() -> object:
|
|
try:
|
|
yield "data: connected\n\n"
|
|
while True:
|
|
message = await queue.get()
|
|
yield f"data: {message}\n\n"
|
|
finally:
|
|
if queue in _reload_queues:
|
|
_reload_queues.remove(queue)
|
|
|
|
return StreamingResponse(
|
|
generate(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
|
|
)
|
|
|
|
@app.get("/favicon.ico", include_in_schema=False)
|
|
async def favicon() -> FileResponse:
|
|
path = WEB_DIR / "favicon.ico"
|
|
if path.is_file():
|
|
return FileResponse(path)
|
|
raise HTTPException(status_code=404)
|
|
|
|
app.mount("/static", StaticFiles(directory=WEB_DIR), name="static")
|
|
return app
|
|
|
|
|
|
app = create_app()
|