path: root/vrising-cam.py

#!/usr/bin/env python3
"""V Rising camera helper.

Hold Q / E to rotate the camera: emulates holding the in-game "Rotate camera"
key (right mouse button) while streaming relative mouse motion, so it behaves
the same on X11 and Wayland. Single file, no services: the first run creates a
venv next to this script and installs evdev; the script only acts while this
process is running (Ctrl+C to stop, RMB is always released on exit).
"""

import argparse
import atexit
import errno
import fcntl
import glob
import os
import select
import signal
import subprocess
import sys
import time
from pathlib import Path

SCRIPT = Path(__file__).resolve()
VENV_DIR = SCRIPT.parent / ".venv"
VENV_PY = VENV_DIR / "bin" / "python"

VENV_HINT = """\
error: could not create the virtualenv: python3 -m venv {dir}
Debian/Ubuntu: sudo apt install python3-venv
"""

BUILD_HINT = """\
error: pip failed inside the venv (see stderr above).
If the error says "No module named pip": delete {dir} and rerun; you are
likely missing the venv/ensurepip package (Debian/Ubuntu: python3-venv).
Otherwise the evdev source build needs a toolchain, Python headers AND the
kernel input headers (linux/input.h):
  Debian/Ubuntu: sudo apt install build-essential python3-dev linux-headers-amd64
  Fedora:        sudo dnf install gcc python3-devel kernel-headers
  Arch:          sudo pacman -S base-devel linux-headers
  Alpine:        apk add build-base python3-dev linux-headers
"""

UINPUT_HINT = """\
uinput access is missing (/dev/uinput could not be opened for writing).
Fix one of two ways:
  a) quick: rerun with sudo:  sudo {script}
  b) grant your user write access to /dev/uinput: membership in the
     "input" group is enough on most distros (Arch included); if your
     distro ships /dev/uinput root-only (e.g. Debian/Ubuntu), add a udev
     rule for it, e.g.:
       echo 'KERNEL=="uinput", MODE="0660", GROUP="input"' | sudo tee /etc/udev/rules.d/70-vrising-cam.rules
       sudo udevadm control --reload-rules && sudo udevadm trigger
     (then log out and back in if the group membership is new)
"""

DEV_PERM_HINT = """\
some /dev/input/event* devices are not readable by your user. Fix one of:
  a) quick: rerun with sudo:  sudo {script}
  b) one-time: sudo usermod -aG input $USER   (then log out and back in)
"""

MODULE_HINT = """\
the uinput kernel module is not loaded. Fix:
  sudo modprobe uinput
  echo uinput | sudo tee /etc/modules-load.d/uinput.conf    (load at every boot)
If modprobe says the module is unknown, your kernel was built without
CONFIG_INPUT_UINPUT and you would need a kernel that includes it.
"""


def _bootstrap():
    if Path(sys.prefix).resolve() != VENV_DIR:
        if not VENV_PY.exists():
            print(f"vrising-cam: creating venv at {VENV_DIR} ...", flush=True)
            r = subprocess.run([sys.executable, "-m", "venv", str(VENV_DIR)])
            if r.returncode != 0 or not VENV_PY.exists():
                sys.stderr.write(VENV_HINT.format(dir=VENV_DIR))
                sys.exit(1)
        os.execv(str(VENV_PY), [str(VENV_PY), str(SCRIPT), *sys.argv[1:]])
    print("vrising-cam: installing evdev into the venv ...", flush=True)
    r = subprocess.run(
        [str(VENV_PY), "-m", "pip", "install", "evdev-binary"],
        capture_output=True,
        text=True,
    )
    if r.returncode != 0:
        print("vrising-cam: no prebuilt wheel for this platform, building evdev from source ...", flush=True)
        r = subprocess.run(
            [str(VENV_PY), "-m", "pip", "install", "evdev"],
            capture_output=True,
            text=True,
        )
    if r.returncode != 0:
        sys.stderr.write((r.stderr or "")[-2000:])
        sys.stderr.write("\n" + BUILD_HINT.format(dir=VENV_DIR))
        sys.exit(1)
    print("vrising-cam: evdev installed.", flush=True)


