Add a status LED driven by card events

The workstation app lit an LED on GPIO 23 with the card. This brings that
across, with three deliberate differences: it never blocks the reader, it
cannot be fatal, and its state is reported.

WMT toggled the LED directly inside card_inserted, on the reader thread, where
a slow write stalls card detection. Here every change is queued to a worker, so
insert() returns immediately (tested with a deliberately slow pin).

Two modes, because the light can mean two things:
  error    (default) lit while the reader runs, dark while reading, blinks on a
                     bad read - a health indicator
  presence           lit only while a card is on the reader

gpiozero raises BadPinFactory on a machine with no GPIO, and that subclasses
ImportError - so a single except ImportError reports 'gpiozero is not
installed' when the library is present but has no usable pin provider. The
library check and the pin construction are therefore separate.

Vendors gpiozero + colorzero (~180 KB) so the LED installs offline too, and
deliberately not setuptools: gpiozero declares it via colorzero, but it is only
a packaging requirement and is never imported at runtime. Installing to the
bundled wheels therefore uses --no-deps, saving 818 KB.

Verified end to end over a pseudo-terminal with a fake pin: a real RDM6300
frame drives the LED through the app's own CLI in both modes.

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
 

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())

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()

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