initial commit

1 files changed, 528 insertions, 0 deletions

diff --git a/vrising-cam.py b/vrising-cam.py
new file mode 100755
index 0000000..c2a6666
--- /dev/null
+++ b/vrising-cam.py
@@ -0,0 +1,528 @@
+#!/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. Fix one of two ways:
+  a) quick: rerun with sudo:  sudo {script}
+  b) one-time setup to run as your normal user (recommended):
+       echo 'KERNEL=="uinput", MODE="0660", GROUP="input", TAG+="uaccess"' | sudo tee /etc/udev/rules.d/70-vrising-cam.rules
+       sudo udevadm control --reload-rules && sudo udevadm trigger
+       sudo usermod -aG input $USER
+     (log out and back in so the input group applies)
+"""
+
+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()