151 lines
4.5 KiB
Python
151 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""QuickShell wallpaper picker backend.
|
|
|
|
Wallpapers are the image files in ~/Pictures/wallpapers (hyprpaper reads that
|
|
dir as the initial set; the picker surfaces whatever is there at runtime).
|
|
|
|
Commands
|
|
--------
|
|
list -> JSON array of {name, path, active} (active = matches the state file)
|
|
set <p> -> persist <p> to ~/.cache/quickshell/wallpaper, then apply it now
|
|
apply -> re-apply the persisted choice (login autostart; retries while
|
|
hyprpaper's control socket comes up)
|
|
path -> print the persisted path (empty if never picked)
|
|
|
|
Apply mechanism
|
|
---------------
|
|
hyprpaper >= 0.8 ships a new control protocol: the legacy `preload`/`reload`
|
|
IPC commands are gone, and `hyprctl hyprpaper wallpaper <monitor>,<path>`
|
|
takes a real output name (no `*`). We query `hyprctl monitors -j` for the
|
|
current output names and apply to each. Persisting to a cache file (not
|
|
hyprpaper.conf) keeps the Nix-generated config authoritative; the login
|
|
autostart line in hyprland.nix re-applies the choice after hyprpaper starts,
|
|
defaulting to the config wallpaper until the user picks something.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
HOME = os.path.expanduser("~")
|
|
WALL_DIR = os.path.join(HOME, "Pictures", "wallpapers")
|
|
CACHE_DIR = os.path.join(HOME, ".cache", "quickshell")
|
|
STATE_FILE = os.path.join(CACHE_DIR, "wallpaper")
|
|
|
|
IMAGE_EXT = (".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif")
|
|
HYPRCTL = "hyprctl"
|
|
|
|
|
|
def ensure_cache_dir():
|
|
os.makedirs(CACHE_DIR, exist_ok=True)
|
|
|
|
|
|
def scan_names():
|
|
if not os.path.isdir(WALL_DIR):
|
|
return []
|
|
try:
|
|
names = os.listdir(WALL_DIR)
|
|
except OSError:
|
|
return []
|
|
return sorted(n for n in names if n.lower().endswith(IMAGE_EXT))
|
|
|
|
|
|
def read_state():
|
|
try:
|
|
with open(STATE_FILE) as fh:
|
|
path = fh.read().strip()
|
|
return path if os.path.isfile(path) else ""
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def list_walls():
|
|
current = read_state()
|
|
entries = []
|
|
for name in scan_names():
|
|
path = os.path.join(WALL_DIR, name)
|
|
entries.append({"name": name, "path": path, "active": path == current})
|
|
return entries
|
|
|
|
|
|
def monitors():
|
|
"""Current Hyprland output names (e.g. eDP-1), empty if unavailable."""
|
|
try:
|
|
out = subprocess.run(
|
|
[HYPRCTL, "monitors", "-j"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
except OSError:
|
|
return []
|
|
if out.returncode != 0:
|
|
return []
|
|
try:
|
|
data = json.loads(out.stdout)
|
|
return [m["name"] for m in data if "name" in m]
|
|
except (ValueError, KeyError, TypeError):
|
|
return []
|
|
|
|
|
|
def apply_path(path, retries=10, delay=0.5):
|
|
"""Apply via hyprctl hyprpaper, absorbing the login socket race."""
|
|
if not os.path.isfile(path):
|
|
return False, "no such file: %s" % path
|
|
last = ""
|
|
for _ in range(retries):
|
|
mons = monitors()
|
|
if not mons:
|
|
last = "no monitors available"
|
|
time.sleep(delay)
|
|
continue
|
|
try:
|
|
errs = []
|
|
ok = True
|
|
for mon in mons:
|
|
setw = subprocess.run(
|
|
[HYPRCTL, "hyprpaper", "wallpaper", mon + "," + path],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if setw.returncode != 0:
|
|
ok = False
|
|
errs.append(setw.stderr.strip())
|
|
except OSError as exc:
|
|
return False, "hyprctl unavailable: %s" % exc
|
|
if ok:
|
|
return True, ""
|
|
last = "; ".join(errs) or "hyprctl hyprpaper wallpaper failed"
|
|
time.sleep(delay)
|
|
return False, last
|
|
|
|
|
|
def set_wall(path):
|
|
if not os.path.isfile(path):
|
|
return {"ok": False, "error": "no such file: %s" % path}
|
|
ensure_cache_dir()
|
|
with open(STATE_FILE, "w") as fh:
|
|
fh.write(path + "\n")
|
|
ok, err = apply_path(path)
|
|
return {"ok": ok, "stderr": err}
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
cmd = args[0] if args else "list"
|
|
if cmd == "list":
|
|
print(json.dumps(list_walls()))
|
|
elif cmd == "set":
|
|
print(json.dumps(set_wall(args[1] if len(args) > 1 else "")))
|
|
elif cmd == "apply":
|
|
current = read_state()
|
|
print(json.dumps(set_wall(current) if current else {"ok": True, "stderr": ""}))
|
|
elif cmd == "path":
|
|
print(read_state())
|
|
else:
|
|
print(json.dumps({"ok": False, "error": "unknown command: " + cmd}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |