377 lines
13 KiB
Python
377 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# niri_overview.py
|
||
|
||
import sys
|
||
import os
|
||
import json
|
||
import subprocess
|
||
import signal
|
||
import tempfile
|
||
import fcntl
|
||
from collections import defaultdict
|
||
|
||
# --- Single-instance lock ---
|
||
LOCKFILE = os.path.join(tempfile.gettempdir(), "niri_overview.lock")
|
||
_lock_fp = open(LOCKFILE, "w")
|
||
try:
|
||
fcntl.flock(_lock_fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except BlockingIOError:
|
||
sys.stderr.write("Error: Another instance of niri_overview is already running.\n")
|
||
sys.exit(1)
|
||
|
||
import gi
|
||
gi.require_version("Gtk", "3.0")
|
||
gi.require_version("Gdk", "3.0")
|
||
gi.require_version("Pango", "1.0")
|
||
gi.require_version("Gio", "2.0")
|
||
gi.require_version("GLib", "2.0")
|
||
|
||
from gi.repository import Gtk, Gdk, Pango, Gio, GLib
|
||
|
||
import config
|
||
|
||
# --- Globals ---
|
||
monitor_windows = {}
|
||
niri_event_process = None
|
||
update_pending = False
|
||
|
||
ICON_SIZE_PX = config.ICON_SIZE_PX
|
||
FALLBACK_GLYPHS = config.FALLBACK_GLYPHS
|
||
DEFAULT_GLYPH = config.DEFAULT_GLYPH
|
||
WINDOW_WIDTH = config.WINDOW_WIDTH
|
||
CSS_PATH = config.STYLE_CSS_PATH
|
||
|
||
# --- Run niri commands ---
|
||
def run_niri_command(args, capture_output=True, check=True, timeout=2):
|
||
cmd = ["niri", "msg"] + args
|
||
try:
|
||
res = subprocess.run(cmd,
|
||
capture_output=capture_output,
|
||
text=True,
|
||
check=check,
|
||
timeout=timeout)
|
||
if capture_output and "--json" in args:
|
||
return json.loads(res.stdout) if res.stdout.strip() else []
|
||
return True
|
||
except Exception as e:
|
||
print(f"niri msg error {' '.join(args)}: {e}", file=sys.stderr)
|
||
return None
|
||
|
||
# --- Main window ---
|
||
class NiriMonitorWindow(Gtk.Window):
|
||
def __init__(self, output_name, output_data):
|
||
super().__init__(title=f"Niri Overview – {output_name}")
|
||
self.output_name = output_name
|
||
self.output_info = output_data.get("info", {})
|
||
self.workspaces = output_data.get("workspaces", [])
|
||
self.window_list = output_data.get("window_list", [])
|
||
|
||
# window setup
|
||
self.set_type_hint(Gdk.WindowTypeHint.DOCK)
|
||
self.set_keep_above(True)
|
||
self.set_decorated(False)
|
||
self.set_default_size(WINDOW_WIDTH, 250)
|
||
self.set_resizable(True)
|
||
self.set_can_focus(True)
|
||
self.grab_focus()
|
||
|
||
# position
|
||
pos = self.output_info.get("logical", {})
|
||
self.move(pos.get("x", 0), pos.get("y", 0))
|
||
|
||
# RGBA support
|
||
screen = self.get_screen()
|
||
visual = screen.get_rgba_visual() if screen else None
|
||
if visual and screen.is_composited():
|
||
self.set_visual(visual)
|
||
self.set_app_paintable(True)
|
||
self.connect("draw", self.on_draw)
|
||
|
||
# layout
|
||
self.main_vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
|
||
self.main_vbox.set_border_width(5)
|
||
self.add(self.main_vbox)
|
||
|
||
sc = Gtk.ScrolledWindow()
|
||
sc.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
|
||
sc.set_hexpand(True)
|
||
sc.set_vexpand(True)
|
||
self.main_vbox.pack_start(sc, True, True, 0)
|
||
|
||
self.content_vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
|
||
sc.add(self.content_vbox)
|
||
|
||
self.apply_css()
|
||
self.rebuild_content()
|
||
|
||
self.connect("delete-event", lambda *a: False)
|
||
self.connect("destroy", self.on_destroy)
|
||
self.connect("key-press-event", self.on_key_press)
|
||
|
||
def apply_css(self):
|
||
provider = Gtk.CssProvider()
|
||
provider.load_from_path(CSS_PATH)
|
||
screen = self.get_screen()
|
||
Gtk.StyleContext.add_provider_for_screen(
|
||
screen, provider, Gtk.STYLE_PROVIDER_PRIORITY_USER
|
||
)
|
||
|
||
def on_draw(self, widget, cr):
|
||
rgba = Gdk.RGBA()
|
||
rgba.parse(config.BACKGROUND_RGBA)
|
||
alloc = self.get_allocation()
|
||
cr.set_source_rgba(rgba.red, rgba.green, rgba.blue, rgba.alpha)
|
||
cr.rectangle(0, 0, alloc.width, alloc.height)
|
||
cr.fill()
|
||
return False
|
||
|
||
def on_destroy(self, widget):
|
||
monitor_windows.pop(self.output_name, None)
|
||
if not monitor_windows:
|
||
GLib.idle_add(quit_app)
|
||
|
||
def on_key_press(self, widget, event):
|
||
key = Gdk.keyval_name(event.keyval)
|
||
if key in ("q", "Escape"):
|
||
GLib.idle_add(quit_app)
|
||
return True
|
||
return False
|
||
|
||
def on_workspace_click(self, widget, event):
|
||
if event.button == 1:
|
||
run_niri_command(
|
||
["action", "focus-workspace", str(widget.workspace_index)],
|
||
capture_output=False
|
||
)
|
||
GLib.idle_add(quit_app)
|
||
return True
|
||
return False
|
||
|
||
def on_app_click(self, widget, event):
|
||
if event.button == 1:
|
||
run_niri_command(
|
||
["action", "focus-window", "--id", str(widget.window_id)],
|
||
capture_output=False
|
||
)
|
||
GLib.idle_add(quit_app)
|
||
return True
|
||
return False
|
||
|
||
def rebuild_content(self):
|
||
for child in self.content_vbox.get_children():
|
||
self.content_vbox.remove(child)
|
||
|
||
# output title
|
||
lbl = Gtk.Label(label=f"Output: {self.output_name}")
|
||
lbl.set_xalign(0.0)
|
||
lbl.get_style_context().add_class("output-title")
|
||
self.content_vbox.pack_start(lbl, False, False, 0)
|
||
|
||
ws_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=5)
|
||
self.content_vbox.pack_start(ws_box, False, False, 0)
|
||
|
||
by_ws = defaultdict(list)
|
||
for w in self.window_list:
|
||
by_ws[w.get("workspace_id")].append(w)
|
||
|
||
for ws in sorted(self.workspaces, key=lambda w: w.get("idx",0)):
|
||
eb = Gtk.EventBox()
|
||
eb.get_style_context().add_class("clickable")
|
||
eb.get_style_context().add_class("workspace")
|
||
if ws.get("is_focused"):
|
||
eb.get_style_context().add_class("workspace-focused")
|
||
eb.workspace_index = ws["idx"]
|
||
eb.connect("button-press-event", self.on_workspace_click)
|
||
|
||
vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=5)
|
||
eb.add(vbox)
|
||
|
||
title = f"Ws {ws['idx']}"
|
||
if ws.get("name"):
|
||
title += f" ({ws['name']})"
|
||
if ws.get("is_active"):
|
||
title += " ●"
|
||
t = Gtk.Label(label=title)
|
||
t.set_xalign(0.0)
|
||
t.get_style_context().add_class("ws-title")
|
||
vbox.pack_start(t, False, False, 0)
|
||
|
||
apps = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=5)
|
||
vbox.pack_start(apps, False, False, 0)
|
||
|
||
wins = by_ws.get(ws["id"], [])
|
||
if not wins:
|
||
empty = Gtk.Label(label="(empty)")
|
||
empty.set_halign(Gtk.Align.CENTER)
|
||
apps.pack_start(empty, True, True, 0)
|
||
else:
|
||
for w in sorted(wins, key=lambda x: x.get("id",0)):
|
||
aeb = Gtk.EventBox()
|
||
aeb.get_style_context().add_class("clickable")
|
||
aeb.get_style_context().add_class("app-container")
|
||
if w.get("is_focused"):
|
||
aeb.get_style_context().add_class("window-focused")
|
||
aeb.window_id = w["id"]
|
||
aeb.connect("button-press-event", self.on_app_click)
|
||
|
||
vb = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
|
||
aeb.add(vb)
|
||
|
||
# icon or glyph
|
||
ib = Gtk.Box()
|
||
ib.set_size_request(ICON_SIZE_PX, ICON_SIZE_PX)
|
||
for lookup in (
|
||
w.get("app_id","").lower(),
|
||
f"{w.get('app_id')}.desktop",
|
||
w.get("app_id","").split(".")[-1]
|
||
):
|
||
if Gtk.IconTheme.get_default().has_icon(lookup):
|
||
icon = Gio.ThemedIcon.new(lookup)
|
||
img = Gtk.Image.new_from_gicon(icon, Gtk.IconSize.LARGE_TOOLBAR)
|
||
img.set_pixel_size(ICON_SIZE_PX)
|
||
ib.pack_start(img, True, True, 0)
|
||
break
|
||
else:
|
||
glyph = FALLBACK_GLYPHS.get(w.get("app_id","").lower(), DEFAULT_GLYPH)
|
||
lab = Gtk.Label(label=glyph)
|
||
lab.get_style_context().add_class("glyph-label")
|
||
ib.pack_start(lab, True, True, 0)
|
||
|
||
vb.pack_start(ib, False, False, 0)
|
||
|
||
nm = Gtk.Label(label=w.get("title", w.get("app_id","")))
|
||
nm.set_ellipsize(Pango.EllipsizeMode.END)
|
||
nm.get_style_context().add_class("app-label")
|
||
vb.pack_start(nm, True, False, 0)
|
||
|
||
apps.pack_start(aeb, False, False, 0)
|
||
|
||
ws_box.pack_start(eb, False, False, 0)
|
||
|
||
self.content_vbox.show_all()
|
||
|
||
# --- Data fetch & UI refresh ---
|
||
def fetch_and_process_data():
|
||
outs = run_niri_command(["--json", "outputs"])
|
||
if outs is None:
|
||
return None
|
||
wss = run_niri_command(["--json", "workspaces"]) or []
|
||
wins = run_niri_command(["--json", "windows"]) or []
|
||
|
||
data = {}
|
||
for name, info in (outs.items() if isinstance(outs,dict) else []):
|
||
data[name] = {"info": {"name": name, **info}, "workspaces": [], "window_list": []}
|
||
|
||
ws_map = {}
|
||
for ws in wss:
|
||
out = ws.get("output")
|
||
if out in data:
|
||
data[out]["workspaces"].append(ws)
|
||
ws_map[ws.get("id")] = out
|
||
|
||
for w in wins:
|
||
out = ws_map.get(w.get("workspace_id"))
|
||
if out in data:
|
||
data[out]["window_list"].append(w)
|
||
|
||
return data
|
||
|
||
def refresh_ui_from_niri():
|
||
global monitor_windows, update_pending
|
||
update_pending = False
|
||
new = fetch_and_process_data()
|
||
if new is None:
|
||
if not monitor_windows and (not niri_event_process or niri_event_process.poll() is not None):
|
||
GLib.idle_add(quit_app)
|
||
return False
|
||
|
||
# remove stale
|
||
for out in set(monitor_windows) - set(new):
|
||
monitor_windows[out].destroy()
|
||
|
||
# update/create
|
||
for out, dat in new.items():
|
||
if out in monitor_windows:
|
||
w = monitor_windows[out]
|
||
w.output_info = dat["info"]
|
||
w.workspaces = dat["workspaces"]
|
||
w.window_list = dat["window_list"]
|
||
w.rebuild_content()
|
||
else:
|
||
w = NiriMonitorWindow(out, dat)
|
||
monitor_windows[out] = w
|
||
w.show_all()
|
||
|
||
if not monitor_windows and (not niri_event_process or niri_event_process.poll() is not None):
|
||
GLib.idle_add(quit_app)
|
||
|
||
return False
|
||
|
||
# --- Event stream ---
|
||
def read_niri_event_callback(src, cond):
|
||
global update_pending, niri_event_process
|
||
if cond & GLib.IO_HUP:
|
||
niri_event_process = None
|
||
if not update_pending:
|
||
update_pending = True
|
||
GLib.idle_add(refresh_ui_from_niri)
|
||
return False
|
||
|
||
if cond & GLib.IO_IN and niri_event_process:
|
||
line = niri_event_process.stdout.readline()
|
||
if line.strip():
|
||
try:
|
||
e = json.loads(line)
|
||
t = next(iter(e), None)
|
||
if t in {
|
||
"WorkspacesChanged","WindowsChanged","WindowOpenedOrChanged",
|
||
"WindowClosed","WorkspaceActivated","WorkspaceActiveWindowChanged",
|
||
"WindowFocusChanged","OutputAdded","OutputRemoved","OutputChanged"
|
||
} and not update_pending:
|
||
update_pending = True
|
||
GLib.idle_add(refresh_ui_from_niri)
|
||
except:
|
||
pass
|
||
return True
|
||
|
||
return False
|
||
|
||
def start_event_stream():
|
||
global niri_event_process
|
||
try:
|
||
niri_event_process = subprocess.Popen(
|
||
["niri","msg","--json","event-stream"],
|
||
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
|
||
text=True, bufsize=1
|
||
)
|
||
GLib.io_add_watch(
|
||
niri_event_process.stdout.fileno(),
|
||
GLib.IO_IN | GLib.IO_HUP,
|
||
read_niri_event_callback
|
||
)
|
||
except FileNotFoundError:
|
||
print("Error: 'niri' not found.", file=sys.stderr)
|
||
GLib.idle_add(quit_app)
|
||
|
||
# --- Quit & signal ---
|
||
def quit_app():
|
||
global niri_event_process
|
||
if niri_event_process and niri_event_process.poll() is None:
|
||
niri_event_process.terminate()
|
||
for w in list(monitor_windows.values()):
|
||
w.destroy()
|
||
if Gtk.main_level() > 0:
|
||
Gtk.main_quit()
|
||
|
||
def signal_handler(sig, frame):
|
||
GLib.idle_add(quit_app)
|
||
|
||
# --- Main ---
|
||
if __name__ == "__main__":
|
||
signal.signal(signal.SIGINT, signal_handler)
|
||
signal.signal(signal.SIGTERM, signal_handler)
|
||
start_event_stream()
|
||
refresh_ui_from_niri()
|
||
Gtk.main()
|
||
|