ast10x0/harness: auto-discover UART ports across Pi fixtures

target/ast10x0/harness/evb_config.toml

@@ -1,15 +1,12 @@
 # Licensed under the Apache-2.0 license
 # SPDX-License-Identifier: Apache-2.0
 
-[gpio]
-srst_pin = 23
-fwspick_pin = 18
-
-[uart]
-serial_port = "/dev/ttyUSB0"
 baudrate = 115200
 
-[device_b]
-srst_pin = 25
+[device_a]  # master
+srst_pin    = 23
+fwspick_pin = 18
+
+[device_b]  # slave
+srst_pin    = 25
 fwspick_pin = 24
-serial_port = "/dev/ttyUSB1"

target/ast10x0/harness/pi_test_runner.py

@@ -32,6 +32,107 @@ def _gpio_set(pin: int, state: str) -> None:
     subprocess.run(["pinctrl", "set", str(pin), "op"] + state.split(), check=True)
 
 
+_DISCOVERY_TIMEOUT = 30  # seconds to wait for a reset board to answer
+_FLUSH_DURATION = 0.5  # drain ports this long while reset is held, comfortably
+# longer than the FTDI latency timer (~16ms) so every pre-reset byte lands and
+# is discarded and nothing new can arrive until reset is released
+_COLLECT_WINDOW = 1.0  # once the first bytes arrive, keep reading this long so a
+# running board has time to reveal a non-0x55 byte and be ruled out
+
+
+def _list_uart_ports() -> list[str]:
+    return sorted(str(p) for p in Path("/dev").glob("ttyUSB*"))
+
+
+def _discover_uart_port(
+    srst_pin: int, fwspick_pin: int, baudrate: int, exclude=(), verbose=False
+):
+    """Reset the board on (srst_pin, fwspick_pin) and return the /dev/ttyUSB*
+    that emits the bootloader ready signal.
+
+    The GPIO wiring to each board is fixed, but USB enumeration order differs
+    between Pi fixtures, so ttyUSB numbering can't be hardcoded. The board we
+    just reset comes up in the bootloader and emits only the 'U' (0x55) ready
+    signal; a board still running its app emits other bytes. So the target is
+    the one port whose post-flush buffer is non-empty and entirely 0x55.
+    """
+    candidates = [p for p in _list_uart_ports() if p not in exclude]
+    if not candidates:
+        print("RUNNER: no /dev/ttyUSB* ports found", file=sys.stderr)
+        return None
+
+    opened: list[tuple[str, serial.Serial]] = []
+    for dev in candidates:
+        try:
+            opened.append((dev, serial.Serial(dev, baudrate=baudrate, timeout=0.1)))
+        except serial.SerialException as e:
+            print(f"RUNNER: skipping {dev}: {e}", file=sys.stderr)
+    if not opened:
+        return None
+
+    try:
+        # Assert reset first so every board stops transmitting, then drain each
+        # port for longer than the FTDI latency timer. Held in reset, no board
+        # can emit anything, so once drained the buffers stay empty until reset
+        # is released; no pre-reset byte can leak past the flush.
+        _gpio_set(srst_pin, "dl")
+        flush_deadline = time.time() + _FLUSH_DURATION
+        while time.time() < flush_deadline:
+            for _, port in opened:
+                port.read(4096)
+        _gpio_set(fwspick_pin, "pn dh")
+        time.sleep(1)
+        _gpio_set(srst_pin, "dh")
+
+        buffers = {dev: bytearray() for dev, _ in opened}
+        deadline = time.time() + _DISCOVERY_TIMEOUT
+        collect_deadline = None
+        while time.time() < deadline:
+            for dev, port in opened:
+                data = port.read(256)
+                if data:
+                    buffers[dev].extend(data)
+            if collect_deadline is None:
+                if any(buffers.values()):
+                    collect_deadline = time.time() + _COLLECT_WINDOW
+            elif time.time() >= collect_deadline:
+                break
+
+        if verbose:
+            for dev, buf in buffers.items():
+                print(
+                    f"RUNNER: {dev} post-flush buffer ({len(buf)} bytes): "
+                    f"{bytes(buf[:32])!r}",
+                    file=sys.stderr,
+                )
+
+        # The freshly reset board emits only the bootloader 'U' (0x55); a board
+        # still running its app emits other bytes. So the target is the port
+        # whose buffer is non-empty and entirely 0x55; any non-0x55 rules it out.
+        candidates = [
+            dev for dev, buf in buffers.items() if buf and all(b == 0x55 for b in buf)
+        ]
+        if len(candidates) == 1:
+            return candidates[0]
+
+        if candidates:
+            print(
+                f"RUNNER: multiple ports look like a bootloader: {candidates}",
+                file=sys.stderr,
+            )
+        else:
+            print("RUNNER: no bootloader (U-only) port found", file=sys.stderr)
+        if not verbose:
+            print(
+                "RUNNER: re-run with --test_arg=-v to dump post-flush buffers",
+                file=sys.stderr,
+            )
+        return None
+    finally:
+        for _, port in opened:
+            port.close()
+
+
 def _sequence_to_fwspick_mode(
     srst_pin: int, fwspick_pin: int, port: serial.Serial
 ) -> None:
