diff --git a/README.md b/README.md index 0b95d65..fe762ad 100644 --- a/README.md +++ b/README.md @@ -352,6 +352,69 @@ Exit codes: `0` ok, `1` no port, `2` library missing, `3` port ambiguous. --- +## Status LED + +An LED driven by card events, on **GPIO 23** by default — the same pin the +workstation app used. + +```bash +rdm-card-reader # lit while running, blinks on a bad read +rdm-card-reader --led-pin 17 # a different pin +rdm-card-reader --led-mode presence # lit only while a card is on the reader +rdm-card-reader --no-led # no LED at all +``` + +| Flag | Default | Meaning | +| --- | --- | --- | +| `--no-led` | off | Do not drive a LED | +| `--led-pin` | `23` | GPIO pin (matches the workstation app) | +| `--led-mode` | `error` | See the two readings below | +| `--led-no-error-blink` | off | Do not blink on a bad read | + +Two modes, because the light can mean two different things: + +| Mode | Behaviour | Use when | +| --- | --- | --- | +| `error` (default) | **Lit whenever the reader is running.** Dark while a card is being read, blinks on a bad read | The light should mean "the reader is alive and ready" | +| `presence` | **Lit only while a card is on the reader** | The light should mean "a card is here" | + +Wiring: LED anode to the GPIO pin through a ~330 Ω resistor, cathode to GND. + +### It cannot break the reader + +Three deliberate choices, because a LED is a convenience and the reader is the +point: + +* **Never blocks.** The workstation app toggled the LED directly on the reader + thread, which stalls card detection. Here every change is queued to a worker + thread, so a toggle returns immediately even with slow hardware. +* **Never fatal.** On a machine with no GPIO, `gpiozero` raises + `BadPinFactory`. The app falls back to a no-op LED and says why, rather than + refusing to run. +* **Never crashes on a dead pin.** A failed toggle is logged and ignored. + +Startup reports which of the three states it is in: + +``` +LED on GPIO 23 (error mode, gpiozero) # working +LED unavailable: gpiozero not installed # library missing +LED unavailable: no usable GPIO on pin 23: ... # no pin factory / pin busy +``` + +That distinction matters: `BadPinFactory` subclasses `ImportError`, so a naive +check reports "not installed" when the library is present but has no usable pin +provider — which would send you to the wrong fix. + +> **If the workstation app also runs on this device, use a different pin** — two +> processes driving GPIO 23 will fight, and the LED will flicker. + +`gpiozero` and `colorzero` are installed automatically on first use, from the +bundled wheels, so this works offline. They add ~180 KB. `setuptools`, which +`gpiozero` pulls in as a *packaging* dependency, is deliberately not vendored: +it is never imported at runtime and would add 818 KB for nothing. + +--- + ## Wiring The RDM6300 talks plain serial at **9600 baud, 8N1**. Connect it to the Pi's diff --git a/deploy/install.sh b/deploy/install.sh index 37d49b3..badcd64 100755 --- a/deploy/install.sh +++ b/deploy/install.sh @@ -105,6 +105,24 @@ else fi fi +# Status LED support. Best-effort: the reader works fine without a LED, so a +# failure here must not stop the install. +# +# --no-deps is deliberate: gpiozero declares colorzero, which in turn declares +# setuptools. setuptools is only a packaging requirement and is never imported +# at runtime, so it is not vendored (it would add 818 KB for nothing). +if [[ -d "$WHEEL_DIR" ]] && "$VENV_PY" -m pip install --quiet --no-index \ + --no-deps --find-links "$WHEEL_DIR" gpiozero colorzero 2>/dev/null; then + echo " installed gpiozero + colorzero for the status LED (offline)" +else + if "$VENV_PY" -m pip install --quiet gpiozero colorzero 2>/dev/null; then + echo " installed gpiozero + colorzero from PyPI" + else + warn "could not install LED support - the reader still works, the LED" + warn "will be inactive. The app can install it later on first use." + fi +fi + # Make the package importable. The .pth file adds src/ to sys.path, so # 'python -m rdm_card_reader' resolves from any working directory. SITE_PACKAGES="$("$VENV_PY" -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')" diff --git a/pyproject.toml b/pyproject.toml index 88d15ed..d465dac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,9 @@ dependencies = [ [project.optional-dependencies] dev = ["pytest>=7.0"] +# Status LED support. Installed automatically on first use by the app, or +# explicitly with: pip install '.[led]' +led = ["gpiozero>=2.0", "colorzero>=2.0"] [project.scripts] rdm-card-reader = "rdm_card_reader.cli:main" diff --git a/src/rdm_card_reader/__init__.py b/src/rdm_card_reader/__init__.py index a6dd172..723ad10 100644 --- a/src/rdm_card_reader/__init__.py +++ b/src/rdm_card_reader/__init__.py @@ -13,7 +13,7 @@ On first start the reader installs its own libraries if they are missing, so a fresh device needs no manual setup - see ``bootstrap``. """ -from .bootstrap import ensure_dependencies, missing_packages +from .bootstrap import ensure_dependencies, missing_packages, bundled_wheel_dir from .card import ( CANDIDATE_PORTS, CONFIG_CARD_VALUE, @@ -30,9 +30,10 @@ from .ports import ( discover, resolve, ) +from .led import DEFAULT_LED_PIN, StatusLED, create as create_led from .reader import available, list_ports, open_reader, reload_library -__version__ = "0.3.0" +__version__ = "0.4.0" __all__ = [ "CANDIDATE_PORTS", @@ -40,11 +41,15 @@ __all__ = [ "CardData", "CardEvent", "AmbiguousPort", + "DEFAULT_LED_PIN", "EventHub", "PortError", "PortInfo", "PortNotFound", + "StatusLED", "available", + "bundled_wheel_dir", + "create_led", "create_server", "discover", "ensure_dependencies", diff --git a/src/rdm_card_reader/bootstrap.py b/src/rdm_card_reader/bootstrap.py index eb2a9fa..f2c4cc1 100644 --- a/src/rdm_card_reader/bootstrap.py +++ b/src/rdm_card_reader/bootstrap.py @@ -35,6 +35,21 @@ REQUIRED_PACKAGES = ( ("serial", "pyserial>=3.5"), ) +#: Extra packages needed only when driving a status LED. +#: +#: ``colorzero`` is listed explicitly because ``gpiozero`` requires it. Both are +#: installed with ``--no-deps``: gpiozero's declared dependency chain pulls in +#: ``setuptools`` (via colorzero), which is only a packaging requirement and is +#: not imported at runtime. Vendoring it would add 818 KB for nothing. +LED_PACKAGES = ( + ("gpiozero", "gpiozero>=2.0"), + ("colorzero", "colorzero>=2.0"), +) + +#: The LED extras are optional: refusing to start because a LED cannot be +#: driven would be absurd. +OPTIONAL_IMPORTS = frozenset({"gpiozero", "colorzero"}) + #: Shown when an install fails, so the operator knows what to do by hand. MANUAL_HINT = ( "Install manually with: {python} -m pip install {requirements}" @@ -144,10 +159,17 @@ def in_virtualenv() -> bool: return sys.prefix != getattr(sys, "base_prefix", sys.prefix) -def _pip_command(requirements: Sequence[str], wheel_dir: Optional[str]) -> List[str]: +def _pip_command( + requirements: Sequence[str], + wheel_dir: Optional[str], + no_deps: bool = False, +) -> List[str]: """Build the pip invocation, handling venv vs system Python.""" command = [sys.executable, "-m", "pip", "install"] + if no_deps: + command.append("--no-deps") + if wheel_dir: # Offline install from a local wheel directory: no index, no network. command += ["--no-index", "--find-links", wheel_dir] @@ -168,10 +190,11 @@ def install( wheel_dir: Optional[str] = None, logger: Optional[Callable[[str], None]] = None, timeout: int = 300, + no_deps: bool = False, ) -> bool: """Run pip. Returns True on success; never raises for a failed install.""" say = logger or (lambda message: None) - command = _pip_command(requirements, wheel_dir) + command = _pip_command(requirements, wheel_dir, no_deps=no_deps) say(" $ {}".format(" ".join(command))) try: @@ -198,17 +221,26 @@ def install( return True +def _is_optional(packages: Sequence[tuple]) -> bool: + """True when every package in the group is an optional extra.""" + names = {name for name, _ in packages} + return bool(names) and names <= OPTIONAL_IMPORTS + + def ensure_dependencies( packages: Sequence[tuple] = REQUIRED_PACKAGES, wheel_dir: Optional[str] = None, auto_install: bool = True, logger: Optional[Callable[[str], None]] = None, + use_bundled: bool = True, + no_deps: bool = False, ) -> bool: - """Check for the reader libraries and install them if they are missing. + """Check for the given libraries and install them if they are missing. Returns True when everything is importable at the end - either because it already was, or because the install worked. Returns False when something is - still missing, so the caller can decide what to do. + still missing, so the caller can decide what to do. For an optional group + (the LED extras) False just means the feature is unavailable. """ say = logger or (lambda message: None) @@ -229,7 +261,7 @@ def ensure_dependencies( # Prefer the wheels shipped with the package: a device with no internet # must still be able to set itself up. explicit = wheel_dir is not None - if not explicit: + if wheel_dir is None and use_bundled: wheel_dir = bundled_wheel_dir() if wheel_dir: say("Using the wheels bundled with this package:") @@ -263,7 +295,8 @@ def ensure_dependencies( else: say("Installing from PyPI ...") - install(requirements, wheel_dir=wheel_dir, logger=say) + install(requirements, wheel_dir=wheel_dir, logger=say, + no_deps=no_deps and wheel_dir is not None) # Re-check rather than trusting pip's exit code alone. still_missing = missing_packages(packages) diff --git a/src/rdm_card_reader/cli.py b/src/rdm_card_reader/cli.py index d481c3a..4cec201 100644 --- a/src/rdm_card_reader/cli.py +++ b/src/rdm_card_reader/cli.py @@ -8,7 +8,7 @@ import threading from typing import Optional, Sequence from . import __version__ -from .bootstrap import ensure_dependencies +from .bootstrap import LED_PACKAGES, ensure_dependencies from .bridge import ( DEFAULT_HOST, DEFAULT_PORT, @@ -22,10 +22,12 @@ from .card import ( BAR, CANDIDATE_PORTS, DEFAULT_HEARTBEAT_INTERVAL, + EVENT_INVALID, CardEvent, format_event, ) from .demo import run_demo +from .led import 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 @@ -126,6 +128,33 @@ def _ensure_library(args) -> bool: return False +def _build_led(args): + """Set up the status LED, or a no-op when GPIO is unusable. + + Never fatal: a reader that refuses to start because a LED cannot be driven + would be worse than one with no light. + """ + if args.no_led: + return create_led(enabled=False) + + # Only pull in gpiozero when a LED was actually asked for, so a headless + # install does not grow the dependency set. + ensure_dependencies( + packages=LED_PACKAGES, + wheel_dir=args.wheel_dir, + auto_install=not args.no_install_deps, + logger=print, + no_deps=True, + ) + return create_led( + pin=args.led_pin, + enabled=True, + mode=args.led_mode, + error_blink=not args.led_no_error_blink, + logger=lambda message: print(" {}".format(message)), + ) + + def _run_bridge(args) -> int: """Serve what the reader sees over HTTP for a web app to consume.""" if not _ensure_library(args): @@ -143,7 +172,17 @@ def _run_bridge(args) -> int: hub = EventHub(mode=args.mode, tap_ttl=args.tap_ttl) hub.set_port(device) + led = _build_led(args) + def emit(event: CardEvent) -> None: + # Drive the LED first: it is queued, so it never delays the reader. + if event.event == EVENT_INVALID: + led.error() + elif event.event == "card_removed": + led.remove() + else: + led.insert() + hub.publish(event) if args.verbose: print(format_event(event), flush=True) @@ -195,6 +234,7 @@ def _run_bridge(args) -> int: reader.close() except Exception: pass + led.close() server.shutdown() server.server_close() return EXIT_OK @@ -254,6 +294,19 @@ def build_parser() -> argparse.ArgumentParser: help="permit binding a non-loopback address") bridge.add_argument("--verbose", action="store_true", help="also print each event and each HTTP request") + + led = parser.add_argument_group("status LED") + led.add_argument("--no-led", action="store_true", + help="do not drive a status LED") + 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-no-error-blink", action="store_true", + help="do not blink the LED on a bad read") return parser @@ -273,20 +326,32 @@ def main(argv: Optional[Sequence[str]] = None) -> int: if not _ensure_library(args): return EXIT_NO_LIBRARY + led = _build_led(args) print_event = _make_printer(args.json) ports = [args.port] if args.port else list(CANDIDATE_PORTS) def report_error(port: str, message: str) -> None: print(" x {}: {}".format(port, explain_error(port, message)), flush=True) + def on_event(event: CardEvent) -> None: + # LED first: it is queued, so it never delays the reader loop. + if event.event == EVENT_INVALID: + led.error() + elif event.event == "card_removed": + led.remove() + else: + led.insert() + print_event(event) + for port in ports: print("-> Opening {} ...".format(port), flush=True) - reader = open_reader(print_event, ports=ports, + reader = open_reader(on_event, ports=ports, heartbeat_interval=args.heartbeat, on_error=report_error) if reader is None: + led.close() print("\nx Could not open any serial port. Check:", file=sys.stderr) print(" 1. the reader is connected (rdm-card-reader --list-ports)", file=sys.stderr) @@ -308,6 +373,7 @@ def main(argv: Optional[Sequence[str]] = None) -> int: reader.close() except Exception: pass + led.close() return EXIT_OK diff --git a/src/rdm_card_reader/led.py b/src/rdm_card_reader/led.py new file mode 100644 index 0000000..94fa84a --- /dev/null +++ b/src/rdm_card_reader/led.py @@ -0,0 +1,314 @@ +"""Drive a status LED from card events. + +The workstation app lit an LED (GPIO 23) with the card: on when a card was +inserted, off when it was removed. This reproduces that, with three deliberate +differences: + +* It never blocks the reader. WMT toggled the LED and slept on the reader + thread, which stalls card detection; here every change is queued to a worker. +* It cannot crash the process. On a machine without GPIO, turning the LED on + raises ``BadPinFactory``. WMT worked around this with a ``DummyLED``; this + keeps the same idea but also handles a pin that disappears at runtime. +* It is explicit. Whether an LED is actually attached is reported, so "no light + came on" is distinguishable from "the card was not read". + +The default pin (23) matches the workstation app. **Use a separate LED from the +workstation's if both run on one device**, since two processes driving one pin +will fight. +""" + +from __future__ import annotations + +import queue +import threading +from dataclasses import dataclass +from typing import Any, Callable, Optional + +#: GPIO pin of the status LED. Chosen to match the workstation app. +DEFAULT_LED_PIN = 23 + +#: How long to hold an error indication, in seconds. +ERROR_BLINK_SECONDS = 1.0 + +OFF = "off" +ON = "on" +BLINK = "blink" + +#: 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 = (ERROR, PRESENCE) + +#: Kept for the occasional caller that only cares about the concrete commands. +_VALID_MODES = (OFF, ON, BLINK) + + +@dataclass +class LEDStatus: + """What the LED subsystem is actually doing.""" + + enabled: bool + pin: Optional[int] + backend: str # "gpiozero" | "null" + mode: str = ERROR + state: bool = False + reason: str = "" + + def to_dict(self) -> dict: + return { + "enabled": self.enabled, + "pin": self.pin, + "backend": self.backend, + "mode": self.mode, + "state": "on" if self.state else "off", + "reason": self.reason, + } + + +class NullLED: + """A LED that does nothing, used when there is no usable GPIO. + + Kept deliberately quiet: the reason it is inert is reported once at + startup, not on every card. + """ + + backend = "null" + + def __init__(self, pin: Optional[int], reason: str = ""): + self.pin = pin + self.reason = reason + + def on(self) -> None: + pass + + def off(self) -> None: + pass + + def close(self) -> None: + pass + + +class GPIOLED: + """A real LED on a GPIO pin, via gpiozero.""" + + backend = "gpiozero" + + def __init__(self, pin: int, active_high: bool = True): + from gpiozero import LED as _LED + + self.pin = pin + self._led = _LED(pin, active_high=active_high) + self._is_on = False + + def on(self) -> None: + self._led.on() + self._is_on = True + + def off(self) -> None: + self._led.off() + self._is_on = False + + def close(self) -> None: + try: + self._led.close() + except Exception: + pass + + +class _Blinker(threading.Thread): + """Repeatedly toggles a LED, for an error indication.""" + + def __init__(self, led: Any, period: float = 0.15): + super().__init__(daemon=True, name="led-blink") + self._led = led + self._period = period + self._stop = threading.Event() + + def run(self) -> None: + state = False + while not self._stop.is_set(): + try: + self._led.on() if state else self._led.off() + except Exception: + return + state = not state + self._stop.wait(self._period) + try: + self._led.off() + except Exception: + pass + + def stop(self) -> None: + self._stop.set() + + +class StatusLED: + """Non-blocking LED controller driven by card events. + + Commands are queued and applied by a worker thread, so calling :meth:`insert` + from the reader callback never blocks card detection. + """ + + def __init__( + self, + pin: int = DEFAULT_LED_PIN, + enabled: bool = True, + mode: str = ERROR, + 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._error_blink = error_blink + self._enabled = enabled + self._blinker: Optional[_Blinker] = None + self._state = False + + self._queue: "queue.Queue[Optional[str]]" = queue.Queue() + self._thread: Optional[threading.Thread] = None + + if not enabled: + self.led: Any = NullLED(None, "LED disabled (--no-led)") + self.status = LEDStatus(False, None, "null", self._mode, False, + "disabled") + return + + # Check the library separately from building the pin. + # + # gpiozero raises BadPinFactory when no pin provider works, and that is + # a subclass of ImportError, so catching ImportError around the pin + # construction would report "gpiozero is not installed" even when it is + # installed and simply has no usable pin library. On a Pi missing lgpio + # or RPi.GPIO that message sends you to the wrong fix. + try: + import gpiozero # noqa: F401 + except ImportError as exc: + reason = "gpiozero not installed: {}".format(exc) + self.led = NullLED(pin, reason) + self.status = LEDStatus(False, pin, "null", self._mode, False, + reason) + self._log("LED unavailable: {}".format(reason)) + else: + try: + self.led = GPIOLED(pin, active_high=active_high) + self.status = LEDStatus(True, pin, "gpiozero", self._mode, + False, "ready") + self._log("LED on GPIO {} ({} mode, gpiozero)".format( + pin, self._mode)) + except Exception as exc: + # No usable pin factory (not a Pi), pin already in use, bad pin. + reason = ("no usable GPIO on pin {}: {}".format(pin, exc)) + self.led = NullLED(pin, reason) + self.status = LEDStatus(False, pin, "null", self._mode, False, + reason) + self._log("LED unavailable: {}".format(reason)) + + if self.status.enabled and self._mode == ERROR: + # A health indicator shows 'alive' immediately. In presence mode the + # light means 'a card is here', so it must stay dark until one is. + self.led.on() + self._set_state(True) + self.status.state = True + + self._thread = threading.Thread(target=self._worker, daemon=True, + name="led-worker") + self._thread.start() + + # -- event hooks ------------------------------------------------------ + + def insert(self) -> None: + """A valid card was read: off in 'error' mode, on in 'presence' mode.""" + self._send(OFF if self._mode == ERROR else ON) + + def remove(self) -> None: + """The card left the reader: dark in 'presence' mode; 'error' mode stays on.""" + self._send(ON if self._mode == ERROR else OFF) + + def error(self) -> None: + """A bad read: blink, if blinking is enabled.""" + self._send(BLINK if self._error_blink else ON) + + # -- control ---------------------------------------------------------- + + def _send(self, command: str) -> None: + if not self.status.enabled: + return + try: + self._queue.put_nowait(command) + except Exception: + pass + + def _worker(self) -> None: + while True: + command = self._queue.get() + if command is None: + return + try: + self._apply(command) + except Exception as exc: + # A failed toggle must never take the reader down. + self._log("LED update failed: {}".format(exc)) + finally: + self._queue.task_done() + + def _apply(self, command: str) -> None: + # Blinking is used only for a bad read, and only by 'error' mode. + if command == BLINK and self._mode == ERROR: + self._stop_blink() + self._blinker = _Blinker(self.led) + self._blinker.start() + return + + self._stop_blink() # a steady command cancels any blink + + # Two readings of the light, hence two modes: + # error - a health indicator: lit the whole time the reader runs, + # so "light on" means "ready for a card". + # presence - a card indicator: lit only while a card is on the reader. + if command == OFF: + self.led.off() + self._set_state(False) + elif command == ON: + self.led.on() + self._set_state(True) + + def _stop_blink(self) -> None: + if self._blinker is not None: + self._blinker.stop() + self._blinker = None + + def _set_state(self, value: bool) -> None: + self._state = value + self.status.state = value + + def close(self) -> None: + self._stop_blink() + if self._thread is not None: + self._queue.put(None) + try: + self.led.off() + except Exception: + pass + self._set_state(False) + try: + self.led.close() + except Exception: + pass + + +def create( + enabled: bool = True, + pin: int = DEFAULT_LED_PIN, + mode: str = ERROR, + error_blink: bool = True, + logger: Optional[Callable[[str], None]] = None, +) -> StatusLED: + """Build the LED controller, degrading to a no-op when GPIO is unusable.""" + return StatusLED(pin=pin, enabled=enabled, mode=mode, + error_blink=error_blink, logger=logger) diff --git a/src/rdm_card_reader/wheels/SHA256SUMS b/src/rdm_card_reader/wheels/SHA256SUMS index 296ea5b..b57b74a 100644 --- a/src/rdm_card_reader/wheels/SHA256SUMS +++ b/src/rdm_card_reader/wheels/SHA256SUMS @@ -1,2 +1,4 @@ +0e60d743a6b8071498a56465f7719c96a5e92928f858bab1be2a0d606c9aa0f8 colorzero-2.0-py2.py3-none-any.whl +d491734803a9bc6036602e6b8bbd73fb25c33da88de0fcd11a40e890b8bc2d3b gpiozero-2.0.1.post3-py3-none-any.whl c4451db6ba391ca6ca299fb3ec7bae67a5c55dde170964c7a14ceefec02f2cf0 pyserial-3.5-py2.py3-none-any.whl afb067da8b8d23a477138ca67d4ff1d198f12d157b64668c9dd3cbd7df54e74e rdm6300-0.1.1-py3-none-any.whl diff --git a/src/rdm_card_reader/wheels/colorzero-2.0-py2.py3-none-any.whl b/src/rdm_card_reader/wheels/colorzero-2.0-py2.py3-none-any.whl new file mode 100644 index 0000000..d37236d Binary files /dev/null and b/src/rdm_card_reader/wheels/colorzero-2.0-py2.py3-none-any.whl differ diff --git a/src/rdm_card_reader/wheels/gpiozero-2.0.1.post3-py3-none-any.whl b/src/rdm_card_reader/wheels/gpiozero-2.0.1.post3-py3-none-any.whl new file mode 100644 index 0000000..61ea379 Binary files /dev/null and b/src/rdm_card_reader/wheels/gpiozero-2.0.1.post3-py3-none-any.whl differ diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py index db6912f..d0a25fd 100644 --- a/tests/test_bootstrap.py +++ b/tests/test_bootstrap.py @@ -129,7 +129,7 @@ def test_installs_when_missing(monkeypatch): lambda packages=None: [] if state["installed"] else [("rdm6300", "rdm6300>=0.1.1")], ) - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300): + def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): state["installed"] = True return True @@ -169,7 +169,7 @@ def test_wheel_dir_is_passed_through(monkeypatch): lambda packages=None: [] if seen.get("done") else [("x", "x>=1")], ) - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300): + def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): seen["wheel_dir"] = wheel_dir seen["done"] = True return True @@ -190,7 +190,7 @@ def test_incomplete_explicit_wheel_dir_is_flagged(monkeypatch): lambda packages=None: [] if seen.get("done") else [("x", "x>=1")], ) - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300): + def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): seen["done"] = True return True diff --git a/tests/test_e2e_led.py b/tests/test_e2e_led.py new file mode 100644 index 0000000..0aec5a8 --- /dev/null +++ b/tests/test_e2e_led.py @@ -0,0 +1,209 @@ +"""End-to-end: a real card read must drive the LED on and off. + +Uses a pseudo-terminal for the reader and a fake pin for the LED, so the whole +chain runs — serial frame -> card event -> LED worker -> pin state — without a +Pi and without GPIO hardware. +""" + +import os +import pty +import struct +import subprocess +import sys +import threading +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +PROJECT = os.path.dirname(HERE) +VENV_PY = os.path.join(PROJECT, ".venv", "bin", "python") +PYTHON = VENV_PY if os.path.exists(VENV_PY) else sys.executable + +CARD_VALUE = 0x00C4A0B5 + +# Runs the app's own CLI with gpiozero's LED replaced by a recorder, so we can +# observe what the app would have done to the pin. +DRIVER = r''' +import os, sys, threading, time, json + +sys.path.insert(0, os.environ["PROJECT_SRC"]) + +from rdm_card_reader import led as led_module + +TRANSITIONS = [] +_LOCK = threading.Lock() + + +class RecordingLED: + def __init__(self, pin=None, active_high=True): + self.pin = pin + self._state = False + + def _record(self, value): + self._state = value + with _LOCK: + TRANSITIONS.append(value) + # Write through immediately: the test terminates this process, so + # anything buffered for the end would be lost. + sys.stderr.write("TRANSITION=" + value + "\n") + sys.stderr.flush() + + def on(self): + self._record("on") + + def off(self): + self._record("off") + + def close(self): + pass + + +class FakeGPIOLED(led_module.GPIOLED): + def __init__(self, pin, active_high=True): + self.pin = pin + self._led = RecordingLED(pin, active_high) + self._is_on = False + + def on(self): + self._led.on() + self._is_on = True + + def off(self): + self._led.off() + self._is_on = False + + def close(self): + self._led.close() + + +led_module.GPIOLED = FakeGPIOLED + +from rdm_card_reader.cli import main + +code = main(["--port", os.environ["SERIAL_PORT"], "--heartbeat", "0.3", + "--led-mode", os.environ.get("LED_MODE", "error"), "--json"]) +sys.exit(code) +''' + + +def em4100_frame(value, card_type=0x0F): + id_bytes = struct.pack(">I", value) + checksum = card_type + for b in id_bytes: + checksum ^= b + frame = (bytes([card_type]) + id_bytes + bytes([checksum])).hex().upper() + assert len(frame) == 12, frame + return frame.encode("ascii") + + +def pump(stream, sink): + """Read a stream on its own thread. + + Text-mode pipes buffer, so select() would not report this data. + """ + for line in stream: + sink.append(line.rstrip()) + + +def run_case(mode): + """Drive a card through the app and return (events, led transitions).""" + master, slave = pty.openpty() + serial_port = os.ttyname(slave) + + env = dict(os.environ) + env["SERIAL_PORT"] = serial_port + env["PROJECT_SRC"] = os.path.join(PROJECT, "src") + env["LED_MODE"] = mode + env.pop("PYTHONPATH", None) + + proc = subprocess.Popen( + [PYTHON, "-u", "-c", DRIVER], + cwd=PROJECT, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, bufsize=1, + ) + + stdout, stderr = [], [] + t1 = threading.Thread(target=pump, args=(proc.stdout, stdout), daemon=True) + t2 = threading.Thread(target=pump, args=(proc.stderr, stderr), daemon=True) + t1.start() + t2.start() + + frame = em4100_frame(CARD_VALUE) + try: + time.sleep(2.0) # let the reader open the port + for _ in range(6): # card present + os.write(master, b"\x02" + frame + b"\x03") + time.sleep(0.1) + time.sleep(1.2) # stop writing -> card removed + finally: + try: + proc.terminate() + except Exception: + pass + try: + proc.wait(timeout=5) + except Exception: + proc.kill() + for t in (t1, t2): + t.join(timeout=3) + try: + os.close(master) + os.close(slave) + except Exception: + pass + + transitions = [ + line.split("=", 1)[1] + for line in stderr + if line.startswith("TRANSITION=") + ] + return stdout, stderr, transitions + + +def main(): + failures = 0 + + # --- error mode: lit while running, dark while a card is read ---------- + stdout, stderr, transitions = run_case("error") + print("======== error mode ========") + for line in stdout[:2]: + print(" " + line) + print(" events: {}".format( + [l.split('"event": "')[1].split('"')[0] + for l in stdout if '"event"' in l])) + print(" transitions: {}".format(transitions)) + + checks = { + "card was read": any("card_inserted" in l for l in stdout), + "card removal seen": any("card_removed" in l for l in stdout), + "lit at startup": transitions[:1] == ["on"], + "dark while reading": "off" in transitions, + "lit again after": transitions[-1:] == ["on"], + } + + # --- presence mode: dark until a card arrives ------------------------- + stdout2, stderr2, transitions2 = run_case("presence") + print("\n======== presence mode ========") + print(" transitions: {}".format(transitions2)) + checks.update({ + "presence: read still works": + any("card_inserted" in l for l in stdout2), + "presence: dark at startup": transitions2[:1] == ["on"] or not transitions2, + "presence: lit on the card": "on" in transitions2, + "presence: dark after": transitions2[-1:] == ["off"], + }) + + print() + ok = True + for name, passed in checks.items(): + print(" [{}] {}".format("PASS" if passed else "FAIL", name)) + ok = ok and bool(passed) + + if not ok: + failures = 1 + print("\n--- driver stderr (error mode) ---") + print("\n".join(stderr)[:1200]) + return failures + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_led.py b/tests/test_led.py new file mode 100644 index 0000000..8b52de2 --- /dev/null +++ b/tests/test_led.py @@ -0,0 +1,306 @@ +"""Tests for the status LED. + +Two things matter: the LED must never be able to break the reader, and it must +behave the way the workstation app did (on with the card, off without). Real +GPIO is unavailable here, so a fake pin is injected to test the logic. +""" + +import os +import sys +import threading +import time + +SRC = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" +) +if SRC not in sys.path: + sys.path.insert(0, SRC) + +import pytest # noqa: E402 + +from rdm_card_reader import led as led_module # noqa: E402 +from rdm_card_reader.led import ( # noqa: E402 + BLINK, + DEFAULT_LED_PIN, + ERROR, + OFF, + ON, + PRESENCE, + NullLED, + StatusLED, + create, +) + + +class FakeLED: + """Stands in for gpiozero.LED, recording what it was told to do.""" + + def __init__(self, pin, active_high=True): + self.pin = pin + self.active_high = active_high + self.state = False + self.transitions = [] + self.closed = False + + def on(self): + self.state = True + self.transitions.append("on") + + def off(self): + self.state = False + self.transitions.append("off") + + def close(self): + self.closed = True + + +@pytest.fixture +def fake_gpio(monkeypatch): + """Make StatusLED build a FakeLED instead of touching GPIO.""" + created = {} + + class FakeGPIOLED(led_module.GPIOLED): + def __init__(self, pin, active_high=True): + self.pin = pin + self._led = FakeLED(pin, active_high) + self._is_on = False + created["led"] = self._led + + def on(self): + self._led.on() + self._is_on = True + + def off(self): + self._led.off() + self._is_on = False + + def close(self): + self._led.close() + + monkeypatch.setattr(led_module, "GPIOLED", FakeGPIOLED) + return created + + +def wait_for_state(led_obj, expected, timeout=3.0): + """Wait for the worker thread to apply a command.""" + deadline = time.time() + timeout + while time.time() < deadline: + if led_obj.state is expected: + return True + time.sleep(0.01) + return False + + +# -- defaults --------------------------------------------------------------- + +def test_default_pin_matches_the_workstation_app(): + assert DEFAULT_LED_PIN == 23 + + +# -- graceful degradation --------------------------------------------------- + +def test_no_gpio_gives_a_working_no_op(monkeypatch): + """On a machine without GPIO the LED must be inert, not fatal.""" + def boom(pin, active_high=True): + raise RuntimeError("Unable to load any default pin factory!") + + monkeypatch.setattr(led_module, "GPIOLED", boom) + status = StatusLED(pin=23) + assert status.status.enabled is False + assert status.status.backend == "null" + # Every hook must still be safe to call. + status.insert() + status.remove() + status.error() + status.close() + + +def test_missing_gpiozero_is_reported_not_raised(monkeypatch): + """gpiozero absent: say so, and do not raise.""" + import builtins + + real_import = builtins.__import__ + + def blocked(name, *args, **kwargs): + if name == "gpiozero": + raise ImportError("No module named 'gpiozero'") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", blocked) + status = StatusLED(pin=23) + assert status.status.enabled is False + assert "not installed" in status.status.reason + + +def test_no_pin_factory_is_not_reported_as_missing_library(monkeypatch): + """BadPinFactory subclasses ImportError, so the two must be distinguished. + + Reporting 'gpiozero is not installed' when it is installed but has no + usable pin library would send you to the wrong fix. + """ + class FakeBadPinFactory(ImportError): + pass + + def boom(pin, active_high=True): + raise FakeBadPinFactory( + "Unable to load any default pin factory!" + ) + + monkeypatch.setattr(led_module, "GPIOLED", boom) + status = StatusLED(pin=23) + assert status.status.enabled is False + assert "not installed" not in status.status.reason + assert "pin factory" in status.status.reason + + +def test_disabled_led_is_inert(fake_gpio): + status = StatusLED(pin=23, enabled=False) + assert status.status.enabled is False + assert isinstance(status.led, NullLED) + status.insert() + time.sleep(0.05) + assert "led" not in fake_gpio + + +def test_null_led_methods_are_safe(): + null = NullLED(23, "no gpio") + null.on() + null.off() + null.close() + + +# -- behaviour -------------------------------------------------------------- + +def test_error_mode_lights_up_at_startup(fake_gpio): + """A health indicator should show 'alive' without waiting for a card.""" + status = StatusLED(pin=23, mode=ERROR) + assert fake_gpio["led"].state is True + status.close() + + +def test_error_mode_blinks_on_a_bad_read(fake_gpio): + status = StatusLED(pin=23, mode=ERROR, error_blink=True) + status.error() + time.sleep(0.25) + assert len(fake_gpio["led"].transitions) >= 2 + status.close() + + +def test_error_mode_settles_back_after_a_blink(fake_gpio): + status = StatusLED(pin=23, mode=ERROR, error_blink=True) + status.error() + time.sleep(0.25) + status.remove() # any steady command cancels blinking + assert wait_for_state(fake_gpio["led"], True) + status.close() + + +def test_error_mode_without_blink_stays_lit(fake_gpio): + status = StatusLED(pin=23, mode=ERROR, error_blink=False) + status.error() + assert wait_for_state(fake_gpio["led"], True) + status.close() + + +def test_presence_mode_is_dark_until_a_card_arrives(fake_gpio): + status = StatusLED(pin=23, mode=PRESENCE) + assert fake_gpio["led"].state is False + status.insert() + assert wait_for_state(fake_gpio["led"], True) + status.remove() + assert wait_for_state(fake_gpio["led"], False) + status.close() + + +def test_close_turns_the_led_off(fake_gpio): + status = StatusLED(pin=23, mode=ERROR) + assert fake_gpio["led"].state is True + status.close() + assert fake_gpio["led"].closed is True + assert fake_gpio["led"].state is False + + +def test_pin_and_active_high_are_passed_through(fake_gpio): + status = StatusLED(pin=17, active_high=False) + assert fake_gpio["led"].pin == 17 + assert fake_gpio["led"].active_high is False + status.close() + + +# -- never blocks or crashes ------------------------------------------------ + +def test_insert_does_not_block(monkeypatch): + """The reader thread calls this, so it must return immediately.""" + class SlowLED(led_module.GPIOLED): + def __init__(self, pin, active_high=True): + self.pin = pin + self._is_on = False + + def on(self): + time.sleep(0.5) # deliberately slow hardware + + def off(self): + pass + + def close(self): + pass + + monkeypatch.setattr(led_module, "GPIOLED", SlowLED) + # mode=presence so the startup light-up does not pay the slow 'on' cost + status = StatusLED(pin=23, mode=PRESENCE) + start = time.time() + for _ in range(20): + status.insert() + elapsed = time.time() - start + assert elapsed < 0.2, "insert() blocked for {:.2f}s".format(elapsed) + status.close() + + +def test_failing_toggle_does_not_raise(fake_gpio, monkeypatch): + """A dead pin must not take the reader down.""" + status = StatusLED(pin=23) + assert fake_gpio["led"] is not None + + def boom(): + raise OSError("pin went away") + + monkeypatch.setattr(status.led, "on", boom) + status.insert() + time.sleep(0.15) # worker swallows the error + status.close() # and close() still works + + +def test_hardware_error_is_reported_to_the_logger(monkeypatch): + lines = [] + + def boom(pin, active_high=True): + raise RuntimeError("no pin factory") + + monkeypatch.setattr(led_module, "GPIOLED", boom) + StatusLED(pin=23, logger=lines.append) + assert any("LED unavailable" in line for line in lines) + + +# -- factory ---------------------------------------------------------------- + +def test_create_defaults_to_error_mode_on_the_workstation_pin(fake_gpio): + status = create() + assert status.status.enabled is True + assert status.status.pin == DEFAULT_LED_PIN + assert status.status.mode == ERROR + status.close() + + +def test_create_disabled_returns_a_no_op(): + status = create(enabled=False) + assert status.status.enabled is False + status.close() + + +def test_status_dict_is_json_serialisable(fake_gpio): + import json + + status = create() + payload = json.dumps(status.status.to_dict()) + assert "gpiozero" in payload + status.close() diff --git a/tests/test_wheels.py b/tests/test_wheels.py index be3a439..48cfd40 100644 --- a/tests/test_wheels.py +++ b/tests/test_wheels.py @@ -142,7 +142,7 @@ def test_bundled_wheels_used_automatically(monkeypatch): lambda packages=None: [] if seen.get("done") else [("rdm6300", "rdm6300>=0.1.1")], ) - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300): + def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): seen["wheel_dir"] = wheel_dir seen["done"] = True return True @@ -168,7 +168,7 @@ def test_tampered_bundle_falls_back_to_pypi(monkeypatch, tmp_path): lambda packages=None: [] if seen.get("done") else [("rdm6300", "rdm6300>=0.1.1")], ) - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300): + def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): seen["wheel_dir"] = wheel_dir seen["done"] = True return True @@ -192,7 +192,7 @@ def test_explicit_wheel_dir_overrides_bundled(monkeypatch, tmp_path): lambda packages=None: [] if seen.get("done") else [("rdm6300", "rdm6300>=0.1.1")], ) - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300): + def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): seen["wheel_dir"] = wheel_dir seen["done"] = True return True @@ -216,7 +216,7 @@ def test_bundle_not_covering_a_requirement_falls_back(monkeypatch, tmp_path): else [("rdm6300", "rdm6300>=0.1.1"), ("serial", "pyserial>=3.5")], ) - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300): + def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): seen["wheel_dir"] = wheel_dir seen["wheel_dir_at_call"] = wheel_dir seen["done"] = True