LED 大屏投放控制台
+正在连接控制服务...
+commit 8590fafac188e9676af355f0f82208663897faaa Author: TJY <1585382647@qq.com> Date: Thu Jul 16 08:31:52 2026 +0800 Init diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8740cd8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.venv/ +__pycache__/ +*.pyc +.pytest_cache/ +.ruff_cache/ +dist/ +build/ +*.egg-info/ +logs/ +runtime/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..94094ee --- /dev/null +++ b/README.md @@ -0,0 +1,98 @@ +# LED 大屏投放平台 + +面向 `14880 x 3510` LED 大屏的双 GPU 服务器投放平台。 + +当前架构: + +```text +控制服务 + -> WebSocket 同步协议 + -> 左 GPU 服务器本地渲染 left tile + -> 右 GPU 服务器本地渲染 right tile + -> 每台服务器 4 路 4K 输出 + -> 拼控 / LED 控制器物理拼接 +``` + +WebSocket 只传场景、动作、时间轴、ACK 和性能指标,不传超大视频流。画面由两台 GPU 服务器各自在本地渲染。 + +## 启动 + +第一次启动: + +```bash +python -m venv .venv +python -m pip install -r requirements.txt +python -m led_platform.cli serve +``` + +已有环境后: + +```bash +python -m led_platform.cli serve +``` + +或者直接用 uvicorn: + +```bash +uvicorn led_platform.main:app --host 127.0.0.1 --port 8000 +``` + +查看地址: + +```bash +python -m led_platform.cli urls +``` + +常用地址: + +- 控制台:http://127.0.0.1:8000/ +- 左输出:http://127.0.0.1:8000/output/left +- 右输出:http://127.0.0.1:8000/output/right +- 健康检查:http://127.0.0.1:8000/healthz +- 同步状态:http://127.0.0.1:8000/api/sync/status + +生产输出端直接用浏览器打开: + +```text +左 GPU 服务器:http://控制服务IP:8000/output/left +右 GPU 服务器:http://控制服务IP:8000/output/right +``` + +## 操作 + +1. 启动控制服务。 +2. 打开控制台 `http://127.0.0.1:8000/`。 +3. 左输出端打开 `/output/left`。 +4. 右输出端打开 `/output/right`。 +5. 在控制台点击场景切换或局部动作。 +6. 观察控制台右侧的节点、FPS、frame time、RTT、clock offset、ACK 状态。 + +## 项目结构 + +```text +led_platform/ + api/ HTTP 和 WebSocket 路由 + core/ 配置、时钟 + domain/ 协议和领域模型 + services/ 场景、tile、同步协调、WebSocket hub + web/static/ 控制台和输出端前端 + main.py FastAPI 应用入口 + cli.py 跨平台命令行入口 + runtime.py 应用依赖装配 +config/ + scenes.json 场景配置 + tiles.json 左右 GPU 服务器和 4 路 4K 输出映射 +tests/ + test_api.py + test_sync.py + test_tiles.py +``` + +## 验证 + +```bash +python -m pytest -q +python -m compileall led_platform tests +node --check led_platform/web/static/output/main.js +node --check led_platform/web/static/admin/main.js +``` diff --git a/config/scenes.json b/config/scenes.json new file mode 100644 index 0000000..835354c --- /dev/null +++ b/config/scenes.json @@ -0,0 +1,29 @@ +{ + "default_scene_id": "overview", + "scenes": [ + { + "id": "overview", + "name": "总览大屏", + "url": "/wall-app", + "view": "overview", + "description": "双 GPU 本地渲染、统一时间轴、左右 tile 裁剪的默认场景", + "preload": [] + }, + { + "id": "energy", + "name": "能源驾驶舱", + "url": "/wall-app", + "view": "energy", + "description": "能源、功率、储能和调度策略场景", + "preload": ["energy-models", "energy-textures"] + }, + { + "id": "security", + "name": "安防态势", + "url": "/wall-app", + "view": "security", + "description": "视频墙、告警、地图联动场景", + "preload": ["camera-grid", "security-map"] + } + ] +} diff --git a/config/tiles.json b/config/tiles.json new file mode 100644 index 0000000..1246fdc --- /dev/null +++ b/config/tiles.json @@ -0,0 +1,44 @@ +{ + "wall": { + "width": 14880, + "height": 3510 + }, + "tiles": [ + { + "id": "left", + "name": "Left GPU server", + "x": 0, + "y": 0, + "width": 7440, + "height": 3510, + "wall_width": 14880, + "wall_height": 3510, + "desktop_width": 7680, + "desktop_height": 4320, + "physical_outputs": [ + { "id": "left-1", "index": 1, "x": 0, "y": 0, "width": 3840, "height": 2160, "connector": "HDMI/DP-1" }, + { "id": "left-2", "index": 2, "x": 3840, "y": 0, "width": 3840, "height": 2160, "connector": "HDMI/DP-2" }, + { "id": "left-3", "index": 3, "x": 0, "y": 2160, "width": 3840, "height": 2160, "connector": "HDMI/DP-3" }, + { "id": "left-4", "index": 4, "x": 3840, "y": 2160, "width": 3840, "height": 2160, "connector": "HDMI/DP-4" } + ] + }, + { + "id": "right", + "name": "Right GPU server", + "x": 7440, + "y": 0, + "width": 7440, + "height": 3510, + "wall_width": 14880, + "wall_height": 3510, + "desktop_width": 7680, + "desktop_height": 4320, + "physical_outputs": [ + { "id": "right-1", "index": 1, "x": 0, "y": 0, "width": 3840, "height": 2160, "connector": "HDMI/DP-1" }, + { "id": "right-2", "index": 2, "x": 3840, "y": 0, "width": 3840, "height": 2160, "connector": "HDMI/DP-2" }, + { "id": "right-3", "index": 3, "x": 0, "y": 2160, "width": 3840, "height": 2160, "connector": "HDMI/DP-3" }, + { "id": "right-4", "index": 4, "x": 3840, "y": 2160, "width": 3840, "height": 2160, "connector": "HDMI/DP-4" } + ] + } + ] +} diff --git a/docs/production-architecture.md b/docs/production-architecture.md new file mode 100644 index 0000000..0af72fe --- /dev/null +++ b/docs/production-architecture.md @@ -0,0 +1,210 @@ +# 生产架构说明 + +## 目标 + +将一个前端大屏应用稳定投放到 `14880 x 3510` LED 大屏。 + +硬件约束: + +- 两台 GPU 服务器。 +- 每台服务器 4 路 4K 输出。 +- 两台服务器品牌或 GPU 可能不同,无法依赖硬件级 GPU 同步。 + +因此系统目标不是让两张 GPU 每一帧硬同步,而是: + +- 统一业务状态。 +- 统一提交时间。 +- 统一动画时间轴。 +- 输出端本地渲染。 +- 拼控完成物理拼接和输入对齐。 + +## 推荐架构 + +```text +控制服务 FastAPI + - 场景管理 + - WebSocket 同步 + - 时间校准 + - ACK 追踪 + - 性能监控 + +左 GPU 服务器 + - 打开 /output/left + - 渲染完整大屏应用的 left tile + - 4 路 4K 输出到拼控 + +右 GPU 服务器 + - 打开 /output/right + - 渲染完整大屏应用的 right tile + - 4 路 4K 输出到拼控 + +拼控 / LED 控制器 + - 接收 8 路 4K + - 按物理坐标拼接成 14880 x 3510 +``` + +## 左右屏如何分开 + +完整逻辑画面: + +```text +|-------------------------- 14880 --------------------------| +|------------ left 7440 ------------|-------- right 7440 ----| +``` + +`config/tiles.json` 定义: + +```text +left: + x=0, y=0, width=7440, height=3510 + +right: + x=7440, y=0, width=7440, height=3510 +``` + +输出端页面创建完整逻辑大屏坐标系,然后根据自己的 tile 做视口偏移。真实 Three.js 项目中,应使用: + +```js +camera.setViewOffset( + fullWidth, + fullHeight, + tile.x, + tile.y, + tile.width, + tile.height +); +``` + +## 每台服务器 4 路 4K + +每台服务器建议配置为 `2 x 2` 逻辑桌面: + +```text +7680 x 4320 +``` + +四路输出: + +```text +1: x=0, y=0, 3840 x 2160 +2: x=3840, y=0, 3840 x 2160 +3: x=0, y=2160, 3840 x 2160 +4: x=3840, y=2160, 3840 x 2160 +``` + +单侧业务 tile 是 `7440 x 3510`,可以放入该逻辑桌面。 + +## 同步协议 + +场景切换: + +```text +POST /api/scenes/{scene_id}/switch +``` + +服务端广播: + +```text +prepare_scene(command_id, apply_at_ms) +commit_scene(command_id, apply_at_ms) +``` + +输出端流程: + +```text +1. 收到 prepare,记录命令,可预加载资源。 +2. 收到 commit,等到 apply_at_ms。 +3. 到点提交场景。 +4. 回 ACK。 +``` + +局部动作: + +```text +POST /api/actions +``` + +支持结构化动作: + +```text +page.next +page.prev +energy.mode +security.alert +timeline.pause +timeline.resume +``` + +## 关键渲染原则 + +所有动画、Three.js、地图和视频都应基于统一时间轴: + +```js +const t = serverNowMs() - sceneStartedAtMs; +renderSceneAt(t); +``` + +不要让左右服务器各自自由播放: + +```js +// 不推荐 +animation += localDeltaTime; +``` + +这样即使某台机器偶尔慢一帧,也会在下一帧追到统一时间,不会越播越偏。 + +## 为什么不推荐单机渲染后网络分发 + +单侧半屏未压缩数据量: + +```text +7440 x 3510 x 4 bytes x 60fps ~= 6.3 GB/s +``` + +左右两侧合计超过 `12 GB/s`。普通网络视频流需要编码、传输、解码,会带来延迟、画质损失、文字细线压缩失真,以及新的编码/解码同步问题。 + +## 生产验收指标 + +建议关注 P95/P99,而不是平均值: + +```text +60 FPS: + P95 frame time < 16.67ms + P99 frame time < 20ms + +30 FPS: + P95 frame time < 33.33ms + P99 frame time < 40ms +``` + +同步指标: + +```text +WebSocket RTT < 10ms +clock offset < 5ms +left/right 都返回 committed 或 action_committed +不得频繁出现 late ACK +``` + +硬件建议: + +- 两台输出服务器尽量使用同型号 GPU、同驱动、同浏览器版本。 +- 电源策略设置为高性能。 +- 关闭系统休眠、屏保、自动更新弹窗。 +- 拼控侧开启输入缓存、帧同步或延迟对齐能力。 +- 生产环境使用 PTP 或内网 NTP。 + +## 边界 + +能保证: + +- 两边业务状态一致。 +- 两边按同一服务端时间点提交。 +- 动画长期不漂移。 +- 命令和性能可观测。 + +不能单独保证: + +- 两张不同 GPU 每一帧物理扫描完全同相。 + +如果必须达到广播级帧同步,需要硬件层支持 Genlock / Frame Lock / 专业视频墙控制器。 diff --git a/led_platform/__init__.py b/led_platform/__init__.py new file mode 100644 index 0000000..23000e3 --- /dev/null +++ b/led_platform/__init__.py @@ -0,0 +1,2 @@ +"""LED wall projection platform.""" + diff --git a/led_platform/api/__init__.py b/led_platform/api/__init__.py new file mode 100644 index 0000000..aec74b4 --- /dev/null +++ b/led_platform/api/__init__.py @@ -0,0 +1,2 @@ +"""HTTP and WebSocket API routes.""" + diff --git a/led_platform/api/http.py b/led_platform/api/http.py new file mode 100644 index 0000000..1c3c66e --- /dev/null +++ b/led_platform/api/http.py @@ -0,0 +1,152 @@ +from pathlib import Path + +from fastapi import APIRouter, HTTPException +from fastapi.responses import FileResponse, RedirectResponse + +from led_platform.core.clock import epoch_ms +from led_platform.domain import CommandEnvelope, CommandType, PerformanceSample +from led_platform.runtime import hub, scenes, settings, sync, tiles + +ROOT = Path(__file__).resolve().parents[2] +WEB_DIR = ROOT / "led_platform" / "web" +STATIC_DIR = WEB_DIR / "static" + +router = APIRouter() + +ALLOWED_ACTIONS = { + "page.next", + "page.prev", + "energy.mode", + "security.alert", + "timeline.pause", + "timeline.resume", +} + + +@router.get("/", include_in_schema=False) +async def admin_index() -> FileResponse: + return FileResponse(STATIC_DIR / "admin" / "index.html") + + +@router.get("/output/{tile_id}", include_in_schema=False) +async def output_index(tile_id: str) -> FileResponse: + try: + tiles.get(tile_id) + except KeyError as exc: + raise HTTPException(status_code=404, detail=f"Unknown output tile: {tile_id}") from exc + return FileResponse(STATIC_DIR / "output" / "index.html") + + +@router.get("/wall-app", include_in_schema=False) +async def wall_app() -> FileResponse: + return FileResponse(STATIC_DIR / "output" / "index.html") + + +@router.get("/favicon.ico", include_in_schema=False) +async def favicon() -> RedirectResponse: + return RedirectResponse("/static/favicon.svg") + + +@router.get("/healthz") +async def healthz() -> dict: + sync_status = sync.status() + return { + "ok": True, + "server_time_ms": epoch_ms(), + "clients": await hub.counts(), + "outputs": len(sync_status["nodes"]), + "pending_commands": len(sync_status["commands"]), + } + + +@router.get("/api/bootstrap/{tile_id}") +async def bootstrap(tile_id: str) -> dict: + try: + tile = tiles.get(tile_id) + except KeyError as exc: + raise HTTPException(status_code=404, detail=f"Unknown output tile: {tile_id}") from exc + return { + "server_time_ms": epoch_ms(), + "target_fps": settings.target_fps, + "tile": tile.model_dump(mode="json"), + "tiles": [item.model_dump(mode="json") for item in tiles.all()], + "state": scenes.state().model_dump(mode="json"), + "scenes": [scene.model_dump(mode="json") for scene in scenes.list()], + } + + +@router.get("/api/scenes") +async def list_scenes() -> dict: + return { + "state": scenes.state().model_dump(mode="json"), + "scenes": [scene.model_dump(mode="json") for scene in scenes.list()], + } + + +@router.post("/api/scenes/{scene_id}/switch") +async def switch_scene(scene_id: str) -> dict: + try: + scene = scenes.get(scene_id) + except KeyError as exc: + raise HTTPException(status_code=404, detail=f"Unknown scene: {scene_id}") from exc + + apply_at_ms = epoch_ms() + settings.switch_prepare_delay_ms + state = scenes.switch(scene.id, apply_at_ms) + payload = { + "state": state.model_dump(mode="json"), + "scene": scene.model_dump(mode="json"), + "server_time_ms": epoch_ms(), + "apply_at_ms": apply_at_ms, + } + prepare = CommandEnvelope(type=CommandType.PREPARE_SCENE, payload=payload) + commit = CommandEnvelope( + type=CommandType.COMMIT_SCENE, + command_id=prepare.command_id, + payload=payload, + ) + sync.register_command(commit, payload) + await hub.broadcast(prepare, "admin", "outputs") + await hub.broadcast(commit, "admin", "outputs") + return payload + + +@router.post("/api/actions") +async def run_action(payload: dict) -> dict: + action = payload.get("action") + if not isinstance(action, str) or action not in ALLOWED_ACTIONS: + raise HTTPException(status_code=400, detail=f"Unsupported action: {action!r}") + + apply_at_ms = epoch_ms() + settings.action_prepare_delay_ms + state = scenes.action(action, apply_at_ms) + command_payload = { + "target": payload.get("target") or "wall", + "action": action, + "args": payload.get("args") or {}, + "state": state.model_dump(mode="json"), + "server_time_ms": epoch_ms(), + "apply_at_ms": apply_at_ms, + } + message = CommandEnvelope(type=CommandType.COMPONENT_ACTION, payload=command_payload) + sync.register_command(message, command_payload) + await hub.broadcast(message, "admin", "outputs") + return command_payload + + +@router.get("/api/tiles") +async def list_tiles() -> dict: + return { + "wall": {"width": settings.wall_width, "height": settings.wall_height}, + "tiles": [tile.model_dump(mode="json") for tile in tiles.all()], + } + + +@router.get("/api/sync/status") +async def sync_status() -> dict: + return sync.status() + + +@router.post("/api/telemetry") +async def telemetry(sample: PerformanceSample) -> dict: + sync.telemetry(sample) + return {"ok": True} + diff --git a/led_platform/api/ws.py b/led_platform/api/ws.py new file mode 100644 index 0000000..1542482 --- /dev/null +++ b/led_platform/api/ws.py @@ -0,0 +1,114 @@ +from fastapi import APIRouter, WebSocket, WebSocketDisconnect + +from led_platform.core.clock import epoch_ms +from led_platform.domain import CommandEnvelope, CommandType, PerformanceSample +from led_platform.runtime import hub, scenes, settings, sync, tiles + +router = APIRouter() + + +def initial_state_payload() -> dict: + apply_at_ms = epoch_ms() + settings.initial_state_delay_ms + state = scenes.state().model_copy(update={"apply_at_ms": apply_at_ms}) + return { + "state": state.model_dump(mode="json"), + "server_time_ms": epoch_ms(), + "apply_at_ms": apply_at_ms, + } + + +@router.websocket("/ws/admin") +async def admin_ws(websocket: WebSocket) -> None: + await hub.connect("admin", websocket) + try: + await hub.send(websocket, CommandEnvelope(type=CommandType.STATE, payload=initial_state_payload())) + while True: + raw = await websocket.receive_json() + if raw.get("type") == CommandType.CLOCK_PING: + await hub.send( + websocket, + CommandEnvelope( + type=CommandType.CLOCK_PONG, + payload={ + "client_send_ms": raw.get("client_send_ms"), + "server_time_ms": epoch_ms(), + }, + ), + ) + except WebSocketDisconnect: + await hub.disconnect("admin", websocket) + + +@router.websocket("/ws/output/{tile_id}") +async def output_ws(websocket: WebSocket, tile_id: str) -> None: + try: + tiles.get(tile_id) + except KeyError: + await websocket.close(code=1008) + return + + node_id: str | None = None + await hub.connect("outputs", websocket) + try: + await hub.send(websocket, CommandEnvelope(type=CommandType.STATE, payload=initial_state_payload())) + while True: + raw = await websocket.receive_json() + message_type = raw.get("type") + + if message_type == CommandType.HELLO: + node_id = str(raw.get("node_id") or f"{tile_id}-anonymous") + node = sync.connect_node( + node_id=node_id, + tile_id=tile_id, + app_version=raw.get("app_version"), + user_agent=raw.get("user_agent"), + ) + await hub.broadcast( + CommandEnvelope( + type=CommandType.HEARTBEAT, + payload=node.model_dump(mode="json"), + ), + "admin", + ) + + elif message_type == CommandType.CLOCK_PING: + await hub.send( + websocket, + CommandEnvelope( + type=CommandType.CLOCK_PONG, + payload={ + "node_id": raw.get("node_id") or node_id, + "client_send_ms": raw.get("client_send_ms"), + "server_time_ms": epoch_ms(), + }, + ), + ) + + elif message_type == CommandType.ACK: + node_id = str(raw.get("node_id") or node_id or "") + sync.clock_quality(node_id, raw.get("clock_offset_ms"), raw.get("rtt_ms")) + record = sync.ack(raw) + await hub.broadcast( + CommandEnvelope( + type=CommandType.ACK, + payload={**raw, "record": record.model_dump(mode="json") if record else None}, + ), + "admin", + ) + + elif message_type == CommandType.TELEMETRY: + if raw.get("node_id") and raw.get("tile_id"): + sync.telemetry( + PerformanceSample( + node_id=raw["node_id"], + tile_id=raw["tile_id"], + fps=float(raw.get("fps") or 0), + frame_time_ms=float(raw.get("frame_time_ms") or 0), + dropped_frames=int(raw.get("dropped_frames") or 0), + ) + ) + + except WebSocketDisconnect: + sync.disconnect_node(node_id) + await hub.disconnect("outputs", websocket) + diff --git a/led_platform/cli.py b/led_platform/cli.py new file mode 100644 index 0000000..4946a05 --- /dev/null +++ b/led_platform/cli.py @@ -0,0 +1,56 @@ +import webbrowser + +import typer +import uvicorn + +from led_platform.core.config import get_settings + +app = typer.Typer(help="LED wall projection platform CLI.") + + +@app.command() +def serve( + host: str = typer.Option(None, help="Bind host."), + port: int = typer.Option(None, help="Bind port."), + reload: bool = typer.Option(False, help="Enable uvicorn reload."), +) -> None: + settings = get_settings() + uvicorn.run( + "led_platform.main:app", + host=host or settings.app_host, + port=port or settings.app_port, + reload=reload, + ) + + +@app.command() +def urls( + base: str = typer.Option("http://127.0.0.1:8000", help="Controller base URL."), +) -> None: + typer.echo(f"Admin: {base}/") + typer.echo(f"Left output: {base}/output/left") + typer.echo(f"Right output:{base}/output/right") + typer.echo(f"Health: {base}/healthz") + typer.echo(f"Sync: {base}/api/sync/status") + + +@app.command() +def open( + base: str = typer.Option("http://127.0.0.1:8000", help="Controller base URL."), + target: str = typer.Option("admin", help="admin, left, right, sync"), +) -> None: + targets = { + "admin": f"{base}/", + "left": f"{base}/output/left", + "right": f"{base}/output/right", + "sync": f"{base}/api/sync/status", + } + url = targets.get(target) + if not url: + raise typer.BadParameter("target must be one of: admin, left, right, sync") + webbrowser.open(url) + typer.echo(url) + + +if __name__ == "__main__": + app() diff --git a/led_platform/core/__init__.py b/led_platform/core/__init__.py new file mode 100644 index 0000000..90db460 --- /dev/null +++ b/led_platform/core/__init__.py @@ -0,0 +1,2 @@ +"""Core platform utilities.""" + diff --git a/led_platform/core/clock.py b/led_platform/core/clock.py new file mode 100644 index 0000000..1b93a0a --- /dev/null +++ b/led_platform/core/clock.py @@ -0,0 +1,11 @@ +from datetime import datetime, timezone +from time import time + + +def epoch_ms() -> int: + return int(time() * 1000) + + +def utc_now() -> datetime: + return datetime.now(timezone.utc) + diff --git a/led_platform/core/config.py b/led_platform/core/config.py new file mode 100644 index 0000000..be58b80 --- /dev/null +++ b/led_platform/core/config.py @@ -0,0 +1,30 @@ +from functools import lru_cache +from pathlib import Path + +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + app_host: str = "127.0.0.1" + app_port: int = 8000 + public_base_url: str = "http://127.0.0.1:8000" + + wall_width: int = 14880 + wall_height: int = 3510 + target_fps: int = 60 + + switch_prepare_delay_ms: int = 1400 + action_prepare_delay_ms: int = 650 + initial_state_delay_ms: int = 160 + command_ttl_seconds: int = 45 + + scene_config: Path = Path("config/scenes.json") + tile_config: Path = Path("config/tiles.json") + + model_config = SettingsConfigDict(env_prefix="LED_", env_file=".env", extra="ignore") + + +@lru_cache +def get_settings() -> Settings: + return Settings() + diff --git a/led_platform/domain/__init__.py b/led_platform/domain/__init__.py new file mode 100644 index 0000000..32875ed --- /dev/null +++ b/led_platform/domain/__init__.py @@ -0,0 +1,30 @@ +from .models import ( + AckStatus, + CommandEnvelope, + CommandRecord, + CommandType, + NodeRole, + NodeStatus, + OutputNode, + PerformanceSample, + PhysicalOutput, + Scene, + SceneState, + Tile, +) + +__all__ = [ + "AckStatus", + "CommandEnvelope", + "CommandRecord", + "CommandType", + "NodeRole", + "NodeStatus", + "OutputNode", + "PerformanceSample", + "PhysicalOutput", + "Scene", + "SceneState", + "Tile", +] + diff --git a/led_platform/domain/models.py b/led_platform/domain/models.py new file mode 100644 index 0000000..9f67c3e --- /dev/null +++ b/led_platform/domain/models.py @@ -0,0 +1,141 @@ +from enum import Enum +from typing import Any +from uuid import uuid4 + +from pydantic import BaseModel, Field, HttpUrl + +from led_platform.core.clock import epoch_ms + + +class CommandType(str, Enum): + HELLO = "hello" + STATE = "state" + PREPARE_SCENE = "prepare_scene" + COMMIT_SCENE = "commit_scene" + COMPONENT_ACTION = "component_action" + CLOCK_PING = "clock_ping" + CLOCK_PONG = "clock_pong" + TELEMETRY = "telemetry" + HEARTBEAT = "heartbeat" + ACK = "ack" + ERROR = "error" + + +class AckStatus(str, Enum): + PREPARED = "prepared" + COMMITTED = "committed" + ACTION_COMMITTED = "action_committed" + LATE = "late" + ERROR = "error" + + +class NodeRole(str, Enum): + ADMIN = "admin" + OUTPUT = "output" + RENDERER = "renderer" + + +class NodeStatus(str, Enum): + STARTING = "starting" + READY = "ready" + DEGRADED = "degraded" + ERROR = "error" + + +class PhysicalOutput(BaseModel): + id: str + index: int + x: int + y: int + width: int = 3840 + height: int = 2160 + connector: str | None = None + + +class Tile(BaseModel): + id: str + name: str + x: int + y: int + width: int + height: int + wall_width: int + wall_height: int + desktop_width: int + desktop_height: int + physical_outputs: list[PhysicalOutput] = Field(default_factory=list) + + @property + def right(self) -> int: + return self.x + self.width + + +class Scene(BaseModel): + id: str + name: str + url: str | HttpUrl = "/wall-app" + view: str + description: str | None = None + preload: list[str] = Field(default_factory=list) + + +class SceneState(BaseModel): + active_scene_id: str + active_view: str + active_url: str + version: int + scene_started_at_ms: int + apply_at_ms: int | None = None + action: str | None = None + + +class CommandEnvelope(BaseModel): + type: CommandType + command_id: str = Field(default_factory=lambda: uuid4().hex) + payload: dict[str, Any] = Field(default_factory=dict) + sent_at_ms: int = Field(default_factory=epoch_ms) + + +class OutputNode(BaseModel): + node_id: str + tile_id: str + status: NodeStatus = NodeStatus.READY + connected_at_ms: int = Field(default_factory=epoch_ms) + last_seen_ms: int = Field(default_factory=epoch_ms) + app_version: str | None = None + user_agent: str | None = None + clock_offset_ms: float | None = None + rtt_ms: float | None = None + fps: float | None = None + frame_time_ms: float | None = None + dropped_frames: int | None = None + gpu_hint: str | None = None + + +class PerformanceSample(BaseModel): + node_id: str + tile_id: str + fps: float + frame_time_ms: float + dropped_frames: int = 0 + at_ms: int = Field(default_factory=epoch_ms) + + +class CommandRecord(BaseModel): + command_id: str + command_type: CommandType + target_tiles: list[str] + apply_at_ms: int + created_at_ms: int = Field(default_factory=epoch_ms) + expires_at_ms: int + payload: dict[str, Any] = Field(default_factory=dict) + acks: list[dict[str, Any]] = Field(default_factory=list) + + @property + def complete(self) -> bool: + completed = { + ack.get("tile_id") + for ack in self.acks + if ack.get("status") in {AckStatus.COMMITTED, AckStatus.ACTION_COMMITTED} + } + return set(self.target_tiles).issubset({str(tile) for tile in completed if tile}) diff --git a/led_platform/main.py b/led_platform/main.py new file mode 100644 index 0000000..361cda6 --- /dev/null +++ b/led_platform/main.py @@ -0,0 +1,32 @@ +from fastapi import FastAPI +from fastapi.staticfiles import StaticFiles + +from led_platform.api.http import STATIC_DIR, router as http_router +from led_platform.api.ws import router as ws_router +from led_platform.core.config import get_settings + + +def create_app() -> FastAPI: + app = FastAPI( + title="LED Wall Projection Platform", + version="1.0.0", + description="Tile-aware dual GPU-server LED wall control and synchronization platform.", + ) + app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") + app.include_router(http_router) + app.include_router(ws_router) + return app + + +app = create_app() + + +def main() -> None: + import uvicorn + + settings = get_settings() + uvicorn.run("led_platform.main:app", host=settings.app_host, port=settings.app_port) + + +if __name__ == "__main__": + main() diff --git a/led_platform/runtime.py b/led_platform/runtime.py new file mode 100644 index 0000000..395ae07 --- /dev/null +++ b/led_platform/runtime.py @@ -0,0 +1,12 @@ +from led_platform.core.config import get_settings +from led_platform.services.scene_store import SceneStore +from led_platform.services.sync_coordinator import SyncCoordinator +from led_platform.services.tile_store import TileStore +from led_platform.services.ws_hub import WebSocketHub + +settings = get_settings() +tiles = TileStore(settings.tile_config, settings.wall_width, settings.wall_height) +scenes = SceneStore(settings.scene_config) +hub = WebSocketHub() +sync = SyncCoordinator(tiles.ids(), settings.command_ttl_seconds) + diff --git a/led_platform/services/__init__.py b/led_platform/services/__init__.py new file mode 100644 index 0000000..1add9ae --- /dev/null +++ b/led_platform/services/__init__.py @@ -0,0 +1,2 @@ +"""Application services.""" + diff --git a/led_platform/services/scene_store.py b/led_platform/services/scene_store.py new file mode 100644 index 0000000..76bb9b6 --- /dev/null +++ b/led_platform/services/scene_store.py @@ -0,0 +1,82 @@ +import json +from pathlib import Path +from threading import Lock + +from led_platform.core.clock import epoch_ms +from led_platform.domain import Scene, SceneState + + +class SceneStore: + def __init__(self, config_path: Path) -> None: + self.config_path = config_path + self._lock = Lock() + self._scenes: dict[str, Scene] = {} + self._state: SceneState | None = None + self.reload() + + def reload(self) -> None: + data = json.loads(self.config_path.read_text(encoding="utf-8")) + scenes = [Scene(**item) for item in data["scenes"]] + if not scenes: + raise ValueError("At least one scene is required.") + default_scene_id = data.get("default_scene_id") or scenes[0].id + scene_map = {scene.id: scene for scene in scenes} + if default_scene_id not in scene_map: + raise ValueError(f"Default scene {default_scene_id!r} does not exist.") + default_scene = scene_map[default_scene_id] + with self._lock: + version = self._state.version + 1 if self._state else 1 + self._scenes = scene_map + self._state = self._build_state(default_scene, version=version, apply_at_ms=None) + + def list(self) -> list[Scene]: + with self._lock: + return list(self._scenes.values()) + + def get(self, scene_id: str) -> Scene: + with self._lock: + if scene_id not in self._scenes: + raise KeyError(scene_id) + return self._scenes[scene_id] + + def state(self) -> SceneState: + with self._lock: + if self._state is None: + raise RuntimeError("Scene state is not initialized.") + return self._state + + def switch(self, scene_id: str, apply_at_ms: int) -> SceneState: + with self._lock: + if scene_id not in self._scenes: + raise KeyError(scene_id) + version = self._state.version + 1 if self._state else 1 + self._state = self._build_state( + self._scenes[scene_id], + version=version, + apply_at_ms=apply_at_ms, + ) + return self._state + + def action(self, action: str, apply_at_ms: int) -> SceneState: + with self._lock: + if self._state is None: + raise RuntimeError("Scene state is not initialized.") + self._state = self._state.model_copy( + update={ + "version": self._state.version + 1, + "action": action, + "apply_at_ms": apply_at_ms, + } + ) + return self._state + + @staticmethod + def _build_state(scene: Scene, version: int, apply_at_ms: int | None) -> SceneState: + return SceneState( + active_scene_id=scene.id, + active_view=scene.view, + active_url=str(scene.url), + version=version, + scene_started_at_ms=apply_at_ms or epoch_ms(), + apply_at_ms=apply_at_ms, + ) diff --git a/led_platform/services/sync_coordinator.py b/led_platform/services/sync_coordinator.py new file mode 100644 index 0000000..c8421f5 --- /dev/null +++ b/led_platform/services/sync_coordinator.py @@ -0,0 +1,115 @@ +from threading import Lock + +from led_platform.core.clock import epoch_ms +from led_platform.domain import CommandEnvelope, CommandRecord, OutputNode, PerformanceSample + + +class SyncCoordinator: + def __init__(self, target_tiles: list[str], command_ttl_seconds: int) -> None: + self.target_tiles = target_tiles + self.command_ttl_seconds = command_ttl_seconds + self._lock = Lock() + self._nodes: dict[str, OutputNode] = {} + self._commands: dict[str, CommandRecord] = {} + + def connect_node( + self, + *, + node_id: str, + tile_id: str, + app_version: str | None = None, + user_agent: str | None = None, + ) -> OutputNode: + with self._lock: + node = OutputNode( + node_id=node_id, + tile_id=tile_id, + app_version=app_version, + user_agent=user_agent, + ) + self._nodes[node_id] = node + return node + + def disconnect_node(self, node_id: str | None) -> None: + if not node_id: + return + with self._lock: + self._nodes.pop(node_id, None) + + def register_command(self, message: CommandEnvelope, payload: dict) -> CommandRecord: + with self._lock: + self._prune_locked() + apply_at_ms = int(payload["apply_at_ms"]) + record = CommandRecord( + command_id=message.command_id, + command_type=message.type, + target_tiles=self.target_tiles, + apply_at_ms=apply_at_ms, + expires_at_ms=epoch_ms() + self.command_ttl_seconds * 1000, + payload=payload, + ) + self._commands[message.command_id] = record + return record + + def ack(self, payload: dict) -> CommandRecord | None: + with self._lock: + self._prune_locked() + command_id = payload.get("command_id") + if not isinstance(command_id, str) or command_id not in self._commands: + return None + ack_payload = {**payload, "server_received_ms": epoch_ms()} + record = self._commands[command_id] + record.acks = [ + ack for ack in record.acks if ack.get("node_id") != ack_payload.get("node_id") + ] + record.acks.append(ack_payload) + return record + + def clock_quality( + self, + node_id: str | None, + clock_offset_ms: float | None, + rtt_ms: float | None, + ) -> None: + if not node_id: + return + with self._lock: + node = self._nodes.get(node_id) + if not node: + return + node.last_seen_ms = epoch_ms() + node.clock_offset_ms = clock_offset_ms + node.rtt_ms = rtt_ms + + def telemetry(self, sample: PerformanceSample) -> None: + with self._lock: + node = self._nodes.get(sample.node_id) + if not node: + return + node.last_seen_ms = epoch_ms() + node.fps = sample.fps + node.frame_time_ms = sample.frame_time_ms + node.dropped_frames = sample.dropped_frames + + def status(self) -> dict: + with self._lock: + self._prune_locked() + return { + "server_time_ms": epoch_ms(), + "target_tiles": self.target_tiles, + "nodes": [node.model_dump(mode="json") for node in self._nodes.values()], + "commands": [ + {**record.model_dump(mode="json"), "complete": record.complete} + for record in self._commands.values() + ], + } + + def _prune_locked(self) -> None: + now = epoch_ms() + stale = [ + command_id + for command_id, record in self._commands.items() + if record.expires_at_ms < now + ] + for command_id in stale: + self._commands.pop(command_id, None) diff --git a/led_platform/services/tile_store.py b/led_platform/services/tile_store.py new file mode 100644 index 0000000..c464bbf --- /dev/null +++ b/led_platform/services/tile_store.py @@ -0,0 +1,73 @@ +import json +from pathlib import Path + +from led_platform.domain import PhysicalOutput, Tile + + +class TileStore: + def __init__(self, config_path: Path, wall_width: int, wall_height: int) -> None: + self.config_path = config_path + self.wall_width = wall_width + self.wall_height = wall_height + self._tiles: dict[str, Tile] = {} + self.reload() + + def reload(self) -> None: + if self.config_path.exists(): + data = json.loads(self.config_path.read_text(encoding="utf-8")) + tiles = [Tile(**item) for item in data["tiles"]] + else: + tiles = self._default_tiles() + self._tiles = {tile.id: tile for tile in tiles} + + def all(self) -> list[Tile]: + return list(self._tiles.values()) + + def get(self, tile_id: str) -> Tile: + if tile_id not in self._tiles: + raise KeyError(tile_id) + return self._tiles[tile_id] + + def ids(self) -> list[str]: + return list(self._tiles.keys()) + + def _default_tiles(self) -> list[Tile]: + half_width = self.wall_width // 2 + return [ + Tile( + id="left", + name="Left output server", + x=0, + y=0, + width=half_width, + height=self.wall_height, + wall_width=self.wall_width, + wall_height=self.wall_height, + desktop_width=7680, + desktop_height=4320, + physical_outputs=self._default_outputs("left"), + ), + Tile( + id="right", + name="Right output server", + x=half_width, + y=0, + width=self.wall_width - half_width, + height=self.wall_height, + wall_width=self.wall_width, + wall_height=self.wall_height, + desktop_width=7680, + desktop_height=4320, + physical_outputs=self._default_outputs("right"), + ), + ] + + @staticmethod + def _default_outputs(prefix: str) -> list[PhysicalOutput]: + return [ + PhysicalOutput(id=f"{prefix}-1", index=1, x=0, y=0), + PhysicalOutput(id=f"{prefix}-2", index=2, x=3840, y=0), + PhysicalOutput(id=f"{prefix}-3", index=3, x=0, y=2160), + PhysicalOutput(id=f"{prefix}-4", index=4, x=3840, y=2160), + ] + diff --git a/led_platform/services/ws_hub.py b/led_platform/services/ws_hub.py new file mode 100644 index 0000000..278de49 --- /dev/null +++ b/led_platform/services/ws_hub.py @@ -0,0 +1,50 @@ +import asyncio +from collections import defaultdict + +from fastapi import WebSocket + +from led_platform.domain import CommandEnvelope + + +class WebSocketHub: + def __init__(self) -> None: + self._lock = asyncio.Lock() + self._groups: dict[str, set[WebSocket]] = defaultdict(set) + + async def connect(self, group: str, websocket: WebSocket) -> None: + await websocket.accept() + async with self._lock: + self._groups[group].add(websocket) + + async def disconnect(self, group: str, websocket: WebSocket) -> None: + async with self._lock: + if group in self._groups: + self._groups[group].discard(websocket) + if not self._groups[group]: + self._groups.pop(group, None) + + async def send(self, websocket: WebSocket, message: CommandEnvelope) -> None: + await websocket.send_json(message.model_dump(mode="json")) + + async def broadcast(self, message: CommandEnvelope, *groups: str) -> None: + async with self._lock: + targets: list[tuple[str, WebSocket]] = [] + for group in groups: + targets.extend((group, websocket) for websocket in self._groups.get(group, set())) + + stale: list[tuple[str, WebSocket]] = [] + for group, websocket in targets: + try: + await self.send(websocket, message) + except RuntimeError: + stale.append((group, websocket)) + + if stale: + async with self._lock: + for group, websocket in stale: + self._groups[group].discard(websocket) + + async def counts(self) -> dict[str, int]: + async with self._lock: + return {group: len(websockets) for group, websockets in self._groups.items()} + diff --git a/led_platform/web/static/admin/index.html b/led_platform/web/static/admin/index.html new file mode 100644 index 0000000..de37ec8 --- /dev/null +++ b/led_platform/web/static/admin/index.html @@ -0,0 +1,66 @@ + + +
+ + +正在连接控制服务...
+${scene.view} | ${scene.url}
+${scene.description ?? ""}
+Unified Timeline Rendering
+左右服务器各自渲染半屏,但所有场景、动画、地图相机、视频时间都由同一个服务端时间轴驱动。
+适合 ECharts、Canvas、WebGL 地图和 Three.js 数字孪生。真实项目中应使用 camera.setViewOffset 做 tile-aware 渲染。
+视频墙、告警卡片和地图联动都通过结构化动作同步触发,不在输出端执行任意远程脚本。
+