feat: relative mouse on the HID interface, and the verifier to prove it
The board now enumerates as a boot keyboard plus a relative mouse on the same HID interface. The pointer is captured, never grabbed: Pointer Lock in the page, so Esc (the browser's own, never forwarded) always frees it, and every frame restates the whole button byte so a frame lost in the WiFi cannot leave a button held. Losing the lock, the tab or the link runs the same release funnel as a stuck key, and boot protocol or `mouse off` refuses mouse frames with E11 rather than sending bytes a boot interface would misparse as keys. src/mouse.h owns the report, src/output.h the queues and the funnel; the page, relayctl.py and watch_target.py carry the button bits and the evdev decoding. The verifier's own tables are pinned now — REL_WHEEL is 8, not 2, and the first version guessed from the report's field order, so a working wheel looked dead while the summary said no scroll happened. LLM-Generated: true
This commit is contained in:
+12
-2
@@ -83,8 +83,18 @@ def main() -> int:
|
||||
sys.stdout.flush()
|
||||
return 0
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSAFLUSH, saved)
|
||||
os.close(fd)
|
||||
# `reboot` makes the device disappear underneath us: the CDC node is gone the
|
||||
# instant the board resets, so restoring the port settings can legitimately fail.
|
||||
# The command already reached the board -- do not turn a successful `reboot` into
|
||||
# a Python traceback and a non-zero exit.
|
||||
try:
|
||||
termios.tcsetattr(fd, termios.TCSAFLUSH, saved)
|
||||
except (OSError, termios.error): # termios.error is NOT an OSError here
|
||||
pass
|
||||
try:
|
||||
os.close(fd)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -8,6 +8,9 @@ installed to be useful.
|
||||
relayctl.py --host keyrelay.local status
|
||||
relayctl.py --host 192.168.4.1 arm
|
||||
relayctl.py --host 10.2.1.99 key Enter # one key, press and release
|
||||
relayctl.py --host 10.2.1.99 mouse move 120 -40 # relative movement, split to fit
|
||||
relayctl.py --host 10.2.1.99 mouse click right # press and release
|
||||
relayctl.py --host 10.2.1.99 mouse drag 200 0 # press, move, release, in order
|
||||
relayctl.py --host 10.2.1.99 cad # Ctrl+Alt+Del, with --yes
|
||||
relayctl.py --host keyrelay.local listen # print every status heartbeat
|
||||
relayctl.py --host keyrelay.local disarm
|
||||
@@ -24,6 +27,14 @@ a controller, so no separate `arm` command could have done it), the text is neve
|
||||
echoed back, and the run disarms again afterwards. What is still absent is a script or
|
||||
macro facility: the board cannot store a sequence, cannot be told to type later, and
|
||||
nothing here replays one. See README "What this deliberately is not".
|
||||
|
||||
`mouse` is the same argument with a pointer. Movement is relative -- the board reports how
|
||||
far, never where -- so the target's own pointer speed decides where a given `move` lands,
|
||||
and there is no position for this tool to be wrong about. `click` and `drag` press real
|
||||
buttons and need --yes for that reason; a movement does not, because moving a cursor is
|
||||
reversible by definition. There is no gesture recording, no auto-click and no "move to the
|
||||
OK button", because the moment this tool could actuate a target on its own it would be a
|
||||
different and worse device than a keyboard.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -49,6 +60,22 @@ MAX_SESSION_KEYS = 16
|
||||
TEXT_FRAME_MAX = 120 # characters per P= frame
|
||||
MAX_PASTE_CHARS = 2048 # characters one run holds on the board
|
||||
DEFAULT_TEXT_PACE_MS = 12 # what the board uses when asked for 0
|
||||
|
||||
# Mouse movement. Both numbers are the board's -- keep them in step with src/config.h
|
||||
# (kMouseDeltaMax, kMouseButtonMask) or this tool builds frames the board refuses.
|
||||
# One M= frame is one HID report, and a report carries one signed byte per axis, so a
|
||||
# larger movement is split into several frames here rather than guessed at there.
|
||||
MOUSE_DELTA_MAX = 127 # one signed byte, per axis, per frame
|
||||
MOUSE_BUTTON_MASK = 0x1f # five buttons: all the report can carry
|
||||
MOUSE_FRAME_MS = 8 # the pace the controller page sends at, matched here
|
||||
|
||||
# Button name -> the report's bit, in the order the HID report descriptor lays them out.
|
||||
# NOTE the DOM's MouseEvent.button numbers are NOT this order (there 1 is middle and 2 is
|
||||
# right), which is why src/web_page.h carries its own mapping and both are pinned by
|
||||
# tests: get it wrong and a right-click arrives as a middle-click, cleanly parsed.
|
||||
MOUSE_BUTTONS = {
|
||||
"left": 0x01, "right": 0x02, "middle": 0x04, "back": 0x08, "forward": 0x10,
|
||||
}
|
||||
GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
|
||||
OP_TEXT, OP_PING, OP_PONG, OP_CLOSE = 0x1, 0x9, 0xA, 0x8
|
||||
|
||||
@@ -117,6 +144,87 @@ def untypeable(text: str) -> list[str]:
|
||||
if not (ord(ch) == 9 or ord(ch) == 10 or 32 <= ord(ch) <= 126)]
|
||||
|
||||
|
||||
def mouse_frame(buttons: int, dx: int = 0, dy: int = 0, wheel: int = 0,
|
||||
pan: int = 0) -> str:
|
||||
"""Render `M=<btn>:<dx>:<dy>:<wheel>:<pan>` -- one mouse report, five fields.
|
||||
|
||||
Strict in the same direction the firmware parses: the button byte must fit the five
|
||||
bits the report has and every movement must fit one signed byte, or this raises. A
|
||||
frame the board would refuse is not sent, and nothing is clamped to make it fit --
|
||||
clamp a 300 and the cursor quietly travels half as far as was asked.
|
||||
"""
|
||||
if buttons & ~MOUSE_BUTTON_MASK:
|
||||
raise RelayError(f"buttons {buttons:#02x} does not fit a five-button report "
|
||||
f"(have: {', '.join(MOUSE_BUTTONS)})")
|
||||
for name, v in (("dx", dx), ("dy", dy), ("wheel", wheel), ("pan", pan)):
|
||||
if not -MOUSE_DELTA_MAX <= v <= MOUSE_DELTA_MAX:
|
||||
raise RelayError(f"{name}={v} is more than one report's worth "
|
||||
f"(+/-{MOUSE_DELTA_MAX}); split it")
|
||||
return f"M={buttons:02x}:{dx}:{dy}:{wheel}:{pan}"
|
||||
|
||||
|
||||
def split_moves(dx: int, dy: int, wheel: int = 0, pan: int = 0) -> list[tuple]:
|
||||
"""Greedy split of a movement into one-report chunks, same shape as the page's.
|
||||
|
||||
Both axes travel together in each chunk, so a diagonal keeps its shape rather than
|
||||
arriving as an L, and the sums are exact -- 300 px out means 300 px on the target.
|
||||
"""
|
||||
out: list[tuple] = []
|
||||
while dx or dy or wheel or pan:
|
||||
step = [max(-MOUSE_DELTA_MAX, min(MOUSE_DELTA_MAX, v)) for v in (dx, dy, wheel, pan)]
|
||||
out.append(tuple(step))
|
||||
dx -= step[0]
|
||||
dy -= step[1]
|
||||
wheel -= step[2]
|
||||
pan -= step[3]
|
||||
return out or [(0, 0, 0, 0)]
|
||||
|
||||
|
||||
def mouse_frames(action: str, values: list[str], button: str = "left") -> list[tuple]:
|
||||
"""Turn a `mouse` subcommand into the frames to send, as (buttons, dx, dy, wheel, pan).
|
||||
|
||||
A click is two frames -- the button restated down, then restated up -- with the
|
||||
movement in between, because the byte on the wire is the whole button state and the
|
||||
order is the entire difference between a click and a drag.
|
||||
"""
|
||||
if button not in MOUSE_BUTTONS:
|
||||
# Checked for every action, not only the ones that use it: `--button thumb` is a
|
||||
# typo, and a command that quietly ignores a nonsense option is one that will be
|
||||
# typed again.
|
||||
raise RelayError(f"unknown button {button!r} (have: {', '.join(MOUSE_BUTTONS)})")
|
||||
if button != "left" and action not in ("click", "drag"):
|
||||
# `mouse move 100 0 --button right` would otherwise move the cursor and pretend the
|
||||
# button was involved. It is not: only the verbs that press something take a
|
||||
# button, and saying so beats inventing a meaning for it later.
|
||||
raise RelayError(f"`{action}` does not press a button, so --button has no "
|
||||
f"meaning here (click and drag do take one)")
|
||||
|
||||
if action == "release":
|
||||
return [(0, 0, 0, 0, 0)]
|
||||
if action == "move":
|
||||
if len(values) not in (2, 3, 4):
|
||||
raise RelayError("usage: mouse move DX DY [WHEEL] [PAN]")
|
||||
nums = [int(v) for v in values] + [0] * (4 - len(values))
|
||||
return [(0, *rest) for rest in split_moves(*nums)]
|
||||
if action == "wheel":
|
||||
if len(values) != 1:
|
||||
raise RelayError("usage: mouse wheel NOTCHES (negative scrolls up)")
|
||||
# split_moves returns (dx, dy, wheel, pan); the wheel belongs in the third slot.
|
||||
return [(0, 0, 0, part[2], part[3])
|
||||
for part in split_moves(0, 0, int(values[0]))]
|
||||
bit = MOUSE_BUTTONS[button]
|
||||
if action == "click":
|
||||
return [(bit, 0, 0, 0, 0), (0, 0, 0, 0, 0)]
|
||||
if action == "drag":
|
||||
if len(values) not in (2, 3, 4):
|
||||
raise RelayError("usage: mouse drag DX DY [WHEEL] [PAN]")
|
||||
nums = [int(v) for v in values] + [0] * (4 - len(values))
|
||||
return [(bit, 0, 0, 0, 0)] + [(bit, *rest) for rest in split_moves(*nums)] \
|
||||
+ [(0, 0, 0, 0, 0)]
|
||||
raise RelayError(f"unknown mouse action {action!r} "
|
||||
"(try move, click, drag, wheel or release)")
|
||||
|
||||
|
||||
def text_frame(delay_ms: int, chunk: str) -> str:
|
||||
"""One `P=<ms>:<hex>` command, byte-identical to proto.h's formatText()."""
|
||||
return f"P={delay_ms}:{chunk.encode('utf-8').hex()}"
|
||||
@@ -416,6 +524,25 @@ class Relay:
|
||||
body = ",".join(f"{u:02x}" for u in usages)
|
||||
self._send(f"Y={mods:02x}:{body}")
|
||||
|
||||
def mouse(self, frames, gap_ms: float = MOUSE_FRAME_MS) -> int:
|
||||
"""Send mouse reports, paced so they arrive like a mouse and not a flood.
|
||||
|
||||
No waiting for acknowledgement between frames -- a movement is not a state the
|
||||
board has to confirm, and the board's own queue is what absorbs a burst. But the
|
||||
gap is real: sent as fast as a socket will take them, a few hundred reports would
|
||||
pile into the board's queue and trip the overflow fail-safe, which releases
|
||||
everything and disarms. An error frame (E10 no mouse interface, E11 boot
|
||||
protocol) raises out of the next read rather than being swallowed.
|
||||
"""
|
||||
sent = 0
|
||||
for buttons, dx, dy, wheel, pan in frames:
|
||||
self._send(mouse_frame(buttons, dx, dy, wheel, pan))
|
||||
sent += 1
|
||||
if gap_ms:
|
||||
time.sleep(gap_ms / 1000.0)
|
||||
self._drain(0.0) # pick up anything the board said about them
|
||||
return sent
|
||||
|
||||
def listen(self, seconds: float) -> None:
|
||||
"""Print heartbeats. Sends keepalives: a silent client is one the board disarms
|
||||
after kNoClientTimeoutMs, and watching must not change what is watched."""
|
||||
@@ -436,6 +563,8 @@ class Relay:
|
||||
print(f" net={s.get('net'):4s} ip={s.get('ip'):15s} "
|
||||
f"usb={s.get('usb')} armed={s.get('armed')} "
|
||||
f"keys={s.get('keys')} chords={s.get('chords')} "
|
||||
f"mframes={s.get('mframes', 0)} "
|
||||
f"mbtn=0x{s.get('mbtn', 0):02x} "
|
||||
f"drop={s.get('drop')} leds=0x{s.get('leds'):02x}")
|
||||
else:
|
||||
print(f" {text}")
|
||||
@@ -478,6 +607,8 @@ def describe(status: dict) -> str:
|
||||
f" armed {'YES -- keystrokes are being forwarded' if status.get('armed') else 'no'}",
|
||||
f" counters keys={status.get('keys')} chords={status.get('chords')} "
|
||||
f"dropped={status.get('drop')}",
|
||||
f" mouse {'can report' if status.get('mouse') else 'CANNOT REPORT'} "
|
||||
f"buttons=0x{status.get('mbtn', 0):02x} reports={status.get('mframes', 0)}",
|
||||
f" text {status.get('typed', 0)} characters typed"
|
||||
+ (f", {status['typing']} still to go" if status.get('typing') else ""),
|
||||
f" leds caps={bool(status.get('leds', 0) & 2)} "
|
||||
@@ -485,6 +616,10 @@ def describe(status: dict) -> str:
|
||||
]
|
||||
if status.get("dflt"):
|
||||
lines.append(" WARNING the board still uses the default passphrase")
|
||||
if status.get("usb") and not status.get("mouse"):
|
||||
lines.append(" NOTE no mouse reports: `mouse on` over serial and reboot, or"
|
||||
"\n the target chose boot protocol (it would read a mouse"
|
||||
"\n report as keystrokes, so the board refuses them)")
|
||||
if status.get("roll"):
|
||||
lines.append(" NOTE more than six keys are down: ErrorRollOver is out")
|
||||
return "\n".join(lines)
|
||||
@@ -534,6 +669,26 @@ def main() -> int:
|
||||
help="leave the bridge armed afterwards")
|
||||
p_paste.add_argument("--yes", action="store_true",
|
||||
help="acknowledge that this types into the target")
|
||||
p_mouse = sub.add_parser(
|
||||
"mouse", help="move the target's pointer, or press one of its buttons")
|
||||
p_mouse.add_argument("action", choices=["move", "click", "drag", "wheel", "release"],
|
||||
help="move: relative dx dy (and optional wheel, pan); "
|
||||
"click: press and release one button; drag: press, move, "
|
||||
"release; wheel: NOTCHES (negative scrolls up); "
|
||||
"release: every button up")
|
||||
p_mouse.add_argument("values", nargs="*", metavar="N",
|
||||
help="numbers for move / drag (px, px [wheel] [pan]) or "
|
||||
"wheel (notches)")
|
||||
p_mouse.add_argument("--button", default="left",
|
||||
help="left, right, middle, back or forward (default left)")
|
||||
p_mouse.add_argument("--gap", type=int, default=MOUSE_FRAME_MS, metavar="MS",
|
||||
help=f"milliseconds between reports (default {MOUSE_FRAME_MS}, "
|
||||
"the rate the controller page uses)")
|
||||
p_mouse.add_argument("--keep-armed", action="store_true",
|
||||
help="stay armed afterwards")
|
||||
p_mouse.add_argument("--yes", action="store_true",
|
||||
help="required for click and drag, which can actuate something")
|
||||
|
||||
p_keys = sub.add_parser(
|
||||
"keys",
|
||||
help="verification session: arm, press these keys, disarm again")
|
||||
@@ -646,6 +801,40 @@ def main() -> int:
|
||||
return 3
|
||||
print("sending Ctrl+Alt+Del ...")
|
||||
relay.chord(0x05, [0x4C])
|
||||
elif args.cmd == "mouse":
|
||||
# A click can actuate whatever is under the pointer, so it is gated like the
|
||||
# other things that have consequences beyond a character appearing. A
|
||||
# movement is not: it moves a cursor, which is reversible by definition.
|
||||
if args.action in ("click", "drag") and not args.yes:
|
||||
print(f"`{args.action}` presses a real mouse button on the machine the "
|
||||
"board is plugged into, wherever its pointer happens to be.\n"
|
||||
"Re-run with --yes if a human is looking at it.", file=sys.stderr)
|
||||
return 4
|
||||
try:
|
||||
frames = mouse_frames(args.action, args.values, args.button)
|
||||
except (RelayError, ValueError) as exc:
|
||||
print(f"{exc}", file=sys.stderr)
|
||||
return 2
|
||||
# Armed inside this one connection, like `keys` and `paste`: the board
|
||||
# disarms the moment a controller drops, so "arm in one invocation and use it
|
||||
# in the next" is not a thing that can work -- and the LED is red for exactly
|
||||
# as long as this session lasts.
|
||||
status = relay.status()
|
||||
if not status.get("mouse", 1):
|
||||
print("this board cannot report a mouse: `mouse on` over serial and "
|
||||
"reboot, or the target chose boot protocol (it would read a "
|
||||
"mouse report as keystrokes, so the board refuses them). Nothing "
|
||||
"was armed and nothing was sent.", file=sys.stderr)
|
||||
return 3
|
||||
relay.arm()
|
||||
print("armed -- LED should be solid red on the board")
|
||||
n = relay.mouse(frames, args.gap)
|
||||
print(f"{n} mouse report(s) sent ({args.action}"
|
||||
+ (f", {args.button}" if args.button != "left" else "") + ")")
|
||||
if not args.keep_armed:
|
||||
relay.disarm()
|
||||
print("disarmed, every key and button released")
|
||||
print(describe(relay.status()))
|
||||
elif args.cmd == "keys":
|
||||
# One deliberate session in one connection: the board disarms the moment
|
||||
# a controller drops (that is the safety property that makes the rest of
|
||||
|
||||
+318
-83
@@ -1,25 +1,35 @@
|
||||
#!/usr/bin/env python3
|
||||
"""watch_target.py -- prove what the target actually received, on the target.
|
||||
|
||||
Run this on the machine the keyrelay board is plugged into. It opens the evdev node
|
||||
of the board's keyboard interface and decodes every event, so verification never
|
||||
depends on what happens to have focus: no terminal, no editor, no guessed timing.
|
||||
Run this on the machine the keyrelay board is plugged into. It opens the evdev nodes of
|
||||
the board's HID interfaces and decodes every event, so verification never depends on what
|
||||
happens to have focus: no terminal, no editor, no guessed timing.
|
||||
|
||||
python3 watch_target.py # 20 s, then a summary
|
||||
python3 watch_target.py # 20 s, then a summary
|
||||
python3 watch_target.py --seconds 60
|
||||
python3 watch_target.py --show-all # include unknown key codes
|
||||
python3 watch_target.py --show-all # include unknown key codes
|
||||
python3 watch_target.py --keyboard-only # ignore the mouse interface
|
||||
|
||||
It reads, it does not write: EV_KEY and EV_LED only, and nothing is injected.
|
||||
The board enumerates as a keyboard and a relative mouse on one HID interface, which the
|
||||
kernel splits into two input devices, so both are watched at once. "The mouse works" means
|
||||
*the target received REL_X*, and a keyboard watcher cannot see that. Movement is coalesced
|
||||
into one line per report -- a captured pointer is 125 reports a second, and a stream
|
||||
nobody can read proves nothing -- while buttons and wheel are printed as they happen. The
|
||||
summary reports distance travelled, not event counts, so it can be compared against a
|
||||
movement you actually made.
|
||||
|
||||
It reads, it does not write: EV_KEY, EV_REL and EV_LED only, and nothing is injected.
|
||||
|
||||
Membership of the `input` group is enough to open /dev/input/event*. The board is
|
||||
found by USB id rather than by a device-name path, so a rename or a different port
|
||||
cannot break it (the default id is the Logitech G815 identity the firmware uses,
|
||||
see src/main.cpp; the Espressif default is also accepted for older builds).
|
||||
cannot break it (the default id is the Logitech G815 identity the firmware uses, see
|
||||
src/main.cpp; the Espressif default is also accepted for older builds).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import glob
|
||||
import pathlib
|
||||
import select
|
||||
import struct
|
||||
import sys
|
||||
@@ -28,7 +38,7 @@ import time
|
||||
EVENT_FORMAT = "llHHi" # struct input_event on 64-bit Linux
|
||||
EVENT_SIZE = struct.calcsize(EVENT_FORMAT)
|
||||
|
||||
EV_SYN, EV_KEY, EV_LED = 0x00, 0x01, 0x11
|
||||
EV_SYN, EV_KEY, EV_REL, EV_ABS, EV_LED = 0x00, 0x01, 0x02, 0x03, 0x11
|
||||
|
||||
IDENTITIES = {("046d", "c33f"), ("303a", "1001")}
|
||||
|
||||
@@ -59,6 +69,27 @@ KEY_NAMES = {
|
||||
189: "F19", 190: "F20", 191: "F21", 192: "F22", 193: "F23", 194: "F24",
|
||||
}
|
||||
|
||||
# Mouse button codes. A separate number space from the keyboard's EV_KEY codes above --
|
||||
# which is exactly why the two interfaces are decoded separately instead of through one
|
||||
# table: BTN_LEFT is 272 and no key code is anywhere near it.
|
||||
BTN_NAMES = {272: "Left", 273: "Right", 274: "Middle", 275: "Back", 276: "Forward",
|
||||
277: "Task", 278: "WheelLeft", 279: "WheelRight"}
|
||||
|
||||
# Input-event axis codes. The first version of this table had the wheel at code 2, guessed
|
||||
# from the order the HID report lays its bytes out -- and REL_WHEEL is 8. It was caught by
|
||||
# nothing on the host and by everything on the target: `mouse wheel -3` scrolled the
|
||||
# machine, the watcher printed nothing, and the summary said no wheel happened. So the
|
||||
# numbers are pinned to <linux/input-event-codes.h> by name and the unknown-code path is
|
||||
# loud, because a verifier that quietly drops a code is worse than one that is wrong.
|
||||
REL_X, REL_Y, REL_Z = 0, 1, 2
|
||||
REL_HWHEEL, REL_DIAL, REL_WHEEL, REL_MISC = 6, 7, 8, 9
|
||||
REL_WHEEL_HI_RES, REL_HWHEEL_HI_RES = 11, 12
|
||||
REL_NAMES = {REL_X: "REL_X", REL_Y: "REL_Y", REL_Z: "REL_Z",
|
||||
REL_HWHEEL: "REL_HWHEEL", REL_DIAL: "REL_DIAL",
|
||||
REL_WHEEL: "REL_WHEEL", REL_MISC: "REL_MISC",
|
||||
REL_WHEEL_HI_RES: "REL_WHEEL_HI_RES",
|
||||
REL_HWHEEL_HI_RES: "REL_HWHEEL_HI_RES"}
|
||||
|
||||
# code -> (unshifted, shifted), for reassembling what was actually typed. Anything
|
||||
# not listed still shows up as a named key press, just not as a character.
|
||||
TYPED = {
|
||||
@@ -106,9 +137,42 @@ def read_sysfs(path: str) -> str:
|
||||
return ""
|
||||
|
||||
|
||||
def keyboard_nodes() -> list[str]:
|
||||
"""evdev nodes of every USB device in IDENTITIES."""
|
||||
nodes: list[str] = []
|
||||
def capability_bits(path: str) -> int:
|
||||
"""The EV_* bitmask a device supports, from sysfs. 0 when it cannot be read."""
|
||||
value = 0
|
||||
for word in read_sysfs(path).split():
|
||||
try:
|
||||
value = (value << 32) | int(word, 16)
|
||||
except ValueError:
|
||||
return 0
|
||||
return value
|
||||
|
||||
|
||||
def classify(ev_bits: int) -> str | None:
|
||||
"""Which of the two devices a node is, from its EV_* capability bits.
|
||||
|
||||
Split out of target_nodes() so the rule itself can be tested: a relative pointing
|
||||
device is the one that advertises EV_REL, and the keyboard is everything else that
|
||||
advertises EV_KEY. Nothing here looks at a device *name*, because the name is what
|
||||
changes when a different driver binds to it.
|
||||
"""
|
||||
if ev_bits & (1 << EV_REL):
|
||||
return "mouse"
|
||||
if ev_bits & (1 << EV_KEY):
|
||||
return "keyboard"
|
||||
return None
|
||||
|
||||
|
||||
def target_nodes(want_mouse: bool = True) -> dict[str, str]:
|
||||
"""evdev nodes of the board's interfaces, keyed "keyboard" / "mouse".
|
||||
|
||||
Classified by what the kernel says the device emits, not by its name or by the order
|
||||
the glob found them in. Watching "the first matching node" would have watched the
|
||||
mouse and reported the keyboard's keystrokes as missing, or the other way round,
|
||||
depending on which interface the kernel probed first -- which is not a fact to build a
|
||||
verification on.
|
||||
"""
|
||||
out: dict[str, str] = {}
|
||||
for device in glob.glob("/sys/bus/usb/devices/*"):
|
||||
if (read_sysfs(f"{device}/idVendor"),
|
||||
read_sysfs(f"{device}/idProduct")) not in IDENTITIES:
|
||||
@@ -116,8 +180,183 @@ def keyboard_nodes() -> list[str]:
|
||||
# Depth is spelled out on purpose: a recursive ** glob over sysfs stalls.
|
||||
found = glob.glob(f"{device}/*/input/input*/event*")
|
||||
found += glob.glob(f"{device}/*/*/input/input*/event*")
|
||||
nodes += ["/dev/input/" + n.rsplit("/", 1)[-1] for n in sorted(found)]
|
||||
return sorted(set(nodes))
|
||||
for path in sorted(found):
|
||||
# .../input/inputN/eventM -> .../input/inputN/capabilities/ev
|
||||
caps = pathlib.Path(path).parent / "capabilities" / "ev"
|
||||
kind = classify(capability_bits(str(caps)))
|
||||
if kind is None or (kind == "mouse" and not want_mouse):
|
||||
continue
|
||||
out.setdefault(kind, "/dev/input/" + path.rsplit("/", 1)[-1])
|
||||
return out
|
||||
|
||||
|
||||
class KeyboardWatch:
|
||||
"""Key presses, chords, LEDs, and the text that came out of them."""
|
||||
|
||||
kind = "keyboard"
|
||||
|
||||
def __init__(self, show_all: bool) -> None:
|
||||
self.show_all = show_all
|
||||
self.held: set[int] = set()
|
||||
self.typed: list[str] = []
|
||||
self.presses = 0
|
||||
self.chords: set[str] = set()
|
||||
self.reported: set[str] = set()
|
||||
self.shift = False
|
||||
|
||||
def event(self, stamp: float, code: int, value: int) -> str | None:
|
||||
if code not in KEY_NAMES and not self.show_all:
|
||||
return None # unknown codes are counted, never dropped:
|
||||
if code not in KEY_NAMES: # they still show up in --show-all totals
|
||||
name = f"key{code}"
|
||||
else:
|
||||
name = KEY_NAMES[code]
|
||||
|
||||
if value == 1: # press
|
||||
self.presses += 1
|
||||
self.held.add(code)
|
||||
if code in SHIFTS:
|
||||
self.shift = True
|
||||
current = frozenset(self.held)
|
||||
for combo, label in CHORDS.items():
|
||||
if combo <= current and label not in self.reported:
|
||||
self.reported.add(label)
|
||||
self.chords.add(label)
|
||||
print(f" t={stamp:6.2f}s >>> {label} <<<")
|
||||
if code in TYPED and not (current & CONTROLS):
|
||||
self.typed.append(TYPED[code][1 if self.shift else 0])
|
||||
elif code in TYPED and code == 28:
|
||||
self.typed.append("<Enter-with-modifiers>")
|
||||
state = "press"
|
||||
elif value == 0: # release
|
||||
self.held.discard(code)
|
||||
if code in SHIFTS and not (self.held & SHIFTS):
|
||||
self.shift = False
|
||||
state = "release"
|
||||
else: # autorepeat
|
||||
state = "repeat"
|
||||
return f"{name:>16} {state}"
|
||||
|
||||
def led(self, stamp: float, code: int, value: int) -> str:
|
||||
return f"LED {LED_NAMES.get(code, f'LED{code}')} {'on' if value else 'off'}"
|
||||
|
||||
def flush(self, stamp: float) -> str | None:
|
||||
"""Keyboards have nothing to coalesce; the interface is here so the loop is."""
|
||||
return None
|
||||
|
||||
def report(self) -> list[str]:
|
||||
lines = [f" key presses {self.presses}"]
|
||||
if self.chords:
|
||||
lines.append(f" chords {', '.join(sorted(self.chords))}")
|
||||
if self.typed:
|
||||
text = "".join(self.typed)
|
||||
preview = text if len(text) <= 240 else text[:240] + "..."
|
||||
lines.append(f" typed {preview!r}")
|
||||
else:
|
||||
lines.append(" typed (nothing plain-text -- modifiers or chords "
|
||||
"only)")
|
||||
if self.held:
|
||||
lines.append(" still held "
|
||||
+ ", ".join(KEY_NAMES.get(c, str(c)) for c in self.held)
|
||||
+ " <-- a stuck key: check the controller released it")
|
||||
return lines
|
||||
|
||||
|
||||
class MouseWatch:
|
||||
"""Relative movement, coalesced per report, plus buttons and wheel."""
|
||||
|
||||
kind = "mouse"
|
||||
|
||||
def __init__(self, show_all: bool) -> None:
|
||||
self.show_all = show_all
|
||||
self.reports = 0 # reports that carried movement
|
||||
self.x = 0 # net travel, the thing you can compare with a ruler
|
||||
self.y = 0
|
||||
self.travel = 0 # total path length in device units
|
||||
self.wheel = 0
|
||||
self.hpan = 0
|
||||
self.held: set[int] = set()
|
||||
self.clicks = 0
|
||||
self.seen_buttons: set[str] = set()
|
||||
self.axes: set[str] = set()
|
||||
self._dx = 0 # the report being accumulated
|
||||
self._dy = 0
|
||||
self._hw = 0
|
||||
|
||||
def event(self, stamp: float, code: int, value: int) -> str | None:
|
||||
if code in REL_NAMES:
|
||||
name = REL_NAMES[code]
|
||||
self.axes.add(name)
|
||||
if code == REL_X:
|
||||
self._dx += value
|
||||
elif code == REL_Y:
|
||||
self._dy += value
|
||||
elif code == REL_WHEEL:
|
||||
self.wheel += value
|
||||
return f"{name:>16} {value:+d}"
|
||||
elif code == REL_HWHEEL:
|
||||
self.hpan += value
|
||||
return f"{name:>16} {value:+d}"
|
||||
# Hi-res wheel events are 1/240 of a detent and repeat the notch the plain
|
||||
# event already reported; counting them would multiply the scroll by 240.
|
||||
return None
|
||||
if code in BTN_NAMES:
|
||||
name = BTN_NAMES[code]
|
||||
self.seen_buttons.add(name)
|
||||
if value == 1:
|
||||
self.held.add(code)
|
||||
self.clicks += 1
|
||||
return f"BTN {name:>12} press"
|
||||
if value == 0:
|
||||
self.held.discard(code)
|
||||
return f"BTN {name:>12} release"
|
||||
return None
|
||||
# An axis this tool has no name for is still counted and still shown -- silently
|
||||
# dropping it is how a working wheel looked like a dead one.
|
||||
self.axes.add(f"rel{code}")
|
||||
if not self.show_all:
|
||||
return None
|
||||
return f"{'rel' + str(code):>16} {value:+d}"
|
||||
|
||||
def flush(self, stamp: float) -> str | None:
|
||||
"""One line per report that moved: 125 reports a second is a blur otherwise."""
|
||||
if not (self._dx or self._dy):
|
||||
return None
|
||||
self.reports += 1
|
||||
self.x += self._dx
|
||||
self.y += self._dy
|
||||
self.travel += abs(self._dx) + abs(self._dy)
|
||||
line = f"{'movement':>16} dx {self._dx:+5d} dy {self._dy:+5d}"
|
||||
self._dx = self._dy = 0
|
||||
return line
|
||||
|
||||
def led(self, stamp: float, code: int, value: int) -> str | None:
|
||||
return None # pointing devices have no LEDs
|
||||
|
||||
def report(self) -> list[str]:
|
||||
lines = [f" reports moved {self.reports}"]
|
||||
if self.reports:
|
||||
lines.append(f" net position dx {self.x:+d} dy {self.y:+d} "
|
||||
f"(path length {self.travel})")
|
||||
else:
|
||||
lines.append(" net position (no movement reached the target)")
|
||||
if self.axes:
|
||||
lines.append(f" axes seen {', '.join(sorted(self.axes))}")
|
||||
lines.append(f" wheel {self.wheel:+d} notches, {self.hpan:+d} horizontal")
|
||||
if self.seen_buttons:
|
||||
lines.append(f" buttons {', '.join(sorted(self.seen_buttons))} "
|
||||
f"({self.clicks} press(es))")
|
||||
else:
|
||||
lines.append(" buttons (none -- try `relayctl mouse click --yes`)")
|
||||
if self.held:
|
||||
lines.append(" still held "
|
||||
+ ", ".join(BTN_NAMES.get(c, str(c)) for c in self.held)
|
||||
+ " <-- a stuck button: the controller let go of it and the "
|
||||
"board did not")
|
||||
return lines
|
||||
|
||||
|
||||
WATCHERS = {"keyboard": KeyboardWatch, "mouse": MouseWatch}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
@@ -127,94 +366,90 @@ def main() -> int:
|
||||
ap.add_argument("--show-all", action="store_true",
|
||||
help="also print key codes this tool cannot name (unknown codes "
|
||||
"are counted, never dropped silently: they show in totals)")
|
||||
ap.add_argument("--keyboard-only", action="store_true",
|
||||
help="ignore the mouse interface -- for a board flashed with "
|
||||
"`mouse off`, or to read the keyboard without the pointer "
|
||||
"filling the screen")
|
||||
args = ap.parse_args()
|
||||
|
||||
nodes = keyboard_nodes()
|
||||
nodes = target_nodes(want_mouse=not args.keyboard_only)
|
||||
if not nodes:
|
||||
ids = " or ".join(":".join(i) for i in IDENTITIES)
|
||||
print(f"no keyboard for {ids} under /sys/bus/usb -- is the board plugged in "
|
||||
f"and enumerated here?", file=sys.stderr)
|
||||
return 1
|
||||
if len(nodes) > 1:
|
||||
print(f"note: several matching nodes {nodes}; watching the first")
|
||||
missing = [k for k in ("keyboard", "mouse")
|
||||
if k not in nodes and not (k == "mouse" and args.keyboard_only)]
|
||||
for kind in missing:
|
||||
print(f"note: no {kind} interface found -- the board enumerated without it "
|
||||
f"(`mouse off` over serial does that deliberately)", file=sys.stderr)
|
||||
|
||||
path = nodes[0]
|
||||
print(f"watching {path} for {args.seconds:.0f} s -- press keys on the controller")
|
||||
handles = {}
|
||||
for kind, path in sorted(nodes.items()):
|
||||
try:
|
||||
handles[kind] = open(path, "rb", buffering=0)
|
||||
except OSError as exc:
|
||||
print(f"cannot open {path} for the {kind}: {exc} -- are you in the `input` "
|
||||
f"group?", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
watchers = {kind: WATCHERS[kind](args.show_all) for kind in handles}
|
||||
print(f"watching {', '.join(f'{k}={v}' for k, v in sorted(nodes.items()))} "
|
||||
f"for {args.seconds:.0f} s")
|
||||
print("press keys, or capture the pointer in the controller page and move it")
|
||||
print(f"{'t':>9} event")
|
||||
|
||||
held: set[int] = set()
|
||||
typed: list[str] = []
|
||||
presses = 0
|
||||
chords_seen: set[str] = set()
|
||||
reported_chords: set[str] = set()
|
||||
shift = False
|
||||
start: float | None = None
|
||||
errors = 0
|
||||
start: float | None = None
|
||||
|
||||
with open(path, "rb", buffering=0) as handle:
|
||||
try:
|
||||
deadline = time.time() + args.seconds
|
||||
while time.time() < deadline:
|
||||
remaining = max(0.0, deadline - time.time())
|
||||
if not select.select([handle], [], [], remaining)[0]:
|
||||
ready, _, _ = select.select(list(handles.values()), [], [], remaining)
|
||||
if not ready:
|
||||
continue
|
||||
raw = handle.read(EVENT_SIZE)
|
||||
if raw is None or len(raw) != EVENT_SIZE:
|
||||
errors += 1
|
||||
continue
|
||||
sec, usec, event_type, code, value = struct.unpack(EVENT_FORMAT, raw)
|
||||
if start is None:
|
||||
start = sec + usec / 1e6
|
||||
stamp = (sec + usec / 1e6) - start
|
||||
for kind, handle in handles.items():
|
||||
if handle not in ready:
|
||||
continue
|
||||
raw = handle.read(EVENT_SIZE)
|
||||
if raw is None or len(raw) != EVENT_SIZE:
|
||||
errors += 1
|
||||
continue
|
||||
sec, usec, event_type, code, value = struct.unpack(EVENT_FORMAT, raw)
|
||||
if start is None:
|
||||
start = sec + usec / 1e6
|
||||
stamp = (sec + usec / 1e6) - start
|
||||
watcher = watchers[kind]
|
||||
|
||||
if event_type == EV_LED:
|
||||
name = LED_NAMES.get(code, f"LED{code}")
|
||||
print(f" t={stamp:6.2f}s LED {name} {'on' if value else 'off'}")
|
||||
continue
|
||||
if event_type != EV_KEY:
|
||||
continue
|
||||
|
||||
name = KEY_NAMES.get(code, f"key{code}")
|
||||
if name.startswith("key") and not args.show_all:
|
||||
continue
|
||||
|
||||
if value == 1: # press
|
||||
presses += 1
|
||||
held.add(code)
|
||||
if code in SHIFTS:
|
||||
shift = True
|
||||
current = frozenset(held)
|
||||
for combo, label in CHORDS.items():
|
||||
if combo <= current and label not in reported_chords:
|
||||
reported_chords.add(label)
|
||||
chords_seen.add(label)
|
||||
print(f" t={stamp:6.2f}s >>> {label} <<<")
|
||||
if code in TYPED and not (current & CONTROLS):
|
||||
typed.append(TYPED[code][1 if shift else 0])
|
||||
elif code in TYPED and code == 28:
|
||||
typed.append("<Enter-with-modifiers>")
|
||||
state = "press"
|
||||
elif value == 0: # release
|
||||
held.discard(code)
|
||||
if code in SHIFTS and not (held & SHIFTS):
|
||||
shift = False
|
||||
state = "release"
|
||||
else: # autorepeat
|
||||
state = "repeat"
|
||||
print(f" t={stamp:6.2f}s {name:>16} {state}", flush=True)
|
||||
if event_type == EV_SYN:
|
||||
line = watcher.flush(stamp)
|
||||
if line:
|
||||
print(f" t={stamp:6.2f}s {line}", flush=True)
|
||||
continue
|
||||
if event_type == EV_LED:
|
||||
line = watcher.led(stamp, code, value)
|
||||
if line:
|
||||
print(f" t={stamp:6.2f}s {line}", flush=True)
|
||||
continue
|
||||
if event_type == EV_KEY:
|
||||
line = watcher.event(stamp, code, value)
|
||||
elif event_type == EV_REL:
|
||||
line = watcher.event(stamp, code, value)
|
||||
else:
|
||||
continue
|
||||
if line:
|
||||
print(f" t={stamp:6.2f}s {line}", flush=True)
|
||||
finally:
|
||||
for handle in handles.values():
|
||||
handle.close()
|
||||
|
||||
print()
|
||||
print(f" key presses {presses}")
|
||||
if chords_seen:
|
||||
print(f" chords {', '.join(sorted(chords_seen))}")
|
||||
if typed:
|
||||
text = "".join(typed)
|
||||
preview = text if len(text) <= 240 else text[:240] + "..."
|
||||
print(f" typed {preview!r}")
|
||||
else:
|
||||
print(" typed (nothing plain-text -- modifiers or chords only)")
|
||||
if held:
|
||||
print(f" still held {', '.join(KEY_NAMES.get(c, str(c)) for c in held)}"
|
||||
" <-- a stuck key: check the controller released it")
|
||||
for kind in ("keyboard", "mouse"):
|
||||
if kind in watchers:
|
||||
print(f" --- {kind} ---")
|
||||
for line in watchers[kind].report():
|
||||
print(line)
|
||||
if errors:
|
||||
print(f" short reads {errors}")
|
||||
return 0
|
||||
|
||||
Reference in New Issue
Block a user