feat: example scripts (simple_rx, simple_tx, ping_pong) with shared donglora.py helper

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,
+}

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.")

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.")

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}")