modernize script

Co-authored-by: Copilot <[email protected]>
Signed-off-by: Zixun LI <[email protected]>

1 files changed, 100 insertions, 51 deletions

diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py
index 4d975f6c3..b8d6362bb 100755
--- a/test/hil/hil_test.py
+++ b/test/hil/hil_test.py
@@ -34,6 +34,8 @@ import sys
 import time
 import warnings
 import signal
+from pathlib import Path
+from typing import Any, TypedDict, NotRequired, cast
 
 # Suppress pkg_resources deprecation warning from fs module
 warnings.filterwarnings("ignore", message="pkg_resources is deprecated")
@@ -64,11 +66,51 @@ board_test = {}
 build_dir = 'cmake-build'
 skip_flash = False
 
+class FlasherCfg(TypedDict):
+    name: str
+    uid: str
+    args: str
+
+
+class AttachedDevCfg(TypedDict, total=False):
+    vid_pid: str
+    serial: str
+    is_cdc: bool
+    is_msc: bool
+    block_count: int
+    block_size: int
+
+
+class TestsCfg(TypedDict, total=False):
+    device: bool
+    dual: bool
+    host: bool
+    only: list[str]
+    skip: list[str]
+    dev_attached: list[AttachedDevCfg]
+
+
+class BuildCfg(TypedDict, total=False):
+    flags_on: list[str]
+    args: list[str]
+
+
+class Board(TypedDict):
+    name: str
+    uid: str
+    tests: TestsCfg
+    flasher: FlasherCfg
+    build: NotRequired[BuildCfg]
+
+
+class HilConfig(TypedDict):
+    boards: list[Board]
+
 CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))
 POOL_TIMEOUT = int(os.getenv('HIL_POOL_TIMEOUT', '3000'))
 
 
-def cmd_stdout_text(out):
+def cmd_stdout_text(out: Any) -> str:
     if out is None:
         return ''
     if isinstance(out, bytes):
@@ -103,8 +145,8 @@ issue at github.com/hathach/tinyusb"
 # -------------------------------------------------------------
 # Path
 # -------------------------------------------------------------
-OPENCOD_ADI_PATH = f'{os.getenv("HOME")}/app/openocd_adi'
-TINYUSB_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi'
+TINYUSB_ROOT = Path(__file__).resolve().parents[2]
 
 # get usb serial by id
 def get_serial_dev(id, vendor_str, product_str, ifnum):
@@ -131,7 +173,7 @@ def get_hid_dev(id, vendor_str, product_str, event):
     return f'/dev/input/by-id/usb-{vendor_str}_{product_str}_{id}-{event}'
 
 
-def open_serial_dev(port):
+def open_serial_dev(port: str):
     timeout = ENUM_TIMEOUT
     ser = None
     while timeout > 0:
@@ -146,10 +188,11 @@ def open_serial_dev(port):
         timeout -= 0.1
 
     assert timeout > 0, f'Cannot open port f{port}' if os.path.exists(port) else f'Port {port} not existed'
+    assert ser is not None
     return ser
 
 
-def read_disk_file(uid, lun, fname):
+def read_disk_file(uid: str, lun: int, fname: str) -> bytes:
     # open_fs("fat://{dev}) require 'pip install pyfatfs'
     dev = get_disk_dev(uid, 'TinyUSB', lun)
     timeout = ENUM_TIMEOUT
@@ -166,8 +209,7 @@ def read_disk_file(uid, lun, fname):
         time.sleep(1)
         timeout -= 1
 
-    assert timeout > 0, f'Storage {dev} not existed'
-    return None
+    raise AssertionError(f'Storage {dev} not existed')
 
 
 def open_mtp_dev(uid):
@@ -189,7 +231,7 @@ def open_mtp_dev(uid):
     return None
 
 
-def get_printer_dev(id, vendor_str, product_str, ifnum):
+def get_printer_dev(id: str, vendor_str, product_str, ifnum: int):
     """Find /dev/usb/lpX by matching USB serial, vendor, product, and interface number via sysfs"""
     vendor_str = vendor_str.replace(' ', '_') if vendor_str else ''
     product_str = product_str.replace(' ', '_') if product_str else ''
@@ -203,7 +245,7 @@ def get_printer_dev(id, vendor_str, product_str, ifnum):
     return None
 
 
-def open_printer_dev(id, vendor_str, product_str, ifnum):
+def open_printer_dev(id: str, vendor_str, product_str, ifnum: int) -> str:
     """Wait for printer device to enumerate and return its path"""
     timeout = ENUM_TIMEOUT
     while timeout > 0:
@@ -218,12 +260,15 @@ def open_printer_dev(id, vendor_str, product_str, ifnum):
 # -------------------------------------------------------------
 # Flashing firmware
 # -------------------------------------------------------------
-def run_cmd(cmd, cwd=None, timeout=CMD_TIMEOUT):
+def run_cmd(cmd: str, cwd: str | None = None, timeout: int = CMD_TIMEOUT) -> subprocess.CompletedProcess:
     popen_kwargs = {
         'cwd': cwd,
         'shell': True,
         'stdout': subprocess.PIPE,
         'stderr': subprocess.STDOUT,
+        'text': True,
+        'encoding': 'utf-8',
+        'errors': 'replace',
     }
     if os.name != 'nt':
         popen_kwargs['preexec_fn'] = os.setsid
