ske087/rdm-card-reader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add keyboard-wedge typing and LED event support

Browse Source 
- keyboard.py: emulate a commercial USB RFID reader by typing card ids
  into the focused window via a /dev/uinput virtual keyboard
- led.py / bridge.py: drive the status LED from card events
- cli.py, card.py: wire up the new options and event flow
- deploy: harden install.sh and the systemd unit
- README: document keyboard mode, LED support, and deployment

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
scheianu
2026-10-05 21:14:58 +03:00
parent 48e800adc1
commit c0a354b98e
9 changed files with 973 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/rdm_card_reader/__init__.py
+7 -1
View File
@@ -30,7 +30,12 @@ from .ports import (
discover,
resolve,
)
from .led import DEFAULT_LED_PIN, StatusLED, create as create_led
from .led import (
DEFAULT_LED_MODE,
DEFAULT_LED_PIN,
StatusLED,
create as create_led,
)
from .reader import available, list_ports, open_reader, reload_library
__version__ = "0.4.0"
@@ -41,6 +46,7 @@ __all__ = [
"CardData",
"CardEvent",
"AmbiguousPort",
"DEFAULT_LED_MODE",
"DEFAULT_LED_PIN",
"EventHub",
"PortError",
src/rdm_card_reader/bridge.py
+45
View File
@@ -75,6 +75,12 @@ DEFAULT_HISTORY = 50
#: replaced.
DEFAULT_TAP_TTL = 10.0
#: How long a *claimed* tap stays readable afterwards. The page that claimed it
#: needs a moment to confirm which card logged in, but an id that stays
#: readable forever is indistinguishable from a tap still waiting to be spent -
#: so it is cleared once this window passes. Set to 0 to keep it indefinitely.
DEFAULT_CLAIMED_TTL = 30.0
class EventHub:
"""Tracks reader state and fans card events out to waiting subscribers.
@@ -96,6 +102,7 @@ class EventHub:
history: int = DEFAULT_HISTORY,
mode: str = MODE_LOGIN,
tap_ttl: float = DEFAULT_TAP_TTL,
claimed_ttl: float = DEFAULT_CLAIMED_TTL,
):
self._lock = threading.Lock()
self._subscribers: List["deque"] = []
@@ -107,6 +114,7 @@ class EventHub:
self._counts = {EVENT_INSERTED: 0, EVENT_REMOVED: 0, "invalid_card": 0}
self._port: Optional[str] = None
self._tap_ttl = float(tap_ttl if tap_ttl else 0.0)
self._claimed_ttl = float(claimed_ttl if claimed_ttl else 0.0)
# The latched tap. It deliberately outlives the card leaving the
# reader, so a page that loads after the tap can still claim it.
@@ -177,8 +185,16 @@ class EventHub:
``present`` is whether the card is physically on the reader right now;
``claimable`` is whether this tap can still be spent on a login.
A claimed tap is reported for ``claimed_ttl`` seconds after it was
spent, so the page that claimed it can confirm which card logged in. An
expired unclaimed tap is cleared too: keeping it would leave an id that
can never be claimed again, which is indistinguishable from a tap that
is still waiting to be used.
"""
with self._lock:
self._drop_dead_tap()
if self._tap is None:
return {
"present": False,
@@ -200,6 +216,32 @@ class EventHub:
"card": self._tap.to_dict(),
}
def _drop_dead_tap(self) -> None:
"""Clear the latched tap once it can no longer be acted on.
Called with the lock held. Two independent deadlines: an unclaimed tap
is dead after ``tap_ttl``, a claimed one after ``claimed_ttl``. A tap of
zero TTL means "keep until replaced", which the claimed case uses to
keep the id around indefinitely.
Crucially this drops only the *tap*, never the presence flag: whether a
card is on the reader is a physical fact learnt from insert/remove
events, and a tap ageing out says nothing about it. Clearing presence
here reported "no card" while a card was sitting on the antenna.
"""
if self._tap is None:
return
now = time.time()
if self._consumed_at is not None:
if self._claimed_ttl and now - self._consumed_at > self._claimed_ttl:
self._tap = None
self._tap_at = None
self._consumed_at = None
return
if self._tap_ttl and now - (self._tap_at or 0.0) > self._tap_ttl:
self._tap = None
self._tap_at = None
def consume_card(self):
"""Atomically claim the latched tap for one authentication.
@@ -207,6 +249,7 @@ class EventHub:
this call succeeds, so two concurrent requests cannot both win.
"""
with self._lock:
self._drop_dead_tap()
if self._tap is None:
return None, "no card read yet", 409
@@ -245,6 +288,7 @@ class EventHub:
def health(self) -> Dict[str, Any]:
with self._lock:
self._drop_dead_tap()
last_age = (
round(time.time() - self._last_at, 1) if self._last_at else None
)
@@ -260,6 +304,7 @@ class EventHub:
"card_present": self._on_reader,
"card_consumed": self._consumed_at is not None,
"tap_ttl_seconds": self._tap_ttl or None,
"claimed_ttl_seconds": self._claimed_ttl or None,
}
def last(self) -> Optional[Dict[str, Any]]:
src/rdm_card_reader/card.py
+13 -2
View File
@@ -18,10 +18,13 @@ except ImportError: # pragma: no cover - fallback so the model stays importable
CardData = namedtuple("CardData", ["value", "checksum", "type", "is_valid"])
#: Serial devices tried in order. ``/dev/ttyS0`` first matches the Raspberry Pi
#: default, the others cover USB / GPIO UART dongles.
#: default, the others cover USB / GPIO UART dongles. ``/dev/serial0`` is the
#: canonical Pi alias for the on-board UART - it resolves to ttyAMA0 or ttyS0
#: depending on the model, and is the device the docs tell you to use.
CANDIDATE_PORTS = (
"/dev/ttyS0",
"/dev/ttyAMA0",
"/dev/serial0",
"/dev/ttyUSB0",
"/dev/ttyACM0",
)
@@ -32,7 +35,15 @@ CONFIG_CARD_VALUE = 12886709
#: Seconds without a read before a card is considered removed. Set to ``None``
#: to disable removal detection entirely.
DEFAULT_HEARTBEAT_INTERVAL = 0.5
#:
#: This is also the presence window: the card counts as "present" until this
#: long after the last frame. RDM6300 modules typically report a card once per
#: placement rather than re-reading continuously, so a short value makes
#: ``present`` flicker - and the status LED with it - while the card sits still
#: on the antenna. Longer holds presence (and the light) on longer but makes
#: lifting the card take longer to notice; raise it if a card held in place is
#: reported as removed, lower it if removal feels sluggish.
DEFAULT_HEARTBEAT_INTERVAL = 2.0
BAR = "-" * 58
src/rdm_card_reader/cli.py
+17 -6
View File
@@ -10,6 +10,7 @@ from typing import Optional, Sequence
from . import __version__
from .bootstrap import LED_PACKAGES, ensure_dependencies
from .bridge import (
DEFAULT_CLAIMED_TTL,
DEFAULT_HOST,
DEFAULT_PORT,
DEFAULT_TAP_TTL,
@@ -27,7 +28,12 @@ from .card import (
format_event,
)
from .demo import run_demo
from .led import DEFAULT_LED_PIN, MODES as LED_MODES, create as create_led
from .led import (
DEFAULT_LED_MODE,
DEFAULT_LED_PIN,
MODES as LED_MODES,
create as create_led,
)
from .ports import AmbiguousPort, PortNotFound, discover, resolve
from .reader import available, explain_error, list_ports, open_reader, reload_library
@@ -169,7 +175,8 @@ def _run_bridge(args) -> int:
device = _resolve_port(args)
hub = EventHub(mode=args.mode, tap_ttl=args.tap_ttl)
hub = EventHub(mode=args.mode, tap_ttl=args.tap_ttl,
claimed_ttl=args.claimed_ttl)
hub.set_port(device)
led = _build_led(args)
@@ -285,6 +292,10 @@ def build_parser() -> argparse.ArgumentParser:
bridge.add_argument("--tap-ttl", type=float, default=DEFAULT_TAP_TTL,
help="seconds an unclaimed tap stays claimable; 0 keeps it "
"claimable while the card is present (default: %(default)s)")
bridge.add_argument("--claimed-ttl", type=float, default=DEFAULT_CLAIMED_TTL,
help="seconds a claimed card stays readable so the claiming "
"page can confirm which card logged in; 0 keeps it "
"indefinitely (default: %(default)s)")
bridge.add_argument("--allow-origin", action="append", metavar="ORIGIN",
help="permit this browser origin to call the API; repeatable")
bridge.add_argument("--token",
@@ -301,10 +312,10 @@ def build_parser() -> argparse.ArgumentParser:
led.add_argument("--led-pin", type=int, default=DEFAULT_LED_PIN,
help="GPIO pin of the status LED (default: %(default)s, "
"matching the workstation app)")
led.add_argument("--led-mode", choices=list(LED_MODES), default=LED_MODES[0],
help="'error' (default): lit whenever the reader is running, "
"blinking means a bad read; 'presence': lit only while "
"a card is on the reader")
led.add_argument("--led-mode", choices=list(LED_MODES), default=DEFAULT_LED_MODE,
help="'presence' (default): lit only while a card is on the "
"reader; 'error': lit whenever the reader is running, "
"blinking means a bad read")
led.add_argument("--led-no-error-blink", action="store_true",
help="do not blink the LED on a bad read")
return parser
src/rdm_card_reader/keyboard.py
+339
View File
@@ -0,0 +1,339 @@
"""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)
src/rdm_card_reader/led.py
+13 -9
View File
@@ -34,15 +34,20 @@ OFF = "off"
ON = "on"
BLINK = "blink"
#: Lit only while a card is on the reader, and off otherwise. This is a card
#: indicator, for when the light should mean "a card is here". The default: a
#: light that tracks the card is unambiguous, whereas a light that is always on
#: says nothing about the card.
PRESENCE = "presence"
#: Lit the whole time the app runs, and blinks on a bad read. This is a health
#: indicator: light on means the reader is alive and ready for a card.
ERROR = "error"
#: Lit only while a card is on the reader, and off otherwise. This is a card
#: indicator, for when the light should mean "a card is here".
PRESENCE = "presence"
MODES = (PRESENCE, ERROR)
MODES = (ERROR, PRESENCE)
#: What the light means unless told otherwise.
DEFAULT_LED_MODE = PRESENCE
#: Kept for the occasional caller that only cares about the concrete commands.
_VALID_MODES = (OFF, ON, BLINK)
@@ -59,8 +64,7 @@ class LEDStatus:
state: bool = False
reason: str = ""
def to_dict(self) -> dict:
return {
def to_dict(self) -> dict: return {
"enabled": self.enabled,
"pin": self.pin,
"backend": self.backend,
@@ -158,13 +162,13 @@ class StatusLED:
self,
pin: int = DEFAULT_LED_PIN,
enabled: bool = True,
mode: str = ERROR,
mode: str = DEFAULT_LED_MODE,
active_high: bool = True,
error_blink: bool = True,
logger: Optional[Callable[[str], None]] = None,
):
self._log = logger or (lambda message: None)
self._mode = mode if mode in MODES else ERROR
self._mode = mode if mode in MODES else DEFAULT_LED_MODE
self._error_blink = error_blink
self._enabled = enabled
self._blinker: Optional[_Blinker] = None
@@ -305,7 +309,7 @@ class StatusLED:
def create(
enabled: bool = True,
pin: int = DEFAULT_LED_PIN,
mode: str = ERROR,
mode: str = DEFAULT_LED_MODE,
error_blink: bool = True,
logger: Optional[Callable[[str], None]] = None,
) -> StatusLED: