Nomarchy/pkgs/nomarchy-theme-sync/nomarchy-theme-sync.py

#!/usr/bin/env python3
"""nomarchy-theme-sync — state writer for Nomarchy's declarative theming.

Single source of truth: $NOMARCHY_PATH/theme-state.json (inside the flake,
read purely by Home Manager via the nomarchy.stateFile option).

Theme changes are applied by Home Manager: this tool writes the new state
and runs `home-manager switch` (override with $NOMARCHY_REBUILD, or skip
with --no-switch). All app theming — Hyprland, Waybar, Ghostty, btop,
Stylix — is baked into the generation; nothing is patched at runtime.

The one runtime exception is the wallpaper: swww is imperative by nature,
so `bg next` cycles instantly and `wallpaper` (re-)applies the current one
at session start and after a switch.

Commands:
  list                       list theme presets
  apply <name|file.json>     merge a preset into the state + rebuild
  set <dotted.path> <value>  tweak one key (e.g. `set ui.gapsOut 16`) + rebuild
  get [dotted.path]          print the current state (or one key)
  wallpaper                  apply the current wallpaper via swww
  bg [next|auto]             cycle the theme's wallpapers (instant, no rebuild)
"""

import argparse
import json
import os
import shlex
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

# ─── Paths ────────────────────────────────────────────────────────────────

FLAKE_DIR = Path(os.environ.get("NOMARCHY_PATH", Path.home() / ".nomarchy")).expanduser()
STATE_FILE = FLAKE_DIR / "theme-state.json"

# Preset search path: the user's flake first (their custom themes win),
# then the presets baked into this package by the Nix build.
THEMES_DIRS = [FLAKE_DIR / "themes"]
if os.environ.get("NOMARCHY_DEFAULT_THEMES"):
    THEMES_DIRS.append(Path(os.environ["NOMARCHY_DEFAULT_THEMES"]))

WALLPAPER_EXTS = {".png", ".jpg", ".jpeg", ".webp"}

QUIET = False


def log(msg: str) -> None:
    if not QUIET:
        print(f"nomarchy-theme-sync: {msg}")


def die(msg: str) -> "None":
    print(f"nomarchy-theme-sync: error: {msg}", file=sys.stderr)
    sys.exit(1)


# All Nomarchy theme notifications share this synchronous tag, so the
# notification daemon (swaync) REPLACES the previous one in place instead
# of stacking — the "rebuilding…" toast becomes the "applied ✓" toast.
NOTIFY_SYNC_TAG = "nomarchy-theme-sync"


def notify(body: str, summary: str = "Nomarchy", *,
           persistent: bool = False, urgency: str = "normal") -> None:
    """Desktop notification. persistent (timeout 0) keeps it up for the
    whole rebuild so a slow switch never reads as a failed selection;
    the success/failure call replaces it via the synchronous tag."""
    if shutil.which("notify-send") is None:
        return
    subprocess.run(
        ["notify-send", "-a", "Nomarchy",
         "-u", urgency,
         "-t", "0" if persistent else "5000",
         "-h", f"string:x-canonical-private-synchronous:{NOTIFY_SYNC_TAG}",
         summary, body],
        capture_output=True,
    )


# ─── State management ─────────────────────────────────────────────────────

def load_state(path: Path = STATE_FILE) -> dict:
    try:
        return json.loads(path.read_text())
    except FileNotFoundError:
        die(f"state file not found: {path} (set $NOMARCHY_PATH to your flake checkout)")
    except json.JSONDecodeError as e:
        die(f"invalid JSON in {path}: {e}")


def write_state(state: dict) -> None:
    """Atomic write: render to a temp file in the same dir, then rename."""
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=STATE_FILE.parent, prefix=".theme-state.", suffix=".json")
    try:
        with os.fdopen(fd, "w") as f:
            json.dump(state, f, indent=2)
            f.write("\n")
        os.replace(tmp, STATE_FILE)
    except BaseException:
        os.unlink(tmp)
        raise
    log(f"state written to {STATE_FILE}")

    # Flakes only see git-tracked files. theme-state.json ships tracked,
    # but make sure a fresh checkout / reset can't silently hide it.
    if (FLAKE_DIR / ".git").exists() and shutil.which("git"):
        subprocess.run(
            ["git", "-C", str(FLAKE_DIR), "add", "--intent-to-add", "theme-state.json"],
            capture_output=True,
        )


def deep_merge(base: dict, override: dict) -> dict:
    out = dict(base)
    for k, v in override.items():
        if isinstance(v, dict) and isinstance(out.get(k), dict):
            out[k] = deep_merge(out[k], v)
        else:
            out[k] = v
    return out


# ─── Font verification ────────────────────────────────────────────────────