@@ -269,23 +314,23 @@ def run_cmd(cmd, cwd=None, timeout=CMD_TIMEOUT):
     return r
 
 
-def flash_jlink(board, firmware):
+def flash_jlink(board: Board, firmware: str) -> subprocess.CompletedProcess:
     flasher = board['flasher']
     script = ['halt', 'r', f'loadfile {firmware}.elf', 'r', 'go', 'exit']
-    f_jlink = f'{board["name"]}_{os.path.basename(firmware)}.jlink'
-    with open(f_jlink, 'w') as f:
+    f_jlink = Path(f'{board["name"]}_{Path(firmware).name}.jlink')
+    with f_jlink.open('w') as f:
         f.writelines(f'{s}\n' for s in script)
     ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
-    os.remove(f_jlink)
+    f_jlink.unlink(missing_ok=True)
     return ret
 
 
-def reset_jlink(board):
+def reset_jlink(board: Board) -> subprocess.CompletedProcess:
     flasher = board['flasher']
     script = ['halt', 'r', 'go', 'exit']
-    f_jlink = f'{board["name"]}_reset.jlink'
-    if not os.path.exists(f_jlink):
-        with open(f_jlink, 'w') as f:
+    f_jlink = Path(f'{board["name"]}_reset.jlink')
+    if not f_jlink.exists():
+        with f_jlink.open('w') as f:
             f.writelines(f'{s}\n' for s in script)
     ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
     return ret
@@ -348,16 +393,20 @@ def reset_openocd_wch(board):
     return ret
 
 
-def flash_openocd_adi(board, firmware):
+def flash_openocd_adi(board: Board, firmware: str) -> subprocess.CompletedProcess:
     flasher = board['flasher']
-    ret = run_cmd(f'{OPENCOD_ADI_PATH}/src/openocd -c "adapter serial {flasher["uid"]}" -s {OPENCOD_ADI_PATH}/tcl '
+    openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
+    tcl_dir = OPENCOD_ADI_PATH / 'tcl'
+    ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
                   f'{flasher["args"]} -c "program {firmware}.elf reset exit"')
     return ret
 
 
-def reset_openocd_adi(board):
+def reset_openocd_adi(board: Board) -> subprocess.CompletedProcess:
     flasher = board['flasher']
-    ret = run_cmd(f'{OPENCOD_ADI_PATH}/src/openocd -c "adapter serial {flasher["uid"]}" -s {OPENCOD_ADI_PATH}/tcl '
+    openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
+    tcl_dir = OPENCOD_ADI_PATH / 'tcl'
+    ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
                   f'{flasher["args"]} -c "program reset exit"')
     return ret
 
@@ -376,17 +425,17 @@ def reset_wlink_rs(board):
     return ret
 
 
-def flash_esptool(board, firmware):
+def flash_esptool(board: Board, firmware: str) -> subprocess.CompletedProcess:
     flasher = board['flasher']
     port = get_serial_dev(flasher["uid"], None, None, 0)
-    fw_dir = os.path.dirname(f'{firmware}.bin')
-    with open(f'{fw_dir}/config.env') as f:
+    fw_dir = Path(f'{firmware}.bin').parent
+    with (fw_dir / 'config.env').open() as f:
         idf_target = json.load(f)['IDF_TARGET']
-    with open(f'{fw_dir}/flash_args') as f:
+    with (fw_dir / 'flash_args').open() as f:
         flash_args = f.read().strip().replace('\n', ' ')
     command = (f'esptool --chip {idf_target} -p {port} {flasher["args"]} '
                f'--before=default_reset --after=hard_reset write_flash {flash_args}')
-    ret = run_cmd(command, cwd=fw_dir)
+    ret = run_cmd(command, cwd=str(fw_dir))
     return ret
 
 
@@ -729,7 +778,7 @@ def test_device_cdc_dual_ports(board):
 
     sizes = [32, 64, 128, 256, 512, random.randint(2000, 5000)]
 
-    def write_and_check(writer, payload):
+    def write_and_check(writer, payload : bytes):
         payload_len = len(payload)
         for s in ser:
             s.reset_input_buffer()
@@ -1298,7 +1347,7 @@ host_test = [
 ]
 
 
