Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eb303db316 | ||
|
|
7ad1dcf7cd |
@@ -1 +1,3 @@
|
||||
*.swp
|
||||
__pycache__/
|
||||
*.pyc
|
||||
|
||||
@@ -3,14 +3,15 @@
|
||||
Hotkey and OBS integration on top of [`linkctl`](https://github.com/illegalstudio/linkctl),
|
||||
so AI tracking and gimbal presets are a keypress rather than a terminal command.
|
||||
|
||||
Three pieces, usable independently:
|
||||
Several pieces, usable independently:
|
||||
|
||||
| Path | What it gives you |
|
||||
|---|---|
|
||||
| `bin/linkctl-macro` | A wrapper that makes `linkctl` safe to bind to a hotkey: finds the binary in a desktop session's sparse `PATH`, passes `--force`, and reports the result as a desktop notification. |
|
||||
| `kde/install-shortcuts.sh` | Registers KDE Plasma global shortcuts, so the keys work from anywhere including under Wayland. |
|
||||
| `obs/obs-linkctl.lua` | OBS hotkeys, buttons, and scene automation: a scene switch can set tracking or load a preset on its own. |
|
||||
| `plasmoid/` | A Plasma 6 widget with a virtual joystick, zoom slider, and Home/Center/Tracking buttons. |
|
||||
| `plasmoid/` | A Plasma 6 widget with a virtual joystick, zoom slider, and Home/Center/Tracking buttons. Pointer or WASD/arrow keys. |
|
||||
| `bin/linkctl-gamepad` | Drives the gimbal from a game controller (DualSense and anything else the kernel exposes as a joystick). |
|
||||
|
||||
## Prerequisites
|
||||
|
||||
@@ -37,6 +38,13 @@ Default bindings, editable in the `BINDINGS` table at the top of that script:
|
||||
| `Meta+Alt+0` | tracking off |
|
||||
| `Meta+Alt+H` | `preset load home` |
|
||||
| `Meta+Alt+C` | recenter gimbal |
|
||||
| `Meta+Alt+←/→` | pan 5 degrees |
|
||||
| `Meta+Alt+↑/↓` | tilt 5 degrees |
|
||||
|
||||
The directional ones are discrete nudges: `kglobalaccel` fires once per
|
||||
keypress and offers no press-and-hold, so holding the key will not pan
|
||||
smoothly. Use the widget or a gamepad for continuous motion. `NUDGE=10
|
||||
./kde/install-shortcuts.sh` changes the step.
|
||||
|
||||
Check them afterwards with `./kde/install-shortcuts.sh --list`, which shows the
|
||||
registered key, the default from the desktop entry, and flags any value Plasma
|
||||
@@ -63,12 +71,52 @@ moves, and it stops when you let go. Degrees per step, step interval, deadzone
|
||||
and axis inversion are all in the widget's settings, along with the `linkctl`
|
||||
path and an optional device.
|
||||
|
||||
**WASD and the arrow keys** work whenever the popup has focus, with the same
|
||||
velocity behaviour: hold to move, release to stop. Opposing keys cancel, and a
|
||||
diagonal moves at the same speed as a straight push rather than faster. The
|
||||
buttons and zoom slider are set to take no focus, so the arrow keys keep
|
||||
reaching the pad after you click one. If focus leaves the popup mid-press the
|
||||
held keys are cleared, so the camera cannot be left panning on its own.
|
||||
|
||||
It drives the gimbal with *relative* moves rather than absolute ones, because
|
||||
`linkctl status` reports no pan or tilt at all while the camera is idle, so
|
||||
there is no trustworthy position to aim from. Relative steps work either way
|
||||
and cannot drift out of sync, and linkctl clamps each one to the camera's own
|
||||
range, so holding the stick against a limit is harmless.
|
||||
|
||||
## Game controller
|
||||
|
||||
```sh
|
||||
bin/linkctl-gamepad --list # which joystick devices exist
|
||||
bin/linkctl-gamepad --monitor # print events as you move sticks and press buttons
|
||||
bin/linkctl-gamepad # run it
|
||||
```
|
||||
|
||||
Same velocity behaviour as the widget: push the left stick to move, release to
|
||||
stop. Defaults are a DualSense on the in-tree `hid_playstation` driver, left
|
||||
stick on axes 0 and 1, with buttons 0, 1 and 2 bound to recenter, home preset
|
||||
and tracking toggle.
|
||||
|
||||
**Confirm your mapping with `--monitor` before trusting the defaults.** Axis
|
||||
and button numbering comes from the driver, not the controller, so it varies
|
||||
between kernels and connection modes (a DualSense often enumerates differently
|
||||
over Bluetooth than over USB). `--monitor` prints the number for whatever you
|
||||
touch, and everything is overridable:
|
||||
|
||||
```sh
|
||||
bin/linkctl-gamepad --pan-axis 0 --tilt-axis 1 --step 8 --deadzone 0.2 --button 3="tracking toggle" --button 9="preset load home"
|
||||
```
|
||||
|
||||
It reads `/dev/input/js*` directly, using only the Python standard library.
|
||||
Qt 6 removed QtGamepad with no replacement, so the widget cannot do this
|
||||
itself and this runs as a separate process. You need to be in the `input`
|
||||
group to open the device.
|
||||
|
||||
`--wait` makes it tolerate the controller being absent and reconnect when it
|
||||
returns, which is what you want for a wireless pad that sleeps. There is a
|
||||
user service in `contrib/linkctl-gamepad.service` for running it that way;
|
||||
edit the `ExecStart` path before installing it.
|
||||
|
||||
## OBS
|
||||
|
||||
Tools, Scripts, `+`, then pick `obs/obs-linkctl.lua`.
|
||||
@@ -116,8 +164,11 @@ case where the gimbal is parked facing down.
|
||||
```sh
|
||||
lua tests/test_obs_linkctl.lua # pure helpers: quoting, parsing, commands
|
||||
lua tests/test_obs_lifecycle.lua # the OBS script against a mock OBS API
|
||||
node tests/test_joystick_logic.js # the widget's command building and parsing
|
||||
node tests/test_joystick_logic.js # widget command building, parsing, key vectors
|
||||
python3 tests/test_gamepad.py # gamepad event decoding and deadzone maths
|
||||
tests/test_gamepad_loop.sh # the gamepad read loop, fed through a FIFO
|
||||
bin/linkctl-macro --print tracking on # show the command without running it
|
||||
bin/linkctl-gamepad --dry-run # print moves instead of sending them
|
||||
./kde/install-shortcuts.sh --dry-run # show the KDE changes, apply nothing
|
||||
./kde/install-shortcuts.sh --list # diagnose what is actually registered
|
||||
shellcheck bin/linkctl-macro kde/install-shortcuts.sh plasmoid/install.sh
|
||||
@@ -127,6 +178,12 @@ The lifecycle suite drives the script the way OBS does, so it catches broken
|
||||
wiring (hotkeys not firing, scene actions not matching, the preset name being
|
||||
captured too early) without needing OBS or a camera.
|
||||
|
||||
`test_gamepad_loop.sh` substitutes a FIFO for `/dev/input/js*` and plays
|
||||
scripted events through it, so the select/read/decode/tick path is covered
|
||||
without a controller attached. It is also the test that pins down the
|
||||
initial-state burst: opening a joystick replays every button as currently
|
||||
pressed, and treating that as real input would fire mapped actions at startup.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
**Shortcut does nothing, and System Settings says "No default shortcuts".**
|
||||
@@ -141,6 +198,17 @@ Versions of this repo before that fix wrote the Plasma 5 form.
|
||||
that works but the key does not, check for a conflict in System Settings, and
|
||||
confirm `--list` shows the entry as `present`.
|
||||
|
||||
**Gamepad does nothing.** Check `bin/linkctl-gamepad --list` sees it. If the
|
||||
device is listed but nothing moves, run `--monitor` and push the stick: no
|
||||
events means a permissions or driver problem, events with unexpected axis
|
||||
numbers mean you need `--pan-axis`/`--tilt-axis`. "permission denied" means
|
||||
adding yourself to the `input` group and logging back in.
|
||||
|
||||
**Arrow keys change zoom instead of moving the camera.** That would mean the
|
||||
slider has keyboard focus. The controls are set to `Qt.NoFocus` to prevent it;
|
||||
if you see it, say so, because it means the focus policy is not taking effect
|
||||
on your Plasma build.
|
||||
|
||||
**`tracking` fails with exit 12.** AI tracking is experimental in `linkctl` and
|
||||
goes through the camera's vendor extension unit. Pan, tilt, zoom and presets use
|
||||
standard UVC and are unaffected.
|
||||
@@ -155,7 +223,12 @@ as one argument.
|
||||
|
||||
## Status
|
||||
|
||||
The Lua helpers and the shell scripts are tested offline, and the KDE installer
|
||||
is dry-runnable and reversible. None of it has been exercised against a live
|
||||
camera or a running Plasma session by the author of this commit, so treat the
|
||||
first run as the real test and use `--dry-run` first.
|
||||
The helpers, the shell scripts and the gamepad read loop are tested offline,
|
||||
and the KDE installer is dry-runnable and reversible. Use `--dry-run` first.
|
||||
|
||||
Not verified against real hardware at the time of writing: no live camera, no
|
||||
running Plasma session, and no physical game controller. The gamepad bridge was
|
||||
tested by feeding synthetic kernel events through a FIFO, which exercises every
|
||||
line of the loop but cannot confirm that a DualSense reports the axis and
|
||||
button numbers assumed by the defaults. Treat `--monitor` as a required first
|
||||
step rather than a debugging aid.
|
||||
|
||||
Executable
+427
@@ -0,0 +1,427 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Drive an Insta360 Link gimbal from a game controller.
|
||||
|
||||
Reads the kernel joystick device directly (``/dev/input/js*``) and issues
|
||||
relative ``linkctl`` moves while a stick is held off centre, in the same
|
||||
velocity style as the Plasma widget. Qt 6 removed QtGamepad and offers no
|
||||
replacement, so the applet cannot do this itself.
|
||||
|
||||
Only the standard library is used: the joystick event format is eight bytes
|
||||
and reading it needs no third-party package.
|
||||
|
||||
Discovering the mapping for your controller:
|
||||
|
||||
linkctl-gamepad --list # which devices exist
|
||||
linkctl-gamepad --monitor # print events as you press things
|
||||
|
||||
Then run it, overriding any axis or button number that differs:
|
||||
|
||||
linkctl-gamepad --button 3="tracking toggle"
|
||||
|
||||
Defaults target a DualSense on the in-tree ``hid_playstation`` driver. Button
|
||||
and axis numbering is driver-specific, so treat them as a starting point and
|
||||
confirm with ``--monitor``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import errno
|
||||
import glob
|
||||
import os
|
||||
import select
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
# struct js_event { __u32 time; __s16 value; __u8 type; __u8 number; }
|
||||
JS_EVENT = struct.Struct("<IhBB")
|
||||
JS_EVENT_SIZE = JS_EVENT.size # 8
|
||||
|
||||
JS_EVENT_BUTTON = 0x01
|
||||
JS_EVENT_AXIS = 0x02
|
||||
JS_EVENT_INIT = 0x80 # ORed into type on the initial sync
|
||||
|
||||
AXIS_MAX = 32767.0
|
||||
|
||||
DEFAULT_BUTTONS = {
|
||||
0: "center",
|
||||
1: "preset load home",
|
||||
2: "tracking toggle",
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Pure helpers. Kept free of I/O so tests/test_gamepad.py can exercise them.
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def decode_event(buf: bytes) -> tuple[int, int, int, int]:
|
||||
"""Decode one joystick event into (time, value, type, number).
|
||||
|
||||
The JS_EVENT_INIT flag is stripped from the type and returned separately
|
||||
by `is_init`, because callers almost always want the bare type.
|
||||
"""
|
||||
if len(buf) != JS_EVENT_SIZE:
|
||||
raise ValueError(f"expected {JS_EVENT_SIZE} bytes, got {len(buf)}")
|
||||
tstamp, value, etype, number = JS_EVENT.unpack(buf)
|
||||
return tstamp, value, etype & ~JS_EVENT_INIT, number
|
||||
|
||||
|
||||
def is_init(buf: bytes) -> bool:
|
||||
"""Whether an event is part of the synthetic initial state burst.
|
||||
|
||||
Opening the device replays the current state of every axis and button.
|
||||
Treating those as real presses would fire actions at startup.
|
||||
"""
|
||||
return bool(JS_EVENT.unpack(buf)[2] & JS_EVENT_INIT)
|
||||
|
||||
|
||||
def normalize_axis(raw: int) -> float:
|
||||
"""Scale a raw axis reading to -1.0..1.0."""
|
||||
return max(-1.0, min(1.0, raw / AXIS_MAX))
|
||||
|
||||
|
||||
def apply_deadzone(value: float, deadzone: float) -> float:
|
||||
"""Zero out small readings and rescale the rest.
|
||||
|
||||
Rescaling matters: without it, motion jumps to `deadzone` worth of speed
|
||||
the instant the stick leaves the dead area. With it, speed still starts
|
||||
from zero at the edge of the deadzone.
|
||||
"""
|
||||
if deadzone < 0 or deadzone >= 1:
|
||||
raise ValueError("deadzone must be in [0, 1)")
|
||||
magnitude = abs(value)
|
||||
if magnitude <= deadzone:
|
||||
return 0.0
|
||||
scaled = (magnitude - deadzone) / (1.0 - deadzone)
|
||||
return scaled if value > 0 else -scaled
|
||||
|
||||
|
||||
def movement_commands(
|
||||
nx: float,
|
||||
ny: float,
|
||||
step: float,
|
||||
invert_pan: bool = False,
|
||||
invert_tilt: bool = False,
|
||||
) -> list[list[str]]:
|
||||
"""Build the linkctl argument lists for one tick.
|
||||
|
||||
`ny` is positive upwards. Each axis needs its own call because linkctl's
|
||||
relative commands are single-axis, so a diagonal yields two. Returns an
|
||||
empty list when nothing is worth sending.
|
||||
"""
|
||||
x = -nx if invert_pan else nx
|
||||
y = -ny if invert_tilt else ny
|
||||
out: list[list[str]] = []
|
||||
dx = abs(x) * step
|
||||
dy = abs(y) * step
|
||||
# Below half a degree the camera's own 0.1 degree resolution makes the
|
||||
# step pointless, and it would otherwise creep on a near-centred stick.
|
||||
if dx >= 0.5:
|
||||
out.append(["right" if x > 0 else "left", f"{dx:.1f}"])
|
||||
if dy >= 0.5:
|
||||
out.append(["up" if y > 0 else "down", f"{dy:.1f}"])
|
||||
return out
|
||||
|
||||
|
||||
def parse_button_spec(spec: str) -> tuple[int, str]:
|
||||
"""Parse a ``N=linkctl args`` button mapping."""
|
||||
number, sep, action = spec.partition("=")
|
||||
if not sep:
|
||||
raise argparse.ArgumentTypeError(
|
||||
f"expected NUMBER=ACTION, got {spec!r}")
|
||||
try:
|
||||
index = int(number.strip())
|
||||
except ValueError:
|
||||
raise argparse.ArgumentTypeError(
|
||||
f"button number must be an integer, got {number!r}") from None
|
||||
action = action.strip()
|
||||
if not action:
|
||||
raise argparse.ArgumentTypeError(f"empty action for button {index}")
|
||||
return index, action
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Device discovery
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def device_name(path: str) -> str:
|
||||
"""Read a joystick's advertised name from sysfs, or '' if unavailable."""
|
||||
node = os.path.basename(path)
|
||||
for candidate in (
|
||||
f"/sys/class/input/{node}/device/name",
|
||||
f"/sys/class/input/{node}/name",
|
||||
):
|
||||
try:
|
||||
with open(candidate, "r") as handle:
|
||||
return handle.read().strip()
|
||||
except OSError:
|
||||
continue
|
||||
return ""
|
||||
|
||||
|
||||
def list_devices() -> list[tuple[str, str]]:
|
||||
return [(p, device_name(p))
|
||||
for p in sorted(glob.glob("/dev/input/js*"))]
|
||||
|
||||
|
||||
def pick_device(explicit: str | None, match: str | None) -> str | None:
|
||||
"""Choose a joystick device, preferring an explicit path."""
|
||||
if explicit:
|
||||
return explicit if os.path.exists(explicit) else None
|
||||
devices = list_devices()
|
||||
if not devices:
|
||||
return None
|
||||
if match:
|
||||
needle = match.lower()
|
||||
for path, name in devices:
|
||||
if needle in name.lower():
|
||||
return path
|
||||
return None
|
||||
return devices[0][0]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Runner
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class Runner:
|
||||
"""Invokes linkctl, with a shared base argument list."""
|
||||
|
||||
def __init__(self, binary: str, device: str | None, force: bool,
|
||||
dry_run: bool, verbose: bool):
|
||||
self.base = [binary]
|
||||
if force:
|
||||
self.base.append("--force")
|
||||
if device:
|
||||
self.base += ["--device", device]
|
||||
self.base.append("--quiet")
|
||||
self.dry_run = dry_run
|
||||
self.verbose = verbose
|
||||
|
||||
def __call__(self, args: list[str]) -> None:
|
||||
cmd = self.base + args
|
||||
if self.dry_run:
|
||||
print("would run:", " ".join(cmd), flush=True)
|
||||
return
|
||||
try:
|
||||
# Blocking, with a timeout: these take milliseconds, and letting
|
||||
# them overlap would queue conflicting moves. A hung linkctl
|
||||
# costs one tick rather than wedging the daemon.
|
||||
result = subprocess.run(
|
||||
cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
|
||||
timeout=2.0,
|
||||
)
|
||||
if result.returncode != 0 and self.verbose:
|
||||
err = result.stderr.decode("utf-8", "replace").strip()
|
||||
print(f"linkctl exited {result.returncode}: {err}",
|
||||
file=sys.stderr, flush=True)
|
||||
except subprocess.TimeoutExpired:
|
||||
print("linkctl timed out", file=sys.stderr, flush=True)
|
||||
except FileNotFoundError:
|
||||
print(f"linkctl not found: {self.base[0]}",
|
||||
file=sys.stderr, flush=True)
|
||||
raise SystemExit(127) from None
|
||||
|
||||
|
||||
def monitor(path: str) -> int:
|
||||
"""Print events as they arrive, to discover axis and button numbers."""
|
||||
print(f"Monitoring {path} ({device_name(path) or 'unknown'}).")
|
||||
print("Move sticks and press buttons. Ctrl-C to stop.\n")
|
||||
with open(path, "rb", buffering=0) as handle:
|
||||
while True:
|
||||
buf = handle.read(JS_EVENT_SIZE)
|
||||
if not buf or len(buf) < JS_EVENT_SIZE:
|
||||
return 0
|
||||
_, value, etype, number = decode_event(buf)
|
||||
tag = " (initial)" if is_init(buf) else ""
|
||||
if etype == JS_EVENT_AXIS:
|
||||
print(f"axis {number:>2} value {value:>6} "
|
||||
f"({normalize_axis(value):+.2f}){tag}", flush=True)
|
||||
elif etype == JS_EVENT_BUTTON:
|
||||
state = "pressed" if value else "released"
|
||||
print(f"button {number:>2} {state}{tag}", flush=True)
|
||||
|
||||
|
||||
def run(args: argparse.Namespace, run_cmd: Runner) -> int:
|
||||
buttons = dict(DEFAULT_BUTTONS) if args.default_buttons else {}
|
||||
buttons.update(dict(args.button or []))
|
||||
|
||||
axes: dict[int, float] = {}
|
||||
interval = args.interval / 1000.0
|
||||
|
||||
while True:
|
||||
path = pick_device(args.js_device, args.match)
|
||||
if path is None:
|
||||
if not args.wait:
|
||||
print("no joystick found; use --list to see devices, or "
|
||||
"--wait to keep retrying", file=sys.stderr)
|
||||
return 3
|
||||
time.sleep(2.0)
|
||||
continue
|
||||
|
||||
try:
|
||||
handle = open(path, "rb", buffering=0)
|
||||
except OSError as exc:
|
||||
if exc.errno in (errno.EACCES, errno.EPERM):
|
||||
print(f"cannot open {path}: permission denied. Add yourself "
|
||||
"to the 'input' group.", file=sys.stderr)
|
||||
return 6
|
||||
time.sleep(2.0)
|
||||
continue
|
||||
|
||||
print(f"Using {path} ({device_name(path) or 'unknown'})", flush=True)
|
||||
axes.clear()
|
||||
next_tick = time.monotonic() + interval
|
||||
|
||||
try:
|
||||
with handle:
|
||||
while True:
|
||||
now = time.monotonic()
|
||||
timeout = max(0.0, next_tick - now)
|
||||
ready, _, _ = select.select([handle], [], [], timeout)
|
||||
|
||||
if ready:
|
||||
chunk = handle.read(JS_EVENT_SIZE * 32)
|
||||
if not chunk:
|
||||
raise OSError(errno.ENODEV, "device went away")
|
||||
for offset in range(0, len(chunk) - JS_EVENT_SIZE + 1,
|
||||
JS_EVENT_SIZE):
|
||||
buf = chunk[offset:offset + JS_EVENT_SIZE]
|
||||
_, value, etype, number = decode_event(buf)
|
||||
if etype == JS_EVENT_AXIS:
|
||||
axes[number] = normalize_axis(value)
|
||||
elif (etype == JS_EVENT_BUTTON and value
|
||||
and not is_init(buf)):
|
||||
action = buttons.get(number)
|
||||
if action:
|
||||
run_cmd(action.split())
|
||||
elif args.verbose:
|
||||
print(f"button {number} unmapped",
|
||||
flush=True)
|
||||
|
||||
now = time.monotonic()
|
||||
if now < next_tick:
|
||||
continue
|
||||
# Skip rather than catch up, so a slow linkctl cannot
|
||||
# build a backlog of stale moves.
|
||||
next_tick = now + interval
|
||||
|
||||
raw_x = axes.get(args.pan_axis, 0.0)
|
||||
raw_y = axes.get(args.tilt_axis, 0.0)
|
||||
nx = apply_deadzone(raw_x, args.deadzone)
|
||||
# Sticks report positive downwards; the rest of the
|
||||
# codebase treats positive as up.
|
||||
ny = -apply_deadzone(raw_y, args.deadzone)
|
||||
for cmd in movement_commands(nx, ny, args.step,
|
||||
args.invert_pan,
|
||||
args.invert_tilt):
|
||||
run_cmd(cmd)
|
||||
except KeyboardInterrupt:
|
||||
return 0
|
||||
except OSError as exc:
|
||||
print(f"{path} closed ({exc.strerror or exc}); waiting for the "
|
||||
"controller to come back", file=sys.stderr, flush=True)
|
||||
if not args.wait:
|
||||
return 8
|
||||
time.sleep(2.0)
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
p = argparse.ArgumentParser(
|
||||
prog="linkctl-gamepad",
|
||||
description="Drive an Insta360 Link gimbal from a game controller.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="Run --monitor to discover your controller's axis and button "
|
||||
"numbers, which are driver-specific.",
|
||||
)
|
||||
p.add_argument("--list", action="store_true",
|
||||
help="list joystick devices and exit")
|
||||
p.add_argument("--monitor", action="store_true",
|
||||
help="print controller events and exit")
|
||||
|
||||
p.add_argument("--js-device", metavar="PATH",
|
||||
help="joystick device (default: first match)")
|
||||
p.add_argument("--match", metavar="TEXT", default="DualSense",
|
||||
help="substring of the controller name to select "
|
||||
"(default: %(default)s; use '' for any)")
|
||||
p.add_argument("--wait", action="store_true",
|
||||
help="keep retrying when the controller is absent, and "
|
||||
"reconnect when it returns")
|
||||
|
||||
p.add_argument("--pan-axis", type=int, default=0, metavar="N",
|
||||
help="axis for pan (default: %(default)s)")
|
||||
p.add_argument("--tilt-axis", type=int, default=1, metavar="N",
|
||||
help="axis for tilt (default: %(default)s)")
|
||||
p.add_argument("--deadzone", type=float, default=0.15, metavar="F",
|
||||
help="ignore deflection below this, 0..1 "
|
||||
"(default: %(default)s)")
|
||||
p.add_argument("--step", type=float, default=6.0, metavar="DEG",
|
||||
help="degrees per tick at full deflection "
|
||||
"(default: %(default)s)")
|
||||
p.add_argument("--interval", type=int, default=160, metavar="MS",
|
||||
help="milliseconds between ticks (default: %(default)s)")
|
||||
p.add_argument("--invert-pan", action="store_true")
|
||||
p.add_argument("--invert-tilt", action="store_true")
|
||||
|
||||
p.add_argument("--button", action="append", type=parse_button_spec,
|
||||
metavar="N=ACTION",
|
||||
help="map a button to linkctl arguments, repeatable, "
|
||||
"e.g. --button 3=\"tracking toggle\"")
|
||||
p.add_argument("--no-default-buttons", dest="default_buttons",
|
||||
action="store_false",
|
||||
help="do not apply the built-in button mappings")
|
||||
|
||||
p.add_argument("--linkctl", default="linkctl", metavar="PATH",
|
||||
help="linkctl binary (default: %(default)s)")
|
||||
p.add_argument("--device", metavar="PATH",
|
||||
help="camera video node, passed to linkctl --device")
|
||||
p.add_argument("--no-force", dest="force", action="store_false",
|
||||
help="respect linkctl's idle-camera guard")
|
||||
p.add_argument("--dry-run", action="store_true",
|
||||
help="print commands instead of running them")
|
||||
p.add_argument("-v", "--verbose", action="store_true")
|
||||
return p
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = build_parser().parse_args(argv)
|
||||
|
||||
if args.deadzone < 0 or args.deadzone >= 1:
|
||||
print("--deadzone must be in [0, 1)", file=sys.stderr)
|
||||
return 2
|
||||
if args.interval < 10:
|
||||
print("--interval must be at least 10ms", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
if args.list:
|
||||
devices = list_devices()
|
||||
if not devices:
|
||||
print("no joystick devices found under /dev/input")
|
||||
return 3
|
||||
for path, name in devices:
|
||||
print(f"{path} {name or '(unknown)'}")
|
||||
return 0
|
||||
|
||||
if args.monitor:
|
||||
path = pick_device(args.js_device, args.match or None)
|
||||
if path is None:
|
||||
print("no matching joystick found; try --list or --match ''",
|
||||
file=sys.stderr)
|
||||
return 3
|
||||
try:
|
||||
return monitor(path)
|
||||
except KeyboardInterrupt:
|
||||
return 0
|
||||
|
||||
runner = Runner(args.linkctl, args.device, args.force,
|
||||
args.dry_run, args.verbose)
|
||||
try:
|
||||
return run(args, runner)
|
||||
except KeyboardInterrupt:
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,35 @@
|
||||
# Run the gamepad bridge as a user service.
|
||||
#
|
||||
# Install:
|
||||
# mkdir -p ~/.config/systemd/user
|
||||
# cp contrib/linkctl-gamepad.service ~/.config/systemd/user/
|
||||
# # edit ExecStart below to the absolute path of bin/linkctl-gamepad
|
||||
# systemctl --user daemon-reload
|
||||
# systemctl --user enable --now linkctl-gamepad.service
|
||||
#
|
||||
# Watch it:
|
||||
# journalctl --user -fu linkctl-gamepad.service
|
||||
#
|
||||
# --wait makes it tolerate the controller being absent or asleep, so the
|
||||
# service can stay enabled whether or not the pad is connected.
|
||||
|
||||
[Unit]
|
||||
Description=Insta360 Link gamepad control
|
||||
Documentation=https://gitbox.example.com/insta360-link-tools
|
||||
After=graphical-session.target
|
||||
PartOf=graphical-session.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
# Absolute path required: systemd does not search PATH for ExecStart.
|
||||
ExecStart=%h/workspace/insta360-link-tools/bin/linkctl-gamepad --wait
|
||||
Restart=on-failure
|
||||
RestartSec=5
|
||||
# The joystick device and the camera are the only things it touches.
|
||||
PrivateTmp=true
|
||||
ProtectSystem=strict
|
||||
ProtectHome=read-only
|
||||
NoNewPrivileges=true
|
||||
|
||||
[Install]
|
||||
WantedBy=graphical-session.target
|
||||
@@ -19,6 +19,9 @@
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
# Degrees moved by one press of a directional shortcut.
|
||||
NUDGE=${NUDGE:-5}
|
||||
|
||||
# id | key sequence | linkctl-macro arguments | human label
|
||||
#
|
||||
# Meta+Alt is used because Plasma leaves that combination largely free.
|
||||
@@ -29,6 +32,13 @@ BINDINGS=(
|
||||
"tracking-off|Meta+Alt+0|tracking off|Insta360: AI tracking off"
|
||||
"preset-home|Meta+Alt+H|preset load home|Insta360: Load home preset"
|
||||
"center|Meta+Alt+C|center|Insta360: Recenter gimbal"
|
||||
# Directional nudges. kglobalaccel fires once per keypress and gives no
|
||||
# press-and-hold, so each of these is a discrete step rather than smooth
|
||||
# motion. For continuous movement use the Plasma widget or a gamepad.
|
||||
"pan-left|Meta+Alt+Left|left $NUDGE|Insta360: Pan left"
|
||||
"pan-right|Meta+Alt+Right|right $NUDGE|Insta360: Pan right"
|
||||
"tilt-up|Meta+Alt+Up|up $NUDGE|Insta360: Tilt up"
|
||||
"tilt-down|Meta+Alt+Down|down $NUDGE|Insta360: Tilt down"
|
||||
)
|
||||
|
||||
# The entries are visible by default: KApplicationTrader skips NoDisplay
|
||||
|
||||
@@ -51,6 +51,40 @@ function tickCommand(base, nx, ny, stepMax, invertPan, invertTilt) {
|
||||
return parts.join(" ; ");
|
||||
}
|
||||
|
||||
/*
|
||||
* Turn a set of held direction keys into a stick vector.
|
||||
*
|
||||
* `held` has boolean left/right/up/down members. Opposing keys cancel, and a
|
||||
* diagonal is normalised to length 1 so holding two keys does not move the
|
||||
* camera faster than holding one.
|
||||
*
|
||||
* Key codes are deliberately not mapped here: this file is a `.pragma
|
||||
* library` and so cannot reach the Qt namespace for Qt.Key_*. The QML side
|
||||
* resolves a key code to a direction name and passes the result in.
|
||||
*/
|
||||
function vectorFromKeys(held) {
|
||||
var x = 0;
|
||||
var y = 0;
|
||||
if (held.left) {
|
||||
x -= 1;
|
||||
}
|
||||
if (held.right) {
|
||||
x += 1;
|
||||
}
|
||||
if (held.up) {
|
||||
y += 1;
|
||||
}
|
||||
if (held.down) {
|
||||
y -= 1;
|
||||
}
|
||||
var len = Math.sqrt(x * x + y * y);
|
||||
if (len > 1) {
|
||||
x /= len;
|
||||
y /= len;
|
||||
}
|
||||
return { nx: x, ny: y };
|
||||
}
|
||||
|
||||
/*
|
||||
* Merge every JSON object found in linkctl's stdout into a flat update.
|
||||
*
|
||||
|
||||
@@ -36,6 +36,61 @@ PlasmoidItem {
|
||||
property string statusText: "Not queried yet"
|
||||
property bool busy: false
|
||||
|
||||
// Held movement keys. Separate booleans rather than one object, because
|
||||
// mutating a member of a `var` object does not re-evaluate bindings.
|
||||
property bool keyLeft: false
|
||||
property bool keyRight: false
|
||||
property bool keyUp: false
|
||||
property bool keyDown: false
|
||||
|
||||
readonly property var keyVec: Logic.vectorFromKeys({
|
||||
left: root.keyLeft, right: root.keyRight,
|
||||
up: root.keyUp, down: root.keyDown
|
||||
})
|
||||
readonly property bool keyEngaged: root.keyVec.nx !== 0 || root.keyVec.ny !== 0
|
||||
|
||||
// The pointer wins while it is down, so a drag is never fought over by a
|
||||
// key left held down underneath it.
|
||||
readonly property real inNx: stick.held ? stick.nx : root.keyVec.nx
|
||||
readonly property real inNy: stick.held ? stick.ny : root.keyVec.ny
|
||||
readonly property bool inEngaged: stick.engaged || root.keyEngaged
|
||||
|
||||
// WASD and the arrow keys. Returns "" for anything that is not a
|
||||
// movement key, so the handler can leave it unaccepted.
|
||||
function dirForKey(key) {
|
||||
switch (key) {
|
||||
case Qt.Key_A:
|
||||
case Qt.Key_Left:
|
||||
return "left";
|
||||
case Qt.Key_D:
|
||||
case Qt.Key_Right:
|
||||
return "right";
|
||||
case Qt.Key_W:
|
||||
case Qt.Key_Up:
|
||||
return "up";
|
||||
case Qt.Key_S:
|
||||
case Qt.Key_Down:
|
||||
return "down";
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
||||
function setKey(dir, down) {
|
||||
switch (dir) {
|
||||
case "left": root.keyLeft = down; break;
|
||||
case "right": root.keyRight = down; break;
|
||||
case "up": root.keyUp = down; break;
|
||||
case "down": root.keyDown = down; break;
|
||||
}
|
||||
}
|
||||
|
||||
function clearKeys() {
|
||||
root.keyLeft = false;
|
||||
root.keyRight = false;
|
||||
root.keyUp = false;
|
||||
root.keyDown = false;
|
||||
}
|
||||
|
||||
function baseCmd() {
|
||||
return Logic.baseCmd(root.linkctl, root.device);
|
||||
}
|
||||
@@ -98,10 +153,49 @@ PlasmoidItem {
|
||||
Component.onCompleted: refresh()
|
||||
|
||||
fullRepresentation: ColumnLayout {
|
||||
id: panel
|
||||
Layout.minimumWidth: Kirigami.Units.gridUnit * 14
|
||||
Layout.minimumHeight: Kirigami.Units.gridUnit * 22
|
||||
spacing: Kirigami.Units.smallSpacing
|
||||
|
||||
// Keyboard control lives here rather than on the pad so the keys work
|
||||
// as soon as the popup opens, without having to click the pad first.
|
||||
focus: true
|
||||
Component.onCompleted: forceActiveFocus()
|
||||
|
||||
Keys.onPressed: (event) => {
|
||||
// Auto-repeat would otherwise look like a rapid press/release
|
||||
// cycle and make the motion stutter.
|
||||
if (event.isAutoRepeat) {
|
||||
event.accepted = true;
|
||||
return;
|
||||
}
|
||||
const d = root.dirForKey(event.key);
|
||||
if (d !== "") {
|
||||
root.setKey(d, true);
|
||||
event.accepted = true;
|
||||
}
|
||||
}
|
||||
Keys.onReleased: (event) => {
|
||||
if (event.isAutoRepeat) {
|
||||
event.accepted = true;
|
||||
return;
|
||||
}
|
||||
const d = root.dirForKey(event.key);
|
||||
if (d !== "") {
|
||||
root.setKey(d, false);
|
||||
event.accepted = true;
|
||||
}
|
||||
}
|
||||
|
||||
// Losing focus mid-press means the release never arrives, which would
|
||||
// leave the camera panning on its own.
|
||||
onActiveFocusChanged: {
|
||||
if (!activeFocus) {
|
||||
root.clearKeys();
|
||||
}
|
||||
}
|
||||
|
||||
PlasmaComponents.Label {
|
||||
Layout.fillWidth: true
|
||||
horizontalAlignment: Text.AlignHCenter
|
||||
@@ -128,17 +222,26 @@ PlasmoidItem {
|
||||
onReleased: root.refresh()
|
||||
}
|
||||
|
||||
PlasmaComponents.Label {
|
||||
Layout.fillWidth: true
|
||||
horizontalAlignment: Text.AlignHCenter
|
||||
text: "Drag, or use WASD / arrow keys"
|
||||
opacity: 0.6
|
||||
font: Kirigami.Theme.smallFont
|
||||
}
|
||||
|
||||
Timer {
|
||||
// Drives the stick. Only runs while it is meaningfully displaced.
|
||||
// Drives both input methods. Only runs while something is asking
|
||||
// for movement, so an idle popup costs nothing.
|
||||
interval: root.tickMs
|
||||
repeat: true
|
||||
running: stick.engaged
|
||||
running: root.inEngaged
|
||||
triggeredOnStart: true
|
||||
onTriggered: {
|
||||
if (root.busy) {
|
||||
return; // let the previous step finish, skip this one
|
||||
}
|
||||
const cmd = Logic.tickCommand(root.baseCmd(), stick.nx, stick.ny,
|
||||
const cmd = Logic.tickCommand(root.baseCmd(), root.inNx, root.inNy,
|
||||
root.stepMax, root.invertPan,
|
||||
root.invertTilt);
|
||||
if (cmd !== "") {
|
||||
@@ -147,6 +250,17 @@ PlasmoidItem {
|
||||
}
|
||||
}
|
||||
|
||||
// Refresh the readout once keyboard movement stops, matching what the
|
||||
// pad does on release.
|
||||
Connections {
|
||||
target: root
|
||||
function onKeyEngagedChanged() {
|
||||
if (!root.keyEngaged) {
|
||||
root.refresh();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
RowLayout {
|
||||
Layout.fillWidth: true
|
||||
spacing: Kirigami.Units.smallSpacing
|
||||
@@ -157,6 +271,8 @@ PlasmoidItem {
|
||||
PlasmaComponents.Slider {
|
||||
id: zoomSlider
|
||||
Layout.fillWidth: true
|
||||
// Keep arrow keys with the panel instead of nudging zoom.
|
||||
focusPolicy: Qt.NoFocus
|
||||
from: 1.0
|
||||
to: 4.0
|
||||
stepSize: 0.1
|
||||
@@ -178,6 +294,7 @@ PlasmoidItem {
|
||||
spacing: Kirigami.Units.smallSpacing
|
||||
|
||||
PlasmaComponents.Button {
|
||||
focusPolicy: Qt.NoFocus
|
||||
Layout.fillWidth: true
|
||||
icon.name: "go-home"
|
||||
text: "Home"
|
||||
@@ -185,6 +302,7 @@ PlasmoidItem {
|
||||
+ Logic.shq(Plasmoid.configuration.homePreset))
|
||||
}
|
||||
PlasmaComponents.Button {
|
||||
focusPolicy: Qt.NoFocus
|
||||
Layout.fillWidth: true
|
||||
icon.name: "crosshairs"
|
||||
text: "Center"
|
||||
@@ -193,6 +311,7 @@ PlasmoidItem {
|
||||
}
|
||||
|
||||
PlasmaComponents.Button {
|
||||
focusPolicy: Qt.NoFocus
|
||||
Layout.fillWidth: true
|
||||
icon.name: root.trackingState === 1 ? "view-visible" : "view-hidden"
|
||||
text: root.trackingState === 1 ? "Tracking: on"
|
||||
@@ -202,6 +321,7 @@ PlasmoidItem {
|
||||
}
|
||||
|
||||
PlasmaComponents.Button {
|
||||
focusPolicy: Qt.NoFocus
|
||||
Layout.fillWidth: true
|
||||
icon.name: "view-refresh"
|
||||
text: "Refresh"
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for linkctl-gamepad's pure helpers.
|
||||
|
||||
Run: python3 tests/test_gamepad.py
|
||||
|
||||
These cover what can be checked without a controller: decoding the kernel's
|
||||
joystick event struct, deadzone rescaling, command construction and argument
|
||||
parsing. Device discovery and the read loop need real hardware.
|
||||
"""
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import struct
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
TARGET = os.path.join(HERE, "..", "bin", "linkctl-gamepad")
|
||||
|
||||
# The script has no .py extension, so load it by path.
|
||||
_spec = importlib.util.spec_from_loader(
|
||||
"linkctl_gamepad",
|
||||
importlib.machinery.SourceFileLoader("linkctl_gamepad", TARGET),
|
||||
)
|
||||
gp = importlib.util.module_from_spec(_spec)
|
||||
_spec.loader.exec_module(gp)
|
||||
|
||||
|
||||
def js_event(value, etype, number, tstamp=0):
|
||||
"""Build a raw joystick event the way the kernel would."""
|
||||
return struct.pack("<IhBB", tstamp, value, etype, number)
|
||||
|
||||
|
||||
class TestDecode(unittest.TestCase):
|
||||
def test_axis_event(self):
|
||||
buf = js_event(-32767, gp.JS_EVENT_AXIS, 1)
|
||||
_, value, etype, number = gp.decode_event(buf)
|
||||
self.assertEqual(value, -32767)
|
||||
self.assertEqual(etype, gp.JS_EVENT_AXIS)
|
||||
self.assertEqual(number, 1)
|
||||
self.assertFalse(gp.is_init(buf))
|
||||
|
||||
def test_button_event(self):
|
||||
buf = js_event(1, gp.JS_EVENT_BUTTON, 3)
|
||||
_, value, etype, number = gp.decode_event(buf)
|
||||
self.assertEqual((value, etype, number), (1, gp.JS_EVENT_BUTTON, 3))
|
||||
|
||||
def test_init_flag_is_stripped_but_reported(self):
|
||||
# Opening the device replays current state with INIT set. The type
|
||||
# must still decode as a button, and is_init must flag it, or the
|
||||
# daemon would fire mapped actions on startup.
|
||||
buf = js_event(1, gp.JS_EVENT_BUTTON | gp.JS_EVENT_INIT, 0)
|
||||
_, _, etype, _ = gp.decode_event(buf)
|
||||
self.assertEqual(etype, gp.JS_EVENT_BUTTON)
|
||||
self.assertTrue(gp.is_init(buf))
|
||||
|
||||
def test_init_axis(self):
|
||||
buf = js_event(0, gp.JS_EVENT_AXIS | gp.JS_EVENT_INIT, 2)
|
||||
_, _, etype, number = gp.decode_event(buf)
|
||||
self.assertEqual((etype, number), (gp.JS_EVENT_AXIS, 2))
|
||||
self.assertTrue(gp.is_init(buf))
|
||||
|
||||
def test_short_buffer_rejected(self):
|
||||
with self.assertRaises(ValueError):
|
||||
gp.decode_event(b"\x00" * 7)
|
||||
|
||||
def test_event_size_matches_kernel_struct(self):
|
||||
self.assertEqual(gp.JS_EVENT_SIZE, 8)
|
||||
|
||||
|
||||
class TestNormalize(unittest.TestCase):
|
||||
def test_extremes_and_centre(self):
|
||||
self.assertAlmostEqual(gp.normalize_axis(32767), 1.0)
|
||||
self.assertAlmostEqual(gp.normalize_axis(-32767), -1.0)
|
||||
self.assertAlmostEqual(gp.normalize_axis(0), 0.0)
|
||||
|
||||
def test_clamps_beyond_nominal_range(self):
|
||||
# Some drivers report -32768, one past the nominal minimum.
|
||||
self.assertAlmostEqual(gp.normalize_axis(-32768), -1.0)
|
||||
self.assertAlmostEqual(gp.normalize_axis(40000), 1.0)
|
||||
|
||||
def test_midpoint(self):
|
||||
self.assertAlmostEqual(gp.normalize_axis(16384), 0.5, places=3)
|
||||
|
||||
|
||||
class TestDeadzone(unittest.TestCase):
|
||||
def test_inside_deadzone_is_zero(self):
|
||||
self.assertEqual(gp.apply_deadzone(0.1, 0.15), 0.0)
|
||||
self.assertEqual(gp.apply_deadzone(-0.1, 0.15), 0.0)
|
||||
self.assertEqual(gp.apply_deadzone(0.15, 0.15), 0.0)
|
||||
|
||||
def test_rescales_from_zero_at_the_edge(self):
|
||||
# The point of rescaling: just past the edge must be near zero speed,
|
||||
# not jump straight to 0.15 worth of it.
|
||||
self.assertAlmostEqual(gp.apply_deadzone(0.16, 0.15), 0.01 / 0.85,
|
||||
places=6)
|
||||
|
||||
def test_full_deflection_is_preserved(self):
|
||||
self.assertAlmostEqual(gp.apply_deadzone(1.0, 0.15), 1.0)
|
||||
self.assertAlmostEqual(gp.apply_deadzone(-1.0, 0.15), -1.0)
|
||||
|
||||
def test_sign_is_preserved(self):
|
||||
self.assertLess(gp.apply_deadzone(-0.5, 0.15), 0)
|
||||
self.assertGreater(gp.apply_deadzone(0.5, 0.15), 0)
|
||||
|
||||
def test_zero_deadzone_is_identity(self):
|
||||
self.assertAlmostEqual(gp.apply_deadzone(0.42, 0.0), 0.42)
|
||||
|
||||
def test_invalid_deadzone_rejected(self):
|
||||
for bad in (-0.1, 1.0, 1.5):
|
||||
with self.assertRaises(ValueError):
|
||||
gp.apply_deadzone(0.5, bad)
|
||||
|
||||
|
||||
class TestMovementCommands(unittest.TestCase):
|
||||
def test_centred_stick_sends_nothing(self):
|
||||
self.assertEqual(gp.movement_commands(0.0, 0.0, 6.0), [])
|
||||
|
||||
def test_tiny_deflection_sends_nothing(self):
|
||||
# 0.01 * 6 = 0.06 degrees, below the half-degree floor.
|
||||
self.assertEqual(gp.movement_commands(0.01, 0.01, 6.0), [])
|
||||
|
||||
def test_cardinal_directions(self):
|
||||
self.assertEqual(gp.movement_commands(1.0, 0.0, 6.0),
|
||||
[["right", "6.0"]])
|
||||
self.assertEqual(gp.movement_commands(-1.0, 0.0, 6.0),
|
||||
[["left", "6.0"]])
|
||||
self.assertEqual(gp.movement_commands(0.0, 1.0, 6.0),
|
||||
[["up", "6.0"]])
|
||||
self.assertEqual(gp.movement_commands(0.0, -1.0, 6.0),
|
||||
[["down", "6.0"]])
|
||||
|
||||
def test_partial_deflection_scales_the_step(self):
|
||||
self.assertEqual(gp.movement_commands(0.5, 0.0, 6.0),
|
||||
[["right", "3.0"]])
|
||||
|
||||
def test_diagonal_yields_both_axes(self):
|
||||
self.assertEqual(gp.movement_commands(1.0, 1.0, 6.0),
|
||||
[["right", "6.0"], ["up", "6.0"]])
|
||||
|
||||
def test_off_axis_component_below_floor_is_dropped(self):
|
||||
# A mostly-horizontal push must not also creep vertically.
|
||||
self.assertEqual(gp.movement_commands(1.0, 0.05, 6.0),
|
||||
[["right", "6.0"]])
|
||||
|
||||
def test_inversion(self):
|
||||
self.assertEqual(gp.movement_commands(1.0, 0.0, 6.0, invert_pan=True),
|
||||
[["left", "6.0"]])
|
||||
self.assertEqual(gp.movement_commands(0.0, 1.0, 6.0, invert_tilt=True),
|
||||
[["down", "6.0"]])
|
||||
self.assertEqual(
|
||||
gp.movement_commands(1.0, 1.0, 6.0, invert_pan=True,
|
||||
invert_tilt=True),
|
||||
[["left", "6.0"], ["down", "6.0"]])
|
||||
|
||||
def test_arguments_are_separate_list_items(self):
|
||||
# They are passed to subprocess without a shell, so the degrees must
|
||||
# not be glued onto the direction.
|
||||
for cmd in gp.movement_commands(1.0, 1.0, 6.0):
|
||||
self.assertEqual(len(cmd), 2)
|
||||
|
||||
|
||||
class TestButtonSpec(unittest.TestCase):
|
||||
def test_simple(self):
|
||||
self.assertEqual(gp.parse_button_spec("0=center"), (0, "center"))
|
||||
|
||||
def test_multiword_action(self):
|
||||
self.assertEqual(gp.parse_button_spec("1=preset load home"),
|
||||
(1, "preset load home"))
|
||||
|
||||
def test_whitespace_tolerated(self):
|
||||
self.assertEqual(gp.parse_button_spec(" 2 = tracking toggle "),
|
||||
(2, "tracking toggle"))
|
||||
|
||||
def test_rejects_missing_separator(self):
|
||||
import argparse
|
||||
with self.assertRaises(argparse.ArgumentTypeError):
|
||||
gp.parse_button_spec("center")
|
||||
|
||||
def test_rejects_non_numeric_button(self):
|
||||
import argparse
|
||||
with self.assertRaises(argparse.ArgumentTypeError):
|
||||
gp.parse_button_spec("cross=center")
|
||||
|
||||
def test_rejects_empty_action(self):
|
||||
import argparse
|
||||
with self.assertRaises(argparse.ArgumentTypeError):
|
||||
gp.parse_button_spec("0=")
|
||||
|
||||
|
||||
class TestRunner(unittest.TestCase):
|
||||
def test_base_command_assembly(self):
|
||||
r = gp.Runner("linkctl", None, True, True, False)
|
||||
self.assertEqual(r.base, ["linkctl", "--force", "--quiet"])
|
||||
|
||||
def test_device_and_no_force(self):
|
||||
r = gp.Runner("/usr/bin/linkctl", "/dev/video0", False, True, False)
|
||||
self.assertEqual(
|
||||
r.base, ["/usr/bin/linkctl", "--device", "/dev/video0", "--quiet"])
|
||||
|
||||
def test_dry_run_does_not_execute(self):
|
||||
import io
|
||||
import contextlib
|
||||
r = gp.Runner("linkctl", None, True, True, False)
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
r(["left", "6.0"])
|
||||
self.assertIn("linkctl --force --quiet left 6.0", buf.getvalue())
|
||||
|
||||
|
||||
class TestCliParsing(unittest.TestCase):
|
||||
def test_defaults(self):
|
||||
a = gp.build_parser().parse_args([])
|
||||
self.assertEqual((a.pan_axis, a.tilt_axis), (0, 1))
|
||||
self.assertAlmostEqual(a.deadzone, 0.15)
|
||||
self.assertTrue(a.force)
|
||||
self.assertTrue(a.default_buttons)
|
||||
|
||||
def test_repeatable_buttons(self):
|
||||
a = gp.build_parser().parse_args(
|
||||
["--button", "3=tracking toggle", "--button", "4=center"])
|
||||
self.assertEqual(a.button, [(3, "tracking toggle"), (4, "center")])
|
||||
|
||||
def test_rejects_bad_deadzone(self):
|
||||
self.assertEqual(gp.main(["--deadzone", "1.5"]), 2)
|
||||
|
||||
def test_rejects_tiny_interval(self):
|
||||
self.assertEqual(gp.main(["--interval", "1"]), 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Executable
+114
@@ -0,0 +1,114 @@
|
||||
#!/usr/bin/env bash
|
||||
#
|
||||
# End-to-end test of linkctl-gamepad's read loop, without a controller.
|
||||
#
|
||||
# Feeds synthetic joystick events through a FIFO used in place of
|
||||
# /dev/input/js*, and checks what the daemon would run. This is the only test
|
||||
# that covers select/read/decode/tick/dispatch together; tests/test_gamepad.py
|
||||
# covers the pure helpers.
|
||||
#
|
||||
# Run: tests/test_gamepad_loop.sh
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
HERE=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)
|
||||
readonly HERE
|
||||
readonly GAMEPAD="$HERE/../bin/linkctl-gamepad"
|
||||
|
||||
failures=0
|
||||
checks=0
|
||||
OUT=$(mktemp)
|
||||
FIFO=""
|
||||
|
||||
cleanup() { rm -f "$OUT" ${FIFO:+"$FIFO"}; }
|
||||
trap cleanup EXIT
|
||||
|
||||
check() {
|
||||
local label=$1 pattern=$2
|
||||
checks=$((checks + 1))
|
||||
if grep -qF -- "$pattern" "$OUT"; then
|
||||
printf 'ok %s\n' "$label"
|
||||
else
|
||||
failures=$((failures + 1))
|
||||
printf 'FAIL %s\n expected to find: %s\n' "$label" "$pattern"
|
||||
fi
|
||||
}
|
||||
|
||||
check_absent() {
|
||||
local label=$1 pattern=$2
|
||||
checks=$((checks + 1))
|
||||
if grep -qF -- "$pattern" "$OUT"; then
|
||||
failures=$((failures + 1))
|
||||
printf 'FAIL %s\n should not have found: %s\n' "$label" "$pattern"
|
||||
else
|
||||
printf 'ok %s\n' "$label"
|
||||
fi
|
||||
}
|
||||
|
||||
# Run the daemon against a FIFO fed by the given python writer script.
|
||||
run_scenario() {
|
||||
local script=$1
|
||||
FIFO=$(mktemp -u /tmp/linkctl-gamepad-test.XXXXXX)
|
||||
mkfifo "$FIFO"
|
||||
python3 -c "$script" "$FIFO" &
|
||||
local writer=$!
|
||||
sleep 0.1
|
||||
timeout 5 "$GAMEPAD" --js-device "$FIFO" --match '' --dry-run \
|
||||
--interval 200 --verbose > "$OUT" 2>&1
|
||||
wait "$writer" 2>/dev/null
|
||||
rm -f "$FIFO"
|
||||
FIFO=""
|
||||
}
|
||||
|
||||
readonly PRELUDE='
|
||||
import struct, sys, time
|
||||
AXIS, BUTTON, INIT = 0x02, 0x01, 0x80
|
||||
def ev(value, etype, number):
|
||||
return struct.pack("<IhBB", 0, value, etype, number)
|
||||
f = open(sys.argv[1], "wb", buffering=0)
|
||||
'
|
||||
|
||||
echo "=== scenario 1: full deflection, diagonal, button, disconnect ==="
|
||||
run_scenario "$PRELUDE"'
|
||||
# Initial-state burst. Button 0 reads as pressed; the daemon must not treat
|
||||
# that as a real press and fire its mapped action.
|
||||
f.write(ev(0, AXIS | INIT, 0)); f.write(ev(0, AXIS | INIT, 1))
|
||||
f.write(ev(1, BUTTON | INIT, 0))
|
||||
time.sleep(0.4)
|
||||
f.write(ev(32767, AXIS, 0)) # pan hard right
|
||||
time.sleep(0.5)
|
||||
f.write(ev(-32767, AXIS, 1)) # tilt hard up (negative is up)
|
||||
time.sleep(0.5)
|
||||
f.write(ev(0, AXIS, 0)); f.write(ev(0, AXIS, 1))
|
||||
time.sleep(0.3)
|
||||
f.write(ev(1, BUTTON, 2)) # real press: tracking toggle
|
||||
time.sleep(0.4)
|
||||
f.close()
|
||||
'
|
||||
echo "--- output ---"; cat "$OUT"; echo "---"
|
||||
|
||||
check "selects the device" "Using "
|
||||
check "pans right at full deflection" "--quiet right 6.0"
|
||||
check "tilts up at full deflection" "--quiet up 6.0"
|
||||
check "real button press fires" "--quiet tracking toggle"
|
||||
check "handles disconnect" "waiting for the controller"
|
||||
# Button 0 maps to `center` by default and appeared only in the INIT burst,
|
||||
# so seeing it would mean INIT filtering is broken.
|
||||
check_absent "ignores the initial-state burst" "--quiet center"
|
||||
|
||||
echo
|
||||
echo "=== scenario 2: deflection inside the deadzone ==="
|
||||
# 4000/32767 is about 0.12, under the 0.15 default deadzone, so the daemon
|
||||
# must stay completely still.
|
||||
run_scenario "$PRELUDE"'
|
||||
f.write(ev(0, AXIS | INIT, 0)); f.write(ev(0, AXIS | INIT, 1))
|
||||
time.sleep(0.3)
|
||||
f.write(ev(4000, AXIS, 0))
|
||||
time.sleep(0.8)
|
||||
f.close()
|
||||
'
|
||||
echo "--- output ---"; cat "$OUT"; echo "---"
|
||||
check_absent "deadzone suppresses movement entirely" "would run:"
|
||||
|
||||
printf '\n%d checks, %d failures\n' "$checks" "$failures"
|
||||
[[ $failures -eq 0 ]]
|
||||
@@ -72,6 +72,40 @@ eq(ctx.tickCommand(B, 0, 1, 6, false, true), B + " down 6.0", "invert tilt");
|
||||
eq(ctx.tickCommand(B, 1, 1, 6, true, true), B + " left 6.0 ; " + B + " down 6.0",
|
||||
"invert both");
|
||||
|
||||
// --- vectorFromKeys ----------------------------------------------------
|
||||
const K = (o) => Object.assign(
|
||||
{ left: false, right: false, up: false, down: false }, o);
|
||||
|
||||
deep(ctx.vectorFromKeys(K({})), { nx: 0, ny: 0 }, "no keys held is centred");
|
||||
deep(ctx.vectorFromKeys(K({ right: true })), { nx: 1, ny: 0 }, "right key");
|
||||
deep(ctx.vectorFromKeys(K({ left: true })), { nx: -1, ny: 0 }, "left key");
|
||||
deep(ctx.vectorFromKeys(K({ up: true })), { nx: 0, ny: 1 }, "up key");
|
||||
deep(ctx.vectorFromKeys(K({ down: true })), { nx: 0, ny: -1 }, "down key");
|
||||
|
||||
// Opposing keys must cancel, not fight or latch.
|
||||
deep(ctx.vectorFromKeys(K({ left: true, right: true })), { nx: 0, ny: 0 },
|
||||
"left and right cancel");
|
||||
deep(ctx.vectorFromKeys(K({ up: true, down: true })), { nx: 0, ny: 0 },
|
||||
"up and down cancel");
|
||||
deep(ctx.vectorFromKeys(K({ left: true, right: true, up: true })),
|
||||
{ nx: 0, ny: 1 }, "cancelling one axis leaves the other");
|
||||
|
||||
// A diagonal must not be faster than a straight push.
|
||||
const diag = ctx.vectorFromKeys(K({ right: true, up: true }));
|
||||
eq(Math.round(Math.sqrt(diag.nx * diag.nx + diag.ny * diag.ny) * 1e6) / 1e6, 1,
|
||||
"diagonal is normalised to length 1");
|
||||
eq(diag.nx.toFixed(4), "0.7071", "diagonal x component");
|
||||
eq(diag.ny.toFixed(4), "0.7071", "diagonal y component");
|
||||
|
||||
// And a diagonal still produces a command on both axes at the default step.
|
||||
eq(ctx.tickCommand(B, diag.nx, diag.ny, 6, false, false),
|
||||
B + " right 4.2 ; " + B + " up 4.2", "diagonal keys move both axes");
|
||||
|
||||
// All four held at once is a standstill, not a crash.
|
||||
deep(ctx.vectorFromKeys(
|
||||
K({ left: true, right: true, up: true, down: true })),
|
||||
{ nx: 0, ny: 0 }, "all four keys cancel out");
|
||||
|
||||
// --- absorb ------------------------------------------------------------
|
||||
deep(ctx.absorb('{"pan":12.3,"tilt":null}'), { pan: 12.3 },
|
||||
"absorb takes pan, ignores null tilt");
|
||||
|
||||
Reference in New Issue
Block a user