From 17721cc6605af86b55eb80254abcb39c3c7c4fab Mon Sep 17 00:00:00 2001 From: Stephen Waits Date: Sun, 29 Mar 2026 10:13:37 -0600 Subject: [PATCH] feat: example scripts (simple_rx, simple_tx, ping_pong) with shared donglora.py helper --- donglora.py | 203 +++++++++++++++++++++++++++++++++++++++++++++++++++ ping_pong.py | 57 +++++++++++++++ simple_rx.py | 34 +++++++++ simple_tx.py | 30 ++++++++ 4 files changed, 324 insertions(+) create mode 100644 donglora.py create mode 100755 ping_pong.py create mode 100755 simple_rx.py create mode 100755 simple_tx.py diff --git a/donglora.py b/donglora.py new file mode 100644 index 0000000..4a279e6 --- /dev/null +++ b/donglora.py @@ -0,0 +1,203 @@ +"""DongLoRa host library — connect, configure, send/receive LoRa packets. + +This is a minimal helper for the example scripts. It implements the +DongLoRa USB protocol (COBS-framed postcard) just enough to be useful. +See docs/PROTOCOL.md for the full specification. +""" +# /// script +# requires-python = ">=3.10" +# dependencies = ["cobs", "pyserial"] +# /// + +import glob +import struct +import subprocess +import serial +import time +from cobs import cobs + +# ── USB device discovery ────────────────────────────────────────── + +USB_VID_PID = "1209:5741" + + +def find_port() -> str | None: + """Find the DongLoRa serial port by USB VID:PID.""" + for path in sorted(glob.glob("/dev/ttyACM*")) + sorted(glob.glob("/dev/ttyUSB*")): + try: + result = subprocess.run( + ["udevadm", "info", "--query=property", f"--name={path}"], + capture_output=True, text=True, timeout=2, + ) + props = dict( + line.split("=", 1) for line in result.stdout.splitlines() if "=" in line + ) + vid = props.get("ID_VENDOR_ID", "").lower() + pid = props.get("ID_MODEL_ID", "").lower() + if f"{vid}:{pid}" == USB_VID_PID: + return path + except Exception: + continue + ports = sorted(glob.glob("/dev/ttyACM*")) + return ports[0] if ports else None + + +def wait_for_device() -> str: + """Block until a DongLoRa device appears.""" + print("Waiting for DongLoRa...", end="", flush=True) + while True: + port = find_port() + if port: + print(f" found {port}") + time.sleep(0.3) + return port + print(".", end="", flush=True) + time.sleep(0.5) + + +# ── COBS framing ───────────────────────────────────────────────── + +def cobs_encode(data: bytes) -> bytes: + return cobs.encode(data) + b"\x00" + + +def read_frame(ser: serial.Serial) -> bytes | None: + """Read one COBS frame. Returns decoded bytes or None on timeout.""" + buf = b"" + while True: + b = ser.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 + + +# ── Postcard encoding ──────────────────────────────────────────── + +def varint(n: int) -> bytes: + out = [] + while n >= 0x80: + out.append((n & 0x7F) | 0x80) + n >>= 7 + out.append(n & 0x7F) + return bytes(out) + + +def zigzag(n: int) -> bytes: + return varint((n << 1) ^ (n >> 31) if n >= 0 else ((-n - 1) << 1) | 1) + + +def decode_varint(data: bytes) -> tuple[int, bytes]: + n, shift = 0, 0 + for i, b in enumerate(data): + n |= (b & 0x7F) << shift + shift += 7 + if not (b & 0x80): + return n, data[i + 1:] + return n, b"" + + +def decode_zigzag(data: bytes) -> tuple[int, bytes]: + n, rest = decode_varint(data) + return (n >> 1) ^ -(n & 1), rest + + +# ── Command encoding ───────────────────────────────────────────── + +def encode_config(cfg: dict) -> bytes: + """Encode a RadioConfig dict to postcard bytes.""" + out = varint(cfg["freq_hz"]) + out += varint(cfg["bw"]) + out += varint(cfg["sf"]) + out += varint(cfg["cr"]) + out += varint(cfg["sync_word"]) + out += zigzag(cfg["tx_power_dbm"]) + return out + + +def encode_command(cmd: str, **kwargs) -> bytes: + """Encode a command name + kwargs to postcard bytes.""" + commands = { + "Ping": 0, "GetConfig": 1, "SetConfig": 2, "StartRx": 3, + "StopRx": 4, "Transmit": 5, "DisplayOn": 6, "DisplayOff": 7, + } + idx = commands[cmd] + out = varint(idx) + if cmd == "SetConfig": + out += encode_config(kwargs["config"]) + elif cmd == "Transmit": + config = kwargs.get("config") + if config is None: + out += b"\x00" + else: + out += b"\x01" + encode_config(config) + payload = kwargs["payload"] + out += varint(len(payload)) + payload + return out + + +# ── Response decoding ──────────────────────────────────────────── + +def decode_response(data: bytes) -> dict: + """Decode a postcard-encoded response into a dict.""" + if not data: + return {"type": "Empty"} + variant = data[0] + rest = data[1:] + if variant == 0: + return {"type": "Pong"} + elif variant == 1: + return {"type": "Config", "raw": rest.hex()} + elif variant == 2: + rssi, rest = decode_zigzag(rest) + snr, rest = decode_zigzag(rest) + plen, rest = decode_varint(rest) + return {"type": "RxPacket", "rssi": rssi, "snr": snr, "payload": rest[:plen]} + elif variant == 3: + return {"type": "TxDone"} + elif variant == 4: + return {"type": "Ok"} + elif variant == 5: + code = rest[0] if rest else -1 + return {"type": "Error", "code": code} + else: + return {"type": f"Unknown({variant})"} + + +# ── High-level helpers ─────────────────────────────────────────── + +def send(ser: serial.Serial, cmd: str, **kwargs) -> dict: + """Send a command and return the response.""" + ser.write(cobs_encode(encode_command(cmd, **kwargs))) + ser.flush() + data = read_frame(ser) + if data is None: + return {"type": "Timeout"} + return decode_response(data) + + +def connect(port: str | None = None, timeout: float = 2.0) -> serial.Serial: + """Open a connection to a DongLoRa device.""" + if port is None: + port = wait_for_device() + ser = serial.Serial(port, timeout=timeout) + ser.reset_input_buffer() + return ser + + +# Default radio config: 915 MHz, 125 kHz BW, SF7, CR 4/5 +DEFAULT_CONFIG = { + "freq_hz": 915_000_000, + "bw": 7, # 125 kHz + "sf": 7, + "cr": 5, # CR 4/5 + "sync_word": 0x1424, + "tx_power_dbm": 14, +} diff --git a/ping_pong.py b/ping_pong.py new file mode 100755 index 0000000..18fde7f --- /dev/null +++ b/ping_pong.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""Two-dongle demo: one transmits, one receives. + +Usage: + uv run examples/ping_pong.py --role tx [PORT] + uv run examples/ping_pong.py --role rx [PORT] +""" +# /// script +# requires-python = ">=3.10" +# dependencies = ["cobs", "pyserial"] +# /// + +import argparse +import sys +import time + +sys.path.insert(0, __import__("pathlib").Path(__file__).parent.as_posix()) +import donglora as dl # noqa: E402 + +parser = argparse.ArgumentParser(description="DongLoRa ping-pong demo") +parser.add_argument("--role", choices=["tx", "rx"], required=True) +parser.add_argument("port", nargs="?", help="Serial port (auto-detect if omitted)") +args = parser.parse_args() + +ser = dl.connect(args.port) +print(dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG)) + +if args.role == "tx": + print("Transmitting every 2 seconds (Ctrl+C to stop)...\n") + seq = 0 + try: + while True: + msg = f"ping #{seq}" + resp = dl.send(ser, "Transmit", payload=msg.encode()) + print(f" TX: {msg!r} → {resp['type']}") + seq += 1 + time.sleep(2) + except KeyboardInterrupt: + print("\nDone.") + +else: + print(dl.send(ser, "StartRx")) + print("Receiving (Ctrl+C to stop)...\n") + ser.timeout = 1 + try: + while True: + data = dl.read_frame(ser) + if data is None: + continue + resp = dl.decode_response(data) + if resp["type"] == "RxPacket": + p = resp["payload"] + text = p.decode("utf-8", errors="replace") + print(f" RX: {text!r} RSSI:{resp['rssi']}dBm SNR:{resp['snr']}dB") + except KeyboardInterrupt: + dl.send(ser, "StopRx") + print("\nDone.") diff --git a/simple_rx.py b/simple_rx.py new file mode 100755 index 0000000..91c50d9 --- /dev/null +++ b/simple_rx.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +"""Receive LoRa packets and print them. + +Usage: + uv run examples/simple_rx.py [PORT] +""" +# /// script +# requires-python = ">=3.10" +# dependencies = ["cobs", "pyserial"] +# /// + +import sys +sys.path.insert(0, __import__("pathlib").Path(__file__).parent.as_posix()) +import donglora as dl # noqa: E402 + +ser = dl.connect(sys.argv[1] if len(sys.argv) > 1 else None) +print(dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG)) +print(dl.send(ser, "StartRx")) + +print("\nListening (Ctrl+C to stop)...\n") +ser.timeout = 1 + +try: + while True: + data = dl.read_frame(ser) + if data is None: + continue + resp = dl.decode_response(data) + if resp["type"] == "RxPacket": + p = resp["payload"] + print(f" RSSI:{resp['rssi']:4d}dBm SNR:{resp['snr']:3d}dB len:{len(p):3d} {p.hex()}") +except KeyboardInterrupt: + dl.send(ser, "StopRx") + print("\nDone.") diff --git a/simple_tx.py b/simple_tx.py new file mode 100755 index 0000000..9d2cbb2 --- /dev/null +++ b/simple_tx.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 +"""Transmit a single LoRa packet. + +Usage: + uv run examples/simple_tx.py [PORT] [MESSAGE] +""" +# /// script +# requires-python = ">=3.10" +# dependencies = ["cobs", "pyserial"] +# /// + +import sys +sys.path.insert(0, __import__("pathlib").Path(__file__).parent.as_posix()) +import donglora as dl # noqa: E402 + +port = None +message = "Hello from DongLoRa!" + +args = sys.argv[1:] +if args and not args[0].startswith("/dev"): + message = " ".join(args) +elif args: + port = args[0] + if len(args) > 1: + message = " ".join(args[1:]) + +ser = dl.connect(port) +print(dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG)) +print(dl.send(ser, "Transmit", payload=message.encode())) +print(f"Sent: {message!r}")