3d36cc5754
- Import encode_animation_cmd and ShowModel at module level - Add show, fired_ids, is_playing per-connection state variables - Extend load handler to load show from show_store by show_id (D-04) - play handler sets is_playing=True - pause handler sets is_playing=False without touching fired_ids - seek handler rebuilds fired_ids from show cues before seek position (D-03)
188 lines
7.7 KiB
Python
188 lines
7.7 KiB
Python
"""WebSocket hub — browser-driven tick/beat protocol (Phase 02-03 redesign).
|
|
|
|
Browser sends position ticks at ~10Hz; server looks up the next beat and
|
|
sends a beat notification back when one is near. Phase 5 adds cue scheduling:
|
|
the tick handler fires UDP animation commands at correct timestamps.
|
|
|
|
Protocol (browser → server):
|
|
{"type": "tick", "position": 42.3} — audio playback position
|
|
{"type": "play", "position": 42.3} — playback started
|
|
{"type": "pause", "position": 42.3} — playback paused
|
|
{"type": "seek", "position": 42.3} — seek event
|
|
{"type": "load", "path": "...", "show_id": "uuid"} — file + show loaded
|
|
|
|
Protocol (server → browser):
|
|
{"type": "beat", "position": 42.3, "tempo": 120.0} — beat detected near position
|
|
{"type": "preview_update", "device_id": "uuid", "animation": "...", "color": [...]}
|
|
{"type": "error", "message": "..."} — error response
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import bisect
|
|
import json
|
|
import logging
|
|
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
|
|
from lightsync.protocol.animation_cmd import encode_animation_cmd
|
|
from lightsync.models.show import ShowModel
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
# Beat detection tolerance window: consider a beat "hit" if audio position
|
|
# is within this many seconds of a known beat timestamp.
|
|
BEAT_WINDOW_SEC = 0.05 # 50ms
|
|
|
|
|
|
class ConnectionManager:
|
|
def __init__(self):
|
|
self.active_connections: list[WebSocket] = []
|
|
|
|
async def connect(self, ws: WebSocket) -> None:
|
|
await ws.accept()
|
|
self.active_connections.append(ws)
|
|
|
|
def disconnect(self, ws: WebSocket) -> None:
|
|
if ws in self.active_connections:
|
|
self.active_connections.remove(ws)
|
|
|
|
async def broadcast(self, message: dict) -> None:
|
|
dead = []
|
|
for connection in self.active_connections:
|
|
try:
|
|
await connection.send_text(json.dumps(message))
|
|
except Exception:
|
|
dead.append(connection)
|
|
for d in dead:
|
|
if d in self.active_connections:
|
|
self.active_connections.remove(d)
|
|
|
|
|
|
manager = ConnectionManager()
|
|
|
|
|
|
def _find_nearest_beat(beats: list[float], position: float) -> float | None:
|
|
"""Return the nearest beat timestamp to position, or None if beats is empty.
|
|
|
|
Uses bisect for O(log n) lookup.
|
|
"""
|
|
if not beats:
|
|
return None
|
|
idx = bisect.bisect_left(beats, position)
|
|
candidates = []
|
|
if idx < len(beats):
|
|
candidates.append(beats[idx])
|
|
if idx > 0:
|
|
candidates.append(beats[idx - 1])
|
|
return min(candidates, key=lambda b: abs(b - position))
|
|
|
|
|
|
@router.websocket("/ws")
|
|
async def websocket_endpoint(websocket: WebSocket):
|
|
await manager.connect(websocket)
|
|
# Per-connection state: which file's beats are we tracking
|
|
current_file: str | None = None
|
|
beats: list[float] = []
|
|
tempo: float = 0.0
|
|
last_beat_reported: float | None = None # avoid re-reporting same beat
|
|
|
|
# Per-connection cue scheduler state (Phase 5 — D-04)
|
|
show: ShowModel | None = None # loaded show for cue scheduling
|
|
fired_ids: set[str] = set() # cues already dispatched this playback pass
|
|
is_playing: bool = False # gate cue scheduling on play state
|
|
|
|
try:
|
|
while True:
|
|
data = await websocket.receive_text()
|
|
try:
|
|
msg = json.loads(data)
|
|
except json.JSONDecodeError:
|
|
await websocket.send_text(json.dumps({"type": "error", "message": "invalid JSON"}))
|
|
continue
|
|
|
|
msg_type = msg.get("type")
|
|
|
|
if msg_type == "tick":
|
|
position = float(msg.get("position", 0.0))
|
|
|
|
# Lazily load beat data from app state when available
|
|
if not beats:
|
|
beats_cache = getattr(websocket.app.state, "beats", None) or {}
|
|
# Use the most recently cached file if we don't have one yet
|
|
if beats_cache:
|
|
# Pick the last uploaded file (most likely what's playing)
|
|
latest_path = list(beats_cache.keys())[-1]
|
|
beat_data = beats_cache[latest_path]
|
|
current_file = latest_path
|
|
beats = sorted(beat_data.get("beats", []))
|
|
tempo = float(beat_data.get("tempo", 0.0))
|
|
|
|
if beats:
|
|
nearest = _find_nearest_beat(beats, position)
|
|
if nearest is not None and abs(nearest - position) <= BEAT_WINDOW_SEC:
|
|
# Only send each beat once per playback pass
|
|
if last_beat_reported != nearest:
|
|
last_beat_reported = nearest
|
|
await websocket.send_text(json.dumps({
|
|
"type": "beat",
|
|
"position": nearest,
|
|
"tempo": tempo,
|
|
}))
|
|
|
|
elif msg_type == "play":
|
|
logger.info("[ws] play event at position=%.3f", float(msg.get("position", 0)))
|
|
last_beat_reported = None # reset beat tracking on play
|
|
is_playing = True # enable cue scheduling
|
|
|
|
elif msg_type == "pause":
|
|
logger.info("[ws] pause event at position=%.3f", float(msg.get("position", 0)))
|
|
is_playing = False # disable cue scheduling; do NOT modify fired_ids
|
|
|
|
elif msg_type == "seek":
|
|
position = float(msg.get("position", 0.0))
|
|
logger.info("[ws] seek to position=%.3f", position)
|
|
last_beat_reported = None # reset so beats near seek target fire
|
|
# Rebuild fired_ids: mark everything before seek position as fired (D-03)
|
|
if show is not None:
|
|
fired_ids = {
|
|
str(cue.id)
|
|
for track in show.tracks
|
|
for cue in track.cues
|
|
if cue.timestamp < position - 0.060
|
|
}
|
|
|
|
elif msg_type == "load":
|
|
# Browser signals which file is now playing — update beat cache lookup
|
|
path = msg.get("path", "")
|
|
if path:
|
|
beats_cache = getattr(websocket.app.state, "beats", None) or {}
|
|
if path in beats_cache:
|
|
current_file = path
|
|
beat_data = beats_cache[path]
|
|
beats = sorted(beat_data.get("beats", []))
|
|
tempo = float(beat_data.get("tempo", 0.0))
|
|
last_beat_reported = None
|
|
logger.info("[ws] loaded beat data for %s: %d beats @ %.1f BPM", path, len(beats), tempo)
|
|
else:
|
|
logger.warning("[ws] no beat data cached for %s", path)
|
|
|
|
# Show cue list loading (Phase 5 — D-04)
|
|
show_id = msg.get("show_id")
|
|
if show_id:
|
|
import lightsync.main as _main
|
|
loaded = await _main.show_store.load(show_id)
|
|
if loaded:
|
|
show = loaded
|
|
fired_ids = set()
|
|
is_playing = False
|
|
logger.info("[ws] loaded show %s with %d tracks", show_id, len(show.tracks))
|
|
else:
|
|
logger.warning("[ws] show %s not found", show_id)
|
|
|
|
else:
|
|
# Unknown message type — echo back for debugging
|
|
await websocket.send_text(json.dumps({"type": "ack", "echo": msg}))
|
|
|
|
except WebSocketDisconnect:
|
|
manager.disconnect(websocket)
|