Y334275 commented on PR #21761:
URL: https://github.com/apache/nuttx/pull/21761#issuecomment-6816717165
## Host-side test script
This is the script used for the TX/RX and reconnect runs. It is not part
of the NuttX tree; run it on the host that owns the CDC ACM port.
```python
#!/usr/bin/env python3
"""Host-side TX/RX reconnect test for the mavlink_otg target.
TX check: HEARTBEAT custom_mode must keep increasing.
RX check: PARAM_REQUEST_LIST must return PARAM_VALUE frames.
Requires pyserial. Run it on the host that owns the CDC ACM port:
python otg_host_test.py --port COMx --cycles 20
python3 otg_host_test.py --port /dev/ttyACM0 --cycles 20
"""
import argparse
import time
import serial
MAV_MAGIC = 0xFD
MSG_HEARTBEAT = 0
MSG_PARAM_REQUEST_LIST = 21
MSG_PARAM_VALUE = 22
CRC_EXTRA_PARAM_REQUEST_LIST = 159
def crc_accumulate(crc, data):
tmp = data ^ (crc & 0xFF)
tmp = (tmp ^ (tmp << 4)) & 0xFF
return ((crc >> 8) ^ (tmp << 8) ^ (tmp << 3) ^ (tmp >> 4)) & 0xFFFF
def request_frame(seq):
payload = bytes((1, 1))
header = bytes((len(payload), 0, 0, seq & 0xFF, 255, 190,
MSG_PARAM_REQUEST_LIST, 0, 0))
crc = 0xFFFF
for byte in header + payload:
crc = crc_accumulate(crc, byte)
crc = crc_accumulate(crc, CRC_EXTRA_PARAM_REQUEST_LIST)
return bytes((MAV_MAGIC,)) + header + payload + bytes(
(crc & 0xFF, crc >> 8))
def parse_frames(data):
frames = []
i = 0
while i + 12 <= len(data):
if data[i] != MAV_MAGIC:
i += 1
continue
payload_len = data[i + 1]
frame_len = 10 + payload_len + 2
if i + frame_len > len(data):
break
msgid = data[i + 7] | (data[i + 8] << 8) | (data[i + 9] << 16)
frames.append((msgid, data[i + 10:i + 10 + payload_len]))
i += frame_len
return frames
def read_for(ser, seconds, quiet=0.3):
data = b""
last = time.time()
end = last + seconds
while time.time() < end:
chunk = ser.read(4096)
if chunk:
data += chunk
last = time.time()
elif time.time() - last > quiet:
break
return data
def open_port(port, baud):
deadline = time.time() + 5.0
while True:
try:
return serial.Serial(port, baud, timeout=0.1)
except serial.SerialException:
if time.time() > deadline:
raise
time.sleep(0.2)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--port", default="/dev/ttyACM0")
parser.add_argument("--baud", type=int, default=115200)
parser.add_argument("--cycles", type=int, default=20)
args = parser.parse_args()
failures = 0
last_counter = None
for cycle in range(args.cycles):
try:
ser = open_port(args.port, args.baud)
time.sleep(0.3)
# Host connects: observe TX without sending anything.
tx_data = read_for(ser, 1.0)
heartbeats = []
for msgid, payload in parse_frames(tx_data):
if msgid == MSG_HEARTBEAT and len(payload) >= 9:
heartbeats.append(int.from_bytes(payload[0:4], "little"))
# Host requests parameters.
ser.write(request_frame(cycle))
rx_data = read_for(ser, 1.0)
params = sum(1 for msgid, _ in parse_frames(rx_data)
if msgid == MSG_PARAM_VALUE)
ser.close()
except Exception as exc: # noqa: BLE001
print("cycle %d: ERROR %s" % (cycle, exc))
failures += 1
time.sleep(0.3)
continue
tx_ok = len(heartbeats) >= 5
tx_seq_ok = True
if heartbeats:
if last_counter is not None and heartbeats[0] <= last_counter:
tx_seq_ok = False
last_counter = heartbeats[-1]
rx_ok = params >= 14
ok = tx_ok and tx_seq_ok and rx_ok
tx_range = ("%d..%d" % (heartbeats[0], heartbeats[-1])
if heartbeats else "none")
print("cycle %d: tx_hb=%d tx_seq=%s params=%d %s"
% (cycle, len(heartbeats), tx_range, params,
"OK" if ok else "FAIL"))
if not ok:
failures += 1
# Host disconnects; the port is already closed above.
time.sleep(0.3)
print("failures=%d/%d" % (failures, args.cycles))
return 1 if failures else 0
if __name__ == "__main__":
raise SystemExit(main())
```
Usage:
```bash
python otg_host_test.py --port COMx --cycles 20
# or
python3 otg_host_test.py --port /dev/ttyACM0 --cycles 20
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
