omarchy-setup/profiles/personal-desktop/plugin-patches/indie.herdr-hud--bin--herdr-web

#!/usr/bin/env python3
"""Show a Herdr pane in the herdr web ui (devswha.herdr-web-ui) as one floating app window.

The shell cannot host a web view (QtWebEngine aborts inside Quickshell), so the HUD
hands agents off to a browser app window instead. The HUD keeps a single window:
asking for the pane it already shows brings it forward; asking for another pane
replaces it in the same place and size. Failures go to a desktop notification,
since the HUD has closed by then.
"""
import json
import os
import subprocess
import sys
import time
from pathlib import Path
from urllib.parse import quote, urlsplit

HOME = Path.home()
WEB_ENV = HOME / ".config/herdr/plugins/config/devswha.herdr-web-ui/env"
CONFIG = HOME / ".config/herdr-hud-native/web.json"
STATE = HOME / ".config/herdr-hud-native/web-window.json"
WAIT_SECONDS = 10


class WebError(RuntimeError):
    pass


def base_url() -> str:
    """HERDR_HUD_WEB_URL, else {"url": ...} in web.json, else the web ui's own listen address."""
    override = os.environ.get("HERDR_HUD_WEB_URL", "").strip()
    if not override:
        try:
            override = str(json.loads(CONFIG.read_text()).get("url") or "").strip()
        except (OSError, ValueError, AttributeError):
            override = ""
    if override:
        return override.rstrip("/")
    values = {}
    try:
        for line in WEB_ENV.read_text().splitlines():
            key, sep, value = line.partition("=")
            if sep and key.strip() in ("HOST", "PORT"):
                values[key.strip()] = value.strip().strip("'\"")
    except OSError:
        pass
    host = values.get("HOST") or "127.0.0.1"
    if host in ("0.0.0.0", "::", "[::]"):
        host = "127.0.0.1"
    return f"http://{host}:{values.get('PORT') or '7317'}"


def hyprctl(*args: str) -> str:
    done = subprocess.run(["hyprctl", *args], capture_output=True, text=True, timeout=5)
    if done.returncode != 0 or done.stdout.startswith("Couldn't connect"):
        raise WebError(done.stdout.strip() or done.stderr.strip() or "hyprctl failed")
    return done.stdout


def fix_hyprland_signature() -> None:
    """A shell that outlived a Hyprland restart keeps a dead signature; use the live one."""
    try:
        hyprctl("version")
        return
    except WebError:
        pass
    runtime = Path(os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")) / "hypr"
    for candidate in sorted(runtime.glob("*/.socket.sock"), key=lambda p: p.stat().st_mtime, reverse=True):
        os.environ["HYPRLAND_INSTANCE_SIGNATURE"] = candidate.parent.name
        try:
            hyprctl("version")
            return
        except WebError:
            continue
    raise WebError("Cannot reach Hyprland")


def dispatch(expression: str) -> None:
    out = hyprctl("dispatch", expression).strip()
    if out != "ok":
        raise WebError(out)


def clients() -> list:
    return json.loads(hyprctl("clients", "-j"))


def web_windows(host: str) -> dict:
    # Chromium-family app windows are classed "<browser>-<host>__<path>-<profile>".
    marker = f"-{host}__"
    return {c["address"]: c for c in clients() if marker in str(c.get("class", ""))}


def load_state() -> dict:
    try:
        state = json.loads(STATE.read_text())
        return state if isinstance(state, dict) else {}
    except (OSError, ValueError):
        return {}


def save_state(state: dict) -> None:
    STATE.parent.mkdir(parents=True, exist_ok=True)
    tmp = STATE.with_suffix(".tmp")
    tmp.write_text(json.dumps(state, indent=2) + "\n")
    tmp.replace(STATE)


def active_workspace() -> str:
    return str(json.loads(hyprctl("activeworkspace", "-j")).get("id", ""))


def bring_forward(address: str) -> None:
    window = f"address:{address}"
    workspace = active_workspace()
    if workspace:
        dispatch(f"hl.dsp.window.move({{ workspace = '{workspace}', window = '{window}' }})")
    dispatch(f"hl.dsp.focus({{ window = '{window}' }})")


def default_size() -> tuple:
    for monitor in json.loads(hyprctl("monitors", "-j")):
        if monitor.get("focused"):
            scale = float(monitor.get("scale") or 1)
            width, height = monitor["width"] / scale, monitor["height"] / scale
            return round(width * 0.72), round(height * 0.82)
    return 1280, 860


def place(address: str, previous: dict | None) -> None:
    window = f"address:{address}"
    dispatch(f"hl.dsp.window.float({{ action = 'enable', window = '{window}' }})")
    if previous and previous.get("floating"):
        width, height = previous["size"]
        x, y = previous["at"]
        dispatch(f"hl.dsp.window.resize({{ x = {width}, y = {height}, window = '{window}' }})")
        dispatch(f"hl.dsp.window.move({{ x = {x}, y = {y}, window = '{window}' }})")
    else:
        width, height = default_size()
        dispatch(f"hl.dsp.window.resize({{ x = {width}, y = {height}, window = '{window}' }})")
        dispatch(f"hl.dsp.window.center({{ window = '{window}' }})")
    dispatch(f"hl.dsp.focus({{ window = '{window}' }})")


def open_pane(pane: str) -> None:
    base = base_url()
    host = urlsplit(base).hostname or ""
    if not host:
        raise WebError(f"Bad web ui address: {base}")
    fix_hyprland_signature()
    existing = web_windows(host)
    state = load_state()
    tracked = existing.get(str(state.get("address", "")))

    if tracked and state.get("pane") == pane:
        bring_forward(tracked["address"])
        return

    if tracked:
        dispatch(f"hl.dsp.window.close({{ window = 'address:{tracked['address']}' }})")

    url = f"{base}/?pane={quote(pane, safe='')}" if pane else f"{base}/"
    subprocess.Popen(["omarchy-launch-webapp", url], stdin=subprocess.DEVNULL,
                     stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)

    deadline = time.monotonic() + WAIT_SECONDS
    while time.monotonic() < deadline:
        time.sleep(0.15)
        fresh = [a for a in web_windows(host) if a not in existing]
        if fresh:
            place(fresh[0], tracked)
            save_state({"address": fresh[0], "pane": pane})
            return
    raise WebError("The web ui window did not appear")


def main() -> int:
    if len(sys.argv) != 3 or sys.argv[1] != "open":
        print("usage: herdr-web open PANE", file=sys.stderr)
        return 2
    try:
        open_pane(sys.argv[2])
    except (WebError, OSError, ValueError, KeyError, subprocess.TimeoutExpired) as error:
        subprocess.run(["notify-send", "Herdr HUD", f"Cannot open the web ui: {error}"], check=False)
        print(error, file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())