348 lines
14 KiB
Python
348 lines
14 KiB
Python
"""libei sender for an EIS fd from xdg-desktop-portal ConnectToEIS."""
|
|
import ctypes
|
|
import select
|
|
import time
|
|
|
|
# libei.h: enum ei_event_type starts at 1, then increments.
|
|
EI_EVENT_CONNECT = 1
|
|
EI_EVENT_DISCONNECT = 2
|
|
EI_EVENT_SEAT_ADDED = 3
|
|
EI_EVENT_SEAT_REMOVED = 4
|
|
EI_EVENT_DEVICE_ADDED = 5
|
|
EI_EVENT_DEVICE_REMOVED = 6
|
|
EI_EVENT_DEVICE_PAUSED = 7
|
|
EI_EVENT_DEVICE_RESUMED = 8
|
|
|
|
# libei.h: enum ei_device_capability values are bitmasks, not 1..n.
|
|
CAP_POINTER = 1 << 0
|
|
CAP_POINTER_ABSOLUTE = 1 << 1
|
|
CAP_KEYBOARD = 1 << 2
|
|
CAP_TOUCH = 1 << 3
|
|
CAP_SCROLL = 1 << 4
|
|
CAP_BUTTON = 1 << 5
|
|
BIND_CAPS = (CAP_POINTER, CAP_POINTER_ABSOLUTE, CAP_KEYBOARD, CAP_BUTTON, CAP_SCROLL)
|
|
|
|
BTN_LEFT = 0x110
|
|
BTN_RIGHT = 0x111
|
|
BTN_MIDDLE = 0x112
|
|
|
|
KEY_ESC = 1
|
|
KEY_ENTER = 28
|
|
KEY_LEFTCTRL = 29
|
|
KEY_LEFTSHIFT = 42
|
|
KEY_LEFTALT = 56
|
|
KEY_SPACE = 57
|
|
KEY_F1 = 59
|
|
KEY_BACKSPACE = 14
|
|
KEY_TAB = 15
|
|
KEY_LEFTMETA = 125
|
|
F_KEYS = {
|
|
'f1': 59, 'f2': 60, 'f3': 61, 'f4': 62, 'f5': 63, 'f6': 64, 'f7': 65, 'f8': 66,
|
|
'f9': 67, 'f10': 68, 'f11': 87, 'f12': 88,
|
|
}
|
|
COMBO_KEYS = {
|
|
'enter': KEY_ENTER, 'return': KEY_ENTER, 'esc': KEY_ESC, 'escape': KEY_ESC,
|
|
'backspace': KEY_BACKSPACE, 'tab': KEY_TAB, 'space': KEY_SPACE,
|
|
'ctrl': KEY_LEFTCTRL, 'control': KEY_LEFTCTRL, 'alt': KEY_LEFTALT,
|
|
'shift': KEY_LEFTSHIFT, 'super': KEY_LEFTMETA, 'meta': KEY_LEFTMETA,
|
|
'left': 105, 'right': 106, 'up': 103, 'down': 108,
|
|
'home': 102, 'end': 107, 'pageup': 104, 'pagedown': 109, 'delete': 111, 'insert': 110,
|
|
**F_KEYS,
|
|
}
|
|
LETTER_KEYS = {ch: code for ch, code in zip('qwertyuiopasdfghjklzxcvbnm', [
|
|
16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 30, 31, 32, 33, 34, 35, 36, 37, 38, 44, 45, 46, 47, 48, 49,
|
|
])}
|
|
DIGIT_KEYS = {str(d): code for d, code in enumerate([11, 2, 3, 4, 5, 6, 7, 8, 9, 10])}
|
|
PUNCT_KEYS = {
|
|
**{ch: (DIGIT_KEYS[d], True) for ch, d in zip('!@#$%^&*()', '1234567890')},
|
|
'-': (12, False), '_': (12, True), '=': (13, False), '+': (13, True),
|
|
'[': (26, False), '{': (26, True), ']': (27, False), '}': (27, True),
|
|
';': (39, False), ':': (39, True), "'": (40, False), '"': (40, True),
|
|
'`': (41, False), '~': (41, True), '\\': (43, False), '|': (43, True),
|
|
',': (51, False), '<': (51, True), '.': (52, False), '>': (52, True),
|
|
'/': (53, False), '?': (53, True),
|
|
}
|
|
|
|
|
|
def _ptr(value):
|
|
return int(value) if value else 0
|
|
|
|
|
|
def _lib():
|
|
lib = ctypes.CDLL('libei.so.1')
|
|
lib.ei_new_sender.restype = ctypes.c_void_p
|
|
lib.ei_new_sender.argtypes = [ctypes.c_void_p]
|
|
lib.ei_configure_name.argtypes = [ctypes.c_void_p, ctypes.c_char_p]
|
|
lib.ei_setup_backend_fd.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
|
lib.ei_setup_backend_fd.restype = ctypes.c_int
|
|
lib.ei_get_fd.argtypes = [ctypes.c_void_p]
|
|
lib.ei_get_fd.restype = ctypes.c_int
|
|
lib.ei_dispatch.argtypes = [ctypes.c_void_p]
|
|
lib.ei_get_event.argtypes = [ctypes.c_void_p]
|
|
lib.ei_get_event.restype = ctypes.c_void_p
|
|
lib.ei_event_get_type.argtypes = [ctypes.c_void_p]
|
|
lib.ei_event_get_type.restype = ctypes.c_int
|
|
lib.ei_event_get_seat.argtypes = [ctypes.c_void_p]
|
|
lib.ei_event_get_seat.restype = ctypes.c_void_p
|
|
lib.ei_event_get_device.argtypes = [ctypes.c_void_p]
|
|
lib.ei_event_get_device.restype = ctypes.c_void_p
|
|
lib.ei_event_unref.argtypes = [ctypes.c_void_p]
|
|
lib.ei_seat_bind_capabilities.argtypes = [ctypes.c_void_p]
|
|
lib.ei_seat_bind_capabilities.restype = None
|
|
lib.ei_seat_has_capability.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
|
lib.ei_seat_has_capability.restype = ctypes.c_int
|
|
lib.ei_device_has_capability.argtypes = [ctypes.c_void_p, ctypes.c_int]
|
|
lib.ei_device_has_capability.restype = ctypes.c_int
|
|
lib.ei_device_ref.argtypes = [ctypes.c_void_p]
|
|
lib.ei_device_ref.restype = ctypes.c_void_p
|
|
lib.ei_device_unref.argtypes = [ctypes.c_void_p]
|
|
lib.ei_device_start_emulating.argtypes = [ctypes.c_void_p, ctypes.c_uint]
|
|
lib.ei_device_stop_emulating.argtypes = [ctypes.c_void_p]
|
|
lib.ei_device_pointer_motion_absolute.argtypes = [ctypes.c_void_p, ctypes.c_double, ctypes.c_double]
|
|
lib.ei_device_pointer_motion.argtypes = [ctypes.c_void_p, ctypes.c_double, ctypes.c_double]
|
|
lib.ei_device_button_button.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_bool]
|
|
lib.ei_device_scroll_discrete.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int]
|
|
lib.ei_device_keyboard_key.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_bool]
|
|
lib.ei_device_frame.argtypes = [ctypes.c_void_p, ctypes.c_uint64]
|
|
lib.ei_now.argtypes = [ctypes.c_void_p]
|
|
lib.ei_now.restype = ctypes.c_uint64
|
|
lib.ei_unref.argtypes = [ctypes.c_void_p]
|
|
return lib
|
|
|
|
|
|
class LibeiSender:
|
|
def __init__(self, fd, name='jarvis'):
|
|
self.lib = _lib()
|
|
self.ei = self.lib.ei_new_sender(None)
|
|
if not self.ei:
|
|
raise RuntimeError('ei_new_sender failed')
|
|
self.lib.ei_configure_name(self.ei, name.encode())
|
|
if self.lib.ei_setup_backend_fd(self.ei, int(fd)) != 0:
|
|
raise RuntimeError('ei_setup_backend_fd failed')
|
|
self.fd = self.lib.ei_get_fd(self.ei)
|
|
self.pointer_abs = None
|
|
self.pointer_rel = None
|
|
self.keyboard = None
|
|
self._emulating = set()
|
|
self._seq = 1
|
|
|
|
@property
|
|
def pointer(self):
|
|
return self.pointer_abs or self.pointer_rel
|
|
|
|
def _has(self, device, cap):
|
|
return bool(device) and bool(self.lib.ei_device_has_capability(device, cap))
|
|
|
|
def _adopt(self, current, device):
|
|
if _ptr(current) == _ptr(device):
|
|
return current
|
|
if current:
|
|
self.lib.ei_device_unref(current)
|
|
return self.lib.ei_device_ref(device)
|
|
|
|
def _drop(self, device):
|
|
key = _ptr(device)
|
|
self._emulating.discard(key)
|
|
dropped = []
|
|
for attr in ('pointer_abs', 'pointer_rel', 'keyboard'):
|
|
current = getattr(self, attr)
|
|
if _ptr(current) == key:
|
|
dropped.append(current)
|
|
setattr(self, attr, None)
|
|
for current in dropped:
|
|
self.lib.ei_device_unref(current)
|
|
|
|
def _remember(self, device):
|
|
if self._has(device, CAP_POINTER_ABSOLUTE):
|
|
self.pointer_abs = self._adopt(self.pointer_abs, device)
|
|
elif self._has(device, CAP_POINTER):
|
|
self.pointer_rel = self._adopt(self.pointer_rel, device)
|
|
if self._has(device, CAP_KEYBOARD):
|
|
self.keyboard = self._adopt(self.keyboard, device)
|
|
|
|
def wait_ready(self, timeout=8.0):
|
|
deadline = time.time() + timeout
|
|
collected = None
|
|
while time.time() < deadline:
|
|
remaining = max(0.05, deadline - time.time())
|
|
readable, _, _ = select.select([self.fd], [], [], remaining)
|
|
if readable:
|
|
self.dispatch()
|
|
if self.pointer_abs and self.keyboard and all(_ptr(d) in self._emulating for d in (self.pointer_abs, self.keyboard)):
|
|
if collected is None:
|
|
collected = time.time()
|
|
elif time.time() - collected >= 0.35 or self.pointer_abs:
|
|
return True
|
|
return bool(self.pointer_abs and self.keyboard and all(_ptr(d) in self._emulating for d in (self.pointer_abs, self.keyboard)))
|
|
|
|
def dispatch(self):
|
|
self.lib.ei_dispatch(self.ei)
|
|
while True:
|
|
event = self.lib.ei_get_event(self.ei)
|
|
if not event:
|
|
break
|
|
kind = self.lib.ei_event_get_type(event)
|
|
if kind == EI_EVENT_SEAT_ADDED:
|
|
seat = self.lib.ei_event_get_seat(event)
|
|
caps = [cap for cap in BIND_CAPS if self.lib.ei_seat_has_capability(seat, cap)] or list(BIND_CAPS)
|
|
self.lib.ei_seat_bind_capabilities(seat, *[ctypes.c_int(cap) for cap in caps], None)
|
|
elif kind == EI_EVENT_DEVICE_ADDED:
|
|
self._remember(self.lib.ei_event_get_device(event))
|
|
elif kind == EI_EVENT_DEVICE_RESUMED:
|
|
device = self.lib.ei_event_get_device(event)
|
|
self.lib.ei_device_start_emulating(device, self._seq)
|
|
self._seq += 1
|
|
self._emulating.add(_ptr(device))
|
|
self._remember(device)
|
|
elif kind == EI_EVENT_DEVICE_PAUSED:
|
|
self._emulating.discard(_ptr(self.lib.ei_event_get_device(event)))
|
|
elif kind in (EI_EVENT_DEVICE_REMOVED, EI_EVENT_DISCONNECT):
|
|
if kind == EI_EVENT_DEVICE_REMOVED:
|
|
self._drop(self.lib.ei_event_get_device(event))
|
|
else:
|
|
for current in (self.pointer_abs, self.pointer_rel, self.keyboard):
|
|
if current:
|
|
self._drop(current)
|
|
self.lib.ei_event_unref(event)
|
|
|
|
def _frame(self, device):
|
|
self.lib.ei_device_frame(device, self.lib.ei_now(self.ei))
|
|
|
|
def _button_code(self, name):
|
|
if name in ('right', 3, '3'):
|
|
return BTN_RIGHT
|
|
if name in ('middle', 2, '2'):
|
|
return BTN_MIDDLE
|
|
return BTN_LEFT
|
|
|
|
def move(self, x, y):
|
|
device = self.pointer_abs
|
|
if not device or _ptr(device) not in self._emulating:
|
|
raise RuntimeError('no active absolute EIS pointer device')
|
|
if self._has(device, CAP_POINTER_ABSOLUTE):
|
|
self.lib.ei_device_pointer_motion_absolute(device, float(x), float(y))
|
|
else:
|
|
self.lib.ei_device_pointer_motion(device, float(x), float(y))
|
|
self._frame(device)
|
|
|
|
def click(self, x, y, button='left', count=1):
|
|
self.move(x, y)
|
|
device = self.pointer
|
|
code = self._button_code(button)
|
|
for _ in range(max(1, int(count))):
|
|
self.lib.ei_device_button_button(device, code, True)
|
|
self._frame(device)
|
|
self.lib.ei_device_button_button(device, code, False)
|
|
self._frame(device)
|
|
|
|
def scroll(self, x, y, dx=0, dy=0):
|
|
self.move(x, y)
|
|
self.lib.ei_device_scroll_discrete(self.pointer, int(dx) * 120, int(dy) * 120)
|
|
self._frame(self.pointer)
|
|
|
|
def drag(self, start, end):
|
|
self.move(start[0], start[1])
|
|
self.lib.ei_device_button_button(self.pointer, BTN_LEFT, True)
|
|
self._frame(self.pointer)
|
|
try:
|
|
for step in range(1, 13):
|
|
self.move(start[0] + (end[0] - start[0]) * step / 12, start[1] + (end[1] - start[1]) * step / 12)
|
|
time.sleep(0.012)
|
|
finally:
|
|
self.lib.ei_device_button_button(self.pointer, BTN_LEFT, False)
|
|
self._frame(self.pointer)
|
|
|
|
def _tap(self, code, mods=()):
|
|
if not self.keyboard or _ptr(self.keyboard) not in self._emulating:
|
|
raise RuntimeError('no EIS keyboard device')
|
|
for mod in mods:
|
|
self.lib.ei_device_keyboard_key(self.keyboard, mod, True)
|
|
self._frame(self.keyboard)
|
|
self.lib.ei_device_keyboard_key(self.keyboard, code, True)
|
|
self._frame(self.keyboard)
|
|
self.lib.ei_device_keyboard_key(self.keyboard, code, False)
|
|
self._frame(self.keyboard)
|
|
for mod in reversed(mods):
|
|
self.lib.ei_device_keyboard_key(self.keyboard, mod, False)
|
|
self._frame(self.keyboard)
|
|
|
|
def type_text(self, text, submit=False):
|
|
for ch in str(text):
|
|
if ch == '\n':
|
|
self._tap(KEY_ENTER)
|
|
continue
|
|
if ch == ' ':
|
|
self._tap(KEY_SPACE)
|
|
continue
|
|
if ch == '\t':
|
|
self._tap(KEY_TAB)
|
|
continue
|
|
lower = ch.lower()
|
|
code = LETTER_KEYS.get(lower) or DIGIT_KEYS.get(ch)
|
|
if code is not None:
|
|
self._tap(code, (KEY_LEFTSHIFT,) if ch.isupper() else ())
|
|
continue
|
|
punct = PUNCT_KEYS.get(ch)
|
|
if punct:
|
|
code, shifted = punct
|
|
self._tap(code, (KEY_LEFTSHIFT,) if shifted else ())
|
|
continue
|
|
# GNOME/GTK Unicode input method; never silently drop characters.
|
|
self._tap(LETTER_KEYS['u'], (KEY_LEFTCTRL, KEY_LEFTSHIFT))
|
|
time.sleep(0.08)
|
|
for digit in format(ord(ch), 'x'):
|
|
self._tap(LETTER_KEYS.get(digit) or DIGIT_KEYS[digit])
|
|
time.sleep(0.01)
|
|
self._tap(KEY_ENTER)
|
|
time.sleep(0.1)
|
|
if submit:
|
|
self._tap(KEY_ENTER)
|
|
|
|
def key_combo(self, combo):
|
|
parts = [p.strip().lower() for p in str(combo).replace('-', '+').split('+') if p.strip()]
|
|
if not parts:
|
|
raise ValueError('keyboard combination is required')
|
|
modifiers = {'ctrl', 'control', 'alt', 'shift', 'super', 'meta'}
|
|
if any(part not in modifiers for part in parts[:-1]):
|
|
raise ValueError('unsupported keyboard modifier')
|
|
mods = list(dict.fromkeys(COMBO_KEYS[part] for part in parts[:-1]))
|
|
key = parts[-1]
|
|
code = COMBO_KEYS.get(key) or LETTER_KEYS.get(key) or DIGIT_KEYS.get(key)
|
|
if code is None:
|
|
raise ValueError(f'unsupported key: {key}')
|
|
self._tap(code, tuple(mods))
|
|
|
|
def handle(self, action):
|
|
self.dispatch()
|
|
kind = action.get('type')
|
|
name = action.get('action')
|
|
if kind == 'pointer' and name in ('click', 'double_click'):
|
|
self.click(action.get('x') or 0, action.get('y') or 0, action.get('button') or 'left', 2 if name == 'double_click' else 1)
|
|
elif kind == 'pointer' and name == 'move':
|
|
self.move(action.get('x') or 0, action.get('y') or 0)
|
|
elif kind == 'pointer' and name == 'scroll':
|
|
self.scroll(action.get('x') or 0, action.get('y') or 0, action.get('dx') or 0, action.get('dy') or 0)
|
|
elif kind == 'pointer' and name == 'drag':
|
|
self.drag(action.get('from') or [0, 0], action.get('to') or [0, 0])
|
|
elif kind == 'keyboard' and name == 'type':
|
|
self.type_text(action.get('text') or '', action.get('submit'))
|
|
elif kind == 'keyboard' and name == 'key':
|
|
self.key_combo(action.get('combo') or '')
|
|
else:
|
|
raise RuntimeError(f'unsupported EIS action {kind} {name}')
|
|
|
|
def close(self):
|
|
seen = []
|
|
for device in (self.pointer_abs, self.pointer_rel, self.keyboard):
|
|
if device and _ptr(device) not in seen:
|
|
seen.append(_ptr(device))
|
|
try:
|
|
self.lib.ei_device_unref(device)
|
|
except Exception:
|
|
pass
|
|
self.pointer_abs = self.pointer_rel = self.keyboard = None
|
|
self._emulating.clear()
|
|
if self.ei:
|
|
self.lib.ei_unref(self.ei)
|
|
self.ei = None
|