try:
    from evdev import InputDevice, UInput, ecodes, list_devices
except ImportError:
    _bootstrap()
    from evdev import InputDevice, UInput, ecodes, list_devices


def _on_sigterm(signum, frame):
    raise SystemExit(130)


def _say(msg):
    print(msg, flush=True)


def _list_event_devices():
    try:
        return sorted(glob.glob("/dev/input/event*"))
    except OSError:
        return []


def scan_keyboards():
    keyboards, unreadable = [], []
    for path in _list_event_devices():
        try:
            dev = InputDevice(path)
        except (OSError, PermissionError):
            unreadable.append(path)
            continue
        try:
            keys = set(dev.capabilities(absinfo=False).get(ecodes.EV_KEY, ()))
        except (OSError, PermissionError):
            unreadable.append(path)
            continue
        if {ecodes.KEY_Q, ecodes.KEY_E} <= keys:
            keyboards.append(dev)
        else:
            dev.close()
    return keyboards, unreadable


def check_grabs(keyboards, log):
    """Detect devices that another process has grabbed exclusively (EVIOCGRAB).

    If someone else holds an exclusive grab (e.g. keyd/kanata/kmonad/remappers),
    our reads on that device return nothing and Q/E will never be detected.
    We try to grab for a microsecond and release immediately; EBUSY means taken.
    """
    grabbed = getattr(ecodes, "EVIOCGRAB", None)
    if grabbed is None:
        grabbed = 0x40044592  # _IOW('E', 0x92, int)
    for dev in keyboards:
        try:
            fcntl.ioctl(dev.fd, grabbed, 1)
        except OSError as e:
            if e.errno == errno.EBUSY:
                log(f"WARNING: {dev.name!r} ({dev.path}) is exclusively grabbed by "
                    "another process (keyd/kanata/kmonad/solaar ...); its key events "
                    "cannot be seen here and Q/E on it will be missed.")
        else:
            try:
                fcntl.ioctl(dev.fd, grabbed, 0)
            except OSError as e:
                log(f"warning: could not release grab probe on {dev.path}: {e}")


def uinput_status():
    p = Path("/dev/uinput")
    if not p.exists():
        return False, "/dev/uinput does not exist", MODULE_HINT
    try:
        fd = os.open(p, os.O_RDWR | os.O_NONBLOCK)
    except OSError as e:
        if e.errno in (errno.ENXIO, errno.ENODEV):
            return False, (
                f"/dev/uinput cannot be opened ({os.strerror(e.errno)}); "
                "the uinput kernel module is probably not loaded"
            ), MODULE_HINT
        return False, f"/dev/uinput cannot be opened ({e})", UINPUT_HINT
    os.close(fd)
    return True, None, None


class Injector:
    def __init__(self, verbose=False, log=None):
        self.ui = UInput(
            {
                ecodes.EV_KEY: [
                    ecodes.BTN_LEFT,
                    ecodes.BTN_RIGHT,
                    ecodes.BTN_MIDDLE,
                ],
                ecodes.EV_REL: [ecodes.REL_X, ecodes.REL_Y],
            },
            name="vrising-cam virtual mouse",
            bustype=ecodes.BUS_USB,
            vendor=0x1209,
            product=0x5356,
            version=1,
        )
        self.rmb = False
        self.verbose = verbose
        self.log = log or _say
        self._closed = False

    def press(self):
        self.ui.write(ecodes.EV_KEY, ecodes.BTN_RIGHT, 1)
        self.ui.syn()
        self.rmb = True
        if self.verbose:
            self.log("inject: RMB press")

    def release(self):
        if self.rmb:
            self.ui.write(ecodes.EV_KEY, ecodes.BTN_RIGHT, 0)
            self.ui.syn()
            self.rmb = False
            if self.verbose:
                self.log("inject: RMB release")

    def move(self, dx):
        self.ui.write(ecodes.EV_REL, ecodes.REL_X, dx)
        self.ui.syn()

    def close(self):
        if self._closed:
            return
        self._closed = True
        self.release()
        self.ui.close()


