145 lines
5.5 KiB
Python
145 lines
5.5 KiB
Python
"""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)
|