donglora/examples
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat!: examples on donglora 1.0 async API with shared _common helpers

Browse Source 
Rewrite all scripts against the DongLoRa Protocol v2 async Dongle surface, and extract the shared MeshCore US preset into a single module.

* _common.py: MESHCORE_US preset (910.525 MHz / BW 62.5 / SF 7 / CR 4/5 / sync 0x1424 / 20 dBm) plus small helpers, imported by every example.

* New: explore.py, which prints the device capabilities and runs a single TX + RX roundtrip for quick hardware validation.

* simple_rx.py, simple_tx.py, ping_pong.py, lora_bridge.py all switch to donglora.Dongle and the new Modulation / LoRaConfig types.

* all_commands.py removed; it was an index of the old v0.1 command enum that no longer exists after the Protocol v2 split.

* justfile: drop stale recipes that referenced removed scripts.
This commit is contained in:
swaits committed 2026-04-22 09:59:16 -06:00
1 parent 80c02788cc
commit 978553e5ac
8 files changed
+204 -209
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

_common.py
+36
View File
@@ -0,0 +1,36 @@
"""Shared defaults for the example scripts.
MeshCore's "USA/Canada (Recommended)" preset, per
https://github.com/ripplebiz/MeshCore/blob/main/docs/faq.md §6.1
(the "narrow" channel the project moved to in October 2025):
* 910.525 MHz — inside the 902-928 MHz ISM band
* 62.5 kHz bandwidth
* SF 7 — lower SF for faster transmissions and better SNR headroom
* CR 4/5 — RadioLib's ``LORA_CR=5``
* 16-symbol preamble — matches every board variant's ``radio.begin(...)``
* Sync word 0x1424 — ``RADIOLIB_SX126X_SYNC_WORD_PRIVATE``, the SX126x
register encoding of the "private" byte 0x12 MeshCore uses to
distinguish its traffic from LoRaWAN (byte 0x34, encoded 0x3444).
* 20 dBm TX power — MeshCore's compile-time default (see
``examples/companion_radio/MyMesh.h``).
If you operate in the EU, swap ``freq_hz`` for ``869_525_000`` (the
EU-equivalent MeshCore channel) and cap ``tx_power_dbm`` at ``14`` to
respect the 868 MHz duty-cycle / ERP rules.
"""
import donglora as dl
MESHCORE_US = dl.LoRaConfig(
freq_hz=910_525_000,
sf=7,
bw=dl.LoRaBandwidth.KHZ_62_5,
cr=dl.LoRaCodingRate.CR_4_5,
preamble_len=16,
sync_word=0x1424,
tx_power_dbm=20,
header_mode=dl.LoRaHeaderMode.EXPLICIT,
payload_crc=True,
iq_invert=False,
)
all_commands.py
-50
View File
@@ -1,50 +0,0 @@
#!/usr/bin/env python3
"""Exercise every DongLoRa command to verify the API works.
Usage:
python examples/all_commands.py [PORT]
"""
import sys
import serial
import donglora as dl
try:
ser = dl.connect(sys.argv[1] if len(sys.argv) > 1 else None)
print("── Ping ──")
print(dl.send(ser, "Ping"))
print("\n── SetConfig ──")
print(dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG))
print("\n── GetConfig ──")
print(dl.send(ser, "GetConfig"))
print("\n── StartRx ──")
print(dl.send(ser, "StartRx"))
print("\n── StopRx ──")
print(dl.send(ser, "StopRx"))
print("\n── Transmit ──")
print(dl.send(ser, "Transmit", payload=b"Hello from all_commands.py!"))
print("\n── DisplayOff ──")
print(dl.send(ser, "DisplayOff"))
print("\n── DisplayOn ──")
print(dl.send(ser, "DisplayOn"))
print("\n── GetMac ──")
print(dl.send(ser, "GetMac"))
print("\nAll 9 commands exercised successfully.")
except KeyboardInterrupt:
print("\nInterrupted.")
except serial.SerialException as e:
print(f"\nSerial error: {e}", file=sys.stderr)
sys.exit(1)
explore.py
+72
View File
@@ -0,0 +1,72 @@
#!/usr/bin/env python3
"""Probe a DongLoRa device: print capabilities, then TX + RX once.
Configures the radio for the MeshCore US/Canada channel so the TX
probe is audible to other MeshCore nodes on the same band.
Usage:
python examples/explore.py [PORT]
"""
import sys
import time
import donglora as dl
from _common import MESHCORE_US
def describe_info(info: dl.Info) -> None:
chip = info.chip.name if isinstance(info.chip, dl.RadioChipId) else f"0x{info.chip:04X}"
print(f" Protocol: DongLoRa Protocol v{info.proto_major}.{info.proto_minor}")
print(f" Firmware: v{info.fw_major}.{info.fw_minor}.{info.fw_patch}")
print(f" Chip: {chip}")
print(f" Freq: {info.freq_min_hz / 1e6:.1f} - {info.freq_max_hz / 1e6:.1f} MHz")
print(f" TX power: {info.tx_power_min_dbm}..{info.tx_power_max_dbm} dBm")
print(f" Max pkt: {info.max_payload_bytes} bytes")
print(f" SF range: {info.supported_sf()}")
print(f" MCU UID: {info.mcu_uid.hex()}")
caps = []
if info.supports(dl.Capability.LORA):
caps.append("LoRa")
if info.supports(dl.Capability.FSK):
caps.append("FSK")
if info.supports(dl.Capability.CAD_BEFORE_TX):
caps.append("CAD")
print(f" Features: {', '.join(caps) or 'none advertised'}")
port = sys.argv[1] if len(sys.argv) > 1 else None
with dl.connect(port, config=MESHCORE_US) as d:
print("── GET_INFO ──")
describe_info(d.info)
cfg = d.config
print(
f"\n── Radio configured: {cfg.freq_hz / 1e6:.3f} MHz "
f"SF{cfg.sf} BW{cfg.bw.khz:g}kHz "
f"sync=0x{cfg.sync_word:04X} {cfg.tx_power_dbm} dBm ──"
)
print("\n── TX ──")
td = d.tx(b"hello from explore.py")
print(
f" result={td.result.name} airtime_us={td.airtime_us} "
f"(~{td.airtime_us / 1000:.1f} ms on air)"
)
print("\n── RX (5 s window) ──")
start = time.monotonic()
any_rx = False
for pkt in d.rx(timeout=0.25):
any_rx = True
print(
f" pkt: rssi={pkt.rssi_dbm:.1f}dBm snr={pkt.snr_db:.1f}dB "
f"len={len(pkt.data)} origin={pkt.origin.name}"
)
if time.monotonic() - start > 5:
break
if not any_rx:
print(" (no packets received — try pairing with ping_pong.py --role tx)")
print("\nDone.")
justfile
+3 -3
View File
@@ -34,9 +34,9 @@ tx *args: _ensure_tools
ping-pong *args: _ensure_tools
@{{_uv}} {{source_directory()}}/ping_pong.py {{args}}
# Exercise all DongLoRa commands
test-commands *args: _ensure_tools
@{{_uv}} {{source_directory()}}/all_commands.py {{args}}
# Probe the connected DongLoRa: dump GET_INFO and exercise one TX + RX
explore *args: _ensure_tools
@{{_uv}} {{source_directory()}}/explore.py {{args}}
# Two-way LoRa bridge over TCP
bridge *args: _ensure_tools
lora_bridge.py
+58 -83
View File
@@ -2,20 +2,22 @@
"""Two-way LoRa bridge over TCP — relay packets between two DongLoRa dongles
across any IP network (LAN, Tailscale, WireGuard, internet).
Architecture:
Architecture::
[Radio A] <-USB-> [DongLoRa A] <-TCP-> [DongLoRa B] <-USB-> [Radio B]
Usage:
Usage::
# Machine A (server):
python examples/lora_bridge.py --mode server --port 9100
# Machine B (client):
python examples/lora_bridge.py --mode client --host machineA --port 9100
Both sides:
- Open local DongLoRa, configure radio, start RX
- Forward received LoRa packets -> TCP -> remote side -> TX
- Bidirectional: both sides receive AND transmit
Both sides open a local DongLoRa, listen for incoming LoRa packets,
forward them over TCP to the peer, and transmit any TCP-arriving
bytes over the air. The Dongle class handles keepalive + recovery
internally, so no extra bookkeeping is needed here.
"""
import argparse
@@ -25,9 +27,8 @@ import sys
import threading
import time
import serial
import donglora as dl
from _common import MESHCORE_US
def tcp_send(sock: socket.socket, payload: bytes):
@@ -45,7 +46,7 @@ def tcp_recv(sock: socket.socket) -> bytes | None:
header += chunk
length = struct.unpack("<I", header)[0]
if length > 65536:
return None # sanity check
return None
data = b""
while len(data) < length:
chunk = sock.recv(length - len(data))
@@ -55,81 +56,50 @@ def tcp_recv(sock: socket.socket) -> bytes | None:
return data
_ser_lock = threading.Lock()
def radio_to_tcp(ser: serial.Serial, sock: socket.socket):
def radio_to_tcp(d: dl.Dongle, sock: socket.socket):
"""Thread: forward LoRa RX packets to TCP."""
ser.timeout = 1
while True:
for pkt in d.rx(timeout=1.0):
try:
with _ser_lock:
data = dl.read_frame(ser)
if data is None:
sf = d.config.sf if isinstance(d.config, dl.LoRaConfig) else 7
min_snr = -2.5 * (sf - 4)
if pkt.snr_db < -32 or pkt.snr_db > 32:
grade = "INVALID"
elif pkt.snr_db < min_snr:
grade = "UNRELIABLE"
elif pkt.snr_db < min_snr + 3:
grade = "MARGINAL"
else:
grade = "GOOD"
if grade in ("INVALID", "UNRELIABLE"):
print(
f" RX drop len:{len(pkt.data):3d} "
f"rssi:{pkt.rssi_dbm:.1f}dBm snr:{pkt.snr_db:.1f}dB [{grade}]"
)
continue
resp = dl.decode_response(data)
if resp["type"] == "RxPacket":
payload = resp["payload"]
snr = resp["snr"]
rssi = resp["rssi"]
sf = dl.DEFAULT_CONFIG["sf"]
min_snr = -2.5 * (sf - 4)
# Grade per PROTOCOL.md — drop likely-corrupt packets
if snr < -32 or snr > 32:
grade = "INVALID"
elif snr < min_snr:
grade = "UNRELIABLE"
elif snr < min_snr + 3:
grade = "MARGINAL"
else:
grade = "GOOD"
if grade in ("INVALID", "UNRELIABLE"):
print(
f" RX drop len:{len(payload):3d} rssi:{rssi}dBm snr:{snr}dB [{grade}]"
)
continue
tag = f" [{grade}]" if grade == "MARGINAL" else ""
print(f" RX→TCP len:{len(payload):3d} rssi:{rssi}dBm snr:{snr}dB{tag}")
tcp_send(sock, payload)
except (serial.SerialException, OSError) as e:
print(f" [radio→tcp error: {e}]")
break
tag = f" [{grade}]" if grade == "MARGINAL" else ""
print(
f" RX→TCP len:{len(pkt.data):3d} "
f"rssi:{pkt.rssi_dbm:.1f}dBm snr:{pkt.snr_db:.1f}dB{tag}"
)
tcp_send(sock, pkt.data)
except OSError as exc:
print(f" [radio→tcp error: {exc}]")
return
def tcp_to_radio(ser: serial.Serial, sock: socket.socket):
def tcp_to_radio(d: dl.Dongle, sock: socket.socket):
"""Thread: forward TCP packets to LoRa TX."""
while True:
try:
payload = tcp_recv(sock)
if payload is None:
print(" [TCP disconnected]")
break
return
print(f" TCP→TX len:{len(payload):3d}")
with _ser_lock:
dl.send(ser, "Transmit", payload=payload)
except (serial.SerialException, OSError) as e:
print(f" [tcp→radio error: {e}]")
break
def run_bridge(ser: serial.Serial, sock: socket.socket):
"""Run bidirectional bridge between serial and TCP."""
print("Bridge active — forwarding packets bidirectionally\n")
t1 = threading.Thread(target=radio_to_tcp, args=(ser, sock), daemon=True)
t2 = threading.Thread(target=tcp_to_radio, args=(ser, sock), daemon=True)
t1.start()
t2.start()
try:
while t1.is_alive() and t2.is_alive():
time.sleep(0.5)
except KeyboardInterrupt:
pass
print("\nBridge stopped.")
d.tx(payload)
except (OSError, dl.DongloraError) as exc:
print(f" [tcp→radio error: {exc}]")
return
def main():
@@ -141,12 +111,6 @@ def main():
args = parser.parse_args()
try:
# Open DongLoRa
ser = dl.connect(args.serial)
print(dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG))
print(dl.send(ser, "StartRx"))
# Establish TCP connection
if args.mode == "server":
print(f"Listening on port {args.port}...")
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -162,15 +126,26 @@ def main():
sock.connect((args.host, args.port))
print("Connected")
run_bridge(ser, sock)
with dl.connect(args.serial, config=MESHCORE_US) as d:
print(f"Radio ready on {d.config.freq_hz / 1e6:.3f} MHz — bridging packets\n")
t1 = threading.Thread(target=radio_to_tcp, args=(d, sock), daemon=True)
t2 = threading.Thread(target=tcp_to_radio, args=(d, sock), daemon=True)
t1.start()
t2.start()
try:
while t1.is_alive() and t2.is_alive():
time.sleep(0.5)
except KeyboardInterrupt:
pass
print("\nBridge stopped.")
except KeyboardInterrupt:
print("\nInterrupted.")
except serial.SerialException as e:
print(f"\nSerial error: {e}", file=sys.stderr)
except dl.DongloraError as exc:
print(f"\nDevice error: {exc}", file=sys.stderr)
sys.exit(1)
except OSError as e:
print(f"\nNetwork error: {e}", file=sys.stderr)
except OSError as exc:
print(f"\nNetwork error: {exc}", file=sys.stderr)
sys.exit(1)
ping_pong.py
+19 -36
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Two-dongle demo: one transmits, one receives.
"""Two-dongle demo on the MeshCore US/Canada channel: one TX, one RX.
Usage:
python examples/ping_pong.py --role tx [PORT]
@@ -7,51 +7,34 @@ Usage:
"""
import argparse
import contextlib
import sys
import time
import serial
import donglora as dl
from _common import MESHCORE_US
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))
with dl.connect(args.port, config=MESHCORE_US) as d:
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)
try:
seq = 0
while True:
msg = f"ping #{seq}"
td = d.tx(msg.encode())
print(f" TX: {msg!r} ({td.airtime_us} µs)")
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
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)
try:
for pkt in d.rx():
text = pkt.data.decode("utf-8", errors="replace")
print(f" RX: {text!r} RSSI:{pkt.rssi_dbm:.1f}dBm SNR:{pkt.snr_db:.1f}dB")
except KeyboardInterrupt:
print("\nDone.")
simple_rx.py
+11 -26
View File
@@ -1,39 +1,24 @@
#!/usr/bin/env python3
"""Receive LoRa packets and print them.
"""Receive LoRa packets on the MeshCore US/Canada channel.
Usage:
python examples/simple_rx.py [PORT]
"""
import contextlib
import sys
import serial
import donglora as dl
from _common import MESHCORE_US
try:
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"))
port = sys.argv[1] if len(sys.argv) > 1 else None
print("\nListening (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"]
with dl.connect(port, config=MESHCORE_US) as d:
print(f"Listening on {d.config.freq_hz / 1e6:.3f} MHz (Ctrl+C to stop)...\n")
try:
for pkt in d.rx():
print(
f" RSSI:{resp['rssi']:4d}dBm SNR:{resp['snr']:3d}dB len:{len(p):3d} {p.hex()}"
f" RSSI:{pkt.rssi_dbm:6.1f} dBm SNR:{pkt.snr_db:5.1f} dB "
f"len:{len(pkt.data):3d} {pkt.data.hex()}"
)
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)
except KeyboardInterrupt:
print("\nDone.")
simple_tx.py
+5 -11
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Transmit a single LoRa packet.
"""Transmit a single LoRa packet on the MeshCore US/Canada channel.
Usage:
python examples/simple_tx.py [PORT] [MESSAGE]
@@ -7,9 +7,8 @@ Usage:
import sys
import serial
import donglora as dl
from _common import MESHCORE_US
port = None
message = "Hello from DongLoRa!"
@@ -22,11 +21,6 @@ elif args:
if len(args) > 1:
message = " ".join(args[1:])
try:
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}")
except serial.SerialException as e:
print(f"\nSerial error: {e}", file=sys.stderr)
sys.exit(1)
with dl.connect(port, config=MESHCORE_US) as d:
td = d.tx(message.encode())
print(f"Sent {message!r} ({td.airtime_us} µs on air)")