cbb/ESP32_COMM
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(wifi_ota): 添加CAN帧转发功能

Browse Source 
添加了HTTP接口用于转发CAN帧，包括/can/send POST接口用于发送CAN帧，
以及/can/status GET接口用于查询CAN通道状态。同时集成了SPI master驱动
以支持与GD32之间的CANFD通信，并定义了完整的CAN帧格式和通道管理机制。

BREAKING CHANGE: 移除了OTA上位机工具脚本
This commit is contained in:
小龙虾
2026-04-27 03:22:49 +08:00
parent df27ef3a98
commit 47ffe1bd54
5 changed files with 358 additions and 660 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/server/wifi_ota.cpp
+111
View File
@@ -21,6 +21,8 @@
#include "ota_relay.h"
#include "health_monitor.h"
#include "monitor/debug_log.h"
#include "spi_master.h"
#include "cJSON.h"
static ota_wifi_mode_t s_wifi_mode = OTA_WIFI_MODE_AP;
static bool s_ota_storage_done = false;
@@ -252,6 +254,98 @@ static esp_err_t health_handler(httpd_req_t *req)
return ESP_OK;
}
// HTTP POST /can/send 处理器
// 上位机发送 CAN 帧，经 ESP32 转发到 GD32
// 请求体: {"ch": 0, "id": 123456, "data": "0102030405060708", "dlc": 8}
// data 为 hex 字符串（可不填，自动用 dlc 决定实际发送长度）
static esp_err_t can_send_handler(httpd_req_t *req)
{
char buf[256] = {0};
int ret = httpd_req_recv(req, buf, sizeof(buf) - 1);
if (ret <= 0) {
httpd_resp_send_err(req, HTTPD_400_BAD_REQUEST, "接收数据失败");
return ESP_FAIL;
}
buf[ret] = '\0';
cJSON *root = cJSON_Parse(buf);
if (!root) {
httpd_resp_send_err(req, HTTPD_400_BAD_REQUEST, "JSON 解析失败");
return ESP_FAIL;
}
cJSON *ch_item = cJSON_GetObjectItem(root, "ch");
cJSON *id_item = cJSON_GetObjectItem(root, "id");
cJSON *data_item = cJSON_GetObjectItem(root, "data");
cJSON *dlc_item = cJSON_GetObjectItem(root, "dlc");
if (!ch_item || !id_item) {
cJSON_Delete(root);
httpd_resp_send_err(req, HTTPD_400_BAD_REQUEST, "缺少 ch 或 id 字段");
return ESP_FAIL;
}
can_channel_t ch = (can_channel_t)cJSON_GetNumberValue(ch_item);
uint32_t ext_id = (uint32_t)cJSON_GetNumberValue(id_item);
uint8_t dlc = dlc_item ? (uint8_t)cJSON_GetNumberValue(dlc_item) : 0;
uint8_t tx_data[CAN_MAX_DLC] = {0};
uint8_t tx_dlc = 0;
if (data_item && cJSON_IsString(data_item)) {
const char *hex_str = cJSON_GetStringValue(data_item);
size_t hex_len = strlen(hex_str);
for (size_t i = 0; i < hex_len / 2 && i < CAN_MAX_DLC; i++) {
char byte_str[3] = { hex_str[i * 2], hex_str[i * 2 + 1], '\0' };
tx_data[i] = (uint8_t)strtol(byte_str, NULL, 16);
tx_dlc++;
}
}
if (tx_dlc == 0 && dlc > 0) {
tx_dlc = (dlc > CAN_MAX_DLC) ? CAN_MAX_DLC : dlc;
}
cJSON_Delete(root);
// 发送 CAN 帧到 GD32
uint8_t resp_data[CAN_MAX_DLC] = {0};
int resp_len = gd32_can_send(ch, ext_id, tx_data, tx_dlc,
resp_data, sizeof(resp_data), 1000);
char resp[512];
if (resp_len >= 0) {
// 将响应 data 转为 hex 字符串
char resp_hex[CAN_MAX_DLC * 2 + 1] = {0};
for (int i = 0; i < resp_len; i++) {
sprintf(&resp_hex[i * 2], "%02X", resp_data[i]);
}
snprintf(resp, sizeof(resp),
"{\"status\": \"ok\", \"ch\": %d, \"id\": %lu, \"sent\": %d, \"recv_len\": %d, \"recv\": \"%s\"}",
ch, (unsigned long)ext_id, tx_dlc, resp_len, resp_hex);
} else {
snprintf(resp, sizeof(resp),
"{\"status\": \"error\", \"code\": %d}", resp_len);
}
httpd_resp_set_type(req, "application/json");
httpd_resp_send(req, resp, strlen(resp));
return ESP_OK;
}
// HTTP GET /can/status 处理器
static esp_err_t can_status_handler(httpd_req_t *req)
{
char resp[128];
snprintf(resp, sizeof(resp),
"{\"channels\": 3, \"max_dlc\": %d, \"spi_init\": %s}",
CAN_MAX_DLC,
g_spi_device != NULL ? "true" : "false");
httpd_resp_set_type(req, "application/json");
httpd_resp_send(req, resp, strlen(resp));
return ESP_OK;
}
// HTTP GET /wifi/status 处理器
static esp_err_t wifi_status_handler(httpd_req_t *req)
{
@@ -340,6 +434,23 @@ static esp_err_t http_ota_server_start(void)
};
httpd_register_uri_handler(s_http_server, &wifi_status_uri);
// CAN 帧转发
httpd_uri_t can_send_uri = {
.uri = "/can/send",
.method = HTTP_POST,
.handler = can_send_handler,
.user_ctx = NULL
};
httpd_register_uri_handler(s_http_server, &can_send_uri);
httpd_uri_t can_status_uri = {
.uri = "/can/status",
.method = HTTP_GET,
.handler = can_status_handler,
.user_ctx = NULL
};
httpd_register_uri_handler(s_http_server, &can_status_uri);
DEBUG_INFO("WIFI", "HTTP 服务器已启动");
return ESP_OK;
main/transport/spi_master.cpp
+85
View File
@@ -2,12 +2,15 @@
* spi_master.c - SPI Master 驱动实现
* T1.1 - T1.4
* ESP-IDF 6.0 适配版本
* CANFD 3通道发送接口
*/
#include <string.h>
#include <stdio.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "driver/spi_master.h"
#include "soc/spi_struct.h"
#include "esp_err.h"
#include "spi_master.h"
#include "monitor/debug_log.h"
@@ -99,6 +102,88 @@ esp_err_t spi_recv(uint8_t cmd, uint8_t *recv_buf, size_t len, uint32_t timeout_
return spi_sync_transfer(cmd, dummy, len, recv_buf, len, timeout_ms);
}
// CANFD 帧发送
// 帧格式: [通道(1B)] [扩展帧ID(4B BE)] [数据(1~64B)]
static int can_frame_send(can_channel_t channel, uint32_t ext_id,
const uint8_t *data, uint8_t dlc,
uint8_t *recv_buf, uint8_t recv_max,
uint32_t timeout_ms)
{
if (g_spi_device == NULL) {
return -1;
}
if (dlc == 0 || dlc > CAN_MAX_DLC) {
DEBUG_ERR("SPI", "DLC 无效: %d", dlc);
return -2;
}
if (channel >= CAN_CHANNEL_MAX) {
DEBUG_ERR("SPI", "通道号无效: %d", channel);
return -3;
}
// 构建 CANFD 帧: [通道(1B)] [EXT_ID(4B)] [DATA(dlc B)]
uint8_t tx_buf[1 + 4 + CAN_MAX_DLC] = {0};
tx_buf[0] = (uint8_t)channel;
tx_buf[1] = (uint8_t)((ext_id >> 24) & 0xFF);
tx_buf[2] = (uint8_t)((ext_id >> 16) & 0xFF);
tx_buf[3] = (uint8_t)((ext_id >> 8) & 0xFF);
tx_buf[4] = (uint8_t)(ext_id & 0xFF);
memcpy(&tx_buf[5], data, dlc);
size_t tx_len = 5 + dlc;
uint8_t rx_buf[5 + CAN_MAX_DLC] = {0};
spi_transaction_t t = {};
t.tx_buffer = tx_buf;
t.rx_buffer = rx_buf;
t.length = tx_len * 8;
t.rxlength = (recv_max > 0) ? (5 + recv_max) * 8 : 0;
esp_err_t ret = spi_device_transmit(g_spi_device, &t);
if (ret != ESP_OK) {
DEBUG_ERR("SPI", "CAN[ch%d] ID=0x%08X 发送失败: %s",
channel, ext_id, esp_err_to_name(ret));
return -4;
}
if (recv_buf && recv_max > 0) {
// 解析 GD32 响应帧
uint8_t resp_channel = rx_buf[0];
uint32_t resp_id = (rx_buf[1] << 24) | (rx_buf[2] << 16)
| (rx_buf[3] << 8) | rx_buf[4];
uint8_t resp_dlc = resp_id >> 24; // 复用了 ID 字段低字节作为 DLC
resp_id &= 0x00FFFFFF;
uint8_t copy_len = (resp_dlc < recv_max) ? resp_dlc : recv_max;
memcpy(recv_buf, &rx_buf[5], copy_len);
DEBUG_INFO("SPI", "CAN[ch%d] ID=0x%08X 响应 DLC=%d",
resp_channel, resp_id, copy_len);
return copy_len;
}
return 0;
}
int gd32_can_send(can_channel_t channel, uint32_t ext_id,
const uint8_t *data, uint8_t dlc,
uint8_t *recv_buf, uint8_t recv_max,
uint32_t timeout_ms)
{
if (dlc == 0) {
dlc = (data != NULL) ? 64 : 0; // 避免 dlc=0 的情况
}
return can_frame_send(channel, ext_id, data, dlc, recv_buf, recv_max, timeout_ms);
}
int gd32_can_send_no_reply(can_channel_t channel, uint32_t ext_id,
const uint8_t *data, uint8_t dlc)
{
return can_frame_send(channel, ext_id, data, dlc, NULL, 0, 0);
}
// 带重试的 SPI 传输
esp_err_t spi_transfer_with_retry(uint8_t cmd, const uint8_t *payload,
size_t payload_len, uint8_t *recv_buf,
main/transport/spi_master.h
+49
View File
@@ -1,5 +1,6 @@
/*
* spi_master.h - SPI Master 驱动头文件
* CANFD 3通道发送接口 + 底层 SPI 总线
*/
#ifndef SPI_MASTER_H
#define SPI_MASTER_H
@@ -16,6 +17,54 @@
// SPI 设备句柄（外部引用）
extern spi_device_handle_t g_spi_device;
// ============================================================
// CANFD 帧格式（3通道）
// [通道编号: 1B] [扩展帧ID: 4B BE] [数据: 1~64B]
// 长度自动识别，无需发送 DLC
// ============================================================
#define CAN_EXTENDED_ID_BITS (29) // CAN 扩展帧 ID 位数
#define CAN_MAX_DLC (64) // 最大数据长度
#define CAN_CHANNEL_MASK (0x03) // 通道号掩码 (0~2)
// 通道定义
typedef enum {
CAN_CHANNEL_0 = 0,
CAN_CHANNEL_1 = 1,
CAN_CHANNEL_2 = 2,
CAN_CHANNEL_MAX = 3,
} can_channel_t;
// CANFD 帧结构
typedef struct {
can_channel_t channel; // CAN 通道 (0~2)
uint32_t ext_id; // 扩展帧 ID (29bit, Big-Endian 存储)
uint8_t data[CAN_MAX_DLC];
uint8_t dlc; // 数据长度 (1~64)
} can_frame_t;
// ============================================================
// 发送 CANFD 帧
// channel: CAN 通道 (0, 1, 2)
// ext_id: 扩展帧 ID (29bit)
// data: 数据内容 (1~64 字节)
// dlc: 数据长度 (自动识别，可传 0 由 data 长度决定)
// recv_buf: 接收缓冲区 (可传 NULL)
// recv_max: 期望接收最大字节数
// timeout_ms: 超时时间 (ms)
// 返回: >= 0 成功(实际接收字节数), < 0 失败
// ============================================================
int gd32_can_send(can_channel_t channel, uint32_t ext_id,
const uint8_t *data, uint8_t dlc,
uint8_t *recv_buf, uint8_t recv_max,
uint32_t timeout_ms);
// 快速发送接口（不带响应接收）
// 用于广播/通知类 CAN 帧
int gd32_can_send_no_reply(can_channel_t channel, uint32_t ext_id,
const uint8_t *data, uint8_t dlc);
// ============================================================
// 初始化 SPI2 (VSPI) 总线和设备
esp_err_t spi_master_init(void);
tools/ota_host_tool.py
-660
View File
@@ -1,660 +0,0 @@
"""
ESP32-C3 OTA 上位机工具
通过 WiFi 连接 ESP32，发送 .hex 固件文件并触发 OTA
依赖: Python 3.8+, tkinter (标准库), requests
用法:
python ota_host_tool.py
"""
import os
import sys
import socket
import threading
import traceback
import struct
from datetime import datetime
from typing import Optional, List, Tuple
# 尝试导入 tkinter，失败时给出提示
try:
import tkinter as tk
from tkinter import filedialog, messagebox, ttk
except ImportError:
print("错误: 缺少 tkinter 模块")
print("Linux/Mac 请安装: pip install tkinter")
print("Windows 通常自带 tkinter")
sys.exit(1)
try:
import requests
except ImportError:
print("错误: 缺少 requests 模块")
print("请安装: pip install requests")
sys.exit(1)
# ===================== 配置 =====================
ESP32_IP = "192.168.4.1" # ESP32 在 AP 模式下的 IP
ESP32_PORT = 80 # HTTP 端口
DEBUG_LOG_IP = "0.0.0.0" # 监听所有接口
DEBUG_LOG_PORT = 6666 # UDP 日志监听端口
TARGET_CHIP = "GD32E508" # 目标芯片（用于显示）
CHUNK_SIZE = 4096 # 每包大小（字节）
# =============================================
class LogHandler:
"""UDP 日志接收器"""
def __init__(self, text_widget: 'tk.Text', port: int = DEBUG_LOG_PORT):
self.text_widget = text_widget
self.port = port
self.running = False
self.sock: Optional[socket.socket] = None
self.thread: Optional[threading.Thread] = None
# 颜色标签
self.tag_configs = {
'[E]': {'foreground': '#FF4444', 'background': '#FFE4E4'},
'[W]': {'foreground': '#FF8800', 'background': '#FFF4E4'},
'[I]': {'foreground': '#0088FF', 'background': '#E4F0FF'},
'[D]': {'foreground': '#888888', 'background': '#F0F0F0'},
'[C]': {'foreground': '#00CC66', 'background': '#E4FFE4'},
}
def start(self):
"""启动日志接收线程"""
self.running = True
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.bind((DEBUG_LOG_IP, self.port))
self.sock.settimeout(0.5)
self.thread = threading.Thread(target=self._receive_loop, daemon=True)
self.thread.start()
def stop(self):
"""停止日志接收"""
self.running = False
if self.sock:
self.sock.close()
if self.thread:
self.thread.join(timeout=1)
def _receive_loop(self):
"""日志接收循环"""
while self.running:
try:
data, addr = self.sock.recvfrom(1024)
msg = data.decode('utf-8', errors='replace').strip()
if msg:
self._append_log(msg)
except socket.timeout:
continue
except Exception:
pass
def _append_log(self, msg: str):
"""在 Text 组件中追加日志"""
# 根据日志级别应用颜色
applied = False
for prefix, config in self.tag_configs.items():
if f'{prefix}' in msg or prefix in msg:
self.text_widget.insert('end', msg + '\n', prefix)
applied = True
break
if not applied:
self.text_widget.insert('end', msg + '\n')
# 自动滚动到底部
self.text_widget.see('end')
class IntelHexParser:
"""Intel HEX 文件解析器
支持标准 Intel HEX 格式（:BB AAAA TT DD... CC）
"""
def __init__(self, filepath: str):
self.filepath = filepath
self.data: List[Optional[bytes]] = [] # 按地址索引的字节数据
self.max_addr = 0
self.min_addr = 0xFFFFFFFF
def parse(self) -> Tuple[bool, str]:
"""解析 HEX 文件，返回 (成功标志, 消息)"""
try:
with open(self.filepath, 'r', encoding='utf-8') as f:
for line_no, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
ok, msg = self._parse_line(line, line_no)
if not ok:
return False, msg
# 计算实际数据范围
if self.min_addr == 0xFFFFFFFF:
return False, "HEX 文件为空或不包含数据"
return True, f"解析成功，地址范围: 0x{self.min_addr:06X} - 0x{self.max_addr:06X}，共 {self.max_addr - self.min_addr + 1} 字节"
except FileNotFoundError:
return False, f"文件不存在: {self.filepath}"
except Exception as e:
return False, f"解析失败: {e}"
def _parse_line(self, line: str, line_no: int) -> Tuple[bool, str]:
"""解析单行 HEX"""
if not line.startswith(':'):
return False, f"第 {line_no} 行格式错误，缺少 ':' 前缀"
if len(line) < 11:
return False, f"第 {line_no} 行长度不足"
try:
# 解析头部
byte_count = int(line[1:3], 16)
address = int(line[3:7], 16)
record_type = int(line[7:9], 16)
# 验证数据长度
expected_len = 9 + byte_count * 2 + 2
if len(line) != expected_len:
return False, f"第 {line_no} 行长度不匹配，期望 {expected_len} 实际 {len(line)}"
# 解析数据
data_bytes = bytes.fromhex(line[9:9 + byte_count * 2])
# 验证 CRC
checksum = int(line[-2:], 16)
calc_checksum = self._calc_checksum(line[1:-2])
if calc_checksum != checksum:
return False, f"第 {line_no} 行 CRC 校验失败"
except ValueError as e:
return False, f"第 {line_no} 行数据解析错误: {e}"
# 处理不同类型的记录
if record_type == 0x00:
# 数据记录
abs_addr = address
if self.min_addr == 0xFFFFFFFF or abs_addr < self.min_addr:
self.min_addr = abs_addr
if abs_addr + byte_count - 1 > self.max_addr:
self.max_addr = abs_addr + byte_count - 1
# 存储数据（简化处理：直接存储）
if len(self.data) < abs_addr + byte_count:
self.data.extend([None] * (abs_addr + byte_count - len(self.data)))
for i, b in enumerate(data_bytes):
self.data[abs_addr + i] = b
elif record_type == 0x01:
# 文件结束记录
pass
elif record_type == 0x02:
# 扩展段地址
pass
elif record_type == 0x04:
# 扩展线性地址
pass
return True, ""
def _calc_checksum(self, hex_str: str) -> int:
"""计算校验和"""
return sum(int(hex_str[i:i+2], 16) for i in range(0, len(hex_str), 2)) & 0xFF
def get_data_segment(self, start: int, length: int) -> bytes:
"""获取指定范围的数据，不足部分填充 0xFF"""
result = bytearray()
for i in range(length):
addr = start + i
if addr < len(self.data) and self.data[addr] is not None:
result.append(self.data[addr])
else:
result.append(0xFF)
return bytes(result)
def get_total_size(self) -> int:
"""获取总数据大小"""
if self.min_addr == 0xFFFFFFFF:
return 0
return self.max_addr - self.min_addr + 1
class OTAClient:
"""OTA 客户端，与 ESP32 通信"""
def __init__(self, esp_ip: str = ESP32_IP, esp_port: int = ESP32_PORT):
self.esp_ip = esp_ip
self.esp_port = esp_port
self.base_url = f"http://{esp_ip}:{esp_port}"
def upload_firmware(self, filepath: str,
progress_callback=None,
chunk_size: int = CHUNK_SIZE) -> Tuple[bool, str]:
"""上传固件到 ESP32
Args:
filepath: HEX 文件路径
progress_callback: 进度回调函数，参数为 (当前字节, 总字节)
Returns:
(成功标志, 消息)
"""
# 解析 HEX 文件
parser = IntelHexParser(filepath)
ok, msg = parser.parse()
if not ok:
return False, msg
total_size = parser.get_total_size()
print(f"[OTA] 固件大小: {total_size} 字节")
# 分块发送
addr = parser.min_addr
sent_bytes = 0
seq = 0
while addr <= parser.max_addr:
# 计算本块大小
remaining = parser.max_addr - addr + 1
chunk = min(chunk_size, remaining)
# 读取数据
data = parser.get_data_segment(addr, chunk)
# 构造数据包: SEQ(2B) + SIZE(4B) + DATA
packet = struct.pack('<H', seq) + struct.pack('<I', total_size) + data
# 发送
try:
response = requests.post(
f"{self.base_url}/ota/upload",
data=packet,
timeout=10,
headers={'Content-Type': 'application/octet-stream'}
)
if response.status_code != 200:
return False, f"上传失败，HTTP {response.status_code}: {response.text}"
except requests.exceptions.RequestException as e:
return False, f"网络错误: {e}"
sent_bytes += chunk
addr += chunk
seq += 1
if progress_callback:
progress_callback(sent_bytes, total_size)
# 检查存储完成状态
status_ok, status_msg = self.check_status()
if not status_ok:
return False, f"固件已发送但状态检查失败: {status_msg}"
return True, f"固件上传完成，共 {seq} 包，{sent_bytes} 字节"
def trigger_relay(self) -> Tuple[bool, str]:
"""触发 ESP32 开始 OTA 中继"""
try:
response = requests.post(f"{self.base_url}/ota/relay", timeout=5)
if response.status_code == 200:
return True, "OTA 中继已触发"
return False, f"触发失败: HTTP {response.status_code}"
except Exception as e:
return False, f"网络错误: {e}"
def check_status(self) -> Tuple[bool, str]:
"""查询 OTA 状态"""
try:
response = requests.get(f"{self.base_url}/ota/status", timeout=5)
if response.status_code == 200:
return True, response.text
return False, f"查询失败: HTTP {response.status_code}"
except Exception as e:
return False, f"网络错误: {e}"
def health_check(self) -> Tuple[bool, str]:
"""查询 ESP32 健康状态"""
try:
response = requests.get(f"{self.base_url}/health", timeout=5)
if response.status_code == 200:
return True, response.text
return False, f"查询失败: HTTP {response.status_code}"
except Exception as e:
return False, f"网络错误: {e}"
class OTAToolGUI:
"""OTA 上位机主界面"""
def __init__(self):
self.root = tk.Tk()
self.root.title(f"ESP32 OTA 上位机工具 - {TARGET_CHIP}")
self.root.geometry("900x700")
# ESP32 连接配置
self.esp_ip = tk.StringVar(value=ESP32_IP)
# OTA 客户端
self.ota_client = OTAClient()
# 日志接收器
self.log_handler: Optional[LogHandler] = None
# 当前的 HEX 文件
self.current_hex_file = None
self._create_widgets()
self._start_log_receiver()
def _create_widgets(self):
"""创建界面组件"""
# ----- 顶部：连接配置 -----
frame_conn = tk.LabelFrame(self.root, text="ESP32 连接配置", padx=10, pady=5)
frame_conn.pack(fill='x', padx=10, pady=5)
tk.Label(frame_conn, text="ESP32 IP:").grid(row=0, column=0, sticky='e')
tk.Entry(frame_conn, textvariable=self.esp_ip, width=20).grid(row=0, column=1, padx=5)
tk.Button(frame_conn, text="连接测试", command=self._test_connection).grid(row=0, column=2, padx=5)
tk.Button(frame_conn, text="健康检查", command=self._health_check).grid(row=0, column=3, padx=5)
# ----- 中部左侧：文件选择 -----
frame_file = tk.LabelFrame(self.root, text="固件文件", padx=10, pady=5)
frame_file.pack(fill='both', expand=True, side='left', padx=10, pady=5)
self.file_path_var = tk.StringVar(value="未选择文件")
tk.Label(frame_file, textvariable=self.file_path_var, fg='blue', wraplength=300).pack(fill='x')
tk.Button(frame_file, text="选择 .hex 文件", command=self._select_hex_file).pack(pady=5)
tk.Button(frame_file, text="选择 .bin 文件", command=self._select_bin_file).pack(pady=5)
# 文件信息
self.file_info_var = tk.StringVar(value="")
tk.Label(frame_file, textvariable=self.file_info_var, fg='gray').pack(fill='x')
# 上传按钮
self.upload_btn = tk.Button(frame_file, text="开始上传固件", bg='green', fg='white',
font=('Arial', 12, 'bold'), state='disabled',
command=self._start_upload)
self.upload_btn.pack(fill='x', pady=10)
# 进度条
self.progress_var = tk.DoubleVar(value=0)
self.progress_bar = ttk.Progressbar(frame_file, variable=self.progress_var, maximum=100)
self.progress_bar.pack(fill='x', pady=5)
self.progress_label = tk.Label(frame_file, text="", fg='gray')
self.progress_label.pack()
# 触发 OTA 中继
tk.Button(frame_file, text="触发 ESP32 OTA 中继", bg='orange',
command=self._trigger_relay).pack(fill='x', pady=5)
# ----- 中部右侧：调试日志 -----
frame_log = tk.LabelFrame(self.root, text="调试日志 (UDP:6666)", padx=5, pady=5)
frame_log.pack(fill='both', expand=True, side='right', padx=10, pady=5)
# 日志文本框
self.log_text = tk.Text(frame_log, height=25, width=70, font=('Courier New', 9))
self.log_text.pack(fill='both', expand=True)
# 配置日志颜色标签
self.log_text.tag_config('[E]', foreground='#FF4444')
self.log_text.tag_config('[W]', foreground='#FF8800')
self.log_text.tag_config('[I]', foreground='#0088FF')
self.log_text.tag_config('[D]', foreground='#888888')
# 滚动条
scrollbar = tk.Scrollbar(self.log_text, command=self.log_text.yview)
scrollbar.pack(side='right', fill='y')
self.log_text.config(yscrollcommand=scrollbar.set)
# 清除日志按钮
tk.Button(frame_log, text="清除日志", command=lambda: self.log_text.delete('1.0', 'end')).pack()
def _start_log_receiver(self):
"""启动 UDP 日志接收"""
self.log_handler = LogHandler(self.log_text, DEBUG_LOG_PORT)
self.log_handler.start()
self._log_info("上位机启动", "UDP 日志接收已启动，监听 0.0.0.0:6666")
def _test_connection(self):
"""测试与 ESP32 的连接"""
ip = self.esp_ip.get().strip()
if not ip:
messagebox.showerror("错误", "请输入 ESP32 的 IP 地址")
return
self.ota_client = OTAClient(esp_ip=ip)
ok, msg = self.ota_client.health_check()
if ok:
messagebox.showinfo("连接成功", f"ESP32 连接正常\n{msg}")
self._log_info("连接", f"ESP32 ({ip}) 连接成功")
else:
messagebox.showerror("连接失败", f"无法连接到 ESP32\n{msg}")
self._log_error("连接", f"ESP32 ({ip}) 连接失败: {msg}")
def _health_check(self):
"""查询 ESP32 健康状态"""
ip = self.esp_ip.get().strip()
self.ota_client = OTAClient(esp_ip=ip)
ok, msg = self.ota_client.health_check()
if ok:
self._log_info("健康检查", msg)
messagebox.showinfo("健康状态", msg)
else:
self._log_error("健康检查", f"查询失败: {msg}")
messagebox.showerror("健康检查", f"查询失败\n{msg}")
def _select_hex_file(self):
"""选择 HEX 文件"""
filepath = filedialog.askopenfilename(
title="选择 HEX 文件",
filetypes=[("HEX 文件", "*.hex"), ("所有文件", "*.*")]
)
if filepath:
self.current_hex_file = (filepath, 'hex')
self.file_path_var.set(os.path.basename(filepath))
self._log_info("文件", f"已选择: {filepath}")
# 解析文件信息
parser = IntelHexParser(filepath)
ok, msg = parser.parse()
if ok:
self.file_info_var.set(msg)
self.upload_btn.config(state='normal')
else:
self.file_info_var.set(f"解析失败: {msg}")
self.upload_btn.config(state='disabled')
def _select_bin_file(self):
"""选择 BIN 文件（简化支持）"""
filepath = filedialog.askopenfilename(
title="选择 BIN 文件",
filetypes=[("BIN 文件", "*.bin"), ("所有文件", "*.*")]
)
if filepath:
self.current_hex_file = (filepath, 'bin')
self.file_path_var.set(os.path.basename(filepath))
self._log_info("文件", f"已选择: {filepath}")
# 获取文件大小
size = os.path.getsize(filepath)
self.file_info_var.set(f"BIN 文件，大小: {size} 字节 ({size/1024:.1f} KB)")
self.upload_btn.config(state='normal')
def _start_upload(self):
"""开始上传固件"""
if not self.current_hex_file:
messagebox.showerror("错误", "请先选择固件文件")
return
filepath, file_type = self.current_hex_file
self.upload_btn.config(state='disabled', text="上传中...")
self.progress_var.set(0)
self.progress_label.config(text="正在连接 ESP32...")
# 在后台线程执行上传
threading.Thread(target=self._upload_worker, args=(filepath, file_type), daemon=True).start()
def _upload_worker(self, filepath: str, file_type: str):
"""上传工作线程"""
try:
if file_type == 'hex':
ok, msg = self._upload_hex(filepath)
else:
ok, msg = self._upload_bin(filepath)
if ok:
self.root.after(0, lambda: self.progress_label.config(text="上传完成"))
self.root.after(0, lambda: messagebox.showinfo("完成", msg))
self._log_info("上传", msg)
else:
self.root.after(0, lambda: self.progress_label.config(text="上传失败"))
self.root.after(0, lambda: messagebox.showerror("上传失败", msg))
self._log_error("上传", f"失败: {msg}")
except Exception as e:
self.root.after(0, lambda: self.progress_label.config(text="上传异常"))
self.root.after(0, lambda: messagebox.showerror("异常", f"上传异常: {e}\n{traceback.format_exc()}"))
self._log_error("上传", f"异常: {e}")
finally:
self.root.after(0, lambda: self.upload_btn.config(state='normal', text="开始上传固件"))
def _upload_hex(self, filepath: str) -> Tuple[bool, str]:
"""上传 HEX 文件"""
parser = IntelHexParser(filepath)
ok, msg = parser.parse()
if not ok:
return False, msg
total_size = parser.get_total_size()
addr = parser.min_addr
seq = 0
while addr <= parser.max_addr:
remaining = parser.max_addr - addr + 1
chunk = min(CHUNK_SIZE, remaining)
data = parser.get_data_segment(addr, chunk)
packet = struct.pack('<H', seq) + struct.pack('<I', total_size) + data
response = requests.post(
f"http://{self.esp_ip.get()}:{ESP32_PORT}/ota/upload",
data=packet,
timeout=10,
headers={'Content-Type': 'application/octet-stream'}
)
if response.status_code != 200:
return False, f"HTTP {response.status_code}: {response.text}"
sent = min(addr - parser.min_addr + chunk, total_size)
progress = (sent / total_size) * 100
self.root.after(0, lambda p=progress, s=sent, t=total_size:
[self.progress_var.set(p),
self.progress_label.config(text=f"上传中: {s}/{t} 字节 ({p:.1f}%)")])
addr += chunk
seq += 1
return True, f"HEX 文件上传完成，共 {seq} 包"
def _upload_bin(self, filepath: str) -> Tuple[bool, str]:
"""上传 BIN 文件（直接二进制）"""
file_size = os.path.getsize(filepath)
with open(filepath, 'rb') as f:
seq = 0
offset = 0
while offset < file_size:
chunk = f.read(CHUNK_SIZE)
packet = struct.pack('<H', seq) + struct.pack('<I', file_size) + chunk
response = requests.post(
f"http://{self.esp_ip.get()}:{ESP32_PORT}/ota/upload",
data=packet,
timeout=10,
headers={'Content-Type': 'application/octet-stream'}
)
if response.status_code != 200:
return False, f"HTTP {response.status_code}: {response.text}"
offset += len(chunk)
seq += 1
progress = (offset / file_size) * 100
self.root.after(0, lambda p=progress, s=offset, t=file_size:
[self.progress_var.set(p),
self.progress_label.config(text=f"上传中: {s}/{t} 字节 ({p:.1f}%)")])
return True, f"BIN 文件上传完成，共 {seq} 包"
def _trigger_relay(self):
"""触发 OTA 中继"""
ip = self.esp_ip.get().strip()
self.ota_client = OTAClient(esp_ip=ip)
ok, msg = self.ota_client.trigger_relay()
if ok:
self._log_info("触发", msg)
messagebox.showinfo("成功", msg)
else:
self._log_error("触发", f"失败: {msg}")
messagebox.showerror("失败", msg)
def _log_info(self, tag: str, msg: str):
"""记录信息日志"""
self._log(tag, msg, '[I]')
def _log_error(self, tag: str, msg: str):
"""记录错误日志"""
self._log(tag, msg, '[E]')
def _log(self, tag: str, msg: str, level: str):
"""追加日志到文本框"""
timestamp = datetime.now().strftime("%H:%M:%S")
log_line = f"[{timestamp}]{level}[{tag}] {msg}"
self.log_handler._append_log(log_line)
def run(self):
"""启动界面主循环"""
self.root.mainloop()
def cleanup(self):
"""清理资源"""
if self.log_handler:
self.log_handler.stop()
def main():
"""主函数"""
print(f"="*50)
print(f"ESP32 OTA 上位机工具")
print(f"ESP32 IP: {ESP32_IP}:{ESP32_PORT}")
print(f"日志监听: {DEBUG_LOG_IP}:{DEBUG_LOG_PORT}")
print(f"="*50)
app = OTAToolGUI()
try:
app.run()
finally:
app.cleanup()
if __name__ == '__main__':
main()
tools/ota_host_web.html
+113
View File
@@ -868,6 +868,46 @@
</div>
</div>
<!-- CAN 帧发送面板 -->
<div class="card">
<div class="card-header">
<div class="card-title">
<div class="icon">📨</div>
CAN 帧发送
</div>
</div>
<div style="display:grid; grid-template-columns:80px 1fr 80px 1fr; gap:12px; margin-bottom:16px;">
<div class="form-group" style="margin:0;">
<label style="font-size:11px;color:var(--text-muted);margin-bottom:6px;display:block;">通道</label>
<select id="canChannel" style="width:100%;background:var(--bg-secondary);border:1px solid var(--border);border-radius:8px;padding:10px 12px;color:var(--text-primary);font-size:13px;">
<option value="0">CH0</option>
<option value="1">CH1</option>
<option value="2">CH2</option>
</select>
</div>
<div class="form-group" style="margin:0;">
<label style="font-size:11px;color:var(--text-muted);margin-bottom:6px;display:block;">扩展帧 ID</label>
<input type="text" id="canId" value="0x18DA00F1" placeholder="例如: 0x18DA00F1" style="width:100%;background:var(--bg-secondary);border:1px solid var(--border);border-radius:8px;padding:10px 12px;color:var(--text-primary);font-size:13px;">
</div>
<div class="form-group" style="margin:0;">
<label style="font-size:11px;color:var(--text-muted);margin-bottom:6px;display:block;">DLC</label>
<input type="number" id="canDlc" value="8" min="1" max="64" style="width:100%;background:var(--bg-secondary);border:1px solid var(--border);border-radius:8px;padding:10px 12px;color:var(--text-primary);font-size:13px;">
</div>
<div class="form-group" style="margin:0;">
<label style="font-size:11px;color:var(--text-muted);margin-bottom:6px;display:block;">数据 (HEX)</label>
<input type="text" id="canData" value="0102030405060708" placeholder="01020304..." style="width:100%;background:var(--bg-secondary);border:1px solid var(--border);border-radius:8px;padding:10px 12px;color:var(--text-primary);font-size:13px;">
</div>
</div>
<div style="display:flex;gap:12px;">
<button class="btn btn-primary" style="flex:1;" onclick="sendCanFrame()">📤 发送</button>
<button class="btn btn-secondary" onclick="clearCanFields()">清空</button>
</div>
<div id="canResponse" style="margin-top:16px;background:var(--bg-secondary);border-radius:10px;padding:12px 16px;font-family:monospace;font-size:12px;display:none;">
<div style="color:var(--text-muted);margin-bottom:6px;">响应:</div>
<div id="canRespContent" style="color:var(--accent);"></div>
</div>
</div>
<!-- 调试日志 -->
<div class="card log-panel">
<div class="card-header">
@@ -1051,6 +1091,79 @@ async function healthCheck() {
}
}
// ==================== CAN 帧发送 ====================
async function sendCanFrame() {
const ch = parseInt(document.getElementById('canChannel').value);
const idStr = document.getElementById('canId').value.trim();
const dlc = parseInt(document.getElementById('canDlc').value) || 8;
const dataStr = document.getElementById('canData').value.trim();
// 解析 ID (支持 0x 前缀)
let extId = 0;
if (idStr.startsWith('0x') || idStr.startsWith('0X')) {
extId = parseInt(idStr, 16);
} else {
extId = parseInt(idStr, 10);
if (isNaN(extId)) extId = parseInt(idStr, 16);
}
if (isNaN(extId)) {
showToast('ID 格式错误', 'error');
return;
}
// 解析 data (hex 字符串)
let dataHex = dataStr.replace(/\s/g, '');
if (dataHex.length % 2 !== 0) {
showToast('数据长度必须为偶数', 'error');
return;
}
addLog('I', 'CAN', `发送 CH${ch} ID=0x${extId.toString(16).toUpperCase()} DLC=${dlc}`);
try {
const response = await fetch(`${getBaseUrl()}/can/send`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
ch: ch,
id: extId,
dlc: dlc,
data: dataHex
}),
signal: AbortSignal.timeout(5000)
});
const result = await response.json();
if (result.status === 'ok') {
addLog('I', 'CAN', `响应: ${result.recv || '(无数据)'} (${result.recv_len || 0}B)`);
const respDiv = document.getElementById('canResponse');
const respContent = document.getElementById('canRespContent');
respDiv.style.display = 'block';
respContent.textContent = result.recv
? `0x${result.recv} (${result.recv_len} 字节)`
: '无响应数据';
showToast(`CH${ch} 发送成功`, 'success');
} else {
addLog('E', 'CAN', `发送失败: code=${result.code}`);
showToast(`发送失败: code=${result.code}`, 'error');
}
} catch (e) {
addLog('E', 'CAN', `发送失败: ${e.message}`);
showToast('发送失败', 'error');
}
}
function clearCanFields() {
document.getElementById('canId').value = '0x18DA00F1';
document.getElementById('canDlc').value = '8';
document.getElementById('canData').value = '0102030405060708';
document.getElementById('canResponse').style.display = 'none';
}
// ==================== 文件处理 ====================
function handleFileSelect(event) {
const file = event.target.files[0];