"""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)