Formatting

dynamo/_types.py

@@ -1,5 +1,5 @@
 from collections.abc import Callable, Coroutine, Mapping
-from typing import Any, ParamSpec, TypeVar
+from typing import Any, ParamSpec, Protocol, TypeVar
 
 from discord import app_commands
 from discord.ext import commands
@@ -22,6 +22,15 @@
 V = TypeVar("V", bound="View", covariant=True)
 
 
+class WrappedCoroutine[**P, R](Protocol):
+    __name__: str
+    __call__: Callable[P, Coroutine[Any, Any, R]]
+
+
+class DecoratedCoroutine[**P, R](Protocol):
+    __call__: Callable[[WrappedCoroutine[P, R]], R]
+
+
 class NotFoundWithHelp(commands.CommandError): ...
 
 

dynamo/extensions/cogs/dev.py

@@ -79,14 +79,15 @@ async def load(self, ctx: Context, *, module: str) -> None:
         module: str
             The name of the cog to load.
         """
-        m = get_cog(module)
+        message = ctx.message
+        cog = get_cog(module)
         try:
-            await self.bot.load_extension(m)
+            await self.bot.load_extension(cog)
         except commands.ExtensionError as ex:
             await ctx.send(f"{ex.__class__.__name__}: {ex}")
-            self.log.exception("Failed to load %s", m)
+            self.log.exception("Failed to load %s", cog)
         else:
-            await ctx.send(ctx.Status.OK)
+            await message.add_reaction(ctx.Status.OK)
 
     @commands.hybrid_command(aliases=("ul",))
     @is_owner()
@@ -98,14 +99,15 @@ async def unload(self, ctx: Context, *, module: str) -> None:
         module: str
             The name of the cog to unload.
         """
-        m = get_cog(module)
+        message = ctx.message
+        cog = get_cog(module)
         try:
-            await self.bot.unload_extension(m)
+            await self.bot.unload_extension(cog)
         except commands.ExtensionError as ex:
             await ctx.send(f"{ex.__class__.__name__}: {ex}")
-            self.log.exception("Failed to unload %s", m)
+            self.log.exception("Failed to unload %s", cog)
         else:
-            await ctx.send(ctx.Status.OK)
+            await message.add_reaction(ctx.Status.OK)
 
     @commands.hybrid_group(name="reload", aliases=("r",), invoke_without_command=True)
     @is_owner()
@@ -117,15 +119,15 @@ async def _reload(self, ctx: Context, *, module: str) -> None:
         module: str
             The name of the cog to reload.
         """
-        m = get_cog(module)
-
+        message: discord.Message | None = ctx.message
+        cog = get_cog(module)
         try:
-            await self.bot.reload_extension(m)
+            await self.bot.reload_extension(cog)
         except commands.ExtensionError as ex:
             await ctx.send(f"{ex.__class__.__name__}: {ex}")
-            self.log.exception("Failed to reload %s", m)
+            self.log.exception("Failed to reload %s", cog)
         else:
-            await ctx.send(ctx.Status.OK)
+            await message.add_reaction(ctx.Status.OK)
 
     async def reload_or_load_extension(self, module: str) -> None:
         try:

dynamo/extensions/cogs/errors.py

@@ -10,16 +10,16 @@
 from dynamo.core import Dynamo, DynamoCog
 from dynamo.utils.context import Context
 
+AppCommandErrorMethod = Callable[[Interaction[Dynamo], app_commands.AppCommandError], Coroutine[Any, Any, None]]
+
 
 class Errors(DynamoCog):
     """Handles errors for the bot."""
 
     def __init__(self, bot: Dynamo) -> None:
         super().__init__(bot)
 
-        self.old_tree_error: Callable[
-            [Interaction[Dynamo], app_commands.AppCommandError], Coroutine[Any, Any, None]
-        ] = self.bot.tree.on_error
+        self.old_tree_error: AppCommandErrorMethod = self.bot.tree.on_error
         self.bot.tree.on_error = self.on_app_command_error
         self.command_error_messages = command_error_messages
         self.app_command_error_messages = app_command_error_messages

dynamo/extensions/cogs/events.py

@@ -16,11 +16,7 @@
 def event_to_option(event: discord.ScheduledEvent) -> discord.SelectOption:
     """Convert a ScheduledEvent to a SelectOption to be used in a dropdown menu"""
     description = event.description if event.description is not None else ""
-    return discord.SelectOption(
-        label=event.name,
-        value=str(event.id),
-        description=shorten_string(description),
-    )
+    return discord.SelectOption(label=event.name, value=str(event.id), description=shorten_string(description))
 
 
 class EventsDropdown[V: discord.ui.View](discord.ui.Select[V]):
@@ -63,7 +59,7 @@ async def on_timeout(self) -> None:
 
 class InterestedDropdown(EventsDropdown[EventsView]):
     async def callback(self, interaction: discord.Interaction) -> None:
-        event: discord.ScheduledEvent | None = next((e for e in self.events if e.id == int(self.values[0])), None)
+        event: discord.ScheduledEvent | None = next(filter(lambda e: e.id == int(self.values[0]), self.events), None)
         response = "No users found" if event is None else await get_interested(event)
         await interaction.response.send_message(response, ephemeral=True)
 
@@ -138,7 +134,7 @@ async def event_check(
         return await self.fetch_events(guild) or f"{Context.Status.FAILURE} No events found!"
 
     @commands.hybrid_command(name="event")
-    @commands.cooldown(1, 35, commands.BucketType.guild)
+    @commands.cooldown(1, 35, commands.BucketType.user)
     @commands.guild_only()
     async def event(self, ctx: Context, event: int | None = None) -> None:
         """Get a list of members subscribed to an event

