refactor(cli): ota 预检免设备连接,132 测试全绿
This commit is contained in:
1 parent
6615696c2d
commit
146ba140ed
1 file changed
+19
-12
+19
-12
@@ -248,26 +248,31 @@ async def _run_command(args, t):
|
|||||||
await t.write_rxtx(data, response=not args.no_response)
|
await t.write_rxtx(data, response=not args.no_response)
|
||||||
return {"sent": len(data)}
|
return {"sent": len(data)}
|
||||||
if c == "ota":
|
if c == "ota":
|
||||||
import zlib
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
image = Path(args.file).read_bytes()
|
image = Path(args.file).read_bytes()
|
||||||
# 独立预检(pQ 头 + CRC32(body)),与 tools/fw.py 同规则
|
_ota_precheck(args) # 不合格直接抛 ValueError
|
||||||
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 执行烧录"}
|
|
||||||
progress_log = []
|
progress_log = []
|
||||||
result = await t.run_ota(image, on_progress=progress_log.append)
|
result = await t.run_ota(image, on_progress=progress_log.append)
|
||||||
return {"ota": "done", **result, "blocks_ok": len(progress_log)}
|
return {"ota": "done", **result, "blocks_ok": len(progress_log)}
|
||||||
raise ValueError(f"未知命令 {c}")
|
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:
|
async def _run_batch(args) -> list:
|
||||||
lines = (sys.stdin.read() if args.file == "-"
|
lines = (sys.stdin.read() if args.file == "-"
|
||||||
else open(args.file, encoding="utf-8").read())
|
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)}
|
return {"devices": await scan(timeout=args.timeout)}
|
||||||
if args.cmd == "batch":
|
if args.cmd == "batch":
|
||||||
return {"results": await _run_batch(args)}
|
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))
|
return await _run_with_transport(args, lambda t: _run_command(args, t))
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user