From b15b661c2d1d487334ec1ec02c3275547e5fbd15 Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Thu, 16 Apr 2026 15:21:40 -0600 Subject: [PATCH] feat!: mux-py on DongLoRa Protocol v2 with tag-aware dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 1.0 Python mux, mirroring the Rust donglora-mux 1.0 one-for-one and implementing the full §13 multi-client semantics on the DongLoRa Protocol v2 wire format. * Structural rewrite. donglora_mux/{daemon,intercept,session,tagmap}.py replace the 601-line __init__.py monolith. * Tag-aware dispatch. TagMapper with skip-zero allocation and per-client bulk drop on disconnect. Multiple clients can have commands in flight concurrently; the single-oneshot response slot is gone. * Full §13.2 multi-client SET_CONFIG semantics. Tracks (client_id, modulation) lock owner and synthesizes OK(APPLIED/MINE), OK(ALREADY_MATCHED/OTHER), or OK(LOCKED_MISMATCH/OTHER). * TX loopback fan-out (§13.4). On TX_DONE(TRANSMITTED) the mux synthesizes an RX frame with origin = LOCAL_LOOPBACK for every connected client except the transmitter. * Async ERR(EFRAME) fan-out: tag-0 framing errors are broadcast to all clients. * Per-client TX cache on ClientSession so the loopback fan-out does not need the sender to re-buffer. * Socket and serial fcntl.flock to prevent multiple mux instances from fighting over the same dongle or socket path. * 500 ms keepalive matching PROTOCOL.md §3.4 (2x margin on the 1 s inactivity timer), replacing the 5 s cadence from 0.x. * Clean SIGINT / SIGTERM shutdown: per-client handler tasks are tracked and cancelled, so blocked reader.read() calls no longer pin the event loop open. --- CHANGELOG.md | 65 +++- README.md | 47 ++- donglora_mux/__init__.py | 569 +++---------------------------- donglora_mux/daemon.py | 695 ++++++++++++++++++++++++++++++++++++++ donglora_mux/intercept.py | 200 +++++++++++ donglora_mux/session.py | 77 +++++ donglora_mux/tagmap.py | 75 ++++ pyproject.toml | 15 +- tests/test_intercept.py | 318 +++++++++-------- tests/test_multiclient.py | 214 ++++++++++++ tests/test_tagmap.py | 77 +++++ uv.lock | 35 +- 12 files changed, 1693 insertions(+), 694 deletions(-) create mode 100644 donglora_mux/daemon.py create mode 100644 donglora_mux/intercept.py create mode 100644 donglora_mux/session.py create mode 100644 donglora_mux/tagmap.py create mode 100644 tests/test_multiclient.py create mode 100644 tests/test_tagmap.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a03828..5d60938 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,19 +1,66 @@ # Changelog -## Unreleased +## 1.0.0 — 2026-04-21 + +### Breaking + +- **Upgraded to the DongLoRa Protocol v2 wire format.** The 0.x line spoke the legacy + v0.1 protocol with hardcoded tag constants (`TAG_OK=4`, `TAG_ERROR=5`, + `CMD_TAG_SET_CONFIG=2`, …) that don't exist in v1.0. This release is + wire-incompatible with 0.x clients and firmware. +- **Pins `donglora>=1.0.0`** and uses its `donglora.frame` / + `donglora.events` / `donglora.commands` modules as the wire-level + backbone. No more in-package reimplementation of the wire protocol. +- **Structural rewrite** mirroring the Rust sibling + [`donglora-mux`](https://crates.io/crates/donglora-mux) one-to-one: + `donglora_mux/{daemon,intercept,session,tagmap}.py` replace the single + 601-line `__init__.py`. ### Added -- Keepalive ping: detects unresponsive dongles (e.g. UART boards behind a - CP2102 bridge where the serial port stays open across ESP32 resets). Pings - every 5 seconds of idle, declares dongle lost after 2-second timeout. -- Radio state restoration after reconnect: saved SetConfig and StartRx are - re-sent to the dongle so clients don't need to know a reset happened. +- **`TagMapper`** — per-command device-tag allocator. Enables concurrent + out-of-order command/response correlation for multi-client setups + (the v0.1-era single-oneshot-response slot is gone). +- **Full `§13.2` multi-client `SET_CONFIG` semantics** — tracks the + `(client_id, modulation)` lock owner. Cross-client calls synthesize: + - `OK(APPLIED / MINE, params)` when the caller owns the lock + (forwarded to dongle). + - `OK(ALREADY_MATCHED / OTHER, current)` when another client holds + the lock and the requested params byte-match. + - `OK(LOCKED_MISMATCH / OTHER, current)` when they differ; `current` + echoes the lock owner's actual modulation. + - Lock auto-releases when the owning client disconnects. +- **TX loopback fan-out (`§13.4`)** — on `TX_DONE(TRANSMITTED)` the mux + synthesizes an `RX` frame with `origin = LOCAL_LOOPBACK` and sends it + to every connected client except the transmitter. +- **Async `ERR(EFRAME)` fan-out** — if the dongle emits async framing + errors (tag 0), they're broadcast to all clients rather than swallowed. +- **Per-client TX cache** on `ClientSession` so the loopback fan-out + doesn't need the sender to re-buffer the payload. +- **Socket and serial file locking** (`fcntl.flock`) to prevent multiple + mux instances from fighting over the same dongle or socket path. +- **500 ms keepalive** matching `PROTOCOL.md §3.4` (2× margin on the + 1 s inactivity timer). Replaces the 5 s cadence from 0.x. +- **Clean ^C / SIGTERM shutdown** — per-client handler tasks are tracked + and cancelled on shutdown, so blocked `reader.read()` calls no longer + pin the event loop open. -### Fixed +### Removed -- Serial port opened without baud rate (defaulted to 9600 instead of 115200), - preventing connection to UART boards behind CP2102 bridges. +- Every v0.1 wire constant and the inline `cobs` + raw `serial.Serial` + frame codec. +- The single `pending_response: asyncio.Future` slot (fundamentally + incompatible with v1.0's per-tag concurrency). + +### Test coverage + +- 28 tests (up from 11): + - `test_tagmap.py` (6) — tag allocator behaviour. + - `test_intercept.py` (13) — every branch of the `decide()` matrix, + including the `ALREADY_MATCHED` / `LOCKED_MISMATCH` synthesis paths. + - `test_multiclient.py` (9) — asyncio-driven daemon dispatch tests: + concurrent tag routing, `SET_CONFIG` lock installation, TX loopback + to peer clients, lock release on disconnect, unsolicited-frame drop. ## 0.2.0 — 2026-04-08 diff --git a/README.md b/README.md index 561db7f..a722340 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,27 @@ # DongLoRa Mux (Python) USB multiplexer daemon — lets multiple applications share one DongLoRa -dongle simultaneously. +dongle simultaneously. Speaks the **DongLoRa Protocol v2** wire protocol +(see [`PROTOCOL.md`](https://github.com/donglora/firmware/blob/main/PROTOCOL.md)). ## What It Does -- Owns the USB serial connection exclusively -- Exposes a Unix domain socket (and optional TCP) speaking the same - COBS-framed protocol -- RxPacket frames broadcast to all connected clients -- StartRx/StopRx reference-counted across clients -- SetConfig locked once set (single client can change freely) +- Owns the USB serial connection exclusively (flock-protected). +- Exposes a Unix domain socket and optional TCP listener speaking the + DongLoRa Protocol v2 frame format. +- **Tag-correlated solicited responses** — concurrent commands from + multiple clients are routed back to the originator via a mux-allocated + device-tag map (`PROTOCOL.md §13.6`). +- **`RX` fan-out** — every over-the-air RX is broadcast to all clients. +- **TX loopback** (`§13.4`) — a successful TX from one client is + delivered to every other client as an `RX` event with + `origin = LOCAL_LOOPBACK`. +- **Smart `SET_CONFIG` arbitration** (`§13.2`) — the first successful + call locks the config; cross-client calls get `ALREADY_MATCHED` / + `LOCKED_MISMATCH` with the lock owner's params echoed. A single + connected client can still retune freely (scanner mode). +- **`RX_START` / `RX_STOP` ref-counting** — the radio stays in RX as + long as any client wants it. ## Running @@ -22,5 +33,23 @@ just run --tcp 5741 --port /dev/ttyACM0 # with options ## Depends On -- [`donglora`](https://github.com/donglora/client-py) — the Python client library - (protocol encoding, device discovery) +- [`donglora`](https://github.com/donglora/client-py) (>= 1.0) — the + Python client library. Provides the wire-level frame codec, event + parsers, and USB discovery helpers this mux is built on. + +## Relationship to `donglora-mux` (Rust) + +This package is a Python port of the Rust +[`donglora-mux`](https://crates.io/crates/donglora-mux) daemon. The two +implementations are **functionally equivalent** — they speak the same +DongLoRa Protocol v2 wire protocol, implement the same `§13.2` / `§13.4` multi-client +semantics, and are drop-in interchangeable in front of the same dongle. +Pick whichever fits your ops environment: + +- **Rust (`cargo install donglora-mux`)** — single binary, no runtime + deps, ~5 MB release build. Better for production / embedded Linux. +- **Python (`pip install donglora-mux`)** — easy to hack on, integrates + with a Python toolchain, useful when Rust isn't an option. + +If you find a behavioural divergence between the two, that's a bug — +please file it. diff --git a/donglora_mux/__init__.py b/donglora_mux/__init__.py index fd52fd9..38c0dd7 100644 --- a/donglora_mux/__init__.py +++ b/donglora_mux/__init__.py @@ -1,531 +1,70 @@ """DongLoRa USB Multiplexer — share one dongle with multiple applications. -Owns the USB serial connection and exposes a Unix domain socket that speaks -the same COBS-framed protocol. Clients connect via the socket (or just call -dl.connect() which auto-detects the mux) and get: +Owns the USB serial connection and exposes a Unix domain socket (and +optional TCP listener) that speaks the DongLoRa Protocol v2 wire format. Clients +connect and get: - • Solicited responses routed back to the commanding client - • RxPacket frames broadcast to ALL connected clients - • StartRx/StopRx reference-counted so the radio stays in RX as long as - anyone wants it - • SetConfig locked once set — subsequent clients must use the same config - (a single client can change config freely, like a scanner) +* **Tag-correlated solicited responses** — concurrent commands from + multiple clients are routed back to the originator via a mux-owned + device-tag allocator (``PROTOCOL.md §13.6``). +* **`RX` fan-out** — every `RX` packet is broadcast to all clients. +* **TX loopback** (``PROTOCOL.md §13.4``) — a successful over-the-air + TX from one client is delivered to every other connected client as + an `RX` event with ``origin = LOCAL_LOOPBACK``. +* **Smart `SET_CONFIG` arbitration** (``PROTOCOL.md §13.2``) — the + first successful call locks the config; cross-client calls get + `ALREADY_MATCHED` / `LOCKED_MISMATCH` with the lock owner echoed in + ``current``. +* **`RX_START` / `RX_STOP` ref-counting** — the radio stays in RX as + long as any client wants it. + +Usage:: -Usage: donglora-mux [--port /dev/ttyACM0] [--socket /tmp/donglora-mux.sock] [--tcp 5741] + +See ``PROTOCOL.md`` for the wire protocol and the Rust sibling +``donglora-mux`` (``cargo install donglora-mux``) for a drop-in +compatible implementation. """ from __future__ import annotations import argparse import asyncio -import contextlib import logging -import os import signal -from pathlib import Path - -import serial -from cobs import cobs import donglora as dl +from donglora_mux.daemon import MuxDaemon + +__all__ = ["MuxDaemon", "main"] log = logging.getLogger("donglora-mux") -# ── Protocol constants ─────────────────────────────────────────── - -TAG_RXPACKET = 2 -TAG_OK = 4 -TAG_ERROR = 5 -CMD_TAG_SET_CONFIG = 2 -CMD_TAG_START_RX = 3 -CMD_TAG_STOP_RX = 4 - -ERROR_INVALID_CONFIG = 0 - -# Keepalive: ping every 5s of idle, declare lost after 2s timeout -KEEPALIVE_INTERVAL = 5.0 -KEEPALIVE_TIMEOUT = 2.0 - - -def default_socket_path() -> str: - """Resolve the mux socket path in priority order.""" - env = os.environ.get("DONGLORA_MUX") - if env: - return env - xdg = os.environ.get("XDG_RUNTIME_DIR") - if xdg: - return os.path.join(xdg, "donglora", "mux.sock") - return "/tmp/donglora-mux.sock" - - -# ── Per-client state ───────────────────────────────────────────── - - -class Client: - """Tracks one connected client.""" - - _next_id = 0 - - def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter): - self.id = Client._next_id - Client._next_id += 1 - self.reader = reader - self.writer = writer - self.rx_interested = False # has this client called StartRx? - self.send_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=256) - self._sender_task: asyncio.Task | None = None - - @property - def label(self) -> str: - return f"client-{self.id}" - - def start_sender(self) -> None: - self._sender_task = asyncio.create_task(self._sender(), name=f"{self.label}-sender") - - async def _sender(self) -> None: - """Drain send_queue → socket. Runs until cancelled or write fails.""" - try: - while True: - frame = await self.send_queue.get() - self.writer.write(frame) - await self.writer.drain() - except (ConnectionError, asyncio.CancelledError): - pass - - async def enqueue(self, cobs_frame: bytes) -> None: - """Enqueue a COBS frame for sending. Drops oldest RxPacket on overflow.""" - try: - self.send_queue.put_nowait(cobs_frame) - except asyncio.QueueFull: - # Drop oldest to make room (best-effort for slow clients) - with contextlib.suppress(asyncio.QueueEmpty): - self.send_queue.get_nowait() - with contextlib.suppress(asyncio.QueueFull): - self.send_queue.put_nowait(cobs_frame) - - def close(self) -> None: - if self._sender_task: - self._sender_task.cancel() - with contextlib.suppress(Exception): - self.writer.close() - - -# ── Mux daemon ─────────────────────────────────────────────────── - - -class MuxDaemon: - def __init__(self, serial_port: str, socket_path: str, tcp_addr: tuple[str, int] | None = None): - self.serial_port = serial_port - self.socket_path = socket_path - self.tcp_addr = tcp_addr - self.ser: serial.Serial | None = None - self.clients: dict[int, Client] = {} - self.cmd_queue: asyncio.Queue[tuple[int, bytes]] = asyncio.Queue() - self.pending_response: asyncio.Future[bytes] | None = None - self.locked_config: bytes | None = None # raw RadioConfig bytes (13 bytes) once set - self._shutdown = asyncio.Event() - - # ── Serial I/O (blocking, run in threads) ──────────────────── - - def _serial_read_frame(self) -> bytes | None: - """Blocking read of one COBS frame from dongle.""" - assert self.ser is not None - return dl.read_frame(self.ser) - - def _serial_write(self, data: bytes) -> None: - """Blocking write to dongle.""" - assert self.ser is not None - self.ser.write(data) - self.ser.flush() - - # ── Dongle reader task ─────────────────────────────────────── - - async def dongle_reader(self) -> None: - """Read frames from the dongle, dispatch solicited vs broadcast.""" - consecutive_errors = 0 - max_retries = 3 - while not self._shutdown.is_set(): - try: - raw = await asyncio.to_thread(self._serial_read_frame) - except (serial.SerialException, OSError) as e: - consecutive_errors += 1 - if consecutive_errors >= max_retries: - log.error( - "Dongle read error (giving up after %d consecutive failures): %s", - consecutive_errors, - e, - ) - self._handle_dongle_disconnect() - return - log.warning("Dongle read glitch (%d/%d): %s", consecutive_errors, max_retries, e) - await asyncio.sleep(0.1) - continue - - consecutive_errors = 0 - - if raw is None: - continue - - tag = raw[0] - cobs_frame = dl.cobs_encode(raw) - - if tag == TAG_RXPACKET: - # Broadcast to all clients - for client in list(self.clients.values()): - await client.enqueue(cobs_frame) - else: - # Solicited response — resolve the pending future - if self.pending_response and not self.pending_response.done(): - self.pending_response.set_result(raw) - else: - log.warning("Unsolicited non-RxPacket response (tag %d) — dropped", tag) - - # ── Dongle writer task ─────────────────────────────────────── - - async def _keepalive_ping(self) -> bool: - """Send a Ping and wait for Pong. Returns True if dongle responded.""" - loop = asyncio.get_running_loop() - self.pending_response = loop.create_future() - - try: - ping_frame = dl.cobs_encode(dl.encode_command("Ping")) - await asyncio.to_thread(self._serial_write, ping_frame) - except (serial.SerialException, OSError): - self.pending_response = None - return False - - try: - await asyncio.wait_for(self.pending_response, timeout=KEEPALIVE_TIMEOUT) - return True - except asyncio.TimeoutError: - self.pending_response = None - return False - - async def dongle_writer(self) -> None: - """Pull commands from queue, send one at a time, route responses.""" - loop = asyncio.get_running_loop() - - while not self._shutdown.is_set(): - try: - client_id, raw_cmd = await asyncio.wait_for( - self.cmd_queue.get(), timeout=KEEPALIVE_INTERVAL - ) - except asyncio.TimeoutError: - log.debug("keepalive ping") - if not await self._keepalive_ping(): - log.error("Keepalive ping failed - dongle lost") - self._handle_dongle_disconnect() - return - continue - - # Check for StartRx/StopRx ref-counting intercept - intercepted = await self._maybe_intercept(client_id, raw_cmd) - if intercepted is not None: - # Send the synthetic response to the client - if client_id in self.clients: - await self.clients[client_id].enqueue(dl.cobs_encode(intercepted)) - continue - - # Create future for the solicited response - self.pending_response = loop.create_future() - - # Send COBS-encoded command to dongle - cobs_frame = dl.cobs_encode(raw_cmd) - try: - await asyncio.to_thread(self._serial_write, cobs_frame) - except (serial.SerialException, OSError) as e: - log.error("Dongle write error: %s", e) - self.pending_response.set_result(bytes([5, 1])) # Error(RadioBusy) - self._handle_dongle_disconnect() - return - - # Wait for dongle_reader to resolve the future - try: - response = await asyncio.wait_for(self.pending_response, timeout=10.0) - except asyncio.TimeoutError: - log.error("Dongle response timeout") - response = bytes([5, 1]) # Error(RadioBusy) - finally: - self.pending_response = None - - # Track state on successful forward - cmd_tag = raw_cmd[0] - resp_tag = response[0] if response else 0xFF - if cmd_tag == CMD_TAG_SET_CONFIG and resp_tag == TAG_OK: - self.locked_config = raw_cmd[1:] - log.info("Radio config locked: %s", self.locked_config.hex()) - elif cmd_tag == CMD_TAG_START_RX and resp_tag == TAG_OK: - if client_id in self.clients: - self.clients[client_id].rx_interested = True - elif cmd_tag == CMD_TAG_STOP_RX and resp_tag == TAG_OK: - if client_id in self.clients: - self.clients[client_id].rx_interested = False - - # Route response to commanding client (if still connected) - if client_id in self.clients: - await self.clients[client_id].enqueue(dl.cobs_encode(response)) - - # ── StartRx / StopRx reference counting ────────────────────── - - def _rx_interest_count(self) -> int: - return sum(1 for c in self.clients.values() if c.rx_interested) - - async def _maybe_intercept(self, client_id: int, raw_cmd: bytes) -> bytes | None: - """Intercept SetConfig/StartRx/StopRx for smart multiplexing. - - Returns a synthetic response if intercepted, None to forward normally. - """ - if not raw_cmd: - return None - cmd_tag = raw_cmd[0] - client = self.clients.get(client_id) - - if cmd_tag == CMD_TAG_SET_CONFIG: - config_bytes = raw_cmd[1:] # 13-byte RadioConfig - if len(self.clients) <= 1: - # Single client — allow free config changes (scanner mode) - # Forward to dongle, update locked_config on success - return None - if self.locked_config is None: - # First SetConfig with multiple clients — forward and lock - return None - if config_bytes == self.locked_config: - # Same config — reply Ok without hitting the dongle - log.debug( - "%s: SetConfig matches locked config — Ok", - client.label if client else f"id-{client_id}", - ) - return bytes([TAG_OK]) - # Different config — reject - log.warning( - "%s: SetConfig rejected (conflicts with locked config)", - client.label if client else f"id-{client_id}", - ) - return bytes([TAG_ERROR, ERROR_INVALID_CONFIG]) - - if cmd_tag == CMD_TAG_START_RX: - if client and client.rx_interested: - # Already interested — no-op, reply Ok - return bytes([TAG_OK]) - if self._rx_interest_count() > 0: - # Someone else is already receiving — just mark this client, reply Ok - if client: - client.rx_interested = True - return bytes([TAG_OK]) - # First interested client — forward to dongle (return None) - return None - - if cmd_tag == CMD_TAG_STOP_RX: - if client and not client.rx_interested: - # Wasn't interested — no-op, reply Ok - return bytes([TAG_OK]) - if client: - client.rx_interested = False - if self._rx_interest_count() > 0: - # Others still interested — don't stop, reply Ok - return bytes([TAG_OK]) - # Last interested client — forward to dongle (return None) - return None +def _parse_tcp_addr(value: str) -> tuple[str, int] | None: + if value.lower() == "none": return None - - # ── Client handling ────────────────────────────────────────── - - async def client_handler( - self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter - ) -> None: - client = Client(reader, writer) - self.clients[client.id] = client - client.start_sender() - log.info("%s connected (%d total)", client.label, len(self.clients)) - - try: - buf = b"" - while True: - data = await reader.read(4096) - if not data: - break - buf += data - # Extract complete COBS frames (delimited by 0x00) - while b"\x00" in buf: - idx = buf.index(b"\x00") - encoded = buf[:idx] - buf = buf[idx + 1 :] - if not encoded: - continue - try: - raw_cmd = cobs.decode(encoded) - except cobs.DecodeError: - log.warning("%s: bad COBS frame — skipped", client.label) - continue - await self.cmd_queue.put((client.id, raw_cmd)) - except (ConnectionError, asyncio.CancelledError): - pass - finally: - await self._remove_client(client) - - async def _remove_client(self, client: Client) -> None: - """Clean up after a disconnected client.""" - was_interested = client.rx_interested - client.close() - self.clients.pop(client.id, None) - log.info("%s disconnected (%d remain)", client.label, len(self.clients)) - - # If this client had RX interest and was the last one, stop RX - if was_interested and self._rx_interest_count() == 0: - log.info("Last RX-interested client gone — sending StopRx") - await self.cmd_queue.put((-1, bytes([CMD_TAG_STOP_RX]))) - - # Reset locked config when all clients are gone - if not self.clients and self.locked_config is not None: - log.info("All clients gone — config lock released") - self.locked_config = None - - # ── Dongle disconnect ──────────────────────────────────────── - - _dongle_lost: asyncio.Event - - def _handle_dongle_disconnect(self) -> None: - log.error("Dongle disconnected — will reconnect") - if self.pending_response and not self.pending_response.done(): - self.pending_response.set_result(bytes([5, 1])) # Error(RadioBusy) - if self.ser: - with contextlib.suppress(Exception): - self.ser.close() - self.ser = None - self._dongle_lost.set() - - # ── Dongle connection ────────────────────────────────────────── - - def _open_dongle(self) -> bool: - """Open and ping the dongle. Returns True on success.""" - try: - self.ser = serial.Serial(self.serial_port, 115200, timeout=2.0, exclusive=True) - self.ser.reset_input_buffer() - self._serial_write(dl.cobs_encode(dl.encode_command("Ping"))) - pong = self._serial_read_frame() - if not pong or pong[0] != 0: - log.warning("Dongle did not respond to Ping") - self.ser.close() - self.ser = None - return False - log.info("Dongle responded to Ping") - return True - except (serial.SerialException, OSError) as e: - log.warning("Could not open dongle: %s", e) - self.ser = None - return False - - # ── Main entry ─────────────────────────────────────────────── - - async def run(self) -> None: - self._dongle_lost = asyncio.Event() - - # Prepare socket path - sock_path = Path(self.socket_path) - sock_path.parent.mkdir(parents=True, exist_ok=True) - if sock_path.exists(): - sock_path.unlink() - - # Start servers (these stay up for the lifetime of the daemon) - server = await asyncio.start_unix_server(self.client_handler, path=str(sock_path)) - log.info("Unix socket listening on %s", self.socket_path) - - tcp_server = None - if self.tcp_addr: - tcp_server = await asyncio.start_server( - self.client_handler, - self.tcp_addr[0], - self.tcp_addr[1], - ) - log.info("TCP listening on %s:%d", *self.tcp_addr) - - # Connect/reconnect loop — servers stay up, dongle comes and goes - saved_config: bytes | None = None - had_rx_interest = False - - try: - while not self._shutdown.is_set(): - # Connect to dongle (retry until it appears) - log.info("Opening dongle on %s", self.serial_port) - while not self._shutdown.is_set(): - if await asyncio.to_thread(self._open_dongle): - break - await asyncio.sleep(2) - - if self._shutdown.is_set(): - break - - # Run dongle reader/writer until dongle is lost - self._dongle_lost.clear() - reader_task = asyncio.create_task(self.dongle_reader(), name="dongle-reader") - writer_task = asyncio.create_task(self.dongle_writer(), name="dongle-writer") - - # Re-establish radio state from previous session - if saved_config is not None: - log.info("Restoring radio config after reconnect") - await self.cmd_queue.put((-1, bytes([CMD_TAG_SET_CONFIG]) + saved_config)) - if had_rx_interest: - await self.cmd_queue.put((-1, bytes([CMD_TAG_START_RX]))) - saved_config = None - had_rx_interest = False - - # Wait for dongle loss or shutdown - shutdown_task = asyncio.create_task(self._shutdown.wait()) - lost_task = asyncio.create_task(self._dongle_lost.wait()) - await asyncio.wait( - [shutdown_task, lost_task], - return_when=asyncio.FIRST_COMPLETED, - ) - shutdown_task.cancel() - lost_task.cancel() - - # Tear down dongle tasks - reader_task.cancel() - writer_task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await asyncio.gather(reader_task, writer_task, return_exceptions=True) - - # Drain any pending commands with errors - while not self.cmd_queue.empty(): - try: - self.cmd_queue.get_nowait() - except asyncio.QueueEmpty: - break - - # Save radio state for restoration after reconnect - saved_config = self.locked_config - had_rx_interest = self._rx_interest_count() > 0 - self.locked_config = None - - if not self._shutdown.is_set(): - log.info("Reconnecting in 2 seconds...") - await asyncio.sleep(2) - finally: - # Cleanup - log.info("Shutting down...") - for client in list(self.clients.values()): - client.close() - server.close() - await server.wait_closed() - if tcp_server: - tcp_server.close() - await tcp_server.wait_closed() - if self.ser: - self.ser.close() - if sock_path.exists(): - sock_path.unlink() - log.info("Stopped.") - - -# ── CLI ────────────────────────────────────────────────────────── + if ":" in value: + host, _, port_str = value.rpartition(":") + return (host, int(port_str)) + return ("0.0.0.0", int(value)) def main() -> None: - parser = argparse.ArgumentParser(description="DongLoRa USB Multiplexer") - parser.add_argument("--port", "-p", default=None, help="Serial port (auto-detect if omitted)") - parser.add_argument("--socket", "-s", default=None, help="Unix socket path") + parser = argparse.ArgumentParser(description="DongLoRa USB Multiplexer (DongLoRa Protocol v2)") + parser.add_argument( + "--port", + "-p", + default=None, + help="Serial port (auto-detect by USB VID:PID if omitted)", + ) + parser.add_argument( + "--socket", + "-s", + default=None, + help="Unix socket path (default: $XDG_RUNTIME_DIR/donglora/mux.sock or /tmp/donglora-mux.sock)", + ) parser.add_argument( "--tcp", "-t", @@ -544,29 +83,17 @@ def main() -> None: port = args.port or dl.find_port() if not port: - log.info("Waiting for DongLoRa device...") + log.info("waiting for DongLoRa device...") port = dl.wait_for_device() - socket_path = args.socket or default_socket_path() - - tcp_addr = None - if args.tcp and args.tcp.lower() != "none": - if ":" in args.tcp: - host, _, p = args.tcp.rpartition(":") - tcp_addr = (host, int(p)) - else: - tcp_addr = ("0.0.0.0", int(args.tcp)) + socket_path = args.socket or dl.default_socket_path() + tcp_addr = _parse_tcp_addr(args.tcp) daemon = MuxDaemon(port, socket_path, tcp_addr=tcp_addr) loop = asyncio.new_event_loop() - - def _signal_handler() -> None: - daemon._shutdown.set() - for sig in (signal.SIGINT, signal.SIGTERM): - loop.add_signal_handler(sig, _signal_handler) - + loop.add_signal_handler(sig, daemon.request_shutdown) try: loop.run_until_complete(daemon.run()) finally: diff --git a/donglora_mux/daemon.py b/donglora_mux/daemon.py new file mode 100644 index 0000000..2da7330 --- /dev/null +++ b/donglora_mux/daemon.py @@ -0,0 +1,695 @@ +"""Core mux daemon — DongLoRa Protocol v2 tag-aware frame routing. + +Owns the USB serial connection and exposes Unix-domain + TCP listeners. +Every inbound client frame goes through a :class:`TagMapper` that +allocates a fresh device tag and remembers the owning +``(client, upstream_tag)`` so the dongle's reply can be routed back. +Async frames from the dongle (``tag == 0``: ``RX``, async +``ERR(EFRAME)``) fan out to every client. + +Additional spec-mandated behaviour: + +* **§13.2** — ``SET_CONFIG`` arbitration with ``APPLIED`` / + ``ALREADY_MATCHED`` / ``LOCKED_MISMATCH`` results and owner tracking. + Handled in :mod:`donglora_mux.intercept`. +* **§13.4** — TX loopback: on ``TX_DONE(TRANSMITTED)`` the mux + synthesizes an ``RX`` frame with ``origin = LOCAL_LOOPBACK`` and fans + it out to every client that didn't send the original TX. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import fcntl +import logging +import os +import time +from dataclasses import dataclass +from pathlib import Path + +import serial + +from donglora.commands import ( + TYPE_PING, + TYPE_RX_START, + TYPE_RX_STOP, + TYPE_SET_CONFIG, + TYPE_TX, +) +from donglora.errors import ErrorCode +from donglora.events import ( + TYPE_ERR, + TYPE_OK, + TYPE_RX, + TYPE_TX_DONE, + Owner, + RxEvent, + RxOrigin, + SetConfigResult, + SetConfigResultCode, + TxDone, + TxResult, + encode_err_payload, + parse_ok_payload, +) +from donglora.frame import Frame, FrameError, decode_frame, encode_frame +from donglora_mux.intercept import DecisionKind, MuxState, decide, rx_interest_count +from donglora_mux.session import ClientSession +from donglora_mux.tagmap import TagMapper + +log = logging.getLogger("donglora-mux") + +# ── Timing constants ─────────────────────────────────────────────── + +# Spec §3.4 inactivity timer is 1 s; ping at 500 ms for 2x margin. +KEEPALIVE_INTERVAL = 0.5 + +# How long to wait between dongle reconnect attempts. Reset on a clean +# long-running session. +RECONNECT_DELAY = 2.0 + +# Consecutive serial read errors before declaring the dongle lost. +MAX_CONSECUTIVE_ERRORS = 3 + +# Sentinel "client id" used by mux-internal commands (keepalive, synthetic +# RX_STOP on last-listener-disconnect). Never collides with a real client +# because ``itertools.count`` in session.py starts at 0 and climbs. +SYNTHETIC_CLIENT_ID = -1 + + +# ── Writer work queue ────────────────────────────────────────────── + + +@dataclass +class _ForwardWork: + client_id: int + type_id: int + upstream_tag: int + payload: bytes + + +@dataclass +class _KeepaliveWork: + pass + + +_WriterWork = _ForwardWork | _KeepaliveWork + + +# ── The daemon ───────────────────────────────────────────────────── + + +class MuxDaemon: + def __init__( + self, + serial_port: str, + socket_path: str, + tcp_addr: tuple[str, int] | None = None, + ) -> None: + self.serial_port = serial_port + self.socket_path = socket_path + self.tcp_addr = tcp_addr + self._shutdown = asyncio.Event() + + # Shared state. Accessed from the reader, writer, and each + # per-client task; asyncio is single-threaded so a plain lock + # isn't needed for the shared dicts, but the TagMapper and the + # MuxState are accessed from async methods via direct refs. + self.sessions: dict[int, ClientSession] = {} + self.tagmap = TagMapper() + self.state = MuxState() + + # Work queue: client handlers → dongle writer. + self._work: asyncio.Queue[_WriterWork] = asyncio.Queue(maxsize=64) + + # Set by the reader/writer tasks when the dongle stops talking. + self._dongle_lost: asyncio.Event = asyncio.Event() + + # Held file handles for the two flocks (socket + serial). Their + # descriptors need to stay open for the life of the daemon — + # closing frees the flock automatically. + self._socket_lock_fd: int | None = None + self._serial_lock_fd: int | None = None + + # Per-connection accept-handler tasks. Tracked so the shutdown + # path can cancel them en masse — otherwise a client blocked in + # ``reader.read()`` keeps the event loop alive indefinitely. + self._client_tasks: set[asyncio.Task] = set() + + # ── Entry ───────────────────────────────────────────────────── + + def request_shutdown(self) -> None: + """Idempotent; safe to call from a signal handler.""" + self._shutdown.set() + + async def run(self) -> None: + """Run the daemon to completion. Returns when shutdown is signalled.""" + self._acquire_socket_lock() + sock_path = Path(self.socket_path) + sock_path.parent.mkdir(parents=True, exist_ok=True) + if sock_path.exists(): + log.info("removing stale socket %s", self.socket_path) + with contextlib.suppress(FileNotFoundError): + sock_path.unlink() + + unix_server = await asyncio.start_unix_server(self._accept_client, path=str(sock_path)) + log.info("Unix socket listening on %s", self.socket_path) + + tcp_server: asyncio.AbstractServer | None = None + if self.tcp_addr is not None: + tcp_server = await asyncio.start_server( + self._accept_client, self.tcp_addr[0], self.tcp_addr[1] + ) + log.info("TCP listening on %s:%d", *self.tcp_addr) + + try: + await self._reconnect_loop() + finally: + log.info("shutting down...") + # Stop accepting new clients first so nothing new piles up + # while we're tearing down in-flight ones. + unix_server.close() + if tcp_server is not None: + tcp_server.close() + # Cancel every accepted-client handler. Without this, a + # client blocked in ``reader.read()`` pins the event loop + # open and ^C appears to hang. + for task in list(self._client_tasks): + task.cancel() + if self._client_tasks: + await asyncio.gather(*self._client_tasks, return_exceptions=True) + await unix_server.wait_closed() + if tcp_server is not None: + await tcp_server.wait_closed() + with contextlib.suppress(FileNotFoundError): + sock_path.unlink() + self._release_locks() + log.info("stopped.") + + # ── Dongle reconnect loop ───────────────────────────────────── + + async def _reconnect_loop(self) -> None: + while not self._shutdown.is_set(): + log.info("opening dongle on %s", self.serial_port) + ser = await self._open_dongle_with_retry() + if ser is None: # shutdown requested mid-retry + return + + self._dongle_lost.clear() + reader_task = asyncio.create_task(self._dongle_reader(ser), name="dongle-reader") + writer_task = asyncio.create_task(self._dongle_writer(ser), name="dongle-writer") + keepalive_task = asyncio.create_task(self._keepalive_loop(), name="keepalive") + + # Wait for either loss or shutdown. + await self._race([self._dongle_lost.wait(), self._shutdown.wait()]) + + reader_task.cancel() + writer_task.cancel() + keepalive_task.cancel() + await asyncio.gather(reader_task, writer_task, keepalive_task, return_exceptions=True) + + # Drain queued work. + while not self._work.empty(): + with contextlib.suppress(asyncio.QueueEmpty): + self._work.get_nowait() + + # The reader/writer tore down their view of `tagmap`'s + # pending entries; reset so reconnect starts clean. + self.tagmap = TagMapper() + + # Close the serial port. + with contextlib.suppress(Exception): + ser.close() + + if not self._shutdown.is_set(): + log.info("reconnecting in %.1fs...", RECONNECT_DELAY) + try: + await asyncio.wait_for(self._shutdown.wait(), timeout=RECONNECT_DELAY) + return # shutdown fired + except asyncio.TimeoutError: + pass + + async def _open_dongle_with_retry(self) -> serial.Serial | None: + while not self._shutdown.is_set(): + ser = await asyncio.to_thread(self._open_and_ping) + if ser is not None: + return ser + try: + await asyncio.wait_for(self._shutdown.wait(), timeout=RECONNECT_DELAY) + return None # shutdown fired + except asyncio.TimeoutError: + continue + return None + + def _open_and_ping(self) -> serial.Serial | None: + """Blocking: acquire the serial flock, open the port, PING. + + Returns the serial handle on success or ``None`` on failure. + """ + try: + self._acquire_serial_lock() + except OSError as e: + log.warning("serial port %s is locked by another process: %s", self.serial_port, e) + return None + + try: + ser = serial.Serial(self.serial_port, 115_200, timeout=2.0, exclusive=True) + except (serial.SerialException, OSError) as e: + log.warning("could not open %s: %s", self.serial_port, e) + self._release_serial_lock() + return None + + try: + ser.reset_input_buffer() + ser.write(encode_frame(TYPE_PING, 1, b"")) + ser.flush() + # Read frames until we see any OK (the PING response). + deadline = time.monotonic() + 2.0 + buf = bytearray() + while time.monotonic() < deadline: + remaining = max(0.05, deadline - time.monotonic()) + ser.timeout = remaining + chunk = ser.read(64) + if not chunk: + continue + buf.extend(chunk) + for frame in _drain_frames(buf): + if frame.type_id == TYPE_OK: + log.info("dongle responded to PING") + ser.timeout = None + return ser + log.warning("dongle did not respond to PING within 2 s") + except (serial.SerialException, OSError, FrameError) as e: + log.warning("PING round-trip failed: %s", e) + + with contextlib.suppress(Exception): + ser.close() + self._release_serial_lock() + return None + + # ── Dongle I/O tasks ────────────────────────────────────────── + + async def _dongle_reader(self, ser: serial.Serial) -> None: + """Pump bytes from the dongle; dispatch decoded frames.""" + buf = bytearray() + errors = 0 + while not self._shutdown.is_set() and not self._dongle_lost.is_set(): + try: + chunk = await asyncio.to_thread(_serial_read_chunk, ser, 512) + except (serial.SerialException, OSError) as e: + errors += 1 + if errors >= MAX_CONSECUTIVE_ERRORS: + log.error("dongle read error (giving up after %d): %s", errors, e) + self._dongle_lost.set() + return + log.warning("dongle read glitch (%d/%d): %s", errors, MAX_CONSECUTIVE_ERRORS, e) + await asyncio.sleep(0.1) + continue + if not chunk: + # Timeout — harmless in reader pump; keep reading. + await asyncio.sleep(0) + continue + errors = 0 + buf.extend(chunk) + + # Drain every complete frame we can see. + frame_errors = 0 + for frame_or_err in _drain_frames_or_errors(buf): + if isinstance(frame_or_err, Frame): + await self._dispatch_device_frame(frame_or_err) + else: + frame_errors += 1 + if frame_errors: + log.warning("%d corrupt frame(s) from dongle (CRC/COBS)", frame_errors) + await self._fanout_async_err(ErrorCode.EFRAME) + + async def _dispatch_device_frame(self, frame: Frame) -> None: + # Async events (tag 0): fan out verbatim. + if frame.tag == 0: + wire = encode_frame(frame.type_id, 0, frame.payload) + for session in self.sessions.values(): + session.enqueue(wire) + return + + # Tagged: look up pending mapping. + pending = self.tagmap.peek(frame.tag) + if pending is None: + log.warning( + "unsolicited frame type=0x%02X tag=%d — no pending mapping", + frame.type_id, + frame.tag, + ) + return + + # TX's two-phase completion: intermediate OK keeps the mapping + # alive; only TX_DONE / ERR frees it. + terminal = not (pending.cmd_type == TYPE_TX and frame.type_id == TYPE_OK) + if terminal: + self.tagmap.take(frame.tag) + + # Install SET_CONFIG lock on APPLIED/MINE. + if pending.cmd_type == TYPE_SET_CONFIG and frame.type_id == TYPE_OK: + try: + parsed = parse_ok_payload(TYPE_SET_CONFIG, frame.payload) + if ( + isinstance(parsed, SetConfigResult) + and parsed.result == SetConfigResultCode.APPLIED + and parsed.owner == Owner.MINE + ): + self.state.locked = (pending.client_id, parsed.current) + log.info("SET_CONFIG applied by client %d — lock installed", pending.client_id) + except Exception as e: # pragma: no cover — parser errors are logged, not fatal + log.debug("parse_ok_payload(SET_CONFIG) failed: %s", e) + + # Track RX interest on RX_START/RX_STOP OK. + if frame.type_id == TYPE_OK: + session = self.sessions.get(pending.client_id) + if session is not None: + if pending.cmd_type == TYPE_RX_START: + session.rx_interested = True + elif pending.cmd_type == TYPE_RX_STOP: + session.rx_interested = False + + # TX loopback: on TX_DONE(TRANSMITTED), synthesize an RX to every OTHER + # client so peers see their neighbour's transmission (spec §13.4). + tx_loopback_bytes: bytes | None = None + if pending.cmd_type == TYPE_TX and frame.type_id == TYPE_TX_DONE: + try: + td = TxDone.decode(frame.payload) + session = self.sessions.get(pending.client_id) + if session is not None: + data = session.tx_cache.pop(pending.upstream_tag, None) + if data is not None and td.result == TxResult.TRANSMITTED: + tx_loopback_bytes = data + except Exception as e: # pragma: no cover + log.debug("TxDone.decode failed: %s", e) + + # Route response to owning client with upstream tag restored. + wire = encode_frame(frame.type_id, pending.upstream_tag, frame.payload) + owning = self.sessions.get(pending.client_id) + if owning is not None: + owning.enqueue(wire) + + if tx_loopback_bytes is not None: + rx = RxEvent( + rssi_dbm=0.0, + snr_db=0.0, + freq_err_hz=0, + timestamp_us=_now_us(), + crc_valid=True, + packets_dropped=0, + origin=RxOrigin.LOCAL_LOOPBACK, + data=tx_loopback_bytes, + ) + loop_wire = encode_frame(TYPE_RX, 0, rx.encode()) + for cid, session in self.sessions.items(): + if cid != pending.client_id: + session.enqueue(loop_wire) + + async def _fanout_async_err(self, code: ErrorCode) -> None: + wire = encode_frame(TYPE_ERR, 0, encode_err_payload(code)) + for session in self.sessions.values(): + session.enqueue(wire) + + async def _dongle_writer(self, ser: serial.Serial) -> None: + while not self._shutdown.is_set() and not self._dongle_lost.is_set(): + try: + work = await self._work.get() + except asyncio.CancelledError: + return + + if isinstance(work, _KeepaliveWork): + # Synthetic mapping keeps the reply absorption clean. + device_tag = self.tagmap.alloc(SYNTHETIC_CLIENT_ID, 0, TYPE_PING) + wire = encode_frame(TYPE_PING, device_tag, b"") + else: + # If this is a TX, stash the payload (sans flags byte) so + # the reader can fan out on TX_DONE(TRANSMITTED). + if work.type_id == TYPE_TX and len(work.payload) > 1: + session = self.sessions.get(work.client_id) + if session is not None: + session.tx_cache[work.upstream_tag] = work.payload[1:] + device_tag = self.tagmap.alloc(work.client_id, work.upstream_tag, work.type_id) + wire = encode_frame(work.type_id, device_tag, work.payload) + + try: + await asyncio.to_thread(_serial_write_all, ser, wire) + except (serial.SerialException, OSError) as e: + log.error("dongle write error: %s", e) + self._dongle_lost.set() + return + + async def _keepalive_loop(self) -> None: + while not self._shutdown.is_set() and not self._dongle_lost.is_set(): + try: + await asyncio.wait_for(self._shutdown.wait(), timeout=KEEPALIVE_INTERVAL) + return + except asyncio.TimeoutError: + pass + if self._dongle_lost.is_set(): + return + # Writer is behind if full; skip this tick rather than block. + with contextlib.suppress(asyncio.QueueFull): + self._work.put_nowait(_KeepaliveWork()) + + # ── Client accept / handler ─────────────────────────────────── + + async def _accept_client( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + # Disable Nagle on TCP to keep small DongLoRa Protocol frames snappy. + sock = writer.get_extra_info("socket") + if sock is not None: + with contextlib.suppress(OSError): + import socket as _socket # local import; only needed here + + sock.setsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1) + + session = ClientSession() + self.sessions[session.id] = session + log.info("%s connected (%d total)", session.label, len(self.sessions)) + + # Register this task so the daemon's shutdown path can cancel + # us out of the blocking reader.read(). + task = asyncio.current_task() + if task is not None: + self._client_tasks.add(task) + + writer_task = asyncio.create_task( + self._client_writer(session, writer), name=f"{session.label}-writer" + ) + try: + await self._client_reader(session, reader) + except asyncio.CancelledError: + # Shutdown path cancelled us — fall through to cleanup. + pass + finally: + writer_task.cancel() + with contextlib.suppress(BaseException): + await writer_task + with contextlib.suppress(Exception): + writer.close() + await self._remove_client(session) + if task is not None: + self._client_tasks.discard(task) + + async def _client_reader(self, session: ClientSession, reader: asyncio.StreamReader) -> None: + buf = bytearray() + while True: + try: + chunk = await reader.read(4096) + except (ConnectionError, OSError) as e: + log.debug("%s reader exit: %s", session.label, e) + return + if not chunk: + log.debug("%s reader exit: EOF", session.label) + return + buf.extend(chunk) + for frame_or_err in _drain_frames_or_errors(buf): + if isinstance(frame_or_err, Frame): + await self._handle_client_frame(session, frame_or_err) + # Corrupt client frame: silently drop. The firmware pattern is + # ERR(EFRAME, tag=0) back to the client, but the tag is + # unrecoverable here so best-effort ignore. + + async def _handle_client_frame(self, session: ClientSession, frame: Frame) -> None: + decision = decide(frame.type_id, frame.payload, session.id, self.state, self.sessions) + if decision.kind == DecisionKind.SYNTHESIZE: + assert decision.type_id is not None and decision.payload is not None + wire = encode_frame(decision.type_id, frame.tag, decision.payload) + session.enqueue(wire) + return + if decision.kind == DecisionKind.DROP: + return + + # FORWARD — enqueue work for the dongle writer. + work = _ForwardWork( + client_id=session.id, + type_id=frame.type_id, + upstream_tag=frame.tag, + payload=frame.payload, + ) + try: + await asyncio.wait_for(self._work.put(work), timeout=2.0) + except asyncio.TimeoutError: + # Writer is stalled (dongle disconnected) — synthesize EBUSY + # back to the client with the upstream tag so their session + # doesn't hang forever waiting. + wire = encode_frame(TYPE_ERR, frame.tag, encode_err_payload(ErrorCode.EBUSY)) + session.enqueue(wire) + + async def _client_writer(self, session: ClientSession, writer: asyncio.StreamWriter) -> None: + while True: + try: + frame = await session.next_outgoing() + except asyncio.CancelledError: + return + try: + writer.write(frame) + await writer.drain() + except (ConnectionError, OSError) as e: + log.debug("%s writer exit: %s", session.label, e) + return + + async def _remove_client(self, session: ClientSession) -> None: + was_interested = session.rx_interested + self.sessions.pop(session.id, None) + self.tagmap.drop_client(session.id) + + # Release config lock if this client held it. + if self.state.locked is not None and self.state.locked[0] == session.id: + log.info("config lock released (owner %s disconnected)", session.label) + self.state.locked = None + + if session.drops: + log.info( + "%s disconnected (%d remain, %d dropped frames)", + session.label, + len(self.sessions), + session.drops, + ) + else: + log.info("%s disconnected (%d remain)", session.label, len(self.sessions)) + + # If this was the last RX-interested client, tell the dongle to stop. + if was_interested and rx_interest_count(self.sessions) == 0: + with contextlib.suppress(asyncio.QueueFull): + self._work.put_nowait( + _ForwardWork( + client_id=SYNTHETIC_CLIENT_ID, + type_id=TYPE_RX_STOP, + upstream_tag=0, + payload=b"", + ) + ) + + # ── File locking ────────────────────────────────────────────── + + def _acquire_socket_lock(self) -> None: + lock_path = f"{self.socket_path}.lock" + Path(lock_path).parent.mkdir(parents=True, exist_ok=True) + fd = os.open(lock_path, os.O_CREAT | os.O_WRONLY, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as e: + os.close(fd) + raise SystemExit( + f"another donglora-mux is already running (lock held on {lock_path}): {e}" + ) from e + self._socket_lock_fd = fd + + def _acquire_serial_lock(self) -> None: + lock_path = f"/tmp/donglora-serial-{self.serial_port.replace('/', '_')}.lock" + fd = os.open(lock_path, os.O_CREAT | os.O_WRONLY, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + os.close(fd) + raise + # Close any previous serial lock fd (reconnect cycle). + if self._serial_lock_fd is not None: + os.close(self._serial_lock_fd) + self._serial_lock_fd = fd + + def _release_serial_lock(self) -> None: + if self._serial_lock_fd is not None: + os.close(self._serial_lock_fd) + self._serial_lock_fd = None + + def _release_locks(self) -> None: + self._release_serial_lock() + if self._socket_lock_fd is not None: + os.close(self._socket_lock_fd) + self._socket_lock_fd = None + + # ── Utilities ───────────────────────────────────────────────── + + @staticmethod + async def _race(aws: list) -> None: + tasks = [asyncio.ensure_future(a) for a in aws] + try: + _done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + for t in pending: + t.cancel() + finally: + for t in tasks: + with contextlib.suppress(BaseException): + await t + + +# ── Module-level helpers ─────────────────────────────────────────── + + +def _now_us() -> int: + return int(time.time() * 1_000_000) + + +def _serial_read_chunk(ser: serial.Serial, n: int) -> bytes: + """Read up to ``n`` bytes. Blocking — run in a thread.""" + ser.timeout = 0.1 # short poll so the reader task can check shutdown + return ser.read(n) + + +def _serial_write_all(ser: serial.Serial, data: bytes) -> None: + ser.write(data) + ser.flush() + + +def _drain_frames(buf: bytearray): + """Yield every complete :class:`Frame` in *buf*; advances *buf* in place. + + Raises on corrupt frames. Used by the bootstrap PING reader where we + want to bubble up the error. + """ + while True: + try: + idx = buf.index(0) + except ValueError: + return + encoded = bytes(buf[:idx]) + del buf[: idx + 1] + if not encoded: + continue + yield decode_frame(encoded) + + +def _drain_frames_or_errors(buf: bytearray): + """Yield every complete frame (or :class:`FrameError`) in *buf*. + + Used by the long-running reader, which tolerates occasional + corruption by counting + reporting rather than exploding. + """ + while True: + try: + idx = buf.index(0) + except ValueError: + return + encoded = bytes(buf[:idx]) + del buf[: idx + 1] + if not encoded: + continue + try: + yield decode_frame(encoded) + except FrameError as e: + yield e diff --git a/donglora_mux/intercept.py b/donglora_mux/intercept.py new file mode 100644 index 0000000..9aaccce --- /dev/null +++ b/donglora_mux/intercept.py @@ -0,0 +1,200 @@ +"""Mux-level command interception (spec §13.2 / §13.4). + +Certain commands can't be blindly forwarded when multiple clients share +one dongle. The mux absorbs them locally and synthesizes a response so +the dongle isn't perturbed. Relevant cases: + +* **``SET_CONFIG``** (§13.2) — the first successful call locks the + config; subsequent calls from *other* clients get + ``ALREADY_MATCHED`` / ``LOCKED_MISMATCH`` with ``owner = OTHER``. +* **``RX_START`` / ``RX_STOP``** — reference-counted. The mux keeps RX + running until the last interested client stops. + +Mirrors ``mux-rs/src/intercept.rs``. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from enum import Enum, auto + +from donglora.commands import ( + TYPE_PING, + TYPE_RX_START, + TYPE_RX_STOP, + TYPE_SET_CONFIG, + TYPE_TX, +) +from donglora.errors import ErrorCode +from donglora.events import ( + TYPE_ERR, + TYPE_OK, + Owner, + SetConfigResult, + SetConfigResultCode, + encode_err_payload, +) +from donglora.modulation import Modulation, decode_modulation, encode_modulation +from donglora_mux.session import ClientSession + +log = logging.getLogger("donglora-mux") + + +class DecisionKind(Enum): + FORWARD = auto() + SYNTHESIZE = auto() + DROP = auto() + + +@dataclass(frozen=True) +class Decision: + """Outcome of :func:`decide` for a single command frame.""" + + kind: DecisionKind + # Only set when kind == SYNTHESIZE. + type_id: int | None = None + payload: bytes | None = None + + @classmethod + def forward(cls) -> Decision: + return cls(kind=DecisionKind.FORWARD) + + @classmethod + def synthesize(cls, type_id: int, payload: bytes) -> Decision: + return cls(kind=DecisionKind.SYNTHESIZE, type_id=type_id, payload=payload) + + @classmethod + def drop(cls) -> Decision: + return cls(kind=DecisionKind.DROP) + + +class MuxState: + """Mux-level state shared across sessions.""" + + __slots__ = ("locked",) + + def __init__(self) -> None: + # ``None`` when no client holds the config lock, else + # ``(client_id, current_modulation)``. + self.locked: tuple[int, Modulation] | None = None + + +def rx_interest_count(sessions: dict[int, ClientSession]) -> int: + return sum(1 for s in sessions.values() if s.rx_interested) + + +def decide( + type_id: int, + payload: bytes, + client_id: int, + state: MuxState, + sessions: dict[int, ClientSession], +) -> Decision: + """Decide what to do with an incoming client command. + + Returns :attr:`Decision.forward` to pass it to the dongle (after the + caller rewrites the tag), :meth:`Decision.synthesize` to reply + directly to the client without touching the dongle, or + :meth:`Decision.drop` to silently swallow (reserved). + + Unknown commands forward: the firmware returns ``ERR(EUNKNOWN_CMD)`` + with the echoed tag, which keeps the mux forward-compatible with + minor-version command additions. + """ + if type_id == TYPE_SET_CONFIG: + # Parse the modulation once for the lock comparison. A malformed + # payload goes through and lets the firmware reject it. + try: + requested = decode_modulation(payload) + except Exception: + return Decision.forward() + return _decide_set_config(requested, client_id, state, sessions) + if type_id == TYPE_RX_START: + return _decide_rx_start(client_id, sessions) + if type_id == TYPE_RX_STOP: + return _decide_rx_stop(client_id, sessions) + # PING, GET_INFO, TX: forward unchanged. + _ = (TYPE_PING, TYPE_TX) # referenced for parity with mux-rs pattern + return Decision.forward() + + +def _decide_set_config( + requested: Modulation, + client_id: int, + state: MuxState, + sessions: dict[int, ClientSession], +) -> Decision: + # Single client: unconditionally forward (scanner mode — a lone + # client can re-tune freely). + if len(sessions) <= 1: + return Decision.forward() + + if state.locked is None: + # First SET_CONFIG with multiple clients — forward; the daemon + # installs the lock once APPLIED comes back. + return Decision.forward() + + owner, current = state.locked + if owner == client_id: + # Lock owner is reconfiguring — forward; daemon updates lock on APPLIED. + return Decision.forward() + + # Another client holds the lock. Synthesize an OK with the + # appropriate result code. + matches = _modulations_match(requested, current) + code = SetConfigResultCode.ALREADY_MATCHED if matches else SetConfigResultCode.LOCKED_MISMATCH + result = SetConfigResult(result=code, owner=Owner.OTHER, current=current) + label = sessions[client_id].label if client_id in sessions else f"id-{client_id}" + log.debug("%s: SET_CONFIG %s (owner=OTHER) — synthesized", label, code.name) + return Decision.synthesize(type_id=TYPE_OK, payload=result.encode()) + + +def _decide_rx_start(client_id: int, sessions: dict[int, ClientSession]) -> Decision: + client = sessions.get(client_id) + if client is None: + return Decision.forward() + if client.rx_interested: + # Already in RX — synthesize empty OK. + return Decision.synthesize(type_id=TYPE_OK, payload=b"") + if rx_interest_count(sessions) > 0: + # Others already have RX running on the dongle; just mark this + # client interested and synthesize OK. + client.rx_interested = True + return Decision.synthesize(type_id=TYPE_OK, payload=b"") + # First interested client — turn RX on at the dongle. + return Decision.forward() + + +def _decide_rx_stop(client_id: int, sessions: dict[int, ClientSession]) -> Decision: + client = sessions.get(client_id) + if client is None: + return Decision.forward() + if not client.rx_interested: + # Wasn't listening — nothing to do. + return Decision.synthesize(type_id=TYPE_OK, payload=b"") + client.rx_interested = False + if rx_interest_count(sessions) > 0: + # Others still interested — don't stop RX. + return Decision.synthesize(type_id=TYPE_OK, payload=b"") + # Last interested client — forward so the dongle actually stops. + return Decision.forward() + + +def _modulations_match(a: Modulation, b: Modulation) -> bool: + """Byte-wise equality on the encoded wire form. + + Spec definition of "matches" is "same params on the wire" — this + dodges any Python-level equality differences between dataclass + instances that share the same wire representation but differ in + default-valued fields. + """ + try: + return encode_modulation(a) == encode_modulation(b) + except Exception: + return False + + +def synthesize_err(code: ErrorCode | int) -> tuple[int, bytes]: + """Build an ``ERR(code)`` synthesis payload (type_id + payload bytes).""" + return (TYPE_ERR, encode_err_payload(code)) diff --git a/donglora_mux/session.py b/donglora_mux/session.py new file mode 100644 index 0000000..673eead --- /dev/null +++ b/donglora_mux/session.py @@ -0,0 +1,77 @@ +"""Per-client session state and bounded send queue. + +Each connected client gets a :class:`ClientSession` tracking its id, +RX-interest flag, a per-client TX cache (used by the §13.4 loopback +fan-out), and a bounded asyncio queue of outgoing wire-encoded frames. + +Mirrors ``mux-rs/src/session.rs``. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import itertools +import logging + +log = logging.getLogger("donglora-mux") + +# Bounded send queue capacity per client. Matches mux-rs SEND_QUEUE_CAP. +SEND_QUEUE_CAP = 256 + +# Emit a drop-count summary after this many drops per client — a flood of +# RX while the client is slow shouldn't produce one log line per frame. +DROP_LOG_EVERY = 32 + +# Monotonic client-id source. Shared across every session in the process. +_next_client_id = itertools.count() + + +class ClientSession: + """State carried for a single connected client. + + Does NOT own the underlying stream; the daemon spawns a per-client + reader/writer pair of asyncio tasks that do I/O directly. This type + is the shared rendez-vous point those tasks enqueue frames through. + """ + + __slots__ = ("_queue", "drops", "id", "rx_interested", "tx_cache") + + def __init__(self) -> None: + self.id: int = next(_next_client_id) + self.rx_interested: bool = False + # Keyed by the client's upstream tag. Populated when a TX is + # forwarded to the dongle; consumed on TX_DONE(TRANSMITTED) by + # the reader so it can synthesize a loopback RX for other + # clients without asking the sender to re-buffer the payload. + self.tx_cache: dict[int, bytes] = {} + self._queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=SEND_QUEUE_CAP) + self.drops: int = 0 + + @property + def label(self) -> str: + return f"client-{self.id}" + + async def next_outgoing(self) -> bytes: + """Await the next frame to send. Intended for the client-writer task.""" + return await self._queue.get() + + def enqueue(self, frame: bytes) -> None: + """Best-effort non-blocking enqueue of a wire frame. + + Full queue → drop the frame, bump the counter, occasionally log. + This prevents one slow client from back-pressuring the whole mux. + """ + try: + self._queue.put_nowait(frame) + except asyncio.QueueFull: + self.drops += 1 + if self.drops == 1 or self.drops % DROP_LOG_EVERY == 0: + log.warning("%s: send queue full, dropped %d frames so far", self.label, self.drops) + + def close(self, writer: asyncio.StreamWriter | None = None) -> None: + """Best-effort close. The daemon normally drives shutdown via + task cancellation + stream close; this is a safety net.""" + if writer is not None: + with contextlib.suppress(Exception): + writer.close() diff --git a/donglora_mux/tagmap.py b/donglora_mux/tagmap.py new file mode 100644 index 0000000..c687f2b --- /dev/null +++ b/donglora_mux/tagmap.py @@ -0,0 +1,75 @@ +"""Device-tag ↔ (client, upstream_tag) mapping. + +DongLoRa Protocol v2 identifies each request/response pair with a 16-bit tag chosen +by the host. The mux sits between N clients and one dongle, so each +client picks its own tags but those tag spaces overlap. The mux reserves +a fresh *device tag* for every forwarded command, remembers which client +owned the originating upstream tag, and rewrites tags in both directions. + +Mirrors ``mux-rs/src/daemon.rs::TagMapper``. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(slots=True) +class Pending: + """A command the mux has forwarded to the dongle and is waiting on.""" + + client_id: int + upstream_tag: int + cmd_type: int + + +class TagMapper: + """Monotonic u16 tag allocator with collision probing.""" + + def __init__(self) -> None: + self._next: int = 1 # 0 is reserved for async device→host events + self._pending: dict[int, Pending] = {} + + def alloc(self, client_id: int, upstream_tag: int, cmd_type: int) -> int: + """Allocate a fresh device tag and record the mapping. + + Skips 0 (reserved for async events) and wraps at ``u16::MAX``. + On collision with an in-flight tag, probes forward linearly. In + practice the pending map is small (tens of entries at most) so + the probe is cheap. + """ + while True: + tag = self._next + self._next = (self._next + 1) & 0xFFFF + if self._next == 0: + self._next = 1 + if tag == 0 or tag in self._pending: + continue + self._pending[tag] = Pending( + client_id=client_id, upstream_tag=upstream_tag, cmd_type=cmd_type + ) + return tag + + def peek(self, device_tag: int) -> Pending | None: + """Return the pending mapping without removing it. + + Used for TX's two-phase completion: the intermediate ``OK`` + shouldn't free the slot because ``TX_DONE`` will land on the + same tag later. + """ + return self._pending.get(device_tag) + + def take(self, device_tag: int) -> Pending | None: + """Final-resolve the mapping (remove + return).""" + return self._pending.pop(device_tag, None) + + def drop_client(self, client_id: int) -> None: + """Remove every in-flight mapping owned by this client. + + Called on disconnect so stale replies from the dongle don't get + routed to dead file descriptors. + """ + self._pending = {tag: p for tag, p in self._pending.items() if p.client_id != client_id} + + def __len__(self) -> int: + return len(self._pending) diff --git a/pyproject.toml b/pyproject.toml index 6d73f24..4abe907 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "donglora-mux" -version = "0.2.0" +version = "1.0.0" description = "DongLoRa USB Multiplexer — share one dongle with multiple applications" license = "MIT" requires-python = ">=3.10" @@ -8,16 +8,18 @@ readme = "README.md" authors = [{ name = "Stephen Waits", email = "steve@waits.net" }] keywords = ["lora", "radio", "usb", "multiplexer"] classifiers = [ - "Development Status :: 3 - Alpha", + "Development Status :: 4 - Beta", "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3", "Topic :: Communications :: Ham Radio", "Topic :: System :: Hardware", ] dependencies = [ - "donglora", - "cobs", - "pyserial", + # DongLoRa Protocol v2 wire primitives — encode_frame, decode_frame, Command/DeviceMessage + # parsers, SetConfigResult, RxEvent, etc. Pinned tightly; the mux knows + # intimate details about the v1.0 shape and will break if the protocol + # moves. + "donglora>=1.0.0,<2.0.0", ] [project.scripts] @@ -28,7 +30,7 @@ Repository = "https://github.com/donglora/mux-py" Homepage = "https://github.com/donglora/mux-py" [dependency-groups] -dev = ["ruff", "pytest", "pytest-cov"] +dev = ["ruff", "pytest", "pytest-cov", "pytest-asyncio"] [build-system] requires = ["hatchling"] @@ -50,3 +52,4 @@ known-first-party = ["donglora", "donglora_mux"] [tool.pytest.ini_options] testpaths = ["tests"] +asyncio_mode = "auto" diff --git a/tests/test_intercept.py b/tests/test_intercept.py index dba3a0b..50af440 100644 --- a/tests/test_intercept.py +++ b/tests/test_intercept.py @@ -1,158 +1,194 @@ -"""Tests for mux interception logic — ported from mux-rs/src/intercept.rs tests.""" +"""Unit tests for :mod:`donglora_mux.intercept`. + +Ports the ``intercept`` test cases from ``mux-rs/src/intercept.rs``, +adapted for Python / asyncio. +""" from __future__ import annotations -import asyncio +import pytest -from donglora.protocol import RADIO_CONFIG_SIZE -from donglora_mux import ( - CMD_TAG_SET_CONFIG, - CMD_TAG_START_RX, - CMD_TAG_STOP_RX, - ERROR_INVALID_CONFIG, - TAG_ERROR, - TAG_OK, - Client, - MuxDaemon, +from donglora.commands import ( + TYPE_GET_INFO, + TYPE_PING, + TYPE_RX_START, + TYPE_RX_STOP, + TYPE_SET_CONFIG, + encode_set_config_payload, ) +from donglora.events import TYPE_OK, Owner, SetConfigResult, SetConfigResultCode +from donglora.modulation import LoRaBandwidth, LoRaCodingRate, LoRaConfig +from donglora_mux.intercept import DecisionKind, MuxState, decide +from donglora_mux.session import ClientSession -def _make_daemon() -> MuxDaemon: - """Create a MuxDaemon with no real serial port (for interception tests only).""" - daemon = MuxDaemon.__new__(MuxDaemon) - daemon.serial_port = "/dev/null" - daemon.socket_path = "/tmp/test-mux.sock" - daemon.tcp_addr = None - daemon.ser = None - daemon.clients = {} - daemon.cmd_queue = asyncio.Queue() - daemon.pending_response = None - daemon.locked_config = None - daemon._shutdown = asyncio.Event() - return daemon +def _lora(freq_hz: int = 910_525_000) -> LoRaConfig: + return LoRaConfig( + freq_hz=freq_hz, + sf=7, + bw=LoRaBandwidth.KHZ_62_5, + cr=LoRaCodingRate.CR_4_5, + preamble_len=16, + sync_word=0x1424, + tx_power_dbm=20, + ) -def _make_client(daemon: MuxDaemon) -> Client: - """Add a mock client to the daemon and return it.""" - reader = asyncio.StreamReader() - # We can't easily create a StreamWriter without a transport, so we'll - # test the interception logic directly via _maybe_intercept - client = Client.__new__(Client) - client.id = Client._next_id - Client._next_id += 1 - client.reader = reader - client.writer = None # type: ignore[assignment] - client.rx_interested = False - client.send_queue = asyncio.Queue(maxsize=256) - client._sender_task = None - daemon.clients[client.id] = client - return client +def _make_sessions(n: int) -> dict[int, ClientSession]: + return {s.id: s for s in (ClientSession() for _ in range(n))} -class TestSetConfigIntercept: - def test_single_client_forwards(self) -> None: - daemon = _make_daemon() - c1 = _make_client(daemon) - cmd = bytes([CMD_TAG_SET_CONFIG]) + bytes(RADIO_CONFIG_SIZE) - result = asyncio.get_event_loop().run_until_complete(daemon._maybe_intercept(c1.id, cmd)) - assert result is None # should forward to dongle - - def test_first_with_multiple_forwards(self) -> None: - daemon = _make_daemon() - _make_client(daemon) - c2 = _make_client(daemon) - cmd = bytes([CMD_TAG_SET_CONFIG]) + bytes([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]) - result = asyncio.get_event_loop().run_until_complete(daemon._maybe_intercept(c2.id, cmd)) - assert result is None # no locked config yet, forward - - def test_matching_returns_ok(self) -> None: - daemon = _make_daemon() - _make_client(daemon) - c2 = _make_client(daemon) - config_bytes = bytes([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]) - daemon.locked_config = config_bytes - cmd = bytes([CMD_TAG_SET_CONFIG]) + config_bytes - result = asyncio.get_event_loop().run_until_complete(daemon._maybe_intercept(c2.id, cmd)) - assert result == bytes([TAG_OK]) - - def test_conflicting_returns_error(self) -> None: - daemon = _make_daemon() - _make_client(daemon) - c2 = _make_client(daemon) - daemon.locked_config = bytes([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]) - cmd = bytes([CMD_TAG_SET_CONFIG]) + bytes([99, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]) - result = asyncio.get_event_loop().run_until_complete(daemon._maybe_intercept(c2.id, cmd)) - assert result == bytes([TAG_ERROR, ERROR_INVALID_CONFIG]) +def _first_id(sessions: dict[int, ClientSession]) -> int: + return min(sessions.keys()) -class TestStartRxIntercept: - def test_first_client_forwards(self) -> None: - daemon = _make_daemon() - c1 = _make_client(daemon) - result = asyncio.get_event_loop().run_until_complete( - daemon._maybe_intercept(c1.id, bytes([CMD_TAG_START_RX])) - ) - assert result is None # first interested client, forward - - def test_already_interested_returns_ok(self) -> None: - daemon = _make_daemon() - c1 = _make_client(daemon) - c1.rx_interested = True - result = asyncio.get_event_loop().run_until_complete( - daemon._maybe_intercept(c1.id, bytes([CMD_TAG_START_RX])) - ) - assert result == bytes([TAG_OK]) - - def test_others_interested_marks_and_returns_ok(self) -> None: - daemon = _make_daemon() - c1 = _make_client(daemon) - c2 = _make_client(daemon) - c1.rx_interested = True - result = asyncio.get_event_loop().run_until_complete( - daemon._maybe_intercept(c2.id, bytes([CMD_TAG_START_RX])) - ) - assert result == bytes([TAG_OK]) - assert c2.rx_interested is True +def _second_id(sessions: dict[int, ClientSession]) -> int: + ids = sorted(sessions.keys()) + return ids[1] -class TestStopRxIntercept: - def test_not_interested_returns_ok(self) -> None: - daemon = _make_daemon() - c1 = _make_client(daemon) - result = asyncio.get_event_loop().run_until_complete( - daemon._maybe_intercept(c1.id, bytes([CMD_TAG_STOP_RX])) - ) - assert result == bytes([TAG_OK]) - - def test_others_remain_returns_ok(self) -> None: - daemon = _make_daemon() - c1 = _make_client(daemon) - c2 = _make_client(daemon) - c1.rx_interested = True - c2.rx_interested = True - result = asyncio.get_event_loop().run_until_complete( - daemon._maybe_intercept(c1.id, bytes([CMD_TAG_STOP_RX])) - ) - assert result == bytes([TAG_OK]) - assert c1.rx_interested is False - - def test_last_interested_forwards(self) -> None: - daemon = _make_daemon() - c1 = _make_client(daemon) - c1.rx_interested = True - result = asyncio.get_event_loop().run_until_complete( - daemon._maybe_intercept(c1.id, bytes([CMD_TAG_STOP_RX])) - ) - assert result is None # last client, forward to dongle - assert c1.rx_interested is False +# ── SET_CONFIG ───────────────────────────────────────────────────── -class TestOtherCommands: - def test_not_intercepted(self) -> None: - daemon = _make_daemon() - c1 = _make_client(daemon) - for tag in [0, 1, 5, 6, 7, 8]: - result = asyncio.get_event_loop().run_until_complete( - daemon._maybe_intercept(c1.id, bytes([tag])) - ) - assert result is None +def test_set_config_single_client_forwards(): + sessions = _make_sessions(1) + state = MuxState() + cid = _first_id(sessions) + payload = encode_set_config_payload(_lora()) + d = decide(TYPE_SET_CONFIG, payload, cid, state, sessions) + assert d.kind == DecisionKind.FORWARD + + +def test_set_config_first_multi_client_forwards(): + sessions = _make_sessions(2) + state = MuxState() + cid = _first_id(sessions) + payload = encode_set_config_payload(_lora()) + d = decide(TYPE_SET_CONFIG, payload, cid, state, sessions) + assert d.kind == DecisionKind.FORWARD + + +def test_set_config_matching_synthesizes_already_matched(): + sessions = _make_sessions(2) + id_a = _first_id(sessions) + id_b = _second_id(sessions) + state = MuxState() + state.locked = (id_a, _lora()) + payload = encode_set_config_payload(_lora()) + d = decide(TYPE_SET_CONFIG, payload, id_b, state, sessions) + assert d.kind == DecisionKind.SYNTHESIZE + assert d.type_id == TYPE_OK + result = SetConfigResult.decode(d.payload) + assert result.result == SetConfigResultCode.ALREADY_MATCHED + assert result.owner == Owner.OTHER + + +def test_set_config_conflict_synthesizes_locked_mismatch_with_current_echoed(): + sessions = _make_sessions(2) + id_a = _first_id(sessions) + id_b = _second_id(sessions) + state = MuxState() + # A's lock is 910.525 MHz; B requests 915 MHz. + state.locked = (id_a, _lora(910_525_000)) + payload = encode_set_config_payload(_lora(915_000_000)) + d = decide(TYPE_SET_CONFIG, payload, id_b, state, sessions) + assert d.kind == DecisionKind.SYNTHESIZE + result = SetConfigResult.decode(d.payload) + assert result.result == SetConfigResultCode.LOCKED_MISMATCH + assert result.owner == Owner.OTHER + # The echoed `current` must reflect A's locked modulation, not B's request. + assert isinstance(result.current, LoRaConfig) + assert result.current.freq_hz == 910_525_000 + + +def test_set_config_owner_reconfig_forwards(): + sessions = _make_sessions(2) + id_a = _first_id(sessions) + state = MuxState() + state.locked = (id_a, _lora(910_525_000)) + # A re-tunes — should forward, daemon updates lock on APPLIED. + payload = encode_set_config_payload(_lora(915_000_000)) + d = decide(TYPE_SET_CONFIG, payload, id_a, state, sessions) + assert d.kind == DecisionKind.FORWARD + + +# ── RX_START ─────────────────────────────────────────────────────── + + +def test_rx_start_first_forwards(): + sessions = _make_sessions(1) + state = MuxState() + cid = _first_id(sessions) + d = decide(TYPE_RX_START, b"", cid, state, sessions) + assert d.kind == DecisionKind.FORWARD + + +def test_rx_start_when_others_listening_marks_and_synthesizes_ok(): + sessions = _make_sessions(2) + state = MuxState() + id_a = _first_id(sessions) + id_b = _second_id(sessions) + sessions[id_a].rx_interested = True + d = decide(TYPE_RX_START, b"", id_b, state, sessions) + assert d.kind == DecisionKind.SYNTHESIZE + assert d.type_id == TYPE_OK + assert d.payload == b"" + assert sessions[id_b].rx_interested is True + + +def test_rx_start_already_interested_synthesizes_ok(): + sessions = _make_sessions(1) + state = MuxState() + cid = _first_id(sessions) + sessions[cid].rx_interested = True + d = decide(TYPE_RX_START, b"", cid, state, sessions) + assert d.kind == DecisionKind.SYNTHESIZE + assert d.type_id == TYPE_OK + + +# ── RX_STOP ──────────────────────────────────────────────────────── + + +def test_rx_stop_not_interested_synthesizes_ok(): + sessions = _make_sessions(1) + state = MuxState() + cid = _first_id(sessions) + d = decide(TYPE_RX_STOP, b"", cid, state, sessions) + assert d.kind == DecisionKind.SYNTHESIZE + assert d.type_id == TYPE_OK + + +def test_rx_stop_with_others_listening_clears_caller_and_synthesizes_ok(): + sessions = _make_sessions(2) + state = MuxState() + id_a = _first_id(sessions) + id_b = _second_id(sessions) + sessions[id_a].rx_interested = True + sessions[id_b].rx_interested = True + d = decide(TYPE_RX_STOP, b"", id_a, state, sessions) + assert d.kind == DecisionKind.SYNTHESIZE + assert d.type_id == TYPE_OK + assert sessions[id_a].rx_interested is False + assert sessions[id_b].rx_interested is True + + +def test_rx_stop_last_interested_forwards_and_clears(): + sessions = _make_sessions(1) + state = MuxState() + cid = _first_id(sessions) + sessions[cid].rx_interested = True + d = decide(TYPE_RX_STOP, b"", cid, state, sessions) + assert d.kind == DecisionKind.FORWARD + assert sessions[cid].rx_interested is False + + +# ── Other commands ───────────────────────────────────────────────── + + +@pytest.mark.parametrize("type_id", [TYPE_PING, TYPE_GET_INFO]) +def test_other_commands_forward(type_id): + sessions = _make_sessions(1) + state = MuxState() + cid = _first_id(sessions) + d = decide(type_id, b"", cid, state, sessions) + assert d.kind == DecisionKind.FORWARD diff --git a/tests/test_multiclient.py b/tests/test_multiclient.py new file mode 100644 index 0000000..051a3d5 --- /dev/null +++ b/tests/test_multiclient.py @@ -0,0 +1,214 @@ +"""Integration tests for :class:`donglora_mux.daemon.MuxDaemon`. + +Drives the daemon's dispatch logic directly without spinning up a real +serial port — we push synthetic dongle frames through +``_dispatch_device_frame`` and inspect what ends up in each client +session's queue. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from donglora.commands import ( + TYPE_PING, + TYPE_SET_CONFIG, + TYPE_TX, + encode_set_config_payload, + encode_tx_payload, +) +from donglora.events import ( + TYPE_OK, + TYPE_RX, + TYPE_TX_DONE, + Owner, + RxEvent, + RxOrigin, + SetConfigResult, + SetConfigResultCode, + TxDone, + TxResult, +) +from donglora.frame import Frame, decode_frame, encode_frame +from donglora.modulation import LoRaBandwidth, LoRaCodingRate, LoRaConfig +from donglora_mux.daemon import MuxDaemon +from donglora_mux.session import ClientSession +from donglora_mux.tagmap import TagMapper + + +def _lora(freq: int = 910_525_000) -> LoRaConfig: + return LoRaConfig( + freq_hz=freq, + sf=7, + bw=LoRaBandwidth.KHZ_62_5, + cr=LoRaCodingRate.CR_4_5, + preamble_len=16, + sync_word=0x1424, + tx_power_dbm=20, + ) + + +def _make_daemon() -> MuxDaemon: + # Construct without touching filesystem / serial — we're exercising + # the dispatch surface directly. + return MuxDaemon("/dev/null", "/tmp/test-mux.sock", tcp_addr=None) + + +def _drain_queue(session: ClientSession) -> list[Frame]: + """Non-blocking: drain whatever is currently in the session's queue.""" + out: list[Frame] = [] + while True: + try: + raw = session._queue.get_nowait() # type: ignore[attr-defined] + except asyncio.QueueEmpty: + return out + # Strip trailing 0x00 sentinel before decode_frame. + assert raw.endswith(b"\x00") + out.append(decode_frame(raw[:-1])) + + +@pytest.fixture +def daemon(): + d = _make_daemon() + # Two sessions, manually added. + a = ClientSession() + b = ClientSession() + d.sessions[a.id] = a + d.sessions[b.id] = b + yield d, a, b + + +async def test_dispatch_routes_tagged_response_to_owning_client(daemon): + d, a, b = daemon + # A issued a PING with upstream tag 0x0042. + device_tag = d.tagmap.alloc(a.id, 0x0042, TYPE_PING) + # Dongle replies with OK on the device_tag. + await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=device_tag, payload=b"")) + + a_frames = _drain_queue(a) + b_frames = _drain_queue(b) + assert len(a_frames) == 1 + assert a_frames[0].type_id == TYPE_OK + assert a_frames[0].tag == 0x0042 + assert b_frames == [] # Only A gets the routed response. + + +async def test_dispatch_fans_out_async_rx_to_all_clients(daemon): + d, a, b = daemon + rx = RxEvent( + rssi_dbm=-80.0, + snr_db=7.5, + freq_err_hz=0, + timestamp_us=1_000_000, + crc_valid=True, + packets_dropped=0, + origin=RxOrigin.OTA, + data=b"hello", + ) + await d._dispatch_device_frame(Frame(type_id=TYPE_RX, tag=0, payload=rx.encode())) + + a_frames = _drain_queue(a) + b_frames = _drain_queue(b) + assert len(a_frames) == 1 + assert len(b_frames) == 1 + assert a_frames[0].type_id == TYPE_RX + assert a_frames[0].tag == 0 + + +async def test_set_config_applied_installs_lock(daemon): + d, a, _b = daemon + cfg = _lora() + device_tag = d.tagmap.alloc(a.id, 0x0099, TYPE_SET_CONFIG) + result = SetConfigResult(result=SetConfigResultCode.APPLIED, owner=Owner.MINE, current=cfg) + await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=device_tag, payload=result.encode())) + # Lock should now be installed with A as owner and the applied cfg. + assert d.state.locked is not None + owner_id, locked_cfg = d.state.locked + assert owner_id == a.id + assert locked_cfg.freq_hz == cfg.freq_hz + + +async def test_tx_done_transmitted_loopbacks_to_peers(daemon): + d, a, b = daemon + # A sent a TX of "hi" with upstream tag 0x00AA. + payload = encode_tx_payload(b"hi") + # Stash like the writer would. + a.tx_cache[0x00AA] = payload[1:] # strip flags byte + device_tag = d.tagmap.alloc(a.id, 0x00AA, TYPE_TX) + # Absorb the intermediate OK (mapping kept, tx_cache still populated). + await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=device_tag, payload=b"")) + # Terminal TX_DONE(TRANSMITTED). + td = TxDone(result=TxResult.TRANSMITTED, airtime_us=5_000) + await d._dispatch_device_frame(Frame(type_id=TYPE_TX_DONE, tag=device_tag, payload=td.encode())) + + a_frames = _drain_queue(a) + b_frames = _drain_queue(b) + + # A got OK (intermediate) + TX_DONE. + assert [f.type_id for f in a_frames] == [TYPE_OK, TYPE_TX_DONE] + assert all(f.tag == 0x00AA for f in a_frames) + # B got a loopback RX (tag=0, origin=LOCAL_LOOPBACK, data="hi"). + assert len(b_frames) == 1 + assert b_frames[0].type_id == TYPE_RX + assert b_frames[0].tag == 0 + decoded = RxEvent # noqa: F841 — just a hint for readers + # Inline decode (RxEvent.decode if available, else pull fields from the wire payload). + buf = b_frames[0].payload + assert buf[19] == int(RxOrigin.LOCAL_LOOPBACK) + assert buf[20:] == b"hi" + + +async def test_disconnect_releases_lock(daemon): + d, a, _b = daemon + d.state.locked = (a.id, _lora()) + # Simulate the post-writer-cancel cleanup step in _remove_client. + await d._remove_client(a) + assert d.state.locked is None + + +async def test_concurrent_tagged_commands_route_independently(daemon): + """Two concurrent PINGs from different clients must correlate correctly.""" + d, a, b = daemon + tag_a = d.tagmap.alloc(a.id, 0x0001, TYPE_PING) + tag_b = d.tagmap.alloc(b.id, 0x0001, TYPE_PING) + # Dongle replies for B first, then A. + await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=tag_b, payload=b"")) + await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=tag_a, payload=b"")) + a_frames = _drain_queue(a) + b_frames = _drain_queue(b) + assert len(a_frames) == 1 + assert len(b_frames) == 1 + # Each sees their own upstream tag, not the mux's internal device tag. + assert a_frames[0].tag == 0x0001 + assert b_frames[0].tag == 0x0001 + + +async def test_unsolicited_tagged_frame_is_dropped(daemon): + """A frame with a tag that isn't in the pending map should be ignored, + not routed to a random client or crash the dispatcher.""" + d, a, b = daemon + # Tag 9999 was never allocated. + await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=9999, payload=b"")) + assert _drain_queue(a) == [] + assert _drain_queue(b) == [] + + +def test_tag_mapper_isolation(daemon): + """Regression check: tagmap should start empty per daemon instance.""" + d, _a, _b = daemon + assert isinstance(d.tagmap, TagMapper) + assert len(d.tagmap) == 0 + + +def test_encode_decode_roundtrip_sanity(): + """Sanity check: encode_frame + strip sentinel + decode_frame are inverses + for the shapes we push through the mux.""" + payload = encode_set_config_payload(_lora()) + wire = encode_frame(TYPE_SET_CONFIG, 42, payload) + assert wire.endswith(b"\x00") + frame = decode_frame(wire[:-1]) + assert frame.type_id == TYPE_SET_CONFIG + assert frame.tag == 42 + assert frame.payload == payload diff --git a/tests/test_tagmap.py b/tests/test_tagmap.py new file mode 100644 index 0000000..3626f8b --- /dev/null +++ b/tests/test_tagmap.py @@ -0,0 +1,77 @@ +"""Unit tests for :mod:`donglora_mux.tagmap`. + +Mirrors the ``TagMapper`` test cases in ``mux-rs/src/daemon.rs``. +""" + +from __future__ import annotations + +from donglora.commands import TYPE_PING, TYPE_TX +from donglora_mux.tagmap import TagMapper + + +def test_alloc_skips_zero(): + m = TagMapper() + # Allocating many tags in a row, each must be non-zero. Even after + # many alloc/take cycles, we should never hand out 0 (reserved for + # async events). + for _ in range(1024): + tag = m.alloc(1, 100, TYPE_PING) + assert tag != 0 + m.take(tag) + + +def test_take_returns_mapping_and_removes(): + m = TagMapper() + tag = m.alloc(42, 7, TYPE_TX) + p = m.take(tag) + assert p is not None + assert p.client_id == 42 + assert p.upstream_tag == 7 + assert p.cmd_type == TYPE_TX + assert len(m) == 0 + + +def test_peek_keeps_mapping(): + m = TagMapper() + tag = m.alloc(42, 7, TYPE_TX) + p = m.peek(tag) + assert p is not None + assert p.client_id == 42 + assert len(m) == 1 + + +def test_drop_client_removes_only_that_clients_tags(): + m = TagMapper() + t_a1 = m.alloc(1, 10, TYPE_PING) + t_a2 = m.alloc(1, 11, TYPE_PING) + t_b = m.alloc(2, 12, TYPE_PING) + m.drop_client(1) + assert m.take(t_a1) is None + assert m.take(t_a2) is None + assert m.take(t_b) is not None + + +def test_concurrent_alloc_yields_unique_tags(): + m = TagMapper() + tags = {m.alloc(1, i, TYPE_PING) for i in range(100)} + assert len(tags) == 100 + assert 0 not in tags + + +def test_allocations_recover_after_take(): + # After a take(), the freed tag can be re-used (post-wrap scenario). + m = TagMapper() + first = m.alloc(1, 0, TYPE_PING) + m.take(first) + # Keep filling until we wrap and get `first` back. Don't assert + # eagerly because the allocator is monotonic; just prove it + # terminates within the u16 space. + seen = {first} + while True: + t = m.alloc(1, 0, TYPE_PING) + if t in seen: + break + seen.add(t) + m.take(t) + if len(seen) > 0x10000: # pragma: no cover — safety + raise AssertionError("allocator failed to recycle slots") diff --git a/uv.lock b/uv.lock index 59b8b34..290a4d7 100644 --- a/uv.lock +++ b/uv.lock @@ -2,6 +2,15 @@ version = 1 revision = 3 requires-python = ">=3.10" +[[package]] +name = "backports-asyncio-runner" +version = "1.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/8e/ff/70dca7d7cb1cbc0edb2c6cc0c38b65cba36cccc491eca64cabd5fe7f8670/backports_asyncio_runner-1.2.0.tar.gz", hash = "sha256:a5aa7b2b7d8f8bfcaa2b57313f70792df84e32a2a746f585213373f900b42162", size = 69893, upload-time = "2025-07-02T02:27:15.685Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a0/59/76ab57e3fe74484f48a53f8e337171b4a2349e506eabe136d7e01d059086/backports_asyncio_runner-1.2.0-py3-none-any.whl", hash = "sha256:0da0a936a8aeb554eccb426dc55af3ba63bcdc69fa1a600b5bb305413a4477b5", size = 12313, upload-time = "2025-07-02T02:27:14.263Z" }, +] + [[package]] name = "cobs" version = "1.2.2" @@ -173,31 +182,27 @@ dev = [ [[package]] name = "donglora-mux" -version = "0.2.0" +version = "1.0.0" source = { editable = "." } dependencies = [ - { name = "cobs" }, { name = "donglora" }, - { name = "pyserial" }, ] [package.dev-dependencies] dev = [ { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "ruff" }, ] [package.metadata] -requires-dist = [ - { name = "cobs" }, - { name = "donglora", editable = "../client-py" }, - { name = "pyserial" }, -] +requires-dist = [{ name = "donglora", editable = "../client-py" }] [package.metadata.requires-dev] dev = [ { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "ruff" }, ] @@ -277,6 +282,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" }, ] +[[package]] +name = "pytest-asyncio" +version = "1.3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "backports-asyncio-runner", marker = "python_full_version < '3.11'" }, + { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/90/2c/8af215c0f776415f3590cac4f9086ccefd6fd463befeae41cd4d3f193e5a/pytest_asyncio-1.3.0.tar.gz", hash = "sha256:d7f52f36d231b80ee124cd216ffb19369aa168fc10095013c6b014a34d3ee9e5", size = 50087, upload-time = "2025-11-10T16:07:47.256Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" }, +] + [[package]] name = "pytest-cov" version = "7.1.0"