@@ -62,6 +163,7 @@ def _wait_for_uart_ready(port: serial.Serial, timeout: int = 30) -> bool:
 
 
 def _upload_firmware(port: serial.Serial, firmware_path: Path) -> None:
+    print(f"RUNNER: uploading firmware to {port.port}", file=sys.stderr)
     data = firmware_path.read_bytes()
     size = len(data)
     aligned = (size + 3) & ~3
@@ -166,8 +268,13 @@ def main() -> int:
         description="AST1060 EVB hardware layer: GPIO, firmware upload, UART stream"
     )
     parser.add_argument(
-        "uart_device",
-        help="Serial port device path (e.g. /dev/ttyUSB0)",
+        "--uart-device",
+        default=None,
+        help=(
+            "Serial port device path (e.g. /dev/ttyUSB0). If omitted, the port "
+            "is auto-discovered by resetting the board on --srst-pin/--fwspick-pin "
+            "and listening for the bootloader. Required with --stream-only."
+        ),
     )
     parser.add_argument(
         "firmware",
@@ -219,6 +326,12 @@ def main() -> int:
         default=None,
         help="BCM GPIO pin connected to device B FWSPICK",
     )
+    parser.add_argument(
+        "-v",
+        "--verbose-runner",
+        action="store_true",
+        help="Dump each port's post-flush buffer during UART discovery",
+    )
     args = parser.parse_args()
 
     if not args.stream_only:
@@ -231,11 +344,28 @@ def main() -> int:
     else:
         firmware_path = None
 
+    # Resolve the master UART. An explicit --uart-device is trusted as-is;
+    # otherwise reset the board and discover which /dev/ttyUSB* answers, so the
+    # harness tolerates USB enumeration order differing between Pi fixtures.
+    if args.stream_only:
+        if not args.uart_device:
+            parser.error("--uart-device is required with --stream-only")
+    elif not args.uart_device:
+        args.uart_device = _discover_uart_port(
+            args.srst_pin,
+            args.fwspick_pin,
+            args.baudrate,
+            verbose=args.verbose_runner,
+        )
+        if args.uart_device is None:
+            print("RUNNER: could not discover master UART", file=sys.stderr)
+            return 1
+        print(f"RUNNER: discovered master UART at {args.uart_device}", file=sys.stderr)
+
     if args.slave_firmware:
         missing = [
             name
             for name, val in [
-                ("--slave-uart-device", args.slave_uart_device),
                 ("--slave-srst-pin", args.slave_srst_pin),
                 ("--slave-fwspick-pin", args.slave_fwspick_pin),
             ]
@@ -250,6 +380,21 @@ def main() -> int:
                 file=sys.stderr,
             )
             return 1
