148 lines
6.1 KiB
Python
148 lines
6.1 KiB
Python
"""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"""
|
|
<node>
|
|
<interface name="{_IFACE}">
|
|
<method name="RegisterStatusNotifierItem">
|
|
<arg name="service" type="s" direction="in"/>
|
|
</method>
|
|
<method name="RegisterStatusNotifierHost">
|
|
<arg name="service" type="s" direction="in"/>
|
|
</method>
|
|
<property name="RegisteredStatusNotifierItems" type="as" access="read"/>
|
|
<property name="IsStatusNotifierHostRegistered" type="b" access="read"/>
|
|
<property name="ProtocolVersion" type="i" access="read"/>
|
|
<signal name="StatusNotifierItemRegistered"><arg name="service" type="s"/></signal>
|
|
<signal name="StatusNotifierItemUnregistered"><arg name="service" type="s"/></signal>
|
|
<signal name="StatusNotifierHostRegistered"/>
|
|
<signal name="StatusNotifierHostUnregistered"/>
|
|
</interface>
|
|
</node>
|
|
"""
|
|
|
|
|
|
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, [])))
|