linux observer
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Install prerequisite checks for solstone-linux.
5
6Exit code rule: fail anywhere -> 1; otherwise 0. Warn does not flip exit code.
7"""
8
9from __future__ import annotations
10
11import asyncio
12import os
13import shutil
14import subprocess
15import sys
16from typing import Callable, NamedTuple
17
18CheckResult = NamedTuple(
19 "CheckResult",
20 [("name", str), ("severity", str), ("detail", str)],
21)
22
23_PORTAL_CHECK_TIMEOUT_SEC: float = 2.0
24
25
26def check_python_version() -> CheckResult:
27 version = tuple(sys.version_info[:2])
28 if version >= (3, 10):
29 return CheckResult("python version", "ok", f"{version[0]}.{version[1]}")
30 return CheckResult(
31 "python version",
32 "fail",
33 f"need >=3.10, got {version[0]}.{version[1]}",
34 )
35
36
37def check_gtk4_typelib() -> CheckResult:
38 try:
39 import gi
40
41 gi.require_version("Gtk", "4.0")
42 from gi.repository import Gtk # noqa: F401
43 except (ImportError, ValueError):
44 return CheckResult(
45 "gtk4 typelib",
46 "fail",
47 "install gir1.2-gtk-4.0 (or distro equivalent)",
48 )
49 return CheckResult("gtk4 typelib", "ok", "Gtk 4.0 available")
50
51
52def check_gstreamer() -> CheckResult:
53 if shutil.which("gst-launch-1.0") is None:
54 return CheckResult(
55 "gstreamer",
56 "fail",
57 "gst-launch-1.0 not on PATH; install gstreamer1.0-tools or equivalent",
58 )
59 try:
60 import gi
61
62 gi.require_version("Gst", "1.0")
63 from gi.repository import Gst # noqa: F401
64 except (ImportError, ValueError):
65 return CheckResult("gstreamer", "fail", "gir1.2-gstreamer-1.0 missing")
66 return CheckResult("gstreamer", "ok", "gst-launch-1.0 and Gst typelib available")
67
68
69def check_cairo() -> CheckResult:
70 try:
71 import cairo # noqa: F401
72 except ImportError:
73 return CheckResult(
74 "cairo binding",
75 "fail",
76 "install python3-cairo (or distro equivalent)",
77 )
78 return CheckResult("cairo binding", "ok", "cairo import ok")
79
80
81def check_pipewire() -> CheckResult:
82 try:
83 result = subprocess.run(
84 ["pactl", "info"],
85 capture_output=True,
86 timeout=5,
87 text=True,
88 )
89 except FileNotFoundError:
90 return CheckResult(
91 "pipewire (pactl)",
92 "fail",
93 "pactl missing; install pipewire-pulse or pulseaudio-utils",
94 )
95 if result.returncode != 0:
96 detail = result.stderr.strip().splitlines()[0] if result.stderr.strip() else ""
97 return CheckResult("pipewire (pactl)", "fail", detail)
98 detail = result.stdout.strip().splitlines()[0] if result.stdout.strip() else ""
99 return CheckResult("pipewire (pactl)", "ok", detail)
100
101
102async def check_portal() -> CheckResult:
103 from dbus_next.aio import MessageBus
104 from dbus_next.constants import BusType
105 from dbus_next.errors import AuthError, DBusError, InvalidAddressError
106
107 async def _body() -> CheckResult:
108 bus = None
109 try:
110 try:
111 bus = await MessageBus(bus_type=BusType.SESSION).connect()
112 except (OSError, AuthError, InvalidAddressError, DBusError) as e:
113 return CheckResult(
114 "xdg-desktop-portal",
115 "fail",
116 f"session bus unreachable: {e}",
117 )
118 try:
119 intro = await bus.introspect(
120 "org.freedesktop.DBus", "/org/freedesktop/DBus"
121 )
122 obj = bus.get_proxy_object(
123 "org.freedesktop.DBus", "/org/freedesktop/DBus", intro
124 )
125 iface = obj.get_interface("org.freedesktop.DBus")
126 owned = await iface.call_name_has_owner(
127 "org.freedesktop.portal.Desktop"
128 )
129 except (DBusError, OSError) as e:
130 return CheckResult(
131 "xdg-desktop-portal",
132 "fail",
133 f"session bus unreachable: {e}",
134 )
135 if owned:
136 return CheckResult(
137 "xdg-desktop-portal",
138 "ok",
139 "org.freedesktop.portal.Desktop registered on session bus",
140 )
141 return CheckResult(
142 "xdg-desktop-portal",
143 "fail",
144 "org.freedesktop.portal.Desktop not registered on session bus",
145 )
146 finally:
147 if bus is not None:
148 bus.disconnect()
149
150 try:
151 return await asyncio.wait_for(_body(), timeout=_PORTAL_CHECK_TIMEOUT_SEC)
152 except asyncio.TimeoutError:
153 return CheckResult(
154 "xdg-desktop-portal",
155 "fail",
156 f"timed out after {_PORTAL_CHECK_TIMEOUT_SEC:g}s",
157 )
158
159
160def check_user_systemd() -> CheckResult:
161 try:
162 result = subprocess.run(
163 ["systemctl", "--user", "is-system-running"],
164 capture_output=True,
165 timeout=5,
166 text=True,
167 )
168 except FileNotFoundError:
169 return CheckResult(
170 "systemd --user",
171 "fail",
172 "systemctl --user not reachable",
173 )
174 detail = result.stdout.strip().splitlines()[0] if result.stdout.strip() else ""
175 if detail:
176 return CheckResult("systemd --user", "ok", detail)
177 return CheckResult("systemd --user", "fail", "systemctl --user not reachable")
178
179
180def check_pipx() -> CheckResult:
181 if shutil.which("pipx") is None:
182 return CheckResult(
183 "pipx",
184 "fail",
185 "pipx missing; install via 'python3 -m pip install --user pipx' or distro package",
186 )
187 return CheckResult("pipx", "ok", "pipx on PATH")
188
189
190def check_appindicator_ext() -> CheckResult:
191 desktop = os.environ.get("XDG_CURRENT_DESKTOP", "")
192 if "GNOME" not in desktop:
193 return CheckResult(
194 "appindicator ext (soft)",
195 "ok",
196 "not applicable (non-GNOME desktop)",
197 )
198 try:
199 result = subprocess.run(
200 ["gnome-extensions", "list"],
201 capture_output=True,
202 timeout=5,
203 text=True,
204 )
205 except FileNotFoundError:
206 return CheckResult(
207 "appindicator ext (soft)",
208 "warn",
209 "install gnome-shell-extension-appindicator",
210 )
211 if "appindicator" in result.stdout.lower():
212 return CheckResult(
213 "appindicator ext (soft)", "ok", "appindicator extension present"
214 )
215 return CheckResult(
216 "appindicator ext (soft)",
217 "warn",
218 "install gnome-shell-extension-appindicator",
219 )
220
221
222def run_doctor() -> int:
223 checks: list[tuple[str, Callable[[], CheckResult]]] = [
224 ("python version", check_python_version),
225 ("gtk4 typelib", check_gtk4_typelib),
226 ("gstreamer", check_gstreamer),
227 ("cairo binding", check_cairo),
228 ("pipewire (pactl)", check_pipewire),
229 ("xdg-desktop-portal", lambda: asyncio.run(check_portal())),
230 ("systemd --user", check_user_systemd),
231 ("pipx", check_pipx),
232 ("appindicator ext (soft)", check_appindicator_ext),
233 ]
234 fail_count = 0
235 warn_count = 0
236
237 for name, fn in checks:
238 try:
239 result = fn()
240 except Exception as e:
241 result = CheckResult(name, "fail", repr(e))
242 if not result.name or result.name != name:
243 result = CheckResult(name, result.severity, result.detail)
244 print(f"{result.severity:<4} {result.name:<28} {result.detail}")
245 if result.severity == "fail":
246 fail_count += 1
247 elif result.severity == "warn":
248 warn_count += 1
249
250 print()
251 print(f"doctor: {len(checks)} checks, {fail_count} failed, {warn_count} warnings")
252 return 1 if fail_count else 0