diff --git a/lightsync/api/ws.py b/lightsync/api/ws.py index c22557a..7138a43 100644 --- a/lightsync/api/ws.py +++ b/lightsync/api/ws.py @@ -1,9 +1,33 @@ -import asyncio +"""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. + +Protocol (browser → server): + {"type": "tick", "position": 42.3} — audio playback position + {"type": "play"} — playback started (log only for now) + {"type": "pause"} — playback paused (log only for now) + {"type": "seek", "position": 42.3} — seek event (log only for now) + +Protocol (server → browser): + {"type": "beat", "position": 42.3, "tempo": 120.0} — beat detected near position + {"type": "error", "message": "..."} — error response +""" +from __future__ import annotations + +import bisect import json +import logging from fastapi import APIRouter, WebSocket, WebSocketDisconnect +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): @@ -32,54 +56,99 @@ class ConnectionManager: manager = ConnectionManager() -async def broadcast_loop(get_engine) -> None: - """Broadcast playback state to all WebSocket clients at 10Hz (AUD-04). +def _find_nearest_beat(beats: list[float], position: float) -> float | None: + """Return the nearest beat timestamp to position, or None if beats is empty. - Args: - get_engine: callable returning the MPVEngine instance (avoids import cycle) + Uses bisect for O(log n) lookup. """ - while True: - await asyncio.sleep(0.1) - engine = get_engine() - if engine and manager.active_connections: - state = engine.get_state() - await manager.broadcast({"type": "tick", **state}) + 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 + try: while True: data = await websocket.receive_text() - msg = json.loads(data) - msg_type = msg.get("type") - - # Import engine reference from main (module-level var) - from lightsync.main import engine - - if engine is None: - await websocket.send_text(json.dumps({"type": "error", "message": "engine not ready"})) + try: + msg = json.loads(data) + except json.JSONDecodeError: + await websocket.send_text(json.dumps({"type": "error", "message": "invalid JSON"})) continue - if msg_type == "play": - engine.play() - await websocket.send_text(json.dumps({"type": "ack", "command": "play"})) + 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 + elif msg_type == "pause": - engine.pause() - await websocket.send_text(json.dumps({"type": "ack", "command": "pause"})) + logger.info("[ws] pause event at position=%.3f", float(msg.get("position", 0))) + elif msg_type == "seek": - position = float(msg.get("position", 0)) - engine.seek(position) - await websocket.send_text(json.dumps({"type": "ack", "command": "seek", "position": position})) + 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 + elif msg_type == "load": + # Browser signals which file is now playing — update beat cache lookup path = msg.get("path", "") if path: - engine.load(path) - await websocket.send_text(json.dumps({"type": "ack", "command": "load", "path": path})) - else: - await websocket.send_text(json.dumps({"type": "error", "message": "missing 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) + else: + # Unknown message type — echo back for debugging await websocket.send_text(json.dumps({"type": "ack", "echo": msg})) + except WebSocketDisconnect: manager.disconnect(websocket)