Bernardo Magri
76941c55b7
nixpkgs wraps waybar, so the running comm is `.waybar-wrapped`: `pkill -x waybar` (state-sync restart/SIGUSR2 branches, tz-watch, updates) matched nothing — a theme switch neither restarted nor reloaded the bar, which then kept the gtk.css of whatever theme it started under (the stale workspace-digit colors reported 2026-07-18; companion gtk fix in the next commit). The unanchored pokes (recording, airplane, theme-shot) had the opposite bug: `waybar` substring-matches the supervisor's `nomarchy-waybar` comm too, and an unhandled RTMIN terminates its bash — the pokes were killing the crash guard. All seven sites now use `pkill … -x 'waybar|\.waybar-wrapped'`. state-sync 0.5.1 → 0.5.2. Verified live on Newton (V3-grade): new pattern restarts the bar under the supervisor (fresh pid, supervisor alive); RTMIN+10 poke reaches the bar, supervisor survives. V0 `nix flake check --no-build` + py_compile; V1 HM generation + system toplevel build. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
751 lines
32 KiB
Python
751 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
"""nomarchy-state-sync — state writer for Nomarchy's declarative theming.
|
|
|
|
Single source of truth: $NOMARCHY_PATH/state.json (inside the flake,
|
|
read purely by Home Manager via the nomarchy.stateFile option).
|
|
Legacy name `theme-state.json` is still read (and migrated on write).
|
|
The old binary name `nomarchy-theme-sync` is a symlink to this tool.
|
|
|
|
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, Kitty, 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)
|
|
auto [--which] apply the day/night theme for the current time
|
|
validate check the state file against the schema (read-only)
|
|
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 re
|
|
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()
|
|
# Canonical name after #107. Legacy theme-state.json (and a brief theme.json
|
|
# misnomer some notes used) are still accepted for existing checkouts.
|
|
STATE_NAME = "state.json"
|
|
LEGACY_STATE_NAMES = ("theme-state.json", "theme.json")
|
|
|
|
|
|
def resolve_state_file(flake_dir: Path = FLAKE_DIR) -> Path:
|
|
preferred = flake_dir / STATE_NAME
|
|
if preferred.exists():
|
|
return preferred
|
|
for name in LEGACY_STATE_NAMES:
|
|
legacy = flake_dir / name
|
|
if legacy.exists():
|
|
return legacy
|
|
return preferred
|
|
|
|
|
|
STATE_FILE = resolve_state_file()
|
|
|
|
# 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
|
|
|
|
# argv[0] may still be the compatibility name nomarchy-theme-sync.
|
|
_TOOL = Path(sys.argv[0]).name or "nomarchy-state-sync"
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
if not QUIET:
|
|
print(f"{_TOOL}: {msg}")
|
|
|
|
|
|
def die(msg: str) -> "None":
|
|
print(f"{_TOOL}: 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-state-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 | None = None) -> dict:
|
|
path = path or resolve_state_file()
|
|
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"{path.name} has a JSON syntax error at line {e.lineno}, "
|
|
f"column {e.colno}: {e.msg}\n"
|
|
f" file: {path}\n"
|
|
f" fix: correct the syntax by hand (a trailing comma is the "
|
|
f"usual culprit) — nothing was changed")
|
|
|
|
|
|
def write_state(state: dict) -> None:
|
|
"""Atomic write: render to a temp file in the same dir, then rename.
|
|
Validates first — an invalid state never reaches disk (the old file
|
|
stays untouched), so a bad `set` can't brick the next rebuild.
|
|
Always writes the canonical state.json; removes a legacy-named sibling
|
|
so a checkout never has two competing sources of truth."""
|
|
errors, warnings = validate_state(state)
|
|
for w in warnings:
|
|
log(f"warning: {w}")
|
|
if errors:
|
|
die("refusing to write an invalid state (nothing changed):\n ✖ "
|
|
+ "\n ✖ ".join(errors))
|
|
out = FLAKE_DIR / STATE_NAME
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp = tempfile.mkstemp(dir=out.parent, prefix=".state.", suffix=".json")
|
|
try:
|
|
with os.fdopen(fd, "w") as f:
|
|
json.dump(state, f, indent=2)
|
|
f.write("\n")
|
|
os.replace(tmp, out)
|
|
except BaseException:
|
|
os.unlink(tmp)
|
|
raise
|
|
for name in LEGACY_STATE_NAMES:
|
|
legacy = FLAKE_DIR / name
|
|
if legacy.exists() and legacy.resolve() != out.resolve():
|
|
try:
|
|
legacy.unlink()
|
|
log(f"migrated: removed legacy {name} (now {STATE_NAME})")
|
|
except OSError as e:
|
|
log(f"warning: could not remove legacy {name}: {e}")
|
|
log(f"state written to {out}")
|
|
|
|
# Flakes only see git-tracked files. 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", STATE_NAME],
|
|
capture_output=True,
|
|
)
|
|
for name in LEGACY_STATE_NAMES:
|
|
subprocess.run(
|
|
["git", "-C", str(FLAKE_DIR), "rm", "-f", "--ignore-unmatch", name],
|
|
capture_output=True,
|
|
)
|
|
|
|
|
|
def auto_commit_enabled(state: dict) -> bool:
|
|
return bool((state.get("settings") or {}).get("autoCommit"))
|
|
|
|
|
|
def auto_commit(message: str) -> None:
|
|
"""Opt-in (settings.autoCommit): commit state.json — and nothing
|
|
else — after a mutation, so settings history is `git log`. The pathspec
|
|
keeps unrelated dirty files out of the commit; a missing git identity
|
|
falls back to a Nomarchy one so a fresh machine never errors. Callers
|
|
fire this when the flag is on before OR after the write, so the
|
|
disable-toggle itself lands in history instead of staying dirty.
|
|
`bg` is deliberately excluded (runtime wallpaper churn); the wallpaper
|
|
path rides along with the next apply/set commit. The rest of a dirty
|
|
tree (hand edits, lock bumps) is swept by nomarchy-lifecycle's
|
|
nomarchy-autocommit right before a pull/rebuild/home switch — same
|
|
flag, honest message — so it never rides a settings-named commit."""
|
|
if not (FLAKE_DIR / ".git").exists() or shutil.which("git") is None:
|
|
return
|
|
git = ["git", "-C", str(FLAKE_DIR)]
|
|
# Pathspec: preferred name plus any legacy file git still knows — in the
|
|
# index or in HEAD (a migration's staged `git rm` needs the pathspec to
|
|
# ride along). Names known to neither MUST be dropped: unlike `git diff`,
|
|
# `git commit` aborts on a pathspec that matches nothing, which silently
|
|
# killed every auto-commit once the #107 rename removed the legacy files.
|
|
def known(name: str) -> bool:
|
|
return subprocess.run(git + ["ls-files", "--error-unmatch", "--", name],
|
|
capture_output=True).returncode == 0 \
|
|
or subprocess.run(git + ["cat-file", "-e", f"HEAD:{name}"],
|
|
capture_output=True).returncode == 0
|
|
paths = [STATE_NAME] + [n for n in LEGACY_STATE_NAMES if known(n)]
|
|
# No-op when the file already matches HEAD (a `set` to the same value).
|
|
# On a repo with no commits yet this diff errors — then just commit.
|
|
if subprocess.run(git + ["diff", "--quiet", "HEAD", "--"] + paths,
|
|
capture_output=True).returncode == 0:
|
|
return
|
|
if not subprocess.run(git + ["config", "user.email"],
|
|
capture_output=True, text=True).stdout.strip():
|
|
git += ["-c", "user.name=Nomarchy", "-c", "user.email=nomarchy@localhost"]
|
|
result = subprocess.run(
|
|
git + ["commit", "--quiet", "-m", message, "--"] + paths,
|
|
capture_output=True, text=True)
|
|
if result.returncode == 0:
|
|
log(f"auto-committed: {message}")
|
|
else:
|
|
log(f"auto-commit skipped: {(result.stderr or result.stdout).strip()}")
|
|
|
|
|
|
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
|
|
|
|
|
|
# ─── Schema validation ─────────────────────────────────────────────────────
|
|
# The friendly-errors contract (with theme.nix, which enforces the same
|
|
# rules at eval time): a hand-edited state file fails HERE, before it is
|
|
# written, with the field, the problem, and the fix — never as a Nix
|
|
# stack trace at rebuild. Hard errors cover the appearance schema the
|
|
# modules consume; settings.* (menu-writer territory) and unknown keys
|
|
# only warn, so custom keys and newer/older schemas keep working.
|
|
|
|
HEX_RE = re.compile(r"^#[0-9a-fA-F]{6}$")
|
|
COLOR_ROLES = ("base", "mantle", "surface", "overlay", "text", "subtext",
|
|
"muted", "accent", "accentAlt", "good", "warn", "bad")
|
|
TOP_KEYS = {"version", "name", "slug", "mode", "wallpaper", "colors", "ansi",
|
|
"fonts", "ui", "icons", "border", "settings"}
|
|
UI_UINTS = ("gapsIn", "gapsOut", "borderSize", "rounding", "iconSize")
|
|
UI_OPACITIES = ("activeOpacity", "inactiveOpacity", "terminalOpacity")
|
|
UI_BOOLS = ("blur", "shadow")
|
|
|
|
|
|
def _is_hex(v) -> bool:
|
|
return isinstance(v, str) and bool(HEX_RE.match(v))
|
|
|
|
|
|
def _is_num(v) -> bool:
|
|
return isinstance(v, (int, float)) and not isinstance(v, bool)
|
|
|
|
|
|
def validate_state(state) -> tuple:
|
|
"""Return (errors, warnings) — friendly strings, one problem each."""
|
|
errors, warnings = [], []
|
|
|
|
def err(field, problem, fix):
|
|
errors.append(f"{field}: {problem}\n fix: {fix}")
|
|
|
|
if not isinstance(state, dict):
|
|
err("top level", "must be a JSON object",
|
|
"re-apply a preset: nomarchy-state-sync apply boreal")
|
|
return errors, warnings
|
|
|
|
for k in state:
|
|
if k not in TOP_KEYS:
|
|
warnings.append(f"unknown top-level key '{k}' — a typo? (it is ignored)")
|
|
|
|
for field, want in (("name", "the theme's display name"),
|
|
("slug", "the theme's directory name"),
|
|
("wallpaper", "a filename in the theme's backgrounds/ (or \"\")"),
|
|
("icons", "an icon theme name (or \"\" for pick-by-mode)")):
|
|
v = state.get(field)
|
|
if v is not None and not isinstance(v, str):
|
|
err(field, f"must be a string ({want}), got {v!r}", 'quote it, e.g. "boreal"')
|
|
|
|
mode = state.get("mode")
|
|
if mode is not None and mode not in ("dark", "light"):
|
|
err("mode", f'must be "dark" or "light", got {mode!r}', 'set it to "dark" or "light"')
|
|
|
|
colors = state.get("colors", {})
|
|
if not isinstance(colors, dict):
|
|
err("colors", "must be an object of palette roles",
|
|
"re-apply a preset to restore the palette")
|
|
else:
|
|
for k, v in colors.items():
|
|
if k not in COLOR_ROLES:
|
|
warnings.append(f"colors.{k}: not a Nomarchy palette role (it is ignored)")
|
|
elif not _is_hex(v):
|
|
err(f"colors.{k}", f'must be "#RRGGBB", got {v!r}',
|
|
'use a 6-digit hex color like "#7aa2f7"')
|
|
|
|
ansi = state.get("ansi")
|
|
if ansi is not None and not (isinstance(ansi, list) and len(ansi) == 16
|
|
and all(_is_hex(c) for c in ansi)):
|
|
err("ansi", 'must be a list of exactly 16 "#RRGGBB" strings',
|
|
"re-apply a preset to restore the terminal palette")
|
|
|
|
fonts = state.get("fonts", {})
|
|
if isinstance(fonts, dict):
|
|
for k in ("mono", "ui"):
|
|
v = fonts.get(k)
|
|
if v is not None and not isinstance(v, str):
|
|
err(f"fonts.{k}", f"must be a font family name (string), got {v!r}",
|
|
'quote it, e.g. "JetBrainsMono Nerd Font"')
|
|
size = fonts.get("size")
|
|
if size is not None and not (_is_num(size) and size > 0):
|
|
err("fonts.size", f"must be a positive number (points), got {size!r}",
|
|
"use a plain number like 11 (no quotes)")
|
|
else:
|
|
err("fonts", "must be an object", 're-add: {"mono": "...", "ui": "...", "size": 11}')
|
|
|
|
ui = state.get("ui", {})
|
|
if isinstance(ui, dict):
|
|
for k in UI_UINTS:
|
|
v = ui.get(k)
|
|
if v is not None and not (isinstance(v, int) and not isinstance(v, bool) and v >= 0):
|
|
err(f"ui.{k}", f"must be a non-negative whole number (pixels), got {v!r}",
|
|
"use a plain number like 12 (no quotes)")
|
|
for k in UI_OPACITIES:
|
|
v = ui.get(k)
|
|
if v is not None and not (_is_num(v) and 0 <= v <= 1):
|
|
err(f"ui.{k}", f"must be a number between 0 and 1, got {v!r}",
|
|
"use e.g. 0.95 (no quotes)")
|
|
for k in UI_BOOLS:
|
|
v = ui.get(k)
|
|
if v is not None and not isinstance(v, bool):
|
|
err(f"ui.{k}", f"must be true or false, got {v!r}",
|
|
"use true or false (no quotes)")
|
|
else:
|
|
err("ui", "must be an object", "re-apply a preset to restore the UI block")
|
|
|
|
border = state.get("border")
|
|
if border is not None:
|
|
if not isinstance(border, dict):
|
|
err("border", "must be an object", 're-add: {"active": "accent", "inactive": "overlay"}')
|
|
else:
|
|
for k in ("active", "inactive"):
|
|
v = border.get(k)
|
|
if v is not None and not (isinstance(v, str)
|
|
and (v in COLOR_ROLES or _is_hex(v))):
|
|
err(f"border.{k}",
|
|
f'must be a palette role ({", ".join(COLOR_ROLES)}) or "#RRGGBB", got {v!r}',
|
|
'use e.g. "accent" or "#7aa2f7"')
|
|
|
|
settings = state.get("settings")
|
|
if settings is not None and not isinstance(settings, dict):
|
|
err("settings", "must be an object (the menu writes feature flags here)",
|
|
're-add: "settings": {}')
|
|
|
|
slug = state.get("slug")
|
|
if isinstance(slug, str) and slug and find_preset(slug) is None:
|
|
warnings.append(f"slug '{slug}' matches no shipped/custom preset — "
|
|
f"fine for a hand-rolled theme, but per-theme assets won't resolve")
|
|
|
|
return errors, warnings
|
|
|
|
|
|
def cmd_validate(_args) -> None:
|
|
state = load_state() # dies with the friendly syntax error on bad JSON
|
|
errors, warnings = validate_state(state)
|
|
for w in warnings:
|
|
print(f" ● {w}")
|
|
if errors:
|
|
print(f"nomarchy-state-sync: {STATE_FILE} has "
|
|
f"{len(errors)} problem(s):", file=sys.stderr)
|
|
for e in errors:
|
|
print(f" ✖ {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
log(f"{STATE_FILE} is valid ({len(warnings)} warning(s))")
|
|
|
|
|
|
# ─── 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-state-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:
|
|
# BACKLOG #56: point at doctor, not a silent wall of Nix noise.
|
|
notify(
|
|
"Rebuild FAILED — scroll up for last lines; run nomarchy-doctor",
|
|
urgency="critical",
|
|
)
|
|
# Live ISO pins only the *default* theme's HM generation. Applying
|
|
# another preset offline tries to build the world from source (#113).
|
|
offline_hint = ""
|
|
try:
|
|
import socket
|
|
socket.create_connection(("1.1.1.1", 53), timeout=1.5).close()
|
|
except OSError:
|
|
offline_hint = (
|
|
" Offline? Only the live ISO's already-pinned theme is "
|
|
"guaranteed without a network — other themes need builds "
|
|
"or a binary cache. Connect and retry, or apply the default "
|
|
"theme (boreal)."
|
|
)
|
|
die(
|
|
"rebuild failed (state already written; fix and re-run). "
|
|
"Diagnose: nomarchy-doctor. Recovery: docs/RECOVERY.md."
|
|
+ offline_hint
|
|
)
|
|
notify("Changes applied ✓")
|
|
# Waybar runs from Hyprland's exec-once (not a systemd unit HM would
|
|
# restart on switch), so nudge the running bar onto the freshly rebuilt
|
|
# config/style. Under the nomarchy-waybar supervisor a plain kill IS a
|
|
# clean restart with the new files — waybar's in-place SIGUSR2 reload
|
|
# is what crashed the bar on hardware (double-reload race with
|
|
# reload_style_on_change while the symlinks flip), so prefer the
|
|
# restart whenever the supervisor is there to catch it; fall back to
|
|
# SIGUSR2 for unsupervised/custom bars. No-ops if nothing is running.
|
|
# The pattern matches both comm names: nixpkgs wraps the binary, so
|
|
# the process is `.waybar-wrapped` — a bare `pkill -x waybar` matched
|
|
# nothing, theme switches never touched the bar, and it kept the
|
|
# gtk.css of whatever theme it was started under (the 2026-07-19
|
|
# stale-digit-color bug). Keep -x: unanchored `waybar` also matches
|
|
# the supervisor's own `nomarchy-waybar` comm and would kill it.
|
|
if shutil.which("pkill"):
|
|
supervised = subprocess.run(
|
|
["pgrep", "-f", "nomarchy-waybar"], capture_output=True
|
|
).returncode == 0
|
|
if supervised:
|
|
subprocess.run(["pkill", "-x", r"waybar|\.waybar-wrapped"], check=False)
|
|
else:
|
|
subprocess.run(
|
|
["pkill", "--signal", "SIGUSR2", "-x", r"waybar|\.waybar-wrapped"],
|
|
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 apply_named(theme: str, no_switch: bool = False) -> None:
|
|
"""Merge a preset (name or JSON file) into the state and rebuild. Shared
|
|
by `apply` and `auto` so the day/night switch is the *same* one engine."""
|
|
candidate = Path(theme).expanduser()
|
|
preset_path = candidate if candidate.is_file() else find_preset(theme)
|
|
if preset_path is None:
|
|
die(f"unknown theme '{theme}' (try `nomarchy-state-sync list`)")
|
|
|
|
preset = json.loads(preset_path.read_text())
|
|
# Merge over current state. Presets carry a full appearance block —
|
|
# palette, border, fonts and ui — so a switch always replaces the last
|
|
# theme's look (a theme can ship a bespoke font/opacity/geometry, e.g.
|
|
# Boreal, without it leaking into the next). settings.* (feature state,
|
|
# keyboard/monitor memory) live outside any preset and survive. A
|
|
# per-key `set ui.<k>`/`fonts.<k>` tweak holds until the next apply.
|
|
old = load_state()
|
|
state = deep_merge(old, preset)
|
|
write_state(state)
|
|
if auto_commit_enabled(old) or auto_commit_enabled(state):
|
|
auto_commit(f"nomarchy: apply theme {state.get('name', theme)}")
|
|
log(f"theme: {state.get('name', theme)}")
|
|
check_fonts(state)
|
|
if not no_switch:
|
|
run_switch()
|
|
apply_wallpaper(state)
|
|
|
|
|
|
def cmd_apply(args) -> None:
|
|
apply_named(args.theme, no_switch=args.no_switch)
|
|
|
|
|
|
def _hhmm_to_min(s: str, default: str) -> int:
|
|
"""'HH:MM' → minutes since midnight; falls back to `default` on garbage."""
|
|
for candidate in (s, default):
|
|
try:
|
|
h, m = str(candidate).split(":")
|
|
return int(h) * 60 + int(m)
|
|
except (ValueError, AttributeError):
|
|
continue
|
|
return 0
|
|
|
|
|
|
def cmd_auto(args) -> None:
|
|
"""Apply the day or night theme for the current local time, per
|
|
settings.autoTheme = {enable, day, night, sunrise, sunset}. Slice 1 of
|
|
the auto time-of-day pair (BACKLOG #79): the timer (slice 2) and menu
|
|
(slice 3) drive this; it can also be run by hand or from a user cron."""
|
|
state = load_state()
|
|
at = (state.get("settings") or {}).get("autoTheme") or {}
|
|
if not at.get("enable"):
|
|
if not args.which:
|
|
log("auto-theme: off (enable via settings.autoTheme.enable)")
|
|
return
|
|
day, night = at.get("day"), at.get("night")
|
|
if not day or not night:
|
|
die("auto-theme needs settings.autoTheme.day and .night (theme slugs)")
|
|
|
|
sunrise = _hhmm_to_min(at.get("sunrise", "07:00"), "07:00")
|
|
sunset = _hhmm_to_min(at.get("sunset", "20:00"), "20:00")
|
|
now = time.localtime()
|
|
mins = now.tm_hour * 60 + now.tm_min
|
|
# Daytime is the span between sunrise and sunset. Also handle a
|
|
# wrap-around pair (sunrise after sunset → the day window crosses
|
|
# midnight), so an inverted schedule still switches both ways.
|
|
if sunrise <= sunset:
|
|
daytime = sunrise <= mins < sunset
|
|
else:
|
|
daytime = mins >= sunrise or mins < sunset
|
|
target = day if daytime else night
|
|
|
|
if args.which:
|
|
print(target)
|
|
return
|
|
if state.get("slug") == target and not args.force:
|
|
log(f"auto-theme: already on {target} ({'day' if daytime else 'night'})")
|
|
return
|
|
apply_named(target, no_switch=args.no_switch)
|
|
|
|
|
|
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)
|
|
|
|
was_on = auto_commit_enabled(state)
|
|
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)
|
|
if was_on or auto_commit_enabled(state):
|
|
auto_commit(f"nomarchy: set {args.path} = {json.dumps(value)}")
|
|
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}")
|
|
# Booleans and null print JSON-style (true/false/null, not Python's
|
|
# True/False/None) so shell consumers can compare against the same
|
|
# literal they `set`. null matters for genuinely-nullable keys like
|
|
# settings.greeter.autoLogin, where "None" would silently miss every
|
|
# `case ... null)` a caller writes.
|
|
print(json.dumps(node, indent=2)
|
|
if node is None or isinstance(node, (dict, list, bool)) else node)
|
|
else:
|
|
print(json.dumps(state, indent=2))
|
|
|
|
|
|
def cmd_wallpaper(_args) -> None:
|
|
"""Apply the current wallpaper (session start, switch, output hotplug)."""
|
|
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-state-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)
|
|
|
|
p = sub.add_parser("auto", help="apply the day/night theme for the current time (settings.autoTheme)")
|
|
p.add_argument("--which", action="store_true", help="print the theme that WOULD apply, don't switch")
|
|
p.add_argument("--force", action="store_true", help="apply even if already on the target theme")
|
|
p.add_argument("--no-switch", action="store_true", help="write state only, skip the rebuild")
|
|
p.set_defaults(func=cmd_auto)
|
|
|
|
sub.add_parser("validate", help="check state.json against the schema (read-only)").set_defaults(func=cmd_validate)
|
|
|
|
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()
|