Add watcher for development

Dockerfile

@@ -12,8 +12,8 @@ COPY site/ site/
 COPY .git/ .git/
 
 RUN poetry config virtualenvs.create false
-RUN poetry install --no-interaction --no-root
+RUN poetry install --no-interaction --no-root --without dev
-RUN poetry run python3 generator/ site/
+RUN poetry run python3 generator/ generate site/
 
 # ===================================================================
 

generator/__main__.py

@@ -7,55 +7,85 @@ import fire
 from collections.abc import Callable
 import process
 import functools
+import shutil
 
 
 def make_absurl_filter(site_conf: any) -> Callable[[str], str]:
     return functools.partial(get_absolute_url, site_conf)
 
+class CLI:
+    @staticmethod
+    def g(*args, **kwargs):
+        return CLI.generate(*args, **kwargs)
 
-def run(base_dir: str, output_dir: str = "_dist"):
+    @staticmethod
-    base_dir = Path(base_dir)
+    def generate(base_dir: str, output_dir: str = "_dist"):
-    output_dir = Path(output_dir)
+        base_dir = Path(base_dir)
-    html_dir = output_dir / "html"
+        output_dir = Path(output_dir)
+        html_dir = output_dir / "html"
 
-    if output_dir.exists() and not is_directory_empty(output_dir):
+        if output_dir.exists() and not is_directory_empty(output_dir):
-        rprint(
+            rprint(
-            WARN_LEADER
+                WARN_LEADER
-            + f"A directory called [bold]{output_dir}[/bold] already exists and is not "
+                + f"A directory called [bold]{output_dir}[/bold] already exists and is not "
-            f"empty"
+                f"empty"
+            )
+
+        with open(base_dir / "config.yml") as f:
+            site_config = load_yaml(f.read())
+            assert (
+                "build" not in site_config
+            ), "build section of site config must not exist so it can be written at runtime"
+            site_config["build"] = {
+                "hash": git.get_latest_commit_hash(base_dir)[:7],
+                "date": datetime.datetime.utcnow(),
+            }
+
+        jinja_env = Environment(
+            loader=FileSystemLoader("site/content"), autoescape=select_autoescape()
         )
 
-    with open(base_dir / "config.yml") as f:
+        jinja_env.filters["absurl"] = make_absurl_filter(site_config)
-        site_config = load_yaml(f.read())
+        jinja_env.filters["fmtdate"] = lambda x: x.strftime("%Y-%m-%d")
-        assert (
-            "build" not in site_config
-        ), "build section of site config must not exist so it can be written at runtime"
-        site_config["build"] = {
-            "hash": git.get_latest_commit_hash(base_dir)[:7],
-            "date": datetime.datetime.utcnow(),
-        }
 
-    jinja_env = Environment(
+        process.content(base_dir, html_dir, jinja_env, site_config)
-        loader=FileSystemLoader("site/content"), autoescape=select_autoescape()
+        process.blog(base_dir, html_dir, jinja_env, site_config)
-    )
+        process.caddy_config(base_dir, output_dir, jinja_env, site_config)
 
-    jinja_env.filters["absurl"] = make_absurl_filter(site_config)
+        thing_overrides = {"rendered": "page"}
-    jinja_env.filters["fmtdate"] = lambda x: x.strftime("%Y-%m-%d")
+        res_parts = []
+        for (key, count) in get_counts().items():
+            s = "" if count == 1 else "s"
+            res_parts.append(f"{key} {count} {thing_overrides.get(key, 'file')}{s}")
 
-    process.content(base_dir, html_dir, jinja_env, site_config)
+        rprint(INFO_LEADER + "Finished working, " + ", ".join(res_parts))
-    process.blog(base_dir, html_dir, jinja_env, site_config)
-    process.caddy_config(base_dir, output_dir, jinja_env, site_config)
 
-    thing_overrides = {"rendered": "page"}
+    @staticmethod
-    res_parts = []
+    def watch(base_dir: str, output_dir: str = "_dist"):
-    for (key, count) in get_counts().items():
+        import pyinotify
-        s = "" if count == 1 else "s"
-        res_parts.append(f"{key} {count} {thing_overrides.get(key, 'file')}{s}")
 
-    print()
+        if os.path.exists(output_dir):
+            rprint(WARN_LEADER + f"{output_dir} already exists and will be clobbered when regeneration is triggered")
 
-    rprint(INFO_LEADER + "Finished working, " + ", ".join(res_parts))
+        @debounce(1)
+        def run():
+            if os.path.exists(output_dir):
+                shutil.rmtree(output_dir)
+            reset_counts() # boys and girls, this is why global state is bad
+            CLI.generate(base_dir, output_dir=output_dir)
 
+        class OnWriteHandler(pyinotify.ProcessEvent):
+            def process_IN_MODIFY(self, event):
+                rprint(INFO_LEADER + f"Change detected in {event.pathname}")
+                run()
+
+        wm = pyinotify.WatchManager()
+        notifier = pyinotify.Notifier(wm, default_proc_fun=OnWriteHandler())
+        wm.add_watch(base_dir, pyinotify.ALL_EVENTS, rec=True, auto_add=True)
+        
+        rprint(INFO_LEADER + f"Watching {base_dir}")
+        notifier.loop()
+        
 
 if __name__ == "__main__":
-    fire.Fire(run)
+    fire.Fire(CLI)

generator/process.py

@@ -77,11 +77,11 @@ def content(base_dir: Path, output_dir: Path, jinja_env: Environment, site_confi
                 )
                 os.makedirs(target_path.parent, exist_ok=True)
 
-                rprint(
+                # rprint(
-                    INFO_LEADER
+                #     INFO_LEADER
-                    + f"Rendering [bold]{fpath.relative_to(base_dir)}[/bold]"
+                #     + f"Rendering [bold]{fpath.relative_to(base_dir)}[/bold]"
-                    f"[white] => {target_path}[/white]"
+                #     f"[white] => {target_path}[/white]"
-                )
+                # )
 
                 ctx = {"site": site_config}
                 _template_frontmatter(tpl_frontmatter, jinja_env, ctx)
@@ -172,11 +172,11 @@ def blog(base_dir: Path, output_dir: Path, jinja_env: Environment, site_config:
                 target_path = output_dir / "blog" / post_slug / "index.html"
                 os.makedirs(target_path.parent, exist_ok=True)
 
-                rprint(
+                # rprint(
-                    INFO_LEADER
+                #     INFO_LEADER
-                    + f"Rendering [bold]{fpath.relative_to(base_dir)}[/bold]"
+                #     + f"Rendering [bold]{fpath.relative_to(base_dir)}[/bold]"
-                    f"[white] => {target_path}[/white]"
+                #     f"[white] => {target_path}[/white]"
-                )
+                # )
 
                 if "updatedDate" in post_frontmatter:
                     post_frontmatter["updatedDate"] = list(

generator/util.py

@@ -3,6 +3,7 @@ import os
 from pathlib import Path
 import yaml
 import mistune
+import threading
 
 ERROR_LEADER = "[[red bold]FAIL[/bold red]] "
 WARN_LEADER = "[[yellow bold]WARN[/bold yellow]] "
@@ -20,6 +21,10 @@ def is_directory_empty(path: str | Path) -> bool:
 _counts: dict[str, int] = {}
 
 
+def reset_counts():
+    _counts = {}
+
+
 def update_counts(key: str, delta: int):
     _counts[key] = _counts.get(key, 0) + delta
 
@@ -61,3 +66,30 @@ def render_markdown(raw_content: str, escape: bool = True) -> str:
         escape=escape, plugins=["strikethrough", "table", "footnotes"]
     )
     return markdown(raw_content)
+
+
+def debounce(wait_time):
+    """
+    https://stackoverflow.com/a/66907107
+
+    Decorator that will debounce a function so that it is called after wait_time seconds
+    If it is called multiple times, will wait for the last call to be debounced and run only this one.
+    """
+
+    def decorator(function):
+        def debounced(*args, **kwargs):
+            def call_function():
+                debounced._timer = None
+                return function(*args, **kwargs)
+            # if we already have a call to the function currently waiting to be executed, reset the timer
+            if debounced._timer is not None:
+                debounced._timer.cancel()
+
+            # after wait_time, call the function provided to the decorator with its arguments
+            debounced._timer = threading.Timer(wait_time, call_function)
+            debounced._timer.start()
+
+        debounced._timer = None
+        return debounced
+
+    return decorator

poetry.lock

@@ -158,6 +158,17 @@ files = [
 plugins = ["importlib-metadata"]
 windows-terminal = ["colorama (>=0.4.6)"]
 
+[[package]]
+name = "pyinotify"
+version = "0.9.6"
+description = "Linux filesystem events monitoring"
+category = "dev"
+optional = false
+python-versions = "*"
+files = [
+    {file = "pyinotify-0.9.6.tar.gz", hash = "sha256:9c998a5d7606ca835065cdabc013ae6c66eb9ea76a00a1e3bc6e0cfe2b4f71f4"},
+]
+
 [[package]]
 name = "pyyaml"
 version = "6.0.1"
@@ -267,4 +278,4 @@ tests = ["pytest", "pytest-cov"]
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.10"
-content-hash = "a89183bd8594d0aa4c78441a7c0307ac71a0195e657ddfb94a033a5fe41cb300"
+content-hash = "54f52a4692080706a62ade0de6c704d2919f5e5ae666f4769061b3b99d197889"

pyproject.toml

@@ -14,6 +14,9 @@ rich = "^13.7.0"
 mistune = "^3.0.2"
 
 
+[tool.poetry.group.dev.dependencies]
+pyinotify = "^0.9.6"
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"