def check_fonts(state: dict) -> None:
    """Warn when a configured font family isn't installed. fontconfig
    substitutes silently, so a typo'd or missing fonts.mono would just
    render the whole desktop in some fallback with no error anywhere."""
    if shutil.which("fc-match") is None:
        return
    for key in ("mono", "ui"):
        name = (state.get("fonts") or {}).get(key)
        if not name:
            continue
        result = subprocess.run(["fc-match", "-f", "%{family}", name],
                                capture_output=True, text=True)
        families = [f.strip().lower() for f in result.stdout.split(",")]
        if result.returncode != 0 or name.strip().lower() not in families:
            fallback = result.stdout.strip() or "unknown"
            print(f"nomarchy-theme-sync: warning: fonts.{key} '{name}' is not "
                  f"installed — fontconfig will substitute '{fallback}'",
                  file=sys.stderr)
            notify(f"Font '{name}' is not installed — using '{fallback}'")


# ─── Rebuild dispatch ─────────────────────────────────────────────────────

def run_switch() -> None:
    """Bake the new state into a Home Manager generation."""
    override = os.environ.get("NOMARCHY_REBUILD")
    argv = shlex.split(override) if override else \
        ["home-manager", "switch", "--flake", str(FLAKE_DIR)]

    if shutil.which(argv[0]) is None:
        log(f"'{argv[0]}' not found — state written; rebuild manually to apply")
        return

    log(f"rebuilding: {' '.join(argv)}")
    # Persistent (timeout 0): stays up for the whole rebuild — replaced
    # in place by the success/failure notification below — so a multi-
    # minute switch never looks like it silently failed.
    notify("Applying changes — rebuilding the desktop…", persistent=True)
    result = subprocess.run(argv)  # stream output to the caller's terminal
    if result.returncode != 0:
        notify("Rebuild FAILED — see terminal / journal", urgency="critical")
        die("rebuild failed (state file already updated; fix and re-run)")
    notify("Changes applied ✓")
    # Waybar runs from Hyprland's exec-once (not a systemd unit HM would restart
    # on switch), so nudge the running bar to re-read its freshly rebuilt
    # config/style. SIGUSR2 = reload; harmless no-op if waybar isn't running.
    if shutil.which("pkill"):
        subprocess.run(["pkill", "--signal", "SIGUSR2", "-x", "waybar"], check=False)


# ─── Theme assets (wallpapers) ────────────────────────────────────────────
# Convention: a preset themes/<slug>.json may have a sibling directory
# themes/<slug>/ with assets: backgrounds/ (wallpapers), btop.theme,
# waybar.css / waybar.jsonc. All except backgrounds/ are consumed by the
# Nix modules at eval time; wallpapers are applied here via swww.

def find_asset(slug: str, name: str):
    for themes_dir in THEMES_DIRS:
        candidate = themes_dir / slug / name
        if candidate.exists():
            return candidate
    return None


def backgrounds_for(slug: str) -> list:
    bg_dir = find_asset(slug, "backgrounds")
    if bg_dir is None or not bg_dir.is_dir():
        return []
    return sorted(p for p in bg_dir.iterdir() if p.suffix.lower() in WALLPAPER_EXTS)


def resolve_wallpaper(state: dict):
    """Explicit path in the state wins; empty means 'first theme background'."""
    explicit = state.get("wallpaper", "")
    if explicit:
        path = Path(explicit).expanduser()
        if path.is_file():
            return path
        log(f"wallpaper not found, using theme default: {explicit}")
    backgrounds = backgrounds_for(state.get("slug", ""))
    return backgrounds[0] if backgrounds else None


def apply_wallpaper(state: dict, wait: bool = False) -> None:
    wallpaper = resolve_wallpaper(state)
    if wallpaper is None:
        log(f"no wallpaper for theme '{state.get('slug', '?')}', skipping")
        return
    # nixpkgs renamed swww to awww (CLI-compatible fork); accept either.
    swww = shutil.which("awww") or shutil.which("swww")
    if swww is None:
        log("awww/swww not found, skipping wallpaper")
        return
    # At session start the daemon may still be coming up.
    for _ in range(10 if wait else 1):
        if subprocess.run([swww, "query"], capture_output=True).returncode == 0:
            break
        time.sleep(0.5)
    result = subprocess.run(
        [swww, "img", str(wallpaper),
         "--transition-type", "grow",
         "--transition-pos", "center",
         "--transition-duration", "1",
         "--transition-fps", "60"],
        capture_output=True,
    )
    log(f"wallpaper: {wallpaper.name}" if result.returncode == 0
        else "wallpaper: awww failed (daemon not running?)")


# ─── Commands ─────────────────────────────────────────────────────────────

def find_preset(name: str):
    for themes_dir in THEMES_DIRS:
        candidate = themes_dir / f"{name}.json"
        if candidate.is_file():
            return candidate
    return None


