Something went wrong. Try again.
A local, collaborative music player (main-branch-only mirror of forge.ejuarezg.com/ejuarezg/jam-session)
Something went wrong. Try again.
6.1 kB · 186 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187import uuid as _uuid
from fastapi import WebSocket
from jam_session.models import QueueState
class ConnectionManager: def __init__(self): self.active_connections: dict[str, WebSocket] = {}
async def connect(self, websocket: WebSocket) -> str: await websocket.accept() conn_id = f"conn_{_uuid.uuid4().hex[:8]}" self.active_connections[conn_id] = websocket return conn_id
def disconnect(self, conn_id: str): self.active_connections.pop(conn_id, None)
async def broadcast(self, message: dict): dead = [] for conn_id, ws in self.active_connections.items(): try: await ws.send_json(message) except Exception: dead.append(conn_id)
for conn_id in dead: self.active_connections.pop(conn_id, None)
async def send_to(self, conn_id: str, message: dict): ws = self.active_connections.get(conn_id) if ws is None: return
try: await ws.send_json(message) except Exception: self.active_connections.pop(conn_id, None)
async def send_sync( self, websocket: WebSocket, queue_state: QueueState, playback_time: float, conn_id: str | None = None, host_conn_id: str | None = None, vote_counts: dict[str, int] | None = None, my_votes: list[str] | None = None, skip_state: dict | None = None, settings: dict | None = None, server_url: str | None = None, ): payload: dict = { "queue_state": queue_state.model_dump(mode="json"), "playback_time": playback_time, "connected_count": len(self.active_connections), }
if server_url is not None: payload["server_url"] = server_url
if conn_id is not None: payload["conn_id"] = conn_id payload["nickname"] = queue_state.nicknames.get(conn_id) payload["is_host"] = (host_conn_id is not None and conn_id == host_conn_id)
if vote_counts is not None: payload["vote_counts"] = vote_counts if my_votes is not None: payload["my_votes"] = my_votes
if skip_state is not None: payload["my_skip_vote"] = skip_state["my_skip_vote"] payload["skip_vote_count"] = skip_state["skip_vote_count"] payload["skip_threshold"] = skip_state["skip_threshold"] payload["skip_cooldown_remaining"] = skip_state["skip_cooldown_remaining"]
if settings is not None: payload["settings"] = { "skip_threshold_pct": settings["skip_threshold_pct"], "auto_persist": settings["auto_persist"], "skip_cooldown_seconds": settings["skip_cooldown_seconds"], }
await websocket.send_json({"type": "server.sync", "payload": payload})
async def broadcast_queue_delta( self, action: str, track: dict | None = None, index: int | None = None ): await self.broadcast( { "type": "server.queue_update", "payload": {"action": action, "track": track, "index": index}, } )
async def broadcast_playback_update( self, track: dict | None = None, paused: bool = False ): await self.broadcast( { "type": "server.playback_update", "payload": {"current_track": track, "paused": paused}, } )
async def broadcast_time_update(self, time: float): await self.broadcast( {"type": "server.time_update", "payload": {"playback_time": time}} )
async def send_error(self, websocket: WebSocket, error: str): try: await websocket.send_json( {"type": "server.error", "payload": {"error": error}} ) except Exception: pass
@property def connected_count(self) -> int: return len(self.active_connections)
async def broadcast_nicknames_update(self, nicknames: dict[str, str]): await self.broadcast( { "type": "server.nicknames_update", "payload": {"nicknames": nicknames}, } )
async def send_nickname_error(self, websocket: WebSocket | None, error: str): if websocket is None: return try: await websocket.send_json( { "type": "server.nickname_error", "payload": {"error": error}, } ) except Exception: pass
async def broadcast_votes_update(self, vote_counts: dict[str, int]): await self.broadcast( { "type": "server.votes_update", "payload": {"vote_counts": vote_counts}, } )
async def broadcast_skip_update(self, state: dict): await self.broadcast( { "type": "server.skip_update", "payload": { "vote_count": state["vote_count"], "threshold": state["threshold"], "triggered": state["triggered"], "cooldown_remaining": state["cooldown_remaining"], }, } )
async def broadcast_settings_update(self, config: dict): await self.broadcast( { "type": "server.settings_update", "payload": { "skip_threshold_pct": config["skip_threshold_pct"], "auto_persist": config["auto_persist"], "skip_cooldown_seconds": config["skip_cooldown_seconds"], }, } )
async def broadcast_queue_sort(self, ordered_ids: list[str]): await self.broadcast( { "type": "server.queue_sort", "payload": {"order": ordered_ids}, } )