diff --git a/src/display/render.rs b/src/display/render.rs index 67bfab16..a724a9eb 100644 --- a/src/display/render.rs +++ b/src/display/render.rs @@ -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"); diff --git a/tools/test_ping.py b/tools/test_ping.py index d25823cc..3c86ef16 100644 --- a/tools/test_ping.py +++ b/tools/test_ping.py @@ -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__":