AskaEth
b00e59d5bf
- Settings: UI test mode toggle generates simulated telemetry data (speed/RPM/gear/laps...) - Simulated data broadcasts via WebSocket and forwards to downstream devices - game_ports config: per-game independent port overrides (shown when unified port off) - Settings UI: game port list with save button, hidden when unified port enabled - Sidebar: switch game reads game_ports[gameId] fallback to default_port
388 lines
12 KiB
Python
388 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException, Request, UploadFile, File
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from config.settings import get_config, update_config
|
|
from models.dashboard import dashboard_manager, DashboardTheme
|
|
from models.scene import scene_manager, Scene, SceneCanvas, DashboardPlacement
|
|
from server.telemetry.listener import telemetry_listener
|
|
from server.websocket import ws_manager
|
|
from utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
router = APIRouter(prefix="/api")
|
|
|
|
|
|
# ---- Status ----
|
|
@router.get("/status")
|
|
async def get_status():
|
|
latest = telemetry_listener.latest_data
|
|
cfg = get_config()
|
|
return {
|
|
"server_running": True,
|
|
"telemetry_running": telemetry_listener.is_running,
|
|
"ws_clients": ws_manager.client_count,
|
|
"packet_count": telemetry_listener.packet_count,
|
|
"last_packet_time": telemetry_listener.last_packet_time,
|
|
"selected_game_id": cfg.get("selected_game_id"),
|
|
"forward_active": telemetry_listener.forwarder.active,
|
|
"latest_data": latest.to_dict() if latest else None,
|
|
}
|
|
|
|
|
|
# ---- Config ----
|
|
@router.get("/config")
|
|
async def api_get_config():
|
|
return get_config()
|
|
|
|
|
|
@router.put("/config")
|
|
async def api_update_config(data: dict[str, Any]):
|
|
cfg = get_config()
|
|
for k, v in data.items():
|
|
cfg[k] = v
|
|
from config.settings import save_config
|
|
save_config(cfg)
|
|
|
|
if "selected_game_id" in data:
|
|
telemetry_listener.set_parser_for_game(data["selected_game_id"])
|
|
|
|
if "theme" in data:
|
|
cfg["theme"] = data["theme"]
|
|
|
|
return {"ok": True}
|
|
|
|
|
|
# ---- Telemetry ----
|
|
@router.post("/telemetry/start")
|
|
async def api_start_telemetry():
|
|
ok = await telemetry_listener.start()
|
|
return {"ok": ok, "running": telemetry_listener.is_running}
|
|
|
|
|
|
@router.post("/telemetry/stop")
|
|
async def api_stop_telemetry():
|
|
telemetry_listener.stop()
|
|
return {"ok": True, "running": telemetry_listener.is_running}
|
|
|
|
|
|
@router.post("/test-mode/start")
|
|
async def api_start_test_mode():
|
|
telemetry_listener.start_test_mode()
|
|
return {"ok": True, "testing": True}
|
|
|
|
|
|
@router.post("/test-mode/stop")
|
|
async def api_stop_test_mode():
|
|
telemetry_listener.stop_test_mode()
|
|
return {"ok": True, "testing": False}
|
|
|
|
|
|
@router.get("/telemetry/latest")
|
|
async def api_latest_telemetry():
|
|
latest = telemetry_listener.latest_data
|
|
if latest:
|
|
return {
|
|
"data": latest.to_dict(),
|
|
"raw": latest.raw,
|
|
}
|
|
return {"data": None, "raw": None}
|
|
|
|
|
|
# ---- Dashboards ----
|
|
@router.get("/dashboards")
|
|
async def api_list_dashboards(category: str = "all"):
|
|
return dashboard_manager.list_all(category)
|
|
|
|
|
|
@router.get("/dashboards/categories")
|
|
async def api_dashboard_categories():
|
|
return dashboard_manager.get_categories()
|
|
|
|
|
|
@router.get("/dashboards/{theme_id}")
|
|
async def api_get_dashboard(theme_id: str):
|
|
theme = dashboard_manager.get(theme_id)
|
|
if not theme:
|
|
raise HTTPException(404, "Dashboard not found")
|
|
return theme
|
|
|
|
|
|
@router.get("/dashboards/{theme_id}/template")
|
|
async def api_get_dashboard_template(theme_id: str):
|
|
tmpl = dashboard_manager.get_index_html(theme_id)
|
|
return {"html": tmpl or ""}
|
|
|
|
|
|
@router.post("/dashboards")
|
|
async def api_create_dashboard(data: dict[str, Any]):
|
|
cfg = data.get("config", {})
|
|
theme = DashboardTheme(
|
|
name=data.get("name", "Untitled"),
|
|
category=data.get("category", "basic"),
|
|
description=data.get("description", ""),
|
|
author=data.get("author", ""),
|
|
icon=cfg.get("icon", "📊"),
|
|
aspect_ratio=cfg.get("aspect_ratio", "auto"),
|
|
render_mode=cfg.get("render_mode", "contain"),
|
|
)
|
|
dashboard_manager.save(theme, data.get("index_html", ""))
|
|
return theme.to_dict()
|
|
|
|
|
|
@router.put("/dashboards/{theme_id}")
|
|
async def api_update_dashboard(theme_id: str, data: dict[str, Any]):
|
|
existing = dashboard_manager.get(theme_id)
|
|
if not existing:
|
|
raise HTTPException(404, "Dashboard not found")
|
|
theme = DashboardTheme(
|
|
id=theme_id,
|
|
name=data.get("name", existing.get("name", "")),
|
|
category=data.get("category", existing.get("category", "basic")),
|
|
description=data.get("description", existing.get("description", "")),
|
|
author=data.get("author", existing.get("author", "")),
|
|
icon=data.get("icon", existing.get("icon", "📊")),
|
|
aspect_ratio=data.get("aspect_ratio", existing.get("aspect_ratio", "auto")),
|
|
render_mode=data.get("render_mode", existing.get("render_mode", "contain")),
|
|
)
|
|
dashboard_manager.save(theme, data.get("index_html", ""))
|
|
return theme.to_dict()
|
|
|
|
|
|
@router.delete("/dashboards/{theme_id}")
|
|
async def api_delete_dashboard(theme_id: str):
|
|
ok = dashboard_manager.delete(theme_id)
|
|
return {"ok": ok}
|
|
|
|
|
|
@router.get("/dashboards/{theme_id}/export")
|
|
async def api_export_dashboard(theme_id: str):
|
|
from fastapi.responses import Response
|
|
zip_bytes = dashboard_manager.export_theme_zip(theme_id)
|
|
if not zip_bytes:
|
|
raise HTTPException(404, "Dashboard not found")
|
|
theme = dashboard_manager.get(theme_id)
|
|
filename = f"{theme['id']}.tsd" if theme else "dashboard.tsd"
|
|
return Response(
|
|
content=zip_bytes,
|
|
media_type="application/zip",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
@router.post("/dashboards/import")
|
|
async def api_import_dashboard(file: UploadFile = File(...)):
|
|
if not file.filename or not file.filename.endswith('.tsd'):
|
|
raise HTTPException(400, "Only .tsd files are accepted")
|
|
zip_data = await file.read()
|
|
ok = dashboard_manager.import_theme_zip(zip_data)
|
|
return {"ok": ok}
|
|
|
|
|
|
# ---- Scenes ----
|
|
@router.get("/scenes")
|
|
async def api_list_scenes(game_id: str = ""):
|
|
return scene_manager.list_all(game_id or None)
|
|
|
|
|
|
@router.get("/scenes/{scene_id}")
|
|
async def api_get_scene(scene_id: str):
|
|
scene = scene_manager.get(scene_id)
|
|
if not scene:
|
|
raise HTTPException(404, "Scene not found")
|
|
return scene
|
|
|
|
|
|
@router.post("/scenes")
|
|
async def api_create_scene(data: dict[str, Any]):
|
|
scene = Scene(
|
|
name=data.get("name", "New Scene"),
|
|
game_id=data.get("game_id", ""),
|
|
description=data.get("description", ""),
|
|
canvases=[SceneCanvas(label="16:9", width=1920, height=1080)],
|
|
)
|
|
if data.get("canvases"):
|
|
scene.canvases = [
|
|
SceneCanvas(
|
|
width=c.get("width", 1920),
|
|
height=c.get("height", 1080),
|
|
label=c.get("label", ""),
|
|
placements=[DashboardPlacement(**p) for p in c.get("placements", [])],
|
|
)
|
|
for c in data["canvases"]
|
|
]
|
|
scene_manager.save(scene)
|
|
return scene.to_dict()
|
|
|
|
|
|
@router.put("/scenes/{scene_id}")
|
|
async def api_update_scene(scene_id: str, data: dict[str, Any]):
|
|
existing = scene_manager.get(scene_id)
|
|
if not existing:
|
|
raise HTTPException(404, "Scene not found")
|
|
scene = Scene.from_dict(existing)
|
|
for k in ["name", "game_id", "description"]:
|
|
if k in data:
|
|
setattr(scene, k, data[k])
|
|
if "canvases" in data:
|
|
scene.canvases = [
|
|
SceneCanvas(
|
|
width=c.get("width", 1920),
|
|
height=c.get("height", 1080),
|
|
label=c.get("label", "Custom"),
|
|
placements=[DashboardPlacement(**p) for p in c.get("placements", [])],
|
|
)
|
|
for c in data["canvases"]
|
|
]
|
|
scene_manager.save(scene)
|
|
return scene.to_dict()
|
|
|
|
|
|
@router.delete("/scenes/{scene_id}")
|
|
async def api_delete_scene(scene_id: str):
|
|
ok = scene_manager.delete(scene_id)
|
|
return {"ok": ok}
|
|
|
|
|
|
@router.get("/scenes/{scene_id}/export")
|
|
async def api_export_scene(scene_id: str):
|
|
from fastapi.responses import Response
|
|
zip_bytes = scene_manager.export_scene_zip(scene_id)
|
|
if not zip_bytes:
|
|
raise HTTPException(404, "Scene not found")
|
|
scene = scene_manager.get(scene_id)
|
|
filename = f"{scene['id']}.tss" if scene else "scene.tss"
|
|
return Response(
|
|
content=zip_bytes,
|
|
media_type="application/zip",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
@router.post("/scenes/import")
|
|
async def api_import_scene(file: UploadFile = File(...)):
|
|
if not file.filename or not file.filename.endswith('.tss'):
|
|
raise HTTPException(400, "Only .tss files are accepted")
|
|
zip_data = await file.read()
|
|
ok = scene_manager.import_scene_zip(zip_data)
|
|
return {"ok": ok}
|
|
|
|
|
|
# ---- Game Plugins ----
|
|
from server.game_manager import game_plugin_manager
|
|
|
|
|
|
@router.get("/games")
|
|
async def api_list_games():
|
|
return game_plugin_manager.list_all()
|
|
|
|
|
|
@router.get("/games/{plugin_id}")
|
|
async def api_get_game(plugin_id: str):
|
|
gp = game_plugin_manager.get(plugin_id)
|
|
if not gp:
|
|
raise HTTPException(404, "Game plugin not found")
|
|
return gp.to_dict()
|
|
|
|
|
|
@router.post("/games/install")
|
|
async def api_install_game_plugin(file: UploadFile = File(...)):
|
|
if not file.filename or not file.filename.endswith('.tsp'):
|
|
raise HTTPException(400, "Only .tsp files are accepted")
|
|
zip_data = await file.read()
|
|
ok = game_plugin_manager.install_plugin_zip(zip_data)
|
|
return {"ok": ok}
|
|
|
|
|
|
@router.delete("/games/{plugin_id}")
|
|
async def api_remove_game_plugin(plugin_id: str):
|
|
ok = game_plugin_manager.remove_plugin(plugin_id)
|
|
return {"ok": ok}
|
|
|
|
|
|
@router.get("/games/{plugin_id}/export")
|
|
async def api_export_game_plugin(plugin_id: str):
|
|
from fastapi.responses import Response
|
|
zip_bytes = game_plugin_manager.export_plugin_zip(plugin_id)
|
|
if not zip_bytes:
|
|
raise HTTPException(404, "Game plugin not found")
|
|
return Response(
|
|
content=zip_bytes,
|
|
media_type="application/zip",
|
|
headers={"Content-Disposition": f'attachment; filename="{plugin_id}.tsp"'},
|
|
)
|
|
|
|
|
|
@router.post("/games/reload")
|
|
async def api_reload_game_plugins():
|
|
game_plugin_manager.reload()
|
|
return {"ok": True, "count": len(game_plugin_manager.list_all())}
|
|
|
|
|
|
# ---- Logs ----
|
|
@router.get("/logs")
|
|
async def api_get_logs(lines: int = 50):
|
|
from pathlib import Path
|
|
log_file = Path(__file__).resolve().parent.parent / "logs" / "turbosu.log"
|
|
if not log_file.exists():
|
|
return {"lines": []}
|
|
try:
|
|
with open(log_file, "r", encoding="utf-8", errors="replace") as f:
|
|
all_lines = f.readlines()
|
|
return {"lines": [l.rstrip() for l in all_lines[-lines:]]}
|
|
except Exception:
|
|
return {"lines": []}
|
|
|
|
|
|
# ---- Full Backup (.tsb) ----
|
|
@router.get("/backup/export")
|
|
async def api_export_backup():
|
|
import io, zipfile
|
|
from fastapi.responses import Response
|
|
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
|
|
base = Path(__file__).resolve().parent.parent
|
|
data_dir = base / "data"
|
|
if data_dir.exists():
|
|
for f in data_dir.rglob("*"):
|
|
if f.is_file() and "__pycache__" not in str(f):
|
|
zf.write(f, f.relative_to(base))
|
|
dash_dir = base / "dashboards"
|
|
if dash_dir.exists():
|
|
for f in dash_dir.rglob("*"):
|
|
if f.is_file():
|
|
zf.write(f, f.relative_to(base))
|
|
games_dir = base / "games" / "user"
|
|
if games_dir.exists():
|
|
for f in games_dir.rglob("*"):
|
|
if f.is_file():
|
|
zf.write(f, f.relative_to(base))
|
|
|
|
return Response(content=buf.getvalue(), media_type="application/zip",
|
|
headers={"Content-Disposition": "attachment; filename=turbosu_backup.tsb"})
|
|
|
|
|
|
# ---- Data Forwarding ----
|
|
@router.get("/forward")
|
|
async def api_list_forward_targets():
|
|
cfg = get_config()
|
|
return cfg.get("forward_targets", [])
|
|
|
|
|
|
@router.put("/forward")
|
|
async def api_update_forward_targets(data: dict[str, Any]):
|
|
targets = data.get("targets", [])
|
|
cfg = get_config()
|
|
cfg["forward_targets"] = targets
|
|
from config.settings import save_config
|
|
save_config(cfg)
|
|
telemetry_listener.forwarder.reload_targets()
|
|
return {"ok": True, "active": telemetry_listener.forwarder.active}
|