gig-line/Tools/input/input_emulation.py

#!/usr/bin/env python3
"""Pure-Python uinput driver to inject OS-level keyboard and mouse input.

This is a small, dependency-free input driver used by the demo QA harness
(`Tools/input/input_test.py`) to prove whether OS input actually reaches the game.
It talks to ``/dev/uinput`` directly through ``ctypes``, so it creates a real
virtual input device that the kernel and the desktop session (X11 or Wayland)
treat as genuine hardware. The game cannot tell simulated input from a real
keyboard/mouse.

No third-party packages are required. ``/dev/uinput`` must be writable by the
calling user (on this machine the system ACL grants `zwitschi` rw; if it is
not, add a udev rule or run as a user in the `input` group).

Usage
-----
    from input_emulation import InputDevice

    with InputDevice() as dev:
        dev.key('w')            # press and release W -> types 'w'? no: key code 17
        dev.hold('w', 0.5)      # hold W for half a second (run forward)
        dev.move(120, 0)        # nudge the mouse +120 px on X (look right)

All calls log using Python's `logging` module so a scripted test can show the
exact injection timeline next to the game log.

Design notes
------------
* Uses the newer ``UI_DEV_SETUP`` ioctl with a ``struct uinput_setup``, falling
  back to the legacy ``write(struct uinput_user_dev)`` method if the kernel
  rejects it (very old kernels). On this kernel the setup ioctl path is used.
* The device exposes EV_KEY (keyboard), EV_REL (relative mouse) and
  EV_SYN (synchronisation). It does not declare an absolute axis, so it is
  treated as a relative device and the desktop does not try to give it an
  on-screen pointer role.
* A single device with both keys and relative axes is created; for a game this
  is fine. If a separate gamepad (EV_ABS/EV_FF) is wanted later, extend this
  file with a second device class that also sets ABS bits.
"""

import ctypes
import ctypes.util
import fcntl
import logging
import os
import struct
import sys
import time

log = logging.getLogger("uinput")

# ---------------------------------------------------------------------------
# ioctl machinery
# ---------------------------------------------------------------------------

# _IOC(dir, type, nr, size) is the standard Linux ioctl encoding.
_IOC_NRBITS = 8
_IOC_TYPEBITS = 8
_IOC_SIZEBITS = 14
_IOC_DIRBITS = 2
_IOC_NRSHIFT = 0
_IOC_TYPESHIFT = _IOC_NRSHIFT + _IOC_NRBITS
_IOC_SIZESHIFT = _IOC_TYPESHIFT + _IOC_TYPEBITS
_IOC_DIRSHIFT = _IOC_SIZESHIFT + _IOC_SIZEBITS
_IOC_NONE = 0
_IOC_WRITE = 1
_IOC_READ = 2


def _ioc(direction, size, type_, nr):
    # type_ may be a single character (ioctl base letter) or already an int.
    if isinstance(type_, str):
        type_ = ord(type_) if len(type_) == 1 else 0
    return (
        (size << _IOC_SIZESHIFT)
        | (type_ << _IOC_TYPESHIFT)
        | (nr << _IOC_NRSHIFT)
        | (direction << _IOC_DIRSHIFT)
    )


def _IOW(type_, nr, size):
    return _ioc(_IOC_WRITE, size, type_, nr)


def _IOR(type_, nr, size):
    return _ioc(_IOC_READ, size, type_, nr)


def _IO(type_, nr):
    return _ioc(_IOC_NONE, 0, type_, nr)


# ---------------------------------------------------------------------------
# Linux constants (from linux/input.h, linux/uinput.h, input-event-codes.h)
# ---------------------------------------------------------------------------

UINPUT_IOCTL_BASE = ord("U")
UINPUT_MAX_NAME_SIZE = 80

UI_SET_EVBIT = _IOW(UINPUT_IOCTL_BASE, 100, ctypes.sizeof(ctypes.c_int))
UI_SET_KEYBIT = _IOW(UINPUT_IOCTL_BASE, 101, ctypes.sizeof(ctypes.c_int))
UI_SET_RELBIT = _IOW(UINPUT_IOCTL_BASE, 102, ctypes.sizeof(ctypes.c_int))
UI_DEV_CREATE = _IO(UINPUT_IOCTL_BASE, 1)
UI_DEV_DESTROY = _IO(UINPUT_IOCTL_BASE, 2)
UI_DEV_SETUP = _IOW(UINPUT_IOCTL_BASE, 3, 0)  # size patched below

EV_SYN = 0x00
EV_KEY = 0x01
EV_REL = 0x02

REL_X = 0x00
REL_Y = 0x01
REL_WHEEL = 0x08

BTN_LEFT = 0x110
BTN_RIGHT = 0x111

BUS_USB = 0x03
INPUT_PROP_DIRECT = 0x01
UINPUT_VERSION = 5  # linux <input.h> UINPUT_VERSION


