linux observer
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Activity detection using DBus APIs.
5
6Detects screen lock and display power-save state via DBus, with ordered
7fallback chains that cover GNOME and KDE desktops. Every function
8degrades gracefully — returning a safe default — so the observer keeps
9running regardless of desktop environment.
10"""
11
12import asyncio
13import logging
14import os
15
16from dbus_next import Variant
17from dbus_next.aio import MessageBus
18from dbus_next.errors import (
19 DBusError,
20 InvalidIntrospectionError,
21 InvalidMemberNameError,
22)
23
24logger = logging.getLogger(__name__)
25
26_DBUS_PROBE_TIMEOUT_SEC = 2.0
27
28_SERVICE_MISSING_ERRORS = (
29 "org.freedesktop.DBus.Error.ServiceUnknown",
30 "org.freedesktop.DBus.Error.NameHasNoOwner",
31)
32
33# GTK4/GDK4 — optional, only needed for monitor geometry detection.
34# On systems without GTK4, get_monitor_geometries() will raise RuntimeError
35# but screencast recording still works (monitors labeled as "monitor-N").
36try:
37 import gi
38
39 gi.require_version("Gdk", "4.0")
40 gi.require_version("Gtk", "4.0")
41 from gi.repository import Gdk, Gtk
42
43 _HAS_GTK = True
44except (ImportError, ValueError):
45 _HAS_GTK = False
46
47# DBus service constants — screen lock
48FDO_SCREENSAVER_BUS = "org.freedesktop.ScreenSaver"
49FDO_SCREENSAVER_PATH = "/ScreenSaver"
50FDO_SCREENSAVER_IFACE = "org.freedesktop.ScreenSaver"
51
52GNOME_SCREENSAVER_BUS = "org.gnome.ScreenSaver"
53GNOME_SCREENSAVER_PATH = "/org/gnome/ScreenSaver"
54GNOME_SCREENSAVER_IFACE = "org.gnome.ScreenSaver"
55
56# DBus service constants — power save
57DISPLAY_CONFIG_BUS = "org.gnome.Mutter.DisplayConfig"
58DISPLAY_CONFIG_PATH = "/org/gnome/Mutter/DisplayConfig"
59DISPLAY_CONFIG_IFACE = "org.gnome.Mutter.DisplayConfig"
60
61KDE_POWER_BUS = "org.kde.Solid.PowerManagement"
62KDE_POWER_PATH = "/org/kde/Solid/PowerManagement"
63KDE_POWER_IFACE = "org.kde.Solid.PowerManagement"
64
65# DBus service constants — monitor geometry (KDE)
66KSCREEN_BUS = "org.kde.KScreen"
67KSCREEN_PATH = "/backend"
68KSCREEN_IFACE = "org.kde.kscreen.Backend"
69
70
71def _is_service_missing(exc: BaseException) -> bool:
72 """True if exc is a DBusError meaning the bus name is not currently owned."""
73 return (
74 isinstance(exc, DBusError)
75 and getattr(exc, "type", "") in _SERVICE_MISSING_ERRORS
76 )
77
78
79def _is_gnome_desktop() -> bool:
80 """True if any XDG_CURRENT_DESKTOP token equals 'gnome' (case-insensitive)."""
81 return any(
82 token.strip().casefold() == "gnome"
83 for token in os.environ.get("XDG_CURRENT_DESKTOP", "").split(":")
84 )
85
86
87async def _name_has_owner(bus: MessageBus, bus_name: str) -> bool:
88 """Ask the bus daemon whether a well-known name is currently owned.
89
90 Returns False on any probe failure (daemon unreachable, timeout, parser
91 error) after logging a warning — the service is treated as absent.
92 """
93
94 async def _probe() -> bool:
95 intro = await bus.introspect("org.freedesktop.DBus", "/org/freedesktop/DBus")
96 obj = bus.get_proxy_object(
97 "org.freedesktop.DBus", "/org/freedesktop/DBus", intro
98 )
99 iface = obj.get_interface("org.freedesktop.DBus")
100 return bool(await iface.call_name_has_owner(bus_name))
101
102 try:
103 return await asyncio.wait_for(_probe(), timeout=_DBUS_PROBE_TIMEOUT_SEC)
104 except (DBusError, InvalidMemberNameError, OSError, asyncio.TimeoutError) as exc:
105 logger.warning(
106 "NameHasOwner probe failed: service=%s path=%s: %s: %s",
107 bus_name,
108 "/org/freedesktop/DBus",
109 type(exc).__name__,
110 exc,
111 )
112 return False
113
114
115async def probe_activity_services(bus: MessageBus) -> dict[str, bool]:
116 """Check which activity DBus services are reachable."""
117 services = {
118 "fdo_screensaver": FDO_SCREENSAVER_BUS,
119 "gnome_screensaver": GNOME_SCREENSAVER_BUS,
120 "gnome_display_config": DISPLAY_CONFIG_BUS,
121 "kde_power": KDE_POWER_BUS,
122 "kscreen": KSCREEN_BUS,
123 }
124 results = {}
125 for name, bus_name in services.items():
126 results[name] = await _name_has_owner(bus, bus_name)
127
128 # Log grouped by function
129 lock_backends = ["fdo_screensaver", "gnome_screensaver"]
130 power_backends = ["gnome_display_config", "kde_power"]
131 monitor_backends = ["kscreen"]
132 results["gtk4"] = _HAS_GTK
133
134 def _status(keys):
135 return ", ".join(f"{k} [{'ok' if results[k] else 'missing'}]" for k in keys)
136
137 logger.info("Screen lock backends: %s", _status(lock_backends))
138 logger.info("Power save backends: %s", _status(power_backends))
139 logger.info(
140 "Monitor backends: %s, gtk4 [%s]",
141 _status(monitor_backends),
142 "ok" if results["gtk4"] else "missing",
143 )
144
145 any_lock = any(results[k] for k in lock_backends)
146 any_power = any(results[k] for k in power_backends)
147 if not any_lock and not any_power:
148 logger.warning(
149 "No activity backends available — running in always-capture mode"
150 )
151
152 return results
153
154
155async def is_screen_locked(bus: MessageBus) -> bool:
156 """Check if the screen is locked.
157
158 On GNOME, probes only org.gnome.ScreenSaver — the FDO ScreenSaver bus
159 on GNOME serves idle-inhibit endpoints only and does not implement
160 GetActive. On non-GNOME desktops, tries FDO ScreenSaver first (KDE
161 kwin and other compliant desktops), then falls back to GNOME
162 ScreenSaver. Returns True if locked, False if unlocked or all
163 backends unavailable.
164 """
165 if not _is_gnome_desktop():
166 # Try freedesktop.org ScreenSaver first (KDE kwin and other non-GNOME desktops)
167 try:
168 intro = await bus.introspect(FDO_SCREENSAVER_BUS, FDO_SCREENSAVER_PATH)
169 obj = bus.get_proxy_object(FDO_SCREENSAVER_BUS, FDO_SCREENSAVER_PATH, intro)
170 iface = obj.get_interface(FDO_SCREENSAVER_IFACE)
171 return bool(await iface.call_get_active())
172 except (
173 DBusError,
174 InvalidMemberNameError,
175 InvalidIntrospectionError,
176 OSError,
177 ) as exc:
178 if not _is_service_missing(exc):
179 logger.warning(
180 "is_screen_locked FDO backend failed: service=%s path=%s: %s: %s",
181 FDO_SCREENSAVER_BUS,
182 FDO_SCREENSAVER_PATH,
183 type(exc).__name__,
184 exc,
185 )
186
187 # Fall back to GNOME ScreenSaver
188 try:
189 intro = await bus.introspect(GNOME_SCREENSAVER_BUS, GNOME_SCREENSAVER_PATH)
190 obj = bus.get_proxy_object(GNOME_SCREENSAVER_BUS, GNOME_SCREENSAVER_PATH, intro)
191 iface = obj.get_interface(GNOME_SCREENSAVER_IFACE)
192 return bool(await iface.call_get_active())
193 except (
194 DBusError,
195 InvalidMemberNameError,
196 InvalidIntrospectionError,
197 OSError,
198 ) as exc:
199 if not _is_service_missing(exc):
200 logger.warning(
201 "is_screen_locked GNOME backend failed: service=%s path=%s: %s: %s",
202 GNOME_SCREENSAVER_BUS,
203 GNOME_SCREENSAVER_PATH,
204 type(exc).__name__,
205 exc,
206 )
207 return False
208
209
210async def is_power_save_active(bus: MessageBus) -> bool:
211 """Check display power save via GNOME Mutter, then KDE Solid.
212
213 Returns True if power save is active, False otherwise.
214 """
215 # Try GNOME Mutter DisplayConfig first
216 try:
217 intro = await bus.introspect(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH)
218 obj = bus.get_proxy_object(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH, intro)
219 iface = obj.get_interface("org.freedesktop.DBus.Properties")
220 mode_variant = await iface.call_get(DISPLAY_CONFIG_IFACE, "PowerSaveMode")
221 mode = int(mode_variant.value)
222 return mode != 0
223 except (
224 DBusError,
225 InvalidMemberNameError,
226 InvalidIntrospectionError,
227 OSError,
228 ) as exc:
229 if not _is_service_missing(exc):
230 logger.warning(
231 "is_power_save_active Mutter backend failed: service=%s path=%s: %s: %s",
232 DISPLAY_CONFIG_BUS,
233 DISPLAY_CONFIG_PATH,
234 type(exc).__name__,
235 exc,
236 )
237
238 # Fall back to KDE Solid PowerManagement
239 try:
240 intro = await bus.introspect(KDE_POWER_BUS, KDE_POWER_PATH)
241 obj = bus.get_proxy_object(KDE_POWER_BUS, KDE_POWER_PATH, intro)
242 iface = obj.get_interface(KDE_POWER_IFACE)
243 return bool(await iface.call_is_lid_closed())
244 except (
245 DBusError,
246 InvalidMemberNameError,
247 InvalidIntrospectionError,
248 OSError,
249 ) as exc:
250 if not _is_service_missing(exc):
251 logger.warning(
252 "is_power_save_active KDE backend failed: service=%s path=%s: %s: %s",
253 KDE_POWER_BUS,
254 KDE_POWER_PATH,
255 type(exc).__name__,
256 exc,
257 )
258 return False
259
260
261def get_monitor_geometries() -> list[dict]:
262 """
263 Get structured monitor information.
264
265 Returns:
266 List of dicts with format:
267 [{"id": "connector-id", "box": [x1, y1, x2, y2], "position": "center|left|right|..."}, ...]
268 where box contains [left, top, right, bottom] coordinates
269
270 Raises:
271 RuntimeError: If GTK4/GDK4 is not available.
272 """
273 if not _HAS_GTK:
274 raise RuntimeError("GTK4 not available for monitor geometry detection")
275
276 from .monitor_positions import assign_monitor_positions
277
278 # Initialize GTK before using GDK functions
279 Gtk.init()
280
281 # Get the default display. If it is None, try opening one from the environment.
282 display = Gdk.Display.get_default()
283 if display is None:
284 env_display = os.environ.get("WAYLAND_DISPLAY") or os.environ.get("DISPLAY")
285 if env_display is not None:
286 display = Gdk.Display.open(env_display)
287 if display is None:
288 raise RuntimeError("No display available")
289 monitors = display.get_monitors()
290
291 # Collect monitor geometries
292 geometries = []
293 for monitor in monitors:
294 geom = monitor.get_geometry()
295 connector = monitor.get_connector() or f"monitor-{len(geometries)}"
296 geometries.append(
297 {
298 "id": connector,
299 "box": [geom.x, geom.y, geom.x + geom.width, geom.y + geom.height],
300 }
301 )
302
303 # Assign position labels using shared algorithm
304 return assign_monitor_positions(geometries)
305
306
307def _unwrap_variants(obj):
308 """Recursively unwrap dbus-next Variants in nested DBus structures."""
309 if isinstance(obj, Variant):
310 return _unwrap_variants(obj.value)
311 if isinstance(obj, dict):
312 return {key: _unwrap_variants(value) for key, value in obj.items()}
313 if isinstance(obj, list):
314 return [_unwrap_variants(value) for value in obj]
315 if isinstance(obj, tuple):
316 return tuple(_unwrap_variants(value) for value in obj)
317 return obj
318
319
320async def get_monitor_geometries_kscreen(bus: MessageBus) -> list[dict]:
321 """
322 Get monitor geometry information from KDE KScreen DBus.
323
324 Returns:
325 List of dicts with format:
326 [{"id": "connector-id", "box": [x1, y1, x2, y2], "position": "center|left|right|..."}, ...]
327 """
328 try:
329 from .monitor_positions import assign_monitor_positions
330
331 intro = await bus.introspect(KSCREEN_BUS, KSCREEN_PATH)
332 obj = bus.get_proxy_object(KSCREEN_BUS, KSCREEN_PATH, intro)
333 iface = obj.get_interface(KSCREEN_IFACE)
334 config = _unwrap_variants(await iface.call_get_config())
335 outputs = config.get("outputs", {})
336 output_values = outputs.values() if isinstance(outputs, dict) else outputs
337
338 geometries = []
339 for output in output_values:
340 if not isinstance(output, dict):
341 continue
342 if not output.get("enabled") or not output.get("connected"):
343 continue
344
345 name = output.get("name")
346 pos = output.get("pos", {})
347 size = output.get("size", {})
348 if not isinstance(name, str) or not isinstance(pos, dict):
349 continue
350 if not isinstance(size, dict):
351 continue
352
353 x = int(pos.get("x", 0))
354 y = int(pos.get("y", 0))
355 scale = float(output.get("scale", 1.0) or 1.0)
356 width = int(size.get("width", 0))
357 height = int(size.get("height", 0))
358 logical_width = round(width / scale)
359 logical_height = round(height / scale)
360 geometries.append(
361 {
362 "id": name,
363 "box": [x, y, x + logical_width, y + logical_height],
364 }
365 )
366
367 monitors = assign_monitor_positions(geometries)
368 logger.debug("KScreen monitor geometries found: %d", len(monitors))
369 return monitors
370 except (
371 DBusError,
372 InvalidMemberNameError,
373 InvalidIntrospectionError,
374 OSError,
375 ) as exc:
376 if not _is_service_missing(exc):
377 logger.warning(
378 "get_monitor_geometries_kscreen failed: service=%s path=%s: %s: %s",
379 KSCREEN_BUS,
380 KSCREEN_PATH,
381 type(exc).__name__,
382 exc,
383 )
384 return []