diff --git a/desktopenvs/hyprdrive/station-bar/main.py b/desktopenvs/hyprdrive/station-bar/main.py index 355fc09..50c6e39 100644 --- a/desktopenvs/hyprdrive/station-bar/main.py +++ b/desktopenvs/hyprdrive/station-bar/main.py @@ -35,6 +35,7 @@ import config # noqa: E402 import theme # noqa: E402 from bar import StationBar # noqa: E402 from paths import APP_ID # noqa: E402 +from sni_watcher import SniWatcher # noqa: E402 class StationBarApp(Gtk.Application): @@ -46,10 +47,14 @@ class StationBarApp(Gtk.Application): self.bars: dict[str, StationBar] = {} self._hologram = True self._visible = True + self._sni_watcher: SniWatcher | None = None def do_startup(self) -> None: Gtk.Application.do_startup(self) theme.load_css() + # Provide the StatusNotifierWatcher ourselves (no DE does it here), so tray + # apps have something to register with and the bars' tray fills up. + self._sni_watcher = SniWatcher() self._hologram = config.hologram_enabled() display = Gdk.Display.get_default() if display is not None: diff --git a/desktopenvs/hyprdrive/station-bar/sni_watcher.py b/desktopenvs/hyprdrive/station-bar/sni_watcher.py new file mode 100644 index 0000000..205cfb5 --- /dev/null +++ b/desktopenvs/hyprdrive/station-bar/sni_watcher.py @@ -0,0 +1,147 @@ +"""A minimal StatusNotifierWatcher (org.kde.StatusNotifierWatcher). + +Under a full desktop this bus name is provided by the panel/DE; hyprdrive has no +such component (we replaced waybar/eww), so tray apps have nowhere to register +and every SNI host — including our own bar's tray.py — sees an empty tray. This +owns the name, tracks registered items/hosts, and emits the registration signals ++ PropertiesChanged so hosts stay live as apps come and go. + +Kept deliberately small: it does not proxy the items themselves (the host talks +to each StatusNotifierItem directly), it just is the registry the spec requires. +""" + +from __future__ import annotations + +import gi + +gi.require_version("Gio", "2.0") +from gi.repository import Gio, GLib # noqa: E402 + +_NAME = "org.kde.StatusNotifierWatcher" +_PATH = "/StatusNotifierWatcher" +_IFACE = "org.kde.StatusNotifierWatcher" + +_XML = f""" + + + + + + + + + + + + + + + + + +""" + + +class SniWatcher: + def __init__(self) -> None: + self._conn: Gio.DBusConnection | None = None + self._items: dict[str, str] = {} # "service/path" -> owner unique name + self._hosts: set[str] = set() + self._reg_id = 0 + # NAME_OWNER flags let a real DE watcher take over cleanly if one appears. + self._owner_id = Gio.bus_own_name( + Gio.BusType.SESSION, _NAME, + Gio.BusNameOwnerFlags.ALLOW_REPLACEMENT, + self._on_bus_acquired, None, self._on_name_lost, + ) + + # -- bus lifecycle ----------------------------------------------------- + def _on_bus_acquired(self, conn: Gio.DBusConnection, _name: str) -> None: + self._conn = conn + info = Gio.DBusNodeInfo.new_for_xml(_XML).interfaces[0] + try: + self._reg_id = conn.register_object( + _PATH, info, self._on_method, self._on_get_property, None) + except GLib.Error: + return + # drop items whose owner leaves the bus + conn.signal_subscribe( + "org.freedesktop.DBus", "org.freedesktop.DBus", "NameOwnerChanged", + "/org/freedesktop/DBus", None, Gio.DBusSignalFlags.NONE, + self._on_name_owner_changed) + + def _on_name_lost(self, _conn, _name) -> None: + # another watcher grabbed the name — step aside quietly + self._conn = None + + # -- method / property handlers --------------------------------------- + def _on_method(self, conn, sender, _path, _iface, method, params, invocation) -> None: + if method == "RegisterStatusNotifierItem": + self._register_item(sender, params.unpack()[0]) + invocation.return_value(None) + elif method == "RegisterStatusNotifierHost": + self._hosts.add(params.unpack()[0]) + conn.emit_signal(None, _PATH, _IFACE, "StatusNotifierHostRegistered", None) + self._emit_props({"IsStatusNotifierHostRegistered": GLib.Variant("b", True)}) + invocation.return_value(None) + else: + invocation.return_value(None) + + def _on_get_property(self, _conn, _sender, _path, _iface, prop): + if prop == "RegisteredStatusNotifierItems": + return GLib.Variant("as", list(self._items.keys())) + if prop == "IsStatusNotifierHostRegistered": + return GLib.Variant("b", bool(self._hosts)) + if prop == "ProtocolVersion": + return GLib.Variant("i", 0) + return None + + # -- registry bookkeeping --------------------------------------------- + @staticmethod + def _entry_for(sender: str, service: str) -> str: + # apps register either an object path (use the caller as the service), + # a full "busname/path", or just a bus name (default item path). + if service.startswith("/"): + return sender + service + if "/" in service: + return service + return service + "/StatusNotifierItem" + + def _register_item(self, sender: str, service: str) -> None: + if self._conn is None: + return + entry = self._entry_for(sender, service) + if entry in self._items: + return + self._items[entry] = sender + self._conn.emit_signal(None, _PATH, _IFACE, "StatusNotifierItemRegistered", + GLib.Variant("(s)", (entry,))) + self._emit_items_changed() + + def _on_name_owner_changed(self, _conn, _sender, _path, _iface, _signal, params) -> None: + name, _old, new_owner = params.unpack() + if new_owner: + return # a name was acquired, not lost + gone = [e for e, owner in self._items.items() + if owner == name or e.split("/", 1)[0] == name] + for e in gone: + del self._items[e] + if self._conn is not None: + self._conn.emit_signal(None, _PATH, _IFACE, "StatusNotifierItemUnregistered", + GLib.Variant("(s)", (e,))) + if name in self._hosts: + self._hosts.discard(name) + if gone: + self._emit_items_changed() + + # -- signalling -------------------------------------------------------- + def _emit_items_changed(self) -> None: + self._emit_props({"RegisteredStatusNotifierItems": + GLib.Variant("as", list(self._items.keys()))}) + + def _emit_props(self, changed: dict) -> None: + if self._conn is None: + return + self._conn.emit_signal( + None, _PATH, "org.freedesktop.DBus.Properties", "PropertiesChanged", + GLib.Variant("(sa{sv}as)", (_IFACE, changed, []))) diff --git a/desktopenvs/hyprdrive/station-bar/tray.py b/desktopenvs/hyprdrive/station-bar/tray.py index ba93e19..8832c9a 100644 --- a/desktopenvs/hyprdrive/station-bar/tray.py +++ b/desktopenvs/hyprdrive/station-bar/tray.py @@ -27,22 +27,31 @@ class TrayHost: self._on_change = on_change self._watcher: Optional[Gio.DBusProxy] = None self._registered_host = False - self._connect() + # 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 _connect(self) -> None: + def _on_watcher_appeared(self, _conn, _name, _owner) -> None: try: - proxy = Gio.DBusProxy.new_for_bus_sync( + self._watcher = Gio.DBusProxy.new_for_bus_sync( Gio.BusType.SESSION, Gio.DBusProxyFlags.NONE, None, _WATCHER_BUS, _WATCHER_PATH, _WATCHER_IFACE, None, ) - self._watcher = proxy if proxy.get_name_owner() is not None else None except GLib.Error: self._watcher = None return - if self._watcher is 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: