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.
9.7 kB · 268 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269import asyncioimport randomfrom datetime import datetime, timedelta, timezone
from jam_session.models import QueueState, Track
ADJECTIVES = [ "Amazing", "Bold", "Cosmic", "Dapper", "Electric", "Funky", "Golden", "Happy", "Icy", "Jazzy", "Kind", "Loud", "Mellow", "Neon", "Orange", "Purple", "Quiet", "Rusty", "Sunny", "Turbo",]NOUNS = [ "Badger", "Cobra", "Dolphin", "Eagle", "Falcon", "Grizzly", "Hawk", "Iguana", "Jaguar", "Koala", "Lion", "Moose", "Narwhal", "Otter", "Panda", "Quail", "Raven", "Shark", "Tiger", "Wolf",]
def generate_nickname() -> str: adj = random.choice(ADJECTIVES) noun = random.choice(NOUNS) num = random.randint(1, 99) return f"{adj}-{noun}-{num}"
class NicknameTakenError(Exception): pass
class SessionManager: def __init__(self): self._queue: list[Track] = [] self._current_track: Track | None = None self.paused: bool = False self.playback_time: float = 0.0 self._lock: asyncio.Lock = asyncio.Lock() self._nicknames: dict[str, str] = {} self._nickname_index: dict[str, str] = {} self._held_nicknames: dict[str, datetime] = {} self._votes: dict[str, set[str]] = {} self._manual_order: bool = False self._skip_votes: set[str] = set() self._skip_triggered: bool = False self._skip_cooldown_until: float = 0.0 self._skip_denominator: int = 0 self._cooldown_handle: asyncio.TimerHandle | None = None self._host_token: str | None = None self._host_conn_id: str | None = None
@staticmethod def _compute_skip_threshold(nickname_count: int, threshold_pct: int) -> int: if nickname_count == 1: return 1 if nickname_count == 2: return 2 if nickname_count == 3: return 2 return int(nickname_count * threshold_pct / 100) + 1
async def add_track(self, track: Track) -> Track: async with self._lock: self._queue.append(track) return track
async def remove_track(self, track_id: str) -> Track | None: async with self._lock: for votes in self._votes.values(): votes.discard(track_id) for i, t in enumerate(self._queue): if t.id == track_id: return self._queue.pop(i) return None
async def reorder(self, new_order: list[str]) -> list[Track]: async with self._lock: ordered = [t for tid in new_order for t in self._queue if t.id == tid] if len(ordered) != len(self._queue): raise ValueError("new_order does not match queue content") self._queue = ordered return self._queue
def get_state(self) -> QueueState: return QueueState( queue=list(self._queue), current_track=self._current_track, paused=self.paused, playback_time=self.playback_time, nicknames=dict(self._nicknames), held_nicknames={ n: exp.isoformat() for n, exp in self._held_nicknames.items() }, )
async def set_nickname(self, conn_id: str, nickname: str) -> str: async with self._lock: lower = nickname.lower() if lower in self._nickname_index: existing_conn = self._nickname_index[lower] if existing_conn != conn_id: raise NicknameTakenError( f"Nickname '{nickname}' is already taken" ) old_nickname = self._nicknames.get(conn_id) if old_nickname: old_lower = old_nickname.lower() if old_lower in self._nickname_index and self._nickname_index[old_lower] == conn_id: del self._nickname_index[old_lower] self._nicknames[conn_id] = nickname self._nickname_index[lower] = conn_id self._held_nicknames.pop(nickname, None) return nickname
def get_nickname(self, conn_id: str) -> str | None: return self._nicknames.get(conn_id)
def get_all_nicknames(self) -> dict[str, str]: return dict(self._nicknames)
async def hold_nickname(self, conn_id: str): async with self._lock: nickname = self._nicknames.pop(conn_id, None) if nickname: self._held_nicknames[nickname] = datetime.now(timezone.utc) + timedelta(seconds=30) lower = nickname.lower() self._nickname_index.pop(lower, None)
async def release_expired_holds(self): async with self._lock: now = datetime.now(timezone.utc) expired = [n for n, exp in self._held_nicknames.items() if now >= exp] for nickname in expired: del self._held_nicknames[nickname]
async def toggle_vote(self, conn_id: str, track_id: str) -> tuple[int, bool]: async with self._lock: track_found = False for t in self._queue: if t.id == track_id: track_found = True break if not track_found: raise ValueError(f"Track {track_id} not found in queue")
if conn_id not in self._votes: self._votes[conn_id] = set()
was_upvoted = False if track_id in self._votes[conn_id]: self._votes[conn_id].discard(track_id) was_upvoted = False else: self._votes[conn_id].add(track_id) was_upvoted = True
if not self._manual_order: self._sort_by_votes()
count = sum(1 for votes in self._votes.values() if track_id in votes) return count, was_upvoted
def _sort_by_votes(self): self._queue.sort(key=lambda t: ( -sum(1 for votes in self._votes.values() if t.id in votes), t.added_at ))
def get_votes_for(self, conn_id: str) -> list[str]: return list(self._votes.get(conn_id, set()))
def get_vote_counts(self) -> dict[str, int]: return { t.id: sum(1 for votes in self._votes.values() if t.id in votes) for t in self._queue }
def set_manual_order(self, active: bool): self._manual_order = active
def is_manual_order(self) -> bool: return self._manual_order
def set_host_token(self, token: str): self._host_token = token
def get_host_token(self) -> str | None: return self._host_token
def validate_host_token(self, token: str) -> bool: return self._host_token is not None and token == self._host_token
def set_host_conn_id(self, conn_id: str): self._host_conn_id = conn_id
def get_host_conn_id(self) -> str | None: return self._host_conn_id
def is_host(self, conn_id: str) -> bool: return self._host_conn_id is not None and self._host_conn_id == conn_id
async def advance(self) -> Track | None: async with self._lock: if self._queue: self._current_track = self._queue.pop(0) else: self._current_track = None self.playback_time = 0.0 return self._current_track
async def toggle_skip_vote(self, conn_id: str, threshold_pct: int) -> dict: async with self._lock: now = asyncio.get_running_loop().time() if now < self._skip_cooldown_until: return { "vote_count": len(self._skip_votes), "threshold": self._compute_skip_threshold(self._skip_denominator, threshold_pct), "triggered": False, "cooldown_remaining": int(self._skip_cooldown_until - now), }
if conn_id in self._skip_votes: self._skip_votes.discard(conn_id) else: self._skip_votes.add(conn_id)
threshold = self._compute_skip_threshold(self._skip_denominator, threshold_pct) triggered = len(self._skip_votes) >= threshold and not self._skip_triggered
if triggered: self._skip_triggered = True
return { "vote_count": len(self._skip_votes), "threshold": threshold, "triggered": triggered, "cooldown_remaining": 0, }
def get_skip_state(self, conn_id: str, threshold_pct: int) -> dict: now = asyncio.get_running_loop().time() cooldown_remaining = 0 if now < self._skip_cooldown_until: cooldown_remaining = int(self._skip_cooldown_until - now) return { "my_skip_vote": conn_id in self._skip_votes, "skip_vote_count": len(self._skip_votes), "skip_threshold": self._compute_skip_threshold(self._skip_denominator, threshold_pct), "skip_cooldown_remaining": cooldown_remaining, }
def reset_skip_votes(self): self._skip_votes.clear() self._skip_triggered = False self._skip_denominator = len(self._nicknames)
def start_skip_cooldown(self, seconds: int): if self._cooldown_handle is not None: self._cooldown_handle.cancel() self._cooldown_handle = None
loop = asyncio.get_running_loop() self._skip_cooldown_until = loop.time() + seconds
def _on_cooldown_end(): self._cooldown_handle = None
self._cooldown_handle = loop.call_later(seconds, _on_cooldown_end)