小龙虾
00ce8dbbb5
- 新增 storage/gd32_flash_store.cpp 组件，提供 GD32 Flash 存储抽象层 - 修改 ota_coordinator.cpp/h 使其继承 IRelayListener 接口，实现与 OtaRelay 的解耦设计 - 将 GD32 相关存储操作统一通过协调器访问，解决 H3 绕过协调器的问题 - 添加完整的 Doxygen 注释到所有公共接口和关键函数 - 更新 CMakeLists.txt 构建配置，添加 storage 组件依赖 - 重构元数据加载/清除操作使用 Gd32FlashStore 实例而非 OtaRelay 直接访问 - 添加中继事件回调接口实现，使 OtaRelay 可通知协调器状态变化
62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
ota_upload_test.py - ESP32 OTA 上传集成测试脚本
|
||
|
||
用法: python ota_upload_test.py
|
||
前提: 已编译 build/ESP32_COMM.bin，ESP32 已连接网络。
|
||
步骤: 上传固件 → 触发更新 → 等待重启并验证版本。
|
||
"""
|
||
import urllib.request, json, time, sys
|
||
|
||
ESP_IP = "192.168.31.127" # 目标 ESP32 IP 地址（根据实际网络修改）
|
||
|
||
# Step 1: Upload ESP32 firmware
|
||
bin_path = r"G:\Robot_v2\DCU\ESP32_COMM\build\ESP32_COMM.bin"
|
||
with open(bin_path, "rb") as f:
|
||
fw_data = f.read()
|
||
|
||
header = json.dumps({"type": "esp32", "total_size": len(fw_data)}).encode("utf-8")
|
||
body = header + fw_data
|
||
print(f"[1/3] Uploading {len(fw_data)} bytes...")
|
||
|
||
req = urllib.request.Request(f"http://{ESP_IP}/ota/upload", data=body, method="POST")
|
||
req.add_header("Content-Type", "application/octet-stream")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=120) as resp:
|
||
result = resp.read().decode()
|
||
print(f" Upload OK: {result}")
|
||
except urllib.error.HTTPError as e:
|
||
body_text = e.read().decode()
|
||
print(f" Upload FAILED: HTTP {e.code} - {body_text}")
|
||
sys.exit(1)
|
||
|
||
# Step 2: Trigger update
|
||
print("[2/3] Triggering OTA update...")
|
||
time.sleep(1)
|
||
req2 = urllib.request.Request(f"http://{ESP_IP}/ota/update", data=b"", method="POST")
|
||
try:
|
||
with urllib.request.urlopen(req2, timeout=10) as resp:
|
||
result2 = resp.read().decode()
|
||
print(f" Update OK: {result2}")
|
||
except (ConnectionResetError, urllib.error.URLError, OSError) as e:
|
||
print(f" Update triggered (connection reset expected): {type(e).__name__}")
|
||
|
||
# Step 3: Wait for ESP32 to come back
|
||
print("[3/3] Waiting for ESP32 to reboot with new firmware...")
|
||
for attempt in range(1, 31):
|
||
time.sleep(5)
|
||
try:
|
||
req3 = urllib.request.Request(f"http://{ESP_IP}/system/status")
|
||
with urllib.request.urlopen(req3, timeout=5) as resp:
|
||
data = json.loads(resp.read().decode())
|
||
ver = data.get("version", "?")
|
||
bld = data.get("build_time", "?")
|
||
upt = data.get("uptime_sec", 0)
|
||
print(f" Online! version={ver} build={bld} uptime={upt}s")
|
||
break
|
||
except Exception:
|
||
if attempt % 5 == 0:
|
||
print(f" Still waiting... ({attempt}/30)")
|
||
else:
|
||
print(" ESP32 did not come back within 150 seconds")
|