ytaudio: accurate active queue system

Replace the drained asyncio.Queue with an explicit, always-inspectable
model: GuildPlayer.pending (waiting tracks) + current (now playing).
- queue command shows Now playing plus numbered Up next list
- play reports queue position when adding behind an active track
- stop/leave use clear_pending(); skip relies on the TRACK_END event to
  advance exactly one track (no double-skip)
- queue alias q, nowplaying alias np
This commit is contained in:
owen
2026-09-17 17:44:50 -05:00
parent 70da956fd4
commit b8eb40bc64
+57 -36
View File
@@ -75,38 +75,58 @@ class QueuedTrack:
class GuildPlayer: class GuildPlayer:
"""Per-guild queue + player-loop state. One per guild.""" """Per-guild queue + player-loop state. One per guild.
The queue model is explicit and always inspectable:
* ``pending`` is the list of tracks waiting to play (index 0 is next).
* ``current`` is the track playing right now (or None).
Commands read these directly, so nowplaying/queue always reflect reality.
"""
def __init__(self, cog: "YTAudio", guild_id: int): def __init__(self, cog: "YTAudio", guild_id: int):
self.cog = cog self.cog = cog
self.guild_id = guild_id self.guild_id = guild_id
self.queue: "asyncio.Queue[Optional[QueuedTrack]]" = asyncio.Queue() self.pending: List[QueuedTrack] = []
self.current: Optional[QueuedTrack] = None self.current: Optional[QueuedTrack] = None
self.task: Optional[asyncio.Task] = None self.task: Optional[asyncio.Task] = None
self.next_event = asyncio.Event() self.next_event = asyncio.Event() # fired when the current track ends
self.wakeup = asyncio.Event() # fired when a track is added
self.volume = 100 self.volume = 100
def start(self): def start(self):
if self.task is None or self.task.done(): if self.task is None or self.task.done():
self.task = asyncio.create_task(self._player_loop()) self.task = asyncio.create_task(self._player_loop())
def enqueue(self, track: "QueuedTrack"):
self.pending.append(track)
self.wakeup.set()
def clear_pending(self):
cleared = self.pending
self.pending = []
for t in cleared:
self.cog._cleanup_file(t)
async def _player_loop(self): async def _player_loop(self):
while True: while True:
if not self.pending:
self.wakeup.clear()
if not self.pending:
try:
await self.wakeup.wait()
except asyncio.CancelledError:
return
continue
track = self.pending.pop(0)
self.current = track
self.next_event.clear() self.next_event.clear()
try:
track = await self.queue.get()
except asyncio.CancelledError:
return
if track is None: # shutdown sentinel
return
self.current = track
try: try:
await self.cog._play_track(self.guild_id, track, self.next_event) await self.cog._play_track(self.guild_id, track, self.next_event)
# Wait until the track-end handler fires next_event
await self.next_event.wait() await self.next_event.wait()
except asyncio.CancelledError: except asyncio.CancelledError:
self.cog._cleanup_file(track) self.cog._cleanup_file(track)
self.current = None
return return
except Exception: except Exception:
log.exception("Error playing track in guild %s", self.guild_id) log.exception("Error playing track in guild %s", self.guild_id)
@@ -348,23 +368,26 @@ class YTAudio(commands.Cog):
gp = self._get_gp(ctx.guild.id) gp = self._get_gp(ctx.guild.id)
gp.volume = await self.config.guild(ctx.guild).volume() gp.volume = await self.config.guild(ctx.guild).volume()
track = QueuedTrack(query, ctx.author, ctx.channel) track = QueuedTrack(query, ctx.author, ctx.channel)
already_active = gp.current is not None or not gp.queue.empty() already_active = gp.current is not None or bool(gp.pending)
await gp.queue.put(track) gp.enqueue(track)
if already_active: if already_active:
await ctx.send("Added to queue: {}".format(query)) position = len(gp.pending)
await ctx.send("Added to queue (position {}): {}".format(position, query))
@commands.command() @commands.command()
@commands.guild_only() @commands.guild_only()
async def skip(self, ctx: commands.Context): async def skip(self, ctx: commands.Context):
"""Skip the current track.""" """Skip the current track."""
gp = self.players.get(ctx.guild.id)
if gp is None or gp.current is None:
return await ctx.send("Nothing is playing.")
try: try:
player = lavalink.get_player(ctx.guild.id) player = lavalink.get_player(ctx.guild.id)
except Exception: except Exception:
return await ctx.send("Nothing is playing.") return await ctx.send("Nothing is playing.")
# Stopping fires TRACK_END, which the event handler uses to advance
# the queue. Do not set next_event here or it would skip two tracks.
await player.stop() await player.stop()
gp = self.players.get(ctx.guild.id)
if gp:
gp.next_event.set()
await ctx.send("Skipped.") await ctx.send("Skipped.")
@commands.command() @commands.command()
@@ -373,12 +396,7 @@ class YTAudio(commands.Cog):
"""Stop playback and clear the queue.""" """Stop playback and clear the queue."""
gp = self.players.get(ctx.guild.id) gp = self.players.get(ctx.guild.id)
if gp: if gp:
while not gp.queue.empty(): gp.clear_pending()
try:
t = gp.queue.get_nowait()
self._cleanup_file(t)
except asyncio.QueueEmpty:
break
gp.next_event.set() gp.next_event.set()
try: try:
player = lavalink.get_player(ctx.guild.id) player = lavalink.get_player(ctx.guild.id)
@@ -409,20 +427,27 @@ class YTAudio(commands.Cog):
await player.pause(False) await player.pause(False)
await ctx.send("Resumed.") await ctx.send("Resumed.")
@commands.command(name="queue") @commands.command(name="queue", aliases=["q"])
@commands.guild_only() @commands.guild_only()
async def queue_cmd(self, ctx: commands.Context): async def queue_cmd(self, ctx: commands.Context):
"""Show the queue.""" """Show the current track and the queue."""
gp = self.players.get(ctx.guild.id) gp = self.players.get(ctx.guild.id)
if not gp or (gp.current is None and gp.queue.empty()): if not gp or (gp.current is None and not gp.pending):
return await ctx.send("The queue is empty.") return await ctx.send("The queue is empty.")
lines = [] lines = []
if gp.current: if gp.current:
lines.append("Now playing: {}".format(gp.current.title or gp.current.query)) lines.append("Now playing: {}".format(gp.current.title or gp.current.query))
pending: List[QueuedTrack] = list(gp.queue._queue) # snapshot if gp.pending:
for i, t in enumerate(pending, 1): lines.append("Up next:")
lines.append(f"{i}. {t.title or t.query}") for i, t in enumerate(gp.pending, 1):
await ctx.send("\n".join(lines[:20])) lines.append("{}. {}".format(i, t.title or t.query))
else:
lines.append("Queue is empty.")
# Keep within Discord message limits.
out = "\n".join(lines)
if len(out) > 1900:
out = out[:1900] + "\n..."
await ctx.send(out)
@commands.command(name="nowplaying", aliases=["np"]) @commands.command(name="nowplaying", aliases=["np"])
@commands.guild_only() @commands.guild_only()
@@ -462,11 +487,7 @@ class YTAudio(commands.Cog):
"""Leave the voice channel and clear state.""" """Leave the voice channel and clear state."""
gp = self.players.get(ctx.guild.id) gp = self.players.get(ctx.guild.id)
if gp: if gp:
while not gp.queue.empty(): gp.clear_pending()
try:
self._cleanup_file(gp.queue.get_nowait())
except asyncio.QueueEmpty:
break
gp.next_event.set() gp.next_event.set()
try: try:
player = lavalink.get_player(ctx.guild.id) player = lavalink.get_player(ctx.guild.id)