Python script, ASCII text executable
1
"""
2
Implements the org.freedesktop.StatusNotifierWatcher and nothing else.
3
Copyright 2025, roundabout-host.com <vlad@roundabout-host.com>
4
This program is free software: you can redistribute it and/or modify
6
it under the terms of the GNU General Public Licence as published by
7
the Free Software Foundation, either version 3 of the Licence, or
8
(at your option) any later version.
9
This program is distributed in the hope that it will be useful,
11
but WITHOUT ANY WARRANTY; without even the implied warranty of
12
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13
GNU General Public Licence for more details.
14
You should have received a copy of the GNU General Public Licence
16
along with this program. If not, see <https://www.gnu.org/licenses/>.
17
"""
18
import gi
20
gi.require_version("Gtk", "4.0")
21
from gi.repository import Gtk, Gio, GLib
22
class StatusNotifierWatcher:
24
def __init__(self, connection: Gio.DBusConnection, interface_name: str):
25
self.connection = connection
26
self.interface_name = interface_name
27
self.items = {}
28
self.hosts = set()
29
self.node_info = Gio.DBusNodeInfo.new_for_xml(f"""
31
<node>
32
<interface name="{interface_name}">
33
<method name="RegisterStatusNotifierItem">
34
<arg type="s" name="service" direction="in"/>
35
</method>
36
<method name="RegisterStatusNotifierHost">
37
<arg type="s" name="service" direction="in"/>
38
</method>
39
<property name="RegisteredStatusNotifierItems" type="as" access="read"/>
40
<property name="IsStatusNotifierHostRegistered" type="b" access="read"/>
41
<property name="ProtocolVersion" type="i" access="read"/>
42
<signal name="StatusNotifierItemRegistered">
43
<arg type="s" name="service" direction="out"/>
44
</signal>
45
<signal name="StatusNotifierItemUnregistered">
46
<arg type="s" name="service" direction="out"/>
47
</signal>
48
<signal name="StatusNotifierHostRegistered">
49
<arg type="s" name="service" direction="out"/>
50
</signal>
51
</interface>
52
</node>
53
"""
54
)
55
self.interface_info = self.node_info.interfaces[0]
56
self.connection.register_object(
58
"/StatusNotifierWatcher",
59
self.interface_info,
60
self.handle_method_call,
61
self.get_property,
62
None
63
)
64
self.connection.signal_subscribe(
66
"org.freedesktop.DBus",
67
"org.freedesktop.DBus",
68
"NameOwnerChanged",
69
"/org/freedesktop/DBus",
70
None,
71
Gio.DBusSignalFlags.NONE,
72
self.on_name_owner_changed,
73
)
74
def get_property(self, connection, sender, object_path, interface_name, property_name):
76
if property_name == "IsStatusNotifierHostRegistered":
77
return GLib.Variant("b", bool(self.hosts))
78
elif property_name == "RegisteredStatusNotifierItems":
79
return GLib.Variant("as", list(self.items))
80
elif property_name == "ProtocolVersion":
81
return GLib.Variant("i", 0)
82
else:
83
raise Exception(f"Unknown property {property_name}")
84
def on_name_owner_changed(self, connection, sender_name, object_path, interface_name, signal_name, params):
86
name, old_owner, new_owner = params.unpack()
87
if new_owner == "":
88
for lookup in (name, old_owner):
89
if lookup in self.items:
90
path = self.items[lookup]
91
del self.items[lookup]
92
print(f"Item {lookup} vanished")
93
self.emit_signal("StatusNotifierItemUnregistered", lookup + ":" + path)
94
if lookup in self.hosts:
95
self.hosts.discard(lookup)
96
print(f"Host {lookup} vanished")
97
def handle_method_call(self, conn, sender, obj_path, iface_name, method, params, invocation):
99
if method == "RegisterStatusNotifierItem":
100
(arg_path,) = params.unpack()
101
self._register_item(sender, arg_path)
102
invocation.return_value(None)
103
elif method == "RegisterStatusNotifierHost":
104
(service,) = params.unpack()
105
self._register_host(sender, service)
106
invocation.return_value(None)
107
else:
108
invocation.return_dbus_error("org.freedesktop.DBus.Error.UnknownMethod", "Unknown method")
109
def _register_item(self, sender, path):
111
if sender not in self.items:
112
if not path.startswith("/"):
113
path = "/StatusNotifierItem"
114
self.items[sender] = path
115
print(f"Registered item {sender}")
116
self.emit_signal("StatusNotifierItemRegistered", sender + ":" + path)
117
def _register_host(self, sender, service):
119
if sender not in self.hosts:
120
self.hosts.add(sender)
121
print(f"Registered host {sender}")
122
self.emit_signal("StatusNotifierHostRegistered", sender)
123
def emit_signal(self, name, arg=None):
125
args = GLib.Variant("(s)", (arg,)) if arg else None
126
self.connection.emit_signal(
127
None,
128
"/StatusNotifierWatcher",
129
self.interface_name,
130
name,
131
args
132
)
133
class WatcherApplication(Gtk.Application):
136
def __init__(self):
137
super().__init__(
138
application_id="com.roundabout_host.roundabout.PanoramaIndicatorService",
139
flags=Gio.ApplicationFlags.FLAGS_NONE,
140
)
141
self.fd_watcher = None
142
self.kde_watcher = None
143
def do_startup(self):
145
Gtk.Application.do_startup(self)
146
def do_activate(self):
148
print("StatusNotifierWatcher starting...")
149
Gio.bus_own_name(
151
Gio.BusType.SESSION,
152
"org.freedesktop.StatusNotifierWatcher",
153
Gio.BusNameOwnerFlags.NONE,
154
self.on_bus_acquired_fd,
155
None,
156
None,
157
)
158
Gio.bus_own_name(
159
Gio.BusType.SESSION,
160
"org.kde.StatusNotifierWatcher",
161
Gio.BusNameOwnerFlags.NONE,
162
self.on_bus_acquired_kde,
163
None,
164
None,
165
)
166
self.hold()
168
def on_bus_acquired_fd(self, connection, name):
170
self.fd_watcher = StatusNotifierWatcher(connection, "org.freedesktop.StatusNotifierWatcher")
171
def on_bus_acquired_kde(self, connection, name):
173
self.kde_watcher = StatusNotifierWatcher(connection, "org.kde.StatusNotifierWatcher")
174
def do_shutdown(self):
176
Gtk.Application.do_shutdown(self)
177
if __name__ == "__main__":
180
app = WatcherApplication()
181
app.run(None)
182