P7

computer-use/py/atspi_action.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+import json
+import sys
+from gi.repository import Atspi
+
+target, action = json.loads(sys.argv[1]), sys.argv[2]
+Atspi.init()
+desktop = Atspi.get_desktop(0)
+found = None
+def walk(node, depth=0):
+    global found
+    if found or node is None or depth > 30:
+        return
+    try:
+        if (node.get_name() or '') == target.get('name') and (node.get_role_name() or '') == target.get('role'):
+            found = node
+            return
+        for i in range(node.get_child_count()):
+            walk(node.get_child_at_index(i), depth + 1)
+    except Exception:
+        return
+walk(desktop)
+if not found:
+    raise SystemExit('AT-SPI target not found')
+actions = found.get_action()
+for i in range(actions.get_n_actions()):
+    if actions.get_action_name(i).lower() == action.lower():
+        if actions.do_action(i):
+            print(json.dumps({'ok': True, 'action': action}))
+            raise SystemExit(0)
+raise SystemExit(f'AT-SPI action unavailable: {action}')

computer-use/py/portal_remote_desktop.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+"""RemoteDesktop consent broker for the local EIS input worker."""
+import json
+import sys
+try:
+    import gi
+    gi.require_version('Gio', '2.0')
+    from gi.repository import Gio, GLib
+except Exception as exc:
+    print(json.dumps({'type': 'error', 'reason': f'PyGObject unavailable: {exc}'}), flush=True)
+    raise SystemExit(1)
+
+bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
+proxy = Gio.DBusProxy.new_sync(bus, Gio.DBusProxyFlags.NONE, None, 'org.freedesktop.portal.Desktop', '/org/freedesktop/portal/desktop', 'org.freedesktop.portal.RemoteDesktop', None)
+def request(method, signature, values):
+    request_path = proxy.call_sync(method, GLib.Variant(signature, values), Gio.DBusCallFlags.NONE, -1, None).unpack()[0]
+    loop = GLib.MainLoop(); result = {'code': 1, 'results': {}}
+    def response(_conn, _sender, _path, _interface, _member, params):
+        result['code'], result['results'] = params.unpack(); loop.quit()
+    sub = bus.signal_subscribe(None, 'org.freedesktop.portal.Request', 'Response', request_path, None, Gio.DBusSignalFlags.NONE, response)
+    loop.run(); bus.signal_unsubscribe(sub)
+    if result['code'] != 0: raise RuntimeError(f'{method} portal response {result["code"]}')
+    return result['results']
+try:
+    token = f'jarvis{GLib.get_real_time()}'
+    session = request('CreateSession', '(a{sv})', ({'session_handle_token': GLib.Variant('s', token)},))['session_handle']
+    request('SelectDevices', '(oa{sv})', (session, {'types': GLib.Variant('u', 3), 'persist_mode': GLib.Variant('u', 2)}))
+    request('Start', '(osa{sv})', (session, '', {}))
+    print(json.dumps({'type': 'ready', 'session': session, 'restore_token_present': True}), flush=True)
+except Exception as exc:
+    print(json.dumps({'type': 'error', 'reason': str(exc)}), flush=True)
+for line in sys.stdin:
+    try:
+        action = json.loads(line)
+        if action.get('type') == 'close':
+            break
+    except Exception:
+        continue