class _InputId(ctypes.Structure):
    _fields_ = [
        ("bustype", ctypes.c_uint16),
        ("vendor", ctypes.c_uint16),
        ("product", ctypes.c_uint16),
        ("version", ctypes.c_uint16),
    ]


class _UinputSetup(ctypes.Structure):
    """Modern uinput_setup: input_id + padded name + ff_effects_max."""

    _fields_ = [
        ("id", _InputId),
        ("name", ctypes.c_char * UINPUT_MAX_NAME_SIZE),
        ("ff_effects_max", ctypes.c_uint32),
    ]


# Patch the UI_DEV_SETUP size now that the struct layout is known.
UI_DEV_SETUP = _IOW(UINPUT_IOCTL_BASE, 3, ctypes.sizeof(_UinputSetup))


class _InputEvent(ctypes.Structure):
    """struct input_event on 64-bit Linux: two longs (timeval) + 2x16 + 32-bit."""

    _fields_ = [
        ("tv_sec", ctypes.c_long),
        ("tv_usec", ctypes.c_long),
        ("type", ctypes.c_uint16),
        ("code", ctypes.c_uint16),
        ("value", ctypes.c_int32),
    ]


class _LegacyUserDev(ctypes.Structure):
    """Legacy uinput_user_dev fallback (name + id + ff + absolute ranges)."""

    _fields_ = [
        ("name", ctypes.c_char * UINPUT_MAX_NAME_SIZE),
        ("id", _InputId),
        ("ff_effects_max", ctypes.c_uint32),
        ("absmax", ctypes.c_int32 * 64),
        ("absmin", ctypes.c_int32 * 64),
        ("absfuzz", ctypes.c_int32 * 64),
        ("absflat", ctypes.c_int32 * 64),
    ]


# ---------------------------------------------------------------------------
# libc
# ---------------------------------------------------------------------------

def _ioctl(fd, request, arg):
    """Issue an ioctl via fcntl.ioctl.

    The uinput bit-fill ioctls (UI_SET_EVBIT/KEYBIT/RELBIT) take the value
    directly, not a pointer: pass the int itself. The struct ioctls pass a
    ctypes Structure (buffer protocol).
    """
    try:
        result = fcntl.ioctl(fd, request, arg)
    except OSError as exc:
        raise OSError(exc.errno, os.strerror(exc.errno), request) from exc
    return result


# ---------------------------------------------------------------------------
# Key code lookup
# ---------------------------------------------------------------------------

# Linux key codes referenced by the demo controls. All keys exposed by the
# device are declared in KEYCODES; the driver enables every one of them, which
# is a superset and lets the harness send any.
KEYCODES = {
    "w": 17, "s": 31, "a": 30, "d": 32,
    "space": 57, "r": 19, "esc": 1,
    "up": 103, "down": 108, "left": 105, "right": 106,
    "enter": 28, "tab": 15, "backspace": 14, "shift": 42, "ctrl": 29,
    "alt": 56, "1": 2, "2": 3, "3": 4, "0": 11,
}
KEY_ESC = 1

# All EV_KEY codes we want to claim (just the ones in KEYCODES plus a few).
_ALL_KEYCODES = sorted(set(KEYCODES.values()))
_ALL_RELS = [REL_X, REL_Y, REL_WHEEL]
_ALL_BTNS = [BTN_LEFT, BTN_RIGHT]


class UInputError(RuntimeError):
    pass


