From e9c4bb3e93149ba5334c4669815ed479d3f86d3d Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Thu, 16 Apr 2026 15:18:54 -0600 Subject: [PATCH] feat!: Python client Protocol v2 rewrite with async Dongle surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 1.0 Python client, mirroring the Rust client 1.0 API and the DongLoRa Protocol v2 wire format. * Module split: monolithic protocol.py / client.py replaced with dongle.py (high-level async Dongle), session.py (reader task + tag routing), frame.py (streaming FrameDecoder + encode_frame), commands.py / events.py / errors.py / modulation.py / info.py / crc.py / chip_id (re-export) matching the donglora-protocol Rust crate. * Async API throughout. Serial I/O via asyncio-serial; mux connections via asyncio streams. The blocking client surface is gone. * thiserror-equivalent error hierarchy (DongloraError base with Timeout, TransportClosed, ReaderExited, BadFrame, ErrClientError variants mirroring the Rust ClientError). * Tag-aware dispatch with concurrent in-flight commands: TagAllocator hands out 16-bit device tags, Session routes each response back to the originating coroutine. * Auto-recovery from ERR(ENOTCONFIGURED): the Dongle caches the last SET_CONFIG and silently re-applies on timeout. * 500 ms background keepalive task per PROTOCOL.md §3.4. * Tests rewritten: test_dongle, test_session, test_frame, test_events, test_modulation, test_crc, test_ergonomics; old test_client / test_protocol removed. * Sticky mux reconnect (from the prior fix commit) is preserved: once a mux socket is in use, reconnects commit to it instead of falling back to USB. --- donglora/__init__.py | 231 +++++++++++++--------- donglora/_top_level.py | 70 +++++++ donglora/client.py | 108 ----------- donglora/codec.py | 41 +--- donglora/commands.py | 50 +++++ donglora/connect.py | 303 ++++++++++++++++++----------- donglora/crc.py | 27 +++ donglora/dongle.py | 360 ++++++++++++++++++++++++++++++++++ donglora/errors.py | 123 ++++++++++++ donglora/events.py | 186 ++++++++++++++++++ donglora/frame.py | 150 +++++++++++++++ donglora/info.py | 196 +++++++++++++++++++ donglora/modulation.py | 406 +++++++++++++++++++++++++++++++++++++++ donglora/protocol.py | 234 ---------------------- donglora/session.py | 371 +++++++++++++++++++++++++++++++++++ tests/test_client.py | 135 ------------- tests/test_codec.py | 48 +---- tests/test_crc.py | 23 +++ tests/test_dongle.py | 219 +++++++++++++++++++++ tests/test_ergonomics.py | 75 ++++++++ tests/test_events.py | 265 +++++++++++++++++++++++++ tests/test_frame.py | 175 +++++++++++++++++ tests/test_modulation.py | 202 +++++++++++++++++++ tests/test_protocol.py | 254 ------------------------ tests/test_session.py | 337 ++++++++++++++++++++++++++++++++ 25 files changed, 3579 insertions(+), 1010 deletions(-) create mode 100644 donglora/_top_level.py delete mode 100644 donglora/client.py create mode 100644 donglora/commands.py create mode 100644 donglora/crc.py create mode 100644 donglora/dongle.py create mode 100644 donglora/errors.py create mode 100644 donglora/events.py create mode 100644 donglora/frame.py create mode 100644 donglora/info.py create mode 100644 donglora/modulation.py delete mode 100644 donglora/protocol.py create mode 100644 donglora/session.py delete mode 100644 tests/test_client.py create mode 100644 tests/test_crc.py create mode 100644 tests/test_dongle.py create mode 100644 tests/test_ergonomics.py create mode 100644 tests/test_events.py create mode 100644 tests/test_frame.py create mode 100644 tests/test_modulation.py delete mode 100644 tests/test_protocol.py create mode 100644 tests/test_session.py diff --git a/donglora/__init__.py b/donglora/__init__.py index aebb608..7c78e72 100644 --- a/donglora/__init__.py +++ b/donglora/__init__.py @@ -1,30 +1,48 @@ -"""DongLoRa host library — connect, configure, send/receive LoRa packets. - -Implements the DongLoRa USB protocol (COBS-framed fixed-size LE). -See firmware/PROTOCOL.md for the full specification. +"""DongLoRa host library — talk LoRa from Python in three lines. Quick start:: import donglora as dl - ser = dl.connect() - dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG) - dl.send(ser, "StartRx") + d = dl.connect() # auto-discover, auto-configure, auto-keepalive + d.tx(b"Hello") # blocks until TX_DONE + for pkt in d.rx(): # continuous receive + print(pkt.rssi_dbm, pkt.data) - while True: - pkt = dl.recv(ser) - if pkt: - print(pkt["rssi"], pkt["payload"].hex()) +For trivial scripts, skip the handle:: + + import donglora as dl + dl.tx(b"hi") + for pkt in dl.rx(): + ... + +Customise anything via :func:`connect` kwargs: + +* ``port="/dev/ttyUSB0"`` — explicit serial device +* ``config=LoRaConfig(...)`` — override the default radio config +* ``auto_configure=False`` — skip automatic SET_CONFIG +* ``keepalive=False`` — run your own session-liveness cadence + +The full DongLoRa Protocol v2 surface is available via :class:`Dongle` methods +(:meth:`Dongle.tx` / :meth:`Dongle.rx` / :meth:`Dongle.set_config` / +:meth:`Dongle.ping` / :attr:`Dongle.info`) and the underlying +:mod:`donglora.frame` / :mod:`donglora.events` / :mod:`donglora.info` +modules if you need to speak the wire directly. + +See ``firmware/PROTOCOL.md`` for the wire-level specification. """ -from donglora.client import ( - HANDSHAKE_TIMEOUT, - drain_rx, - recv, - send, - validate, +from __future__ import annotations + +from donglora._top_level import close, recv, rx, tx +from donglora.commands import ( + TYPE_GET_INFO, + TYPE_PING, + TYPE_RX_START, + TYPE_RX_STOP, + TYPE_SET_CONFIG, + TYPE_TX, ) -from donglora.codec import cobs_encode, read_frame from donglora.connect import ( DEFAULT_TIMEOUT, connect, @@ -43,101 +61,128 @@ from donglora.discovery import ( find_port, wait_for_device, ) -from donglora.protocol import ( - CMD_TAG_GET_CONFIG, - CMD_TAG_GET_MAC, - CMD_TAG_PING, - CMD_TAG_SET_CONFIG, - CMD_TAG_START_RX, - CMD_TAG_STOP_RX, - CMD_TAG_TRANSMIT, - DEFAULT_CONFIG, - ERROR_INVALID_CONFIG, - ERROR_NO_DISPLAY, - ERROR_NOT_CONFIGURED, - ERROR_RADIO_BUSY, - ERROR_TX_TIMEOUT, - MAX_PAYLOAD, - PREAMBLE_DEFAULT, - RADIO_CONFIG_SIZE, - RESP_TAG_CONFIG, - RESP_TAG_ERROR, - RESP_TAG_MAC_ADDRESS, - RESP_TAG_OK, - RESP_TAG_PONG, - RESP_TAG_RX_PACKET, - RESP_TAG_TX_DONE, - TX_POWER_MAX, - Bandwidth, +from donglora.dongle import Dongle +from donglora.errors import ( + BusyError, + Cancelled, + ChannelBusy, + DeviceError, + DongloraError, ErrorCode, - RadioConfig, - decode_config, - decode_response, - encode_command, - encode_config, + FrameError, + InternalError, + LengthError, + ModulationError, + NotConfiguredError, + ParamError, + RadioError, + TimeoutError_, + UnknownCommandError, +) +from donglora.events import ( + Owner, + RxEvent, + RxOrigin, + SetConfigResult, + SetConfigResultCode, + TxDone, + TxResult, +) +from donglora.frame import MAX_OTA_PAYLOAD, MAX_PAYLOAD_FIELD, MAX_WIRE_FRAME, Frame, encode_frame +from donglora.info import MAX_MCU_UID_LEN, MAX_RADIO_UID_LEN, Capability, Info, RadioChipId +from donglora.modulation import ( + FlrcBitrate, + FlrcBt, + FlrcCodingRate, + FlrcConfig, + FlrcPreambleLen, + FskConfig, + LoRaBandwidth, + LoRaCodingRate, + LoRaConfig, + LoRaHeaderMode, + LrFhssBandwidth, + LrFhssCodingRate, + LrFhssConfig, + LrFhssGrid, + Modulation, + ModulationId, ) from donglora.transport import MuxConnection __all__ = [ "BRIDGE_VID_PIDS", - "CMD_TAG_GET_CONFIG", - "CMD_TAG_GET_MAC", - # tag constants - "CMD_TAG_PING", - "CMD_TAG_SET_CONFIG", - "CMD_TAG_START_RX", - "CMD_TAG_STOP_RX", - "CMD_TAG_TRANSMIT", - "DEFAULT_CONFIG", - # connect "DEFAULT_TIMEOUT", - # error constants - "ERROR_INVALID_CONFIG", - "ERROR_NOT_CONFIGURED", - "ERROR_NO_DISPLAY", - "ERROR_RADIO_BUSY", - "ERROR_TX_TIMEOUT", - # client - "HANDSHAKE_TIMEOUT", - # protocol - "MAX_PAYLOAD", - "PREAMBLE_DEFAULT", - "RADIO_CONFIG_SIZE", - "RESP_TAG_CONFIG", - "RESP_TAG_ERROR", - "RESP_TAG_MAC_ADDRESS", - "RESP_TAG_OK", - "RESP_TAG_PONG", - "RESP_TAG_RX_PACKET", - "RESP_TAG_TX_DONE", - "TX_POWER_MAX", + "MAX_MCU_UID_LEN", + "MAX_OTA_PAYLOAD", + "MAX_PAYLOAD_FIELD", + "MAX_RADIO_UID_LEN", + "MAX_WIRE_FRAME", + "TYPE_GET_INFO", + "TYPE_PING", + "TYPE_RX_START", + "TYPE_RX_STOP", + "TYPE_SET_CONFIG", + "TYPE_TX", "USB_PID", - # discovery "USB_VID", "USB_VID_PID", - "Bandwidth", + "BusyError", + "Cancelled", + "Capability", + "ChannelBusy", + "DeviceError", + "Dongle", + "DongloraError", "ErrorCode", - # transport + "FlrcBitrate", + "FlrcBt", + "FlrcCodingRate", + "FlrcConfig", + "FlrcPreambleLen", + "Frame", + "FrameError", + "FskConfig", + "Info", + "InternalError", + "LengthError", + "LoRaBandwidth", + "LoRaCodingRate", + "LoRaConfig", + "LoRaHeaderMode", + "LrFhssBandwidth", + "LrFhssCodingRate", + "LrFhssConfig", + "LrFhssGrid", + "Modulation", + "ModulationError", + "ModulationId", "MuxConnection", - "RadioConfig", - # codec - "cobs_encode", + "NotConfiguredError", + "Owner", + "ParamError", + "RadioChipId", + "RadioError", + "RxEvent", + "RxOrigin", + "SetConfigResult", + "SetConfigResultCode", + "TimeoutError_", + "TxDone", + "TxResult", + "UnknownCommandError", + "close", "connect", "connect_default", "connect_mux_auto", - "decode_config", - "decode_response", "default_socket_path", - "drain_rx", - "encode_command", - "encode_config", + "encode_frame", "find_port", "mux_connect", "mux_tcp_connect", - "read_frame", "recv", - "send", + "rx", "try_connect", - "validate", + "tx", "wait_for_device", ] diff --git a/donglora/_top_level.py b/donglora/_top_level.py new file mode 100644 index 0000000..1521923 --- /dev/null +++ b/donglora/_top_level.py @@ -0,0 +1,70 @@ +"""Module-level convenience singletons so ``import donglora as dl`` gives +one-line TX / RX scripts a Dongle without any explicit setup. + +.. code-block:: python + + import donglora as dl + + dl.tx(b"hello") # lazy auto-connect on first call + for pkt in dl.rx(): # same — lazy auto-connect + print(pkt.rssi_dbm, pkt.data) + +Users who need multiple connections, explicit configs, or context-manager +lifetimes should call :func:`donglora.connect` directly and use the +returned :class:`Dongle`. +""" + +from __future__ import annotations + +import atexit +import contextlib +import threading +from collections.abc import Iterator + +from donglora.dongle import Dongle +from donglora.events import RxEvent, TxDone + +_default: Dongle | None = None +_lock = threading.Lock() + + +def _get_default() -> Dongle: + """Return (lazily creating) the process-wide default :class:`Dongle`.""" + global _default + with _lock: + if _default is None: + from donglora.connect import connect + + _default = connect() + atexit.register(_cleanup) + return _default + + +def _cleanup() -> None: + global _default + with _lock: + d = _default + _default = None + if d is not None: + with contextlib.suppress(Exception): + d.close() + + +def tx(data: bytes, *, skip_cad: bool = False, timeout: float = 10.0) -> TxDone: + """One-line transmit. Creates the default dongle on first call.""" + return _get_default().tx(data, skip_cad=skip_cad, timeout=timeout) + + +def rx(timeout: float | None = None) -> Iterator[RxEvent]: + """One-line continuous receive. Lazy-connects on first call.""" + return iter(_get_default().rx(timeout)) + + +def recv(timeout: float | None = None) -> RxEvent | None: + """One packet. Lazy-connects on first call.""" + return _get_default().recv(timeout) + + +def close() -> None: + """Close the default dongle, if any. Safe to call multiple times.""" + _cleanup() diff --git a/donglora/client.py b/donglora/client.py deleted file mode 100644 index 2f14eee..0000000 --- a/donglora/client.py +++ /dev/null @@ -1,108 +0,0 @@ -"""High-level DongLoRa client helpers. - -Mirrors the Rust client's ``client.rs``. Provides :func:`send`, -:func:`recv`, :func:`drain_rx`, and :func:`validate` as module-level -functions that operate on any transport (serial port or mux connection). -""" - -from __future__ import annotations - -import collections -from typing import Any - -from donglora.codec import cobs_encode, read_frame -from donglora.protocol import decode_response, encode_command - -RX_BUFFER_CAP: int = 256 -"""Maximum number of buffered RxPackets before oldest are dropped.""" - -MAX_UNSOLICITED_BEFORE_TIMEOUT: int = 50 -"""Safety bound: max frames to read while waiting for a solicited response.""" - -HANDSHAKE_TIMEOUT: float = 0.2 -"""Timeout (seconds) for the ping-on-connect handshake.""" - -_rx_queue: collections.deque[dict[str, Any]] = collections.deque(maxlen=RX_BUFFER_CAP) -"""Buffer for RxPacket frames encountered while waiting for solicited responses.""" - - -def validate(ser: Any) -> None: - """Ping the device and verify the Pong response. - - Uses a short timeout to fail fast on non-DongLoRa devices. Called - automatically by the ``connect`` family of functions. - - Raises :class:`ConnectionError` if the device does not respond. - """ - saved = ser.timeout - ser.timeout = HANDSHAKE_TIMEOUT - try: - resp = send(ser, "Ping") - if resp["type"] != "Pong": - raise ConnectionError(f"device did not respond to ping — got {resp['type']}") - except ConnectionError: - raise - except Exception as exc: - raise ConnectionError("device did not respond to ping — not a DongLoRa device") from exc - finally: - ser.timeout = saved - - -def send(ser: Any, cmd: str, **kwargs: Any) -> dict[str, Any]: - """Send a command and return the solicited response. - - Any unsolicited ``RxPacket`` frames encountered while waiting are - buffered in the module-level receive queue (retrievable via - :func:`recv` / :func:`drain_rx`). - """ - ser.write(cobs_encode(encode_command(cmd, **kwargs))) - ser.flush() - for _ in range(MAX_UNSOLICITED_BEFORE_TIMEOUT): - data = read_frame(ser) - if data is None: - return {"type": "Timeout"} - resp = decode_response(data) - if resp["type"] == "RxPacket": - _rx_queue.append(resp) - continue - return resp - return {"type": "Timeout"} - - -def recv(ser: Any) -> dict[str, Any] | None: - """Return the next RxPacket from the buffer or the wire. - - Returns ``None`` on timeout (no packet available). - """ - if _rx_queue: - return _rx_queue.popleft() - data = read_frame(ser) - if data is None: - return None - resp = decode_response(data) - if resp["type"] == "RxPacket": - return resp - return None - - -def drain_rx(ser: Any) -> list[dict[str, Any]]: - """Drain all buffered and pending RxPacket frames. - - Temporarily reduces the read timeout to quickly drain any frames - still in flight, then restores the original timeout. - """ - packets: list[dict[str, Any]] = list(_rx_queue) - _rx_queue.clear() - old_timeout = ser.timeout - ser.timeout = 0.01 - try: - while True: - data = read_frame(ser) - if data is None: - break - resp = decode_response(data) - if resp["type"] == "RxPacket": - packets.append(resp) - finally: - ser.timeout = old_timeout - return packets diff --git a/donglora/codec.py b/donglora/codec.py index 169e273..85f12c2 100644 --- a/donglora/codec.py +++ b/donglora/codec.py @@ -1,44 +1,21 @@ -"""COBS framing for the DongLoRa USB protocol. +"""Backwards-compatible COBS codec helpers. -Mirrors the Rust client's ``codec.rs``. Frames are COBS-encoded with a -``0x00`` sentinel byte. +The v1.0 wire layer lives in :mod:`donglora.frame`. This module keeps +the thin helpers that the old client exposed (``cobs_encode``, +``read_frame``) so unrelated transport glue keeps working. """ from __future__ import annotations -from typing import Protocol, runtime_checkable - from cobs import cobs - -@runtime_checkable -class Readable(Protocol): - """Anything with a blocking ``read(n)`` method (serial port, socket wrapper, etc.).""" - - def read(self, n: int = 1) -> bytes: ... +from donglora.frame import Frame, Readable, read_frame # noqa: F401 — re-export def cobs_encode(data: bytes) -> bytes: - """COBS-encode *data* and append the ``0x00`` sentinel.""" - return cobs.encode(data) + b"\x00" + """COBS-encode *data* and append the ``0x00`` sentinel. - -def read_frame(reader: Readable) -> bytes | None: - """Read one COBS frame from *reader*. - - Returns decoded payload bytes, or ``None`` on timeout (zero-length read). + Exposed for transport-level pass-through; regular callers should go + through :func:`donglora.frame.encode_frame` instead. """ - buf = b"" - while True: - b = reader.read(1) - if not b: - return None - if b == b"\x00": - break - buf += b - if not buf: - return None - try: - return cobs.decode(buf) - except cobs.DecodeError: - return None + return cobs.encode(data) + b"\x00" diff --git a/donglora/commands.py b/donglora/commands.py new file mode 100644 index 0000000..6c00020 --- /dev/null +++ b/donglora/commands.py @@ -0,0 +1,50 @@ +"""Host-to-device command payload encoders and type constants.""" + +from __future__ import annotations + +import struct +from dataclasses import dataclass + +from donglora.modulation import Modulation, encode_modulation + +# Message type identifiers (PROTOCOL.md §4) +TYPE_PING: int = 0x01 +TYPE_GET_INFO: int = 0x02 +TYPE_SET_CONFIG: int = 0x03 +TYPE_TX: int = 0x04 +TYPE_RX_START: int = 0x05 +TYPE_RX_STOP: int = 0x06 + + +@dataclass(frozen=True) +class TxFlags: + skip_cad: bool = False + + def as_byte(self) -> int: + return 0b0000_0001 if self.skip_cad else 0 + + +def encode_ping_payload() -> bytes: + return b"" + + +def encode_get_info_payload() -> bytes: + return b"" + + +def encode_set_config_payload(modulation: Modulation) -> bytes: + return encode_modulation(modulation) + + +def encode_tx_payload(data: bytes, *, skip_cad: bool = False) -> bytes: + if not data: + raise ValueError("TX payload must be non-empty") + return struct.pack(" bytes: + return b"" + + +def encode_rx_stop_payload() -> bytes: + return b"" diff --git a/donglora/connect.py b/donglora/connect.py index bdc1460..c9a1a3c 100644 --- a/donglora/connect.py +++ b/donglora/connect.py @@ -1,9 +1,16 @@ -"""Connection auto-detection and mux client helpers. +"""Auto-detect and connect to a DongLoRa device. -Mirrors the Rust client's ``connect.rs``. The :func:`connect` function -tries mux connections first (TCP via env var, then Unix socket), falling -back to direct USB serial. This matches the Rust client's ``connect()`` -behaviour. +``connect()`` is the "60-seconds-to-hacking" entry point. In its zero-argument +form it: + +1. Finds a DongLoRa USB device (or mux socket if one is available). +2. Opens the transport at 115200 baud. +3. Sends ``GET_INFO`` to confirm we're talking DongLoRa Protocol v2. +4. Applies a sensible default :class:`LoRaConfig` via ``SET_CONFIG``. +5. Spawns a keepalive thread to keep the session alive. +6. Returns a :class:`Dongle` ready for ``.tx()`` / ``.rx()``. + +Every step is overrideable via keyword arguments. See ``connect``'s docstring. """ from __future__ import annotations @@ -16,23 +23,23 @@ from typing import Any import serial -from donglora.client import validate from donglora.discovery import find_port, wait_for_device +from donglora.dongle import Dongle +from donglora.errors import DongloraError +from donglora.info import Info +from donglora.modulation import LoRaConfig, Modulation +from donglora.session import Session from donglora.transport import MuxConnection -log = logging.getLogger("donglora") +log = logging.getLogger("donglora.connect") DEFAULT_TIMEOUT: float = 2.0 -"""Default read timeout for connections (seconds).""" +"""Default transport read timeout (seconds). This is the *poll* interval +— overall command deadlines are per-call kwargs on the Dongle API.""" def default_socket_path() -> str: - """Resolve the mux socket path in priority order. - - 1. ``$DONGLORA_MUX`` environment variable - 2. ``$XDG_RUNTIME_DIR/donglora/mux.sock`` - 3. ``/tmp/donglora-mux.sock`` - """ + """Resolve the mux socket path in priority order.""" env = os.environ.get("DONGLORA_MUX") if env: return env @@ -43,7 +50,6 @@ def default_socket_path() -> str: def _find_mux_socket() -> str | None: - """Find an existing mux socket path, or ``None`` if no socket file exists.""" env = os.environ.get("DONGLORA_MUX") if env: return env if os.path.exists(env) else None @@ -57,15 +63,9 @@ def _find_mux_socket() -> str | None: _used_mux = False -"""Set once this process connects via mux. All future connect() calls -will only try mux, never fall through to direct serial.""" def _try_tcp_mux(addr: str, timeout: float) -> MuxConnection | None: - """Try to connect to a TCP mux at *addr* (``[host:]port``). - - Returns ``None`` on failure instead of raising. - """ if ":" in addr: host, _, port_str = addr.rpartition(":") host = host or "localhost" @@ -77,71 +77,35 @@ def _try_tcp_mux(addr: str, timeout: float) -> MuxConnection | None: except ValueError: return None try: - return mux_tcp_connect(host, port, timeout) + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect((host, port)) + return MuxConnection(sock, timeout) except (ConnectionRefusedError, OSError): return None -# ── Public connection functions ─────────────────────────────────── +def _open_transport(port: str | None, timeout: float) -> Any: + """Bottom-level transport opener. Returns a serial/socket-like object. - -def mux_connect(path: str | None = None, timeout: float = DEFAULT_TIMEOUT) -> MuxConnection: - """Connect to the DongLoRa mux daemon via Unix domain socket.""" - if path is None: - path = _find_mux_socket() - if path is None: - raise FileNotFoundError("no mux socket found") - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.connect(path) - conn = MuxConnection(sock, timeout) - validate(conn) - return conn - - -def mux_tcp_connect(host: str, port: int, timeout: float = DEFAULT_TIMEOUT) -> MuxConnection: - """Connect to the DongLoRa mux daemon via TCP.""" - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.connect((host, port)) - conn = MuxConnection(sock, timeout) - validate(conn) - return conn - - -def connect(port: str | None = None, timeout: float = DEFAULT_TIMEOUT) -> Any: - """Auto-detect and connect to a DongLoRa device. - - Priority (first connection): - - 1. ``DONGLORA_MUX_TCP`` env var -> TCP mux connection - 2. Unix socket mux (if socket file exists) - 3. Direct USB serial (auto-detect by VID:PID, blocks until device appears) - - If *port* is given, skips mux detection and connects directly. - - Once a mux connection succeeds, all future calls in this process only - try the mux (waiting for it to reappear if necessary). This prevents - clients from stealing the serial port during a mux restart. + Priority: explicit port → sticky mux → TCP mux env → Unix mux socket → USB auto-detect. """ global _used_mux - # Explicit port — go direct if port is not None: log.debug("opening serial port %s", port) ser = serial.Serial(port, baudrate=115200, timeout=timeout) ser.reset_input_buffer() - validate(ser) return ser - # Previously connected via mux — stay on mux, wait if necessary if _used_mux: mux_path = default_socket_path() while not os.path.exists(mux_path): log.info("Waiting for mux at %s ...", mux_path) time.sleep(0.5) - conn = mux_connect(mux_path, timeout) - return conn + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(mux_path) + return MuxConnection(sock, timeout) - # Try TCP mux via environment variable tcp = os.environ.get("DONGLORA_MUX_TCP") if tcp: conn = _try_tcp_mux(tcp, timeout) @@ -150,94 +114,211 @@ def connect(port: str | None = None, timeout: float = DEFAULT_TIMEOUT) -> Any: _used_mux = True return conn - # Try Unix socket mux — commit if socket exists (no fall-through) sock_path = _find_mux_socket() if sock_path is not None: - log.debug("mux socket found at %s — connecting via mux only", sock_path) - conn = mux_connect(sock_path, timeout) + log.debug("mux socket found at %s", sock_path) + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(sock_path) _used_mux = True - return conn + return MuxConnection(sock, timeout) - # No mux found on first attempt — direct USB serial port_path = find_port() if port_path is None: port_path = wait_for_device() log.debug("opening serial port %s", port_path) ser = serial.Serial(port_path, baudrate=115200, timeout=timeout) ser.reset_input_buffer() - validate(ser) return ser -def connect_default() -> Any: - """Connect with the default timeout. Convenience wrapper for ``connect(None)``.""" - return connect(timeout=DEFAULT_TIMEOUT) +def connect( + port: str | None = None, + *, + timeout: float = DEFAULT_TIMEOUT, + config: Modulation | None = None, + auto_configure: bool = True, + keepalive: bool = True, +) -> Dongle: + """Connect to a DongLoRa device and return a ready-to-use :class:`Dongle`. + The default (zero-argument) invocation: -def connect_mux_auto(timeout: float = DEFAULT_TIMEOUT) -> Any: - """Connect to a mux daemon only (TCP via env var, then Unix socket). + * Auto-discovers a USB device (or mux socket). + * Sends ``GET_INFO`` to confirm DongLoRa Protocol v2. + * Applies ``LoRaConfig.default()`` (EU868 / SF7 / BW125 / CR4/5). + * Starts a background keepalive thread. - Unlike :func:`connect`, this **never** falls back to direct USB serial. - Returns an error if no mux is reachable — the caller can retry with - backoff. + Parameters + ---------- + port + Explicit serial device path. Skips auto-discovery. + timeout + Transport read timeout (polling interval). Default 2 s. + config + Modulation to apply at connect-time. Default: + :meth:`LoRaConfig.default`. Pass ``None`` with + ``auto_configure=False`` to skip configuration entirely. + auto_configure + If False, skip the ``SET_CONFIG`` step. The caller is then + responsible for calling :meth:`Dongle.set_config` before TX/RX. + keepalive + If False, skip the keepalive daemon thread. The caller is + responsible for periodic :meth:`Dongle.ping` calls to stay + under the 1 s inactivity timer. - This is the Python equivalent of the Rust client's "sticky mux" behaviour: - once you decide to use the mux, you stay on the mux. """ - # Try TCP mux via environment variable + + def _open_and_init() -> tuple[Session, Info, Modulation | None]: + """Open the transport, bring up a Session, and verify+configure. + + Used for both the initial connect and any later transparent + reconnect driven by :meth:`Dongle._recover_session`. Respects + the sticky-mux state at call time — so a mux-first first call + followed by a reconnect will stay on mux even if USB came back. + """ + transport = _open_transport(port, timeout) + session = Session(transport) + try: + info = session.get_info(timeout=max(timeout, 2.0)) + except DongloraError: + session.close() + raise + if not isinstance(info, Info): + session.close() + raise DongloraError(f"GET_INFO returned unexpected payload: {info!r}") + if info.proto_major != 1: + session.close() + raise DongloraError( + f"device speaks DongLoRa Protocol v{info.proto_major}.{info.proto_minor}; " + "this client requires v1.x", + ) + + applied: Modulation | None = None + if auto_configure: + applied = config if config is not None else LoRaConfig.default() + try: + session.set_config(applied, timeout=max(timeout, 2.0)) + except DongloraError: + session.close() + raise + return session, info, applied + + session, info, applied = _open_and_init() + return Dongle( + session, + info, + applied_config=applied, + keepalive=keepalive, + _reopener=_open_and_init, + ) + + +def connect_default() -> Dongle: + """Convenience wrapper for ``connect()`` with all defaults.""" + return connect() + + +def connect_mux_auto(timeout: float = DEFAULT_TIMEOUT) -> Dongle: + """Connect via mux only — never falls through to direct USB serial. + + Sets the sticky-mux flag on success so any later :func:`connect` + call also stays on the mux. + """ + global _used_mux tcp = os.environ.get("DONGLORA_MUX_TCP") if tcp: conn = _try_tcp_mux(tcp, timeout) if conn is not None: - log.debug("connected to TCP mux at %s", tcp) - return conn + _used_mux = True + return _session_on(conn) - # Try Unix socket mux sock_path = _find_mux_socket() if sock_path is None: raise FileNotFoundError("no mux socket found") - return mux_connect(sock_path, timeout) + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(sock_path) + _used_mux = True + return _session_on(MuxConnection(sock, timeout)) -def try_connect(timeout: float = DEFAULT_TIMEOUT) -> Any: - """Like :func:`connect` but non-blocking when no USB device is present. - - Runs the same fallback chain (TCP mux, Unix socket mux, direct USB - serial) but uses a single non-blocking scan instead of polling - indefinitely for a USB device. Raises if nothing is found. - - If a previous call connected via mux, only tries mux (raises if - the mux socket is not currently available). - """ +def try_connect(timeout: float = DEFAULT_TIMEOUT) -> Dongle: + """Like :func:`connect` but single-scan USB (no blocking wait).""" global _used_mux - - # Previously connected via mux — only try mux if _used_mux: - return mux_connect(None, timeout) + return connect_mux_auto(timeout) - # Try TCP mux via environment variable tcp = os.environ.get("DONGLORA_MUX_TCP") if tcp: conn = _try_tcp_mux(tcp, timeout) if conn is not None: - log.debug("connected to TCP mux at %s", tcp) _used_mux = True - return conn + return _session_on(conn) - # Try Unix socket mux — commit if socket exists sock_path = _find_mux_socket() if sock_path is not None: - log.debug("mux socket found at %s — connecting via mux only", sock_path) - conn = mux_connect(sock_path, timeout) + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(sock_path) _used_mux = True - return conn + return _session_on(MuxConnection(sock, timeout)) - # No mux found on first attempt — direct USB serial (single non-blocking scan) port_path = find_port() if port_path is None: raise FileNotFoundError("no DongLoRa device found (no mux, no USB device)") - log.debug("opening serial port %s", port_path) ser = serial.Serial(port_path, baudrate=115200, timeout=timeout) ser.reset_input_buffer() - validate(ser) - return ser + return _session_on(ser) + + +def mux_connect(path: str | None = None, timeout: float = DEFAULT_TIMEOUT) -> Dongle: + """Connect to the mux daemon via Unix domain socket. + + Sets the sticky-mux flag on success so any later :func:`connect` + call also stays on the mux. + """ + global _used_mux + if path is None: + path = _find_mux_socket() + if path is None: + raise FileNotFoundError("no mux socket found") + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(path) + _used_mux = True + return _session_on(MuxConnection(sock, timeout)) + + +def mux_tcp_connect(host: str, port: int, timeout: float = DEFAULT_TIMEOUT) -> Dongle: + """Connect to the mux daemon via TCP. + + Sets the sticky-mux flag on success so any later :func:`connect` + call also stays on the mux. + """ + global _used_mux + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect((host, port)) + _used_mux = True + return _session_on(MuxConnection(sock, timeout)) + + +def _session_on(transport: Any) -> Dongle: + """Bring up a :class:`Dongle` on an already-open transport.""" + session = Session(transport) + try: + info = session.get_info() + except DongloraError: + session.close() + raise + if not isinstance(info, Info): + session.close() + raise DongloraError(f"GET_INFO returned unexpected payload: {info!r}") + if info.proto_major != 1: + session.close() + raise DongloraError( + f"device speaks DongLoRa Protocol v{info.proto_major}.{info.proto_minor}; this client requires v1.x", + ) + applied = LoRaConfig.default() + try: + session.set_config(applied) + except DongloraError: + session.close() + raise + return Dongle(session, info, applied_config=applied, keepalive=True) diff --git a/donglora/crc.py b/donglora/crc.py new file mode 100644 index 0000000..6616bd8 --- /dev/null +++ b/donglora/crc.py @@ -0,0 +1,27 @@ +"""CRC-16/CCITT-FALSE — the wire-level integrity check used by DongLoRa Protocol v2. + +Polynomial ``0x1021``, initial value ``0xFFFF``, no reflection, +XOR-out ``0x0000``. Also known as CRC-16/AUTOSAR or CRC-16/IBM-3740. + +**Not** the same as CRC-16/KERMIT, CRC-16/XMODEM, or plain +"CRC-16/CCITT" — those use different initial values or reflect input. + +Check value: ``crc16(b"123456789") == 0x29B1``. The module-level +assertion below fails at import if the implementation drifts. +""" + +from __future__ import annotations + + +def crc16(data: bytes) -> int: + """Compute CRC-16/CCITT-FALSE over *data*. Returns a u16.""" + crc = 0xFFFF + for byte in data: + crc ^= byte << 8 + for _ in range(8): + crc = ((crc << 1) ^ 0x1021) & 0xFFFF if crc & 0x8000 else (crc << 1) & 0xFFFF + return crc + + +# Self-check: the check value from PROTOCOL.md Appendix B. +assert crc16(b"123456789") == 0x29B1, "CRC-16/CCITT-FALSE check value drift" diff --git a/donglora/dongle.py b/donglora/dongle.py new file mode 100644 index 0000000..3fc58ce --- /dev/null +++ b/donglora/dongle.py @@ -0,0 +1,360 @@ +"""The user-facing :class:`Dongle` — a connected, configured, ready-to-TX radio. + +Designed for "stupidly easy" ergonomics: once you have a ``Dongle`` from +:func:`donglora.connect`, you call ``.tx(data)`` and ``.rx()`` and that's +it. The keepalive daemon, inactivity-timeout auto-recovery, tag +bookkeeping, CRC, COBS — all hidden. +""" + +from __future__ import annotations + +import logging +import threading +from collections.abc import Callable, Iterator + +from donglora.errors import DongloraError, NotConfiguredError +from donglora.events import RxEvent, TxDone +from donglora.info import Info +from donglora.modulation import LoRaConfig, Modulation +from donglora.session import Session + +log = logging.getLogger("donglora.dongle") + +# Type alias for the reopener closure. Returns (new session, info, +# applied config) — the same triple ``connect()`` produces initially. +Reopener = Callable[[], tuple[Session, Info, "Modulation | None"]] + +_KEEPALIVE_INTERVAL_S: float = 0.5 +"""Default keepalive cadence. Spec §3.4 sets the inactivity timer to +1000 ms — we ping at 500 ms for 2x margin. Hosts that send commands +naturally won't trigger keepalive PINGs.""" + + +class Dongle: + """Live DongLoRa radio session. + + Normally constructed via :func:`donglora.connect`. Users who need + full control can pass in a prebuilt transport + :class:`Session`, + but the happy path is: + + .. code-block:: python + + import donglora as dl + d = dl.connect() + d.tx(b"Hello") + for pkt in d.rx(): + print(pkt.rssi_dbm, pkt.data) + + Thread-safe: multiple threads can call ``tx`` / ``rx`` concurrently. + """ + + def __init__( + self, + session: Session, + info: Info, + *, + applied_config: Modulation | None, + keepalive: bool = True, + _reopener: Reopener | None = None, + ): + self._session = session + self._info = info + self._applied_config: Modulation | None = applied_config + self._rx_started = False + self._closed = False + + # Reopener closure supplied by :func:`donglora.connect`. When + # the session's reader thread dies (e.g. mux restart, USB blip) + # and the caller invokes any method, the Dongle transparently + # calls this to build a fresh session. ``None`` disables + # transparent reconnect (users who constructed a Dongle + # manually from an existing Session keep the old behaviour). + self._reopener: Reopener | None = _reopener + # Serialises reconnect work across concurrent method calls, so + # multiple threads hitting a dead session only drive one + # reconnect attempt. + self._recover_lock = threading.Lock() + + # Keepalive daemon: sends PING every ~500 ms unless the caller + # issued other traffic more recently. We track "time of last + # transmit" as a simple monotonic counter. + self._last_write_lock = threading.Lock() + self._last_write_time = _now() + self._keepalive_stop = threading.Event() + self._keepalive_thread: threading.Thread | None = None + if keepalive: + self._keepalive_thread = threading.Thread( + target=self._keepalive_loop, + name="donglora-keepalive", + daemon=True, + ) + self._keepalive_thread.start() + + # ── User-facing API ───────────────────────────────────────────── + + @property + def info(self) -> Info: + """Cached ``GET_INFO`` snapshot from connect-time.""" + return self._info + + @property + def config(self) -> Modulation | None: + """The most-recently-applied radio config, or ``None`` if none.""" + return self._applied_config + + def tx(self, data: bytes, *, skip_cad: bool = False, timeout: float = 10.0) -> TxDone: + """Transmit *data*. Blocks until TX_DONE arrives. + + Raises :class:`donglora.errors.ChannelBusy` if CAD detected + activity, :class:`donglora.errors.Cancelled` if SET_CONFIG + cancelled a queued TX, or other :class:`DongloraError` + subclasses for device errors. + """ + self._check_open() + self._mark_write() + return self._with_recovery( + lambda: self._session.transmit(data, skip_cad=skip_cad, timeout=timeout), + ) + + def rx(self, timeout: float | None = None) -> _RxIterator: + """Return an iterator over incoming packets. + + On first call, lazily sends ``RX_START`` to put the radio into + continuous receive. ``timeout`` is the per-packet deadline; set + to ``None`` to block forever (the common case for listening + scripts). + """ + self._check_open() + if not self._rx_started: + self._mark_write() + self._with_recovery(lambda: self._session.rx_start()) + self._rx_started = True + return _RxIterator(self, timeout) + + def recv(self, timeout: float | None = None) -> RxEvent | None: + """Block on a single RX event. Convenience for ``next(iter(d.rx(timeout)))``. + + Returns ``None`` on timeout. Lazily starts continuous RX on + first call. If the session dies (mux restart, transport blip) + while waiting, transparently reconnects and returns ``None`` — + the caller just loops and calls ``recv()`` again. + """ + self._check_open() + self._ensure_session_alive() + if not self._rx_started: + self._mark_write() + self._with_recovery(lambda: self._session.rx_start()) + self._rx_started = True + pkt = self._session.next_rx(timeout) + if pkt is None and not self._session.is_alive: + # Session died during the wait (reader posted the dead + # sentinel which next_rx filtered to None). Reconnect now + # so the *next* recv() sees a live session. + self._ensure_session_alive() + return pkt + + def rx_stop(self) -> None: + """Stop continuous RX. Rare; most programs just close the dongle.""" + self._check_open() + self._ensure_session_alive() + self._mark_write() + self._session.rx_stop() + self._rx_started = False + + def set_config(self, config: Modulation) -> None: + """Apply a new radio configuration. Blocks on the ``OK`` response.""" + self._check_open() + self._ensure_session_alive() + self._mark_write() + self._session.set_config(config) + self._applied_config = config + # SET_CONFIG aborts any continuous RX on the device side. + self._rx_started = False + + def ping(self) -> None: + """Send a PING. Rarely needed directly — the keepalive daemon + handles session-liveness automatically. + """ + self._check_open() + self._ensure_session_alive() + self._mark_write() + self._session.ping() + + def close(self) -> None: + """Close the session. Idempotent.""" + if self._closed: + return + self._closed = True + self._keepalive_stop.set() + self._session.close() + + # Context manager + def __enter__(self) -> Dongle: + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + # ── Internals ─────────────────────────────────────────────────── + + def _check_open(self) -> None: + if self._closed: + raise DongloraError("dongle is closed") + + def _mark_write(self) -> None: + with self._last_write_lock: + self._last_write_time = _now() + + def _ensure_session_alive(self) -> None: + """If the session's reader thread has died, rebuild the session. + + Blocks until reconnect succeeds or the Dongle is closed. + No-op when the session is healthy. Safe to call concurrently — + only one thread drives the actual reconnect, the rest wait on + the shared lock. + """ + if self._session.is_alive: + return + if self._reopener is None: + # Manually-constructed Dongle without a reopener — surface + # the death as an error rather than silently swallowing it. + raise DongloraError("session closed; no reopener configured") + with self._recover_lock: + # Another thread may have already reconnected by the time + # we got the lock. + if self._session.is_alive: + return + self._recover_session_locked() + + def _recover_session_locked(self) -> None: + """Replace ``self._session`` with a fresh one. + + Called under ``_recover_lock``. Retries the reopener with a + backoff until success or the Dongle is closed. Re-applies the + cached config (if any) and restarts RX (if it was running) + before returning, so the caller's next command sees a live + dongle in the same logical state. + """ + assert self._reopener is not None + log.warning("dongle: session died, reconnecting transparently") + old_session = self._session + import contextlib as _contextlib + + with _contextlib.suppress(Exception): + old_session.close() + + backoff = 0.5 + max_backoff = 2.0 + while not self._closed and not self._keepalive_stop.is_set(): + try: + new_session, new_info, applied = self._reopener() + except (DongloraError, OSError) as exc: + log.info("dongle: reconnect attempt failed (%s); retrying...", exc) + if self._keepalive_stop.wait(backoff): + return # close() fired during backoff + backoff = min(backoff * 1.5, max_backoff) + continue + + self._session = new_session + self._info = new_info + if applied is not None: + self._applied_config = applied + # Restart RX if the caller had it running. + if self._rx_started: + try: + new_session.rx_start() + except DongloraError as exc: + log.info("dongle: restart rx after reconnect failed: %s", exc) + self._rx_started = False + log.info("dongle: reconnect successful") + return + + def _with_recovery(self, op): + """Run *op*, transparently recovering from two kinds of drop: + + * Session reader death (mux restart, transport blip) — + rebuild the session via :meth:`_ensure_session_alive` and + retry once. + * ``ERR(ENOTCONFIGURED)`` from an inactivity timeout — + re-apply the cached config and retry once. + + Other errors pass through untouched. + """ + self._ensure_session_alive() + try: + return op() + except NotConfiguredError: + if self._applied_config is None: + raise + log.info("dongle: auto-recovering after inactivity timeout") + self._session.set_config(self._applied_config) + return op() + except DongloraError: + # If the session died mid-call, reconnect and retry once. + if not self._session.is_alive and self._reopener is not None: + self._ensure_session_alive() + return op() + raise + + def _keepalive_loop(self) -> None: + while not self._keepalive_stop.wait(_KEEPALIVE_INTERVAL_S): + # If the session died, let the keepalive thread drive the + # reconnect proactively — without this, we'd only notice + # when the caller next invoked a public method. + if not self._session.is_alive and self._reopener is not None: + try: + self._ensure_session_alive() + except DongloraError as exc: + log.debug("keepalive reconnect failed: %s", exc) + continue + with self._last_write_lock: + since = _now() - self._last_write_time + if since < _KEEPALIVE_INTERVAL_S: + continue + try: + self._session.ping(timeout=2.0) + with self._last_write_lock: + self._last_write_time = _now() + except NotConfiguredError: + # Session timed out during quiet period. Try to re-apply + # the cached config so the next user call sees a live + # session. If the caller issues a command first, the + # with_recovery path will handle it instead — so treat + # failures here as best-effort. + if self._applied_config is not None: + try: + self._session.set_config(self._applied_config) + except Exception as exc: + log.debug("keepalive recovery failed: %s", exc) + except DongloraError as exc: + log.debug("keepalive ping failed: %s", exc) + except Exception as exc: + log.debug("keepalive thread quiet error: %s", exc) + + +class _RxIterator: + """Iterator view over a Dongle's incoming RX events.""" + + def __init__(self, dongle: Dongle, timeout: float | None): + self._dongle = dongle + self._timeout = timeout + + def __iter__(self) -> Iterator[RxEvent]: + return self + + def __next__(self) -> RxEvent: + pkt = self._dongle._session.next_rx(self._timeout) + if pkt is None: + raise StopIteration + return pkt + + +def _now() -> float: + import time + + return time.monotonic() + + +# Re-export the default LoRa config builder so users don't have to +# import from modulation.py. +__all__ = ["Dongle", "LoRaConfig"] diff --git a/donglora/errors.py b/donglora/errors.py new file mode 100644 index 0000000..ceb7e47 --- /dev/null +++ b/donglora/errors.py @@ -0,0 +1,123 @@ +"""Wire error codes + Python exception hierarchy. + +``ErrorCode`` mirrors PROTOCOL.md §7. ``DongloraError`` subclasses are +raised by the high-level :class:`donglora.dongle.Dongle` so callers can +catch specific failures (e.g. ``ChannelBusy``) without inspecting codes. +""" + +from __future__ import annotations + +import enum + + +class ErrorCode(enum.IntEnum): + """Wire-level error code (u16 LE). Values assigned by PROTOCOL.md §7.""" + + # Synchronous (0x0001..=0x00FF) + EPARAM = 0x0001 + ELENGTH = 0x0002 + ENOT_CONFIGURED = 0x0003 + EMODULATION = 0x0004 + EUNKNOWN_CMD = 0x0005 + EBUSY = 0x0006 + + # Asynchronous (0x0100..=0x01FF) + ERADIO = 0x0101 + EFRAME = 0x0102 + EINTERNAL = 0x0103 + + @classmethod + def from_u16(cls, raw: int) -> ErrorCode | int: + """Return the named variant if recognised, otherwise the raw int.""" + try: + return cls(raw) + except ValueError: + return raw + + +# ── Python exception hierarchy ───────────────────────────────────── + + +class DongloraError(Exception): + """Base class for all DongLoRa client errors.""" + + +class FrameError(DongloraError): + """COBS decode, CRC check, or length check failed on an inbound frame.""" + + +class TimeoutError_(DongloraError): + """Command response not received within the configured deadline.""" + + +class DeviceError(DongloraError): + """The device returned an ``ERR`` frame.""" + + def __init__(self, code: ErrorCode | int, *, tag: int | None = None): + self.code = code + self.tag = tag + name = code.name if isinstance(code, ErrorCode) else f"0x{int(code):04X}" + super().__init__(f"device error {name} (tag={tag})") + + +class NotConfiguredError(DeviceError): + """Mapped from ``ERR(ENOTCONFIGURED)``.""" + + +class BusyError(DeviceError): + """Mapped from ``ERR(EBUSY)``.""" + + +class ParamError(DeviceError): + """Mapped from ``ERR(EPARAM)``.""" + + +class LengthError(DeviceError): + """Mapped from ``ERR(ELENGTH)``.""" + + +class ModulationError(DeviceError): + """Mapped from ``ERR(EMODULATION)``.""" + + +class UnknownCommandError(DeviceError): + """Mapped from ``ERR(EUNKNOWN_CMD)``.""" + + +class RadioError(DeviceError): + """Mapped from ``ERR(ERADIO)`` — SPI error or unexpected hardware state.""" + + +class InternalError(DeviceError): + """Mapped from ``ERR(EINTERNAL)`` — firmware bug or invariant violation.""" + + +_SPECIALISATIONS: dict[int, type[DeviceError]] = { + ErrorCode.EPARAM: ParamError, + ErrorCode.ELENGTH: LengthError, + ErrorCode.ENOT_CONFIGURED: NotConfiguredError, + ErrorCode.EMODULATION: ModulationError, + ErrorCode.EUNKNOWN_CMD: UnknownCommandError, + ErrorCode.EBUSY: BusyError, + ErrorCode.ERADIO: RadioError, + ErrorCode.EINTERNAL: InternalError, +} + + +def device_error(code: ErrorCode | int, *, tag: int | None = None) -> DeviceError: + """Construct the most specific :class:`DeviceError` subclass for *code*.""" + cls = _SPECIALISATIONS.get(int(code), DeviceError) + return cls(code, tag=tag) + + +# ── TX-specific non-error results ────────────────────────────────── + + +class ChannelBusy(DongloraError): + """CAD detected activity and TX was not attempted. Retry with new tag.""" + + +class Cancelled(DongloraError): + """A queued TX was cancelled before it reached the air (SET_CONFIG + or disconnect interrupted the queue). + """ diff --git a/donglora/events.py b/donglora/events.py new file mode 100644 index 0000000..edafa38 --- /dev/null +++ b/donglora/events.py @@ -0,0 +1,186 @@ +"""Device-to-host message payloads: ``OK`` / ``ERR`` / ``RX`` / ``TX_DONE``.""" + +from __future__ import annotations + +import enum +import struct +from dataclasses import dataclass + +from donglora.errors import ErrorCode +from donglora.info import Info +from donglora.modulation import Modulation, decode_modulation, encode_modulation + +# Message type identifiers +TYPE_OK: int = 0x80 +TYPE_ERR: int = 0x81 +TYPE_RX: int = 0xC0 +TYPE_TX_DONE: int = 0xC1 + + +# ── SET_CONFIG response payload ──────────────────────────────────── + + +class SetConfigResultCode(enum.IntEnum): + APPLIED = 0 + ALREADY_MATCHED = 1 + LOCKED_MISMATCH = 2 + + +class Owner(enum.IntEnum): + NONE = 0 + MINE = 1 + OTHER = 2 + + +@dataclass(frozen=True) +class SetConfigResult: + result: SetConfigResultCode + owner: Owner + current: Modulation + + def encode(self) -> bytes: + return bytes([int(self.result), int(self.owner)]) + encode_modulation(self.current) + + @classmethod + def decode(cls, data: bytes) -> SetConfigResult: + if len(data) < 2: + raise ValueError("SetConfigResult too short") + return cls( + result=SetConfigResultCode(data[0]), + owner=Owner(data[1]), + current=decode_modulation(data[2:]), + ) + + +# ── TX_DONE payload ──────────────────────────────────────────────── + + +class TxResult(enum.IntEnum): + TRANSMITTED = 0 + CHANNEL_BUSY = 1 + CANCELLED = 2 + + +@dataclass(frozen=True) +class TxDone: + result: TxResult + airtime_us: int + + def encode(self) -> bytes: + return bytes([int(self.result)]) + struct.pack(" TxDone: + if len(data) != 5: + raise ValueError(f"TxDone wire size must be 5, got {len(data)}") + result = TxResult(data[0]) + (airtime_us,) = struct.unpack_from(" bytes: + return ( + struct.pack(" RxEvent: + if len(data) < cls.METADATA_SIZE: + raise ValueError(f"RxEvent payload too short: {len(data)}") + (rssi_raw,) = struct.unpack_from(" bytes: + return struct.pack(" ErrorCode | int: + if len(data) != 2: + raise ValueError(f"ERR payload must be 2 bytes, got {len(data)}") + (raw,) = struct.unpack_from(" bytes: + """Encode a complete wire-ready DongLoRa Protocol frame. + + Returns ``COBS(type || tag_le || payload || crc_le) || 0x00``. + """ + if not (0 <= type_id <= 0xFF): + raise ValueError(f"type_id out of range: {type_id}") + if not (0 <= tag <= 0xFFFF): + raise ValueError(f"tag out of range: {tag}") + if len(payload) > MAX_PAYLOAD_FIELD: + raise FrameTooLargeError(f"payload too large ({len(payload)} > {MAX_PAYLOAD_FIELD})") + body = struct.pack(" Frame: + """Decode a single COBS+sentinel-terminated DongLoRa Protocol frame. + + *frame_bytes* must NOT include the trailing ``0x00``. + """ + try: + decoded = cobs.decode(frame_bytes) + except cobs.DecodeError as exc: + raise FrameCobsError(str(exc)) from exc + if len(decoded) < FRAME_HEADER_SIZE + FRAME_TRAILER_SIZE: + raise FrameTooShortError(f"decoded frame too short: {len(decoded)} bytes") + body, crc_bytes = decoded[:-FRAME_TRAILER_SIZE], decoded[-FRAME_TRAILER_SIZE:] + expected = crc16(body) + (got,) = struct.unpack(" bytes: ... + + +def read_frame(reader: Readable) -> Frame | None: + """Block on *reader* until a full COBS-delimited frame arrives. + + Returns the decoded :class:`Frame`, or ``None`` on transport timeout. + Raises :class:`FrameError` on CRC/COBS/length failure so the caller + can distinguish corruption from timeout. + """ + buf = bytearray() + while True: + b = reader.read(1) + if not b: + return None + if b == b"\x00": + if not buf: + return None + return decode_frame(bytes(buf)) + buf.extend(b) + + +def iter_frames(data: bytes) -> Iterator[Frame]: + """Yield each complete frame embedded in *data* (for tests / replay). + + Raises :class:`FrameError` on the first corrupt frame encountered. + """ + start = 0 + for i, byte in enumerate(data): + if byte == 0x00: + if i > start: + yield decode_frame(bytes(data[start:i])) + start = i + 1 diff --git a/donglora/info.py b/donglora/info.py new file mode 100644 index 0000000..caa6112 --- /dev/null +++ b/donglora/info.py @@ -0,0 +1,196 @@ +"""``GET_INFO`` response payload + capability bitmap constants.""" + +from __future__ import annotations + +import enum +import struct +from dataclasses import dataclass, field + +MAX_MCU_UID_LEN: int = 32 +MAX_RADIO_UID_LEN: int = 16 + + +# ── Capability bits (PROTOCOL.md §9) ────────────────────────────── + + +class Capability: + """Bit masks for ``Info.capability_bitmap``. + + Use like: ``if info.supports(Capability.LORA | Capability.CAD_BEFORE_TX): ...``. + """ + + # Modulations + LORA = 1 << 0 + FSK = 1 << 1 + GFSK = 1 << 2 + LR_FHSS = 1 << 3 + FLRC = 1 << 4 + MSK = 1 << 5 + GMSK = 1 << 6 + BLE_COMPATIBLE = 1 << 7 + + # Radio features + CAD_BEFORE_TX = 1 << 16 + IQ_INVERSION = 1 << 17 + RANGING = 1 << 18 + GNSS_SCAN = 1 << 19 + WIFI_MAC_SCAN = 1 << 20 + SPECTRAL_SCAN = 1 << 21 + FULL_DUPLEX = 1 << 22 + + # Protocol features + MULTI_CLIENT = 1 << 32 + + +# ── Radio chip IDs (PROTOCOL.md §8) ─────────────────────────────── + + +class RadioChipId(enum.IntEnum): + UNKNOWN = 0x0000 + SX1261 = 0x0001 + SX1262 = 0x0002 + SX1268 = 0x0003 + LLCC68 = 0x0004 + SX1272 = 0x0010 + SX1276 = 0x0011 + SX1277 = 0x0012 + SX1278 = 0x0013 + SX1279 = 0x0014 + SX1280 = 0x0020 + SX1281 = 0x0021 + LR1110 = 0x0030 + LR1120 = 0x0031 + LR1121 = 0x0032 + LR2021 = 0x0040 + + +# ── Info struct ─────────────────────────────────────────────────── + + +@dataclass(frozen=True) +class Info: + """``GET_INFO`` response payload. Stable for the lifetime of a session.""" + + proto_major: int + proto_minor: int + fw_major: int + fw_minor: int + fw_patch: int + radio_chip_id: int + capability_bitmap: int + supported_sf_bitmap: int + supported_bw_bitmap: int + max_payload_bytes: int + rx_queue_capacity: int + tx_queue_capacity: int + freq_min_hz: int + freq_max_hz: int + tx_power_min_dbm: int + tx_power_max_dbm: int + mcu_uid: bytes + radio_uid: bytes + + MIN_WIRE_SIZE: int = field(default=37, init=False, repr=False, compare=False) + + @property + def chip(self) -> RadioChipId | int: + """Project ``radio_chip_id`` into the :class:`RadioChipId` enum if + recognised; otherwise return the raw u16 so nothing is lost for + forward-compat with future chips. + """ + try: + return RadioChipId(self.radio_chip_id) + except ValueError: + return self.radio_chip_id + + def supports(self, mask: int) -> bool: + """Return True if all bits in *mask* are set in the capability bitmap.""" + return (self.capability_bitmap & mask) == mask + + def supported_sf(self) -> list[int]: + """List of spreading factors (5..12) the device supports.""" + return [sf for sf in range(16) if self.supported_sf_bitmap & (1 << sf)] + + def encode(self) -> bytes: + if len(self.mcu_uid) > MAX_MCU_UID_LEN: + raise ValueError(f"mcu_uid too long: {len(self.mcu_uid)}") + if len(self.radio_uid) > MAX_RADIO_UID_LEN: + raise ValueError(f"radio_uid too long: {len(self.radio_uid)}") + header = struct.pack( + " Info: + if len(data) < 37: + raise ValueError(f"Info payload too short: {len(data)}") + ( + proto_major, + proto_minor, + fw_major, + fw_minor, + fw_patch, + radio_chip_id, + capability_bitmap, + supported_sf_bitmap, + supported_bw_bitmap, + max_payload_bytes, + rx_queue_capacity, + tx_queue_capacity, + freq_min_hz, + freq_max_hz, + tx_power_min_dbm, + tx_power_max_dbm, + mcu_uid_len, + ) = struct.unpack_from(" MAX_MCU_UID_LEN: + raise ValueError(f"mcu_uid_len out of range: {mcu_uid_len}") + if len(data) < 36 + mcu_uid_len + 1: + raise ValueError(f"Info payload truncated after mcu_uid: {len(data)}") + mcu_uid = bytes(data[36 : 36 + mcu_uid_len]) + radio_len_idx = 36 + mcu_uid_len + radio_uid_len = data[radio_len_idx] + if radio_uid_len > MAX_RADIO_UID_LEN: + raise ValueError(f"radio_uid_len out of range: {radio_uid_len}") + radio_start = radio_len_idx + 1 + if len(data) < radio_start + radio_uid_len: + raise ValueError(f"Info payload truncated after radio_uid: {len(data)}") + radio_uid = bytes(data[radio_start : radio_start + radio_uid_len]) + return cls( + proto_major=proto_major, + proto_minor=proto_minor, + fw_major=fw_major, + fw_minor=fw_minor, + fw_patch=fw_patch, + radio_chip_id=radio_chip_id, + capability_bitmap=capability_bitmap, + supported_sf_bitmap=supported_sf_bitmap, + supported_bw_bitmap=supported_bw_bitmap, + max_payload_bytes=max_payload_bytes, + rx_queue_capacity=rx_queue_capacity, + tx_queue_capacity=tx_queue_capacity, + freq_min_hz=freq_min_hz, + freq_max_hz=freq_max_hz, + tx_power_min_dbm=tx_power_min_dbm, + tx_power_max_dbm=tx_power_max_dbm, + mcu_uid=mcu_uid, + radio_uid=radio_uid, + ) diff --git a/donglora/modulation.py b/donglora/modulation.py new file mode 100644 index 0000000..4c62b6c --- /dev/null +++ b/donglora/modulation.py @@ -0,0 +1,406 @@ +"""Per-modulation config dataclasses for ``SET_CONFIG`` payloads. + +Mirrors the Rust crate's ``modulation`` module byte-for-byte. Users +build a :class:`LoRaConfig` (or FSK/LR-FHSS/FLRC variant) and pass it +to :meth:`donglora.dongle.Dongle.set_config` or to :func:`connect` as +an override. + +The `.default()` classmethod on each returns a sensible starting point +— for LoRa that's the Appendix C.2.3 canonical EU868 config, which is +also what :func:`connect()` auto-applies. +""" + +from __future__ import annotations + +import enum +import struct +from dataclasses import dataclass, field + +# ── Modulation ID (u8 prefix byte on every SET_CONFIG payload) ──── + + +class ModulationId(enum.IntEnum): + LORA = 0x01 + FSK_GFSK = 0x02 + LR_FHSS = 0x03 + FLRC = 0x04 + + +# ── LoRa enums ───────────────────────────────────────────────────── + + +class LoRaBandwidth(enum.IntEnum): + """LoRa bandwidth enum — integer is the wire value, not kHz.""" + + KHZ_7_81 = 0 + KHZ_10_42 = 1 + KHZ_15_63 = 2 + KHZ_20_83 = 3 + KHZ_31_25 = 4 + KHZ_41_67 = 5 + KHZ_62_5 = 6 + KHZ_125 = 7 + KHZ_250 = 8 + KHZ_500 = 9 + KHZ_200 = 10 # SX128x + KHZ_400 = 11 # SX128x + KHZ_800 = 12 # SX128x + KHZ_1600 = 13 # SX128x + + @property + def khz(self) -> float: + """Nominal bandwidth in kHz. Use with ``:g`` or ``:.1f`` for display.""" + return _BW_KHZ[self] + + +_BW_KHZ: dict[LoRaBandwidth, float] = { + LoRaBandwidth.KHZ_7_81: 7.81, + LoRaBandwidth.KHZ_10_42: 10.42, + LoRaBandwidth.KHZ_15_63: 15.63, + LoRaBandwidth.KHZ_20_83: 20.83, + LoRaBandwidth.KHZ_31_25: 31.25, + LoRaBandwidth.KHZ_41_67: 41.67, + LoRaBandwidth.KHZ_62_5: 62.5, + LoRaBandwidth.KHZ_125: 125.0, + LoRaBandwidth.KHZ_250: 250.0, + LoRaBandwidth.KHZ_500: 500.0, + LoRaBandwidth.KHZ_200: 200.0, + LoRaBandwidth.KHZ_400: 400.0, + LoRaBandwidth.KHZ_800: 800.0, + LoRaBandwidth.KHZ_1600: 1600.0, +} + + +class LoRaCodingRate(enum.IntEnum): + CR_4_5 = 0 + CR_4_6 = 1 + CR_4_7 = 2 + CR_4_8 = 3 + + +class LoRaHeaderMode(enum.IntEnum): + EXPLICIT = 0 + IMPLICIT = 1 + + +# ── LoRa config ──────────────────────────────────────────────────── + + +@dataclass(frozen=True) +class LoRaConfig: + """LoRa modulation parameters. 15-byte wire struct.""" + + freq_hz: int + sf: int # 5-12 + bw: LoRaBandwidth + cr: LoRaCodingRate + preamble_len: int = 8 + sync_word: int = 0x1424 + tx_power_dbm: int = 14 + header_mode: LoRaHeaderMode = LoRaHeaderMode.EXPLICIT + payload_crc: bool = True + iq_invert: bool = False + + WIRE_SIZE: int = field(default=15, init=False, repr=False, compare=False) + + @classmethod + def default(cls) -> LoRaConfig: + """PROTOCOL.md §C.2.3 canonical EU868 config — also what + :func:`donglora.connect` auto-applies. + """ + return cls( + freq_hz=868_100_000, + sf=7, + bw=LoRaBandwidth.KHZ_125, + cr=LoRaCodingRate.CR_4_5, + preamble_len=8, + sync_word=0x1424, + tx_power_dbm=14, + header_mode=LoRaHeaderMode.EXPLICIT, + payload_crc=True, + iq_invert=False, + ) + + def encode(self) -> bytes: + return ( + struct.pack(" LoRaConfig: + if len(data) != 15: + raise ValueError(f"LoRaConfig wire size must be 15 bytes, got {len(data)}") + (freq_hz,) = struct.unpack_from(" bytes: + if len(self.sync_word) > 8: + raise ValueError("FSK sync word max 8 bytes") + return ( + struct.pack(" FskConfig: + if len(data) < 16: + raise ValueError(f"FskConfig too short: {len(data)}") + (freq_hz,) = struct.unpack_from(" 8: + raise ValueError(f"FSK sync_word_len out of range: {sync_word_len}") + if len(data) != 16 + sync_word_len: + raise ValueError(f"FskConfig wrong total length: {len(data)}") + return cls( + freq_hz=freq_hz, + bitrate_bps=bitrate_bps, + freq_dev_hz=freq_dev_hz, + rx_bw=rx_bw, + preamble_len=preamble_len, + sync_word=bytes(data[16 : 16 + sync_word_len]), + ) + + +# ── LR-FHSS config ───────────────────────────────────────────────── + + +class LrFhssBandwidth(enum.IntEnum): + KHZ_39 = 0 + KHZ_86 = 1 + KHZ_137 = 2 + KHZ_184 = 3 + KHZ_336 = 4 + KHZ_387 = 5 + KHZ_723 = 6 + KHZ_1523 = 7 + + +class LrFhssCodingRate(enum.IntEnum): + CR_5_6 = 0 + CR_2_3 = 1 + CR_1_2 = 2 + CR_1_3 = 3 + + +class LrFhssGrid(enum.IntEnum): + KHZ_25 = 0 + KHZ_3_9 = 1 + + +@dataclass(frozen=True) +class LrFhssConfig: + """LR-FHSS modulation parameters. 10-byte wire struct. TX-only on all + current Semtech silicon. + """ + + freq_hz: int + bw: LrFhssBandwidth + cr: LrFhssCodingRate + grid: LrFhssGrid + hopping: bool = True + tx_power_dbm: int = 14 + + def encode(self) -> bytes: + return ( + struct.pack(" LrFhssConfig: + if len(data) != 10: + raise ValueError(f"LrFhssConfig wire size must be 10 bytes, got {len(data)}") + (freq_hz,) = struct.unpack_from(" bytes: + return ( + struct.pack(" FlrcConfig: + if len(data) != 13: + raise ValueError(f"FlrcConfig wire size must be 13 bytes, got {len(data)}") + (freq_hz,) = struct.unpack_from(" bytes: + """Encode a modulation-agnostic ``[modulation_id || params]`` payload.""" + if isinstance(mod, LoRaConfig): + return bytes([ModulationId.LORA]) + mod.encode() + if isinstance(mod, FskConfig): + return bytes([ModulationId.FSK_GFSK]) + mod.encode() + if isinstance(mod, LrFhssConfig): + return bytes([ModulationId.LR_FHSS]) + mod.encode() + if isinstance(mod, FlrcConfig): + return bytes([ModulationId.FLRC]) + mod.encode() + raise TypeError(f"unknown modulation type: {type(mod)}") + + +def decode_modulation(data: bytes) -> Modulation: + if not data: + raise ValueError("modulation payload is empty") + mod_id = data[0] + params = data[1:] + if mod_id == ModulationId.LORA: + return LoRaConfig.decode(params) + if mod_id == ModulationId.FSK_GFSK: + return FskConfig.decode(params) + if mod_id == ModulationId.LR_FHSS: + return LrFhssConfig.decode(params) + if mod_id == ModulationId.FLRC: + return FlrcConfig.decode(params) + raise ValueError(f"unknown modulation_id: 0x{mod_id:02X}") diff --git a/donglora/protocol.py b/donglora/protocol.py deleted file mode 100644 index 1d765a1..0000000 --- a/donglora/protocol.py +++ /dev/null @@ -1,234 +0,0 @@ -"""Wire protocol types and fixed-size little-endian serialization. - -Mirrors the firmware's protocol and the Rust client's ``protocol.rs`` -using Python types. Every integer is fixed-width LE. -""" - -from __future__ import annotations - -import enum -import struct -from typing import Any, TypedDict - -# ── Size constants ──────────────────────────────────────────────── - -MAX_PAYLOAD: int = 256 -"""Maximum LoRa payload size in bytes.""" - -RADIO_CONFIG_SIZE: int = 13 -"""RadioConfig wire size (fixed).""" - -TX_POWER_MAX: int = -128 -"""Sentinel: set TX power to the board's maximum (i8::MIN on the wire).""" - -PREAMBLE_DEFAULT: int = 0 -"""Sentinel: use firmware default preamble length (16 symbols).""" - -# ── Tag constants ───────────────────────────────────────────────── - -CMD_TAG_PING: int = 0 -CMD_TAG_GET_CONFIG: int = 1 -CMD_TAG_SET_CONFIG: int = 2 -CMD_TAG_START_RX: int = 3 -CMD_TAG_STOP_RX: int = 4 -CMD_TAG_TRANSMIT: int = 5 -CMD_TAG_DISPLAY_ON: int = 6 -CMD_TAG_DISPLAY_OFF: int = 7 -CMD_TAG_GET_MAC: int = 8 - -RESP_TAG_PONG: int = 0 -RESP_TAG_CONFIG: int = 1 -RESP_TAG_RX_PACKET: int = 2 -RESP_TAG_TX_DONE: int = 3 -RESP_TAG_OK: int = 4 -RESP_TAG_ERROR: int = 5 -RESP_TAG_MAC_ADDRESS: int = 6 - -# ── Enums ───────────────────────────────────────────────────────── - - -class Bandwidth(enum.IntEnum): - """LoRa signal bandwidth.""" - - Khz7 = 0 - Khz10 = 1 - Khz15 = 2 - Khz20 = 3 - Khz31 = 4 - Khz41 = 5 - Khz62 = 6 - Khz125 = 7 - Khz250 = 8 - Khz500 = 9 - - -class ErrorCode(enum.IntEnum): - """Error codes reported by the firmware.""" - - InvalidConfig = 0 - RadioBusy = 1 - TxTimeout = 2 - CrcError = 3 - NotConfigured = 4 - NoDisplay = 5 - - -ERROR_INVALID_CONFIG: int = ErrorCode.InvalidConfig -ERROR_RADIO_BUSY: int = ErrorCode.RadioBusy -ERROR_TX_TIMEOUT: int = ErrorCode.TxTimeout -ERROR_NOT_CONFIGURED: int = ErrorCode.NotConfigured -ERROR_NO_DISPLAY: int = ErrorCode.NoDisplay - -# ── RadioConfig ─────────────────────────────────────────────────── - - -class RadioConfig(TypedDict): - """Complete LoRa radio configuration. - - Wire layout (13 bytes, all little-endian):: - - [freq_hz:4] [bw:1] [sf:1] [cr:1] [sync_word:2] - [tx_power_dbm:1] [preamble_len:2] [cad:1] - """ - - freq_hz: int - bw: int - sf: int - cr: int - sync_word: int - tx_power_dbm: int - preamble_len: int - cad: int - - -DEFAULT_CONFIG: RadioConfig = { - "freq_hz": 915_000_000, - "bw": Bandwidth.Khz125, - "sf": 7, - "cr": 5, - "sync_word": 0x1424, - "tx_power_dbm": TX_POWER_MAX, - "preamble_len": PREAMBLE_DEFAULT, - "cad": 1, -} -"""Default radio config: 915 MHz, 125 kHz BW, SF7, CR 4/5, max power, CAD on.""" - -# ── Config wire encoding ────────────────────────────────────────── - -_CONFIG_STRUCT = struct.Struct(" bytes: - """Encode a RadioConfig to 13 fixed-size LE bytes.""" - return _CONFIG_STRUCT.pack( - cfg["freq_hz"], - cfg["bw"], - cfg["sf"], - cfg["cr"], - cfg["sync_word"], - cfg["tx_power_dbm"] & 0xFF, - cfg.get("preamble_len", PREAMBLE_DEFAULT), - cfg.get("cad", 1), - ) - - -def decode_config(data: bytes) -> RadioConfig | None: - """Decode a RadioConfig from 13 LE bytes. Returns ``None`` if too short or invalid.""" - if len(data) < RADIO_CONFIG_SIZE: - return None - freq_hz, bw_raw, sf, cr, sync_word, _pwr_unsigned, preamble_len, cad = ( - _CONFIG_STRUCT.unpack_from(data, 0) - ) - try: - Bandwidth(bw_raw) - except ValueError: - return None - tx_power_dbm = struct.unpack_from(" bytes: - """Encode a command to fixed-size LE bytes. - - Raises ``ValueError`` for unknown commands or oversized payloads. - """ - tag = _COMMAND_TAGS.get(cmd) - if tag is None: - raise ValueError(f"unknown command: {cmd!r}") - out = bytes([tag]) - if cmd == "SetConfig": - out += encode_config(kwargs["config"]) - elif cmd == "Transmit": - config: RadioConfig | None = kwargs.get("config") - if config is None: - out += b"\x00" - else: - out += b"\x01" + encode_config(config) - payload: bytes = kwargs["payload"] - if len(payload) > MAX_PAYLOAD: - raise ValueError(f"payload too large ({len(payload)} bytes, max {MAX_PAYLOAD})") - out += struct.pack(" dict[str, Any]: - """Decode a fixed-size LE response into a dict.""" - if not data: - return {"type": "Empty"} - tag = data[0] - rest = data[1:] - if tag == RESP_TAG_PONG: - return {"type": "Pong"} - if tag == RESP_TAG_CONFIG: - cfg = decode_config(rest) - if cfg is not None: - return {"type": "Config", **cfg} - return {"type": "Config", "raw": rest.hex()} - if tag == RESP_TAG_RX_PACKET: - if len(rest) < 6: - return {"type": "RxPacket", "rssi": 0, "snr": 0, "payload": b""} - rssi = struct.unpack_from("= 6: - mac = ":".join(f"{b:02X}" for b in rest[:6]) - return {"type": "MacAddress", "mac": mac} - return {"type": "MacAddress", "raw": rest.hex()} - return {"type": f"Unknown({tag})"} diff --git a/donglora/session.py b/donglora/session.py new file mode 100644 index 0000000..92c21c1 --- /dev/null +++ b/donglora/session.py @@ -0,0 +1,371 @@ +"""Internal session plumbing: tag counter, outstanding-tag correlation, +background reader thread, async event queue. + +Not exposed to end users — :class:`donglora.dongle.Dongle` owns a +Session and handles all request/response traffic through it. +""" + +from __future__ import annotations + +import logging +import queue +import threading +from dataclasses import dataclass +from typing import Any + +from donglora.commands import ( + TYPE_PING, + encode_get_info_payload, + encode_ping_payload, + encode_rx_start_payload, + encode_rx_stop_payload, + encode_set_config_payload, + encode_tx_payload, +) +from donglora.errors import ( + DongloraError, + ErrorCode, + FrameError, + TimeoutError_, + device_error, +) +from donglora.events import ( + TYPE_ERR, + TYPE_OK, + TYPE_RX, + TYPE_TX_DONE, + RxEvent, + TxDone, + decode_err_payload, + parse_ok_payload, +) +from donglora.frame import ( + Frame, + encode_frame, + read_frame, +) +from donglora.frame import ( + FrameError as FrameCodecError, +) +from donglora.modulation import Modulation + +log = logging.getLogger("donglora.session") + + +# Sentinel placed on ``_rx_queue`` when the reader thread exits, so any +# blocked ``next_rx`` caller unblocks promptly instead of waiting out +# its full timeout. The sentinel is filtered out before reaching user +# code. +_DEAD_SENTINEL: object = object() + + +# ── Pending-command tracking ─────────────────────────────────────── + + +@dataclass +class _Pending: + """One outstanding command awaiting a response.""" + + tag: int + cmd_type: int + event: threading.Event + # Result slots (one of these is filled by the reader thread): + ok_payload: Any = None # parsed OK payload (None for empty OKs) + err: DongloraError | None = None + tx_done: TxDone | None = None # only for TX + # TX commands get two responses: OK (enqueued) then TX_DONE. The + # event is signalled after TX_DONE for TX; after OK/ERR for others. + + +class Session: + """Owns the transport, demultiplexes inbound frames, correlates tags + to pending commands, queues async events. + + Public API is thread-safe. + """ + + def __init__(self, transport: Any): + self._transport = transport + self._transport_lock = threading.Lock() + + # Tag generator: monotonic u16, skips 0. + self._next_tag = 1 + self._tag_lock = threading.Lock() + + # Outstanding commands by tag. + self._pending: dict[int, _Pending] = {} + self._pending_lock = threading.Lock() + + # Async RX events + async ERR frames. The queue may also hold + # `_DEAD_SENTINEL` to signal reader death without a user value. + self._rx_queue: queue.Queue[Any] = queue.Queue() + self._async_errors: queue.Queue[DongloraError] = queue.Queue() + + # Reader thread. `_reader_dead` fires when the reader loop + # exits for any reason (transport closed, session closed, + # unexpected exception). The Dongle watches this to drive + # transparent reconnect. + self._closed = threading.Event() + self._reader_dead = threading.Event() + self._reader = threading.Thread( + target=self._reader_loop, + name="donglora-reader", + daemon=True, + ) + self._reader.start() + + # ── Public: send / send_tx / rx / close ────────────────────────── + + def close(self) -> None: + self._closed.set() + # Wake any pending commands so they don't hang forever. + with self._pending_lock: + for p in self._pending.values(): + p.err = DongloraError("session closed") + p.event.set() + self._pending.clear() + # Close the underlying transport so the reader thread's blocking + # read unblocks and it can exit cleanly. + import contextlib as _contextlib + + with _contextlib.suppress(Exception): + self._transport.close() + + @property + def is_alive(self) -> bool: + """True while the reader thread is running and no close was requested.""" + return not self._reader_dead.is_set() and not self._closed.is_set() + + @property + def reader_dead_event(self) -> threading.Event: + """Event that fires when the reader thread exits (for any reason).""" + return self._reader_dead + + def ping(self, *, timeout: float = 2.0) -> None: + self._send_and_wait(TYPE_PING, encode_ping_payload(), timeout=timeout) + + def get_info(self, *, timeout: float = 2.0): + return self._send_and_wait(0x02, encode_get_info_payload(), timeout=timeout) + + def set_config(self, modulation: Modulation, *, timeout: float = 2.0): + return self._send_and_wait( + 0x03, + encode_set_config_payload(modulation), + timeout=timeout, + ) + + def transmit(self, data: bytes, *, skip_cad: bool = False, timeout: float = 10.0) -> TxDone: + return self._send_tx(data, skip_cad=skip_cad, timeout=timeout) + + def rx_start(self, *, timeout: float = 2.0) -> None: + self._send_and_wait(0x05, encode_rx_start_payload(), timeout=timeout) + + def rx_stop(self, *, timeout: float = 2.0) -> None: + self._send_and_wait(0x06, encode_rx_stop_payload(), timeout=timeout) + + def next_rx(self, timeout: float | None) -> RxEvent | None: + """Pop the next RX event from the queue, or ``None`` on timeout. + + ``timeout=None`` blocks forever. Returns ``None`` if the reader + thread has died — the caller (normally :class:`Dongle`) should + then trigger a reconnect. + """ + try: + item = self._rx_queue.get(block=True, timeout=timeout) + except queue.Empty: + return None + if item is _DEAD_SENTINEL: + # Reader is gone — re-post the sentinel for any other + # waiters and return None. + self._rx_queue.put(_DEAD_SENTINEL) + return None + return item + + def drain_async_errors(self) -> list[DongloraError]: + """Return and clear any async errors observed since the last call.""" + drained = [] + while True: + try: + drained.append(self._async_errors.get_nowait()) + except queue.Empty: + break + return drained + + # ── Internal: send machinery ───────────────────────────────────── + + def _alloc_tag(self) -> int: + with self._tag_lock: + tag = self._next_tag + self._next_tag = (tag + 1) & 0xFFFF + if self._next_tag == 0: + self._next_tag = 1 + return tag + + def _write_frame(self, type_id: int, tag: int, payload: bytes) -> None: + wire = encode_frame(type_id, tag, payload) + with self._transport_lock: + self._transport.write(wire) + flush = getattr(self._transport, "flush", None) + if callable(flush): + flush() + + def _register_pending(self, tag: int, cmd_type: int) -> _Pending: + p = _Pending(tag=tag, cmd_type=cmd_type, event=threading.Event()) + with self._pending_lock: + self._pending[tag] = p + return p + + def _forget_pending(self, tag: int) -> None: + with self._pending_lock: + self._pending.pop(tag, None) + + def _send_and_wait(self, type_id: int, payload: bytes, *, timeout: float) -> Any: + tag = self._alloc_tag() + pending = self._register_pending(tag, type_id) + try: + self._write_frame(type_id, tag, payload) + if not pending.event.wait(timeout=timeout): + raise TimeoutError_(f"command 0x{type_id:02X} tag={tag} timed out") + if pending.err is not None: + raise pending.err + return pending.ok_payload + finally: + self._forget_pending(tag) + + def _send_tx(self, data: bytes, *, skip_cad: bool, timeout: float) -> TxDone: + """TX has a two-step completion: OK (enqueued) then TX_DONE. + + We re-use the same pending slot for both: the reader flags ``err`` + (early ERR during enqueue) or ``tx_done`` (after TX_DONE arrives) + and signals the event. We skip the intermediate OK entirely — the + reader treats the OK as "keep waiting for TX_DONE" on this tag. + """ + tag = self._alloc_tag() + pending = self._register_pending(tag, 0x04) + try: + self._write_frame(0x04, tag, encode_tx_payload(data, skip_cad=skip_cad)) + if not pending.event.wait(timeout=timeout): + raise TimeoutError_(f"TX tag={tag} timed out waiting for TX_DONE") + if pending.err is not None: + raise pending.err + if pending.tx_done is None: + raise DongloraError("TX completed without TX_DONE (internal error)") + return pending.tx_done + finally: + self._forget_pending(tag) + + # ── Reader thread ──────────────────────────────────────────────── + + def _reader_loop(self) -> None: + try: + while not self._closed.is_set(): + try: + frame = read_frame(self._transport) + except FrameCodecError as exc: + # CRC/COBS/length failure on an inbound frame. Surface as + # an async error so the caller can see it. + log.debug("inbound frame error: %s", exc) + self._async_errors.put(FrameError(str(exc))) + continue + except OSError: + # Transport dropped (USB unplug, socket closed, etc.). + if not self._closed.is_set(): + log.debug("session reader: transport closed") + return + if frame is None: + # Transport timeout — loop and keep reading. + continue + try: + self._dispatch(frame) + except Exception as exc: + log.exception("reader dispatch failed: %s", exc) + finally: + # Reader is going away — wake anyone still waiting so they + # don't hang on .event.wait() forever, and signal the + # Dongle so it can trigger transparent reconnect. + with self._pending_lock: + for p in self._pending.values(): + if p.err is None and p.tx_done is None: + p.err = DongloraError("transport closed") + p.event.set() + self._pending.clear() + self._reader_dead.set() + # Poison the RX queue with a sentinel so any blocked + # next_rx() unblocks promptly instead of waiting for its + # full timeout. The sentinel is filtered out in next_rx. + self._rx_queue.put(_DEAD_SENTINEL) + + def _dispatch(self, frame: Frame) -> None: + type_id = frame.type_id + tag = frame.tag + payload = frame.payload + + # Async events: tag==0 or RX (which always has tag 0). + if tag == 0: + if type_id == TYPE_RX: + try: + rx = RxEvent.decode(payload) + except ValueError as exc: + log.debug("bad RX event: %s", exc) + self._async_errors.put(FrameError(f"bad RX payload: {exc}")) + return + self._rx_queue.put(rx) + return + if type_id == TYPE_ERR: + code = decode_err_payload(payload) + self._async_errors.put(device_error(code, tag=0)) + return + log.debug("unknown async frame type 0x%02X", type_id) + return + + # Tag-correlated: OK / ERR / TX_DONE / RX-with-tag (never expected). + with self._pending_lock: + pending = self._pending.get(tag) + if pending is None: + log.debug("no pending command for tag 0x%04X (type 0x%02X)", tag, type_id) + return + + if type_id == TYPE_OK: + try: + ok = parse_ok_payload(pending.cmd_type, payload) + except (ValueError, KeyError) as exc: + pending.err = DongloraError(f"bad OK payload: {exc}") + pending.event.set() + return + if pending.cmd_type == 0x04: + # TX: OK means "enqueued" — do NOT signal yet, wait for TX_DONE. + pending.ok_payload = ok + return + pending.ok_payload = ok + pending.event.set() + return + + if type_id == TYPE_ERR: + code = decode_err_payload(payload) + pending.err = device_error(code, tag=tag) + pending.event.set() + return + + if type_id == TYPE_TX_DONE: + try: + td = TxDone.decode(payload) + except ValueError as exc: + pending.err = DongloraError(f"bad TX_DONE payload: {exc}") + pending.event.set() + return + from donglora.errors import Cancelled, ChannelBusy + from donglora.events import TxResult + + if td.result == TxResult.CHANNEL_BUSY: + pending.err = ChannelBusy() + elif td.result == TxResult.CANCELLED: + pending.err = Cancelled() + else: + pending.tx_done = td + pending.event.set() + return + + log.debug("unexpected tag-correlated type 0x%02X for tag 0x%04X", type_id, tag) + + +_ = ErrorCode # re-export so users can still catch specific codes diff --git a/tests/test_client.py b/tests/test_client.py deleted file mode 100644 index 68b1528..0000000 --- a/tests/test_client.py +++ /dev/null @@ -1,135 +0,0 @@ -"""Tests for donglora.client — send/recv/drain_rx/validate.""" - -from __future__ import annotations - -import struct -from io import BytesIO -from typing import Any - -from donglora.client import _rx_queue, drain_rx, recv, send, validate -from donglora.codec import cobs_encode - - -def _make_mock(responses: list[bytes]) -> Any: - """Create a mock transport that returns pre-encoded COBS frames.""" - wire = b"" - for resp in responses: - wire += cobs_encode(resp) - - class Mock: - def __init__(self) -> None: - self._stream = BytesIO(wire) - self.timeout = 2.0 - self._written: list[bytes] = [] - - def read(self, n: int = 1) -> bytes: - return self._stream.read(n) - - def write(self, data: bytes) -> int: - self._written.append(data) - return len(data) - - def flush(self) -> None: - pass - - return Mock() - - -def _pong() -> bytes: - return bytes([0]) - - -def _ok() -> bytes: - return bytes([4]) - - -def _rx_packet(payload: bytes, rssi: int = -80, snr: int = 10) -> bytes: - return ( - bytes([2]) - + struct.pack(" None: - _rx_queue.clear() - - def test_basic_command(self) -> None: - mock = _make_mock([_pong()]) - resp = send(mock, "Ping") - assert resp["type"] == "Pong" - - def test_buffers_rx_packets(self) -> None: - mock = _make_mock([_rx_packet(b"pkt1"), _rx_packet(b"pkt2"), _ok()]) - resp = send(mock, "StartRx") - assert resp["type"] == "Ok" - assert len(_rx_queue) == 2 - - def test_timeout(self) -> None: - mock = _make_mock([]) - resp = send(mock, "Ping") - assert resp["type"] == "Timeout" - - -class TestRecv: - def setup_method(self) -> None: - _rx_queue.clear() - - def test_from_buffer(self) -> None: - _rx_queue.append({"type": "RxPacket", "rssi": -80, "snr": 10, "payload": b"buf"}) - mock = _make_mock([]) - pkt = recv(mock) - assert pkt is not None - assert pkt["payload"] == b"buf" - - def test_from_wire(self) -> None: - mock = _make_mock([_rx_packet(b"wire")]) - pkt = recv(mock) - assert pkt is not None - assert pkt["payload"] == b"wire" - - def test_timeout(self) -> None: - mock = _make_mock([]) - assert recv(mock) is None - - -class TestDrainRx: - def setup_method(self) -> None: - _rx_queue.clear() - - def test_drains_buffer_and_wire(self) -> None: - _rx_queue.append({"type": "RxPacket", "rssi": 0, "snr": 0, "payload": b"buf"}) - mock = _make_mock([_rx_packet(b"wire")]) - packets = drain_rx(mock) - assert len(packets) == 2 - assert packets[0]["payload"] == b"buf" - assert packets[1]["payload"] == b"wire" - assert len(_rx_queue) == 0 - - -class TestValidate: - def setup_method(self) -> None: - _rx_queue.clear() - - def test_success(self) -> None: - mock = _make_mock([_pong()]) - validate(mock) - - def test_failure(self) -> None: - mock = _make_mock([_ok()]) - try: - validate(mock) - assert False, "should have raised" # noqa: B011 - except ConnectionError: - pass - - def test_timeout(self) -> None: - mock = _make_mock([]) - try: - validate(mock) - assert False, "should have raised" # noqa: B011 - except ConnectionError: - pass diff --git a/tests/test_codec.py b/tests/test_codec.py index 5184ff6..743e456 100644 --- a/tests/test_codec.py +++ b/tests/test_codec.py @@ -1,12 +1,13 @@ -"""Tests for donglora.codec — COBS framing.""" +"""COBS helper tests — the thin re-export layer. + +Frame-level codec tests live in ``test_frame.py``. +""" from __future__ import annotations -from io import BytesIO - from cobs import cobs -from donglora.codec import cobs_encode, read_frame +from donglora.codec import cobs_encode class TestCobsEncode: @@ -34,42 +35,3 @@ class TestCobsEncode: encoded = cobs_encode(data) decoded = cobs.decode(encoded[:-1]) assert decoded == data - - -class _MockSerial: - """Minimal mock that behaves like serial.Serial for read_frame.""" - - def __init__(self, data: bytes) -> None: - self._stream = BytesIO(data) - - def read(self, n: int = 1) -> bytes: - return self._stream.read(n) - - -class TestReadFrame: - def test_basic(self) -> None: - data = b"hello" - frame = cobs_encode(data) - result = read_frame(_MockSerial(frame)) - assert result == data - - def test_timeout(self) -> None: - result = read_frame(_MockSerial(b"")) - assert result is None - - def test_empty_frame(self) -> None: - result = read_frame(_MockSerial(b"\x00")) - assert result is None - - def test_corrupt_frame(self) -> None: - result = read_frame(_MockSerial(b"\xff\x00")) - # cobs.decode of [0xff] may or may not raise — just verify we get bytes or None - assert result is None or isinstance(result, bytes) - - def test_multiple_frames(self) -> None: - frame1 = cobs_encode(b"first") - frame2 = cobs_encode(b"second") - mock = _MockSerial(frame1 + frame2) - assert read_frame(mock) == b"first" - assert read_frame(mock) == b"second" - assert read_frame(mock) is None diff --git a/tests/test_crc.py b/tests/test_crc.py new file mode 100644 index 0000000..ce8b860 --- /dev/null +++ b/tests/test_crc.py @@ -0,0 +1,23 @@ +"""CRC-16/CCITT-FALSE conformance tests.""" + +from donglora.crc import crc16 + + +def test_check_value_matches_spec(): + # PROTOCOL.md §2.2 / Appendix B. + assert crc16(b"123456789") == 0x29B1 + + +def test_empty_input_is_initial_value(): + # No reflection, no XOR-out: empty leaves init untouched. + assert crc16(b"") == 0xFFFF + + +def test_single_null_byte(): + # A naive init=0 impl would yield 0; correct init=0xFFFF yields 0xE1F0. + assert crc16(b"\x00") == 0xE1F0 + + +def test_differs_from_kermit_variant(): + # CRC-16/KERMIT of "123456789" is 0x2189. + assert crc16(b"123456789") != 0x2189 diff --git a/tests/test_dongle.py b/tests/test_dongle.py new file mode 100644 index 0000000..4e2fc4b --- /dev/null +++ b/tests/test_dongle.py @@ -0,0 +1,219 @@ +"""Dongle-level tests: keepalive thread, auto-recovery, context manager.""" + +from __future__ import annotations + +import time + +import pytest + +from donglora.commands import TYPE_PING, TYPE_TX +from donglora.dongle import Dongle +from donglora.errors import DongloraError +from donglora.events import ( + TYPE_ERR, + TxResult, +) +from donglora.frame import encode_frame +from donglora.modulation import LoRaConfig +from donglora.session import Session +from tests.test_session import FakeTransport, _default_script + + +def _make_dongle(transport: FakeTransport) -> Dongle: + session = Session(transport) + info = session.get_info() + cfg = LoRaConfig.default() + session.set_config(cfg) + return Dongle(session, info, applied_config=cfg, keepalive=False) + + +def test_dongle_tx_returns_tx_done(): + t = FakeTransport() + try: + d = _make_dongle(t) + with d: + td = d.tx(b"hello") + assert td.airtime_us == 30_976 + finally: + t.close() + + +def test_dongle_info_property_exposes_cached_info(): + t = FakeTransport() + try: + with _make_dongle(t) as d: + assert d.info.proto_major == 1 + finally: + t.close() + + +def test_dongle_auto_recovers_from_not_configured(): + """First TX returns ENOTCONFIGURED; Dongle re-applies cached config, + then retries and succeeds. + """ + + class FlipScript: + def __init__(self): + self._first_tx = True + + def __call__(self, type_id, tag, payload): + if type_id == TYPE_TX and self._first_tx: + self._first_tx = False + return [encode_frame(TYPE_ERR, tag, b"\x03\x00")] # ENOTCONFIGURED + return _default_script(type_id, tag, payload) + + t = FakeTransport(script=FlipScript()) + try: + with _make_dongle(t) as d: + td = d.tx(b"retry-me") + assert td.result == TxResult.TRANSMITTED + finally: + t.close() + + +def test_dongle_context_manager_closes_session(): + t = FakeTransport() + try: + d = _make_dongle(t) + with d: + pass + # After exit, further operations raise DongloraError. + with pytest.raises(DongloraError): + d.tx(b"too-late") + finally: + t.close() + + +def test_dongle_close_is_idempotent(): + t = FakeTransport() + try: + d = _make_dongle(t) + d.close() + d.close() # must not raise + finally: + t.close() + + +def test_dongle_keepalive_daemon_sends_pings(): + """With keepalive enabled, a quiet dongle emits PINGs at ~500ms cadence.""" + t = FakeTransport() + try: + session = Session(t) + info = session.get_info() + cfg = LoRaConfig.default() + session.set_config(cfg) + # Record frames sent after setup. keepalive=True starts the + # background thread. + t.log.clear() + d = Dongle(session, info, applied_config=cfg, keepalive=True) + time.sleep(0.8) # allow at least one keepalive interval + d.close() + # Expect at least one PING in the log. + from donglora.frame import decode_frame + + ping_count = sum(1 for raw in t.log if decode_frame(raw).type_id == TYPE_PING) + assert ping_count >= 1 + finally: + t.close() + + +def test_dongle_config_property_tracks_set_config(): + t = FakeTransport() + try: + with _make_dongle(t) as d: + assert d.config == LoRaConfig.default() + custom = LoRaConfig( + freq_hz=915_000_000, + sf=10, + bw=d.config.bw, + cr=d.config.cr, + ) + d.set_config(custom) + assert d.config == custom + finally: + t.close() + + +class _EOFTransport(FakeTransport): + """Like FakeTransport, but `read()` raises ConnectionError after close. + + Matches the real-world ``MuxConnection.read`` / ``serial.Serial.read`` + behaviour when the remote hangs up, which is what the Dongle's + transparent-reconnect path needs to see. + """ + + def read(self, n: int = 1) -> bytes: + data = super().read(n) + if not data and self._closed: + raise ConnectionError("transport closed") + return data + + +def test_dongle_transparent_reconnect_on_session_death(): + """When the session's reader dies (mux restart / transport blip), + the Dongle transparently reopens and a subsequent ``tx()`` call + just works. The caller never sees an exception. + + Drives it with a reopener that returns a *fresh* transport for the + second connect — simulating the mux coming back up under a new + process with the original radio state intact. + """ + # Transport #1 — will be "killed" mid-test to simulate mux death. + t1 = _EOFTransport() + # Transport #2 — stands in for the replacement mux coming back. + t2 = FakeTransport() + + session1 = Session(t1) + info1 = session1.get_info() + cfg = LoRaConfig.default() + session1.set_config(cfg) + + def reopener(): + s = Session(t2) + i = s.get_info() + s.set_config(cfg) + return (s, i, cfg) + + d = Dongle( + session1, + info1, + applied_config=cfg, + keepalive=False, + _reopener=reopener, + ) + try: + # Sanity: first TX works through session #1. + assert d.tx(b"hi").result == TxResult.TRANSMITTED + + # Simulate mux going away. Reader thread sees ConnectionError + # on its next read and exits. + t1.close() + # Give the reader a moment to notice. + time.sleep(0.2) + + # Second TX MUST succeed — Dongle reconnects through the + # reopener transparently. + assert d.tx(b"after-reconnect").result == TxResult.TRANSMITTED + # And the cached config is preserved across reconnect. + assert d.config == cfg + finally: + d.close() + t2.close() + + +def test_dongle_without_reopener_surfaces_dead_session(): + """Manually-constructed Dongles (no reopener) must not swallow + session death — the caller gets a real error.""" + t = _EOFTransport() + try: + session = Session(t) + info = session.get_info() + cfg = LoRaConfig.default() + session.set_config(cfg) + d = Dongle(session, info, applied_config=cfg, keepalive=False) # no reopener + t.close() + time.sleep(0.2) # let the reader die + with pytest.raises(DongloraError): + d.tx(b"no-reopener") + finally: + t.close() diff --git a/tests/test_ergonomics.py b/tests/test_ergonomics.py new file mode 100644 index 0000000..7f20105 --- /dev/null +++ b/tests/test_ergonomics.py @@ -0,0 +1,75 @@ +"""Sanity check that the three-line happy path works. + +If these tests need anything beyond ``import donglora as dl`` + three +method calls, the client lost its reason for existing. +""" + +from __future__ import annotations + +import donglora as dl +from donglora.events import TYPE_RX, RxOrigin + + +def test_happy_path_tx_is_three_lines(): + """Three lines: get a Dongle (from the test harness), tx, done.""" + from donglora.dongle import Dongle + from donglora.modulation import LoRaConfig + from donglora.session import Session + from tests.test_session import FakeTransport + + t = FakeTransport() + try: + session = Session(t) + info = session.get_info() + cfg = LoRaConfig.default() + session.set_config(cfg) + with Dongle(session, info, applied_config=cfg, keepalive=False) as d: + td = d.tx(b"Hello") + assert td.result.name == "TRANSMITTED" + finally: + t.close() + + +def test_rx_iterator_yields_incoming_packets(): + """The `.rx()` iterator terminates cleanly on timeout.""" + from donglora.dongle import Dongle + from donglora.events import RxEvent + from donglora.modulation import LoRaConfig + from donglora.session import Session + from tests.test_session import FakeTransport + + t = FakeTransport() + try: + session = Session(t) + info = session.get_info() + cfg = LoRaConfig.default() + session.set_config(cfg) + rx = RxEvent( + rssi_dbm=-72.0, + snr_db=8.0, + freq_err_hz=0, + timestamp_us=100, + crc_valid=True, + packets_dropped=0, + origin=RxOrigin.OTA, + data=b"ping", + ) + with Dongle(session, info, applied_config=cfg, keepalive=False) as d: + # Inject two RX events, then time out. + t.push_async(TYPE_RX, 0, rx.encode()) + t.push_async(TYPE_RX, 0, rx.encode()) + received: list[bytes] = [] + for pkt in d.rx(timeout=0.5): + received.append(pkt.data) + if len(received) == 2: + break + assert received == [b"ping", b"ping"] + finally: + t.close() + + +def test_public_api_names_are_stable(): + # The top-level module exposes the core happy-path surface. If any + # of these disappear it's a breaking change we should notice. + for name in ("connect", "Dongle", "LoRaConfig", "tx", "rx", "recv"): + assert hasattr(dl, name), f"donglora.{name} must stay exported" diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..1b806fa --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,265 @@ +"""Event payload tests: Info, SetConfigResult, TxDone, RxEvent, ERR.""" + +import pytest + +from donglora.commands import TYPE_PING, TYPE_SET_CONFIG +from donglora.errors import ErrorCode +from donglora.events import ( + Owner, + RxEvent, + RxOrigin, + SetConfigResult, + SetConfigResultCode, + TxDone, + TxResult, + decode_err_payload, + encode_err_payload, + parse_ok_payload, +) +from donglora.info import Capability, Info, RadioChipId +from donglora.modulation import LoRaConfig + +# ── SetConfigResult ─────────────────────────────────────────── + + +def test_set_config_result_c23_bytes(): + # PROTOCOL.md §C.2.3: OK in reply to SET_CONFIG, LoRa applied. + r = SetConfigResult( + result=SetConfigResultCode.APPLIED, + owner=Owner.MINE, + current=LoRaConfig.default(), + ) + expected = bytes( + [ + 0x00, + 0x01, # result=APPLIED, owner=MINE + 0x01, # modulation_id = LoRa + 0xA0, + 0x27, + 0xBE, + 0x33, # freq + 0x07, + 0x07, + 0x00, # sf / bw / cr + 0x08, + 0x00, # preamble_len + 0x24, + 0x14, # sync_word + 0x0E, # tx_power + 0x00, + 0x01, + 0x00, # header / crc / iq + ], + ) + assert r.encode() == expected + assert SetConfigResult.decode(expected) == r + + +# ── TxDone ──────────────────────────────────────────────────── + + +def test_tx_done_transmitted_airtime(): + td = TxDone(result=TxResult.TRANSMITTED, airtime_us=30_976) + assert td.encode() == bytes([0x00, 0x00, 0x79, 0x00, 0x00]) + assert TxDone.decode(td.encode()) == td + + +def test_tx_done_channel_busy(): + td = TxDone(result=TxResult.CHANNEL_BUSY, airtime_us=0) + assert TxDone.decode(td.encode()) == td + + +# ── RxEvent ─────────────────────────────────────────────────── + + +def test_rx_event_c26_bytes(): + rx = RxEvent( + rssi_dbm=-73.5, + snr_db=9.5, + freq_err_hz=-125, + timestamp_us=42_000_000, + crc_valid=True, + packets_dropped=0, + origin=RxOrigin.OTA, + data=bytes([0x01, 0x02, 0x03, 0x04]), + ) + # PROTOCOL.md §C.2.6 payload (strip frame header + CRC). + expected = bytes( + [ + 0x21, + 0xFD, # rssi = -735 LE + 0x5F, + 0x00, # snr = 95 LE + 0x83, + 0xFF, + 0xFF, + 0xFF, # freq_err = -125 LE + 0x80, + 0xDE, + 0x80, + 0x02, + 0x00, + 0x00, + 0x00, + 0x00, # ts = 42_000_000 LE + 0x01, # crc_valid + 0x00, + 0x00, # packets_dropped + 0x00, # origin OTA + 0x01, + 0x02, + 0x03, + 0x04, + ], + ) + assert rx.encode() == expected + decoded = RxEvent.decode(expected) + assert decoded.rssi_dbm == pytest.approx(-73.5) + assert decoded.snr_db == pytest.approx(9.5) + assert decoded.data == bytes([0x01, 0x02, 0x03, 0x04]) + + +def test_rx_event_rejects_invalid_crc_byte(): + buf = bytearray( + RxEvent( + rssi_dbm=0.0, + snr_db=0.0, + freq_err_hz=0, + timestamp_us=0, + crc_valid=True, + packets_dropped=0, + origin=RxOrigin.OTA, + data=b"", + ).encode(), + ) + buf[16] = 2 + with pytest.raises(ValueError): + RxEvent.decode(bytes(buf)) + + +# ── ERR payload ─────────────────────────────────────────────── + + +def test_err_payload_roundtrip(): + wire = encode_err_payload(ErrorCode.ENOT_CONFIGURED) + assert wire == bytes([0x03, 0x00]) + assert decode_err_payload(wire) == ErrorCode.ENOT_CONFIGURED + + +def test_err_payload_preserves_unknown_code(): + wire = encode_err_payload(0xABCD) + assert decode_err_payload(wire) == 0xABCD + + +# ── OK payload context dispatch ─────────────────────────────── + + +def test_parse_ok_for_ping_requires_empty(): + assert parse_ok_payload(TYPE_PING, b"") is None + with pytest.raises(ValueError): + parse_ok_payload(TYPE_PING, b"\x00") + + +def test_parse_ok_for_set_config_returns_result(): + r = SetConfigResult( + result=SetConfigResultCode.APPLIED, + owner=Owner.MINE, + current=LoRaConfig.default(), + ) + parsed = parse_ok_payload(TYPE_SET_CONFIG, r.encode()) + assert parsed == r + + +# ── Info ────────────────────────────────────────────────────── + + +def _sample_info() -> Info: + return Info( + proto_major=1, + proto_minor=0, + fw_major=0, + fw_minor=1, + fw_patch=0, + radio_chip_id=RadioChipId.SX1262, + capability_bitmap=Capability.LORA | Capability.FSK | Capability.CAD_BEFORE_TX, + supported_sf_bitmap=0x1FE0, + supported_bw_bitmap=0x03FF, + max_payload_bytes=255, + rx_queue_capacity=64, + tx_queue_capacity=16, + freq_min_hz=150_000_000, + freq_max_hz=960_000_000, + tx_power_min_dbm=-9, + tx_power_max_dbm=22, + mcu_uid=bytes([0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x23, 0x45, 0x67]), + radio_uid=b"", + ) + + +def test_info_appendix_c22_bytes(): + info = _sample_info() + expected = bytes( + [ + 0x01, + 0x00, + 0x00, + 0x01, + 0x00, + 0x02, + 0x00, # chip id + 0x03, + 0x00, + 0x01, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, # cap bitmap + 0xE0, + 0x1F, # supported SF + 0xFF, + 0x03, # supported BW + 0xFF, + 0x00, # max payload 255 + 0x40, + 0x00, # rx q 64 + 0x10, + 0x00, # tx q 16 + 0x80, + 0xD1, + 0xF0, + 0x08, # freq min + 0x00, + 0x70, + 0x38, + 0x39, # freq max + 0xF7, + 0x16, # tx power min/max + 0x08, # mcu_uid_len + 0xDE, + 0xAD, + 0xBE, + 0xEF, + 0x01, + 0x23, + 0x45, + 0x67, + 0x00, # radio_uid_len + ], + ) + assert info.encode() == expected + assert Info.decode(expected) == info + + +def test_info_tolerates_trailing_bytes(): + info = _sample_info() + wire = info.encode() + b"\xaa" * 10 + assert Info.decode(wire) == info + + +def test_info_supports(): + info = _sample_info() + assert info.supports(Capability.LORA) + assert info.supports(Capability.LORA | Capability.FSK) + assert not info.supports(Capability.MULTI_CLIENT) + assert info.chip == RadioChipId.SX1262 diff --git a/tests/test_frame.py b/tests/test_frame.py new file mode 100644 index 0000000..fa66f58 --- /dev/null +++ b/tests/test_frame.py @@ -0,0 +1,175 @@ +"""Frame codec tests — every Appendix C hex vector pinned. + +These match the Rust crate's ``tests/vectors.rs`` byte-for-byte. If the +client-py and firmware ever disagree, one of these tests fails first. +""" + +import pytest + +from donglora.commands import ( + TYPE_PING, + TYPE_TX, + encode_ping_payload, + encode_tx_payload, +) +from donglora.events import TYPE_ERR, TYPE_OK, TYPE_TX_DONE +from donglora.frame import ( + Frame, + FrameCobsError, + FrameCrcError, + FrameTooShortError, + decode_frame, + encode_frame, + read_frame, +) + +# ── §C.2.1 PING / OK ───────────────────────────────────────────── + + +def test_c21_ping_encodes_to_spec_bytes(): + wire = encode_frame(TYPE_PING, 0x0001, encode_ping_payload()) + assert wire == bytes([0x03, 0x01, 0x01, 0x03, 0x9D, 0xC8, 0x00]) + + +def test_c21_ok_encodes_to_spec_bytes(): + wire = encode_frame(TYPE_OK, 0x0001, b"") + assert wire == bytes([0x03, 0x80, 0x01, 0x03, 0xF7, 0xC4, 0x00]) + + +# ── §C.2.4 TX / OK / TX_DONE ───────────────────────────────────── + + +def test_c24_tx_with_cad(): + wire = encode_frame(TYPE_TX, 0x0004, encode_tx_payload(b"Hello", skip_cad=False)) + assert wire == bytes( + [0x03, 0x04, 0x04, 0x01, 0x08, 0x48, 0x65, 0x6C, 0x6C, 0x6F, 0x26, 0x40, 0x00], + ) + + +def test_c24_ok(): + wire = encode_frame(TYPE_OK, 0x0004, b"") + assert wire == bytes([0x03, 0x80, 0x04, 0x03, 0x02, 0x3B, 0x00]) + + +def test_c24_tx_done(): + # Payload = result(0=TRANSMITTED) + airtime_us(30976 LE). + payload = bytes([0x00, 0x00, 0x79, 0x00, 0x00]) + wire = encode_frame(TYPE_TX_DONE, 0x0004, payload) + assert wire == bytes( + [0x03, 0xC1, 0x04, 0x01, 0x01, 0x02, 0x79, 0x01, 0x03, 0xE3, 0xFA, 0x00], + ) + + +# ── §C.2.5 TX skip_cad ────────────────────────────────────────── + + +def test_c25_tx_skip_cad(): + wire = encode_frame(TYPE_TX, 0x0005, encode_tx_payload(b"URGENT", skip_cad=True)) + assert wire == bytes( + [0x03, 0x04, 0x05, 0x0A, 0x01, 0x55, 0x52, 0x47, 0x45, 0x4E, 0x54, 0xDB, 0x1C, 0x00], + ) + + +# ── §C.5.1 ENOTCONFIGURED ─────────────────────────────────────── + + +def test_c51_err_notconfigured(): + # Payload = 0x0003 LE. + wire = encode_frame(TYPE_ERR, 0x0028, bytes([0x03, 0x00])) + assert wire == bytes([0x03, 0x81, 0x28, 0x02, 0x03, 0x03, 0x53, 0x7E, 0x00]) + + +# ── §C.6.4 Async EFRAME (tag=0) ───────────────────────────────── + + +def test_c64_async_err_eframe(): + wire = encode_frame(TYPE_ERR, 0x0000, bytes([0x02, 0x01])) + assert wire == bytes([0x02, 0x81, 0x01, 0x05, 0x02, 0x01, 0xCE, 0xEF, 0x00]) + + +# ── §C.8.1 COBS edge: no zeros in body ───────────────────────── + + +def test_c81_no_zeros_in_body(): + # PING tag=0x0101 → tag bytes have no 0x00, body has no zeros. + wire = encode_frame(TYPE_PING, 0x0101, b"") + assert wire == bytes([0x06, 0x01, 0x01, 0x01, 0xBC, 0xD8, 0x00]) + + +# ── Decode-side round-trip ────────────────────────────────────── + + +def test_decode_roundtrip_empty_payload(): + wire = encode_frame(TYPE_PING, 0x0042, b"") + frame = decode_frame(wire[:-1]) # strip sentinel + assert frame == Frame(type_id=TYPE_PING, tag=0x0042, payload=b"") + + +def test_decode_roundtrip_with_payload(): + payload = b"hello-world" + wire = encode_frame(TYPE_TX, 0x0123, payload) + frame = decode_frame(wire[:-1]) + assert frame.type_id == TYPE_TX + assert frame.tag == 0x0123 + assert frame.payload == payload + + +def test_decode_rejects_crc_flip(): + wire = bytearray(encode_frame(TYPE_PING, 0x0001, b"")) + # Flip one byte of the CRC region. + wire[-3] ^= 0x01 + with pytest.raises((FrameCrcError, FrameCobsError)): + decode_frame(bytes(wire[:-1])) + + +def test_decode_rejects_too_short_frame(): + # COBS of a 4-byte body: [0x05, 0xAA, 0xBB, 0xCC, 0xDD]. Decodes to + # 4 bytes, under the 5-byte minimum (type + tag + crc). + wire = bytes([0x05, 0xAA, 0xBB, 0xCC, 0xDD]) + with pytest.raises(FrameTooShortError): + decode_frame(wire) + + +# ── read_frame with a BytesIO-like reader ────────────────────── + + +class _FakeReader: + def __init__(self, data: bytes): + self._data = bytearray(data) + + def read(self, n: int = 1) -> bytes: + if not self._data: + return b"" + chunk = bytes(self._data[:n]) + del self._data[:n] + return chunk + + +def test_read_frame_yields_decoded_frame(): + wire = encode_frame(TYPE_PING, 0x0001, b"") + r = _FakeReader(wire) + frame = read_frame(r) + assert frame == Frame(type_id=TYPE_PING, tag=0x0001, payload=b"") + + +def test_read_frame_returns_none_on_timeout(): + r = _FakeReader(b"") + assert read_frame(r) is None + + +def test_read_frame_handles_back_to_back_frames(): + a = encode_frame(TYPE_PING, 1, b"") + b = encode_frame(TYPE_PING, 2, b"") + r = _FakeReader(a + b) + assert read_frame(r).tag == 1 + assert read_frame(r).tag == 2 + + +# ── Oversized payloads rejected ──────────────────────────────── + + +def test_encode_rejects_oversized_payload(): + from donglora.frame import MAX_PAYLOAD_FIELD, FrameTooLargeError + + with pytest.raises(FrameTooLargeError): + encode_frame(0xC0, 0x0000, b"\x42" * (MAX_PAYLOAD_FIELD + 1)) diff --git a/tests/test_modulation.py b/tests/test_modulation.py new file mode 100644 index 0000000..929c374 --- /dev/null +++ b/tests/test_modulation.py @@ -0,0 +1,202 @@ +"""Per-modulation config encode/decode tests. + +The LoRa test pins the Appendix C.2.3 canonical EU868 byte sequence. +The others round-trip through encode → decode. +""" + +import pytest + +from donglora.modulation import ( + FlrcBitrate, + FlrcBt, + FlrcCodingRate, + FlrcConfig, + FlrcPreambleLen, + FskConfig, + LoRaBandwidth, + LoRaCodingRate, + LoRaConfig, + LoRaHeaderMode, + LrFhssBandwidth, + LrFhssCodingRate, + LrFhssConfig, + LrFhssGrid, + decode_modulation, + encode_modulation, +) + +# ── LoRa — spec bytes ─────────────────────────────────────────── + + +def test_lora_default_matches_appendix_c23(): + cfg = LoRaConfig.default() + expected = bytes( + [ + 0xA0, + 0x27, + 0xBE, + 0x33, # freq_hz = 868_100_000 + 0x07, # sf + 0x07, # bw = Khz125 + 0x00, # cr = 4/5 + 0x08, + 0x00, # preamble_len = 8 + 0x24, + 0x14, # sync_word = 0x1424 + 0x0E, # tx_power_dbm = 14 + 0x00, # header explicit + 0x01, # payload_crc on + 0x00, # iq normal + ], + ) + assert cfg.encode() == expected + + +def test_lora_roundtrip(): + cfg = LoRaConfig( + freq_hz=915_000_000, + sf=10, + bw=LoRaBandwidth.KHZ_250, + cr=LoRaCodingRate.CR_4_8, + preamble_len=32, + sync_word=0xABCD, + tx_power_dbm=22, + header_mode=LoRaHeaderMode.IMPLICIT, + payload_crc=False, + iq_invert=True, + ) + assert LoRaConfig.decode(cfg.encode()) == cfg + + +def test_lora_decode_rejects_wrong_length(): + with pytest.raises(ValueError): + LoRaConfig.decode(b"\x00" * 14) + + +def test_lora_decode_rejects_invalid_enum(): + buf = bytearray(LoRaConfig.default().encode()) + buf[5] = 99 # invalid bandwidth + with pytest.raises(ValueError): + LoRaConfig.decode(bytes(buf)) + + +# ── FSK ──────────────────────────────────────────────────────── + + +def test_fsk_roundtrip_with_sync_word(): + cfg = FskConfig( + freq_hz=868_000_000, + bitrate_bps=50_000, + freq_dev_hz=25_000, + rx_bw=0x1A, + preamble_len=32, + sync_word=b"\x12\x34\x56\x78", + ) + assert FskConfig.decode(cfg.encode()) == cfg + + +def test_fsk_rejects_oversized_sync_word(): + cfg = FskConfig( + freq_hz=0, + bitrate_bps=0, + freq_dev_hz=0, + rx_bw=0, + preamble_len=0, + sync_word=b"\x00" * 9, + ) + with pytest.raises(ValueError): + cfg.encode() + + +# ── LR-FHSS ──────────────────────────────────────────────────── + + +def test_lr_fhss_roundtrip(): + cfg = LrFhssConfig( + freq_hz=915_000_000, + bw=LrFhssBandwidth.KHZ_137, + cr=LrFhssCodingRate.CR_2_3, + grid=LrFhssGrid.KHZ_25, + hopping=True, + tx_power_dbm=14, + ) + assert LrFhssConfig.decode(cfg.encode()) == cfg + + +def test_lr_fhss_rejects_nonzero_reserved(): + cfg = LrFhssConfig( + freq_hz=0, + bw=LrFhssBandwidth.KHZ_39, + cr=LrFhssCodingRate.CR_1_3, + grid=LrFhssGrid.KHZ_25, + hopping=False, + tx_power_dbm=0, + ) + buf = bytearray(cfg.encode()) + buf[9] = 1 + with pytest.raises(ValueError): + LrFhssConfig.decode(bytes(buf)) + + +# ── FLRC ─────────────────────────────────────────────────────── + + +def test_flrc_roundtrip(): + cfg = FlrcConfig( + freq_hz=2_400_000_000, + bitrate=FlrcBitrate.KBPS_1300, + cr=FlrcCodingRate.CR_3_4, + bt=FlrcBt.BT_0_5, + preamble_len=FlrcPreambleLen.BITS_24, + sync_word=0x12345678, + tx_power_dbm=10, + ) + assert FlrcConfig.decode(cfg.encode()) == cfg + + +# ── Sum type (modulation_id prefix) ─────────────────────────── + + +def test_encode_modulation_prepends_lora_id(): + cfg = LoRaConfig.default() + wire = encode_modulation(cfg) + assert wire[0] == 0x01 + assert wire[1:] == cfg.encode() + + +def test_decode_modulation_rejects_unknown_id(): + with pytest.raises(ValueError): + decode_modulation(bytes([0x99]) + b"\x00" * 15) + + +def test_decode_modulation_dispatches_correctly(): + for cfg in ( + LoRaConfig.default(), + FskConfig( + freq_hz=868_000_000, + bitrate_bps=9_600, + freq_dev_hz=5_000, + rx_bw=0, + preamble_len=16, + sync_word=b"\x12\x34", + ), + LrFhssConfig( + freq_hz=915_000_000, + bw=LrFhssBandwidth.KHZ_137, + cr=LrFhssCodingRate.CR_2_3, + grid=LrFhssGrid.KHZ_25, + hopping=True, + tx_power_dbm=14, + ), + FlrcConfig( + freq_hz=2_400_000_000, + bitrate=FlrcBitrate.KBPS_1300, + cr=FlrcCodingRate.CR_3_4, + bt=FlrcBt.BT_0_5, + preamble_len=FlrcPreambleLen.BITS_24, + sync_word=0x12345678, + tx_power_dbm=10, + ), + ): + wire = encode_modulation(cfg) + assert decode_modulation(wire) == cfg diff --git a/tests/test_protocol.py b/tests/test_protocol.py deleted file mode 100644 index 67fcab4..0000000 --- a/tests/test_protocol.py +++ /dev/null @@ -1,254 +0,0 @@ -"""Tests for donglora.protocol — ported from client-rs/src/protocol.rs tests.""" - -from __future__ import annotations - -import struct - -import pytest - -from donglora.protocol import ( - DEFAULT_CONFIG, - MAX_PAYLOAD, - PREAMBLE_DEFAULT, - RADIO_CONFIG_SIZE, - TX_POWER_MAX, - Bandwidth, - ErrorCode, - RadioConfig, - decode_config, - decode_response, - encode_command, - encode_config, -) - - -def _make_config() -> RadioConfig: - return { - "freq_hz": 915_000_000, - "bw": Bandwidth.Khz125, - "sf": 7, - "cr": 5, - "sync_word": 0x3444, - "tx_power_dbm": 22, - "preamble_len": 16, - "cad": 1, - } - - -# ── RadioConfig ────────────────────────────────────────────────── - - -class TestRadioConfig: - def test_roundtrip(self) -> None: - cfg = _make_config() - encoded = encode_config(cfg) - assert len(encoded) == RADIO_CONFIG_SIZE - decoded = decode_config(encoded) - assert decoded == cfg - - def test_default_roundtrip(self) -> None: - encoded = encode_config(DEFAULT_CONFIG) - decoded = decode_config(encoded) - assert decoded is not None - assert decoded["freq_hz"] == DEFAULT_CONFIG["freq_hz"] - assert decoded["bw"] == DEFAULT_CONFIG["bw"] - assert decoded["sf"] == DEFAULT_CONFIG["sf"] - assert decoded["cr"] == DEFAULT_CONFIG["cr"] - assert decoded["sync_word"] == DEFAULT_CONFIG["sync_word"] - assert decoded["tx_power_dbm"] == DEFAULT_CONFIG["tx_power_dbm"] - assert decoded["cad"] == DEFAULT_CONFIG["cad"] - - def test_all_bandwidths(self) -> None: - for bw in Bandwidth: - cfg = {**_make_config(), "bw": bw} - encoded = encode_config(cfg) - decoded = decode_config(encoded) - assert decoded is not None - assert decoded["bw"] == bw - - def test_invalid_bandwidth(self) -> None: - buf = bytearray(encode_config(_make_config())) - buf[4] = 255 - assert decode_config(bytes(buf)) is None - - def test_negative_power(self) -> None: - cfg = {**_make_config(), "tx_power_dbm": TX_POWER_MAX} - encoded = encode_config(cfg) - decoded = decode_config(encoded) - assert decoded is not None - assert decoded["tx_power_dbm"] == TX_POWER_MAX - - def test_short_buffer(self) -> None: - assert decode_config(bytes(12)) is None - assert decode_config(b"") is None - - -# ── Command encoding ───────────────────────────────────────────── - - -class TestCommandEncoding: - def test_simple_tags(self) -> None: - assert encode_command("Ping") == bytes([0]) - assert encode_command("GetConfig") == bytes([1]) - assert encode_command("StartRx") == bytes([3]) - assert encode_command("StopRx") == bytes([4]) - assert encode_command("DisplayOn") == bytes([6]) - assert encode_command("DisplayOff") == bytes([7]) - assert encode_command("GetMac") == bytes([8]) - - def test_set_config(self) -> None: - cfg = _make_config() - cmd = encode_command("SetConfig", config=cfg) - assert cmd[0] == 2 - assert len(cmd) == 1 + RADIO_CONFIG_SIZE - decoded = decode_config(cmd[1:]) - assert decoded == cfg - - def test_transmit_no_config(self) -> None: - cmd = encode_command("Transmit", payload=b"hello") - assert cmd[0] == 5 - assert cmd[1] == 0 # has_config = false - length = struct.unpack_from(" None: - cfg = _make_config() - cmd = encode_command("Transmit", config=cfg, payload=b"test") - assert cmd[0] == 5 - assert cmd[1] == 1 # has_config = true - decoded = decode_config(cmd[2:]) - assert decoded == cfg - offset = 2 + RADIO_CONFIG_SIZE - length = struct.unpack_from(" None: - cmd = encode_command("Transmit", payload=b"") - assert cmd[0] == 5 - length = struct.unpack_from(" None: - with pytest.raises(ValueError, match="payload too large"): - encode_command("Transmit", payload=bytes(MAX_PAYLOAD + 1)) - - def test_transmit_max_payload(self) -> None: - cmd = encode_command("Transmit", payload=bytes(MAX_PAYLOAD)) - length = struct.unpack_from(" None: - with pytest.raises(ValueError, match="unknown command"): - encode_command("FakeCommand") - - -# ── Response decoding ──────────────────────────────────────────── - - -class TestResponseDecoding: - def test_pong(self) -> None: - assert decode_response(bytes([0])) == {"type": "Pong"} - - def test_config(self) -> None: - cfg = _make_config() - resp = decode_response(bytes([1]) + encode_config(cfg)) - assert resp["type"] == "Config" - assert resp["freq_hz"] == cfg["freq_hz"] - assert resp["bw"] == cfg["bw"] - assert resp["sf"] == cfg["sf"] - assert resp["cr"] == cfg["cr"] - assert resp["sync_word"] == cfg["sync_word"] - assert resp["tx_power_dbm"] == cfg["tx_power_dbm"] - assert resp["preamble_len"] == cfg["preamble_len"] - assert resp["cad"] == cfg["cad"] - - def test_rx_packet(self) -> None: - payload = b"data" - data = bytes([2]) - data += struct.pack(" None: - data = bytes([2]) - data += struct.pack(" None: - assert decode_response(bytes([3])) == {"type": "TxDone"} - - def test_ok(self) -> None: - assert decode_response(bytes([4])) == {"type": "Ok"} - - def test_error_codes(self) -> None: - for code in ErrorCode: - resp = decode_response(bytes([5, code])) - assert resp["type"] == "Error" - assert resp["code"] == code - - def test_mac_address(self) -> None: - mac_bytes = bytes([0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]) - resp = decode_response(bytes([6]) + mac_bytes) - assert resp["type"] == "MacAddress" - assert resp["mac"] == "AA:BB:CC:DD:EE:FF" - - def test_empty(self) -> None: - assert decode_response(b"") == {"type": "Empty"} - - def test_unknown_tag(self) -> None: - resp = decode_response(bytes([255])) - assert resp["type"] == "Unknown(255)" - - -# ── Cross-compatibility with firmware encoding ──────────────────── - - -class TestFirmwareCompatibility: - def test_worked_example(self) -> None: - """From PROTOCOL.md: 915 MHz, 125 kHz BW, SF7, CR 4/5, sync 0x1424, max power.""" - cfg: RadioConfig = { - "freq_hz": 915_000_000, - "bw": Bandwidth.Khz125, - "sf": 7, - "cr": 5, - "sync_word": 0x1424, - "tx_power_dbm": TX_POWER_MAX, - "preamble_len": PREAMBLE_DEFAULT, - "cad": 1, - } - cmd = encode_command("SetConfig", config=cfg) - expected = bytes( - [0x02, 0xC0, 0xCA, 0x89, 0x36, 0x07, 0x07, 0x05, 0x24, 0x14, 0x80, 0x00, 0x00, 0x01] - ) - assert cmd == expected - - -# ── Enum coverage ──────────────────────────────────────────────── - - -class TestEnums: - def test_bandwidth_values(self) -> None: - assert Bandwidth.Khz7 == 0 - assert Bandwidth.Khz500 == 9 - assert len(Bandwidth) == 10 - - def test_error_code_values(self) -> None: - assert ErrorCode.InvalidConfig == 0 - assert ErrorCode.RadioBusy == 1 - assert ErrorCode.TxTimeout == 2 - assert ErrorCode.CrcError == 3 - assert ErrorCode.NotConfigured == 4 - assert ErrorCode.NoDisplay == 5 - assert len(ErrorCode) == 6 diff --git a/tests/test_session.py b/tests/test_session.py new file mode 100644 index 0000000..bc5553b --- /dev/null +++ b/tests/test_session.py @@ -0,0 +1,337 @@ +"""Session-level tests: tag correlation, async events, auto-recovery. + +Uses a `FakeTransport` — a thread-safe in-memory duplex pipe with a +simple script of pre-canned firmware responses. Every test exercises +the real frame codec; only the transport is faked. +""" + +from __future__ import annotations + +import threading +from collections.abc import Callable + +import pytest + +from donglora.commands import ( + TYPE_GET_INFO, + TYPE_PING, + TYPE_RX_START, + TYPE_SET_CONFIG, + TYPE_TX, +) +from donglora.errors import ( + BusyError, + Cancelled, + ChannelBusy, + NotConfiguredError, +) +from donglora.events import ( + TYPE_ERR, + TYPE_OK, + TYPE_RX, + TYPE_TX_DONE, + Owner, + RxEvent, + RxOrigin, + SetConfigResult, + SetConfigResultCode, + TxDone, + TxResult, +) +from donglora.frame import decode_frame, encode_frame +from donglora.info import Info, RadioChipId +from donglora.modulation import LoRaConfig +from donglora.session import Session + +# ── A tiny in-memory transport that simulates a device ───────── + + +class FakeTransport: + """Thread-safe in-memory duplex pipe with a scripted device.""" + + def __init__(self, *, script: Callable[[int, int, bytes], list[bytes]] | None = None): + self._read_buf = bytearray() + self._read_cond = threading.Condition() + self._write_buf = bytearray() + self._write_cond = threading.Condition() + self._closed = False + self._script = script or _default_script + self._writer_thread = threading.Thread( + target=self._writer_loop, + name="fake-transport-writer", + daemon=True, + ) + self._writer_thread.start() + self.log: list[bytes] = [] + + def write(self, data: bytes) -> int: + with self._write_cond: + self._write_buf.extend(data) + self._write_cond.notify_all() + return len(data) + + def flush(self) -> None: + pass + + def read(self, n: int = 1) -> bytes: + with self._read_cond: + while not self._read_buf and not self._closed: + self._read_cond.wait(timeout=0.1) + if self._closed: + return b"" + chunk = bytes(self._read_buf[:n]) + del self._read_buf[:n] + return chunk + + def close(self) -> None: + with self._read_cond: + self._closed = True + self._read_cond.notify_all() + with self._write_cond: + self._write_cond.notify_all() + + def _writer_loop(self) -> None: + """Read framed commands off _write_buf, feed to the script, + push responses onto _read_buf. + """ + while not self._closed: + with self._write_cond: + while not self._write_buf and not self._closed: + self._write_cond.wait(timeout=0.1) + if self._closed: + return + # Collect complete frames (ending in 0x00) from write_buf. + frames_raw: list[bytes] = [] + while True: + try: + idx = self._write_buf.index(0) + except ValueError: + break + frames_raw.append(bytes(self._write_buf[:idx])) + del self._write_buf[: idx + 1] + for raw in frames_raw: + try: + frame = decode_frame(raw) + except Exception: + continue + self.log.append(raw) + responses = self._script(frame.type_id, frame.tag, frame.payload) + if responses: + with self._read_cond: + for r in responses: + self._read_buf.extend(r) + self._read_cond.notify_all() + + # Kick an async event (e.g. RX) into the read stream. + def push_async(self, type_id: int, tag: int, payload: bytes) -> None: + wire = encode_frame(type_id, tag, payload) + with self._read_cond: + self._read_buf.extend(wire) + self._read_cond.notify_all() + + +def _default_script(type_id: int, tag: int, payload: bytes) -> list[bytes]: # noqa: ARG001 + """Reasonable default: ACK every command with a shaped OK.""" + if type_id == TYPE_PING: + return [encode_frame(TYPE_OK, tag, b"")] + if type_id == TYPE_GET_INFO: + info = _make_info() + return [encode_frame(TYPE_OK, tag, info.encode())] + if type_id == TYPE_SET_CONFIG: + result = SetConfigResult( + result=SetConfigResultCode.APPLIED, + owner=Owner.MINE, + current=LoRaConfig.default(), + ) + return [encode_frame(TYPE_OK, tag, result.encode())] + if type_id == TYPE_RX_START: + return [encode_frame(TYPE_OK, tag, b"")] + if type_id == TYPE_TX: + td = TxDone(result=TxResult.TRANSMITTED, airtime_us=30_976) + return [ + encode_frame(TYPE_OK, tag, b""), + encode_frame(TYPE_TX_DONE, tag, td.encode()), + ] + return [encode_frame(TYPE_OK, tag, b"")] + + +def _make_info() -> Info: + return Info( + proto_major=1, + proto_minor=0, + fw_major=0, + fw_minor=1, + fw_patch=0, + radio_chip_id=RadioChipId.SX1262, + capability_bitmap=1 | (1 << 16), # LoRa + CAD_BEFORE_TX + supported_sf_bitmap=0x1FE0, + supported_bw_bitmap=0x03FF, + max_payload_bytes=255, + rx_queue_capacity=32, + tx_queue_capacity=1, + freq_min_hz=150_000_000, + freq_max_hz=960_000_000, + tx_power_min_dbm=-9, + tx_power_max_dbm=22, + mcu_uid=b"\xde\xad\xbe\xef\x01\x23", + radio_uid=b"", + ) + + +# ── Session tests ────────────────────────────────────────────── + + +@pytest.fixture +def transport(): + t = FakeTransport() + yield t + t.close() + + +def test_session_ping(transport): + s = Session(transport) + try: + s.ping(timeout=2.0) + finally: + s.close() + + +def test_session_get_info_returns_info_struct(transport): + s = Session(transport) + try: + info = s.get_info(timeout=2.0) + assert isinstance(info, Info) + assert info.proto_major == 1 + finally: + s.close() + + +def test_session_set_config_returns_set_config_result(transport): + s = Session(transport) + try: + result = s.set_config(LoRaConfig.default()) + assert isinstance(result, SetConfigResult) + assert result.result == SetConfigResultCode.APPLIED + finally: + s.close() + + +def test_session_transmit_returns_tx_done(transport): + s = Session(transport) + try: + td = s.transmit(b"Hello") + assert td.result == TxResult.TRANSMITTED + assert td.airtime_us == 30_976 + finally: + s.close() + + +def test_session_transmit_channel_busy(transport): + def script(type_id: int, tag: int, payload: bytes) -> list[bytes]: + if type_id == TYPE_TX: + td = TxDone(result=TxResult.CHANNEL_BUSY, airtime_us=0) + return [ + encode_frame(TYPE_OK, tag, b""), + encode_frame(TYPE_TX_DONE, tag, td.encode()), + ] + return _default_script(type_id, tag, payload) + + transport._script = script + s = Session(transport) + try: + with pytest.raises(ChannelBusy): + s.transmit(b"x") + finally: + s.close() + + +def test_session_transmit_cancelled(transport): + def script(type_id: int, tag: int, payload: bytes) -> list[bytes]: + if type_id == TYPE_TX: + td = TxDone(result=TxResult.CANCELLED, airtime_us=0) + return [ + encode_frame(TYPE_OK, tag, b""), + encode_frame(TYPE_TX_DONE, tag, td.encode()), + ] + return _default_script(type_id, tag, payload) + + transport._script = script + s = Session(transport) + try: + with pytest.raises(Cancelled): + s.transmit(b"x") + finally: + s.close() + + +def test_session_err_not_configured_raises(transport): + def script(type_id: int, tag: int, payload: bytes) -> list[bytes]: + if type_id == TYPE_TX: + return [encode_frame(TYPE_ERR, tag, b"\x03\x00")] + return _default_script(type_id, tag, payload) + + transport._script = script + s = Session(transport) + try: + with pytest.raises(NotConfiguredError): + s.transmit(b"x") + finally: + s.close() + + +def test_session_err_ebusy_raises(transport): + def script(type_id: int, tag: int, payload: bytes) -> list[bytes]: + if type_id == TYPE_TX: + return [encode_frame(TYPE_ERR, tag, b"\x06\x00")] + return _default_script(type_id, tag, payload) + + transport._script = script + s = Session(transport) + try: + with pytest.raises(BusyError): + s.transmit(b"x") + finally: + s.close() + + +def test_session_async_rx_event_arrives_on_queue(transport): + s = Session(transport) + try: + rx = RxEvent( + rssi_dbm=-73.5, + snr_db=9.5, + freq_err_hz=0, + timestamp_us=100, + crc_valid=True, + packets_dropped=0, + origin=RxOrigin.OTA, + data=b"\x01\x02\x03", + ) + transport.push_async(TYPE_RX, 0, rx.encode()) + delivered = s.next_rx(timeout=2.0) + assert delivered is not None + assert delivered.data == b"\x01\x02\x03" + finally: + s.close() + + +def test_session_tag_correlation_across_concurrent_sends(transport): + # Two threads each send a PING. Both must resolve against their own tag. + s = Session(transport) + try: + errors: list[Exception] = [] + + def do_ping() -> None: + try: + s.ping(timeout=2.0) + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=do_ping) for _ in range(10)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=3.0) + assert errors == [] + finally: + s.close()