86 lines
2.4 KiB
Python
86 lines
2.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
import argparse
|
||
|
|
import glob
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
|
||
|
|
try:
|
||
|
|
import serial
|
||
|
|
except ImportError:
|
||
|
|
print("need pyserial: sudo apt install python3-serial (or pip3 install pyserial)")
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
|
||
|
|
def list_ports():
|
||
|
|
ports = sorted(glob.glob("/dev/ttyUSB*") + glob.glob("/dev/ttyACM*") + glob.glob("/dev/ttyS*"))
|
||
|
|
for p in ports:
|
||
|
|
print(p)
|
||
|
|
if not ports:
|
||
|
|
print("no serial ports found")
|
||
|
|
return ports
|
||
|
|
|
||
|
|
|
||
|
|
def open_port(name, baud):
|
||
|
|
sp = serial.Serial(None, baud, bytesize=8, parity="N", stopbits=1, timeout=0.3)
|
||
|
|
sp.port = name
|
||
|
|
sp.open()
|
||
|
|
return sp
|
||
|
|
|
||
|
|
|
||
|
|
def drain(sp, seconds):
|
||
|
|
deadline = time.time() + seconds
|
||
|
|
buf = b""
|
||
|
|
while time.time() < deadline:
|
||
|
|
chunk = sp.read(256)
|
||
|
|
if chunk:
|
||
|
|
buf += chunk
|
||
|
|
return buf
|
||
|
|
|
||
|
|
|
||
|
|
def show(buf):
|
||
|
|
if not buf:
|
||
|
|
return "(no bytes)"
|
||
|
|
text = buf.decode("ascii", "replace").replace("\n", "\\n").replace("\r", "\\r")
|
||
|
|
hexs = " ".join(f"{b:02X}" for b in buf)
|
||
|
|
return f"ascii[{text}] hex[{hexs}]"
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
ap = argparse.ArgumentParser(description="RS485 bus test tool for LPC verification")
|
||
|
|
ap.add_argument("-l", "--list", action="store_true", help="list serial ports")
|
||
|
|
ap.add_argument("-p", "--port", default="/dev/ttyUSB0")
|
||
|
|
ap.add_argument("-b", "--baud", type=int, default=9600)
|
||
|
|
ap.add_argument("-s", "--send", help='text to send, \\n \\r escaped')
|
||
|
|
ap.add_argument("-w", "--listen", type=float, default=0, help="seconds to listen after send")
|
||
|
|
ap.add_argument("-i", "--interactive", action="store_true", help="read line from stdin, send, show reply")
|
||
|
|
args = ap.parse_args()
|
||
|
|
|
||
|
|
if args.list:
|
||
|
|
list_ports()
|
||
|
|
return
|
||
|
|
|
||
|
|
sp = open_port(args.port, args.baud)
|
||
|
|
print(f"opened {args.port} @ {args.baud} 8N1, Ctrl+C to exit")
|
||
|
|
try:
|
||
|
|
if args.send:
|
||
|
|
payload = args.send.replace("\\n", "\n").replace("\\r", "\r").encode("ascii")
|
||
|
|
sp.write(payload)
|
||
|
|
sp.flush()
|
||
|
|
print(f"TX: {payload!r}")
|
||
|
|
if args.interactive:
|
||
|
|
while True:
|
||
|
|
line = input("TX> ")
|
||
|
|
if not line:
|
||
|
|
continue
|
||
|
|
sp.write((line + "\n").encode("ascii"))
|
||
|
|
sp.flush()
|
||
|
|
print("RX:", show(drain(sp, 0.3)))
|
||
|
|
if args.listen > 0:
|
||
|
|
print("RX:", show(drain(sp, args.listen)))
|
||
|
|
finally:
|
||
|
|
sp.close()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|