Merge pull request #48 from r-Norge/feature/more_type_hints

Add more typing information. Split user queue with index into separate functions

musicbot/cogs/music/__init__.py

@@ -4,11 +4,13 @@ from discord.ext import commands, tasks
 from lavalink.events import (
     NodeChangedEvent, NodeConnectedEvent, NodeDisconnectedEvent, PlayerUpdateEvent, QueueEndEvent, TrackEndEvent,
     TrackStartEvent, TrackStuckEvent)
+from lavalink.models import AudioTrack
 
 import asyncio
 import json
 import re
 import urllib
+from typing import List
 
 from bs4 import BeautifulSoup
 
@@ -32,8 +34,7 @@ class Music(commands.Cog):
 
         self.bot = bot
         self.leave_timer.start()
-        self.logger = self.bot.main_logger.bot_logger.getChild("Errors")
-        bot.lavalink.add_event_hook(self.track_hook)
+        self.bot.lavalink.add_event_hook(self.track_hook)
 
     async def cog_check(self, ctx):
         if not ctx.guild:
@@ -279,12 +280,12 @@ class Music(commands.Cog):
         player: MixPlayer = self.bot.lavalink.player_manager.get(ctx.guild.id)
 
         # We create a new selector for each selection
-        def move_selector(ctx, queue, title, first: bool):
+        def move_selector(ctx, queue: List[AudioTrack], title: str, first: bool):
             async def return_track(track):
                 return track
 
             choices = []
-            for index, (track, _) in enumerate(queue, start=1):
+            for index, track in enumerate(queue, start=1):
                 blank = " " * 6
                 if first:
                     prefix = f'`{blank}`\n`{index:<6}`'
@@ -301,7 +302,7 @@ class Music(commands.Cog):
         page = 0
         while True:
             # Prompt the user for which track to move
-            selector = move_selector(ctx, player.user_queue(ctx.author.id, dual=True), "{moved.choose_pos}", True)
+            selector = move_selector(ctx, player.user_queue(ctx.author.id), "{moved.choose_pos}", True)
             message, track_to_move = await selector.start_scrolling(ClearOn.Timeout, message, page)
             page = selector.page_number
 
@@ -309,7 +310,7 @@ class Music(commands.Cog):
                 break
 
             # Prompt the user for where to move it
-            selector = move_selector(ctx, player.user_queue(ctx.author.id, dual=True), "{moved.choose_song}", False)
+            selector = move_selector(ctx, player.user_queue(ctx.author.id), "{moved.choose_song}", False)
             message, track_to_replace = await selector.start_scrolling(ClearOn.Timeout, message, page)
             page = selector.page_number
 
@@ -320,11 +321,10 @@ class Music(commands.Cog):
             # one in the queue. Find the index of both songs in the current queue
 
             # Keep updating the user queue in case it changes while the user is selecting songs
-            user_queue = player.user_queue(ctx.author.id, dual=True)
-            user_queue_identifiers = [track for (track, _) in user_queue]
+            user_queue = player.user_queue(ctx.author.id)
             try:
-                pos_initial = user_queue_identifiers.index(track_to_move[0])
-                pos_final = user_queue_identifiers.index(track_to_replace[0])
+                pos_initial = user_queue.index(track_to_move[0])
+                pos_final = user_queue.index(track_to_replace[0])
                 player.move_user_track(ctx.author.id, pos_initial, pos_final)
             except ValueError:
                 # Track not found in queue. Queue could have changed during selection
@@ -340,7 +340,7 @@ class Music(commands.Cog):
     async def _remove(self, ctx):
         player: MixPlayer = self.bot.lavalink.player_manager.get(ctx.guild.id)
 
-        user_queue = player.user_queue(ctx.author.id, dual=True)
+        user_queue: List[AudioTrack] = player.user_queue(ctx.author.id)
 
         tracks_to_remove = []
 
@@ -363,7 +363,7 @@ class Music(commands.Cog):
             return inner
 
         selector_buttons = []
-        for index, (track, _) in enumerate(user_queue, start=1):
+        for index, track in enumerate(user_queue, start=1):
             selector_buttons.append(
                 SelectorItem(f'`{index}` [{track.title}]({track.uri})', str(index),
                              add_change_button_color(update_remove_list, tracks_to_remove, track)))
@@ -377,13 +377,18 @@ class Music(commands.Cog):
             paginator = TextPaginator(color=ctx.me.color, title="Removed")
 
             # If the user spent a long time selecting songs the queue might have changed
-            user_queue = player.user_queue(ctx.author.id, dual=False)
+            user_queue = player.user_queue(ctx.author.id)
             track_indexes_to_remove = [i for i, track in enumerate(user_queue) if track in tracks_to_remove]
 
             # Sort in reverse order since songs shift index when deleting