def run_check():
    ok_ui, ui_msg, ui_hint = uinput_status()
    keyboards, unreadable = scan_keyboards()
    if ok_ui:
        print("/dev/uinput: present and openable for writing")
    else:
        print("/dev/uinput: NOT USABLE")
        if ui_msg:
            print(f"  {ui_msg}")
        if ui_hint:
            sys.stderr.write(ui_hint)
    print(f"keyboards exposing Q+E: {len(keyboards)}")
    for dev in keyboards:
        print(f"  {dev.path}  {dev.name}")
    if keyboards:
        check_grabs(keyboards, print)
    if unreadable:
        print(f"unreadable event devices: {len(unreadable)}")
        for path in unreadable[:8]:
            print(f"  {path}")
        if len(unreadable) > 8:
            print(f"  ... and {len(unreadable) - 8} more")
        sys.stderr.write(DEV_PERM_HINT.format(script=SCRIPT))
    if not Path("/dev/input").is_dir():
        print("  note: /dev/input does not exist (running in a container?)")
    ready = ok_ui and bool(keyboards)
    print("READY" if ready else "NOT READY")
    return 0 if ready else 1


def parse_args():
    p = argparse.ArgumentParser(
        prog="vrising-cam",
        description=(
            "Hold Q / E to rotate the V Rising camera: emulates holding the in-game "
            "'Rotate camera' key (right mouse button) while dragging the mouse. "
            "Runs only while this process is alive; the first run auto-creates "
            ".venv next to the script and installs evdev."
        ),
    )
    p.add_argument("--speed", type=int, default=2000, metavar="UNITS/SEC",
                   help="rotation speed in relative X units per second "
                        "(default: 2000)")
    p.add_argument("--tick-ms", type=float, default=1.0, metavar="MS",
                   help="milliseconds between motion ticks (default: 1.0; "
                        "lower = smoother, higher = lighter)")
    p.add_argument("--ease", type=float, default=0.0, metavar="SECONDS",
                   help="ramp the rotation speed up over SECONDS on press and "
                        "decay it on release (0 = off, try 0.15)")
    p.add_argument("--invert", action="store_true",
                   help="swap the rotation direction of Q and E")
    p.add_argument("--verbose", "-v", action="store_true",
                   help="log key presses/releases, RMB injection and motion ticks")
    p.add_argument("--test-rmb", type=float, nargs="?", const=3.0, default=0, metavar="SECONDS",
                   help="hold the emulated right mouse button for SECONDS "
                        "(default 3, or 0 to disable), then exit; ignores Q/E "
                        "and is used to check that the desktop reacts to the "
                        "injected button at all")
    p.add_argument("--check", action="store_true",
                   help="show detected keyboards and permission status, then exit")
    return p.parse_args()


