alexei.dolgolyov
51ec1503f4
Lint & Test / test (push) Successful in 10s
Frontend hot path (player.js, background.js):
- visualizer rAF: drop per-frame getComputedStyle('--accent') (cached on
applyAccentColor), build canvas LinearGradient once per accent change
instead of 32× per frame, batch all bars into a single beginPath/fill
- FPS-gate canvas redraw via frequencyDataVersion so 60-144 Hz monitors
stop re-rendering identical frames produced at 30 Hz on the backend
- editorial spectrum bars: replace style.height (layout) with
transform: scaleY (compositor-only); cache bar refs, pre-compute
per-bar gain/range, dedup writes at 1/1000 quantization
- coalesce VU needle into the visualizer rAF; cache vuNeedle ref;
dedup angle writes at 0.1°
- updateUI: status-payload fingerprint short-circuits the redundant
status_update broadcasts that fire during a track change
- swapArtworkSrc: only force layout reflow when keyframe is in flight;
drop the ?_=Date.now() cache-buster so identical artwork URLs reuse
the decoded bitmap; mini/glow imgs only re-set src when changed
- drop the fullscreen MutationObserver — fs-bloom-art is mirrored
directly from the artwork-swap path, eliminating the second blur paint
- updateProgress: skip text writes when the rounded second hasn't moved;
POSITION_INTERPOLATION_MS 100 → 250
- background.js: lift resizeBackgroundCanvas out of the rAF body, cache
step, accept new int-scaled wire format
CSS:
- spectrum bars use transform: scaleY(var(--bar-h-scale)) + transition
on transform; will-change updated to transform
- album-art-glow and fs-bloom-art switched to small-source-blur trick
(render at 20-25% size, scale 4-6×, lower blur radius) — visually
equivalent, ~10-25× cheaper repaint on track change
- drop unused transition: filter on .vinyl-stage #album-art
Backend (audio_analyzer.py, websocket_manager.py):
- pre-allocate windowed and cumsum buffers; replace
np.concatenate(([0.0], np.cumsum(...))) with cumsum[0]=0 +
np.cumsum(out=cumsum[1:]); float32 hanning window
- RMS via np.dot(mono, mono) — no astype copy, no ** temp
- int16 wire format (scale=1000) — smaller JSON, no Python float boxing
- versioned data + threading.Event so _audio_broadcast_loop is event-
driven (ev.wait + monotonic seq dedup) instead of polling on a timer
with the always-false `data is _last_data` identity check
ruff clean, pytest 7 passed / 3 numpy-skipped, esbuild bundle 113.6 kB.
348 lines
13 KiB
Python
348 lines
13 KiB
Python
"""WebSocket connection manager and status broadcaster."""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import time
|
|
from typing import Any, Callable, Coroutine
|
|
|
|
from fastapi import WebSocket
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ConnectionManager:
|
|
"""Manages WebSocket connections and broadcasts status updates."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the connection manager."""
|
|
self._active_connections: set[WebSocket] = set()
|
|
self._lock = asyncio.Lock()
|
|
self._last_status: dict[str, Any] | None = None
|
|
self._get_status_func: Callable[[], Coroutine[Any, Any, Any]] | None = None
|
|
self._broadcast_task: asyncio.Task | None = None
|
|
self._poll_interval: float = 0.5 # Internal poll interval for change detection
|
|
self._position_broadcast_interval: float = 5.0 # Send position updates every 5s during playback
|
|
self._last_broadcast_time: float = 0.0
|
|
self._running: bool = False
|
|
# Audio visualizer
|
|
self._visualizer_subscribers: set[WebSocket] = set()
|
|
self._audio_task: asyncio.Task | None = None
|
|
self._audio_analyzer = None
|
|
|
|
async def connect(self, websocket: WebSocket) -> None:
|
|
"""Accept a new WebSocket connection."""
|
|
await websocket.accept()
|
|
async with self._lock:
|
|
self._active_connections.add(websocket)
|
|
logger.info(
|
|
"WebSocket client connected. Total: %d", len(self._active_connections)
|
|
)
|
|
|
|
# Send current status immediately upon connection
|
|
status = self._last_status
|
|
if not status and self._get_status_func:
|
|
try:
|
|
result = await self._get_status_func()
|
|
status = result.model_dump()
|
|
self._last_status = status
|
|
except Exception as e:
|
|
logger.debug("Failed to fetch initial status: %s", e)
|
|
if status:
|
|
try:
|
|
await websocket.send_json({"type": "status", "data": status})
|
|
except Exception as e:
|
|
logger.debug("Failed to send initial status: %s", e)
|
|
|
|
async def disconnect(self, websocket: WebSocket) -> None:
|
|
"""Remove a WebSocket connection. Stops audio capture if last visualizer subscriber."""
|
|
should_stop = False
|
|
async with self._lock:
|
|
self._active_connections.discard(websocket)
|
|
was_subscriber = websocket in self._visualizer_subscribers
|
|
self._visualizer_subscribers.discard(websocket)
|
|
if was_subscriber and len(self._visualizer_subscribers) == 0:
|
|
should_stop = True
|
|
if should_stop:
|
|
await self._maybe_stop_capture()
|
|
logger.info(
|
|
"WebSocket client disconnected. Total: %d", len(self._active_connections)
|
|
)
|
|
|
|
async def broadcast(self, message: dict[str, Any]) -> None:
|
|
"""Broadcast a message to all connected clients concurrently."""
|
|
async with self._lock:
|
|
connections = list(self._active_connections)
|
|
|
|
if not connections:
|
|
return
|
|
|
|
async def _send(ws: WebSocket) -> WebSocket | None:
|
|
try:
|
|
await ws.send_json(message)
|
|
return None
|
|
except Exception as e:
|
|
logger.debug("Failed to send to client: %s", e)
|
|
return ws
|
|
|
|
results = await asyncio.gather(*(_send(ws) for ws in connections))
|
|
|
|
# Clean up disconnected clients
|
|
for ws in results:
|
|
if ws is not None:
|
|
await self.disconnect(ws)
|
|
|
|
async def broadcast_scripts_changed(self) -> None:
|
|
"""Notify all connected clients that scripts have changed."""
|
|
message = {"type": "scripts_changed", "data": {}}
|
|
await self.broadcast(message)
|
|
logger.info("Broadcast sent: scripts_changed")
|
|
|
|
async def broadcast_links_changed(self) -> None:
|
|
"""Notify all connected clients that links have changed."""
|
|
message = {"type": "links_changed", "data": {}}
|
|
await self.broadcast(message)
|
|
logger.info("Broadcast sent: links_changed")
|
|
|
|
async def subscribe_visualizer(self, websocket: WebSocket) -> None:
|
|
"""Subscribe a client to audio visualizer data. Starts capture on first subscriber."""
|
|
should_start = False
|
|
async with self._lock:
|
|
self._visualizer_subscribers.add(websocket)
|
|
if len(self._visualizer_subscribers) == 1 and self._audio_analyzer:
|
|
should_start = True
|
|
if should_start:
|
|
await self._maybe_start_capture()
|
|
logger.debug("Visualizer subscriber added. Total: %d", len(self._visualizer_subscribers))
|
|
|
|
async def unsubscribe_visualizer(self, websocket: WebSocket) -> None:
|
|
"""Unsubscribe a client from audio visualizer data. Stops capture on last subscriber."""
|
|
should_stop = False
|
|
async with self._lock:
|
|
self._visualizer_subscribers.discard(websocket)
|
|
if len(self._visualizer_subscribers) == 0:
|
|
should_stop = True
|
|
if should_stop:
|
|
await self._maybe_stop_capture()
|
|
logger.debug("Visualizer subscriber removed. Total: %d", len(self._visualizer_subscribers))
|
|
|
|
async def _maybe_start_capture(self) -> None:
|
|
"""Start audio capture if not already running (called on first subscriber)."""
|
|
if self._audio_analyzer and not self._audio_analyzer.running:
|
|
loop = asyncio.get_event_loop()
|
|
started = await loop.run_in_executor(None, self._audio_analyzer.start)
|
|
if started:
|
|
logger.info("Audio capture started (first subscriber)")
|
|
else:
|
|
logger.warning("Audio capture failed to start")
|
|
|
|
async def _maybe_stop_capture(self) -> None:
|
|
"""Stop audio capture if running (called when last subscriber leaves)."""
|
|
if self._audio_analyzer and self._audio_analyzer.running:
|
|
loop = asyncio.get_event_loop()
|
|
await loop.run_in_executor(None, self._audio_analyzer.stop)
|
|
logger.info("Audio capture stopped (no subscribers)")
|
|
|
|
async def start_audio_monitor(self, analyzer) -> None:
|
|
"""Register the audio analyzer. Capture starts on-demand when clients subscribe."""
|
|
self._audio_analyzer = analyzer
|
|
if analyzer and analyzer.available:
|
|
self._audio_task = asyncio.create_task(self._audio_broadcast_loop())
|
|
logger.info("Audio visualizer broadcast loop started (capture on-demand)")
|
|
|
|
async def stop_audio_monitor(self) -> None:
|
|
"""Stop audio frequency broadcasting."""
|
|
if self._audio_task:
|
|
self._audio_task.cancel()
|
|
try:
|
|
await self._audio_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._audio_task = None
|
|
|
|
async def _audio_broadcast_loop(self) -> None:
|
|
"""Background loop: read frequency data from analyzer and broadcast to subscribers.
|
|
|
|
Event-driven: blocks on the analyzer's data_event so it wakes up
|
|
exactly once per produced frame, instead of polling on a timer.
|
|
Backstop sleep applies when capture is idle / has no subscribers.
|
|
"""
|
|
from ..config import settings
|
|
idle_interval = 1.0 / max(1, settings.visualizer_fps)
|
|
# Bounded wait so we still notice subscribe/unsubscribe transitions.
|
|
wake_timeout = max(0.05, idle_interval)
|
|
loop = asyncio.get_event_loop()
|
|
|
|
last_seq = -1
|
|
|
|
while True:
|
|
try:
|
|
async with self._lock:
|
|
subscribers = list(self._visualizer_subscribers)
|
|
|
|
analyzer = self._audio_analyzer
|
|
if not subscribers or not analyzer or not analyzer.running:
|
|
await asyncio.sleep(idle_interval)
|
|
continue
|
|
|
|
# Wait off-loop for a fresh frame. The capture thread sets
|
|
# data_event after each FFT update; we clear it before the
|
|
# next wait so we never burn a wake on stale data.
|
|
ev = analyzer.data_event
|
|
|
|
def _wait() -> bool:
|
|
return ev.wait(wake_timeout)
|
|
|
|
got = await loop.run_in_executor(None, _wait)
|
|
if not got:
|
|
# Timeout — loop around to re-check subscriber state.
|
|
continue
|
|
ev.clear()
|
|
|
|
data, seq = analyzer.get_frequency_data_versioned()
|
|
if data is None or seq == last_seq:
|
|
continue
|
|
last_seq = seq
|
|
|
|
# Pre-serialize once for all subscribers (avoids per-client JSON encoding)
|
|
text = json.dumps({"type": "audio_data", "data": data}, separators=(',', ':'))
|
|
|
|
async def _send(ws: WebSocket) -> WebSocket | None:
|
|
try:
|
|
await ws.send_text(text)
|
|
return None
|
|
except Exception:
|
|
return ws
|
|
|
|
results = await asyncio.gather(*(_send(ws) for ws in subscribers))
|
|
|
|
failed = [ws for ws in results if ws is not None]
|
|
for ws in failed:
|
|
await self.disconnect(ws)
|
|
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as e:
|
|
logger.error("Error in audio broadcast: %s", e)
|
|
await asyncio.sleep(idle_interval)
|
|
|
|
def status_changed(
|
|
self, old: dict[str, Any] | None, new: dict[str, Any]
|
|
) -> bool:
|
|
"""Detect if media status has meaningfully changed.
|
|
|
|
Position is NOT included for normal playback (let HA interpolate).
|
|
But seeks (large unexpected jumps) are detected.
|
|
"""
|
|
if old is None:
|
|
return True
|
|
|
|
# Fields to compare for changes (NO position - let HA interpolate)
|
|
significant_fields = [
|
|
"state",
|
|
"title",
|
|
"artist",
|
|
"album",
|
|
"volume",
|
|
"muted",
|
|
"duration",
|
|
"source",
|
|
"album_art_url",
|
|
]
|
|
|
|
for field in significant_fields:
|
|
if old.get(field) != new.get(field):
|
|
return True
|
|
|
|
# Detect seeks - large position jumps that aren't normal playback
|
|
old_pos = old.get("position") or 0
|
|
new_pos = new.get("position") or 0
|
|
pos_diff = new_pos - old_pos
|
|
|
|
# During playback, position should increase by ~0.5s (our poll interval)
|
|
# A seek is when position jumps backwards OR forward by more than expected
|
|
if new.get("state") == "playing":
|
|
# Backward seek or forward jump > 3s indicates seek
|
|
if pos_diff < -1.0 or pos_diff > 3.0:
|
|
return True
|
|
else:
|
|
# When paused, any significant position change is a seek
|
|
if abs(pos_diff) > 1.0:
|
|
return True
|
|
|
|
return False
|
|
|
|
async def start_status_monitor(
|
|
self,
|
|
get_status_func: Callable[[], Coroutine[Any, Any, Any]],
|
|
) -> None:
|
|
"""Start the background status monitoring loop."""
|
|
if self._running:
|
|
return
|
|
|
|
self._get_status_func = get_status_func
|
|
self._running = True
|
|
self._broadcast_task = asyncio.create_task(
|
|
self._status_monitor_loop(get_status_func)
|
|
)
|
|
logger.info("WebSocket status monitor started")
|
|
|
|
async def stop_status_monitor(self) -> None:
|
|
"""Stop the background status monitoring loop."""
|
|
self._running = False
|
|
if self._broadcast_task:
|
|
self._broadcast_task.cancel()
|
|
try:
|
|
await self._broadcast_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
logger.info("WebSocket status monitor stopped")
|
|
|
|
async def _status_monitor_loop(
|
|
self,
|
|
get_status_func: Callable[[], Coroutine[Any, Any, Any]],
|
|
) -> None:
|
|
"""Background loop that polls for status changes and broadcasts."""
|
|
while self._running:
|
|
try:
|
|
# Only poll if we have connected clients
|
|
async with self._lock:
|
|
has_clients = len(self._active_connections) > 0
|
|
|
|
if not has_clients:
|
|
await asyncio.sleep(2.0) # Sleep longer when no clients connected
|
|
continue
|
|
|
|
status = await get_status_func()
|
|
status_dict = status.model_dump()
|
|
|
|
# Only broadcast on actual state changes
|
|
# Let HA handle position interpolation during playback
|
|
if self.status_changed(self._last_status, status_dict):
|
|
self._last_status = status_dict
|
|
self._last_broadcast_time = time.time()
|
|
await self.broadcast(
|
|
{"type": "status_update", "data": status_dict}
|
|
)
|
|
logger.debug("Broadcast sent: status change")
|
|
else:
|
|
# Update cached status even without broadcast
|
|
self._last_status = status_dict
|
|
|
|
await asyncio.sleep(self._poll_interval)
|
|
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as e:
|
|
logger.error("Error in status monitor: %s", e)
|
|
await asyncio.sleep(self._poll_interval)
|
|
|
|
@property
|
|
def client_count(self) -> int:
|
|
"""Return the number of connected clients."""
|
|
return len(self._active_connections)
|
|
|
|
|
|
# Global instance
|
|
ws_manager = ConnectionManager()
|