fix: handle SerialException gracefully in all example scripts
All examples now catch serial.SerialException instead of crashing with a traceback. This covers port contention (multiple processes), device disconnect, and USB errors. meshcore_rx.py already had this.
This commit is contained in:
1 parent
68753b5dfe
commit
84be92f263
3 files changed
+46
-27
No files matched your search
+20
-15
@@ -11,6 +11,7 @@ Usage:
|
|||||||
# ///
|
# ///
|
||||||
|
|
||||||
import argparse
|
import argparse
|
||||||
|
import serial
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
|
|
||||||
@@ -22,27 +23,24 @@ parser.add_argument("--role", choices=["tx", "rx"], required=True)
|
|||||||
parser.add_argument("port", nargs="?", help="Serial port (auto-detect if omitted)")
|
parser.add_argument("port", nargs="?", help="Serial port (auto-detect if omitted)")
|
||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
|
|
||||||
ser = dl.connect(args.port)
|
try:
|
||||||
print(dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG))
|
ser = dl.connect(args.port)
|
||||||
|
print(dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG))
|
||||||
|
|
||||||
if args.role == "tx":
|
if args.role == "tx":
|
||||||
print("Transmitting every 2 seconds (Ctrl+C to stop)...\n")
|
print("Transmitting every 2 seconds (Ctrl+C to stop)...\n")
|
||||||
seq = 0
|
seq = 0
|
||||||
try:
|
|
||||||
while True:
|
while True:
|
||||||
msg = f"ping #{seq}"
|
msg = f"ping #{seq}"
|
||||||
resp = dl.send(ser, "Transmit", payload=msg.encode())
|
resp = dl.send(ser, "Transmit", payload=msg.encode())
|
||||||
print(f" TX: {msg!r} → {resp['type']}")
|
print(f" TX: {msg!r} → {resp['type']}")
|
||||||
seq += 1
|
seq += 1
|
||||||
time.sleep(2)
|
time.sleep(2)
|
||||||
except KeyboardInterrupt:
|
|
||||||
print("\nDone.")
|
|
||||||
|
|
||||||
else:
|
else:
|
||||||
print(dl.send(ser, "StartRx"))
|
print(dl.send(ser, "StartRx"))
|
||||||
print("Receiving (Ctrl+C to stop)...\n")
|
print("Receiving (Ctrl+C to stop)...\n")
|
||||||
ser.timeout = 1
|
ser.timeout = 1
|
||||||
try:
|
|
||||||
while True:
|
while True:
|
||||||
data = dl.read_frame(ser)
|
data = dl.read_frame(ser)
|
||||||
if data is None:
|
if data is None:
|
||||||
@@ -52,6 +50,13 @@ else:
|
|||||||
p = resp["payload"]
|
p = resp["payload"]
|
||||||
text = p.decode("utf-8", errors="replace")
|
text = p.decode("utf-8", errors="replace")
|
||||||
print(f" RX: {text!r} RSSI:{resp['rssi']}dBm SNR:{resp['snr']}dB")
|
print(f" RX: {text!r} RSSI:{resp['rssi']}dBm SNR:{resp['snr']}dB")
|
||||||
except KeyboardInterrupt:
|
|
||||||
|
except KeyboardInterrupt:
|
||||||
|
try:
|
||||||
dl.send(ser, "StopRx")
|
dl.send(ser, "StopRx")
|
||||||
print("\nDone.")
|
except Exception:
|
||||||
|
pass
|
||||||
|
print("\nDone.")
|
||||||
|
except serial.SerialException as e:
|
||||||
|
print(f"\nSerial error: {e}", file=sys.stderr)
|
||||||
|
sys.exit(1)
|
||||||
+16
-8
@@ -9,18 +9,20 @@ Usage:
|
|||||||
# dependencies = ["cobs", "pyserial"]
|
# dependencies = ["cobs", "pyserial"]
|
||||||
# ///
|
# ///
|
||||||
|
|
||||||
|
import serial
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
sys.path.insert(0, __import__("pathlib").Path(__file__).parent.as_posix())
|
sys.path.insert(0, __import__("pathlib").Path(__file__).parent.as_posix())
|
||||||
import donglora as dl # noqa: E402
|
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:
|
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"))
|
||||||
|
|
||||||
|
print("\nListening (Ctrl+C to stop)...\n")
|
||||||
|
ser.timeout = 1
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
data = dl.read_frame(ser)
|
data = dl.read_frame(ser)
|
||||||
if data is None:
|
if data is None:
|
||||||
@@ -30,5 +32,11 @@ try:
|
|||||||
p = resp["payload"]
|
p = resp["payload"]
|
||||||
print(f" RSSI:{resp['rssi']:4d}dBm SNR:{resp['snr']:3d}dB len:{len(p):3d} {p.hex()}")
|
print(f" RSSI:{resp['rssi']:4d}dBm SNR:{resp['snr']:3d}dB len:{len(p):3d} {p.hex()}")
|
||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
dl.send(ser, "StopRx")
|
try:
|
||||||
|
dl.send(ser, "StopRx")
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
print("\nDone.")
|
print("\nDone.")
|
||||||
|
except serial.SerialException as e:
|
||||||
|
print(f"\nSerial error: {e}", file=sys.stderr)
|
||||||
|
sys.exit(1)
|
||||||
+10
-4
@@ -9,7 +9,9 @@ Usage:
|
|||||||
# dependencies = ["cobs", "pyserial"]
|
# dependencies = ["cobs", "pyserial"]
|
||||||
# ///
|
# ///
|
||||||
|
|
||||||
|
import serial
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
sys.path.insert(0, __import__("pathlib").Path(__file__).parent.as_posix())
|
sys.path.insert(0, __import__("pathlib").Path(__file__).parent.as_posix())
|
||||||
import donglora as dl # noqa: E402
|
import donglora as dl # noqa: E402
|
||||||
|
|
||||||
@@ -24,7 +26,11 @@ elif args:
|
|||||||
if len(args) > 1:
|
if len(args) > 1:
|
||||||
message = " ".join(args[1:])
|
message = " ".join(args[1:])
|
||||||
|
|
||||||
ser = dl.connect(port)
|
try:
|
||||||
print(dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG))
|
ser = dl.connect(port)
|
||||||
print(dl.send(ser, "Transmit", payload=message.encode()))
|
print(dl.send(ser, "SetConfig", config=dl.DEFAULT_CONFIG))
|
||||||
print(f"Sent: {message!r}")
|
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)
|
||||||
Reference in new issue
Block a user