def run_test_rmb(hold_s, lead_s=3.0):
    ok_ui, ui_msg, ui_hint = uinput_status()
    if not ok_ui:
        print(f"error: {ui_msg}", flush=True)
        if ui_hint:
            sys.stderr.write(ui_hint)
        return 1
    try:
        inj = Injector()
    except (OSError, PermissionError) as e:
        print(f"error: could not create the uinput device: {e}", flush=True)
        sys.stderr.write(UINPUT_HINT.format(script=SCRIPT))
        return 1
    atexit.register(inj.close)
    reader = inj.ui.device
    if reader is None:
        print("note: could not open a read-back handle for the virtual device; "
              "kernel self-check will be skipped", flush=True)
    print(f"move the mouse over something useful NOW; injected RMB press in "
          f"{lead_s:.0f} s ...", flush=True)
    for i in range(int(lead_s), 0, -1):
        print(f"  {i} s ...", flush=True)
        time.sleep(1)

    def drain():
        read = []
        if reader is None:
            return read
        while True:
            ready, _, _ = select.select([reader.fd], [], [], 0)
            if not ready:
                break
            try:
                events = list(reader.read())
            except (OSError, BlockingIOError):
                break
            if not events:
                break
            read.extend(events)
        return read

    print(f"holding injected RIGHT mouse button for {hold_s:.1f} s ...", flush=True)
    print("watch the CURSOR: it should twitch left/right during the hold (if "
          "it does not move at all, the session is not receiving this "
          "device's motion either).", flush=True)
    try:
        inj.press()
        steps = max(2, int(hold_s / 0.25))
        for i in range(steps):
            inj.move(1 if i % 2 else -1)
            time.sleep(0.25 if i < steps - 1 else 0.05)
    except KeyboardInterrupt:
        pass
    seen = drain()
    inj.release()
    seen += drain()
    inj.close()
    buttons = [ev for ev in seen if ev.type == ecodes.EV_KEY and ev.code == ecodes.BTN_RIGHT]
    rels = [ev for ev in seen if ev.type == ecodes.EV_REL and ev.code == ecodes.REL_X]
    values = sorted(ev.value for ev in buttons)
    if reader is not None and values and {0, 1} <= set(values):
        print(f"kernel read-back: OK - {len(buttons)} BTN_RIGHT events "
              f"(down+up) and {len(rels)} REL_X events came back from the "
              "kernel. Injection works at kernel level; if the desktop still "
              "showed nothing, the problem is session-side delivery "
              "(KWin/libinput), not this script.", flush=True)
    elif reader is not None:
        print(f"kernel read-back: FAILED - kernel delivered "
              f"{len(buttons)} BTN_RIGHT / {len(rels)} REL_X events back "
              "despite injection; the writes never made it into the input "
              "stack. Report this output.", flush=True)
    else:
        print(f"test-rmb done, RMB released. ({len(rels)} REL_X events were "
              "not self-verified: read-back handle unavailable.)", flush=True)
    print("if the desktop showed NO reaction to the hold/release, install the "
          "debug tool on the PC (it is in the 'libinput-tools' package on Arch:\n"
          "  sudo pacman -S libinput-tools\nthen keep this running in another "
          "terminal\n  sudo libinput debug-events\nwhile rerunning "
          "python vrising-cam.py --test-rmb  and check whether a 'BTN_RIGHT' "
          "line appears there. Also useful:\n  udevadm info /sys/devices/virtual/input/inputNN "
          "(the 'vrising-cam virtual mouse' sysfs node from /proc/bus/input/devices)",
          flush=True)
    if reader is not None:
        print(f"(virtual device node for read-back: {reader.path})", flush=True)
    return 0


