from __future__ import annotations import json import os from contextlib import asynccontextmanager from pathlib import Path from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Request from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.responses import HTMLResponse, FileResponse from config.settings import get_config from server.api import router as api_router from server.websocket import ws_manager from server.telemetry.listener import telemetry_listener from utils.logger import get_logger logger = get_logger(__name__) BASE_DIR = Path(__file__).resolve().parent STATIC_DIR = BASE_DIR / "static" TEMPLATES_DIR = BASE_DIR / "templates" os.makedirs(BASE_DIR / "logs", exist_ok=True) @asynccontextmanager async def lifespan(app: FastAPI): import asyncio import warnings from websockets.exceptions import ConnectionClosedError def _handle_async_exception(loop, ctx): exc = ctx.get("exception") if isinstance(exc, ConnectionClosedError): return loop.default_exception_handler(ctx) asyncio.get_event_loop().set_exception_handler(_handle_async_exception) warnings.filterwarnings("ignore", message=".*ping timeout.*") logger.info("=" * 50) logger.info(" TurboSu - Racing Telemetry Dashboard") logger.info("=" * 50) cfg = get_config() logger.info("Server config: %s:%d", cfg.get("server_host", "0.0.0.0"), cfg.get("server_port", 9527)) await telemetry_listener.start() logger.info("Telemetry listener auto-started") if cfg.get("ui_test_mode"): telemetry_listener.start_test_mode() logger.info("Test mode auto-started (config)") yield telemetry_listener.stop() logger.info("TurboSu shutdown complete") app = FastAPI(title="TurboSu", version="1.0.0", lifespan=lifespan) app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") templates = Jinja2Templates(directory=str(TEMPLATES_DIR)) app.include_router(api_router) @app.get("/", response_class=HTMLResponse) async def index(request: Request): return templates.TemplateResponse("index.html", {"request": request}) @app.get("/dashboard/{theme_id}", response_class=HTMLResponse) async def dashboard_render(request: Request, theme_id: str): from models.dashboard import dashboard_manager index_html = dashboard_manager.get_index_html(theme_id) if index_html: return HTMLResponse(content=index_html) theme = dashboard_manager.get(theme_id) if not theme: raise HTTPException(status_code=404, detail="Dashboard not found") return HTMLResponse(content=f""" 仪表盘 "{theme['name']}" 缺少 index.html """) @app.get("/scene/{scene_id}", response_class=HTMLResponse) async def scene_render(request: Request, scene_id: str): from models.scene import scene_manager scene = scene_manager.get(scene_id) return templates.TemplateResponse("scene.html", { "request": request, "scene": scene, }) @app.websocket("/ws") async def websocket_endpoint(ws: WebSocket): cid = await ws_manager.connect(ws) try: cfg = get_config() await ws.send_json({ "type": "connected", "client_id": cid, "selected_game_id": cfg.get("selected_game_id"), }) while True: msg = await ws.receive_json() if msg.get("type") == "ping": await ws.send_json({"type": "pong"}) except WebSocketDisconnect: pass except Exception: pass finally: ws_manager.disconnect(cid) @app.get("/health") async def health(): return {"status": "ok"} if __name__ == "__main__": import uvicorn cfg = get_config() uvicorn.run( "app:app", host=cfg.get("server_host", "0.0.0.0"), port=cfg.get("server_port", 9527), reload=False, log_level="info", )