build.py

#!/usr/bin/env python3

import argparse
import os
import sys
import threading
from pathlib import Path

from watchdog.events import (
    FileSystemEventHandler,
)
from watchdog.observers import Observer

ENCODING = "utf-8"
SEPARATOR_WIDTH = 60

JOBS: list[tuple[list[Path], Path]] = [
    (
        [
            Path("src/BreitbandGraphics/breitbandgraphics.lua"),
            Path("src/BreitbandGraphics/types.lua"),
            Path("src/BreitbandGraphics/core.lua"),
            Path("src/BreitbandGraphics/internal.lua"),
            Path("src/BreitbandGraphics/backends/mupen64_d2d.lua"),
            Path("src/BreitbandGraphics/epilogue.lua"),
        ],
        Path("build/breitbandgraphics-amalgamated.lua"),
    ),
    (
        [
            Path("src/ugui/ugui.lua"),
            Path("src/ugui/environment/clipboard.lua"),
            Path("src/ugui/environment/keycode.lua"),
            Path("src/ugui/environment/static_env.lua"),
            Path("src/ugui/environment/env.lua"),
            Path("src/ugui/core.lua"),
            Path("src/ugui/internal.lua"),
            Path("src/ugui/helpers.lua"),
            Path("src/ugui/styler.lua"),
            Path("src/ugui/controls/control.lua"),
            Path("src/ugui/controls/label.lua"),
            Path("src/ugui/controls/button.lua"),
            Path("src/ugui/controls/toggle_button.lua"),
            Path("src/ugui/controls/carrousel_button.lua"),
            Path("src/ugui/controls/scrollbar.lua"),
            Path("src/ugui/controls/listbox.lua"),
            Path("src/ugui/controls/combobox.lua"),
            Path("src/ugui/controls/textbox.lua"),
            Path("src/ugui/controls/joystick.lua"),
            Path("src/ugui/controls/trackbar.lua"),
            Path("src/ugui/controls/menu.lua"),
            Path("src/ugui/controls/tabcontrol.lua"),
            Path("src/ugui/controls/numberbox.lua"),
            Path("src/ugui/controls/spinner.lua"),
            Path("src/ugui/nineslice.lua"),
            Path("src/ugui/epilogue.lua"),
        ],
        Path("build/ugui-amalgamated.lua"),
    ),
]


def make_header(text: str) -> str:
    dashes = "-" * SEPARATOR_WIDTH
    return f"-- {dashes}\n--   {text}\n-- {dashes}\n"


def amalgamate(in_paths, out_path) -> bool:
    chunks: list[str] = []

    chunks.append(
        make_header("THIS FILE IS AUTOMATICALLY GENERATED - DO NOT EDIT DIRECTLY")
    )

    for path in in_paths:
        if not path.exists():
            print(f"[ERROR] File not found: {path}")
            return False
        if not path.is_file():
            print(f"[ERROR] Path is not a file: {path}")
            return False

        try:
            content = path.read_text(encoding=ENCODING)
        except OSError as e:
            print(f"[ERROR] Could not read '{path}': {e}")
            return False

        chunks.append(make_header(path))
        chunks.append(content.rstrip("\n") + "\n")

    amalgamated = "\n".join(chunks)

    try:
        os.makedirs(os.path.dirname(out_path), exist_ok=True)
        out_path.write_text(amalgamated, encoding=ENCODING)
        print(f"[OK] Written to '{out_path}' ({len(in_paths)} file(s) merged).")
        return True
    except OSError as e:
        print(f"[ERROR] Could not write to '{out_path}': {e}")
        return False


def run_all_jobs(exit_on_error: bool = True) -> None:
    """Run every amalgamation job once."""
    for in_paths, out_path in JOBS:
        ok = amalgamate(in_paths, out_path)
        if not ok and exit_on_error:
            sys.exit(1)


def watch() -> None:
    path_to_jobs: dict[Path, list[tuple[list[Path], Path]]] = {}
    for job in JOBS:
        in_paths, _ = job
        for p in in_paths:
            key = p.resolve()
            path_to_jobs.setdefault(key, []).append(job)

    watched_dirs: set[Path] = {p.parent for p in path_to_jobs}

    DEBOUNCE_SECONDS = 0.05
    pending: dict[Path, threading.Timer] = {}
    pending_lock = threading.Lock()

    def handle_change(src_path: str) -> None:
        resolved = Path(src_path).resolve()
        if resolved not in path_to_jobs:
            return

        def fire():
            import time

            timestamp = time.strftime("%H:%M:%S")
            print(f"[WATCH] [{timestamp}] Changed: {resolved}")
            for in_paths, out_path in path_to_jobs[resolved]:
                amalgamate(in_paths, out_path)
            with pending_lock:
                pending.pop(resolved, None)

        with pending_lock:
            existing = pending.pop(resolved, None)
            if existing:
                existing.cancel()
            t = threading.Timer(DEBOUNCE_SECONDS, fire)
            pending[resolved] = t
            t.start()

    class Handler(FileSystemEventHandler):
        def on_modified(self, event):
            if not event.is_directory:
                handle_change(str(event.src_path))

        def on_created(self, event):
            if not event.is_directory:
                handle_change(str(event.src_path))

    print("[WATCH] Starting initial build...")
    run_all_jobs(exit_on_error=False)

    observer = Observer()
    handler = Handler()
    for directory in watched_dirs:
        observer.schedule(handler, str(directory), recursive=False)

    observer.start()
    total_files = sum(len(in_paths) for in_paths, _ in JOBS)
    print(
        f"[WATCH] Watching {total_files} file(s) across {len(watched_dirs)} director(ies). Press Ctrl+C to stop."
    )

    try:
        observer.join()
    except KeyboardInterrupt:
        print("\n[WATCH] Stopping...")
        observer.stop()
        observer.join()
        print("[WATCH] Stopped.")


def main() -> None:
    parser = argparse.ArgumentParser(description="Builds ugui.")
    parser.add_argument(
        "--watch",
        action="store_true",
        help=(
            "Keep running and watch source files for changes, rebuilding when necessary."
        ),
    )
    args = parser.parse_args()

    if args.watch:
        watch()
    else:
        run_all_jobs(exit_on_error=True)


if __name__ == "__main__":
    main()