+            tracks_removed = []
             for track in sorted(track_indexes_to_remove, reverse=True):
                 if removed_track := player.remove_user_track(ctx.author.id, track):
-                    paginator.add_line(f"{removed_track.title}")
+                    tracks_removed.append(f"{removed_track.title}")
+
+            # We want to display the removed tracks in the original queue order, so reverse the list once again
+            for removed in reversed(tracks_removed):
+                paginator.add_line(removed)
 
             paginator.close_page()
             scroller = Scroller(ctx, paginator)

musicbot/utils/mixplayer/mixqueue.py

@@ -2,7 +2,7 @@ import logging
 from collections import OrderedDict, deque
 from itertools import chain, cycle, islice
 from random import shuffle
-from typing import Generic, Iterable, Iterator, List, Tuple, TypeVar, Union
+from typing import Generic, Iterable, Iterator, List, Optional, Tuple, TypeVar
 
 # Would like to ensure the T has a "requester" attribute, but don't know if that is possible
 T = TypeVar('T')
@@ -80,7 +80,7 @@ class MixQueue(Generic[T]):
         combined = zip(queue, pos)
         return list(combined)
 
-    def pop_first(self) -> Union[T, None]:
+    def pop_first(self) -> Optional[T]:
         if self.priority_queue:
             next_track = self.priority_queue.pop(0)
             self._history.append(next_track)
@@ -110,7 +110,7 @@ class MixQueue(Generic[T]):
         except KeyError:
             pass
 
-    def add_track(self, requester: int, track: T, pos: Union[int, None] = None) -> Tuple[T, int, int]:
+    def add_track(self, requester: int, track: T, pos: Optional[int] = None) -> Tuple[T, int, int]:
         user_queue = self.queues.get(requester)
         if user_queue is None:
             self.queues[requester] = [track]
@@ -138,7 +138,7 @@ class MixQueue(Generic[T]):
             else:
                 self.queues.pop(requester)
 
-    def remove_user_track(self, requester: int, pos: int) -> Union[None, T]:
+    def remove_user_track(self, requester: int, pos: int) -> Optional[T]:
         user_queue = self.queues.get(requester)
         if user_queue is not None:
             if pos < len(user_queue):
@@ -149,7 +149,7 @@ class MixQueue(Generic[T]):
                 self._clear_empty()
                 return track
 
-    def remove_global_track(self, pos: int) -> Union[T, None]:
+    def remove_global_track(self, pos: int) -> Optional[T]:
         # Get the actual index of the song
         index = list(self).index(self.get_queue()[pos])
 
@@ -217,7 +217,7 @@ class MixQueue(Generic[T]):
                     globpos += 1
         return globpos
 
-    def _glob_to_loc(self, pos: int) -> Tuple[Union[T, None], Union[int, None]]:
+    def _glob_to_loc(self, pos: int) -> Tuple[Optional[T], Optional[int]]:
         try:
             song = next(islice(self, pos, pos + 1))
         except (ValueError, StopIteration):
@@ -234,7 +234,7 @@ class MixQueue(Generic[T]):
         return None, None
 
     @property
-    def first_queue(self) -> Union[int, None]:
+    def first_queue(self) -> Optional[int]:
         try:
             return list(self.queues.keys())[0]
         except IndexError:

musicbot/utils/mixplayer/player.py

@@ -5,8 +5,7 @@ from lavalink import AudioTrack, DefaultPlayer, Node
 from lavalink.events import QueueEndEvent, TrackEndEvent, TrackExceptionEvent, TrackStartEvent, TrackStuckEvent
 from lavalink.filters import Equalizer, Timescale
 
-import typing
-from typing import Union
+from typing import Dict, List, Optional, Set, Tuple
 
 from .mixqueue import MixQueue
 
@@ -15,13 +14,13 @@ class MixPlayer(DefaultPlayer):
     def __init__(self, guild_id: int, node: Node):
         super().__init__(guild_id, node)
 
-        self.queue: MixQueue = MixQueue()
+        self.queue: MixQueue[AudioTrack] = MixQueue()
 
-        self.listeners = set()
-        self.voteables = {}
-        self.skip_voters = set()
-        self.boosted = False
-        self.nightcore_enabled = False
+        self.listeners: Set[int] = set()
+        self.voteables: Dict[int, Set[int]] = {}
+        self.skip_voters: Set[int] = set()
+        self.boosted: bool = False
+        self.nightcore_enabled: bool = False
         bands = [
             (0, 0.15),
             (1, 0.15),
@@ -31,17 +30,17 @@ class MixPlayer(DefaultPlayer):
             (5, -0.1),
             (6, -0.05)
         ]
-        self.bass_boost_filter = Equalizer()
+        self.bass_boost_filter: lavalink.Filter = Equalizer()
         self.bass_boost_filter.update(bands=bands)
 
