feat: add beet.contrib.livereload and --reload option

beet/contrib/livereload.py

@@ -0,0 +1,164 @@
+"""Plugin for automatically reloading the data pack after the build."""
+
+
+__all__ = [
+    "livereload",
+    "create_livereload_data_pack",
+    "livereload_server",
+    "LogCallback",
+    "LogWatcher",
+]
+
+
+import json
+import logging
+import re
+import time
+from pathlib import Path
+from threading import Event, Thread
+from typing import Any, Callable, Optional, overload
+
+from beet import Connection, Context, DataPack, Function
+from beet.contrib.autosave import Autosave
+from beet.core.utils import FileSystemPath, remove_path
+
+logger = logging.getLogger("livereload")
+
+
+LIVERELOAD_REGEX = re.compile(r"\[CHAT\] \[livereload\] (Ready|Reloaded)")
+
+
+def beet_default(ctx: Context):
+    autosave = ctx.inject(Autosave)
+    autosave.add_link(livereload)
+
+
+def livereload(ctx: Context):
+    link_cache = ctx.cache["link"]
+
+    link_directory = link_cache.json.get("data_pack")
+    if not link_directory or not ctx.data:
+        return
+
+    data = create_livereload_data_pack()
+    livereload_path = data.save(link_directory)
+    link_cache.json.setdefault("dirty", []).append(str(livereload_path))
+
+    with ctx.worker(livereload_server) as channel:
+        channel.send(livereload_path)
+
+
+def create_livereload_data_pack() -> DataPack:
+    data = DataPack("livereload")
+
+    prefix = {"text": "[livereload]", "color": "red"}
+    ready = ["", prefix, " ", {"text": "Ready", "color": "gold"}]
+    changes = ["", prefix, " ", {"text": f"Changes detected", "color": "gold"}]
+    reloaded = ["", prefix, " ", {"text": f"Reloaded", "color": "gold"}]
+
+    data["livereload:load"] = Function(
+        [
+            "schedule clear livereload:poll",
+            "scoreboard objectives add livereload dummy",
+            f"execute unless score delta livereload matches 1.. run tellraw @a {json.dumps(ready)}",
+            f"execute if score delta livereload matches 1.. run tellraw @a {json.dumps(reloaded)}",
+            "schedule function livereload:poll 1s",
+        ],
+        tags=["minecraft:load"],
+    )
+
+    data["livereload:poll"] = Function(
+        [
+            "execute store result score new_pack_count livereload run datapack list available",
+            "scoreboard players operation delta livereload = new_pack_count livereload",
+            "scoreboard players operation delta livereload -= pack_count livereload",
+            "scoreboard players operation pack_count livereload = new_pack_count livereload",
+            "execute unless score delta livereload matches 1.. run schedule function livereload:poll 1s",
+            f"execute if score delta livereload matches 1.. run tellraw @a {json.dumps(changes)}",
+            "execute if score delta livereload matches 1.. run schedule function livereload:reload 1s",
+        ]
+    )
+
+    data["livereload:reload"] = Function(["reload"])
+
+    return data
+
+
+def livereload_server(connection: Connection[Path, None]):
+    livereload_path = None
+
+    with LogWatcher() as log_watcher:
+        for client in connection:
+            for path in client:
+                if path == livereload_path:
+                    continue
+
+                livereload_path = path
+                log_file_path = path.parent.parent.parent.parent / "logs" / "latest.log"
+
+                if not log_file_path.is_file():
+                    logger.warning("Couldn't monitor game log. Reloading disabled.")
+                    continue
+
+                @log_watcher.tail(log_file_path)
+                def _(line: str):
+                    if LIVERELOAD_REGEX.search(line) and livereload_path:
+                        remove_path(livereload_path)
+
+
+LogCallback = Callable[[str], Any]
+
+
+class LogWatcher:
+    event: Event
+    thread: Optional[Thread]
+
+    def __init__(self):
+        self.event = Event()
+        self.thread = None
+
+    @overload
+    def tail(self, path: FileSystemPath) -> Callable[[LogCallback], None]:
+        ...
+
+    @overload
+    def tail(self, path: FileSystemPath, callback: LogCallback):
+        ...
+
+    def tail(
+        self,
+        path: FileSystemPath,
+        callback: Optional[LogCallback] = None,
+    ) -> Any:
+        def decorator(callback: LogCallback):
+            if self.thread:
+                self.event.set()
+                self.thread.join()
+            self.event.clear()
+            self.thread = Thread(target=self.target, args=(path, callback))
+            self.thread.start()
+
+        if callback:
+            decorator(callback)
+        else:
+            return decorator
+
+    def target(self, path: FileSystemPath, callback: LogCallback):
+        with open(path) as f:
+            f.seek(0, 2)
+            while not self.event.is_set():
+                lines = f.read().splitlines()
+                time.sleep(0.5)
+                for line in lines:
+                    callback(line)
+
+    def close(self):
+        if self.thread:
+            self.event.set()
+            self.thread.join()
+
+    def __enter__(self) -> "LogWatcher":
+        return self
+
+    def __exit__(self, *_):
+        self.close()

beet/toolchain/commands.py

@@ -34,6 +34,12 @@ def build(project: Project, link: Optional[str], no_link: bool):
 
 @beet.command()
 @pass_project
+@click.option(
+    "-r",
+    "--reload",
+    is_flag=True,
+    help="Enable live data pack reloading.",
+)
 @click.option(
     "-l",
     "--link",
@@ -53,7 +59,13 @@ def build(project: Project, link: Optional[str], no_link: bool):
     default=0.6,
     help="Configure the polling interval.",
 )
-def watch(project: Project, link: Optional[str], no_link: bool, interval: float):
+def watch(
+    project: Project,
+    reload: bool,
+    link: Optional[str],
+    no_link: bool,
+    interval: float,
+):
     """Watch the project directory and build on file changes."""
     text = "Linking and watching project..." if link else "Watching project..."
     with message_fence(text):
@@ -73,6 +85,9 @@ def watch(project: Project, link: Optional[str], no_link: bool, interval: float)
             change_time = click.style(now, fg="green", bold=True)
             echo(f"{change_time} {text}")
 
+            if reload:
+                project.config.pipeline.append("beet.contrib.livereload")
+
             with error_handler(format_padding=1):
                 project.build(no_link)