Dotfiles/desktopenvs/hyprdrive/station-bar/hypr_ipc.py

"""Event-driven Hyprland state tracking via the IPC event socket (.socket2.sock).

A status bar is always visible, so polling hyprctl on a timer (like horizon-dock's
"refresh when revealed" windows.py) would waste cycles or lag behind real events.
Instead this opens one persistent connection and streams parsed events as they
arrive — the same source `scripts/workspace` (legacy eww fallback) reads via
`socat`, but consumed natively through GLib/Gio so no subprocess is needed.

Initial state (on startup, before any event has fired) comes from a one-off
`hyprctl -j` round trip, same pattern as horizon-dock's windows.py.
"""

from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path
from typing import Callable, Optional

import gi

gi.require_version("Gio", "2.0")
from gi.repository import Gio, GLib  # noqa: E402


def _socket_path() -> Optional[Path]:
    sig = os.environ.get("HYPRLAND_INSTANCE_SIGNATURE")
    if not sig:
        return None
    runtime = os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")
    modern = Path(runtime) / "hypr" / sig / ".socket2.sock"
    if modern.exists():
        return modern
    legacy = Path("/tmp/hypr") / sig / ".socket2.sock"  # older Hyprland versions
    return legacy if legacy.exists() else None


def _hyprctl_json(*args: str) -> object:
    try:
        out = subprocess.run(["hyprctl", "-j", *args], capture_output=True,
                             text=True, timeout=1.5, check=True).stdout
        return json.loads(out)
    except (subprocess.SubprocessError, json.JSONDecodeError, OSError):
        return None


def initial_workspaces() -> list[dict]:
    data = _hyprctl_json("workspaces")
    return data if isinstance(data, list) else []


def initial_active_workspace_id() -> Optional[int]:
    data = _hyprctl_json("activeworkspace")
    return data.get("id") if isinstance(data, dict) else None


def initial_monitors() -> list[dict]:
    """Each monitor dict carries `name` (the connector, e.g. "DP-1", matching a
    Gdk.Monitor's connector) and `activeWorkspace: {id, name}` — the workspace
    that monitor is currently showing, independent of which monitor has focus.
    Used to give each per-monitor bar its own workspace list and active-highlight."""
    data = _hyprctl_json("monitors")
    return data if isinstance(data, list) else []


def initial_active_window_title() -> str:
    data = _hyprctl_json("activewindow")
    return (data.get("title") or "") if isinstance(data, dict) else ""


def focus_workspace(ws_id: int) -> None:
    subprocess.Popen(
        ["hyprctl", "dispatch", f"hl.dsp.focus({{ workspace = {ws_id} }})"],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )


class HyprIPC:
    """Connects once to .socket2.sock and dispatches parsed events.

    Callbacks:
      on_workspaces_changed()   a workspace was created/destroyed/moved — re-fetch
                                 the list via initial_workspaces()
      on_active_workspace(id)   focus moved to workspace `id`
      on_active_window(title)   the focused window's title changed (empty string
                                 when focus moved to no window)
    """

    def __init__(self, on_workspaces_changed: Callable[[], None],
                on_active_workspace: Callable[[int], None],
                on_active_window: Callable[[str], None]) -> None:
        self._on_workspaces_changed = on_workspaces_changed
        self._on_active_workspace = on_active_workspace
        self._on_active_window = on_active_window
        self._stream: Optional[Gio.DataInputStream] = None
        self._connect()

    def _connect(self) -> None:
        path = _socket_path()
        if path is None:
            return
        try:
            conn = Gio.SocketClient().connect(Gio.UnixSocketAddress.new(str(path)), None)
        except GLib.Error:
            return
        self._stream = Gio.DataInputStream.new(conn.get_input_stream())
        self._read_next()

    def _read_next(self) -> None:
        if self._stream is not None:
            self._stream.read_line_async(GLib.PRIORITY_DEFAULT, None, self._on_line)

    def _on_line(self, stream: Gio.DataInputStream, result: Gio.AsyncResult) -> None:
        try:
            line, _len = stream.read_line_finish_utf8(result)
        except GLib.Error:
            line = None
        if line is None:
            self._stream = None  # socket closed — stop rather than spin
            return
        self._dispatch(line)
        self._read_next()

    def _dispatch(self, line: str) -> None:
        if ">>" not in line:
            return
        event, _, payload = line.partition(">>")
        if event == "workspace":
            try:
                self._on_active_workspace(int(payload))
            except ValueError:
                pass  # special:* workspaces aren't shown as numbered stations
        elif event in ("createworkspace", "destroyworkspace", "moveworkspace",
                       "openwindow", "closewindow", "movewindow", "movewindowv2",
                       "focusedmon"):
            # any of these can change a workspace's window count OR which monitor
            # a workspace lives on (moveworkspace) OR a monitor's active workspace
            # (focusedmon) — all of which per-monitor bars (bar.py) filter on, so
            # re-fetch the list + re-derive each bar's active from `monitors`.
            self._on_workspaces_changed()
        elif event == "activewindow":
            _cls, _, title = payload.partition(",")
            self._on_active_window(title)