-        self.nightcore_filter = Timescale()
+        self.nightcore_filter: lavalink.Filter = Timescale()
         self.nightcore_filter.update(speed=1.25, pitch=1.25)
 
-    def add(self, requester: int, track: typing.Union[dict, AudioTrack], pos: int = None):
+    def add(self, requester: int, track: AudioTrack, pos: Optional[int] = None) -> Tuple[AudioTrack, int, int]:
         """ Adds a track to the queue. """
         return self.queue.add_track(requester, track, pos)
 
-    def add_next(self, requester: int, track: typing.Union[dict, AudioTrack], pos: int = None):
+    def add_priority(self, track: AudioTrack):
         """ Adds a track to beginning of the queue """
         self.queue.add_next_track(track)
 
@@ -52,11 +51,11 @@ class MixPlayer(DefaultPlayer):
     def remove_user_queue(self, requester: int):
         self.queue.remove_user_queue(requester)
 
-    def remove_user_track(self, requester: int, pos: int) -> Union[None, AudioTrack]:
+    def remove_user_track(self, requester: int, pos: int) -> Optional[AudioTrack]:
         """ Removes the song at <pos> from the queue of requester """
         return self.queue.remove_user_track(requester, pos)
 
-    def remove_global_track(self, pos: int):
+    def remove_global_track(self, pos: int) -> Optional[AudioTrack]:
         """ Removes the song at <pos> in the global queue """
         return self.queue.remove_global_track(pos)
 
@@ -64,18 +63,20 @@ class MixPlayer(DefaultPlayer):
         """ Randomly reorders the queue of requester """
         self.queue.shuffle_user_queue(requester)
 
-    def user_queue(self, user: int, dual: bool = False):
-        if dual:
-            return self.queue.get_user_queue_with_index(user)
+    def user_queue(self, user: int) -> List[AudioTrack]:
         return self.queue.get_user_queue(user)
 
-    def global_queue(self):
+    def user_queue_with_global_index(self, user: int) -> List[Tuple[AudioTrack, int]]:
+        return self.queue.get_user_queue_with_index(user)
+
+    def global_queue(self) -> List[AudioTrack]:
         return self.queue.get_queue()
 
-    def get_history(self):
+    def get_history(self) -> List[AudioTrack]:
         return self.queue.history
 
-    def queue_duration(self, include_current: bool = False, member: discord.Member = None, end_pos: int = None):
+    def queue_duration(self, include_current: bool = False,
+                       member: Optional[discord.Member] = None, end_pos: Optional[int] = None):
         duration = 0
         queue = self.user_queue(member.id) if member else self.queue
 
@@ -84,14 +85,13 @@ class MixPlayer(DefaultPlayer):
                 break
             duration += int(track.duration)
 
-        remaining = self.current.duration - self.position
         if include_current:
             if self.current:
-                remaining = self.current.duration - self.position
+                remaining = self.current.duration - int(self.position)
                 return lavalink.utils.format_time(duration + remaining)
         return lavalink.utils.format_time(duration)
 
-    async def play(self, track: AudioTrack = None, start_time: int = 0):
+    async def play(self, track: Optional[AudioTrack] = None, start_time: int = 0):
 
         self.current = None
         self.last_update = 0
@@ -107,10 +107,12 @@ class MixPlayer(DefaultPlayer):
                 await self.node._dispatch_event(QueueEndEvent(self))
                 return
             else:
+                # At this point track will not be None, as the queue is not empty
                 track = self.queue.pop_first()
 
         self.current = track
-        if track.track is None:
+        if track is None or track.track is None:
+            # Ignore, if the queue was empty we would have dispatched the event already
             return
         await self.node._send(op='play', guildId=str(self.guild_id),
                               track=track.track, startTime=start_time)
@@ -131,7 +133,7 @@ class MixPlayer(DefaultPlayer):
         self.clear_votes()
 
     def update_listeners(self, member, voice_state):
-        if self.is_connected:
+        if self.channel_id is not None:
             vc = int(self.channel_id)
             if voice_state.channel is None or voice_state.channel.id != vc:
                 self.listeners.discard(member)
@@ -153,7 +155,7 @@ class MixPlayer(DefaultPlayer):
             self.voteables[category].add(member)
 
     def remove_member_votes(self, member):
-        for category, votes in self.voteables.items():
+        for _, votes in self.voteables.items():
             votes.discard(member)
 
     def get_voters(self, category):
@@ -162,7 +164,7 @@ class MixPlayer(DefaultPlayer):
         return self.voteables[category]
 
     def clear_votes(self):
-        for category, votes in self.voteables.items():
+        for _, votes in self.voteables.items():
             votes.clear()
 
     def enable_looping(self, looping):
