solpbc.org
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 logging
13import os
14
15from dbus_next import Variant
16from dbus_next.aio import MessageBus
17
18logger = logging.getLogger(__name__)
19
20# GTK4/GDK4 — optional, only needed for monitor geometry detection.
21# On systems without GTK4, get_monitor_geometries() will raise RuntimeError
22# but screencast recording still works (monitors labeled as "monitor-N").
23try:
24 import gi
25
26 gi.require_version("Gdk", "4.0")
27 gi.require_version("Gtk", "4.0")
28 from gi.repository import Gdk, Gtk
29
30 _HAS_GTK = True
31except (ImportError, ValueError):
32 _HAS_GTK = False
33
34# DBus service constants — screen lock
35FDO_SCREENSAVER_BUS = "org.freedesktop.ScreenSaver"
36FDO_SCREENSAVER_PATH = "/ScreenSaver"
37FDO_SCREENSAVER_IFACE = "org.freedesktop.ScreenSaver"
38
39GNOME_SCREENSAVER_BUS = "org.gnome.ScreenSaver"
40GNOME_SCREENSAVER_PATH = "/org/gnome/ScreenSaver"
41GNOME_SCREENSAVER_IFACE = "org.gnome.ScreenSaver"
42
43# DBus service constants — power save
44DISPLAY_CONFIG_BUS = "org.gnome.Mutter.DisplayConfig"
45DISPLAY_CONFIG_PATH = "/org/gnome/Mutter/DisplayConfig"
46DISPLAY_CONFIG_IFACE = "org.gnome.Mutter.DisplayConfig"
47
48KDE_POWER_BUS = "org.kde.Solid.PowerManagement"
49KDE_POWER_PATH = "/org/kde/Solid/PowerManagement"
50KDE_POWER_IFACE = "org.kde.Solid.PowerManagement"
51
52# DBus service constants — monitor geometry (KDE)
53KSCREEN_BUS = "org.kde.KScreen"
54KSCREEN_PATH = "/backend"
55KSCREEN_IFACE = "org.kde.kscreen.Backend"
56
57
58async def probe_activity_services(bus: MessageBus) -> dict[str, bool]:
59 """Check which activity DBus services are reachable."""
60 services = {
61 "fdo_screensaver": FDO_SCREENSAVER_BUS,
62 "gnome_screensaver": GNOME_SCREENSAVER_BUS,
63 "gnome_display_config": DISPLAY_CONFIG_BUS,
64 "kde_power": KDE_POWER_BUS,
65 "kscreen": KSCREEN_BUS,
66 }
67 results = {}
68 for name, bus_name in services.items():
69 try:
70 await bus.introspect(bus_name, "/")
71 results[name] = True
72 except Exception:
73 results[name] = False
74
75 # Log grouped by function
76 lock_backends = ["fdo_screensaver", "gnome_screensaver"]
77 power_backends = ["gnome_display_config", "kde_power"]
78 monitor_backends = ["kscreen"]
79 results["gtk4"] = _HAS_GTK
80
81 def _status(keys):
82 return ", ".join(f"{k} [{'ok' if results[k] else 'missing'}]" for k in keys)
83
84 logger.info("Screen lock backends: %s", _status(lock_backends))
85 logger.info("Power save backends: %s", _status(power_backends))
86 logger.info(
87 "Monitor backends: %s, gtk4 [%s]",
88 _status(monitor_backends),
89 "ok" if results["gtk4"] else "missing",
90 )
91
92 any_lock = any(results[k] for k in lock_backends)
93 any_power = any(results[k] for k in power_backends)
94 if not any_lock and not any_power:
95 logger.warning(
96 "No activity backends available — running in always-capture mode"
97 )
98
99 return results
100
101
102async def is_screen_locked(bus: MessageBus) -> bool:
103 """Check if the screen is locked via FDO ScreenSaver, then GNOME ScreenSaver.
104
105 Returns True if locked, False if unlocked or all backends unavailable.
106 """
107 # Try freedesktop.org ScreenSaver first (KDE kwin + GNOME)
108 try:
109 intro = await bus.introspect(FDO_SCREENSAVER_BUS, FDO_SCREENSAVER_PATH)
110 obj = bus.get_proxy_object(FDO_SCREENSAVER_BUS, FDO_SCREENSAVER_PATH, intro)
111 iface = obj.get_interface(FDO_SCREENSAVER_IFACE)
112 return bool(await iface.call_get_active())
113 except Exception:
114 pass
115
116 # Fall back to GNOME ScreenSaver
117 try:
118 intro = await bus.introspect(GNOME_SCREENSAVER_BUS, GNOME_SCREENSAVER_PATH)
119 obj = bus.get_proxy_object(GNOME_SCREENSAVER_BUS, GNOME_SCREENSAVER_PATH, intro)
120 iface = obj.get_interface(GNOME_SCREENSAVER_IFACE)
121 return bool(await iface.call_get_active())
122 except Exception:
123 return False
124
125
126async def is_power_save_active(bus: MessageBus) -> bool:
127 """Check display power save via GNOME Mutter, then KDE Solid.
128
129 Returns True if power save is active, False otherwise.
130 """
131 # Try GNOME Mutter DisplayConfig first
132 try:
133 intro = await bus.introspect(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH)
134 obj = bus.get_proxy_object(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH, intro)
135 iface = obj.get_interface("org.freedesktop.DBus.Properties")
136 mode_variant = await iface.call_get(DISPLAY_CONFIG_IFACE, "PowerSaveMode")
137 mode = int(mode_variant.value)
138 return mode != 0
139 except Exception:
140 pass
141
142 # Fall back to KDE Solid PowerManagement
143 try:
144 intro = await bus.introspect(KDE_POWER_BUS, KDE_POWER_PATH)
145 obj = bus.get_proxy_object(KDE_POWER_BUS, KDE_POWER_PATH, intro)
146 iface = obj.get_interface(KDE_POWER_IFACE)
147 return bool(await iface.call_is_lid_closed())
148 except Exception:
149 return False
150
151
152def get_monitor_geometries() -> list[dict]:
153 """
154 Get structured monitor information.
155
156 Returns:
157 List of dicts with format:
158 [{"id": "connector-id", "box": [x1, y1, x2, y2], "position": "center|left|right|..."}, ...]
159 where box contains [left, top, right, bottom] coordinates
160
161 Raises:
162 RuntimeError: If GTK4/GDK4 is not available.
163 """
164 if not _HAS_GTK:
165 raise RuntimeError("GTK4 not available for monitor geometry detection")
166
167 from .monitor_positions import assign_monitor_positions
168
169 # Initialize GTK before using GDK functions
170 Gtk.init()
171
172 # Get the default display. If it is None, try opening one from the environment.
173 display = Gdk.Display.get_default()
174 if display is None:
175 env_display = os.environ.get("WAYLAND_DISPLAY") or os.environ.get("DISPLAY")
176 if env_display is not None:
177 display = Gdk.Display.open(env_display)
178 if display is None:
179 raise RuntimeError("No display available")
180 monitors = display.get_monitors()
181
182 # Collect monitor geometries
183 geometries = []
184 for monitor in monitors:
185 geom = monitor.get_geometry()
186 connector = monitor.get_connector() or f"monitor-{len(geometries)}"
187 geometries.append(
188 {
189 "id": connector,
190 "box": [geom.x, geom.y, geom.x + geom.width, geom.y + geom.height],
191 }
192 )
193
194 # Assign position labels using shared algorithm
195 return assign_monitor_positions(geometries)
196
197
198def _unwrap_variants(obj):
199 """Recursively unwrap dbus-next Variants in nested DBus structures."""
200 if isinstance(obj, Variant):
201 return _unwrap_variants(obj.value)
202 if isinstance(obj, dict):
203 return {key: _unwrap_variants(value) for key, value in obj.items()}
204 if isinstance(obj, list):
205 return [_unwrap_variants(value) for value in obj]
206 if isinstance(obj, tuple):
207 return tuple(_unwrap_variants(value) for value in obj)
208 return obj
209
210
211async def get_monitor_geometries_kscreen(bus: MessageBus) -> list[dict]:
212 """
213 Get monitor geometry information from KDE KScreen DBus.
214
215 Returns:
216 List of dicts with format:
217 [{"id": "connector-id", "box": [x1, y1, x2, y2], "position": "center|left|right|..."}, ...]
218 """
219 try:
220 from .monitor_positions import assign_monitor_positions
221
222 intro = await bus.introspect(KSCREEN_BUS, KSCREEN_PATH)
223 obj = bus.get_proxy_object(KSCREEN_BUS, KSCREEN_PATH, intro)
224 iface = obj.get_interface(KSCREEN_IFACE)
225 config = _unwrap_variants(await iface.call_get_config())
226 outputs = config.get("outputs", {})
227 output_values = outputs.values() if isinstance(outputs, dict) else outputs
228
229 geometries = []
230 for output in output_values:
231 if not isinstance(output, dict):
232 continue
233 if not output.get("enabled") or not output.get("connected"):
234 continue
235
236 name = output.get("name")
237 pos = output.get("pos", {})
238 size = output.get("size", {})
239 if not isinstance(name, str) or not isinstance(pos, dict):
240 continue
241 if not isinstance(size, dict):
242 continue
243
244 x = int(pos.get("x", 0))
245 y = int(pos.get("y", 0))
246 scale = float(output.get("scale", 1.0) or 1.0)
247 width = int(size.get("width", 0))
248 height = int(size.get("height", 0))
249 logical_width = round(width / scale)
250 logical_height = round(height / scale)
251 geometries.append(
252 {
253 "id": name,
254 "box": [x, y, x + logical_width, y + logical_height],
255 }
256 )
257
258 monitors = assign_monitor_positions(geometries)
259 logger.debug("KScreen monitor geometries found: %d", len(monitors))
260 return monitors
261 except Exception:
262 return []