138 lines
5.1 KiB
Python
138 lines
5.1 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:
|
|
try:
|
|
self._watcher = Gio.DBusProxy.new_for_bus_sync(
|
|
Gio.BusType.SESSION, Gio.DBusProxyFlags.NONE, None,
|
|
_WATCHER_BUS, _WATCHER_PATH, _WATCHER_IFACE, None,
|
|
)
|
|
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
|
|
try:
|
|
self._watcher.call_sync(
|
|
"RegisterStatusNotifierHost", GLib.Variant("(s)", (_HOST_NAME,)),
|
|
Gio.DBusCallFlags.NONE, 500, None,
|
|
)
|
|
except GLib.Error:
|
|
pass
|
|
self._registered_host = True
|
|
|
|
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: self._context_menu(p))
|
|
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)
|
|
|
|
@staticmethod
|
|
def _context_menu(proxy: Gio.DBusProxy) -> None:
|
|
proxy.call("ContextMenu", GLib.Variant("(ii)", (0, 0)),
|
|
Gio.DBusCallFlags.NONE, -1, None, None, None)
|