Remove the unused keyboard wedge module
keyboard.py was never wired in: no module imported it, the CLI had no --type flag, and neither the docs nor install.sh referenced it, so the virtual-keyboard path could never run. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -1,339 +0,0 @@
|
||||
"""Type a card id into whatever window has focus, like a keyboard-wedge reader.
|
||||
|
||||
A commercial USB RFID reader presents itself to the machine as a keyboard: the
|
||||
card code appears in the focused window as if it had been typed. This does the
|
||||
same in software, by creating a virtual keyboard through the kernel's
|
||||
``/dev/uinput`` interface.
|
||||
|
||||
Two consequences of doing it that way, both deliberate:
|
||||
|
||||
* **No external tool and no extra dependency.** ``python-evdev`` would be the
|
||||
usual choice, but it is a C extension, so its wheels are tied to one
|
||||
interpreter and one CPU architecture. This project ships pure-Python wheels
|
||||
precisely so the same files work on 32- and 64-bit ARM; pulling in a compiled
|
||||
dependency would break that. ``ctypes`` and ``struct`` are enough.
|
||||
* **It works from the reader's system service.** uinput registers the device
|
||||
with the kernel, not with a login session, so the desktop picks it up exactly
|
||||
as it would a real keyboard. No ``DISPLAY``, no session bus, no
|
||||
``systemctl --user`` unit is involved - unlike X11-only tools such as
|
||||
``xdotool``, which also fail on Wayland.
|
||||
|
||||
The known limit, which a real wedge reader shares: the text goes to whatever
|
||||
holds focus, so a tap while a password prompt is focused types the card id
|
||||
there. That is why the feature is **opt-in** (``--type``) rather than default.
|
||||
|
||||
Like the LED, this never blocks the reader thread, never crashes the process,
|
||||
and never pretends to work when it cannot: whether a virtual keyboard was
|
||||
actually created is reported at startup.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ctypes
|
||||
import fcntl
|
||||
import os
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
#: Where the kernel exposes the uinput control device.
|
||||
UINPUT_PATH = "/dev/uinput"
|
||||
|
||||
#: Suppress repeated identical text, e.g. the same card held on the antenna.
|
||||
DEFAULT_KEY_DELAY = 0.002
|
||||
|
||||
#: Output formats for the card id.
|
||||
DECIMAL = "decimal"
|
||||
HEX = "hex"
|
||||
FORMATS = (DECIMAL, HEX)
|
||||
|
||||
#: What to send after the id, so the target field is submitted.
|
||||
ENTER = "enter"
|
||||
TAB = "tab"
|
||||
NONE = "none"
|
||||
SUFFIXES = (ENTER, TAB, NONE)
|
||||
|
||||
# -- kernel constants (linux/input-event-codes.h and linux/uinput.h) ---------
|
||||
|
||||
EV_SYN = 0x00
|
||||
EV_KEY = 0x01
|
||||
SYN_REPORT = 0x00
|
||||
|
||||
_KEY_PRESS = 1
|
||||
_KEY_RELEASE = 0
|
||||
|
||||
_UI_DEV_CREATE = 0x5501
|
||||
_UI_DEV_DESTROY = 0x5502
|
||||
_UI_SET_EVBIT = 0x40045564
|
||||
_UI_SET_KEYBIT = 0x40045565
|
||||
|
||||
_UINPUT_MAX_NAME_SIZE = 80
|
||||
|
||||
#: Character -> key code. Digits cover the default output; a-f cover ``--hex``
|
||||
#: (lower case, since upper case would need a shift modifier for no real gain).
|
||||
_KEYCODES: Dict[str, int] = {
|
||||
"1": 2, "2": 3, "3": 4, "4": 5, "5": 6,
|
||||
"6": 7, "7": 8, "8": 9, "9": 10, "0": 11,
|
||||
"a": 30, "b": 48, "c": 46, "d": 32, "e": 18, "f": 33,
|
||||
"x": 45,
|
||||
}
|
||||
|
||||
_SPECIAL: Dict[str, int] = {
|
||||
ENTER: 28,
|
||||
TAB: 15,
|
||||
}
|
||||
|
||||
|
||||
class _InputEvent(ctypes.Structure):
|
||||
"""``struct input_event``.
|
||||
|
||||
``tv_sec``/``tv_usec`` are ``c_long`` so the layout is right on both 32- and
|
||||
64-bit kernels - the struct is 16 bytes on one and 24 on the other, and
|
||||
hard-coding either would corrupt the event.
|
||||
"""
|
||||
|
||||
_fields_ = [
|
||||
("tv_sec", ctypes.c_long),
|
||||
("tv_usec", ctypes.c_long),
|
||||
("type", ctypes.c_uint16),
|
||||
("code", ctypes.c_uint16),
|
||||
("value", ctypes.c_int32),
|
||||
]
|
||||
|
||||
|
||||
class _InputId(ctypes.Structure):
|
||||
_fields_ = [
|
||||
("bustype", ctypes.c_uint16),
|
||||
("vendor", ctypes.c_uint16),
|
||||
("product", ctypes.c_uint16),
|
||||
("version", ctypes.c_uint16),
|
||||
]
|
||||
|
||||
|
||||
class _UInputUserDev(ctypes.Structure):
|
||||
"""``struct uinput_user_dev``, used to name and describe the device."""
|
||||
|
||||
_fields_ = [
|
||||
("name", ctypes.c_char * _UINPUT_MAX_NAME_SIZE),
|
||||
("id", _InputId),
|
||||
("ff_effects_max", ctypes.c_uint32),
|
||||
("absmax", ctypes.c_int32 * 64),
|
||||
("absmin", ctypes.c_int32 * 64),
|
||||
("absfuzz", ctypes.c_int32 * 64),
|
||||
("absflat", ctypes.c_int32 * 64),
|
||||
]
|
||||
|
||||
|
||||
def _as_bytes(struct: ctypes.Structure) -> bytes:
|
||||
return ctypes.string_at(ctypes.byref(struct), ctypes.sizeof(struct))
|
||||
|
||||
|
||||
@dataclass
|
||||
class WedgeStatus:
|
||||
"""What the wedge subsystem is actually doing."""
|
||||
|
||||
enabled: bool
|
||||
backend: str # "uinput" | "null"
|
||||
format: str = DECIMAL
|
||||
suffix: str = ENTER
|
||||
reason: str = ""
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"enabled": self.enabled,
|
||||
"backend": self.backend,
|
||||
"format": self.format,
|
||||
"suffix": self.suffix,
|
||||
"reason": self.reason,
|
||||
}
|
||||
|
||||
|
||||
class NullWedge:
|
||||
"""A wedge that does nothing, used when uinput is unavailable."""
|
||||
|
||||
backend = "null"
|
||||
|
||||
def __init__(self, reason: str = ""):
|
||||
self.reason = reason
|
||||
|
||||
def type_text(self, text: str) -> None:
|
||||
pass
|
||||
|
||||
def close(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
class UInputWedge:
|
||||
"""A virtual keyboard created through ``/dev/uinput``."""
|
||||
|
||||
backend = "uinput"
|
||||
|
||||
def __init__(self, name: str = "rdm-card-reader", key_delay: float = DEFAULT_KEY_DELAY):
|
||||
self.key_delay = key_delay
|
||||
self._fd = os.open(UINPUT_PATH, os.O_WRONLY | os.O_NONBLOCK)
|
||||
|
||||
# Declare the only kinds of event this device will ever send.
|
||||
fcntl.ioctl(self._fd, _UI_SET_EVBIT, EV_KEY)
|
||||
for code in sorted(set(_KEYCODES.values()) | set(_SPECIAL.values())):
|
||||
fcntl.ioctl(self._fd, _UI_SET_KEYBIT, code)
|
||||
|
||||
dev = _UInputUserDev()
|
||||
dev.name = name.encode("utf-8")[:_UINPUT_MAX_NAME_SIZE - 1]
|
||||
# BUS_USB keeps the device looking like an ordinary keyboard to udev.
|
||||
dev.id = _InputId(bustype=0x03, vendor=0x1, product=0x1, version=1)
|
||||
os.write(self._fd, _as_bytes(dev))
|
||||
|
||||
fcntl.ioctl(self._fd, _UI_DEV_CREATE)
|
||||
# udev needs a moment to notice a brand-new input device; typing before
|
||||
# it does means the first characters go nowhere.
|
||||
time.sleep(0.2)
|
||||
|
||||
def _send(self, event_type: int, code: int, value: int) -> None:
|
||||
os.write(self._fd, _as_bytes(_InputEvent(0, 0, event_type, code, value)))
|
||||
|
||||
def _tap(self, code: int) -> None:
|
||||
self._send(EV_KEY, code, _KEY_PRESS)
|
||||
self._send(EV_SYN, SYN_REPORT, 0)
|
||||
self._send(EV_KEY, code, _KEY_RELEASE)
|
||||
self._send(EV_SYN, SYN_REPORT, 0)
|
||||
if self.key_delay:
|
||||
time.sleep(self.key_delay)
|
||||
|
||||
def type_text(self, text: str) -> None:
|
||||
for character in text:
|
||||
code = _KEYCODES.get(character.lower())
|
||||
if code is not None:
|
||||
self._tap(code)
|
||||
|
||||
def press(self, name: str) -> None:
|
||||
code = _SPECIAL.get(name)
|
||||
if code is not None:
|
||||
self._tap(code)
|
||||
|
||||
def close(self) -> None:
|
||||
try:
|
||||
fcntl.ioctl(self._fd, _UI_DEV_DESTROY)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
os.close(self._fd)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def format_card(event: Any, fmt: str = DECIMAL) -> str:
|
||||
"""Render a card event as the text a wedge reader would have typed."""
|
||||
if fmt == HEX:
|
||||
return event.hex_value
|
||||
return str(event.value)
|
||||
|
||||
|
||||
class KeyboardWedge:
|
||||
"""Non-blocking wedge controller driven by card events.
|
||||
|
||||
Text is queued and typed by a worker thread, so a card read never waits on
|
||||
the desktop - the same arrangement the LED uses, and for the same reason.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
enabled: bool = True,
|
||||
fmt: str = DECIMAL,
|
||||
suffix: str = ENTER,
|
||||
key_delay: float = DEFAULT_KEY_DELAY,
|
||||
logger: Optional[Callable[[str], None]] = None,
|
||||
):
|
||||
self._log = logger or (lambda message: None)
|
||||
self._fmt = fmt if fmt in FORMATS else DECIMAL
|
||||
self._suffix = suffix if suffix in SUFFIXES else ENTER
|
||||
self._queue: "queue.Queue[Optional[str]]" = queue.Queue()
|
||||
self._thread: Optional[threading.Thread] = None
|
||||
|
||||
if not enabled:
|
||||
self.wedge: Any = NullWedge("typing disabled (--no-type)")
|
||||
self.status = WedgeStatus(False, "null", self._fmt, self._suffix,
|
||||
"disabled")
|
||||
return
|
||||
|
||||
try:
|
||||
self.wedge = UInputWedge(key_delay=key_delay)
|
||||
self.status = WedgeStatus(True, "uinput", self._fmt, self._suffix,
|
||||
"ready")
|
||||
self._log("Keyboard wedge ready (uinput, {} text, {} suffix)".format(
|
||||
self._fmt, self._suffix))
|
||||
except PermissionError as exc:
|
||||
reason = (
|
||||
"cannot write {}: {} - add the user to the 'input' group "
|
||||
"and re-run the installer".format(UINPUT_PATH, exc)
|
||||
)
|
||||
self.wedge = NullWedge(reason)
|
||||
self.status = WedgeStatus(False, "null", self._fmt, self._suffix,
|
||||
reason)
|
||||
self._log("Keyboard wedge unavailable: {}".format(reason))
|
||||
except Exception as exc:
|
||||
reason = "no usable uinput: {}".format(exc)
|
||||
self.wedge = NullWedge(reason)
|
||||
self.status = WedgeStatus(False, "null", self._fmt, self._suffix,
|
||||
reason)
|
||||
self._log("Keyboard wedge unavailable: {}".format(reason))
|
||||
|
||||
self._thread = threading.Thread(target=self._worker, daemon=True,
|
||||
name="wedge-worker")
|
||||
self._thread.start()
|
||||
|
||||
# -- event hooks ------------------------------------------------------
|
||||
|
||||
def insert(self, event: Any) -> None:
|
||||
"""A card was read: queue its id, then the configured suffix."""
|
||||
self._send(format_card(event, self._fmt))
|
||||
|
||||
@property
|
||||
def suffix(self) -> str:
|
||||
return self._suffix
|
||||
|
||||
# -- control ----------------------------------------------------------
|
||||
|
||||
def _send(self, text: str) -> None:
|
||||
if not self.status.enabled:
|
||||
return
|
||||
try:
|
||||
self._queue.put_nowait(text)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _worker(self) -> None:
|
||||
while True:
|
||||
text = self._queue.get()
|
||||
if text is None:
|
||||
return
|
||||
try:
|
||||
self.wedge.type_text(text)
|
||||
if self._suffix != NONE:
|
||||
self.wedge.press(self._suffix)
|
||||
except Exception as exc:
|
||||
# A failed keystroke must never take the reader down.
|
||||
self._log("Keyboard wedge failed: {}".format(exc))
|
||||
finally:
|
||||
self._queue.task_done()
|
||||
|
||||
def close(self) -> None:
|
||||
if self._thread is not None:
|
||||
try:
|
||||
self._queue.put_nowait(None)
|
||||
except Exception:
|
||||
pass
|
||||
self.wedge.close()
|
||||
|
||||
|
||||
def create(
|
||||
enabled: bool = True,
|
||||
fmt: str = DECIMAL,
|
||||
suffix: str = ENTER,
|
||||
key_delay: float = DEFAULT_KEY_DELAY,
|
||||
logger: Optional[Callable[[str], None]] = None,
|
||||
) -> KeyboardWedge:
|
||||
"""Build the wedge controller, falling back to a no-op when uinput fails."""
|
||||
return KeyboardWedge(enabled=enabled, fmt=fmt, suffix=suffix,
|
||||
key_delay=key_delay, logger=logger)
|
||||
Reference in New Issue
Block a user