"""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, })) # ── Cue scheduler (Phase 5 — D-01, D-02) ────────────────── if is_playing and show is not None: LOOKAHEAD = 0.060 # 60ms look-ahead window (D-02) udp = websocket.app.state.udp_sender preview_updates = [] # batch preview messages for track in show.tracks: device = next( (d for d in show.devices if str(d.id) == str(track.device_id)), None, ) if device is None: continue active_cue = None # track which cue is active for preview for cue in track.cues: cue_id = str(cue.id) if cue_id in fired_ids: # Check if this already-fired cue is still active (for preview) if cue.timestamp <= position < cue.timestamp + cue.duration: active_cue = cue continue if cue.timestamp > position + LOOKAHEAD: continue # not due yet if cue.timestamp < position - 0.1: # Too far in the past — mark fired without sending fired_ids.add(cue_id) continue # Fire this cue (D-01) fired_ids.add(cue_id) active_cue = cue if cue.animation: try: payload = encode_animation_cmd(cue.animation, cue.params) udp.send(payload, device.ip, device.port) except (ValueError, KeyError): logger.warning("[ws] unknown animation %r", cue.animation) # Build preview update for this device (D-07) if active_cue and active_cue.animation: color = active_cue.params.get("color", [0, 255, 180]) preview_updates.append({ "type": "preview_update", "device_id": str(track.device_id), "animation": active_cue.animation, "color": color, }) else: # No active cue — send clear (D-05: turns dark) preview_updates.append({ "type": "preview_update", "device_id": str(track.device_id), "animation": None, "color": None, }) # Broadcast all preview updates for pu in preview_updates: await manager.broadcast(pu) 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)