mirror of
https://github.com/DeeJanuz/frametop.git
synced 2026-10-06 13:00:10 +02:00
DeeJanuzandClaude Opus 5.5
6df6f97937
Each check starts two gdbus processes (about 5.5 ms of CPU each on the Frame), so checking once a second cost about 1% of a core for as long as the desktop ran. On SteamOS 0.3.0 native activation fails, so the watcher always runs. A registry left behind now goes within 5 seconds instead of 1. design.md also says where the watcher matters: started from the VR launcher, the desktop runs in steam.service, which doesn't stop with it, so keep-apps.sh and the unit's stop don't clean up there. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
88 lines
3.6 KiB
Python
Executable File
88 lines
3.6 KiB
Python
Executable File
#!/usr/bin/python3
|
|
"""Start AT-SPI on this nested desktop's live bus, after KWin has set its displays."""
|
|
import ast
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
DBUS = "org.freedesktop.DBus"
|
|
DBUS_PATH = "/org/freedesktop/DBus"
|
|
REGISTRY = "org.a11y.atspi.Registry"
|
|
|
|
|
|
def call(address, destination, path, method, *args):
|
|
return subprocess.run(
|
|
["gdbus", "call", "--address", address, "--dest", destination,
|
|
"--object-path", path, "--method", method, "--timeout", "5", *args],
|
|
check=True, capture_output=True, text=True, timeout=6).stdout.strip()
|
|
|
|
|
|
def stop(child):
|
|
if child.poll() is None:
|
|
child.terminate()
|
|
try:
|
|
child.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
child.kill()
|
|
child.wait()
|
|
|
|
|
|
def run_registry(address, session):
|
|
# SteamOS's registryd stays alive after a bus disconnect. Keep only our own
|
|
# child tied to both private buses; no host PID lookup or broad process kill.
|
|
registry = subprocess.Popen(["/usr/lib/at-spi2-registryd"])
|
|
try:
|
|
while registry.poll() is None:
|
|
try:
|
|
call(session, DBUS, DBUS_PATH, DBUS + ".GetId")
|
|
call(address, DBUS, DBUS_PATH, DBUS + ".GetId")
|
|
except subprocess.SubprocessError:
|
|
break # Normal session shutdown or loss of its accessibility bus.
|
|
# Each check starts two gdbus processes, and SteamVR needs the CPU.
|
|
time.sleep(5)
|
|
finally:
|
|
stop(registry)
|
|
|
|
|
|
def start():
|
|
session = os.environ.get("DBUS_SESSION_BUS_ADDRESS")
|
|
runtime = os.environ.get("XDG_RUNTIME_DIR", "")
|
|
if not session or os.path.basename(runtime) != "frametop":
|
|
return # Do not autolaunch a bus or touch the host desktop.
|
|
os.environ.pop("AT_SPI_BUS_ADDRESS", None)
|
|
reply = call(session, "org.a11y.Bus", "/org/a11y/bus", "org.a11y.Bus.GetAddress")
|
|
address, = ast.literal_eval(reply)
|
|
if not isinstance(address, str) or not address.startswith("unix:"):
|
|
raise ValueError("org.a11y.Bus returned no local accessibility bus")
|
|
if call(address, DBUS, DBUS_PATH, DBUS + ".NameHasOwner", REGISTRY) == "(true,)":
|
|
return
|
|
# Activation otherwise inherits the accessibility daemon's old environment (and
|
|
# DBUS_SESSION_BUS_ADDRESS is the accessibility bus itself). Pin the live bus.
|
|
env = {key: os.environ.get(key, "") for key in (
|
|
"DISPLAY", "WAYLAND_DISPLAY", "XAUTHORITY", "XDG_RUNTIME_DIR",
|
|
"XDG_CONFIG_HOME", "DBUS_SESSION_BUS_ADDRESS")}
|
|
env["AT_SPI_BUS_ADDRESS"] = address
|
|
call(address, DBUS, DBUS_PATH, DBUS + ".UpdateActivationEnvironment", repr(env))
|
|
try:
|
|
call(address, DBUS, DBUS_PATH, DBUS + ".StartServiceByName", REGISTRY, "0")
|
|
except subprocess.CalledProcessError:
|
|
# SteamOS's launcher can select dbus-broker because we are in a user unit,
|
|
# but the private session bus has no systemd activation manager. Reuse its
|
|
# bus rather than starting another launcher/bus. Never replace an owner.
|
|
if call(address, DBUS, DBUS_PATH, DBUS + ".NameHasOwner", REGISTRY) == "(true,)":
|
|
return
|
|
os.environ["AT_SPI_BUS_ADDRESS"] = address
|
|
# registryd itself refuses to queue behind an existing registry, including
|
|
# races with native activation. The watcher cleans up only our own child.
|
|
run_registry(address, session)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
signal.signal(signal.SIGTERM, lambda signum, frame: sys.exit(0))
|
|
try:
|
|
start()
|
|
except (OSError, ValueError, SyntaxError, subprocess.SubprocessError) as error:
|
|
print(f"frametop: accessibility unavailable: {error}", file=sys.stderr)
|