Files
examples/ping_pong.py
T
swaitsandClaude Opus 4.6 ef8261da57 fix: update dependency path, remove dead recipes, apply lint fixes
Fixed broken dependency path (../clients/python -> ../client-py).
Added strict ruff config. Removed dead meshcore/orac/telemetry
justfile recipes (scripts never existed). Applied SIM105
(contextlib.suppress) and I001 (import sorting) lint fixes.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-08 13:34:00 -06:00

58 lines
1.7 KiB
Python
Executable File

#!/usr/bin/env python3
"""Two-dongle demo: one transmits, one receives.
Usage:
python examples/ping_pong.py --role tx [PORT]
python examples/ping_pong.py --role rx [PORT]
"""
import argparse
import contextlib
import sys
import time
import serial
import donglora as dl
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()
try:
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
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)
else:
print(dl.send(ser, "StartRx"))
print("Receiving (Ctrl+C to stop)...\n")
ser.timeout = 1
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:
with contextlib.suppress(Exception):
dl.send(ser, "StopRx")
print("\nDone.")
except serial.SerialException as e:
print(f"\nSerial error: {e}", file=sys.stderr)
sys.exit(1)