@@ -179,7 +181,7 @@ class MixPlayer(DefaultPlayer):
         if isinstance(event, (TrackStuckEvent, TrackExceptionEvent)) or \
                 isinstance(event, TrackEndEvent) and event.reason == 'FINISHED':
             self.skip_voters.clear()
-            for category, votes in self.voteables.items():
+            for _, votes in self.voteables.items():
                 votes.clear()
             await self.play()
 

musicbot/utils/userinteraction/paginators/queuepaginator.py

@@ -1,33 +1,39 @@
 # Discord Packages
 import discord
 
+from typing import Optional
+
+from ...mixplayer.player import MixPlayer
 from .textpaginator import TextPaginator
 
 
 class QueuePaginator(TextPaginator):
-    def __init__(self, localizer, player, color, member: discord.Member = None, include_current: bool = False):
+    def __init__(self, localizer, player: MixPlayer, color: discord.Color,
+                 member: Optional[discord.Member] = None, include_current: bool = False):
         self.localizer = localizer
 
-        queue = player.global_queue() if not member else player.user_queue(member.id, dual=True)
+        # Duration is calculated the same way for both global and user queues
         duration = player.queue_duration(member=member, include_current=include_current)
 
         if member:
-            title = localizer.format_str("{queue.userqueue}",  _user=member.display_name, _length=len(queue),
-                                         _duration=duration)
+            member_queue = player.user_queue_with_global_index(member.id)
+            title = localizer.format_str("{queue.userqueue}",
+                                         _user=member.display_name, _length=len(member_queue), _duration=duration)
+
+            super().__init__(max_lines=10, **{"color": color, "title": title})
+
+            # The user queues also inclue the global position of the tracks
+            for index, (track, global_pos) in enumerate(member_queue):
+                queued_track = localizer.format_str("{queue.usertrack}", _index=index+1, _globalindex=global_pos+1,
+                                                    _title=track.title, _uri=track.uri)
+                self.add_line(queued_track)
         else:
+            queue = player.global_queue()
             title = localizer.format_str("{queue.length}", _length=len(queue), _duration=duration)
 
-        super().__init__(max_lines=10, **{"color": color, "title": title})
-
-        for index, temp in enumerate(queue):
-            if member is None:
-                track = temp
-                queued_track = localizer.format_str("{queue.globaltrack}", _index=index+1,  _title=track.title,
+            super().__init__(max_lines=10, **{"color": color, "title": title})
+            for index, track in enumerate(queue, start=1):
+                queued_track = localizer.format_str("{queue.globaltrack}", _index=index,  _title=track.title,
                                                     _uri=track.uri, _user_id=track.requester)
-            else:
-                track, globpos = temp
-                queued_track = localizer.format_str("{queue.usertrack}", _index=index+1, _globalindex=globpos+1,
-                                                    _title=track.title, _uri=track.uri)
-
-            self.add_line(queued_track)
+                self.add_line(queued_track)
         self.add_page_indicator(self.localizer, "{queue.pageindicator}")

musicbot/utils/userinteraction/scroller.py

@@ -5,7 +5,7 @@ import discord
 
 import asyncio
 from enum import Flag, auto
-from typing import List, Union
+from typing import List, Optional
 
 from .paginators import BasePaginator, CantScrollException
 
@@ -44,7 +44,7 @@ class Scroller:
 
         self.bot = ctx.bot
         self.channel = ctx.channel
-        self.message: Union[discord.Message, None] = None
+        self.message: Optional[discord.Message] = None
 
         # Initialize the view we'll be using for scrolling
         self.view = discord.ui.View(timeout=timeout)
@@ -89,7 +89,7 @@ class Scroller:
             raise CantScrollException('Bot cannot send messages.')
 
     async def start_scrolling(self, clear_mode: ClearOn,
-                              message: Union[discord.Message, None] = None,
+                              message: Optional[discord.Message] = None,
                               start_page: int = 0) -> discord.Message:
         self.clear_mode = clear_mode
         self.page_number = start_page

musicbot/utils/userinteraction/selector.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 import discord
 
 from enum import Enum, auto
-from typing import Callable, Coroutine, List, Union
+from typing import Callable, Coroutine, List, Optional
 
 from .paginators import TextPaginator
 from .scroller import ClearOn, Scroller
@@ -117,7 +117,7 @@ class Selector(TextPaginator, Scroller):
             self.currently_visible_buttons.append(button)
 
     async def start_scrolling(self, clear_mode: ClearOn = ClearOn.Timeout,
-                              message: Union[discord.Message, None] = None,
+                              message: Optional[discord.Message] = None,
                               start_page: int = 0):
         message = await super().start_scrolling(clear_mode, message, start_page)
         return message, self.callback_results