-def test_example(board, f1, example):
+def test_example(board: Board, f1: str, example: str) -> int:
     """
     Test example firmware
     :param board: board dict
@@ -1313,11 +1362,11 @@ def test_example(board, f1, example):
     if f1 != "":
         f1_str = '-f1_' + f1.replace(' ', '_')
 
-    fw_dir = f'{TINYUSB_ROOT}/{build_dir}/cmake-build-{name}{f1_str}/{example}'
-    fw_name = f'{fw_dir}/{os.path.basename(example)}'
+    fw_dir = TINYUSB_ROOT / build_dir / f'cmake-build-{name}{f1_str}' / example
+    fw_name = fw_dir / Path(example).name
     print(f'{name+f1_str:40} {example:30} ...', end='')
 
-    if not os.path.exists(fw_dir) or not (os.path.exists(f'{fw_name}.elf') or os.path.exists(f'{fw_name}.bin')):
+    if not fw_dir.exists() or not ((fw_name.with_suffix('.elf')).exists() or (fw_name.with_suffix('.bin')).exists()):
         print('Skip (no binary)')
         return 0
 
@@ -1330,7 +1379,7 @@ def test_example(board, f1, example):
     flash_ok = True
     for i in range(max_retry):
         if not skip_flash:
-            ret = globals()[f'flash_{board["flasher"]["name"].lower()}'](board, fw_name)
+            ret = globals()[f'flash_{board["flasher"]["name"].lower()}'](board, str(fw_name))
             flash_ok = (ret.returncode == 0)
         if flash_ok:
             try:
@@ -1360,18 +1409,18 @@ def test_example(board, f1, example):
     return err_count
 
 
-def build_board(board):
+def build_board(board: Board) -> tuple[str, int]:
     """Build firmware for this board via tools/build.py.
     Honors board config's build.flags_on variants and build.args defines.
     Output goes to cmake-build/cmake-build-BOARD[-f1_...]/ (tools/build.py layout)."""
     name = board['name']
-    bcfg = board.get('build', {})
+    bcfg = cast(BuildCfg, board.get('build', {}))
     flags_on_list = bcfg.get('flags_on', [''])
     extra_defs = bcfg.get('args', [])
 
     failed = 0
     for f1 in flags_on_list:
-        cmd = [sys.executable, f'{TINYUSB_ROOT}/tools/build.py', '-b', name]
+        cmd = [sys.executable, str(TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name]
         for d in extra_defs:
             cmd += ['-D', d]
         if f1:
@@ -1386,7 +1435,7 @@ def build_board(board):
     return name, failed
 
 
-def test_board(board):
+def test_board(board: Board) -> tuple[str, int, list[str]]:
     name = board['name']
     flasher = board['flasher']
 
@@ -1400,11 +1449,11 @@ def test_board(board):
     else:
         if 'tests' in board:
             board_tests = board['tests']
-            if 'device' in board_tests and board_tests['device'] == True:
+            if board_tests.get('device') is True:
                 test_list += list(device_tests)
-            if 'dual' in board_tests and board_tests['dual'] == True:
+            if board_tests.get('dual') is True:
                 test_list += dual_tests
-            if 'host' in board_tests and board_tests['host'] == True:
+            if board_tests.get('host') is True:
                 test_list += host_test
             if 'only' in board_tests:
                 test_list = board_tests['only']
@@ -1434,7 +1483,7 @@ def test_board(board):
     return name, err_count, sorted(set(failed_tests))
 
 
-def main():
+def main() -> None:
     """
     Hardware test on specified boards
     """
@@ -1461,7 +1510,7 @@ def main():
     parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')
     args = parser.parse_args()
 
-    config_file = args.config_file
+    config_file = Path(args.config_file)
     boards = args.board
     skip_boards = args.skip_board
     verbose = args.verbose
@@ -1476,10 +1525,10 @@ def main():
     skip_flash = args.skip_flash
 
     # if config file is not found, try to find it in the same directory as this script
-    if not os.path.exists(config_file):
-        config_file = os.path.join(os.path.dirname(__file__), config_file)
-    with open(config_file) as f:
-        config = json.load(f)
+    if not config_file.exists():
+        config_file = Path(__file__).resolve().parent / config_file
+    with config_file.open() as f:
+        config = cast(HilConfig, json.load(f))
 
     if len(boards) == 0:
         config_boards = [e for e in config['boards'] if e['name'] not in skip_boards]
@@ -1501,7 +1550,7 @@ def main():
         print(f'Build phase done: {build_err} failed')
         print('-' * 30)
 
-    with Pool(processes=os.cpu_count()) as pool:
+    with Pool(processes=os.cpu_count() or 1) as pool:
         async_ret = pool.map_async(test_board, config_boards)
         try:
             mret = async_ret.get(timeout=POOL_TIMEOUT)
@@ -1512,15 +1561,15 @@ def main():
         err_count = build_err + sum(e[1] for e in mret)
         # generate skip list for next re-run if failed: skip boards that fully passed,
         # and emit -bt BOARD:t1,t2 so each failed board only re-runs its own failed tests.
-        skip_fname = f'{config_file}.skip'
+        skip_fname = config_file.with_suffix(config_file.suffix + '.skip')
         if err_count > 0:
             skip_boards += [name for name, err, _ in mret if err == 0]
             parts = [f'--skip-board {i}' for i in skip_boards]
             parts += [f'-bt {name}:{",".join(fts)}' for name, err, fts in mret if err > 0 and fts]
-            with open(skip_fname, 'w') as f:
+            with skip_fname.open('w') as f:
                 f.write(' '.join(parts))
-        elif os.path.exists(skip_fname):
-            os.remove(skip_fname)
+        elif skip_fname.exists():
+            skip_fname.unlink()
 
     duration = time.time() - duration
     print()