def main():
    args = parse_args()
    if args.check:
        sys.exit(run_check())
    if args.test_rmb:
        sys.exit(run_test_rmb(args.test_rmb))
    ok_ui, ui_msg, ui_hint = uinput_status()
    if not ok_ui:
        if ui_msg:
            print(f"error: {ui_msg}", flush=True)
        if ui_hint:
            sys.stderr.write(ui_hint)
        sys.exit(1)
    keyboards, unreadable = scan_keyboards()
    if not keyboards:
        print("error: no readable keyboard device exposing Q and E was found.", flush=True)
        if unreadable:
            sys.stderr.write(DEV_PERM_HINT.format(script=SCRIPT))
        sys.exit(1)
    keymap = {ecodes.KEY_Q: -1, ecodes.KEY_E: 1}
    if args.invert:
        keymap = {code: -sign for code, sign in keymap.items()}
    try:
        inj = Injector(verbose=args.verbose)
    except (OSError, PermissionError) as e:
        print(f"error: could not create the uinput device: {e}", flush=True)
        sys.stderr.write(UINPUT_HINT.format(script=SCRIPT))
        sys.exit(1)
    atexit.register(inj.close)
    signal.signal(signal.SIGTERM, _on_sigterm)
    if args.verbose:
        check_grabs(keyboards, _say)
    watched = ", ".join(f"{dev.name!r} ({dev.path})" for dev in keyboards)
    print("vrising-cam running - hold Q to rotate left, E to rotate right. Ctrl+C to stop.", flush=True)
    print(f"watching {len(keyboards)} keyboard(s): {watched}", flush=True)
    print(f"emulation: RMB hold + {args.speed} units/s over {args.tick_ms} ms ticks"
          + (f", {args.ease:.2f} s ease" if args.ease > 0 else ""), flush=True)
    fds = {dev.fd: dev for dev in keyboards}
    state = {code: False for code in keymap}
    tick_s = min(0.05, max(0.0005, args.tick_ms / 1000.0))
    max_dt = max(0.05, 4.0 * tick_s)  # clamp late wake-ups, never throttle ticks
    last = 0.0               # last emission timestamp while active
    acc = 0.0                # fractional unit accumulator
    idle = 0.25
    verbose = args.verbose
    lit = {ecodes.KEY_Q: "Q", ecodes.KEY_E: "E"}
    win_total = 0            # units emitted in the verbose log window
    window_t0 = 0.0
    ease_s = args.ease
    press_t = 0.0            # ramp anchor for the current ease-in
    mult = 1.0               # current speed multiplier (0..1, ease only)
    decel = False            # keys released: ease-out until RMB unpresses
    last_dir = 0             # direction to keep decaying after release
    try:
        while True:
            active = state[ecodes.KEY_Q] or state[ecodes.KEY_E]
            emitting = active or decel
            now = time.monotonic()
            timeout = idle if not emitting else max(0.0, last + tick_s - now)
            ready, _, _ = select.select(list(fds), [], [], timeout)
            for fd in ready:
                dev = fds.get(fd)
                if dev is None:
                    continue
                try:
                    events = dev.read()
                except OSError:
                    print(f"warning: lost device {dev.name} ({dev.path})", flush=True)
                    fds.pop(fd, None)
                    continue
                for ev in events:
                    if ev.type == ecodes.EV_KEY and ev.code in keymap and ev.value in (0, 1):
                        state[ev.code] = ev.value == 1
                        if verbose:
                            _say(f"key: {lit[ev.code]} {'press' if ev.value else 'release'} "
                                 f"({dev.name!r}, {dev.path})")
            if not fds:
                print("error: all watched devices are gone, exiting.", flush=True)
                sys.exit(1)
            active = state[ecodes.KEY_Q] or state[ecodes.KEY_E]
            if active:
                now = time.monotonic()
                if not inj.rmb:
                    inj.press()
                    last = now
                    acc = 0.0
                    win_total = 0
                    window_t0 = now
                    press_t = now
                    decel = False
                elif decel:
                    decel = False
                    press_t = now - mult * ease_s   # continue ramp seamless
                dt = min(max(0.0, now - last), max_dt)
                last = now
                direction = sum(state[c] * s for c, s in keymap.items())
                if direction:
                    last_dir = direction
            else:
                now = time.monotonic()
                if inj.rmb and not decel:
                    if ease_s > 0:
                        decel = True     # keys gone: ease-out, keep last_dir
                    else:
                        inj.release()
                dt = min(max(0.0, now - last), max_dt)
                last = now
                direction = last_dir if decel else 0
            if ease_s > 0:
                if decel:
                    mult = max(0.0, mult - dt / ease_s)
                elif active:
                    t = (now - press_t) / ease_s
                    mult = min(1.0, t * t * (3.0 - 2.0 * t)) if t < 1.0 else 1.0
                else:
                    mult = 1.0
            else:
                mult = 1.0
            if decel and mult <= 0.0:
                inj.release()
                decel = False
                direction = 0
            if direction and mult > 0.0:
                acc += args.speed * mult * dt
                n = int(acc)
                if n:
                    acc -= n
                    dx = n * direction
                    inj.move(dx)
                    win_total += n
                    if verbose and now - window_t0 >= 2.0:
                        measured = win_total / (now - window_t0)
                        _say(f"motion: {win_total:+d} units in "
                             f"{now - window_t0:.1f} s "
                             f"(~{measured:+.0f} actual vs {args.speed:+d} "
                             "requested)")
                        win_total = 0
                        window_t0 = now
    except KeyboardInterrupt:
        pass
    finally:
        inj.close()
        print("vrising-cam stopped, RMB released.", flush=True)


if __name__ == "__main__":
    main()