92 lines
3.7 KiB
Python
92 lines
3.7 KiB
Python
|
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.")
|