roundabout,
created on Saturday, 18 October 2025, 10:31:07 (1760783467),
received on Saturday, 18 October 2025, 10:31:15 (1760783475)
Author identity: Vlad <vlad.muntoiu@gmail.com>
0d3a95207c918cc52b0f87f7ce757f75b7ed2849
main.py
@@ -18,145 +18,161 @@ along with this program. If not, see <https://www.gnu.org/licenses/>.
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import GLib, Gio, GObject, Gtk
from pydbus import SessionBus
import pydbus.generic
class StatusNotifierWatcher:
StatusNotifierItemRegistered = pydbus.generic.signal()
StatusNotifierHostRegistered = pydbus.generic.signal()
StatusNotifierItemUnregistered = pydbus.generic.signal()
def __init__(self, bus):
super().__init__()
def __init__(self, connection: Gio.DBusConnection, interface_name: str):
self.connection = connection
self.interface_name = interface_name
self.items = {}
self.hosts = set()
self.items = set()
self.bus = bus
dbus_proxy = self.bus.get("org.freedesktop.DBus", "/org/freedesktop/DBus")
dbus_proxy.NameOwnerChanged.connect(self._on_name_owner_changed)
self.node_info = Gio.DBusNodeInfo.new_for_xml(f"""
<node>
<interface name="{interface_name}">
<method name="RegisterStatusNotifierItem">
<arg type="s" name="service" direction="in"/>
</method>
<method name="RegisterStatusNotifierHost">
<arg type="s" name="service" 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 type="s" name="service" direction="out"/>
</signal>
<signal name="StatusNotifierItemUnregistered">
<arg type="s" name="service" direction="out"/>
</signal>
<signal name="StatusNotifierHostRegistered">
<arg type="s" name="service" direction="out"/>
</signal>
</interface>
</node>
"""
)
self.interface_info = self.node_info.interfaces[0]
self.connection.register_object(
"/StatusNotifierWatcher",
self.interface_info,
self.handle_method_call,
self.get_property,
None
)
self.connection.signal_subscribe(
"org.freedesktop.DBus",
"org.freedesktop.DBus",
"NameOwnerChanged",
"/org/freedesktop/DBus",
None,
Gio.DBusSignalFlags.NONE,
self.on_name_owner_changed,
)
def _on_name_owner_changed(self, name, old_owner, new_owner):
def get_property(self, connection, sender, object_path, interface_name, property_name):
if property_name == "IsStatusNotifierHostRegistered":
return GLib.Variant("b", bool(self.hosts))
elif property_name == "RegisteredStatusNotifierItems":
return GLib.Variant("as", list(self.items))
elif property_name == "ProtocolVersion":
return GLib.Variant("i", 0)
else:
raise Exception(f"Unknown property {property_name}")
def on_name_owner_changed(self, connection, sender_name, object_path, interface_name, signal_name, params):
name, old_owner, new_owner = params.unpack()
if new_owner == "":
removed = False
for lookup in (name, old_owner):
if lookup in self.items:
self.items.discard(lookup)
self.StatusNotifierItemUnregistered(lookup)
path = self.items[lookup]
del self.items[lookup]
print(f"Item {lookup} vanished")
self.emit_signal("StatusNotifierItemUnregistered", lookup + ":" + path)
if lookup in self.hosts:
self.hosts.discard(lookup)
def RegisterStatusNotifierItem(self, service_name):
self.items.add(service_name)
self.StatusNotifierItemRegistered(service_name)
def RegisterStatusNotifierHost(self, service_name):
self.hosts.add(service_name)
self.StatusNotifierHostRegistered(service_name)
@property
def RegisteredStatusNotifierItems(self):
return sorted(self.items)
@property
def IsStatusNotifierHostRegistered(self):
return bool(self.hosts)
@property
def ProtocolVersion(self):
return 0
class KDEStatusNotifierWatcher(StatusNotifierWatcher):
"""
<node>
<interface name="org.kde.StatusNotifierWatcher">
<method name="RegisterStatusNotifierItem">
<arg type="s" name="service_name" direction="in"/>
</method>
<method name="RegisterStatusNotifierHost">
<arg type="s" name="service_name" 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 type="s" name="service" direction="out"/>
</signal>
<signal name="StatusNotifierItemUnregistered">
<arg type="s" name="service" direction="out"/>
</signal>
<signal name="StatusNotifierHostRegistered">
<arg type="s" name="service" direction="out"/>
</signal>
</interface>
</node>
"""
class FDStatusNotifierWatcher(StatusNotifierWatcher):
"""
<node>
<interface name="org.freedesktop.StatusNotifierWatcher">
<method name="RegisterStatusNotifierItem">
<arg type="s" name="service_name" direction="in"/>
</method>
<method name="RegisterStatusNotifierHost">
<arg type="s" name="service_name" 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 type="s" name="service" direction="out"/>
</signal>
<signal name="StatusNotifierItemUnregistered">
<arg type="s" name="service" direction="out"/>
</signal>
<signal name="StatusNotifierHostRegistered">
<arg type="s" name="service" direction="out"/>
</signal>
</interface>
</node>
"""
print(f"Host {lookup} vanished")
def handle_method_call(self, conn, sender, obj_path, iface_name, method, params, invocation):
if method == "RegisterStatusNotifierItem":
(arg_path,) = params.unpack()
self._register_item(sender, arg_path)
invocation.return_value(None)
elif method == "RegisterStatusNotifierHost":
(service,) = params.unpack()
self._register_host(sender, service)
invocation.return_value(None)
else:
invocation.return_dbus_error("org.freedesktop.DBus.Error.UnknownMethod", "Unknown method")
def _register_item(self, sender, path):
if sender not in self.items:
if not path.startswith("/"):
path = "/StatusNotifierItem"
self.items[sender] = path
print(f"Registered item {sender}")
self.emit_signal("StatusNotifierItemRegistered", sender + ":" + path)
def _register_host(self, sender, service):
if sender not in self.hosts:
self.hosts.add(sender)
print(f"Registered host {sender}")
self.emit_signal("StatusNotifierHostRegistered", sender)
def emit_signal(self, name, arg=None):
args = GLib.Variant("(s)", (arg,)) if arg else None
self.connection.emit_signal(
None,
"/StatusNotifierWatcher",
self.interface_name,
name,
args
)
class WatcherApplication(Gtk.Application):
def __init__(self):
super().__init__(application_id="com.roundabout_host.roundabout.PanoramaIndicatorService",
flags=Gio.ApplicationFlags.FLAGS_NONE)
self.bus = None
self.kde_publishing = None
self.fd_publishing = None
self.kde_watcher = None
super().__init__(
application_id="com.roundabout_host.roundabout.PanoramaIndicatorService",
flags=Gio.ApplicationFlags.FLAGS_NONE,
)
self.fd_watcher = None
self.kde_watcher = None
def do_startup(self):
Gtk.Application.do_startup(self)
self.bus = SessionBus()
self.kde_watcher = KDEStatusNotifierWatcher(self.bus)
self.fd_watcher = FDStatusNotifierWatcher(self.bus)
self.kde_publishing = self.bus.publish(
f"org.kde.StatusNotifierWatcher",
("/StatusNotifierWatcher", self.kde_watcher)
def do_activate(self):
print("StatusNotifierWatcher starting...")
Gio.bus_own_name(
Gio.BusType.SESSION,
"org.freedesktop.StatusNotifierWatcher",
Gio.BusNameOwnerFlags.NONE,
self.on_bus_acquired_fd,
None,
None,
)
self.fd_publishing = self.bus.publish(
f"org.freedesktop.StatusNotifierWatcher",
("/StatusNotifierWatcher", self.fd_watcher)
Gio.bus_own_name(
Gio.BusType.SESSION,
"org.kde.StatusNotifierWatcher",
Gio.BusNameOwnerFlags.NONE,
self.on_bus_acquired_kde,
None,
None,
)
def do_activate(self):
print("StatusNotifierWatcher running...")
self.hold()
def do_shutdown(self):
if self.bus:
if self.kde_publishing:
self.bus.unpublish(self.kde_publishing)
if self.fd_publishing:
self.bus.unpublish(self.fd_publishing)
print("Unregistered")
Gtk.Application.do_shutdown(self)