Tidy up

dynamo/core/cog.py

@@ -11,6 +11,8 @@
 if TYPE_CHECKING:
     from dynamo.core import Dynamo
 
+type Submittables = dict[str, type[RawSubmittable]]
+
 
 class Cog(commands.Cog):
     """Dynamo cog. Sets up logging and any existing raw submittables."""
@@ -20,8 +22,8 @@ class Cog(commands.Cog):
     def __init__(
         self,
         bot: Dynamo,
-        raw_modal_submits: dict[str, type[RawSubmittable]] | None = None,
-        raw_button_submits: dict[str, type[RawSubmittable]] | None = None,
+        raw_modal_submits: Submittables | None = None,
+        raw_button_submits: Submittables | None = None,
     ) -> None:
         self.bot: Dynamo = bot
         self.log = logging.getLogger(get_cog(self.__class__.__name__))

dynamo/extensions/cogs/dev.py

@@ -1,5 +1,7 @@
 import contextlib
 import importlib
+import importlib.abc
+import importlib.metadata
 import sys
 from collections.abc import AsyncGenerator, Callable
 from typing import Literal, cast
@@ -28,7 +30,7 @@ async def _execute_extension_action(self, action: Callable[[str], Coro[None]], c
         try:
             await action(get_cog(cog))
         except commands.ExtensionError:
-            self.log.exception("Extension %s failed for cog %s", action.__name__, cog)
+            self.log.exception("Action '%s' failed for cog %s", action.__name__, cog)
             return False
         return True
 
@@ -97,57 +99,63 @@ async def _reload(self, ctx: Context, *, module: str) -> None:
         await ctx.message.add_reaction(ctx.Status.OK if success else ctx.Status.FAILURE)
 
     async def _reload_extension(self, module: str) -> bool:
-        try:
-            await self.bot.reload_extension(module)
-        except commands.ExtensionNotLoaded:
-            self.log.warning("Extension %s is not loaded. Attempting to load...", module)
-            return await self._execute_extension_action(self.bot.load_extension, module)
-        return True
+        result = await self._execute_extension_action(self.bot.reload_extension, module)
+        if not result:
+            result = await self._execute_extension_action(self.bot.load_extension, module)
+        return result
+
+    def _reload_utils(self) -> list[tuple[Context.Status, str]]:
+        modules_to_reload = frozenset(sys.modules[m] for m in sys.modules if m.startswith("dynamo.utils."))
+        result: list[tuple[Context.Status, str]] = []
+        for module in modules_to_reload:
+            try:
+                importlib.reload(module)
+                result.append((Context.Status.SUCCESS, module.__name__))
+            except Exception:
+                self.log.exception("Failed to reload module %s. Never imported?", module.__name__)
+                result.append((Context.Status.FAILURE, module.__name__))
+        return result
+
+    async def _reload_extensions(self) -> list[tuple[Context.Status, str]]:
+        extensions = frozenset(self.bot.cogs)
+        result: list[tuple[Context.Status, str]] = []
+        for extension in extensions:
+            try:
+                await self._reload_extension(extension)
+                result.append((Context.Status.SUCCESS, extension))
+            except Exception:
+                self.log.exception("Failed to reload extension %s", extension)
+                result.append((Context.Status.FAILURE, extension))
+        return result
 
     @_reload.command(name="all")
     @is_owner()
-    async def _reload_all(self, ctx: Context) -> None:
-        """Reload all cogs"""
-        if not await ctx.prompt("Are you sure you want to reload all cogs?"):
+    async def reload_all(self, ctx: Context) -> None:
+        """Reload extensions and utils"""
+        confirm = await ctx.prompt("Are you sure you want to reload all extensions and utils?")
+        if not confirm:
             return
 
-        utils_modules = self._reload_utils_modules()
-        extensions_status = await self._reload_all_extensions()
-
-        await ctx.send(self._format_reload_results(utils_modules, extensions_status))
-
-    def _reload_utils_modules(self) -> tuple[int, int]:
-        utils_modules = [mod for mod in sys.modules if mod.startswith("dynamo.utils.")]
-        success = sum(self._reload_module(mod) for mod in utils_modules)
-        return success, len(utils_modules)
+        extensions = await self._reload_extensions()
+        utils = self._reload_utils()
 
-    def _reload_module(self, module: str) -> bool:
-        try:
-            importlib.reload(sys.modules[module])
-        except (KeyError, ModuleNotFoundError, NameError):
-            self.log.exception("Failed to reload %s", module)
-            return False
-        return True
+        if not extensions and not utils:
+            await ctx.send("No extensions or utils to reload.")
+            return
 
-    async def _reload_all_extensions(self) -> list[tuple[Context.Status, str]]:
-        extensions = list(self.bot.extensions)
-        success = Context.Status.SUCCESS
-        failure = Context.Status.FAILURE
-        return [(success if await self._reload_extension(ext) else failure, ext) for ext in extensions]
-
-    def _format_reload_results(
-        self, utils_result: tuple[int, int], extensions_status: list[tuple[Context.Status, str]]
-    ) -> str:
-        utils_success, utils_total = utils_result
-        extensions_success = sum(1 for status, _ in extensions_status if status == Context.Status.SUCCESS)
-        extensions_total = len(extensions_status)
-
-        result = [
-            f"Reloaded {utils_success}/{utils_total} utilities",
-            f"Reloaded {extensions_success}/{extensions_total} extensions",
-            "\n".join(f"{status} `{ext}`" for status, ext in extensions_status),
-        ]
-        return "\n".join(result)
+        await ctx.send(self._pretty_results(extensions, utils))
+
+    @staticmethod
+    def _pretty_results(extensions: list[tuple[Context.Status, str]], utils: list[tuple[Context.Status, str]]) -> str:
+        result = ""
+        if extensions:
+            result += "### Extensions\n"
+            result += "\n".join(f"> {status.value}\t`{get_cog(name)}`" for status, name in extensions)
+        if utils:
+            result += f"{'\n' if extensions else ''}"
+            result += "### Utils\n"
+            result += "\n".join(f"> {status.value}\t`{name}`" for status, name in utils)
+        return result
 
     @commands.hybrid_command(name="quit", aliases=("exit", "shutdown", "q"))
     @is_owner()

dynamo/extensions/cogs/events.py

@@ -54,8 +54,7 @@ async def on_timeout(self) -> None:
         for i in self.children:
             item = cast(EventsDropdown[EventsView], i)
             item.disabled = True
-        await self.message.edit(content="Expired!", view=self)
-        await self.message.delete(delay=10)
+        await self.message.edit(content="Expired!", view=self, delete_after=10)
 
 
 class InterestedDropdown(EventsDropdown[EventsView]):
@@ -71,11 +70,6 @@ async def get_interested(event: discord.ScheduledEvent) -> str:
 
     Get a list of users interested in an event
 
-    Parameters
-    ----------
-    event : discord.ScheduledEvent
-        The event to get interested users of
-
     Returns
     -------
     str
@@ -158,8 +152,7 @@ async def event(self, ctx: Context, event_id: int | None = None) -> None:
         event_exists: str | list[discord.ScheduledEvent] = await event_check(ctx.guild, event_id)
         if isinstance(event_exists, str):
             self.active_users.remove(ctx.author.id)
-            await message.edit(content=event_exists)
-            await message.delete(delay=10)
+            await message.edit(content=event_exists, delete_after=10)
             return
 
         view = EventsView(ctx.author.id, event_exists, InterestedDropdown, timeout=25)
@@ -177,7 +170,7 @@ async def event_error(self, ctx: Context, error: Exception) -> NoReturn:
         """
         if ctx.author.id in self.active_users:
             self.active_users.remove(ctx.author.id)
-        await ctx.send(f"Something went wrong: {error}")
+        await ctx.send(f"Something went wrong: {error!s}")
         raise error
 
 

dynamo/extensions/cogs/general.py

@@ -10,7 +10,7 @@
 from dynamo.utils import spotify
 from dynamo.utils.context import Context
 from dynamo.utils.converter import MemberLikeConverter
-from dynamo.utils.identicon import Identicon, derive_seed, get_colors, get_identicon, seed_from_time
+from dynamo.utils.identicon import derive_seed, get_colors, get_identicon, seed_from_time
 
 
 class General(Cog):
@@ -42,10 +42,10 @@ async def generate_identicon(
         name = seed_to_use if (isinstance(seed_to_use, str | int)) else seed_to_use.display_name
 
         seed_to_use = derive_seed(name)
-        fg, bg = get_colors(seed=seed_to_use)
 
-        identicon: bytes = await get_identicon(Identicon(5, fg, bg, 0.4, seed_to_use))
+        identicon: bytes = await get_identicon(seed_to_use)
         file = discord.File(BytesIO(identicon), filename="identicon.png")
+        fg, _ = get_colors(seed_to_use)
 
         cmd_mention = await self.bot.tree.find_mention_for("identicon", guild=guild)
         prefix = "d!" if guild is None else self.bot.prefixes.get(guild.id, ["d!", "d?"])[0]
@@ -54,11 +54,6 @@ async def generate_identicon(
         e.set_image(url="attachment://identicon.png")
         return e, file
 
-    @commands.hybrid_command(name="ping")
-    async def ping(self, ctx: Context) -> None:
-        """Get the bot's latency"""
-        await ctx.send(f"\N{TABLE TENNIS PADDLE AND BALL} {round(self.bot.latency * 1000)}ms")
-
     @commands.hybrid_command(name="identicon", aliases=("i", "idt"))
     async def identicon(
         self,

dynamo/extensions/cogs/info.py

@@ -1,5 +1,7 @@
 import datetime
 import itertools
+import os
+import time
 from importlib import metadata
 
 import discord
@@ -48,7 +50,16 @@ class Info(Cog):
     def __init__(self, bot: Dynamo) -> None:
         super().__init__(bot)
 
-        self.process = psutil.Process()
+        self.process = psutil.Process(os.getpid())
+
+    @commands.hybrid_command(name="ping")
+    async def ping(self, ctx: Context) -> None:
+        """Get the bot's latency"""
+        before = time.monotonic()
+        before_ws = int(round(self.bot.latency * 1000, 1))
+        message = await ctx.send("\N{TABLE TENNIS PADDLE AND BALL} Pong")
+        ping = (time.monotonic() - before) * 1000
+        await message.edit(content=f"\N{TABLE TENNIS PADDLE AND BALL} WS: `{before_ws}ms`\t|\tREST: `{int(ping)}ms`")
 
     @commands.hybrid_command(name="about")
     async def about(self, ctx: Context) -> None:

dynamo/utils/cache.py

@@ -111,8 +111,7 @@ def async_cache[**P, T](coro: WrappedCoro[P, T], /) -> CachedTask[P, T]: ...
 def async_cache[**P, T](
     maxsize: int | WrappedCoro[P, T] | None = 128, ttl: float | None = None
 ) -> CachedTask[P, T] | DecoratedCoro[P, T]:
-    """
-    Decorator to cache the result of a coroutine.
+    """Decorator to cache the result of a coroutine.
 
     This decorator caches the result of a coroutine to improve performance
     by avoiding redundant computations. It is functionally similar to :func:`functools.cache`

dynamo/utils/checks.py

@@ -8,7 +8,7 @@
 
 
 class Check(Protocol):
-    predicate: Callable[..., Coro[bool]]
+    predicate: Callable[[Context], Coro[bool]]
 
     def __call__[T](self, coro_or_commands: T) -> T: ...
 

dynamo/utils/context.py

@@ -19,16 +19,6 @@ class ConfirmationView(View):
     message: discord.Message | None
 
     def __init__(self, *, timeout: float, author_id: int, delete_after: bool) -> None:
-        """
-        Parameters
-        ----------
-        timeout: float
-            The timeout for the view.
-        author_id: int
-            The ID of the author of the view.
-        delete_after: bool
-            Whether to delete the message after the view times out.
-        """
         super().__init__(timeout=timeout)
         self.author_id: int = author_id
         self.delete_after: bool = delete_after
@@ -92,25 +82,10 @@ async def prompt(
         delete_after: bool = True,
         author_id: int | None = None,
     ) -> bool:
-        """|coro|
+        """
+        |coro|
 
         Prompt the user to confirm an action
-
-        Parameters
-        ----------
-        message: str
-            The message to send to the user.
-        timeout: float
-            The timeout for the view.
-        author_id: int | None
-            The ID of the author of the view. If not provided, the author of the context is used.
-        delete_after: bool
-            Whether to delete the message after the view times out.
-
-        Returns
-        -------
-        bool
-            Whether the user confirmed the action.
         """
         author_id = author_id or self.author.id
         view = ConfirmationView(timeout=timeout, author_id=author_id, delete_after=delete_after)