class InputDevice:
    """A virtual USB keyboard+mouse device backed by /dev/uinput.

    Use as a context manager: the device is destroyed on exit, so a hang or
    crash never leaves a phantom input device behind. All methods emit log
    records so a test can reconstruct exactly what was injected and when.
    """

    def __init__(self, name="GigLine QA Input", path="/dev/uinput"):
        self.name = name
        self.path = path
        self._fd = None
        self._open_and_setup()

    # -- lifecycle ----------------------------------------------------------

    def _open_and_setup(self):
        try:
            self._fd = os.open(self.path, os.O_WRONLY | os.O_NONBLOCK)
        except OSError as exc:
            raise UInputError(
                f"cannot open {self.path}: {exc}. "
                "Needs a udev rule or membership in the 'input' group."
            ) from exc

        fd = self._fd
        try:
            # Declare which event types the device can emit.
            # These bit-fill ioctls take the value directly (not a pointer).
            for bit in (EV_KEY, EV_REL, EV_SYN):
                _ioctl(fd, UI_SET_EVBIT, bit)
            for code in _ALL_KEYCODES + _ALL_BTNS:
                _ioctl(fd, UI_SET_KEYBIT, code)
            for rel in _ALL_RELS:
                _ioctl(fd, UI_SET_RELBIT, rel)

            # Modern path.
            setup = _UinputSetup()
            setup.id.bustype = BUS_USB
            setup.id.vendor = 0x1234
            setup.id.product = 0x5678
            setup.id.version = UINPUT_VERSION
            setup.name = self.name.encode()[: UINPUT_MAX_NAME_SIZE - 1]
            setup.ff_effects_max = 0

            try:
                _ioctl(fd, UI_DEV_SETUP, setup)
            except OSError:
                # Very old kernels: fall back to writing uinput_user_dev.
                legacy = _LegacyUserDev()
                legacy.id = setup.id
                legacy.name = setup.name
                legacy.ff_effects_max = 0
                os.write(fd, legacy)
                log.warning("kernel rejected UI_DEV_SETUP; used legacy write path")

            _ioctl(fd, UI_DEV_CREATE, 0)
            # Give the input core a moment to register the device node.
            time.sleep(0.2)
        except Exception:
            self.close()
            raise

        log.info("virtual input device %r created on %s", self.name, self.path)

    def close(self):
        if self._fd is not None:
            try:
                _ioctl(self._fd, UI_DEV_DESTROY, 0)
            except (OSError, Exception):
                pass
            os.close(self._fd)
            self._fd = None
            log.info("virtual input device %r destroyed", self.name)

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()
        return False

    # -- event emission -----------------------------------------------------

    def _emit(self, etype, code, value):
        """Write an input_event (2 longs timeval + u16/u16/s32) + SYN_REPORT."""
        if self._fd is None:
            raise UInputError("device is closed")
        # struct input_event: __kernel_old_timeval (2x long) + type/code/value.
        ev = struct.pack("llHHi", int(time.time()), 0, etype, code, int(value))
        try:
            os.write(self._fd, ev)
        except OSError as exc:
            log.error("write failed: %s", exc)
            raise
        os.write(self._fd, struct.pack("llHHi", 0, 0, EV_SYN, 0, 0))

    # -- high-level actions -------------------------------------------------

    def _key_code(self, key):
        if isinstance(key, int):
            return key
        key = key.lower()
        if key not in KEYCODES:
            raise UInputError(f"unknown key {key!r}; known: {sorted(KEYCODES)}")
        return KEYCODES[key]

    def press(self, key):
        """Press a key (no release)."""
        code = self._key_code(key)
        self._emit(EV_KEY, code, 1)
        log.info("KEY down  %-6s (code %d)", key, code)

    def release(self, key):
        """Release a previously pressed key."""
        code = self._key_code(key)
        self._emit(EV_KEY, code, 0)
        log.info("KEY up    %-6s (code %d)", key, code)

    def key(self, key, delay=None):
        """Full press+release of a single key. Optional delay between them."""
        self.press(key)
        if delay is not None:
            time.sleep(delay)
        self.release(key)

    def hold(self, key, duration):
        """Hold a key down for `duration` seconds, then release it."""
        self.press(key)
        time.sleep(duration)
        self.release(key)
        log.info("HELD %-6s for %.2fs", key, duration)

    def chord(self, *keys):
        """Press all keys together, then release all in reverse."""
        codes = [self._key_code(k) for k in keys]
        for k, c in zip(keys, codes):
            self._emit(EV_KEY, c, 1)
            log.info("KEY down  %-6s", k)
        for k, c in zip(reversed(keys), reversed(codes)):
            self._emit(EV_KEY, c, 0)
            log.info("KEY up    %-6s", k)
        log.info("CHORD %s", "+".join(keys))

    def move(self, dx, dy, delay=None):
        """Move the (relative) mouse by dx,dy pixels. Watches the pointer."""
        self._emit(EV_REL, REL_X, int(dx))
        self._emit(EV_REL, REL_Y, int(dy))
        if delay is not None:
            time.sleep(delay)
        log.info("MOUSE move (%+d, %+d)", int(dx), int(dy))

    def bpress(self, button="left"):
        self._emit(EV_KEY, BTN_LEFT if button.lower() == "left" else BTN_RIGHT, 1)
        log.info("BUTTON down %s", button)

    def brelease(self, button="left"):
        self._emit(EV_KEY, BTN_LEFT if button.lower() == "left" else BTN_RIGHT, 0)
        log.info("BUTTON up   %s", button)

    def bclick(self, button="left"):
        self.bpress(button)
        time.sleep(0.02)
        self.brelease(button)
        log.info("BUTTON click %s", button)

    def wheel(self, ticks):
        self._emit(EV_REL, REL_WHEEL, int(ticks))
        log.info("WHEEL %+d", int(ticks))


def main():
    # A tiny smoke test: create the device, tap W, nudge the mouse, quit.
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(message)s")
    try:
        with InputDevice() as dev:
            print("# device ready; sending W key + mouse nudge ...")
            dev.key("w")
            dev.move(80, 0)
            time.sleep(0.3)
        print("# ok, device destroyed cleanly")
    except UInputError as exc:
        print(f"ERROR: {exc}")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())