+        if not args.slave_uart_device:
+            args.slave_uart_device = _discover_uart_port(
+                args.slave_srst_pin,
+                args.slave_fwspick_pin,
+                args.baudrate,
+                exclude=(args.uart_device,),
+                verbose=args.verbose_runner,
+            )
+            if args.slave_uart_device is None:
+                print("Error: could not discover slave UART", file=sys.stderr)
+                return 1
+            print(
+                f"RUNNER: discovered slave UART at {args.slave_uart_device}",
+                file=sys.stderr,
+            )
         return 0 if _run_paired(args, firmware_path, slave_firmware_path) else 1
 
     try:

target/ast10x0/harness/test_runner.py

@@ -269,9 +269,8 @@ def _run_remote(
 ) -> bool:
     """SCP firmware and runner to Pi; stream UART back over SSH stdout."""
     host = args.pi_host
-    gpio = config["gpio"]
-    uart = config["uart"]
-    baudrate = args.baudrate if args.baudrate else uart["baudrate"]
+    dev_a = config["device_a"]
+    baudrate = args.baudrate if args.baudrate else config["baudrate"]
 
     remote_dir = "/tmp/ast1060_test"
 
@@ -311,21 +310,24 @@ def _run_remote(
             check=True,
         )
 
-    remote_cmd = f"python3 -u {remote_dir}/pi_test_runner.py {uart_device}"
+    remote_cmd = f"python3 -u {remote_dir}/pi_test_runner.py"
     if not args.parse_only:
         remote_cmd += f" {remote_fw}"
     remote_cmd += (
-        f" --srst-pin {gpio['srst_pin']}"
-        f" --fwspick-pin {gpio['fwspick_pin']}"
+        f" --srst-pin {dev_a['srst_pin']}"
+        f" --fwspick-pin {dev_a['fwspick_pin']}"
         f" --baudrate {baudrate}"
     )
+    if uart_device:
+        remote_cmd += f" --uart-device {uart_device}"
+    if args.verbose_runner:
+        remote_cmd += " --verbose-runner"
     if args.parse_only:
         remote_cmd += " --stream-only"
     if remote_slave_fw:
         device_b = config["device_b"]
         remote_cmd += (
             f" --slave-firmware {remote_slave_fw}"
-            f" --slave-uart-device {device_b['serial_port']}"
             f" --slave-srst-pin {device_b['srst_pin']}"
             f" --slave-fwspick-pin {device_b['fwspick_pin']}"
         )
@@ -375,7 +377,7 @@ def main() -> int:
         "--baudrate",
         type=int,
         default=None,
-        help=f"Override baud rate from evb_config.toml (default: {config['uart']['baudrate']})",
+        help=f"Override baud rate from evb_config.toml (default: {config['baudrate']})",
     )
     parser.add_argument(
         "--log-file",
@@ -392,6 +394,15 @@ def main() -> int:
         action="store_true",
         help="Skip GPIO and firmware upload; stream and detokenize UART output only",
     )
+    parser.add_argument(
+        "-v",
+        "--verbose-runner",
+        action="store_true",
+        help=(
+            "Dump each port's post-flush buffer during UART discovery. Pass "
+            "through Bazel with --test_arg=-v"
+        ),
+    )
     args, remaining = parser.parse_known_args()
 
     if os.environ.get("PW_RUNNER_PASSTHROUGH") == "1":
@@ -400,11 +411,20 @@ def main() -> int:
         res = subprocess.run([args.firmware] + remaining)
         return res.returncode
 
-    uart_device = os.environ.get("UART_DEVICE") or config["uart"]["serial_port"]
+    # Normally the Pi auto-discovers the UART by resetting the board, so no
+    # device is configured. UART_DEVICE lets a caller pin an explicit port;
+    # --parse-only can't reset, so it requires one.
+    uart_device = os.environ.get("UART_DEVICE")
 
     if not args.firmware:
         parser.error("firmware is required")
 
+    if args.parse_only and not uart_device:
+        parser.error(
+            "--parse-only requires UART_DEVICE to be set "
+            "(no reset happens, so the port can't be auto-discovered)"
+        )
+
     # system_image emits both .elf and .bin under the same stem; accept either.
     # When invoked via --run_under on a system_image_test, Bazel passes the
     # no-suffix symlink (e.g. threads_test → threads.elf); resolve it first.