swaits/donglora
Archived
1
0 Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

fix: CR encoding bug, display overflow, auto-reconnect script

Browse Source 
- Fix CodingRate encoding: postcard uses 0-based variant index, not
  repr(u8) value. CR 4/5 is variant 0, not 1.
- Shorten RSSI/SNR display line to fit 21-char width
- Python script auto-detects DongLoRa by USB VID:PID (303a:1001),
  reconnects on device reset/disconnect, no more crash on RST

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
swaitsandClaude Opus 4.6 committed 2026-03-29 20:42:56 -06:00
1 parent 102ce417ff
commit e8744e17da
2 files changed
+157 -109
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/display/render.rs
+3 -3
View File
@@ -120,14 +120,14 @@ pub fn dashboard(
.draw(target)
.ok();
// Row 3: RSSI + SNR
// Row 3: RSSI + SNR (compact to fit 21 chars)
buf.clear();
match (status.last_rssi, status.last_snr) {
(Some(rssi), Some(snr)) => {
let _ = write!(buf, "RSSI:{} dBm SNR:{} dB", rssi, snr);
let _ = write!(buf, "{}dBm SNR:{}dB", rssi, snr);
}
(Some(rssi), None) => {
let _ = write!(buf, "RSSI:{} dBm", rssi);
let _ = write!(buf, "{}dBm", rssi);
}
_ => {
let _ = write!(buf, "No signal");
tools/test_ping.py
+154 -106
View File
@@ -5,12 +5,66 @@
# dependencies = ["cobs", "pyserial"]
# ///
import glob
import struct
import serial
import sys
import time
from cobs import cobs
# Espressif USB JTAG/serial debug unit
USB_VID_PID = "303a:1001"
def find_serial_port() -> str | None:
"""Find the serial port for our USB device."""
import subprocess
# Try to find 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,
)
vid = ""
pid = ""
for line in result.stdout.splitlines():
if line.startswith("ID_VENDOR_ID="):
vid = line.split("=", 1)[1].lower()
elif line.startswith("ID_MODEL_ID="):
pid = line.split("=", 1)[1].lower()
if f"{vid}:{pid}" == USB_VID_PID:
return path
except Exception:
continue
# Fallback: first ttyACM device
ports = sorted(glob.glob("/dev/ttyACM*"))
return ports[0] if ports else None
def wait_for_device() -> str:
"""Poll until the USB device appears."""
print("Waiting for DongLoRa...", end="", flush=True)
while True:
port = find_serial_port()
if port:
print(f" found {port}")
time.sleep(0.3) # let the device settle
return port
print(".", end="", flush=True)
time.sleep(0.5)
def open_serial(port: str) -> serial.Serial:
return serial.Serial(port, timeout=2)
# ── COBS framing ───────────────────────────────────────────────────
def cobs_frame(payload: bytes) -> bytes:
return cobs.encode(payload) + b"\x00"
@@ -28,12 +82,10 @@ def read_frame(ser: serial.Serial) -> bytes | None:
return cobs.decode(buf) if buf else None
# ── Postcard serialization helpers ──────────────────────────────────
# Postcard uses varint encoding for enum variants and integers.
# ── Postcard serialization ─────────────────────────────────────────
def varint(n: int) -> bytes:
"""Encode unsigned varint (postcard style)."""
out = []
while n >= 0x80:
out.append((n & 0x7F) | 0x80)
@@ -42,16 +94,28 @@ def varint(n: int) -> bytes:
return bytes(out)
def zigzag(n: int) -> bytes:
return varint((n << 1) ^ (n >> 31) if n >= 0 else ((-n - 1) << 1) | 1)
def encode_radio_config(cfg: dict) -> 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: dict) -> bytes:
"""Encode a Command enum variant to postcard bytes."""
kind = cmd["type"]
if kind == "Ping":
return varint(0)
elif kind == "GetConfig":
return varint(1)
elif kind == "SetConfig":
cfg = cmd["config"]
return varint(2) + encode_radio_config(cfg)
return varint(2) + encode_radio_config(cmd["config"])
elif kind == "StartRx":
return varint(3)
elif kind == "StopRx":
@@ -64,62 +128,8 @@ def encode_command(cmd: dict) -> bytes:
raise ValueError(f"Unknown command type: {kind}")
def encode_radio_config(cfg: dict) -> bytes:
"""Encode RadioConfig struct to postcard bytes."""
# freq_hz: u32 as varint
out = varint(cfg["freq_hz"])
# bw: Bandwidth enum (u8 repr)
out += varint(cfg["bw"])
# sf: u8
out += varint(cfg["sf"])
# cr: CodingRate enum (u8 repr)
out += varint(cfg["cr"])
# sync_word: u16 as varint
out += varint(cfg["sync_word"])
# tx_power_dbm: i8 as zigzag varint
out += zigzag(cfg["tx_power_dbm"])
return out
def zigzag(n: int) -> bytes:
"""Encode signed int as zigzag varint."""
return varint((n << 1) ^ (n >> 31) if n >= 0 else ((-n - 1) << 1) | 1)
def decode_response(data: bytes) -> dict:
"""Decode a Response enum variant from postcard bytes."""
variant = data[0]
rest = data[1:]
if variant == 0:
return {"type": "Pong"}
elif variant == 1:
return {"type": "Config", "raw": rest.hex()}
elif variant == 2:
# RxPacket: rssi(i16 zigzag), snr(i8 zigzag), payload(len-prefixed)
rssi, rest = decode_zigzag_varint(rest)
snr, rest = decode_zigzag_varint(rest)
plen, rest = decode_varint(rest)
payload = rest[:plen]
return {
"type": "RxPacket",
"rssi": rssi,
"snr": snr,
"payload": payload,
}
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})", "raw": rest.hex()}
def decode_varint(data: bytes) -> tuple[int, bytes]:
n = 0
shift = 0
n, shift = 0, 0
for i, b in enumerate(data):
n |= (b & 0x7F) << shift
shift += 7
@@ -133,7 +143,31 @@ def decode_zigzag_varint(data: bytes) -> tuple[int, bytes]:
return (n >> 1) ^ -(n & 1), rest
# ── Main ────────────────────────────────────────────────────────────
def decode_response(data: bytes) -> dict:
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_varint(rest)
snr, rest = decode_zigzag_varint(rest)
plen, rest = decode_varint(rest)
payload = rest[:plen]
return {"type": "RxPacket", "rssi": rssi, "snr": snr, "payload": payload}
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})", "raw": rest.hex()}
# ── Main ───────────────────────────────────────────────────────────
def send_cmd(ser: serial.Serial, cmd: dict, label: str) -> dict | None:
@@ -151,55 +185,69 @@ def send_cmd(ser: serial.Serial, cmd: dict, label: str) -> dict | None:
return resp
def main():
port = sys.argv[1] if len(sys.argv) > 1 else "/dev/ttyACM0"
print(f"Opening {port}")
ser = serial.Serial(port, timeout=2)
ser.reset_input_buffer()
RADIO_CONFIG = {
"freq_hz": 910_525_000,
"bw": 6, # Khz62 = variant index 6
"sf": 7,
"cr": 0, # Cr4_5 = variant index 0
"sync_word": 0x3444,
"tx_power_dbm": 14,
}
# Ping
def configure_and_listen(ser: serial.Serial):
send_cmd(ser, {"type": "Ping"}, "Ping")
# Configure radio: 910.525 MHz, BW 62.5k, SF7, CR 4/5
config = {
"freq_hz": 910_525_000,
"bw": 6, # Khz62 = 6
"sf": 7,
"cr": 1, # Cr4_5 = 1
"sync_word": 0x3444,
"tx_power_dbm": 14,
}
send_cmd(ser, {"type": "SetConfig", "config": config}, "SetConfig 910.525/62.5k/SF7/CR4_5")
# Start receiving
send_cmd(ser, {"type": "SetConfig", "config": RADIO_CONFIG}, "SetConfig 910.525/62.5k/SF7/CR4_5")
send_cmd(ser, {"type": "StartRx"}, "StartRx")
# Listen for packets
print("\nListening for packets (Ctrl+C to stop)...\n")
ser.timeout = None # block forever
try:
while True:
data = read_frame(ser)
if data is None:
continue
resp = decode_response(data)
if resp["type"] == "RxPacket":
payload = resp["payload"]
try:
text = payload.decode("utf-8", errors="replace")
except Exception:
text = payload.hex()
print(
f" RSSI:{resp['rssi']:4d} dBm "
f"SNR:{resp['snr']:3d} dB "
f"len:{len(payload):3d} "
f"payload: {text}"
)
else:
print(f" {resp}")
except KeyboardInterrupt:
print("\nStopping...")
send_cmd(ser, {"type": "StopRx"}, "StopRx")
ser.timeout = None
while True:
data = read_frame(ser)
if data is None:
raise ConnectionError("device disconnected")
resp = decode_response(data)
if resp["type"] == "RxPacket":
payload = resp["payload"]
try:
text = payload.decode("utf-8", errors="replace")
except Exception:
text = payload.hex()
print(
f" RSSI:{resp['rssi']:4d}dBm "
f"SNR:{resp['snr']:3d}dB "
f"len:{len(payload):3d} "
f"{text}"
)
else:
print(f" {resp}")
def main():
port = sys.argv[1] if len(sys.argv) > 1 else None
while True:
if port is None:
port = wait_for_device()
try:
print(f"Opening {port}")
ser = open_serial(port)
ser.reset_input_buffer()
configure_and_listen(ser)
except (serial.SerialException, ConnectionError, OSError) as e:
print(f"\nDisconnected: {e}")
print("Will reconnect when device reappears...")
port = None
time.sleep(1)
except KeyboardInterrupt:
print("\nStopping...")
try:
ser.timeout = 2
send_cmd(ser, {"type": "StopRx"}, "StopRx")
except Exception:
pass
break
if __name__ == "__main__":