Implementing a --watch flag...

alteza/engine.py

@@ -6,11 +6,13 @@
 import sys
 import time
 import types
-from typing import Optional, Generator, List, Dict, Set, Any
+from typing import Callable, Optional, Generator, List, Dict, Set, Any
 
 import sh  # type: ignore
 from pypage import pypage  # type: ignore
 from tap import Tap
+from watchdog.events import FileSystemEventHandler, FileSystemEvent
+from watchdog.observers import Observer as WatchdogObserver
 
 from .fs import (
     FsNode,
@@ -34,6 +36,7 @@ class Args(Tap):  # pyre-ignore[13]
     clear_output_dir: bool = False  # Delete the output directory, if it already exists.
     copy_assets: bool = False  # Copy static assets instead of symlinking to them.
     seed: str = "{}"  # Seed JSON data to add to the initial root env.
+    watch: bool = False  # Watch for content changes, and rebuild.
 
 
 class Content:
@@ -324,73 +327,72 @@ def enterDir(newDir: str) -> Generator[None, None, None]:
 
 
 class Engine:
-    class Generate:
-        # These classes are just here to organize a bunch of related functions together.
-        # This class should never be instantiated. Generate.generate(...) is the entry
-        # point to be called to write the output of a processed Content object.
-        # Similarly, Engine.run(args) is used to invoke Alteza overall.
-        @staticmethod
-        def writeMdContents(md: Md) -> None:
-            if os.path.exists("index.html"):
-                raise AltezaException(
-                    f"An index.html already exists, and conflicts with {md}, at {os.getcwd()}."
-                )
-            with open("index.html", "w", encoding="utf-8") as pageHtml:
-                pageHtml.write(md.getPyPageOutput())
-
-        @staticmethod
-        def writeMd(md: Md) -> None:
-            if not md.isIndex():
-                os.mkdir(md.realName)
-                with enterDir(md.realName):
-                    Engine.Generate.writeMdContents(md)
-            else:
-                Engine.Generate.writeMdContents(md)
+    # This class is just here to organize a bunch of related functions together.
+    # This class should never be instantiated, and most functions not called directly.
+    # Engine.generate(...) is called to write the output of a processed Content object.
+    # Similarly, Engine.run(args) is used to invoke Alteza overall.
+    @staticmethod
+    def generateMdContents(md: Md) -> None:
+        if os.path.exists("index.html"):
+            raise AltezaException(
+                f"An index.html already exists, and conflicts with {md}, at {os.getcwd()}."
+            )
+        with open("index.html", "w", encoding="utf-8") as pageHtml:
+            pageHtml.write(md.getPyPageOutput())
 
-        @staticmethod
-        def writeNonMd(nonMd: NonMd) -> None:
-            fileName = nonMd.rectifiedFileName
-            if os.path.exists(fileName):
-                raise AltezaException(
-                    f"File {fileName} already exists, and conflicts with {nonMd}."
-                )
-            with open(fileName, "w", encoding="utf-8") as nonMdPageFile:
-                nonMdPageFile.write(nonMd.getPyPageOutput())
+    @staticmethod
+    def generateMd(md: Md) -> None:
+        if not md.isIndex():
+            os.mkdir(md.realName)
+            with enterDir(md.realName):
+                Engine.generateMdContents(md)
+        else:
+            Engine.generateMdContents(md)
 
-        @staticmethod
-        def writePyPageNode(pyPageNode: PyPageNode) -> None:
-            if isinstance(pyPageNode, Md):
-                Engine.Generate.writeMd(pyPageNode)
+    @staticmethod
+    def generateNonMd(nonMd: NonMd) -> None:
+        fileName = nonMd.rectifiedFileName
+        if os.path.exists(fileName):
+            raise AltezaException(
+                f"File {fileName} already exists, and conflicts with {nonMd}."
+            )
+        with open(fileName, "w", encoding="utf-8") as nonMdPageFile:
+            nonMdPageFile.write(nonMd.getPyPageOutput())
 
-            elif isinstance(pyPageNode, NonMd):
-                Engine.Generate.writeNonMd(pyPageNode)
+    @staticmethod
+    def generatePyPageNode(pyPageNode: PyPageNode) -> None:
+        if isinstance(pyPageNode, Md):
+            Engine.generateMd(pyPageNode)
 
-            else:
-                raise AltezaException(f"{pyPageNode} pyPage attribute is invalid.")
+        elif isinstance(pyPageNode, NonMd):
+            Engine.generateNonMd(pyPageNode)
 
-        @staticmethod
-        def linkStaticAsset(fileNode: FileNode, shouldCopy: bool) -> None:
-            if shouldCopy:
-                shutil.copyfile(fileNode.absoluteFilePath, fileNode.fileName)
-            else:
-                os.symlink(fileNode.absoluteFilePath, fileNode.fileName)
-
-        @staticmethod
-        def generate(args: Args, content: Content) -> None:
-            def walk(curDir: DirNode) -> None:
-                for subDir in filter(lambda node: node.shouldPublish, curDir.subDirs):
-                    os.mkdir(subDir.dirName)
-                    with enterDir(subDir.dirName):
-                        walk(subDir)
-
-                for fileNode in filter(lambda node: node.shouldPublish, curDir.files):
-                    if isinstance(fileNode, PyPageNode):
-                        Engine.Generate.writePyPageNode(fileNode)
-                    else:
-                        Engine.Generate.linkStaticAsset(fileNode, args.copy_assets)
+        else:
+            raise AltezaException(f"{pyPageNode} pyPage attribute is invalid.")
+
+    @staticmethod
+    def generateStaticAsset(fileNode: FileNode, shouldCopy: bool) -> None:
+        if shouldCopy:
+            shutil.copyfile(fileNode.absoluteFilePath, fileNode.fileName)
+        else:
+            os.symlink(fileNode.absoluteFilePath, fileNode.fileName)
 
-            with enterDir(args.output):
-                walk(content.rootDir)
+    @staticmethod
+    def generate(args: Args, content: Content) -> None:
+        def walk(curDir: DirNode) -> None:
+            for subDir in filter(lambda node: node.shouldPublish, curDir.subDirs):
+                os.mkdir(subDir.dirName)
+                with enterDir(subDir.dirName):
+                    walk(subDir)
+
+            for fileNode in filter(lambda node: node.shouldPublish, curDir.files):
+                if isinstance(fileNode, PyPageNode):
+                    Engine.generatePyPageNode(fileNode)
+                else:
+                    Engine.generateStaticAsset(fileNode, args.copy_assets)
+
+        with enterDir(args.output):
+            walk(content.rootDir)
 
     @staticmethod
     def checkContentDir(contentDir: str) -> None:
@@ -420,7 +422,7 @@ def resetOutputDir(outputDir: str, shouldDelete: bool) -> None:
         os.mkdir(outputDir)
 
     @staticmethod
-    def process(args: Args) -> Content:
+    def processContent(args: Args) -> Content:
         with enterDir(args.content):
             print("Analyzing content directory...")
             fs = Fs()
@@ -436,16 +438,61 @@ def process(args: Args) -> Content:
         return content
 
     @staticmethod
-    def run(args: Args) -> None:
+    def makeSite(args: Args) -> None:
         startTimeNs = time.time_ns()
 
         Engine.checkContentDir(args.content)
         Engine.resetOutputDir(args.output, args.clear_output_dir)
 
-        content = Engine.process(args)
+        content = Engine.processContent(args)
         print("Generating...")
-        Engine.Generate.generate(args, content)
+        Engine.generate(args, content)
 
         elapsedMilliseconds = (time.time_ns() - startTimeNs) / 10**6
         # pylint: disable=consider-using-f-string
         print("\nSite generation complete. Time elapsed: %.2f ms" % elapsedMilliseconds)
+
+    class WatchdogEventHandler(FileSystemEventHandler):
+        def __init__(self) -> None:
+            self.timeOfMostRecentEvent: Optional[int] = None
+
+        def on_any_event(self, event: FileSystemEvent) -> None:
+            self.timeOfMostRecentEvent = max(
+                self.timeOfMostRecentEvent or 0, time.time_ns()
+            )
+
+    @staticmethod
+    def runWatchdog(contentDir: str, action: Callable[[], None]) -> None:
+        timeIntervalNs = 2 * 10**8
+        timeIntervalSecs = 0.2
+
+        def watching() -> None:
+            print(f"\nWatching for changes in {contentDir}...")
+
+        eventHandler = Engine.WatchdogEventHandler()
+        observer = WatchdogObserver()
+        observer.schedule(eventHandler, contentDir, recursive=True)
+        observer.start()
+        try:
+            watching()
+            while True:
+                time.sleep(timeIntervalSecs)
+                if eventHandler.timeOfMostRecentEvent:
+                    timeSinceMostRecentEvent = time.time_ns() - (
+                        eventHandler.timeOfMostRecentEvent or 0
+                    )
+                    if timeSinceMostRecentEvent > timeIntervalNs:
+                        eventHandler.timeOfMostRecentEvent = None
+                        print("\nDetected a change.\n")
+                        action()
+                        watching()
+        finally:
+            observer.stop()
+            observer.join()
+
+    @staticmethod
+    def run(args: Args) -> None:
+        Engine.makeSite(args)
+
+        if args.watch:
+            Engine.runWatchdog(args.content, lambda: Engine.makeSite(args))

requirements.txt

@@ -28,3 +28,5 @@ typing-extensions==4.9.0
     # via typing-inspect
 typing-inspect==0.9.0
     # via typed-argument-parser
+watchdog==4.0.1
+    # via alteza (setup.py)

setup.py

@@ -22,6 +22,7 @@
         "colored >= 2.2.4",
         "sh >= 2.0.7",
         "typed-argument-parser >= 1.10.1",
+        "watchdog >= 4.0.1",
     ],
     long_description=open("README.md").read(),
     long_description_content_type="text/markdown",