def cmd_list(_args) -> None:
    names = sorted({f.stem for d in THEMES_DIRS if d.is_dir() for f in d.glob("*.json")})
    if not names:
        die(f"no theme presets found (searched: {', '.join(map(str, THEMES_DIRS))})")
    print("\n".join(names))


def cmd_apply(args) -> None:
    candidate = Path(args.theme).expanduser()
    preset_path = candidate if candidate.is_file() else find_preset(args.theme)
    if preset_path is None:
        die(f"unknown theme '{args.theme}' (try `nomarchy-theme-sync list`)")

    preset = json.loads(preset_path.read_text())
    # Merge over current state: presets define palette/name/wallpaper,
    # user tweaks (gaps, fonts) outside the preset survive.
    state = deep_merge(load_state(), preset)
    write_state(state)
    log(f"theme: {state.get('name', args.theme)}")
    check_fonts(state)
    if not args.no_switch:
        run_switch()
        apply_wallpaper(state)


def cmd_set(args) -> None:
    state = load_state()
    try:
        value = json.loads(args.value)  # numbers, bools, null, quoted strings
    except json.JSONDecodeError:
        value = args.value  # bare string ("#7aa2f7", font names, paths)

    node = state
    keys = args.path.split(".")
    for key in keys[:-1]:
        node = node.setdefault(key, {})
        if not isinstance(node, dict):
            die(f"cannot descend into non-object at '{key}' in '{args.path}'")
    node[keys[-1]] = value

    write_state(state)
    log(f"set {args.path} = {value!r}")
    if args.path.startswith("fonts."):
        check_fonts(state)
    if not args.no_switch:
        run_switch()
        # Only wallpaper/theme keys change what swww shows; skip the re-apply
        # (and its transition) for unrelated keys like ui.* or settings.* so a
        # gaps tweak or a feature toggle doesn't flash the background.
        if args.path.split(".")[0] in ("wallpaper", "slug"):
            apply_wallpaper(state)


def cmd_get(args) -> None:
    state = load_state()
    if args.path:
        node = state
        for key in args.path.split("."):
            try:
                node = node[key]
            except (KeyError, TypeError):
                die(f"no such key: {args.path}")
        print(json.dumps(node, indent=2) if isinstance(node, (dict, list)) else node)
    else:
        print(json.dumps(state, indent=2))


def cmd_wallpaper(_args) -> None:
    """Apply the current wallpaper (session start, post-switch hook)."""
    apply_wallpaper(load_state(), wait=True)


def cmd_bg(args) -> None:
    """Cycle the theme's wallpapers — instant, no rebuild needed (the
    wallpaper is runtime state for swww; nothing in Nix consumes it)."""
    state = load_state()
    if args.action == "auto":
        state["wallpaper"] = ""
    else:  # next
        backgrounds = backgrounds_for(state.get("slug", ""))
        if not backgrounds:
            die(f"theme '{state.get('slug', '?')}' has no backgrounds/ directory")
        current = resolve_wallpaper(state)
        idx = (backgrounds.index(current) + 1) % len(backgrounds) if current in backgrounds else 0
        state["wallpaper"] = str(backgrounds[idx])
    write_state(state)
    apply_wallpaper(state)


# ─── Entry point ──────────────────────────────────────────────────────────

def main() -> None:
    global QUIET

    parser = argparse.ArgumentParser(
        prog="nomarchy-theme-sync",
        description="Nomarchy theming — state writer + Home Manager rebuild dispatcher.",
    )
    parser.add_argument("--quiet", action="store_true", help="suppress progress output")
    sub = parser.add_subparsers(dest="command", required=True)

    sub.add_parser("list", help="list theme presets").set_defaults(func=cmd_list)

    p = sub.add_parser("apply", help="apply a theme preset (name or JSON file) + rebuild")
    p.add_argument("theme")
    p.add_argument("--no-switch", action="store_true", help="write state only, skip the rebuild")
    p.set_defaults(func=cmd_apply)

    p = sub.add_parser("set", help="set one key, e.g. `set ui.gapsOut 16` + rebuild")
    p.add_argument("path")
    p.add_argument("value")
    p.add_argument("--no-switch", action="store_true", help="write state only, skip the rebuild")
    p.set_defaults(func=cmd_set)

    p = sub.add_parser("get", help="print state (or one dotted key)")
    p.add_argument("path", nargs="?")
    p.set_defaults(func=cmd_get)

    sub.add_parser("wallpaper", help="apply the current wallpaper via swww").set_defaults(func=cmd_wallpaper)

    p = sub.add_parser("bg", help="wallpaper control: `bg next` cycles, `bg auto` resets")
    p.add_argument("action", choices=["next", "auto"], default="next", nargs="?")
    p.set_defaults(func=cmd_bg)

    args = parser.parse_args()
    QUIET = args.quiet
    try:
        args.func(args)
    except KeyboardInterrupt:
        sys.exit(130)


if __name__ == "__main__":
    main()