From 146ba140ed60a938da4cc3d18c7e419091a0b526 Mon Sep 17 00:00:00 2001 From: agent Date: Fri, 24 Jul 2026 11:55:37 +0000 Subject: [PATCH] =?UTF-8?q?refactor(cli):=20ota=20=E9=A2=84=E6=A3=80?= =?UTF-8?q?=E5=85=8D=E8=AE=BE=E5=A4=87=E8=BF=9E=E6=8E=A5=EF=BC=8C132=20?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E5=85=A8=E7=BB=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/ppclock/cli.py | 31 +++++++++++++++++++------------ 1 file changed, 19 insertions(+), 12 deletions(-) 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))