util: FPGA runner: rework to use asyncio

util/clang_tidy.py

@@ -27,7 +27,7 @@ def main():
         )
 
     cmd = "clang-tidy"
-    cmd_args = [cmd, "-p", BUILD_DIR, *sys.argv[1:], *c_files]
+    cmd_args = [cmd, "-p", BUILD_DIR, "--header-filter=.*", *sys.argv[1:], *c_files]
     os.execvp(cmd, cmd_args)
 
 

util/fpga_runner.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import argparse
-import subprocess
+import asyncio
 import sys
 import time
 from pathlib import Path
@@ -16,49 +16,44 @@
 BAUD_RATE: int = 1_000_000
 TIMEOUT: int = 60
 
+RUNNER: str = Path(__file__).name
 
-def load_fpga_test(test: Path, timeout: int = TIMEOUT) -> None:
+
+async def load_fpga_test(test: Path) -> None:
     command = ["openFPGALoader", "--spi", "--offset", str(BOOT_ROM_OFFSET), "--write-flash"]
-    command.append(test.with_suffix(".bin"))
+    command.append(str(test.with_suffix(".bin")))
+    # TODO: This is a workaround to send a reset and start the test, should be removed when we
+    # are able to reset the SoC with the external reset.
+    # The first invocation resets and load the binary, the second resets and the load is
+    # ignored by the bootROM, thus we don't check the return error.
+    p = await asyncio.create_subprocess_exec(*command)
+    if await p.wait() != 0:
+        print(f"[{RUNNER}] SPI load command exited with non-zero exit code {p.returncode}")
+        sys.exit(1)
+    p = await asyncio.create_subprocess_exec(*command)
+    await p.wait()
+
+
+async def run_fpga_test(tty: str, test: Path) -> bool:
+    with serial.Serial(tty, BAUD_RATE, timeout=0) as uart:
+        load = asyncio.create_task(load_fpga_test(test))
+        poll = asyncio.create_task(poll_uart(uart))
+        success = await poll
+        await load
+        return success
+
+
+async def poll_uart(uart: serial.Serial) -> bool:
     start = time.time()
-    while time.time() - start < timeout:
-        try:
-            subprocess.run(command, capture_output=False, check=False)
-        except OSError as e:
-            print(f"[{Path(__file__).name}] Error: {e.strerror}")
-            sys.exit(1)
-
-        # TODO: This is a workaround to send a reset and start the test, should be removed when we
-        # are able to reset the SoC with the external reset.
-        # The first invocation resets and load the binary, the second resets and the load is
-        # ignored by the bootROM, thus we don't check the return error.
-        subprocess.run(command, capture_output=True, check=False)
-        return
-
-    print(f"[{Path(__file__).name}] Load FPGA test timeout")
-    sys.exit(1)
-
-
-def run_fpga_test(tty: str, test: Path, timeout: int = 10) -> int:
-    print(f"Listening to {tty}")
-    with serial.Serial(tty, BAUD_RATE, timeout=1) as uart:
-        start = time.time()
-        load_fpga_test(test)
-        while time.time() - start < timeout:
-            line = uart.readline().decode("utf-8", errors="ignore")
-            print(line, end="")
-            if not line or "TEST RESULT" not in line:
-                continue
-
-            if "PASSED" in line:
-                return 0
-
-            if "FAILED" not in line:
-                print(f"[{Path(__file__).name}] Unknown test result: {line}")
-
-            return 1
-        print(f"[{Path(__file__).name}] Test timeout")
-        return 1
+    while time.time() - start < TIMEOUT:
+        line = await asyncio.to_thread(uart.readline)
+        line = line.decode("utf-8", errors="ignore")
+        print(line, end="")
+        if not line or "TEST RESULT" not in line:
+            continue
+        return "PASSED" in line
+    print(f"[{RUNNER}] Test timeout")
+    return False
 
 
 def find_uart(vid: int = 0x0403, pid: int = 0x6001) -> str | None:
@@ -74,9 +69,13 @@ def main() -> None:
     args = parser.parse_args()
 
     if uart_tty := find_uart():
-        sys.exit(run_fpga_test(uart_tty, args.test, TIMEOUT))
+        try:
+            success = asyncio.run(run_fpga_test(uart_tty, args.test))
+            sys.exit(0 if success else 1)
+        except KeyboardInterrupt:
+            sys.exit(1)
 
-    print(f"[{Path(__file__).name}] Error: UART device not found")
+    print(f"[{RUNNER}] Error: UART device not found")
     sys.exit(1)