From 6c1f276002cd4e1d9ebd47f1d5f69dab08cc6e85 Mon Sep 17 00:00:00 2001 From: wangfq Date: Thu, 20 Aug 2026 11:53:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(DBNMQTTool):=20OTA=20=E9=A1=B5=E7=AD=BE=20?= =?UTF-8?q?+=20=E5=8D=8F=E8=AE=AE=E5=8D=95=E6=B5=8B=20(MQTT=20V1.08=20?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E5=85=88=E8=A1=8C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - protocol.py: OTA 模块 — 命令常量/ota_crc32(ISO-HDLC)/ota_split_bin(256B×96KB)/ data_ota_* 构建器/状态·阶段·错误码描述 - main.py: OTA 页签 — bin 选择(≤96KB+CRC32 显示)/版本·force·slot 参数/ 下载(OtaDownloadThread: begin→data×N→end, 断点续传·缺片重定位·CRC 重发)/ 刷写(二次确认)/中止/状态查询; 进度条 + ota_report/ota_status 实时显示 - tests/test_ota_protocol.py: 22 断言 — CRC32 标准向量/分片边界/构建器/ OtaDeviceSim 设备状态机(顺序·幂等·乱序·单片CRC·全镜像复核·续传)/完整下载流 - devlog 置顶条目 --- DBNMQTTool/dbn_mqtt_tool/protocol.py | 89 +++++++ DBNMQTTool/docs/devlog.md | 25 ++ DBNMQTTool/main.py | 326 +++++++++++++++++++++++++- DBNMQTTool/tests/test_ota_protocol.py | 316 +++++++++++++++++++++++++ 4 files changed, 755 insertions(+), 1 deletion(-) create mode 100644 DBNMQTTool/tests/test_ota_protocol.py diff --git a/DBNMQTTool/dbn_mqtt_tool/protocol.py b/DBNMQTTool/dbn_mqtt_tool/protocol.py index 1b5b467..fa57202 100644 --- a/DBNMQTTool/dbn_mqtt_tool/protocol.py +++ b/DBNMQTTool/dbn_mqtt_tool/protocol.py @@ -9,6 +9,7 @@ DLD960 IoT MQTT 协议定义 import json import time import struct +import zlib from typing import Optional, Any from dataclasses import dataclass, field, asdict @@ -273,6 +274,94 @@ def data_log_clear(stream: str = STREAM_EVENT) -> dict: return {} +# ============================================================ +# OTA 远程升级 (V1.08: Loop MCU 先存后刷, ROADMAP P1.4 ①) +# 协议依据《DLD960_IoT_MQTT协议.md》§4.19~4.24 / §5.5 +# ============================================================ + +CMD_OTA_BEGIN = "ota_begin" +CMD_OTA_DATA = "ota_data" +CMD_OTA_END = "ota_end" +CMD_OTA_ABORT = "ota_abort" +CMD_OTA_FLASH = "ota_flash" +CMD_OTA_STATUS = "ota_status" +CMD_OTA_REPORT = "ota_report" # dev→srv 主动上报 + +OTA_TARGET_LOOP = "loop" +OTA_CHUNK_SIZE = 256 # 单片 256B（协议常量, 不接受协商） +OTA_MAX_SIZE = 96 * 1024 # 镜像上限 96KB (Slot 100KB - 4KB 边界余量) +OTA_SLOT_A = "a" +OTA_SLOT_B = "b" + +OTA_STATE_DESC = { + "idle": "空闲", + "downloading": "下载中", + "ready": "校验通过(可刷写)", + "flashing": "刷写中", + "flash_failed": "刷写失败", + "aborted": "已中止", +} + +OTA_STAGE_DESC = { + "begin": "会话开启", + "downloading": "分片落盘", + "ready": "校验通过", + "flashing": "刷写中", + "done": "刷写成功(Loop 已重启)", + "failed": "刷写失败", +} + +OTA_ERROR_DESC = { + 0x1001: "启动帧无响应", + 0x1002: "地址帧错误", + 0x1003: "数据块 ACK 超限", + 0x1004: "全镜像校验失败", + 0x1005: "安全窗口拒绝后强制失败", +} + + +def ota_crc32(data: bytes) -> int: + """CRC-32/ISO-HDLC (协议 §4.19.1; zlib.crc32 即此算法, 设备查表法一致)""" + return zlib.crc32(data) & 0xFFFFFFFF + + +def ota_split_bin(data: bytes, chunk_size: int = OTA_CHUNK_SIZE) -> list[tuple[int, int, str]]: + """bin → 分片列表 [(offset, crc32, hex_str), ...] (单片 256B, hex 512 字符)""" + if len(data) > OTA_MAX_SIZE: + raise ValueError(f"镜像 {len(data)}B 超上限 {OTA_MAX_SIZE}B (96KB)") + return [(off, ota_crc32(data[off:off + chunk_size]), data[off:off + chunk_size].hex()) + for off in range(0, len(data), chunk_size)] + + +def data_ota_begin(size: int, crc32: int, version: str = "", + target: str = OTA_TARGET_LOOP, force: bool = False) -> dict: + """ota_begin 请求 data (开启会话 / 断点续传定位)""" + return {"target": target, "size": size, "crc32": crc32, + "version": version, "force": force} + + +def data_ota_data(offset: int, crc32: int, data_hex: str, + target: str = OTA_TARGET_LOOP) -> dict: + """ota_data 请求 data (分片下发, offset 256 对齐)""" + return {"target": target, "offset": offset, "crc32": crc32, "data": data_hex} + + +def data_ota_end(crc32: int, target: str = OTA_TARGET_LOOP) -> dict: + """ota_end 请求 data (结束下载, 全镜像 CRC32 复核)""" + return {"target": target, "crc32": crc32} + + +def data_ota_abort(target: str = OTA_TARGET_LOOP) -> dict: + """ota_abort 请求 data (中止会话, 释放暂存)""" + return {"target": target} + + +def data_ota_flash(slot: str = OTA_SLOT_A, force: bool = False, + target: str = OTA_TARGET_LOOP) -> dict: + """ota_flash 请求 data (触发本地 ISP 刷写, 仅 ready 态)""" + return {"target": target, "slot": slot, "force": force} + + # ============================================================ # 解析设备上报 # ============================================================ diff --git a/DBNMQTTool/docs/devlog.md b/DBNMQTTool/docs/devlog.md index ebef6a1..9c81cf2 100644 --- a/DBNMQTTool/docs/devlog.md +++ b/DBNMQTTool/docs/devlog.md @@ -6,6 +6,31 @@ --- +## 2026-08-20 — OTA 页签：Loop 远程 OTA 分片下载（协议 V1.08 工具先行） + +### 背景 + +ROADMAP P1.4 ①：Loop MCU (AT32F421) 远程 OTA，先存后刷。协议设计稿 V1.01 已定（`docs/DLD960_MQTT_OTA协议.md`），并入 MQTT 协议 V1.08。按"工具先行，固件后到"惯例，先在 DBNMQTTool 实现 OTA 页签 + 协议单测，验证协议自洽后再动固件。 + +### 变更 + +- **protocol.py**：新增 OTA 模块——`CMD_OTA_*` 命令常量、`ota_crc32`（CRC-32/ISO-HDLC = zlib.crc32，标准向量 0xCBF43926）、`ota_split_bin`（256B/片 hex，96KB 上限）、`data_ota_begin/data/end/abort/flash` 构建器、状态/阶段/错误码中文描述 +- **main.py**：新增 **OTA 页签**——固件 bin 选择（≤96KB 校验 + 全镜像 CRC32 显示）、目标版本/force/slot 参数、①下载（begin→data×N→end，`OtaDownloadThread` QThread 后台 + 响应队列，断点续传/缺片重定位/CRC 错重发）②刷写（ota_flash 二次确认）③中止 ④查询状态；进度条 + 状态日志（ota_report/ota_status 实时显示） +- **tests/test_ota_protocol.py**：22 断言——CRC32 标准向量、分片边界（1B/256B/257B/超限）、构建器字段、**OtaDeviceSim 设备侧状态机**（顺序/幂等/乱序拒绝/单片 CRC/全镜像复核/断点续传）+ 完整下载流（正常/缺片补发） + +### 验证 + +- `venv/bin/python -m unittest tests.test_ota_protocol`：22 全过 +- offscreen MainWindow 冒烟：OTA tab 构建正常、控件齐全、512B bin → 2 片 CRC 正确 +- 协议规则自洽性验证：设备模拟器按文档规则（offset==received 顺序片 / offsetreceived 拒收 code=1）与工具分片流程对跑，断点续传与缺片重定位全通 + +### 已知约束 + +- 下载线程每片停等 ACK（~512 片 × RTT）；量产平台可改窗口并发，协议层无需变更 +- 固件未实现 ota_* 前，工具下发会收 `code=4 unsupported`——能力探测见协议 §4.19 + +--- + ## 2026-08-18 — log_query 响应改 hex 原始字节解析展示（协议 V1.07 修订） ### 背景 diff --git a/DBNMQTTool/main.py b/DBNMQTTool/main.py index 690ddb7..342c2c9 100644 --- a/DBNMQTTool/main.py +++ b/DBNMQTTool/main.py @@ -7,6 +7,8 @@ DBN MQTT Tool — DLD960 IoT MQTT 设备管理工具 import sys import os import json +import queue +import time from datetime import datetime from typing import Optional @@ -18,8 +20,9 @@ from PySide6.QtWidgets import ( QTabWidget, QTextEdit, QGroupBox, QGridLayout, QCheckBox, QSplitter, QMessageBox, QHeaderView, QFrame, QComboBox, QSpinBox, QPlainTextEdit, + QFileDialog, QProgressBar, ) -from PySide6.QtCore import Qt, QTimer, Signal, Slot +from PySide6.QtCore import Qt, QTimer, Signal, Slot, QThread from PySide6.QtGui import QFont, QColor, QTextCursor import paho.mqtt.client as mqtt @@ -39,8 +42,128 @@ from dbn_mqtt_tool.protocol import ( data_pwd_verify, data_pwd_set, data_report_config, data_log_query, data_log_clear, build_request, next_msg_id, topic_down, topic_up, + # V1.08 OTA + CMD_OTA_BEGIN, CMD_OTA_DATA, CMD_OTA_END, CMD_OTA_ABORT, + CMD_OTA_FLASH, CMD_OTA_STATUS, CMD_OTA_REPORT, + OTA_TARGET_LOOP, OTA_CHUNK_SIZE, OTA_SLOT_A, OTA_SLOT_B, + OTA_STATE_DESC, OTA_STAGE_DESC, OTA_ERROR_DESC, + data_ota_begin, data_ota_data, data_ota_end, data_ota_abort, data_ota_flash, + ota_crc32, ota_split_bin, ) +# OTA 相关命令集合 (用于响应分发) +OTA_CMD_SET = {CMD_OTA_BEGIN, CMD_OTA_DATA, CMD_OTA_END, CMD_OTA_ABORT, + CMD_OTA_FLASH, CMD_OTA_STATUS} + + +class OtaDownloadThread(QThread): + """OTA 分片下载线程 (协议 V1.08): ota_begin → ota_data×N → ota_end + 断点续传: ota_begin 响应 offset 为设备已接收字节数, 从此处续传 + 重定位: ota_data code=1 时按 data.offset 调整 (缺片/乱序/CRC) + """ + progress = Signal(int, int) # sent, total + log = Signal(str) + finished_ok = Signal(str) + finished_err = Signal(str) + + def __init__(self, mqtt_client: MqttClient, dev_serial: str, bin_data: bytes, + version: str, force: bool, resp_queue: "queue.Queue"): + super().__init__() + self._mqtt = mqtt_client + self._sn = dev_serial + self._bin = bin_data + self._version = version + self._force = force + self._resp_queue = resp_queue + self._stop = False + + def stop(self): + self._stop = True + + def _send(self, cmd: str, data: dict) -> int: + from dbn_mqtt_tool.protocol import get_topic_for_cmd + msg = build_request(cmd, data) + self._mqtt.publish(get_topic_for_cmd(cmd, self._sn), msg, qos=1) + return msg["msg_id"] + + def _wait_resp(self, mid: int, cmd: str, timeout: float = 5.0) -> dict: + """等待与 msg_id 匹配的响应 (resp_queue 由 _on_message 填充)""" + deadline = time.time() + timeout + while time.time() < deadline and not self._stop: + try: + c, payload = self._resp_queue.get(timeout=0.2) + except queue.Empty: + continue + if c == cmd and payload.get("msg_id") == mid: + return payload + raise TimeoutError(f"{cmd} 响应超时 (msg_id={mid})") + + def run(self): + try: + size = len(self._bin) + total_crc = ota_crc32(self._bin) + chunks = ota_split_bin(self._bin) + self.log.emit(f"[OTA] 镜像 {size}B CRC32=0x{total_crc:08X} → {len(chunks)} 片 @{OTA_CHUNK_SIZE}B") + + # 1. ota_begin (断点续传定位) + mid = self._send(CMD_OTA_BEGIN, data_ota_begin( + size, total_crc, version=self._version, force=self._force)) + resp = self._wait_resp(mid, CMD_OTA_BEGIN) + if resp.get("code") != 0: + raise RuntimeError(f"ota_begin 失败: code={resp.get('code')} {resp.get('msg')} " + f"(err_code={resp.get('data', {}).get('err_code', '-')})") + d = resp.get("data", {}) + offset = int(d.get("offset", 0)) + self.log.emit(f"[OTA] begin OK slot={d.get('slot', '?')} state={d.get('state', '?')} " + f"offset={offset} (断点续传点)") + + # 2. 逐片下发 (停等: 每片等响应) + retry = 0 + while offset < size and not self._stop: + idx = offset // OTA_CHUNK_SIZE + if idx >= len(chunks): + # 设备 received 溢出, 直接截断到 size + break + off, crc, hexs = chunks[idx] + mid = self._send(CMD_OTA_DATA, data_ota_data(off, crc, hexs)) + resp = self._wait_resp(mid, CMD_OTA_DATA) + code = resp.get("code") + if code == 0: + offset = int(resp.get("data", {}).get("received", offset + OTA_CHUNK_SIZE)) + retry = 0 + self.progress.emit(min(offset, size), size) + elif code == 1: + # 缺片/乱序/CRC 失败 → 按设备指示重定位 (data.offset=received) + new_off = int(resp.get("data", {}).get("offset", offset)) + if new_off == offset: + retry += 1 + if retry > 3: + raise RuntimeError("分片连续失败 (>3 次), 建议 ota_abort 后重来") + else: + retry = 0 + offset = new_off + else: + raise RuntimeError(f"ota_data@{off} 失败: code={code} {resp.get('msg')} " + f"(err_code={resp.get('data', {}).get('err_code', '-')})") + self.log.emit(f"[OTA] data@{off} → received={offset}") + + if self._stop: + self.log.emit("[OTA] 已停止") + return + self.progress.emit(size, size) + + # 3. ota_end (全镜像 CRC32 复核) + mid = self._send(CMD_OTA_END, data_ota_end(total_crc)) + resp = self._wait_resp(mid, CMD_OTA_END, timeout=10) + if resp.get("code") != 0: + raise RuntimeError(f"ota_end 失败: code={resp.get('code')} {resp.get('msg')} " + f"crc_ok={resp.get('data', {}).get('crc_ok', '-')}") + self.log.emit("[OTA] end OK crc_ok=true → state=ready, 可触发 ota_flash") + self.finished_ok.emit("下载+校验完成, 可触发刷写 (ota_flash)") + except Exception as e: + self.log.emit(f"[OTA] ✗ {e}") + self.finished_err.emit(str(e)) + class MainWindow(QMainWindow): _mqtt_status = Signal(bool, str) _mqtt_msg = Signal(str, str, object) @@ -91,6 +214,7 @@ class MainWindow(QMainWindow): self._notebook.addTab(self._build_config_tab(), "参数配置") self._notebook.addTab(self._build_data_tab(), "实时数据") self._notebook.addTab(self._build_log_tab(), "日志") + self._notebook.addTab(self._build_ota_tab(), "OTA") self._notebook.addTab(self._build_simulate_tab(), "模拟上报") self._notebook.addTab(self._build_proto_topic_tab(), "协议Topic") self._notebook.addTab(self._build_custom_topic_tab(), "自定义Topic") @@ -429,6 +553,192 @@ class MainWindow(QMainWindow): layout.addLayout(layout2) return w + def _build_ota_tab(self) -> QWidget: + """OTA 升级页签 (协议 V1.08: Loop 远程 OTA, 先存后刷)""" + w = QWidget() + layout = QVBoxLayout(w) + + # -- 固件选择 -- + row1 = QHBoxLayout() + row1.addWidget(QLabel("固件 bin:")) + self._ota_file = QLineEdit() + self._ota_file.setReadOnly(True) + self._ota_file.setPlaceholderText("选择 Loop 固件 bin (≤96KB, 0x08003400 起 APP 映像)") + row1.addWidget(self._ota_file, 1) + btn_file = QPushButton("选择…") + btn_file.clicked.connect(self._ota_choose_file) + row1.addWidget(btn_file) + layout.addLayout(row1) + + # -- 参数 -- + row2 = QHBoxLayout() + row2.addWidget(QLabel("目标版本:")) + self._ota_version = QLineEdit("1.1.0") + self._ota_version.setMaximumWidth(110) + row2.addWidget(self._ota_version) + self._ota_force = QCheckBox("force") + self._ota_force.setToolTip("强制: 覆盖现有镜像 / 跳过安全检查 (高风险)") + row2.addWidget(self._ota_force) + row2.addWidget(QLabel("slot:")) + self._ota_slot = QComboBox() + self._ota_slot.addItems([OTA_SLOT_A, OTA_SLOT_B]) + self._ota_slot.setMaximumWidth(60) + row2.addWidget(self._ota_slot) + self._ota_info = QLabel("未选择固件") + row2.addWidget(self._ota_info, 1) + layout.addLayout(row2) + + # -- 操作按钮 -- + row3 = QHBoxLayout() + b_begin = QPushButton("① 下载 (begin/data/end)") + b_begin.setToolTip("开启会话→分片下发→全镜像校验; 已下载过则断点续传") + b_begin.clicked.connect(self._ota_start_download) + row3.addWidget(b_begin) + b_flash = QPushButton("② 刷写 (ota_flash)") + b_flash.setToolTip("仅 state=ready 可触发; 安全检查 (有车拒绝) 后异步执行") + b_flash.clicked.connect(self._ota_send_flash) + row3.addWidget(b_flash) + b_abort = QPushButton("中止 (ota_abort)") + b_abort.clicked.connect(self._ota_send_abort) + row3.addWidget(b_abort) + b_status = QPushButton("查询状态 (ota_status)") + b_status.clicked.connect(self._ota_send_status) + row3.addWidget(b_status) + b_clear = QPushButton("清空日志") + b_clear.clicked.connect(lambda: self._ota_log_text.clear()) + row3.addWidget(b_clear) + layout.addLayout(row3) + + # -- 进度 -- + self._ota_progress = QProgressBar() + self._ota_progress.setRange(0, 100) + self._ota_progress.setValue(0) + layout.addWidget(self._ota_progress) + + # -- 状态日志 -- + self._ota_log_text = QPlainTextEdit() + self._ota_log_text.setReadOnly(True) + self._ota_log_text.setFont(QFont("Consolas", 9)) + self._ota_log_text.setMaximumBlockCount(2000) + layout.addWidget(self._ota_log_text, 1) + + # -- 响应队列 (下载线程 ← _on_message) -- + self._ota_resp_queue: "queue.Queue" = queue.Queue() + self._ota_thread: Optional[OtaDownloadThread] = None + return w + + # ================================================================ + # OTA 升级 (V1.08) + # ================================================================ + + def _ota_choose_file(self): + path, _ = QFileDialog.getOpenFileName(self, "选择固件 bin", "", + "固件文件 (*.bin);;所有文件 (*)") + if path: + self._ota_file.setText(path) + try: + with open(path, "rb") as f: + data = f.read() + if len(data) > 96 * 1024: + QMessageBox.warning(self, "固件过大", f"{len(data)}B 超上限 96KB (Slot 100KB-4KB)") + crc = ota_crc32(data) + self._ota_info.setText(f"{len(data)}B · CRC32=0x{crc:08X}") + except OSError as e: + QMessageBox.critical(self, "读取失败", str(e)) + + def _ota_load_bin(self) -> Optional[bytes]: + path = self._ota_file.text() + if not path: + QMessageBox.warning(self, "提示", "请先选择固件 bin") + return None + try: + data = open(path, "rb").read() + except OSError as e: + QMessageBox.critical(self, "读取失败", str(e)) + return None + if len(data) > 96 * 1024: + QMessageBox.warning(self, "固件过大", f"{len(data)}B 超上限 96KB") + return None + return data + + def _ota_log(self, text: str): + self._ota_log_text.appendPlainText(f"[{datetime.now().strftime('%H:%M:%S')}] {text}") + + def _ota_start_download(self): + sn = self._require_dev() + if not sn: + return + if self._ota_thread and self._ota_thread.isRunning(): + QMessageBox.warning(self, "提示", "OTA 下载已在运行, 先中止") + return + data = self._ota_load_bin() + if data is None: + return + self._ota_log(f"开始 OTA 下载: sn={sn} size={len(data)}B") + self._ota_progress.setValue(0) + self._ota_thread = OtaDownloadThread( + self._mqtt, sn, data, + version=self._ota_version.text().strip() or "0.0.0", + force=self._ota_force.isChecked(), + resp_queue=self._ota_resp_queue, + ) + self._ota_thread.progress.connect(self._ota_on_progress) + self._ota_thread.log.connect(self._ota_log) + self._ota_thread.finished_ok.connect( + lambda m: (self._ota_log(f"✓ {m}"), + QMessageBox.information(self, "OTA", m))) + self._ota_thread.finished_err.connect( + lambda m: QMessageBox.critical(self, "OTA 失败", m)) + self._ota_thread.start() + + def _ota_on_progress(self, sent: int, total: int): + pct = int(sent * 100 / total) if total else 0 + self._ota_progress.setValue(pct) + self._ota_progress.setFormat(f"{sent}/{total} B ({pct}%)") + + def _ota_send_flash(self): + sn = self._require_dev() + if not sn: + return + confirm = QMessageBox.question( + self, "确认刷写", + "触发 ota_flash 将把暂存镜像刷入 Loop MCU (约 6~8s 窗口, 期间检测中断)。\n" + "请确认现场无车压线圈。继续?", + QMessageBox.Yes | QMessageBox.No) + if confirm != QMessageBox.Yes: + return + self._send_cmd(CMD_OTA_FLASH, data_ota_flash( + slot=self._ota_slot.currentText(), force=self._ota_force.isChecked())) + + def _ota_send_abort(self): + if self._ota_thread and self._ota_thread.isRunning(): + self._ota_thread.stop() + self._send_cmd(CMD_OTA_ABORT, data_ota_abort()) + + def _ota_send_status(self): + self._send_cmd(CMD_OTA_STATUS) + + def _ota_show_status(self, data: dict): + """显示 ota_status / ota_report 状态""" + line = [] + state = data.get("state", "") + stage = data.get("stage", "") + prog = data.get("progress", {}) + if state: + line.append(f"state={state}({OTA_STATE_DESC.get(state, '?')})") + if stage: + line.append(f"stage={stage}({OTA_STAGE_DESC.get(stage, '?')})") + if prog: + line.append(f"progress={prog.get('sent', 0)}/{prog.get('total', 0)}B") + if data.get("received") is not None: + line.append(f"received={data['received']}B") + if data.get("last_result"): + line.append(f"last_result={data['last_result']}") + if data.get("last_error"): + err = OTA_ERROR_DESC.get(data["last_error"], data["last_error"]) + line.append(f"last_error={err}") + self._ota_log("状态: " + (" | ".join(line) if line else json.dumps(data, ensure_ascii=False))) + # ================================================================ # 模拟设备上报 # ================================================================ @@ -946,7 +1256,15 @@ class MainWindow(QMainWindow): self._apply_log_query_data(data) elif cmd == CMD_LOG_CLEAR: self._log(f"脱机日志已清除 (审计留痕, 需重新统计确认)") + elif cmd in OTA_CMD_SET: + # OTA 响应 → 喂下载线程响应队列 + 状态显示 + self._ota_resp_queue.put((cmd, payload)) + self._ota_show_status(payload.get("data", {})) else: + if cmd in OTA_CMD_SET: + self._ota_resp_queue.put((cmd, payload)) + self._ota_log(f"{cmd} 失败 code={code} {pmsg} " + f"(err_code={payload.get('data', {}).get('err_code', '-')})") self._show_json({"error": f"code={code} {pmsg} ({ERROR_MSGS.get(code, '?')})"}) elif cmd == CMD_LOOP_DATA: @@ -982,6 +1300,12 @@ class MainWindow(QMainWindow): self._log_recv(topic, payload, f"initialize sn={dev_sn} model={model}") self._show_json({"online": dev_sn, "model": model, "extra": extra}) + elif cmd == CMD_OTA_REPORT: + # V1.08: OTA 进度/结果主动上报 (无 code 字段) + data = payload.get("data", {}) + self._ota_show_status(data) + self._log_recv(topic, payload, f"ota_report stage={data.get('stage', '?')}") + # 旧协议 / 无 cmd 字段 的消息（如 Initialize） elif "Method" in payload: self._log_recv(topic, payload, f"Method={payload.get('Method', '?')}") diff --git a/DBNMQTTool/tests/test_ota_protocol.py b/DBNMQTTool/tests/test_ota_protocol.py new file mode 100644 index 0000000..20076cd --- /dev/null +++ b/DBNMQTTool/tests/test_ota_protocol.py @@ -0,0 +1,316 @@ +""" +DLD960 IoT MQTT OTA 协议单测 (V1.08) + +验证对象: dbn_mqtt_tool/protocol.py 的 OTA 模块 + - ota_crc32: CRC-32/ISO-HDLC 标准向量 + - ota_split_bin: 分片边界 (256B 协议常量, 96KB 上限) + - data_ota_*: 命令构建器字段 + - OtaDeviceSim: 设备侧接收状态机模拟 (协议 §4.19~4.21 规则) + - 完整下载流: 正常/断点续传/缺片重定位/CRC 错重发/重复片幂等 + +运行: venv/bin/python -m unittest tests.test_ota_protocol -v +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from dbn_mqtt_tool.protocol import ( + ota_crc32, ota_split_bin, OTA_CHUNK_SIZE, OTA_MAX_SIZE, + data_ota_begin, data_ota_data, data_ota_end, data_ota_abort, data_ota_flash, + OTA_TARGET_LOOP, OTA_SLOT_A, +) + + +class OtaDeviceSim: + """模拟固件 OTA 接收状态机 (协议 §4.19 ota_begin / §4.20 ota_data / §4.21 ota_end) + + 规则 (与协议文档一致): + - 顺序片 (offset == received): 单片 CRC32 校验 → 写暂存 → received += 256 + - 重复片 (offset < received): 幂等回 code=0, 不重写 + - 缺片 (offset > received): code=1, data.offset=received (要求续传) + - 单片 CRC 失败: code=1 (平台重发本片) + - ota_end: received!=size → code=1; 全镜像 CRC 复核 → ready + """ + + def __init__(self, bin_data: bytes): + self.bin = bin_data + self.size = len(bin_data) + self.state = "idle" + self.received = 0 + self.storage = bytearray(self.size) + + def _base(self, offset: int) -> dict: + return {"target": OTA_TARGET_LOOP, "slot": OTA_SLOT_A, + "offset": offset, "received": self.received, + "size": self.size, "crc32": ota_crc32(self.bin), + "state": self.state} + + def ota_begin(self, req: dict) -> dict: + # 镜像一致且已 ready → 直接返回可刷写 + if self.state == "ready" and req["size"] == self.size \ + and req["crc32"] == ota_crc32(self.bin): + return {"code": 0, "data": self._base(self.size)} + # downloading 中断 → 返回已接收偏移 (断点续传) + if self.state == "downloading": + return {"code": 0, "data": self._base(self.received)} + self.state = "downloading" + self.received = 0 + self.storage = bytearray(self.size) + return {"code": 0, "data": self._base(0)} + + def ota_data(self, req: dict) -> dict: + offset, crc, hexs = req["offset"], req["crc32"], req["data"] + data = bytes.fromhex(hexs) + if offset == self.received: + if ota_crc32(data) != crc: + return {"code": 1, "err_code": 1, "msg": "crc mismatch", + "data": self._base(self.received)} + self.storage[offset:offset + len(data)] = data + self.received = min(offset + len(data), self.size) + return {"code": 0, "data": self._base(offset)} + if offset < self.received: + return {"code": 0, "data": self._base(offset)} # 幂等 + return {"code": 1, "err_code": 1, "msg": "gap/out-of-order", + "data": self._base(self.received)} + + def ota_end(self, req: dict) -> dict: + if self.received != self.size: + return {"code": 1, "data": self._base(self.received)} + if ota_crc32(bytes(self.storage)) != req["crc32"]: + return {"code": 5, "err_code": 5, "data": {"crc_ok": False}} + self.state = "ready" + return {"code": 0, "data": {"crc_ok": True}} + + +class TestOtaCrc32(unittest.TestCase): + def test_standard_vector(self): + """CRC-32/ISO-HDLC 标准校验向量: crc32('123456789') = 0xCBF43926""" + self.assertEqual(ota_crc32(b"123456789"), 0xCBF43926) + + def test_empty(self): + self.assertEqual(ota_crc32(b""), 0x00000000) + + def test_zlib_alias(self): + import zlib + for data in (b"", b"a", os.urandom(300)): + self.assertEqual(ota_crc32(data), zlib.crc32(data) & 0xFFFFFFFF) + + +class TestOtaSplitBin(unittest.TestCase): + def test_small_single_chunk(self): + data = b"\x00\x01\x02\x03" + chunks = ota_split_bin(data) + self.assertEqual(len(chunks), 1) + off, crc, hexs = chunks[0] + self.assertEqual(off, 0) + self.assertEqual(hexs, data.hex()) + self.assertEqual(len(hexs), 8) # 4B → 8 hex + self.assertEqual(crc, ota_crc32(data)) + + def test_exact_chunk(self): + data = bytes(range(256)) + chunks = ota_split_bin(data) + self.assertEqual(len(chunks), 1) + self.assertEqual(len(chunks[0][2]), 512) # 256B → 512 hex + + def test_two_chunks(self): + data = bytes(range(256)) + b"\x00" # 257B → 2 片 + chunks = ota_split_bin(data) + self.assertEqual(len(chunks), 2) + self.assertEqual(chunks[0][0], 0) + self.assertEqual(chunks[1][0], 256) + self.assertEqual(len(chunks[1][2]), 2) # 末片 1B + + def test_offsets_256_aligned(self): + data = os.urandom(5000) + for off, _, _ in ota_split_bin(data): + self.assertEqual(off % OTA_CHUNK_SIZE, 0) + + def test_roundtrip_reconstruct(self): + data = os.urandom(4097) + merged = b"".join(bytes.fromhex(hexs) for _, _, hexs in ota_split_bin(data)) + self.assertEqual(merged, data) + + def test_over_max_size(self): + with self.assertRaises(ValueError): + ota_split_bin(os.urandom(OTA_MAX_SIZE + 1)) + + +class TestOtaBuilders(unittest.TestCase): + def test_begin(self): + d = data_ota_begin(46864, 0x12345678, version="1.1.0") + self.assertEqual(d, {"target": "loop", "size": 46864, "crc32": 0x12345678, + "version": "1.1.0", "force": False}) + + def test_begin_force(self): + d = data_ota_begin(100, 1, force=True) + self.assertTrue(d["force"]) + + def test_data(self): + d = data_ota_data(0, 0x9E3779B9, "abcd") + self.assertEqual(d, {"target": "loop", "offset": 0, + "crc32": 0x9E3779B9, "data": "abcd"}) + + def test_end_abort_flash(self): + self.assertEqual(data_ota_end(0x11223344), + {"target": "loop", "crc32": 0x11223344}) + self.assertEqual(data_ota_abort(), {"target": "loop"}) + self.assertEqual(data_ota_flash(slot="b", force=True), + {"target": "loop", "slot": "b", "force": True}) + + +class TestDeviceSim(unittest.TestCase): + """设备侧规则: 顺序/幂等/乱序/CRC/全镜像复核""" + + def setUp(self): + self.bin_data = bytes(range(256)) * 20 # 5120B = 20 片 + self.dev = OtaDeviceSim(self.bin_data) + + def test_normal_flow(self): + dev = self.dev + chunks = ota_split_bin(self.bin_data) + resp = dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data))) + self.assertEqual(resp["code"], 0) + self.assertEqual(resp["data"]["offset"], 0) + for off, crc, hexs in chunks: + resp = dev.ota_data(data_ota_data(off, crc, hexs)) + self.assertEqual(resp["code"], 0, f"片@{off} 应成功") + self.assertEqual(dev.received, len(self.bin_data)) + resp = dev.ota_end(data_ota_end(ota_crc32(self.bin_data))) + self.assertEqual(resp["code"], 0) + self.assertTrue(resp["data"]["crc_ok"]) + self.assertEqual(dev.state, "ready") + self.assertEqual(bytes(dev.storage), self.bin_data) + + def test_resume_after_begin(self): + """断点续传: downloading 中断后 begin 返回已接收偏移""" + dev = self.dev + chunks = ota_split_bin(self.bin_data) + dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data))) + for off, crc, hexs in chunks[:7]: # 只收 7 片 + dev.ota_data(data_ota_data(off, crc, hexs)) + self.assertEqual(dev.received, 7 * OTA_CHUNK_SIZE) + # 重新 begin → 续传点 + resp = dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data))) + self.assertEqual(resp["data"]["offset"], 7 * OTA_CHUNK_SIZE) + # 从续传点继续 + for off, crc, hexs in chunks[7:]: + dev.ota_data(data_ota_data(off, crc, hexs)) + self.assertEqual(dev.received, len(self.bin_data)) + resp = dev.ota_end(data_ota_end(ota_crc32(self.bin_data))) + self.assertTrue(resp["data"]["crc_ok"]) + + def test_dup_chunk_idempotent(self): + """重复片幂等: 已收片重发 → code=0, received 不变""" + dev = self.dev + chunks = ota_split_bin(self.bin_data) + dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data))) + off0, crc0, hexs0 = chunks[0] + dev.ota_data(data_ota_data(off0, crc0, hexs0)) + r1 = dev.ota_data(data_ota_data(off0, crc0, hexs0)) # 重复 + self.assertEqual(r1["code"], 0) + self.assertEqual(r1["data"]["received"], OTA_CHUNK_SIZE) + + def test_gap_rejected(self): + """缺片: offset > received → code=1, data.offset=received""" + dev = self.dev + chunks = ota_split_bin(self.bin_data) + dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data))) + off2, crc2, hexs2 = chunks[2] # 跳过 0,1 + resp = dev.ota_data(data_ota_data(off2, crc2, hexs2)) + self.assertEqual(resp["code"], 1) + self.assertEqual(resp["data"]["offset"], 0) # 要求从 0 续传 + + def test_chunk_crc_error(self): + """单片 CRC 错 → code=1, 不落盘""" + dev = self.dev + chunks = ota_split_bin(self.bin_data) + dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data))) + off0, _, hexs0 = chunks[0] + resp = dev.ota_data(data_ota_data(off0, 0xDEADBEEF, hexs0)) # 错 crc + self.assertEqual(resp["code"], 1) + self.assertEqual(dev.received, 0) # 未落盘 + # 重发正确片 → 成功 + resp = dev.ota_data(data_ota_data(off0, ota_crc32(self.bin_data[:256]), hexs0)) + self.assertEqual(resp["code"], 0) + + def test_end_incomplete(self): + """ota_end 时 received != size → code=1""" + dev = self.dev + dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data))) + resp = dev.ota_end(data_ota_end(ota_crc32(self.bin_data))) + self.assertEqual(resp["code"], 1) + self.assertEqual(resp["data"]["offset"], 0) + + def test_end_crc_fail(self): + """全镜像 CRC 复核失败 → code=5 crc_ok=false""" + dev = self.dev + chunks = ota_split_bin(self.bin_data) + dev.ota_begin(data_ota_begin(len(self.bin_data), ota_crc32(self.bin_data))) + for off, crc, hexs in chunks: + dev.ota_data(data_ota_data(off, crc, hexs)) + resp = dev.ota_end(data_ota_end(0x00000000)) # 错的全镜像 crc + self.assertEqual(resp["code"], 5) + self.assertFalse(resp["data"]["crc_ok"]) + + +class TestDownloadFlow(unittest.TestCase): + """工具侧下载流程 (等价 OtaDownloadThread 核心逻辑, 无 Qt 依赖)""" + + def _run_flow(self, bin_data: bytes, simulate_gap: bool = False): + dev = OtaDeviceSim(bin_data) + chunks = ota_split_bin(bin_data) + total_crc = ota_crc32(bin_data) + + # ota_begin + resp = dev.ota_begin(data_ota_begin(len(bin_data), total_crc)) + assert resp["code"] == 0 + offset = resp["data"]["offset"] + + sent_offsets = [] + gap_done = False + while offset < len(bin_data): + idx = offset // OTA_CHUNK_SIZE + off, crc, hexs = chunks[idx] + if simulate_gap and not gap_done and off == 256: + # 模拟一次丢包: 跳过该片直接发下一片 → 设备回 gap → 重定位 + gap_done = True + resp = dev.ota_data(data_ota_data(chunks[idx + 1][0], + chunks[idx + 1][1], + chunks[idx + 1][2])) + self.assertEqual(resp["code"], 1) + offset = resp["data"]["offset"] # 重定位 + continue + resp = dev.ota_data(data_ota_data(off, crc, hexs)) + if resp["code"] == 0: + offset = resp["data"]["received"] + elif resp["code"] == 1: + offset = resp["data"]["offset"] # 重定位 + else: + self.fail(f"意外 code={resp['code']}") + sent_offsets.append(off) + + # ota_end + resp = dev.ota_end(data_ota_end(total_crc)) + self.assertEqual(resp["code"], 0) + self.assertTrue(resp["data"]["crc_ok"]) + self.assertEqual(dev.state, "ready") + self.assertEqual(bytes(dev.storage), bin_data) + return sent_offsets + + def test_flow_normal(self): + bin_data = os.urandom(3000) + self._run_flow(bin_data) + + def test_flow_with_gap_retransmit(self): + """缺片重传: 丢 1 片后设备要求重定位, 工具补发, 最终全量一致""" + bin_data = os.urandom(3000) + sent = self._run_flow(bin_data, simulate_gap=True) + # 所有片最终都被发送过 (含补发) + self.assertIn(256, sent) + + +if __name__ == "__main__": + unittest.main(verbosity=2)