Remove dead code and use the event constants consistently

- drop unused BAR import, _VALID_MODES, _is_optional(),
  OPTIONAL_IMPORTS and is_loopback()
- replace hand-written event name strings with EVENT_INSERTED,
  EVENT_REMOVED and EVENT_INVALID, so a typo cannot silently
  disable an event path
- drop the now-moot pytest/coverage entries from .gitignore

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
2026-10-05 21:24:45 +03:00
parent 15b43d476a
commit c720f4ca67
7 changed files with 21 additions and 33 deletions
-3
View File
@@ -5,8 +5,5 @@ build/
dist/ dist/
.venv/ .venv/
venv/ venv/
.pytest_cache/
.coverage
htmlcov/
.idea/ .idea/
.vscode/ .vscode/
-10
View File
@@ -46,10 +46,6 @@ LED_PACKAGES = (
("colorzero", "colorzero>=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. #: Shown when an install fails, so the operator knows what to do by hand.
MANUAL_HINT = ( MANUAL_HINT = (
"Install manually with: {python} -m pip install {requirements}" "Install manually with: {python} -m pip install {requirements}"
@@ -221,12 +217,6 @@ def install(
return True 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( def ensure_dependencies(
packages: Sequence[tuple] = REQUIRED_PACKAGES, packages: Sequence[tuple] = REQUIRED_PACKAGES,
wheel_dir: Optional[str] = None, wheel_dir: Optional[str] = None,
+2 -7
View File
@@ -56,7 +56,7 @@ from collections import deque
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any, Callable, Deque, Dict, List, Optional from typing import Any, Callable, Deque, Dict, List, Optional
from .card import EVENT_INSERTED, EVENT_REMOVED, CardEvent from .card import EVENT_INVALID, EVENT_INSERTED, EVENT_REMOVED, CardEvent
#: Reader modes. ``login`` publishes inserts only (a tap = an authentication #: Reader modes. ``login`` publishes inserts only (a tap = an authentication
#: attempt). ``monitor`` publishes inserts and removals both. #: attempt). ``monitor`` publishes inserts and removals both.
@@ -111,7 +111,7 @@ class EventHub:
self._last_at: Optional[float] = None self._last_at: Optional[float] = None
self._mode = mode if mode in VALID_MODES else MODE_LOGIN self._mode = mode if mode in VALID_MODES else MODE_LOGIN
self._started_at = time.time() self._started_at = time.time()
self._counts = {EVENT_INSERTED: 0, EVENT_REMOVED: 0, "invalid_card": 0} self._counts = {EVENT_INSERTED: 0, EVENT_REMOVED: 0, EVENT_INVALID: 0}
self._port: Optional[str] = None self._port: Optional[str] = None
self._tap_ttl = float(tap_ttl if tap_ttl else 0.0) self._tap_ttl = float(tap_ttl if tap_ttl else 0.0)
self._claimed_ttl = float(claimed_ttl if claimed_ttl else 0.0) self._claimed_ttl = float(claimed_ttl if claimed_ttl else 0.0)
@@ -541,8 +541,3 @@ def create_server(
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]
server.logger = logger # type: ignore[attr-defined] server.logger = logger # type: ignore[attr-defined]
return server return server
def is_loopback(host: str) -> bool:
"""True for hosts that are not reachable off-box."""
return host in ("127.0.0.1", "::1", "localhost")
+3 -3
View File
@@ -20,10 +20,10 @@ from .bridge import (
create_server, create_server,
) )
from .card import ( from .card import (
BAR,
CANDIDATE_PORTS, CANDIDATE_PORTS,
DEFAULT_HEARTBEAT_INTERVAL, DEFAULT_HEARTBEAT_INTERVAL,
EVENT_INVALID, EVENT_INVALID,
EVENT_REMOVED,
CardEvent, CardEvent,
format_event, format_event,
) )
@@ -185,7 +185,7 @@ def _run_bridge(args) -> int:
# Drive the LED first: it is queued, so it never delays the reader. # Drive the LED first: it is queued, so it never delays the reader.
if event.event == EVENT_INVALID: if event.event == EVENT_INVALID:
led.error() led.error()
elif event.event == "card_removed": elif event.event == EVENT_REMOVED:
led.remove() led.remove()
else: else:
led.insert() led.insert()
@@ -348,7 +348,7 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
# LED first: it is queued, so it never delays the reader loop. # LED first: it is queued, so it never delays the reader loop.
if event.event == EVENT_INVALID: if event.event == EVENT_INVALID:
led.error() led.error()
elif event.event == "card_removed": elif event.event == EVENT_REMOVED:
led.remove() led.remove()
else: else:
led.insert() led.insert()
+5 -2
View File
@@ -7,6 +7,9 @@ import time
from .card import ( from .card import (
DEFAULT_HEARTBEAT_INTERVAL, DEFAULT_HEARTBEAT_INTERVAL,
EVENT_INVALID,
EVENT_INSERTED,
EVENT_REMOVED,
CardData, CardData,
CardEvent, CardEvent,
format_event, format_event,
@@ -37,12 +40,12 @@ def run_demo(
for _ in range(max(1, count)): for _ in range(max(1, count)):
card = rng.choice(DEMO_CARDS) card = rng.choice(DEMO_CARDS)
event_name = "card_inserted" if card.is_valid else "invalid_card" event_name = EVENT_INSERTED if card.is_valid else EVENT_INVALID
_emit(CardEvent.from_card(event_name, card, "demo"), as_json) _emit(CardEvent.from_card(event_name, card, "demo"), as_json)
time.sleep(max(0.1, heartbeat)) time.sleep(max(0.1, heartbeat))
if card.is_valid: if card.is_valid:
_emit(CardEvent.from_card("card_removed", card, "demo"), as_json) _emit(CardEvent.from_card(EVENT_REMOVED, card, "demo"), as_json)
time.sleep(0.3) time.sleep(0.3)
print("\nDemo finished.", flush=True) print("\nDemo finished.", flush=True)
-4
View File
@@ -49,10 +49,6 @@ MODES = (PRESENCE, ERROR)
#: What the light means unless told otherwise. #: What the light means unless told otherwise.
DEFAULT_LED_MODE = PRESENCE DEFAULT_LED_MODE = PRESENCE
#: Kept for the occasional caller that only cares about the concrete commands.
_VALID_MODES = (OFF, ON, BLINK)
@dataclass @dataclass
class LEDStatus: class LEDStatus:
"""What the LED subsystem is actually doing.""" """What the LED subsystem is actually doing."""
+11 -4
View File
@@ -10,7 +10,14 @@ import importlib
import os import os
from typing import Callable, Dict, List, Optional, Sequence from typing import Callable, Dict, List, Optional, Sequence
from .card import CANDIDATE_PORTS, DEFAULT_HEARTBEAT_INTERVAL, CardEvent from .card import (
CANDIDATE_PORTS,
DEFAULT_HEARTBEAT_INTERVAL,
EVENT_INVALID,
EVENT_INSERTED,
EVENT_REMOVED,
CardEvent,
)
try: # pragma: no cover - depends on the environment try: # pragma: no cover - depends on the environment
import rdm6300 import rdm6300
@@ -58,15 +65,15 @@ def _make_reader_class(callback: EventCallback):
"""Fires *callback* for each insert, removal and checksum failure.""" """Fires *callback* for each insert, removal and checksum failure."""
def card_inserted(self, card) -> None: def card_inserted(self, card) -> None:
callback(CardEvent.from_card("card_inserted", card, self.port)) callback(CardEvent.from_card(EVENT_INSERTED, card, self.port))
def card_removed(self, card) -> None: def card_removed(self, card) -> None:
if card is None: if card is None:
return return
callback(CardEvent.from_card("card_removed", card, self.port)) callback(CardEvent.from_card(EVENT_REMOVED, card, self.port))
def invalid_card(self, card) -> None: def invalid_card(self, card) -> None:
callback(CardEvent.from_card("invalid_card", card, self.port)) callback(CardEvent.from_card(EVENT_INVALID, card, self.port))
return CardReader return CardReader