64c49c9e64
- LOOKAHEAD = 0.060 (60ms window per D-02) - Scheduler iterates show.tracks/cues on each tick, gated on is_playing and show - Cues within lookahead window fire UDP via encode_animation_cmd + UDPSender.send - Past cues (>100ms behind) silently marked fired without UDP dispatch - active_cue tracking per track enables preview state for already-fired blocks - preview_update messages broadcast to all clients on each tick (D-07) - Clear preview_update (animation: null) sent when no active cue on a device
250 lines
11 KiB
Python
250 lines
11 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,
|
|
}))
|
|
|
|
# ── 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)
|