hotreload/__init__.py

@@ -0,0 +1,5 @@
+from .hotreload import HotReload
+
+
+async def setup(bot):
+    await bot.add_cog(HotReload(bot))

hotreload/hotreload.py

@@ -0,0 +1,91 @@
+from asyncio import run_coroutine_threadsafe
+from pathlib import Path
+
+from red_commons.logging import RedTraceLogger, getLogger
+from redbot.core import commands
+from redbot.core.bot import Red
+from redbot.core.core_commands import CoreLogic
+from redbot.core.utils.chat_formatting import bold, humanize_list
+from watchdog.events import FileSystemEvent, RegexMatchingEventHandler
+from watchdog.observers import Observer
+
+
+class HotReload(commands.Cog):
+    """Automatically reload cogs in local cog paths on file change."""
+
+    __author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
+    __git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
+    __version__ = "1.0.0"
+    __documentation__ = "https://seacogs.coastalcommits.com/hotreload/"
+
+    def __init__(self, bot: Red) -> None:
+        super().__init__()
+        self.bot: Red = bot
+        self.logger: RedTraceLogger = getLogger(name="red.SeaCogs.HotReload")
+        self.observer = None
+        watchdog_loggers = [getLogger(name="watchdog.observers.inotify_buffer")]
+        for watchdog_logger in watchdog_loggers:
+            watchdog_logger.setLevel("INFO")  # SHUT UP!!!!
+
+    def cog_load(self) -> None:
+        """Start the observer when the cog is loaded."""
+        self.bot.loop.create_task(self.start_observer())
+
+    def cog_unload(self) -> None:
+        """Stop the observer when the cog is unloaded."""
+        if self.observer:
+            self.observer.stop()
+            self.observer.join()
+            self.logger.info("Stopped observer. No longer watching for file changes.")
+
+    def format_help_for_context(self, ctx: commands.Context) -> str:
+        pre_processed = super().format_help_for_context(ctx) or ""
+        n = "\n" if "\n\n" not in pre_processed else ""
+        text = [
+            f"{pre_processed}{n}",
+            f"{bold('Cog Version:')} [{self.__version__}]({self.__git__})",
+            f"{bold('Author:')} {humanize_list(self.__author__)}",
+            f"{bold('Documentation:')} {self.__documentation__}",
+        ]
+        return "\n".join(text)
+
+    async def get_paths(self) -> tuple[Path]:
+        """Retrieve user defined paths."""
+        cog_manager = self.bot._cog_mgr
+        cog_paths = await cog_manager.user_defined_paths()
+        return (Path(path) for path in cog_paths)
+
+    async def start_observer(self) -> None:
+        """Start the observer to watch for file changes."""
+        self.observer = Observer()
+        paths = await self.get_paths()
+        for path in paths:
+            self.observer.schedule(event_handler=HotReloadHandler(bot=self.bot, path=path), path=path, recursive=True)
+        self.observer.start()
+        self.logger.info("Started observer. Watching for file changes.")
+
+
+class HotReloadHandler(RegexMatchingEventHandler):
+    """Handler for file changes."""
+
+    def __init__(self, bot: Red, path: Path) -> None:
+        super().__init__(regexes=[r".*\.py$"])
+        self.bot: Red = bot
+        self.path: Path = path
+        self.logger: RedTraceLogger = getLogger(name="red.SeaCogs.HotReload.Observer")
+
+    def on_modified(self, event: FileSystemEvent) -> None:
+        """Handle file modification events."""
+        if event.is_directory:
+            return
+        relative_path = Path(event.src_path).relative_to(self.path)
+        package_name = relative_path.parts[0]
+        self.logger.info(f"File {'/'.join(relative_path.parts[1:])} in the cog {package_name} has been modified.")
+        run_coroutine_threadsafe(self.reload_cog(package_name), loop=self.bot.loop)
+
+    async def reload_cog(self, cog_name: str) -> None:
+        """Reload modified cog."""
+        core_logic = CoreLogic(bot=self.bot)
+        self.logger.info(f"Reloading {cog_name} cog.")
+        await core_logic._reload(pkg_names=(cog_name,))
+        self.logger.info(f"Reloaded {cog_name} cog.")

hotreload/info.json

@@ -0,0 +1,17 @@
+{
+  "author" : ["cswimr"],
+  "install_msg" : "Thank you for installing HotReload!",
+  "name" : "HotReload",
+  "short" : "Automatically reload cogs in local cog paths on file change.",
+  "description" : "Automatically reload cogs in local cog paths on file change.",
+  "end_user_data_statement" : "This cog does not store end user data.",
+  "hidden": false,
+  "disabled": false,
+  "min_bot_version": "3.5.0",
+  "min_python_version": [3, 10, 0],
+  "requirements": ["watchdog"],
+  "tags": [
+    "utility",
+    "development"
+  ]
+}

pyproject.toml

@@ -18,6 +18,7 @@ dependencies = [
     "py-dactyl",
     "pydantic>=2.9.2",
     "red-discordbot>=3.5.14",
+    "watchdog>=5.0.3",
     "websockets>=13.1",
 ]
 

uv.lock

@@ -1667,6 +1667,7 @@ dependencies = [
     { name = "py-dactyl" },
     { name = "pydantic" },
     { name = "red-discordbot" },
+    { name = "watchdog" },
     { name = "websockets" },
 ]
 
@@ -1706,6 +1707,7 @@ requires-dist = [
     { name = "py-dactyl", git = "https://github.com/cswimr/pydactyl" },
     { name = "pydantic", specifier = ">=2.9.2" },
     { name = "red-discordbot", specifier = ">=3.5.14" },
+    { name = "watchdog", specifier = ">=5.0.3" },
     { name = "websockets", specifier = ">=13.1" },
 ]