diff --git a/target/ast10x0/harness/evb_config.toml b/target/ast10x0/harness/evb_config.toml index c169c604..3c09ee79 100644 --- a/target/ast10x0/harness/evb_config.toml +++ b/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" diff --git a/target/ast10x0/harness/pi_test_runner.py b/target/ast10x0/harness/pi_test_runner.py index 9f418f9f..bb382bae 100644 --- a/target/ast10x0/harness/pi_test_runner.py +++ b/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: diff --git a/target/ast10x0/harness/test_runner.py b/target/ast10x0/harness/test_runner.py index cba88d74..2dbad47b 100644 --- a/target/ast10x0/harness/test_runner.py +++ b/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.