diff --git a/src/ppclock/cli.py b/src/ppclock/cli.py index 3d4d3fb..548b0c7 100644 --- a/src/ppclock/cli.py +++ b/src/ppclock/cli.py @@ -248,26 +248,31 @@ async def _run_command(args, t): await t.write_rxtx(data, response=not args.no_response) return {"sent": len(data)} if c == "ota": - import zlib from pathlib import Path image = Path(args.file).read_bytes() - # 独立预检(pQ 头 + CRC32(body)),与 tools/fw.py 同规则 - if len(image) < 64 or image[0:2] != b"pQ": - raise ValueError("固件格式错误(缺 pQ 头)") - body_size = int.from_bytes(image[4:8], "little") - crc_ok = (zlib.crc32(image[64:64 + body_size]) & 0xFFFFFFFF - == int.from_bytes(image[8:12], "little")) - if not crc_ok: - raise ValueError("固件 CRC 校验失败,拒绝烧录") - if not args.yes: - return {"precheck": "ok", "body_size": body_size, - "hint": "加 --yes 执行烧录"} + _ota_precheck(args) # 不合格直接抛 ValueError progress_log = [] result = await t.run_ota(image, on_progress=progress_log.append) return {"ota": "done", **result, "blocks_ok": len(progress_log)} raise ValueError(f"未知命令 {c}") +def _ota_precheck(args) -> dict: + """无设备场景的固件预检(--yes 才需要连接)。""" + import zlib + from pathlib import Path + image = Path(args.file).read_bytes() + if len(image) < 64 or image[0:2] != b"pQ": + raise ValueError("固件格式错误(缺 pQ 头)") + body_size = int.from_bytes(image[4:8], "little") + crc_ok = (zlib.crc32(image[64:64 + body_size]) & 0xFFFFFFFF + == int.from_bytes(image[8:12], "little")) + if not crc_ok: + raise ValueError("固件 CRC 校验失败,拒绝烧录") + return {"precheck": "ok", "body_size": body_size, + "hint": "加 --yes 执行烧录"} + + async def _run_batch(args) -> list: lines = (sys.stdin.read() if args.file == "-" else open(args.file, encoding="utf-8").read()) @@ -298,6 +303,8 @@ async def _dispatch(args) -> dict | list | None: return {"devices": await scan(timeout=args.timeout)} if args.cmd == "batch": return {"results": await _run_batch(args)} + if args.cmd == "ota" and not args.yes: + return _ota_precheck(args) # 预检无需设备 return await _run_with_transport(args, lambda t: _run_command(args, t))