dynamo/extensions/cogs/general.py

@@ -8,7 +8,6 @@
 from dynamo.utils import spotify
 from dynamo.utils.context import Context
 from dynamo.utils.converter import SeedConverter
-from dynamo.utils.format import human_join
 from dynamo.utils.identicon import Identicon, derive_seed, get_colors, get_identicon, seed_from_time
 
 
@@ -99,7 +98,7 @@ async def spotify(
             return
 
         activities = getattr(user, "activities", [])
-        activity: discord.Spotify | None = next((a for a in activities if isinstance(a, discord.Spotify)), None)
+        activity: discord.Spotify | None = next(filter(lambda a: isinstance(a, discord.Spotify), activities), None)
 
         if activity is None:
             await ctx.send(f"{user!s} is not listening to Spotify.")
@@ -111,19 +110,8 @@ async def spotify(
             return
 
         buffer, ext = await spotify.draw(activity, album_cover)
-        fname = f"spotify-card.{ext}"
-
-        file = discord.File(buffer, filename=fname)
-        track = f"[{activity.title}](<{activity.track_url}>)"
-        spotify_emoji = self.bot.app_emojis.get("spotify", "🎧")
-        embed = discord.Embed(
-            title=f"{spotify_emoji} Now Playing",
-            description=f"{user.mention} is listening to **{track}** by"
-            f" **{human_join(activity.artists, conjunction="and")}**",
-            color=activity.color,
-        )
-        embed.set_footer(text=f"Requested by {ctx.author!s}", icon_url=ctx.author.display_avatar.url)
-        embed.set_image(url=f"attachment://{fname}")
+        embed, file = spotify.make_embed(user, activity, buffer, self.bot.app_emojis.get("spotify", "🎧"), ext=ext)
+
         await ctx.send(embed=embed, file=file)
 
 

dynamo/extensions/cogs/info.py

@@ -14,6 +14,33 @@
 PYTHON = "https://s3.dualstack.us-east-2.amazonaws.com/pythondotorg-assets/media/community/logos/python-logo-only.png"
 
 
+def format_commit(commit: pygit2.Commit) -> str:
+    """A formatted commit message [`hash`](url) message (offset)"""
+    short, *_ = commit.message.partition("\n")
+    commit_tz = datetime.timezone(datetime.timedelta(minutes=commit.commit_time_offset))
+    commit_time = datetime.datetime.fromtimestamp(commit.commit_time, commit_tz).astimezone(commit_tz)
+
+    offset = format_relative(dt=commit_time)
+    return f"[`{commit.short_id}`](https://github.com/trumully/dynamo/commit/{commit.id}) {short} ({offset})"
+
+
+def get_latest_commits(count: int = 3) -> str:
+    """Get (count) latest commits from the git repository"""
+    repo = pygit2.Repository(".git")
+    commits = list(itertools.islice(repo.walk(repo.head.target, pygit2.enums.SortMode.TOPOLOGICAL), count))
+    return "\n".join(format_commit(c) for c in commits)
+
+
+def embed_from_user(user: discord.Member | discord.User) -> discord.Embed:
+    embed = discord.Embed(color=user.color)
+    avatar = user.display_avatar.with_static_format("png")
+    embed.set_footer(text=f"ID: {user.id}")
+    embed.set_author(name=str(user))
+    embed.set_image(url=avatar.url)
+    embed.timestamp = discord.utils.utcnow()
+    return embed
+
+
 class Info(DynamoCog):
     """Statistics commands"""
 
@@ -22,37 +49,12 @@ def __init__(self, bot: Dynamo) -> None:
 
         self.process = psutil.Process()
 
-    @staticmethod
-    def format_commit(commit: pygit2.Commit) -> str:
-        """A formatted commit message [`hash`](url) message (offset)"""
-        short, *_ = commit.message.partition("\n")
-        commit_tz = datetime.timezone(datetime.timedelta(minutes=commit.commit_time_offset))
-        commit_time = datetime.datetime.fromtimestamp(commit.commit_time, commit_tz).astimezone(commit_tz)
-
-        offset = format_relative(dt=commit_time)
-        return f"[`{commit.short_id}`](https://github.com/trumully/dynamo/commit/{commit.id}) {short} ({offset})"
-
-    def get_latest_commits(self, count: int = 3) -> str:
-        repo = pygit2.Repository(".git")
-        commits = list(itertools.islice(repo.walk(repo.head.target, pygit2.enums.SortMode.TOPOLOGICAL), count))
-        return "\n".join(self.format_commit(c) for c in commits)
-
-    @staticmethod
-    def embed_from_user(user: discord.Member | discord.User) -> discord.Embed:
-        embed = discord.Embed(color=user.color)
-        embed.set_footer(text=f"ID: {user.id}")
-        avatar = user.display_avatar.with_static_format("png")
-        embed.set_author(name=str(user))
-        embed.set_image(url=avatar.url)
-        embed.timestamp = discord.utils.utcnow()
-        return embed
-
     @commands.hybrid_command(name="about")
     async def about(self, ctx: Context) -> None:
         """Get information about the bot"""
 
         bot_name = self.bot.user.display_name
-        revision = self.get_latest_commits()
+        revision = get_latest_commits()
         avatar_url = self.bot.user.display_avatar.with_static_format("png")
         discord_version = metadata.version("discord.py")
 
@@ -81,7 +83,7 @@ async def avatar(self, ctx: Context, user: discord.Member | discord.User) -> Non
         user: discord.Member | discord.User
             The user to get the avatar of
         """
-        await ctx.send(embed=self.embed_from_user(user), ephemeral=True)
+        await ctx.send(embed=embed_from_user(user), ephemeral=True)
 
 
 async def setup(bot: Dynamo) -> None: