Add manual LED control to the status page

"When the light does not come on" has two very different causes -
the pin is wrong, or the wiring is wrong - and nothing so far could
tell them apart. The status page now has an LED panel with buttons
to switch either LED on or off by hand, which takes card events out
of the picture entirely: if the light comes on, the pin, backend and
permissions are all fine.

Backed by GET/POST /api/led. A forced state latches, holding through
card events and the error pulse, until released - otherwise the next
tap would immediately undo it and the test would be useless. A button
whose LED has no usable pin is disabled rather than silently doing
nothing, and the POST returns 409 rather than a misleading 200 when
there is no pin to drive.

Both LED states are recorded at the moment a request is served, not
only when the worker thread gets to it, so a snapshot cannot
contradict the response that was just sent.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
2026-10-05 22:30:28 +03:00
parent a399542a6e
commit c2b3eda649
5 changed files with 286 additions and 6 deletions
+37
View File
@@ -329,6 +329,20 @@ consume a real login just for being open. Claim from your own app instead.
Disable it with `--no-page` if you want the port to serve the API only. Disable it with `--no-page` if you want the port to serve the API only.
The page includes an **LED panel** with buttons to switch either LED on or off
by hand, for the case where the light does not come on and you need to know
whether the software can drive the pin at all. A forced state is held until you
press *hand back to card events*, so card events do not immediately undo it.
The same is reachable without a browser:
```bash
curl -sS http://127.0.0.1:8765/api/led # read the state
curl -sS -X POST -H 'Content-Type: application/json' \
-d '{"led":"status","on":true}' http://127.0.0.1:8765/api/led
curl -sS -X POST -H 'Content-Type: application/json' \
-d '{"release":true}' http://127.0.0.1:8765/api/led
```
### Endpoints ### Endpoints
| Method | Path | Purpose | | Method | Path | Purpose |
@@ -341,6 +355,8 @@ Disable it with `--no-page` if you want the port to serve the API only.
| `GET` | `/api/taps` | Recent tap history (`?limit=N`) | | `GET` | `/api/taps` | Recent tap history (`?limit=N`) |
| `GET` | `/api/health` | Reader state, port, uptime, tap counts, both TTLs — for monitoring | | `GET` | `/api/health` | Reader state, port, uptime, tap counts, both TTLs — for monitoring |
| `POST` | `/api/mode` | `{"mode": "login"}` or `{"mode": "monitor"}` | | `POST` | `/api/mode` | `{"mode": "login"}` or `{"mode": "monitor"}` |
| `GET` | `/api/led` | Both LEDs: pin, polarity, on/off, whether held manually |
| `POST` | `/api/led` | `{"led": "status"\|"error", "on": true\|false}`, or `{"release": true}` |
`login` mode (default) publishes taps only — a tap is an authentication `login` mode (default) publishes taps only — a tap is an authentication
attempt. `monitor` mode also publishes removals, which is what you want for attempt. `monitor` mode also publishes removals, which is what you want for
@@ -782,6 +798,27 @@ confirm with the journal, which now names the polarity:
LED on GPIO 23 (presence mode, active-low, gpiozero) LED on GPIO 23 (presence mode, active-low, gpiozero)
``` ```
**Check 0b — can the software drive the pin at all?** Before suspecting the
wiring, prove the pin works. Open the status page and use the **LED panel** to
switch each light on and off by hand, which takes card events out of the
picture entirely:
```bash
curl -sS -X POST -H 'Content-Type: application/json' \
-d '{"led":"status","on":true}' http://127.0.0.1:8765/api/led
```
- **The light comes on** → the pin, backend and permissions are all fine, so
the problem is card events, not hardware. Check the journal for
`CARD INSERTED` when you tap.
- **Nothing happens, HTTP 200** → the pin is being driven but no light is
connected to it. Wiring, or the wrong pin.
- **HTTP 409 `that LED is not available`** → the LED has no usable pin; look at
`reason` in the response, which is the same text as the journal line.
Press *hand back to card events* (or POST `{"release": true}`) when finished, or
the LEDs stay held where you left them.
To isolate polarity from everything else, run the service in `error` mode, To isolate polarity from everything else, run the service in `error` mode,
where the light is lit the whole time the app runs: where the light is lit the whole time the app runs:
+68 -1
View File
@@ -466,6 +466,15 @@ class _Handler(BaseHTTPRequestHandler):
self._send_json(self._hub.current_card()) self._send_json(self._hub.current_card())
return return
if path == "/api/led":
controller = getattr(self.server, "led", None)
if controller is None:
self._send_json({"error": "LED control not available"},
status=503)
return
self._send_json(controller.snapshot())
return
if path == "/api/taps": if path == "/api/taps":
limit = 20 limit = 20
if "?" in self.path: if "?" in self.path:
@@ -487,6 +496,58 @@ class _Handler(BaseHTTPRequestHandler):
def do_POST(self) -> None: # noqa: N802 def do_POST(self) -> None: # noqa: N802
path = self.path.split("?", 1)[0].rstrip("/") or "/" path = self.path.split("?", 1)[0].rstrip("/") or "/"
if path == "/api/led":
if not self._cors_ok():
self._send_json({"error": "origin not allowed"}, status=403)
return
if not self._token_ok():
self._send_json({"error": "missing or invalid token"}, status=401)
return
controller = getattr(self.server, "led", None)
if controller is None:
self._send_json({"error": "LED control not available"},
status=503)
return
length = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(length) if length else b"{}"
try:
payload = json.loads(raw.decode("utf-8") or "{}")
except ValueError:
self._send_json({"error": "invalid JSON body"}, status=400)
return
which = str(payload.get("led", ""))
if which == "release" or payload.get("release"):
controller.release()
self._send_json(controller.snapshot())
return
if "on" not in payload:
self._send_json(
{"error": "expected {\"led\": \"status|error\", "
"\"on\": true|false}, or {\"release\": true}"},
status=400,
)
return
if which not in ("status", "error"):
self._send_json(
{"error": "unknown led", "valid": ["status", "error"]},
status=400,
)
return
if not controller.set_led(which, bool(payload.get("on"))):
# The LED exists in configuration but has no usable pin, so
# there is nothing to switch. Say so rather than 200.
self._send_json(
{"error": "that LED is not available",
"state": controller.snapshot()},
status=409,
)
return
self._send_json(controller.snapshot())
return
if path == "/api/card/claim": if path == "/api/card/claim":
if not self._cors_ok(): if not self._cors_ok():
self._send_json({"error": "origin not allowed"}, status=403) self._send_json({"error": "origin not allowed"}, status=403)
@@ -588,11 +649,17 @@ def create_server(
token: Optional[str] = None, token: Optional[str] = None,
logger: Optional[Callable[[str], None]] = None, logger: Optional[Callable[[str], None]] = None,
serve_page: bool = True, serve_page: bool = True,
led: Any = None,
) -> ThreadingHTTPServer: ) -> ThreadingHTTPServer:
"""Build (but do not start) the HTTP server around *hub*.""" """Build (but do not start) the HTTP server around *hub*.
*led* is the LED controller, exposed for the manual override at
``/api/led``. It is optional: without it the endpoint reports 503.
"""
server = ThreadingHTTPServer((host, port), _Handler) server = ThreadingHTTPServer((host, port), _Handler)
server.daemon_threads = True server.daemon_threads = True
server.hub = hub # type: ignore[attr-defined] server.hub = hub # type: ignore[attr-defined]
server.led = led # type: ignore[attr-defined]
server.serve_page = serve_page # type: ignore[attr-defined] server.serve_page = serve_page # type: ignore[attr-defined]
server.allowed_origins = list(allowed_origins or []) # type: ignore[attr-defined] server.allowed_origins = list(allowed_origins or []) # type: ignore[attr-defined]
server.token = token or os.environ.get("RDM_BRIDGE_TOKEN") or None # type: ignore[attr-defined] server.token = token or os.environ.get("RDM_BRIDGE_TOKEN") or None # type: ignore[attr-defined]
+2
View File
@@ -224,6 +224,7 @@ def _run_bridge(args) -> int:
token=args.token, token=args.token,
logger=print if args.verbose else None, logger=print if args.verbose else None,
serve_page=not args.no_page, serve_page=not args.no_page,
led=led,
) )
server_thread = threading.Thread(target=server.serve_forever, daemon=True, server_thread = threading.Thread(target=server.serve_forever, daemon=True,
@@ -235,6 +236,7 @@ def _run_bridge(args) -> int:
if not args.no_page: if not args.no_page:
print(" GET / <- status page in a browser") print(" GET / <- status page in a browser")
print(" GET /api/card <- the card on the reader right now") print(" GET /api/card <- the card on the reader right now")
print(" GET /api/led <- LED state; POST to switch one by hand")
print(" POST /api/card/claim (single use, for one login)") print(" POST /api/card/claim (single use, for one login)")
print(" GET /api/events (SSE stream)") print(" GET /api/events (SSE stream)")
print(" GET /api/last /api/health") print(" GET /api/last /api/health")
+73 -2
View File
@@ -226,7 +226,10 @@ class StatusLED:
self._state = False self._state = False
self._error_until: Optional[float] = None self._error_until: Optional[float] = None
self._error_state = False self._error_state = False
#: Set while the status page is forcing a LED, so card events and the
#: error pulse leave it alone.
self._manual_status = False
self._manual_error = False
self._queue: "queue.Queue[Optional[str]]" = queue.Queue() self._queue: "queue.Queue[Optional[str]]" = queue.Queue()
self._thread: Optional[threading.Thread] = None self._thread: Optional[threading.Thread] = None
@@ -295,10 +298,14 @@ class StatusLED:
def insert(self) -> None: def insert(self) -> None:
"""A valid card was read: off in 'error' mode, on in 'presence' mode.""" """A valid card was read: off in 'error' mode, on in 'presence' mode."""
if self._manual_status:
return
self._send(OFF if self._mode == ERROR else ON) self._send(OFF if self._mode == ERROR else ON)
def remove(self) -> None: def remove(self) -> None:
"""The card left the reader: dark in 'presence' mode; 'error' mode stays on.""" """The card left the reader: dark in 'presence' mode; 'error' mode stays on."""
if self._manual_status:
return
self._send(ON if self._mode == ERROR else OFF) self._send(ON if self._mode == ERROR else OFF)
def error(self) -> None: def error(self) -> None:
@@ -310,6 +317,8 @@ class StatusLED:
blinks as before - but only in ``error`` mode, since a blink and "card blinks as before - but only in ``error`` mode, since a blink and "card
present" cannot both be shown on one light. present" cannot both be shown on one light.
""" """
if self._manual_status or self._manual_error:
return
if self.error_status.enabled and self._error_pulse > 0: if self.error_status.enabled and self._error_pulse > 0:
self._error_until = time.monotonic() + self._error_pulse self._error_until = time.monotonic() + self._error_pulse
return return
@@ -328,6 +337,66 @@ class StatusLED:
self._error_state = on self._error_state = on
self.error_status.state = on self.error_status.state = on
# -- manual control (diagnostics) -------------------------------------
def set_led(self, which: str, on: bool) -> bool:
"""Force one LED on or off by hand. Returns False for an unknown name.
This exists to answer "can the software drive the pin at all?", which
separates a wiring fault from a configuration one. It latches: the
forced state holds until :meth:`release` or the next card event, so a
light turned on from the status page does not go dark when the next
poll runs.
"""
if which == "status":
if not self.status.enabled:
return False
self._manual_status = True
self._send(ON if on else OFF)
# Record the requested state now: the worker applies it
# asynchronously, and a snapshot taken before it runs would
# otherwise contradict the response we just sent.
self._set_state(bool(on))
return True
if which == "error":
if not self.error_status.enabled:
return False
self._manual_error = True
# Cancel any pulse in flight so it cannot fight the manual state.
self._error_until = None
self._error_indicate(bool(on))
return True
return False
def release(self) -> None:
"""Drop any manual override and let card events drive the LEDs again."""
self._manual_status = False
self._manual_error = False
self._error_until = None
if self.status.enabled:
# Return to the resting state for the current mode, recorded now so
# a snapshot taken before the worker runs agrees with it.
resting = self._mode == ERROR
self._send(ON if resting else OFF)
self._set_state(resting)
if self.error_status.enabled:
self._error_indicate(False)
def snapshot(self) -> dict:
"""Both LEDs' state, for ``/api/led`` and the status page."""
return {
"manual": self._manual_status or self._manual_error,
"status": self._led_info(self.status, self._manual_status),
"error": self._led_info(self.error_status, self._manual_error),
}
@staticmethod
def _led_info(status: "LEDStatus", manual: bool) -> dict:
info = status.to_dict()
info["manual"] = manual
info["on"] = bool(status.state)
return info
# -- control ---------------------------------------------------------- # -- control ----------------------------------------------------------
def _send(self, command: str) -> None: def _send(self, command: str) -> None:
@@ -358,7 +427,7 @@ class StatusLED:
def _tick_error_pulse(self) -> None: def _tick_error_pulse(self) -> None:
"""Light the error LED for its window, then darken it again.""" """Light the error LED for its window, then darken it again."""
if self._error_until is None: if self._error_until is None or self._manual_error:
return return
now = time.monotonic() now = time.monotonic()
if now < self._error_until: if now < self._error_until:
@@ -402,6 +471,8 @@ class StatusLED:
def close(self) -> None: def close(self) -> None:
self._stop_blink() self._stop_blink()
self._manual_status = False
self._manual_error = False
if self._thread is not None: if self._thread is not None:
try: try:
self._queue.put_nowait(None) self._queue.put_nowait(None)
+106 -3
View File
@@ -81,6 +81,16 @@ PAGE_HTML = """<!doctype html>
.invalid .ev { color: var(--bad); } .invalid .ev { color: var(--bad); }
.detail { color: var(--dim); word-break: break-all; } .detail { color: var(--dim); word-break: break-all; }
.detail b { color: var(--fg); font-weight: 600; } .detail b { color: var(--fg); font-weight: 600; }
.ledbtns { display: flex; flex-wrap: wrap; gap: .4rem; margin-top: .7rem; }
.ledbtns button {
font: inherit; font-size: .8rem; padding: .3rem .6rem; cursor: pointer;
color: var(--fg); background: #23272f; border: 1px solid var(--line);
border-radius: 6px;
}
.ledbtns button:hover:not(:disabled) { border-color: var(--accent); }
.ledbtns button:disabled { opacity: .4; cursor: not-allowed; }
.ledmsg { margin: .6rem 0 0; font-size: .78rem; color: var(--dim); min-height: 1em; }
.ledmsg.err { color: var(--bad); }
footer { margin-top: 1rem; color: var(--dim); font-size: .78rem; } footer { margin-top: 1rem; color: var(--dim); font-size: .78rem; }
footer code { color: var(--fg); background: var(--panel); padding: .1rem .3rem; border-radius: 4px; } footer code { color: var(--fg); background: var(--panel); padding: .1rem .3rem; border-radius: 4px; }
</style> </style>
@@ -127,6 +137,25 @@ PAGE_HTML = """<!doctype html>
<dt>tap / claimed TTL</dt><dd id="n-ttl">&mdash;</dd> <dt>tap / claimed TTL</dt><dd id="n-ttl">&mdash;</dd>
</dl> </dl>
</section> </section>
<section class="panel">
<h2>LEDs</h2>
<dl>
<dt>green (status)</dt> <dd id="led-st-state">&mdash;</dd>
<dt>red (error)</dt> <dd id="led-er-state">&mdash;</dd>
<dt>status pin</dt> <dd id="led-st-pin">&mdash;</dd>
<dt>error pin</dt> <dd id="led-er-pin">&mdash;</dd>
<dt>driven by</dt> <dd id="led-auto">&mdash;</dd>
</dl>
<div class="ledbtns">
<button type="button" data-led="status" data-on="1">Green on</button>
<button type="button" data-led="status" data-on="0">Green off</button>
<button type="button" data-led="error" data-on="1">Red on</button>
<button type="button" data-led="error" data-on="0">Red off</button>
<button type="button" id="led-release">Hand back to card events</button>
</div>
<p class="ledmsg" id="led-msg">Switching here holds until you hand it back.</p>
</section>
</main> </main>
<section class="logwrap"> <section class="logwrap">
@@ -190,6 +219,79 @@ PAGE_HTML = """<!doctype html>
? "\\u2014" : String(value); ? "\\u2014" : String(value);
} }
// ---- LEDs -----------------------------------------------------------
function ledMsg(text, isError) {
var node = el("led-msg");
node.textContent = text;
node.className = "ledmsg" + (isError ? " err" : "");
}
function renderLEDs(data) {
var st = data.status || {}, er = data.error || {};
var stState = !st.enabled ? "off (no pin)"
: (st.on ? "on" : "off") + (st.manual ? " (manual)" : "");
var erState = !er.enabled ? "off (no pin)"
: (er.on ? "on" : "off") + (er.manual ? " (manual)" : "");
setText("led-st-state", stState);
setText("led-er-state", erState);
setText("led-st-pin", st.pin === null || st.pin === undefined
? "\\u2014" : "GPIO " + st.pin + (st.active_high === false ? ", active-low" : ""));
setText("led-er-pin", er.pin === null || er.pin === undefined
? "\\u2014" : "GPIO " + er.pin + (er.active_high === false ? ", active-low" : ""));
setText("led-auto", data.manual ? "held (manual)" : "card events");
// Disable a button when its LED has no usable pin, so a dead control is
// visibly dead rather than silently doing nothing.
var buttons = document.querySelectorAll(".ledbtns button[data-led]");
Array.prototype.forEach.call(buttons, function (b) {
var name = b.getAttribute("data-led");
if (!name) { b.disabled = false; return; }
b.disabled = name === "status" ? !st.enabled : !er.enabled;
});
el("led-release").disabled = !data.manual;
}
function ledCommand(body, message) {
fetch("/api/led", {
method: "POST",
headers: Object.assign({ "Content-Type": "application/json" },
authHeaders()),
body: JSON.stringify(body)
}).then(function (r) {
return r.json().then(function (data) { return { ok: r.ok, status: r.status, data: data }; });
}).then(function (res) {
if (!res.ok) {
ledMsg(res.data && res.data.error ? res.data.error : "failed (" + res.status + ")", true);
if (res.data && res.data.state) renderLEDs(res.data.state);
return;
}
renderLEDs(res.data);
ledMsg(message);
}).catch(function () { ledMsg("request failed", true); });
}
document.querySelectorAll(".ledbtns button[data-led]").forEach(function (button) {
button.addEventListener("click", function () {
var name = button.getAttribute("data-led");
var on = button.getAttribute("data-on") === "1";
ledCommand({ led: name, on: on },
name + " LED forced " + (on ? "on" : "off") +
" - press \\"hand back\\" to restore");
});
});
el("led-release").addEventListener("click", function () {
ledCommand({ release: true }, "back under card-event control");
});
function pollLEDs() {
fetch("/api/led", { headers: authHeaders() })
.then(function (r) { return r.ok ? r.json() : null; })
.then(function (data) { if (data) renderLEDs(data); })
.catch(function () { /* the reader panel already reports failures */ });
}
function renderStatus(health, card) { function renderStatus(health, card) {
setText("mode", health.mode); setText("mode", health.mode);
setText("port", health.port || "not open"); setText("port", health.port || "not open");
@@ -292,9 +394,7 @@ PAGE_HTML = """<!doctype html>
} }
var es = null; var es = null;
function connectEvents() { if (es) es.close();
function connectEvents() {
if (es) es.close();
var t = token(); var t = token();
// EventSource cannot set headers, so the token goes in the query string - // EventSource cannot set headers, so the token goes in the query string -
// the same mechanism the API already accepts for non-local callers. // the same mechanism the API already accepts for non-local callers.
@@ -312,12 +412,15 @@ PAGE_HTML = """<!doctype html>
connectEvents(); connectEvents();
seedLog(); seedLog();
poll(); poll();
pollLEDs();
}); });
connectEvents(); connectEvents();
seedLog(); seedLog();
poll(); poll();
pollLEDs();
setInterval(poll, REFRESH_MS); setInterval(poll, REFRESH_MS);
setInterval(pollLEDs, REFRESH_MS);
})(); })();
</script> </script>
</body> </body>