252 lines
10 KiB
Python
252 lines
10 KiB
Python
"""Minimal StatusNotifierItem (SNI) HOST — same protocol/approach as horizon-
|
|
dock's tray.py, but rendered inline directly on the always-visible bar instead
|
|
of inside a popover. Since there's no "rebuild on reveal" moment to hook, this
|
|
live-subscribes to the watcher's RegisteredStatusNotifierItems property so the
|
|
row updates itself as apps come and go, rather than rebuilding on a timer.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Callable, Optional
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
gi.require_version("Gio", "2.0")
|
|
from gi.repository import Gio, GLib, Gtk # noqa: E402
|
|
|
|
_WATCHER_BUS = "org.kde.StatusNotifierWatcher"
|
|
_WATCHER_PATH = "/StatusNotifierWatcher"
|
|
_WATCHER_IFACE = "org.kde.StatusNotifierWatcher"
|
|
_ITEM_IFACE = "org.kde.StatusNotifierItem"
|
|
_HOST_NAME = "org.kde.StatusNotifierHost-station-bar"
|
|
|
|
|
|
class TrayHost:
|
|
def __init__(self, on_change: Callable[[], None]) -> None:
|
|
self._on_change = on_change
|
|
self._watcher: Optional[Gio.DBusProxy] = None
|
|
self._registered_host = False
|
|
# Watch the name rather than connecting once: our own SniWatcher may not
|
|
# own it yet at construction, and this also recovers if the watcher (ours
|
|
# or a real DE's) restarts.
|
|
self._watch_id = Gio.bus_watch_name(
|
|
Gio.BusType.SESSION, _WATCHER_BUS, Gio.BusNameWatcherFlags.NONE,
|
|
self._on_watcher_appeared, self._on_watcher_vanished)
|
|
|
|
def _on_watcher_appeared(self, _conn, _name, _owner) -> None:
|
|
# MUST be async: this bar hosts the StatusNotifierWatcher itself (see
|
|
# sni_watcher.py), so a *_sync proxy here would block the main loop
|
|
# waiting on a reply only that same loop can produce — a self-deadlock
|
|
# that freezes the whole bar (its clock, its holo intro — everything)
|
|
# until the 25s D-Bus timeout finally lets it go.
|
|
Gio.DBusProxy.new_for_bus(
|
|
Gio.BusType.SESSION, Gio.DBusProxyFlags.NONE, None,
|
|
_WATCHER_BUS, _WATCHER_PATH, _WATCHER_IFACE, None,
|
|
self._on_watcher_ready, None,
|
|
)
|
|
|
|
def _on_watcher_ready(self, _source, result, _user_data) -> None:
|
|
try:
|
|
self._watcher = Gio.DBusProxy.new_for_bus_finish(result)
|
|
except GLib.Error:
|
|
self._watcher = None
|
|
return
|
|
self._watcher.connect("g-properties-changed", lambda *_a: self._on_change())
|
|
self._registered_host = False
|
|
self._ensure_host_registered()
|
|
self._on_change()
|
|
|
|
def _on_watcher_vanished(self, _conn, _name) -> None:
|
|
self._watcher = None
|
|
self._registered_host = False
|
|
self._on_change()
|
|
|
|
def _ensure_host_registered(self) -> None:
|
|
if self._registered_host or self._watcher is None:
|
|
return
|
|
# Async for the same self-deadlock reason as _on_watcher_appeared: this
|
|
# call targets our own in-process watcher. Fire-and-forget.
|
|
self._registered_host = True
|
|
self._watcher.call(
|
|
"RegisterStatusNotifierHost", GLib.Variant("(s)", (_HOST_NAME,)),
|
|
Gio.DBusCallFlags.NONE, 2000, None, None,
|
|
)
|
|
|
|
def available(self) -> bool:
|
|
return self._watcher is not None
|
|
|
|
def registered_items(self) -> list[str]:
|
|
if self._watcher is None:
|
|
return []
|
|
value = self._watcher.get_cached_property("RegisteredStatusNotifierItems")
|
|
return list(value.unpack()) if value is not None else []
|
|
|
|
@staticmethod
|
|
def _parse_item(entry: str) -> tuple[str, str]:
|
|
if "/" in entry:
|
|
bus, _, path = entry.partition("/")
|
|
return bus, "/" + path
|
|
return entry, "/StatusNotifierItem"
|
|
|
|
def build_into(self, box: Gtk.Box) -> None:
|
|
"""(Re)populate `box` with one small icon button per registered tray item."""
|
|
child = box.get_first_child()
|
|
while child is not None:
|
|
nxt = child.get_next_sibling()
|
|
box.remove(child)
|
|
child = nxt
|
|
|
|
for entry in self.registered_items():
|
|
bus, path = self._parse_item(entry)
|
|
btn = self._build_item_button(bus, path)
|
|
if btn is not None:
|
|
box.append(btn)
|
|
|
|
def _build_item_button(self, bus: str, path: str) -> Optional[Gtk.Button]:
|
|
try:
|
|
proxy = Gio.DBusProxy.new_for_bus_sync(
|
|
Gio.BusType.SESSION, Gio.DBusProxyFlags.NONE, None,
|
|
bus, path, _ITEM_IFACE, None,
|
|
)
|
|
except GLib.Error:
|
|
return None
|
|
|
|
icon_name = self._prop_str(proxy, "IconName") or "application-x-executable"
|
|
title = self._prop_str(proxy, "Title") or icon_name or bus
|
|
|
|
btn = Gtk.Button()
|
|
btn.set_has_frame(False)
|
|
btn.add_css_class("station-tray-item")
|
|
btn.set_tooltip_text(title)
|
|
image = Gtk.Image.new_from_icon_name(icon_name)
|
|
image.set_pixel_size(14)
|
|
btn.set_child(image)
|
|
btn.connect("clicked", lambda *_a, p=proxy: self._activate(p))
|
|
|
|
right_click = Gtk.GestureClick(button=3)
|
|
right_click.connect("pressed",
|
|
lambda *_a, p=proxy, b=btn: self._context_menu(p, b))
|
|
btn.add_controller(right_click)
|
|
return btn
|
|
|
|
@staticmethod
|
|
def _prop_str(proxy: Gio.DBusProxy, name: str) -> str:
|
|
value = proxy.get_cached_property(name)
|
|
return value.unpack() if value is not None else ""
|
|
|
|
@staticmethod
|
|
def _activate(proxy: Gio.DBusProxy) -> None:
|
|
proxy.call("Activate", GLib.Variant("(ii)", (0, 0)),
|
|
Gio.DBusCallFlags.NONE, -1, None, None, None)
|
|
|
|
# -- context menu (com.canonical.dbusmenu) --------------------------------
|
|
#
|
|
# SNI items don't hand us a positioned menu — most expose a dbusmenu object
|
|
# (the `Menu` property) that the HOST is expected to render itself. The old
|
|
# code just called the item's ContextMenu(0,0); apps that implement it (e.g.
|
|
# Discord) popped their menu at screen (0,0) — top-left — and apps that only
|
|
# do dbusmenu (nm-applet, blueman) got nothing at all. So instead we fetch
|
|
# the dbusmenu layout and render it in a GTK popover anchored to the icon.
|
|
|
|
def _context_menu(self, proxy: Gio.DBusProxy, button: Gtk.Button) -> None:
|
|
menu_path = self._prop_str(proxy, "Menu")
|
|
if not menu_path:
|
|
# No dbusmenu — last-ditch: ask the item to show its own context menu.
|
|
proxy.call("ContextMenu", GLib.Variant("(ii)", (0, 0)),
|
|
Gio.DBusCallFlags.NONE, -1, None, None, None)
|
|
return
|
|
try:
|
|
dm = Gio.DBusProxy.new_for_bus_sync(
|
|
Gio.BusType.SESSION, Gio.DBusProxyFlags.NONE, None,
|
|
proxy.get_name(), menu_path, "com.canonical.dbusmenu", None,
|
|
)
|
|
except GLib.Error:
|
|
return
|
|
|
|
# Best-effort: let the app refresh its menu before we read it.
|
|
try:
|
|
dm.call_sync("AboutToShow", GLib.Variant("(i)", (0,)),
|
|
Gio.DBusCallFlags.NONE, 800, None)
|
|
except GLib.Error:
|
|
pass
|
|
try:
|
|
res = dm.call_sync("GetLayout", GLib.Variant("(iias)", (0, -1, [])),
|
|
Gio.DBusCallFlags.NONE, 1500, None)
|
|
except GLib.Error:
|
|
return
|
|
_revision, layout = res.unpack()
|
|
|
|
actions = Gio.SimpleActionGroup()
|
|
model = self._build_model(dm, layout, actions)
|
|
|
|
popover = Gtk.PopoverMenu.new_from_model(model)
|
|
popover.add_css_class("station-tray-menu")
|
|
popover.set_parent(button)
|
|
popover.set_position(Gtk.PositionType.BOTTOM)
|
|
popover.insert_action_group("traymenu", actions)
|
|
# Keep a ref while shown, and tear the surface down cleanly on close.
|
|
self._open_popover = popover
|
|
popover.connect("closed", self._on_popover_closed)
|
|
popover.popup()
|
|
|
|
def _on_popover_closed(self, popover: Gtk.Popover) -> None:
|
|
popover.unparent()
|
|
if getattr(self, "_open_popover", None) is popover:
|
|
self._open_popover = None
|
|
|
|
def _build_model(self, dm: Gio.DBusProxy, node, actions: Gio.SimpleActionGroup) -> Gio.Menu:
|
|
"""Turn a dbusmenu (id, props, children) node into a Gio.Menu. Separators
|
|
split the run into sections (PopoverMenu draws a divider between them)."""
|
|
menu = Gio.Menu()
|
|
section = Gio.Menu()
|
|
|
|
def flush() -> None:
|
|
nonlocal section
|
|
if section.get_n_items() > 0:
|
|
menu.append_section(None, section)
|
|
section = Gio.Menu()
|
|
|
|
for child in node[2]:
|
|
cid, props, _kids = child
|
|
if not props.get("visible", True):
|
|
continue
|
|
if props.get("type") == "separator":
|
|
flush()
|
|
continue
|
|
|
|
label = props.get("label", "") or ""
|
|
if props.get("children-display") == "submenu":
|
|
item = Gio.MenuItem.new(label, None)
|
|
item.set_submenu(self._build_model(dm, child, actions))
|
|
section.append_item(item)
|
|
continue
|
|
|
|
aname = f"i{cid}"
|
|
if props.get("toggle-type") in ("checkmark", "radio"):
|
|
state = GLib.Variant("b", bool(props.get("toggle-state", 0)))
|
|
act = Gio.SimpleAction.new_stateful(aname, None, state)
|
|
else:
|
|
act = Gio.SimpleAction.new(aname, None)
|
|
act.set_enabled(bool(props.get("enabled", True)))
|
|
act.connect("activate", lambda _a, _p, i=cid: self._menu_event(dm, i))
|
|
actions.add_action(act)
|
|
|
|
item = Gio.MenuItem.new(label, f"traymenu.{aname}")
|
|
icon_name = props.get("icon-name")
|
|
if icon_name:
|
|
try:
|
|
item.set_icon(Gio.ThemedIcon.new(icon_name))
|
|
except GLib.Error:
|
|
pass
|
|
section.append_item(item)
|
|
|
|
flush()
|
|
return menu
|
|
|
|
@staticmethod
|
|
def _menu_event(dm: Gio.DBusProxy, item_id: int) -> None:
|
|
dm.call("Event",
|
|
GLib.Variant("(isvu)", (item_id, "clicked", GLib.Variant("i", 0), 0)),
|
|
Gio.DBusCallFlags.NONE, -1, None, None, None)
|