diff --git a/scripts/speed_test.py b/scripts/speed_test.py new file mode 100644 index 0000000..ec448ad --- /dev/null +++ b/scripts/speed_test.py @@ -0,0 +1,263 @@ +#!/usr/bin/env python3 + +import serial +import threading +import time + +# ========================== +# Configuration +# ========================== + +PORT = "/dev/ttyACM0" +BAUDRATE = 115200 + +COMMAND_PREFIX = 0x69 + +COMMAND_ACK = 0 +COMMAND_NACK = 1 +LED = 2 + +DEVICE_ID = 0 + +# Test parameters +TEST_DURATION = 2.0 # seconds per stage +LOSS_THRESHOLD = 1.0 # percent + +# Rates to test (packets/second) +RATES = [ + 100, + 200, + 500, + 1000, + 2000, + 4000, + 8000, + 16000, +] + +# ========================== +# Statistics +# ========================== + +lock = threading.Lock() + +tx_packets = 0 +tx_bytes = 0 + +rx_packets = 0 +rx_ack = 0 +rx_nack = 0 +rx_error = 0 + + +# ========================== +# Packet helpers +# ========================== + +def calculate_crc(msg: bytes) -> int: + s = sum(msg) & 0xFF + return (-s) & 0xFF + + +def make_packet(command: int, data: bytes = b"") -> bytes: + pkt = bytearray() + + pkt.append(COMMAND_PREFIX) + pkt.append(len(data)) + pkt.append(DEVICE_ID) + pkt.append(command) + pkt.append(0) + + pkt.extend(data) + + pkt[4] = calculate_crc(pkt[:4] + pkt[5:]) + return bytes(pkt) + + +def verify_crc(packet: bytes) -> bool: + crc = packet[4] + calc = calculate_crc(packet[:4] + packet[5:]) + return crc == calc + + +def packet_size(buf: bytes): + if len(buf) < 2: + return None + return 5 + buf[1] + + +# ========================== +# Receiver +# ========================== + +def reader(ser): + global rx_packets, rx_ack, rx_nack, rx_error + + rx = bytearray() + + while True: + data = ser.read(4096) + if not data: + continue + + rx.extend(data) + + while True: + if not rx: + break + + # ASCII log output + if rx[0] != COMMAND_PREFIX: + nl = rx.find(b"\n") + if nl == -1: + rx.clear() + break + + line = rx[: nl + 1] + del rx[: nl + 1] + + try: + print("[LOG]", line.decode().rstrip()) + except UnicodeDecodeError: + pass + + continue + + size = packet_size(rx) + + if size is None or len(rx) < size: + break + + pkt = bytes(rx[:size]) + del rx[:size] + + if not verify_crc(pkt): + continue + + cmd = pkt[3] + + with lock: + rx_packets += 1 + + if cmd == COMMAND_ACK: + rx_ack += 1 + elif cmd == COMMAND_NACK: + rx_nack += 1 + elif cmd == COMMAND_ERROR: + rx_error += 1 + + +# ========================== +# Benchmark +# ========================== + +def run_stage(ser, packet, target_pps): + global tx_packets, tx_bytes + global rx_packets, rx_ack, rx_nack, rx_error + + with lock: + tx_packets = 0 + tx_bytes = 0 + rx_packets = 0 + rx_ack = 0 + rx_nack = 0 + rx_error = 0 + + interval = 1.0 / target_pps + + start = time.perf_counter() + end = start + TEST_DURATION + next_tx = start + + while True: + now = time.perf_counter() + + if now >= end: + break + + if now >= next_tx: + ser.write(packet) + + with lock: + tx_packets += 1 + tx_bytes += len(packet) + + next_tx += interval + else: + sleep = next_tx - now + if sleep > 0: + time.sleep(min(sleep, 0.0005)) + + # Allow final ACKs to arrive + time.sleep(0.25) + + elapsed = time.perf_counter() - start + + with lock: + tx = tx_packets + ack = rx_ack + nack = rx_nack + err = rx_error + bytes_sent = tx_bytes + + loss = 0.0 + if tx: + loss = (tx - ack) / tx * 100.0 + + print() + print("=" * 60) + print(f"Target Rate : {target_pps:>7} pkt/s") + print(f"Elapsed : {elapsed:.3f} s") + print(f"Sent : {tx}") + print(f"ACK : {ack}") + print(f"NACK : {nack}") + print(f"ERROR : {err}") + print(f"Loss : {loss:.2f}%") + print(f"Actual TX : {tx / elapsed:.0f} pkt/s") + print(f"Throughput : {bytes_sent / elapsed / 1024:.2f} KiB/s") + + return loss + + +# ========================== +# Main +# ========================== + +def main(): + packet = make_packet(LED) + + print(f"Opening {PORT} @ {BAUDRATE} baud...") + ser = serial.Serial(PORT, BAUDRATE, timeout=0.01) + + threading.Thread(target=reader, args=(ser,), daemon=True).start() + + print("\nStarting communication benchmark...\n") + + previous_rate = None + + for rate in RATES: + loss = run_stage(ser, packet, rate) + + if loss > LOSS_THRESHOLD: + print() + print( + f"Link became unreliable (> {LOSS_THRESHOLD:.1f}% loss)." + ) + + if previous_rate is not None: + print(f"Maximum reliable rate ≈ {previous_rate} pkt/s") + + break + + previous_rate = rate + + else: + print() + print("Completed all test stages.") + print(f"Reliable up to at least {RATES[-1]} pkt/s.") + + ser.close() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/scripts/speed_test2.py b/scripts/speed_test2.py new file mode 100644 index 0000000..dda2ba2 --- /dev/null +++ b/scripts/speed_test2.py @@ -0,0 +1,224 @@ +#!/usr/bin/env python3 + +import serial +import threading +import time + +# ========================================================== +# Configuration +# ========================================================== + +PORT = "/dev/ttyACM0" +BAUDRATE = 115200 + +COMMAND_PREFIX = 0x69 + +COMMAND_ACK = 0 +COMMAND_NACK = 1 +LED = 2 + +DEVICE_ID = 0 + +TEST_DURATION = 5.0 +ACK_TIMEOUT = 1.0 + +# ========================================================== +# Packet helpers +# ========================================================== + +def calculate_crc(msg: bytes) -> int: + return (-sum(msg)) & 0xFF + + +def make_packet(command: int, data: bytes = b"") -> bytes: + pkt = bytearray() + + pkt.append(COMMAND_PREFIX) + pkt.append(len(data)) + pkt.append(DEVICE_ID) + pkt.append(command) + pkt.append(0) + + pkt.extend(data) + + pkt[4] = calculate_crc(pkt[:4] + pkt[5:]) + return bytes(pkt) + + +def verify_crc(packet: bytes) -> bool: + return packet[4] == calculate_crc(packet[:4] + packet[5:]) + + +# ========================================================== +# ACK synchronization +# ========================================================== + +ack_event = threading.Event() + +ack_count = 0 +nack_count = 0 +error_count = 0 +timeout_count = 0 + +running = True + + +# ========================================================== +# Receiver +# ========================================================== + +def reader(ser): + global ack_count + global nack_count + global error_count + + rx = bytearray() + + while running: + + data = ser.read(4096) + + if not data: + continue + + rx.extend(data) + + while True: + + if len(rx) < 5: + break + + if rx[0] != COMMAND_PREFIX: + + nl = rx.find(b"\n") + + if nl == -1: + rx.clear() + break + + try: + print("[LOG]", rx[:nl].decode().rstrip()) + except UnicodeDecodeError: + pass + + del rx[: nl + 1] + continue + + length = rx[1] + size = 5 + length + + if len(rx) < size: + break + + pkt = bytes(rx[:size]) + del rx[:size] + + if not verify_crc(pkt): + continue + + cmd = pkt[3] + + if cmd == COMMAND_ACK: + ack_count += 1 + ack_event.set() + + elif cmd == COMMAND_NACK: + nack_count += 1 + ack_event.set() + + elif cmd == COMMAND_ERROR: + error_count += 1 + ack_event.set() + + +# ========================================================== +# Benchmark +# ========================================================== + +def benchmark(ser): + + global timeout_count + + packet = make_packet(LED) + + sent = 0 + + rtt_sum = 0.0 + rtt_min = float("inf") + rtt_max = 0.0 + + start = time.perf_counter() + end = start + TEST_DURATION + + while time.perf_counter() < end: + + ack_event.clear() + + t0 = time.perf_counter() + + ser.write(packet) + + if not ack_event.wait(ACK_TIMEOUT): + timeout_count += 1 + continue + + t1 = time.perf_counter() + + rtt = t1 - t0 + + sent += 1 + rtt_sum += rtt + rtt_min = min(rtt_min, rtt) + rtt_max = max(rtt_max, rtt) + + elapsed = time.perf_counter() - start + + print() + print("=" * 60) + print(f"Elapsed : {elapsed:.3f} s") + print(f"Sent : {sent}") + print(f"ACK : {ack_count}") + print(f"NACK : {nack_count}") + print(f"ERROR : {error_count}") + print(f"Timeouts : {timeout_count}") + print(f"Packet Rate : {sent / elapsed:.1f} pkt/s") + + if sent: + print(f"Mean RTT : {1000*rtt_sum/sent:.3f} ms") + print(f"Min RTT : {1000*rtt_min:.3f} ms") + print(f"Max RTT : {1000*rtt_max:.3f} ms") + + +# ========================================================== +# Main +# ========================================================== + +def main(): + + global running + + print(f"Opening {PORT}") + + ser = serial.Serial( + PORT, + BAUDRATE, + timeout=0.01, + ) + + threading.Thread( + target=reader, + args=(ser,), + daemon=True, + ).start() + + print("Running benchmark...") + + benchmark(ser) + + running = False + + ser.close() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/scripts/speed_test3.py b/scripts/speed_test3.py new file mode 100644 index 0000000..6f33c99 --- /dev/null +++ b/scripts/speed_test3.py @@ -0,0 +1,384 @@ +#!/usr/bin/env python3 +""" +max_command_speed_test.py + +Finds the maximum stable command rate for the UART command protocol described +by `struct command_message_t`, using the LED command (sent with length=0, +i.e. no data bytes -- only the 5-byte header) as the test payload. + +Wire format (matches the firmware's ring-buffer parser): + + [prefix:1][length:1][id:1][command:1][crc:1][data:length] + +Note the firmware only ever reads `length` bytes of data off the wire (see +`usb_thread`), NOT the full 160-byte `data[]` array from the struct -- the +160 bytes only exist in RAM. So we must only transmit `length` bytes. + +CRC: sum of prefix+length+id+command+0(crc placeholder)+data[0..length-1], +mod 256, then crc = (0x100 - sum) & 0xff. This matches +`command_calculate_crc()`, which runs while msg->crc is still 0 (it's set +by the caller only *after* this function returns). + +Two test modes: + + roundtrip - send one command, wait for ACK/NACK, repeat. Measures the + sustainable rate when the host waits for each reply + (safest / simplest way to drive the device). + + flood - send a whole burst of commands back-to-back at a fixed + inter-command delay, then collect all the ACK/NACK replies + afterwards. This stresses the UART IRQ handler + ring buffer + directly and finds the max raw throughput the firmware can + absorb without dropping/desyncing bytes. + +The script sweeps delays (roundtrip) or a binary search on delay (flood) to +find the smallest stable delay (i.e. highest command rate) that still gets +a correct ACK for every command sent. + +Requires: pip install pyserial +""" + +import argparse +import struct +import sys +import time + +import serial + +# ---- Protocol constants (from firmware) ----------------------------------- +COMMAND_PREFIX = 0x69 +COMMAND_ID = 0x00 +COMMAND_DATA_SIZE = 160 + +COMMAND_ACK = 0 +COMMAND_NACK = 1 +LED = 2 + +HEADER_FMT = " int: + """Replicates command_calculate_crc(): sum over + [prefix, length, id, command, crc(=0 at calc time), data[:length]]""" + s = COMMAND_PREFIX + length + COMMAND_ID + command + 0 + s += sum(data[:length]) + return (0x100 - (s & 0xFF)) & 0xFF + + +def build_packet(command: int, data: bytes = b"") -> bytes: + length = len(data) + if length > COMMAND_DATA_SIZE - 1: + raise ValueError(f"data too long: {length} > {COMMAND_DATA_SIZE - 1}") + crc = calc_crc(length, command, data) + return struct.pack(HEADER_FMT, COMMAND_PREFIX, length, COMMAND_ID, command, crc) + data + + +def build_led_packet() -> bytes: + """LED command with no payload (length=0). Only the 5-byte header + [prefix][length=0][id][command][crc] is sent -- no data bytes at all.""" + return build_packet(LED, b"") + + +class ResponseError(Exception): + pass + + +def read_response(ser: serial.Serial, timeout: float): + """Read one full response packet (ACK/NACK) from the wire, honoring + `timeout` seconds total. Returns dict with command/length/data/crc_ok, + or None on timeout.""" + ser.timeout = timeout + deadline = time.monotonic() + timeout + + # Scan for prefix byte + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + ser.timeout = remaining + b = ser.read(1) + if not b: + return None + if b[0] == COMMAND_PREFIX: + break + + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + ser.timeout = remaining + hdr = ser.read(4) # length, id, command, crc + if len(hdr) < 4: + return None + length, resp_id, command, crc = hdr + + data = b"" + if length: + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + ser.timeout = remaining + data = ser.read(length) + if len(data) < length: + return None + + expected_crc = calc_crc(length, command, data) + return { + "id": resp_id, + "command": command, + "length": length, + "data": data, + "crc": crc, + "crc_ok": crc == expected_crc, + } + + +def send_led(ser: serial.Serial): + ser.write(build_led_packet()) + + +# ---- Test modes ------------------------------------------------------------- + +def test_roundtrip(ser, delay, n, timeout): + """Send n commands, waiting for a valid ACK after each before sending + the next (with an *additional* `delay` seconds between send and next + send, on top of whatever the round-trip itself costs). Returns + (success_count, n, elapsed_seconds), where elapsed_seconds is the + *measured* wall-clock time for the whole trial -- not derived from + `delay`. This is what "rate" should be computed from, since at + delay=0 the round-trip latency (write + firmware processing + read) + is the real limiter, not some divide-by-zero fiction.""" + ser.reset_input_buffer() + ok = 0 + t_start = time.monotonic() + for i in range(n): + send_led(ser) + resp = read_response(ser, timeout) + if resp and resp["crc_ok"] and resp["command"] == COMMAND_ACK: + ok += 1 + if delay: + time.sleep(delay) + elapsed = time.monotonic() - t_start + return ok, n, elapsed + + +def test_flood(ser, delay, n, timeout): + """Send n commands back-to-back with only `delay` seconds between + sends (no waiting for replies in between), then collect n replies + afterward. Returns (success_count, n, elapsed_seconds). + + elapsed_seconds covers the whole trial (send phase + collecting all + replies) as actually measured -- at delay=0 this still takes real, + nonzero time (write() syscalls, USB bulk transfer framing, the + firmware's IRQ handler + ring buffer + usb_thread all take time), so + this is the number that should be used to compute cmd/s, never + `1/delay`.""" + ser.reset_input_buffer() + t_start = time.monotonic() + for i in range(n): + send_led(ser) + if delay: + time.sleep(delay) + send_duration = time.monotonic() - t_start + + ok = 0 + # give it time proportional to what we sent, plus per-reply timeout + end_deadline = time.monotonic() + timeout + send_duration + for i in range(n): + remaining = end_deadline - time.monotonic() + if remaining <= 0: + break + resp = read_response(ser, remaining) + if resp and resp["crc_ok"] and resp["command"] == COMMAND_ACK: + ok += 1 + elif resp is None: + break + elapsed = time.monotonic() - t_start + return ok, n, elapsed + + +def achieved_rate(ok, elapsed): + """cmd/s actually measured, based on successful commands over real + wall-clock time. Never derived from the requested delay.""" + if elapsed <= 0: + return 0.0 + return ok / elapsed + + +# ---- Sweep / search logic ---------------------------------------------------- + +def find_max_rate(ser, mode, n, timeout, success_threshold, start_delay, min_delay, verbose): + """Binary search the smallest stable inter-command delay that still + achieves >= success_threshold success ratio. Returns + (best_delay, measured_rate_at_best_delay). + + Rate is always the *measured* cmd/s from the trial, never 1/delay -- + at delay=0 the requested delay tells you nothing about the actual + ceiling, which is set by write()/USB overhead and firmware processing + time, not by our sleep() calls.""" + test_fn = test_roundtrip if mode == "roundtrip" else test_flood + last_rate = 0.0 + + def trial(delay): + nonlocal last_rate + ok, total, elapsed = test_fn(ser, delay, n, timeout) + ratio = ok / total if total else 0.0 + rate = achieved_rate(ok, elapsed) + last_rate = rate + if verbose: + print(f" delay={delay*1000:8.3f} ms measured_rate={rate:9.1f} cmd/s " + f"success={ok}/{total} ({ratio*100:5.1f}%) elapsed={elapsed*1000:.1f} ms") + return ratio >= success_threshold + + lo, hi = min_delay, start_delay + if not trial(hi): + print(f"WARNING: even the slow starting delay ({hi*1000:.3f} ms) failed " + f"the success threshold. Try increasing --start-delay.") + return hi, last_rate + hi_rate = last_rate + + if trial(lo): + # Even the fastest requested delay was stable. That does NOT mean + # the rate is infinite -- it means we've hit the real ceiling + # (write()/USB/firmware), and last_rate is the measured number + # for it. Report that instead of pretending it's unbounded. + print(f"NOTE: even the fastest delay ({lo*1000:.3f} ms) passed, at a " + f"measured {last_rate:.1f} cmd/s. That's likely the true ceiling " + f"(write()/USB overhead + firmware processing), not an artifact " + f"of --min-delay. Run --soak to confirm it holds over a longer run.") + return lo, last_rate + + # Binary search between lo (fails) and hi (passes) for smallest passing delay + best_rate = hi_rate + for _ in range(20): + mid = (lo + hi) / 2.0 + if trial(mid): + hi = mid + best_rate = last_rate + else: + lo = mid + if hi - lo < 1e-5: # 0.01 ms resolution + break + + return hi, best_rate + + +def soak_test(ser, mode, delay, duration_s, timeout): + """Run continuously at a fixed delay for `duration_s` seconds and + report the measured sustained cmd/s and success ratio. Use this to + confirm a delay=0 (or any) result actually holds up over time, since + a short burst can pass while a longer run reveals ring-buffer + overflow or drift.""" + test_fn = test_roundtrip if mode == "roundtrip" else test_flood + t_start = time.monotonic() + total_ok = 0 + total_n = 0 + # run in chunks so we can report progress and stop at duration_s + chunk = 200 + while time.monotonic() - t_start < duration_s: + ok, n, elapsed = test_fn(ser, delay, chunk, timeout) + total_ok += ok + total_n += n + rate = achieved_rate(total_ok, time.monotonic() - t_start) + print(f" soak: {total_ok}/{total_n} ok so far, " + f"sustained rate={rate:.1f} cmd/s, " + f"t={time.monotonic() - t_start:5.1f}s / {duration_s}s") + elapsed = time.monotonic() - t_start + ratio = total_ok / total_n if total_n else 0.0 + rate = achieved_rate(total_ok, elapsed) + return total_ok, total_n, ratio, rate + + +def sweep(ser, mode, n, timeout, delays, verbose=True): + test_fn = test_roundtrip if mode == "roundtrip" else test_flood + results = [] + print(f"\n{'delay (ms)':>12} {'measured cmd/s':>16} {'success':>10} {'ratio':>8}") + print("-" * 52) + for delay in delays: + ok, total, elapsed = test_fn(ser, delay, n, timeout) + ratio = ok / total if total else 0.0 + rate = achieved_rate(ok, elapsed) # measured, never 1/delay + print(f"{delay*1000:12.3f} {rate:16.1f} {ok:>4}/{total:<5} {ratio*100:7.1f}%") + results.append((delay, ok, total, ratio, rate)) + return results + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--port", default="/dev/ttyACM0") + ap.add_argument("--baud", type=int, default=115200, + help="Baud rate (often ignored by USB CDC-ACM, but set for compatibility)") + ap.add_argument("--mode", choices=["roundtrip", "flood"], default="roundtrip", + help="roundtrip: wait for ACK after each send. " + "flood: send a burst, then collect replies (tests raw UART/ring-buffer throughput)") + ap.add_argument("-n", type=int, default=200, help="Commands per trial") + ap.add_argument("--timeout", type=float, default=0.5, help="Per-response read timeout (s)") + ap.add_argument("--success-threshold", type=float, default=1.0, + help="Required success ratio to call a rate 'stable' (0-1)") + ap.add_argument("--start-delay", type=float, default=0.02, + help="Slow starting inter-command delay in seconds for the search (known-good)") + ap.add_argument("--min-delay", type=float, default=0.0, + help="Fastest inter-command delay to try, in seconds (0 = back-to-back)") + ap.add_argument("--sweep", action="store_true", + help="Also print a full table sweeping delays geometrically " + "from --start-delay down to --min-delay-floor") + ap.add_argument("--min-delay-floor", type=float, default=0.0005, + help="Smallest delay used for --sweep table (s)") + ap.add_argument("--sweep-steps", type=int, default=12) + ap.add_argument("--soak", type=float, default=0.0, + help="After the search, run this many seconds at the best " + "delay found to confirm the rate holds up over time " + "(recommended, especially when the best delay is 0)") + args = ap.parse_args() + + print(f"Opening {args.port} @ {args.baud} baud, mode={args.mode}, n={args.n}/trial") + ser = serial.Serial(args.port, args.baud, timeout=args.timeout) + time.sleep(0.2) # let the port settle + ser.reset_input_buffer() + ser.reset_output_buffer() + + try: + if args.sweep: + delays = [] + hi, lo = args.start_delay, args.min_delay_floor + steps = max(args.sweep_steps, 1) + for i in range(steps): + frac = i / (steps - 1) if steps > 1 else 0 + # geometric interpolation from hi -> lo + d = hi * ((lo / hi) ** frac) if hi > 0 else 0 + delays.append(d) + sweep(ser, args.mode, args.n, args.timeout, delays) + print() + + print("Binary-searching for max stable command rate...") + best_delay, best_rate = find_max_rate( + ser, args.mode, args.n, args.timeout, + args.success_threshold, args.start_delay, args.min_delay, + verbose=True, + ) + + print("\n=== Result ===") + print(f"Max stable inter-command delay: {best_delay*1000:.3f} ms") + print(f"Measured rate at that delay: {best_rate:.1f} commands/sec " + f"(measured from actual elapsed time over {args.n} commands, " + f"not derived from the delay)") + print(f"(mode={args.mode}, n={args.n}, success_threshold={args.success_threshold*100:.0f}%)") + + if args.soak > 0: + print(f"\nRunning {args.soak:.0f}s soak test at delay={best_delay*1000:.3f} ms " + f"to confirm this holds up over time...") + ok, total, ratio, rate = soak_test(ser, args.mode, best_delay, args.soak, args.timeout) + print(f"\nSoak result: {ok}/{total} ok ({ratio*100:.1f}%), " + f"sustained rate={rate:.1f} cmd/s over {args.soak:.0f}s") + if ratio < args.success_threshold: + print("WARNING: the short trial passed but the soak test did NOT hold up -- " + "the real stable rate is lower than reported above. Try a slower " + "--start-delay / larger --min-delay and re-run.") + + finally: + ser.close() + + +if __name__ == "__main__": + main() \ No newline at end of file