328 lines
14 KiB
Python
328 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""RemoteDesktop + ScreenCast broker: EIS input and a live PipeWire frame buffer."""
|
|
import json
|
|
import os
|
|
import select
|
|
import sys
|
|
import time
|
|
import signal
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
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)
|
|
|
|
from libei_sender import LibeiSender
|
|
from pw_framebuffer import PipeWireFrameBuffer
|
|
|
|
BTN = {'left': 0x110, 'right': 0x111, 'middle': 0x112}
|
|
MODE = os.environ.get('JARVIS_CU_MODE', 'act')
|
|
# persist_mode 2 made GNOME restore the previous share and skip the screen picker.
|
|
PERSIST = os.environ.get('JARVIS_CU_PERSIST', '0') == '1'
|
|
PERSIST_MODE = 2 if PERSIST else 0
|
|
|
|
|
|
def log_error(reason):
|
|
print(json.dumps({'type': 'error', 'reason': str(reason)}), flush=True)
|
|
|
|
|
|
bus = None
|
|
|
|
|
|
def portal_proxy(iface):
|
|
return Gio.DBusProxy.new_sync(
|
|
bus, Gio.DBusProxyFlags.NONE, None,
|
|
'org.freedesktop.portal.Desktop', '/org/freedesktop/portal/desktop',
|
|
iface, None,
|
|
)
|
|
|
|
|
|
remote = screencast = None
|
|
|
|
|
|
def request(proxy, method, signature, values, timeout_ms=120000):
|
|
# Subscribe before sending the request: fast responses can precede the reply.
|
|
token = f'jarvis{GLib.get_real_time()}'
|
|
sender_name = bus.get_unique_name()[1:].replace('.', '_')
|
|
request_path = f'/org/freedesktop/portal/desktop/request/{sender_name}/{token}'
|
|
values = list(values)
|
|
values[-1] = dict(values[-1], handle_token=GLib.Variant('s', token))
|
|
loop = GLib.MainLoop()
|
|
result = {'code': None, 'results': {}}
|
|
def response(_conn, _sender, _path, _interface, _member, params):
|
|
result['code'], result['results'] = params.unpack()
|
|
loop.quit()
|
|
def expired():
|
|
result['code'] = 'timeout'
|
|
loop.quit()
|
|
return False
|
|
sub = bus.signal_subscribe(None, 'org.freedesktop.portal.Request', 'Response', request_path, None, Gio.DBusSignalFlags.NONE, response)
|
|
timer = GLib.timeout_add(timeout_ms, expired)
|
|
try:
|
|
proxy.call_sync(method, GLib.Variant(signature, tuple(values)), Gio.DBusCallFlags.NONE, timeout_ms, None)
|
|
if result['code'] is None:
|
|
loop.run()
|
|
finally:
|
|
if result['code'] != 'timeout':
|
|
GLib.source_remove(timer)
|
|
bus.signal_unsubscribe(sub)
|
|
if result['code'] != 0:
|
|
raise RuntimeError(f'{method} portal response {result["code"]}')
|
|
return result['results']
|
|
|
|
|
|
def unix_fd(proxy, method, session):
|
|
incoming = Gio.UnixFDList.new()
|
|
variant, outgoing = proxy.call_with_unix_fd_list_sync(
|
|
method,
|
|
GLib.Variant('(oa{sv})', (session, {})),
|
|
Gio.DBusCallFlags.NONE,
|
|
15000,
|
|
incoming,
|
|
None,
|
|
)
|
|
if outgoing is not None and outgoing.get_length() > 0:
|
|
unpacked = variant.unpack()
|
|
index = unpacked[0] if isinstance(unpacked, (tuple, list)) else unpacked
|
|
if isinstance(index, int) and 0 <= index < outgoing.get_length():
|
|
return outgoing.get(index)
|
|
return outgoing.get(0)
|
|
unpacked = variant.unpack()
|
|
fd = unpacked[0] if isinstance(unpacked, (tuple, list)) else unpacked
|
|
if isinstance(fd, int) and fd >= 0:
|
|
return fd
|
|
raise RuntimeError(f'{method} returned no file descriptor')
|
|
|
|
|
|
def notify(method, signature, values):
|
|
remote.call_sync(method, GLib.Variant(signature, values), Gio.DBusCallFlags.NONE, 5000, None)
|
|
|
|
|
|
def stream_node_id(streams):
|
|
if not streams:
|
|
raise RuntimeError('ScreenCast Start returned no PipeWire streams')
|
|
first = streams[0]
|
|
if isinstance(first, (list, tuple)):
|
|
return int(first[0])
|
|
if isinstance(first, dict):
|
|
return int(first.get('node_id') or first.get('id'))
|
|
return int(first)
|
|
|
|
|
|
class PortalNotify:
|
|
def __init__(self, session, node_id, origin=(0, 0)):
|
|
self.session = session
|
|
self.node_id = node_id
|
|
self.origin = origin
|
|
|
|
def handle(self, action):
|
|
kind = action.get('type')
|
|
name = action.get('action')
|
|
opts = {}
|
|
session = self.session
|
|
if kind == 'pointer' and name in ('click', 'double_click', 'move'):
|
|
x = float(action.get('x') or 0)
|
|
y = float(action.get('y') or 0)
|
|
notify('NotifyPointerMotionAbsolute', '(oa{sv}udd)', (session, opts, self.node_id, x - self.origin[0], y - self.origin[1]))
|
|
if name != 'move':
|
|
button = BTN.get(str(action.get('button') or 'left'), BTN['left'])
|
|
repeats = 2 if name == 'double_click' else 1
|
|
for _ in range(repeats):
|
|
notify('NotifyPointerButton', '(oa{sv}iu)', (session, opts, button, 1))
|
|
notify('NotifyPointerButton', '(oa{sv}iu)', (session, opts, button, 0))
|
|
elif kind == 'pointer' and name == 'scroll':
|
|
self.handle(dict(action, action='move'))
|
|
for axis, key in ((0, 'dy'), (1, 'dx')):
|
|
amount = int(action.get(key) or 0)
|
|
if amount:
|
|
notify('NotifyPointerAxisDiscrete', '(oa{sv}ui)', (session, opts, axis, amount))
|
|
elif kind == 'pointer' and name == 'drag':
|
|
start = action.get('from') or [0, 0]
|
|
end = action.get('to') or [0, 0]
|
|
self.handle({'type': 'pointer', 'action': 'move', 'x': start[0], 'y': start[1]})
|
|
notify('NotifyPointerButton', '(oa{sv}iu)', (session, opts, BTN['left'], 1))
|
|
try:
|
|
for step in range(1, 13):
|
|
self.handle({'type': 'pointer', 'action': 'move', 'x': start[0] + (end[0] - start[0]) * step / 12, 'y': start[1] + (end[1] - start[1]) * step / 12})
|
|
time.sleep(0.012)
|
|
finally:
|
|
notify('NotifyPointerButton', '(oa{sv}iu)', (session, opts, BTN['left'], 0))
|
|
elif kind == 'keyboard' and name == 'type':
|
|
for ch in str(action.get('text') or ''):
|
|
if ord(ch) > 127:
|
|
self.handle({'type': 'keyboard', 'action': 'key', 'combo': 'ctrl+shift+u'})
|
|
time.sleep(0.08)
|
|
for digit in format(ord(ch), 'x'):
|
|
self.handle({'type': 'keyboard', 'action': 'type', 'text': digit})
|
|
time.sleep(0.01)
|
|
self.handle({'type': 'keyboard', 'action': 'key', 'combo': 'enter'})
|
|
time.sleep(0.1)
|
|
continue
|
|
keysym = {'\n': 0xff0d, '\t': 0xff09}.get(ch, ord(ch) if ord(ch) <= 255 else 0x01000000 | ord(ch))
|
|
notify('NotifyKeyboardKeysym', '(oa{sv}iu)', (session, opts, keysym, 1))
|
|
notify('NotifyKeyboardKeysym', '(oa{sv}iu)', (session, opts, keysym, 0))
|
|
if action.get('submit'):
|
|
notify('NotifyKeyboardKeysym', '(oa{sv}iu)', (session, opts, 0xff0d, 1))
|
|
notify('NotifyKeyboardKeysym', '(oa{sv}iu)', (session, opts, 0xff0d, 0))
|
|
elif kind == 'keyboard' and name == 'key':
|
|
from libei_sender import COMBO_KEYS, LETTER_KEYS, DIGIT_KEYS
|
|
parts = str(action.get('combo') or '').lower().replace('-', '+').split('+')
|
|
modifiers = {'ctrl', 'control', 'alt', 'shift', 'super', 'meta'}
|
|
if not parts or any(p not in modifiers for p in parts[:-1]):
|
|
raise ValueError('unsupported keyboard combination')
|
|
codes = [COMBO_KEYS.get(p) or LETTER_KEYS.get(p) or DIGIT_KEYS.get(p) for p in parts]
|
|
if any(code is None for code in codes):
|
|
raise ValueError('unsupported keyboard combination')
|
|
held = []
|
|
try:
|
|
for code in codes:
|
|
notify('NotifyKeyboardKeycode', '(oa{sv}iu)', (session, opts, code, 1))
|
|
held.append(code)
|
|
finally:
|
|
for code in reversed(held):
|
|
notify('NotifyKeyboardKeycode', '(oa{sv}iu)', (session, opts, code, 0))
|
|
else:
|
|
raise RuntimeError(f'unsupported portal notify action {kind} {name}')
|
|
|
|
|
|
class SessionHandler:
|
|
def __init__(self, input_handler, framebuffer):
|
|
self.input_handler = input_handler
|
|
self.framebuffer = framebuffer
|
|
|
|
def dispatch(self):
|
|
if hasattr(self.input_handler, 'dispatch'):
|
|
self.input_handler.dispatch()
|
|
|
|
def handle(self, action):
|
|
if action.get('type') == 'frame':
|
|
if not self.framebuffer:
|
|
raise RuntimeError('PipeWire ScreenCast is not attached')
|
|
path = action.get('path')
|
|
if not path:
|
|
raise RuntimeError('frame path is required')
|
|
self.framebuffer.capture(path)
|
|
print(json.dumps({'type': 'frame', 'path': path, 'source': 'pipewire', 'id': action.get('id')}), flush=True)
|
|
return
|
|
if self.input_handler is None:
|
|
raise RuntimeError('desktop input is observe-only')
|
|
self.input_handler.handle(action)
|
|
|
|
|
|
def run_loop(handler, extra_fd=None):
|
|
stdin_fd = sys.stdin.fileno()
|
|
buffer = b''
|
|
while True:
|
|
context = GLib.MainContext.default()
|
|
while context.pending():
|
|
context.iteration(False)
|
|
fds = [stdin_fd] + ([extra_fd] if extra_fd is not None else [])
|
|
readable, _, _ = select.select(fds, [], [], 0.1)
|
|
if extra_fd is not None and extra_fd in readable:
|
|
handler.dispatch()
|
|
if stdin_fd not in readable:
|
|
continue
|
|
chunk = os.read(stdin_fd, 65536)
|
|
if not chunk:
|
|
break
|
|
buffer += chunk
|
|
if len(buffer) > 1024 * 1024:
|
|
raise RuntimeError('input message exceeds limit')
|
|
while b'\n' in buffer:
|
|
line, buffer = buffer.split(b'\n', 1)
|
|
action = {}
|
|
try:
|
|
action = json.loads(line)
|
|
if action.get('type') == 'close':
|
|
return
|
|
handler.handle(action)
|
|
if action.get('type') != 'frame':
|
|
print(json.dumps({'type': 'ack', 'id': action.get('id'), 'ok': True}), flush=True)
|
|
except Exception as exc:
|
|
print(json.dumps({'type': 'error', 'id': action.get('id'), 'reason': str(exc)}), flush=True)
|
|
|
|
|
|
def attach_screencast():
|
|
token = f'jarvissc{GLib.get_real_time()}'
|
|
session = request(screencast, 'CreateSession', '(a{sv})', ({'session_handle_token': GLib.Variant('s', token)},))['session_handle']
|
|
request(screencast, 'SelectSources', '(oa{sv})', (session, {
|
|
'types': GLib.Variant('u', 1),
|
|
'multiple': GLib.Variant('b', False),
|
|
'cursor_mode': GLib.Variant('u', 2),
|
|
'persist_mode': GLib.Variant('u', PERSIST_MODE),
|
|
}))
|
|
return session
|
|
|
|
|
|
def main():
|
|
global bus, remote, screencast
|
|
bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
|
|
remote = portal_proxy('org.freedesktop.portal.RemoteDesktop')
|
|
screencast = portal_proxy('org.freedesktop.portal.ScreenCast')
|
|
sender = framebuffer = None
|
|
session = None
|
|
signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
|
|
try:
|
|
if MODE == 'observe':
|
|
session = attach_screencast()
|
|
results = request(screencast, 'Start', '(osa{sv})', (session, '', {}))
|
|
else:
|
|
token = f'jarvisrd{GLib.get_real_time()}'
|
|
session = request(remote, 'CreateSession', '(a{sv})', ({'session_handle_token': GLib.Variant('s', token)},))['session_handle']
|
|
request(remote, 'SelectDevices', '(oa{sv})', (session, {'types': GLib.Variant('u', 3), 'persist_mode': GLib.Variant('u', PERSIST_MODE)}))
|
|
request(screencast, 'SelectSources', '(oa{sv})', (session, {
|
|
'types': GLib.Variant('u', 1), 'multiple': GLib.Variant('b', False), 'cursor_mode': GLib.Variant('u', 2),
|
|
}))
|
|
results = request(remote, 'Start', '(osa{sv})', (session, '', {}))
|
|
streams = results.get('streams') or []
|
|
node_id = stream_node_id(streams)
|
|
props = streams[0][1]
|
|
origin = props.get('position', (0, 0))
|
|
framebuffer = PipeWireFrameBuffer(unix_fd(screencast, 'OpenPipeWireRemote', session), node_id)
|
|
backend, extra_fd, input_handler, eis_error = 'none', None, None, None
|
|
if MODE != 'observe':
|
|
# Notify remains a tested fallback, selectable for diagnostics.
|
|
if os.environ.get('JARVIS_CU_INPUT_BACKEND') != 'notify':
|
|
try:
|
|
sender = LibeiSender(unix_fd(remote, 'ConnectToEIS', session))
|
|
if not sender.wait_ready():
|
|
raise RuntimeError('EIS did not provide active absolute pointer and keyboard devices')
|
|
input_handler, extra_fd, backend = sender, sender.fd, 'portal-ei'
|
|
except Exception as exc:
|
|
eis_error = str(exc)
|
|
if sender:
|
|
sender.close()
|
|
sender = None
|
|
if input_handler is None:
|
|
input_handler, backend = PortalNotify(session, node_id, origin), 'portal-notify'
|
|
def closed(*_):
|
|
raise SystemExit(0)
|
|
bus.signal_subscribe(None, 'org.freedesktop.portal.Session', 'Closed', session, None, Gio.DBusSignalFlags.NONE, closed)
|
|
print(json.dumps({
|
|
'type': 'ready', 'session': session, 'restore_token_present': bool(results.get('restore_token')),
|
|
'backend': backend, 'screen': True, 'eis_error': eis_error,
|
|
'streams': [{'node_id': int(n), **p} for n, p in streams],
|
|
}), flush=True)
|
|
run_loop(SessionHandler(input_handler, framebuffer), extra_fd)
|
|
except Exception as exc:
|
|
log_error(exc)
|
|
finally:
|
|
if framebuffer:
|
|
framebuffer.close()
|
|
if sender:
|
|
sender.close()
|
|
if session:
|
|
try:
|
|
bus.call_sync('org.freedesktop.portal.Desktop', session, 'org.freedesktop.portal.Session', 'Close', None, None, Gio.DBusCallFlags.NONE, 2000, None)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|