1284 lines
50 KiB
Python
1284 lines
50 KiB
Python
import sys
|
||
import threading
|
||
import time
|
||
import tkinter
|
||
import tkinter.messagebox
|
||
import tkinter.filedialog
|
||
import configparser
|
||
|
||
import os
|
||
import binascii
|
||
import ctypes
|
||
|
||
# 读取Dll动态链接库：
|
||
from ctypes import *
|
||
|
||
def resource_dir():
|
||
"""资源目录: PyInstaller打包后DLL解压于_MEIPASS, 源码模式为脚本目录"""
|
||
if getattr(sys, 'frozen', False):
|
||
return sys._MEIPASS
|
||
return os.path.dirname(os.path.abspath(__file__))
|
||
|
||
def app_dir():
|
||
"""配置目录: 打包后config.ini持久化于exe同目录, 源码模式为脚本目录"""
|
||
if getattr(sys, 'frozen', False):
|
||
return os.path.dirname(sys.executable)
|
||
return os.path.dirname(os.path.abspath(__file__))
|
||
|
||
# 依赖的DLL文件(打包时内嵌, 运行时解压到_MEIPASS)
|
||
CAN_DLL_PATH = os.path.join(resource_dir(), 'ControlCAN.dll')
|
||
# 读取DLL文件
|
||
Can_DLL = windll.LoadLibrary(CAN_DLL_PATH)
|
||
|
||
class ToolTip(object):
|
||
def __init__(self, widget):
|
||
self.widget = widget
|
||
self.tip_window = None
|
||
|
||
def show_tip(self, tip_text):
|
||
"Display text in a tooltip window"
|
||
if self.tip_window or not tip_text:
|
||
return
|
||
x, y, _cx, cy = self.widget.bbox("insert")
|
||
# get size of widget
|
||
x = x + self.widget.winfo_rootx() + 250
|
||
# calculate to display tooptip
|
||
y = y + cy + self.widget.winfo_rooty() + 0
|
||
# below and to the right
|
||
self.tip_window = tw = tkinter.Toplevel(self.widget)
|
||
# create new tooltip window
|
||
tw.wm_overrideredirect(True)
|
||
# remove all Window Manager (wm)
|
||
tw.wm_geometry("+%d+%d" % (x, y))
|
||
# create window size
|
||
|
||
label = tkinter.Label(tw, text=tip_text, justify=tkinter.LEFT,
|
||
background="#ffffe0", relief=tkinter.SOLID,
|
||
borderwidth=1, font=("微软雅黑", 10))
|
||
label.pack(ipadx=1)
|
||
|
||
def hide_tip(self):
|
||
tw = self.tip_window
|
||
self.tip_window = None
|
||
if tw:
|
||
tw.destroy()
|
||
|
||
def create_ToolTip(widget, text):
|
||
tooltip = ToolTip(widget)
|
||
def enter(event):
|
||
tooltip.show_tip(text)
|
||
def leave(event):
|
||
tooltip.hide_tip()
|
||
widget.bind('<Enter>', enter)
|
||
widget.bind('<Leave>', leave)
|
||
|
||
def Show_Update_log():
|
||
LOG_str = "[LogStart] " + "20220327_BETA1.0: 新规作成，基本CAN通信功能实现\n"
|
||
LOG_str = LOG_str + "[LogStart] " + "20220409_BETA1.1: 接收改用线程\n"
|
||
LOG_str = LOG_str + "[LogStart] " + "20220410_BETA1.2: UDS框架接入\n"
|
||
LOG_str = LOG_str + "[LogStart] " + "20260807_BETA2.0: UI重构(可配置CAN ID/按钮集成/文件选择)\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
|
||
# ==================== 配置管理 ====================
|
||
|
||
CONFIG_PATH = os.path.join(app_dir(), 'config.ini')
|
||
|
||
def load_config():
|
||
"""读取配置文件, 不存在则返回默认配置"""
|
||
config = configparser.ConfigParser()
|
||
config.read(CONFIG_PATH, encoding='utf-8')
|
||
if not config.has_section('CAN'):
|
||
config['CAN'] = {'tx_id': '0x180', 'rx_id': '0x181'}
|
||
if not config.has_section('UI'):
|
||
config['UI'] = {'window_width': '900', 'window_height': '600'}
|
||
if not config.has_section('Last'):
|
||
config['Last'] = {'bin_path': ''}
|
||
return config
|
||
|
||
def save_config(config):
|
||
"""保存配置到文件"""
|
||
try:
|
||
with open(CONFIG_PATH, 'w', encoding='utf-8') as f:
|
||
config.write(f)
|
||
except Exception as e:
|
||
print("保存配置失败: %s" % str(e))
|
||
|
||
def parse_hex(s):
|
||
"""解析十六进制字符串(支持0x前缀/十进制)"""
|
||
s = s.strip()
|
||
if s.lower().startswith('0x'):
|
||
return int(s, 16)
|
||
return int(s)
|
||
|
||
def apply_can_id():
|
||
"""应用CAN ID设置"""
|
||
global TRANSMIT_ID, RECEIVE_ID
|
||
try:
|
||
new_tx = parse_hex(tx_id_var.get())
|
||
new_rx = parse_hex(rx_id_var.get())
|
||
if not (0 <= new_tx <= 0x7FF and 0 <= new_rx <= 0x7FF):
|
||
raise ValueError("ID超出标准帧范围(0-0x7FF)")
|
||
TRANSMIT_ID = new_tx
|
||
RECEIVE_ID = new_rx
|
||
# 保存到配置文件
|
||
config = load_config()
|
||
config['CAN']['tx_id'] = hex(new_tx)
|
||
config['CAN']['rx_id'] = hex(new_rx)
|
||
save_config(config)
|
||
log_append("CAN ID已更新: TX=0x%03X RX=0x%03X" % (TRANSMIT_ID, RECEIVE_ID), "iap")
|
||
update_status()
|
||
except Exception as e:
|
||
log_append("CAN ID格式错误: %s" % str(e), "error")
|
||
|
||
def log_append(msg, level="normal"):
|
||
"""日志追加到界面 (level: normal/iap/error)"""
|
||
color = {"normal": "lime", "iap": "cyan", "error": "red"}.get(level, "lime")
|
||
prefix = {"normal": "", "iap": "[IAP]", "error": "[ERR]"}.get(level, "")
|
||
textLOG.tag_config(level, foreground=color)
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "] " \
|
||
+ prefix + " " + msg + "\n"
|
||
textLOG.insert(tkinter.END, LOG_str, (level,))
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update_idletasks()
|
||
|
||
g_tx_count = 0
|
||
g_rx_count = 0
|
||
g_can_connected = False
|
||
|
||
def update_status():
|
||
"""更新状态栏"""
|
||
status_can_label.config(text="● CAN已连接" if g_can_connected else "○ CAN未连接",
|
||
fg="green" if g_can_connected else "red")
|
||
status_tx_label.config(text="TX: %d" % g_tx_count)
|
||
status_rx_label.config(text="RX: %d" % g_rx_count)
|
||
|
||
def browse_bin():
|
||
"""文件选择器: 选择BIN固件文件"""
|
||
filepath = tkinter.filedialog.askopenfilename(
|
||
title="选择BIN固件文件",
|
||
filetypes=[("BIN文件", "*.bin"), ("HEX文件", "*.hex"), ("所有文件", "*.*")])
|
||
if filepath:
|
||
BIN_path_entry.set(filepath)
|
||
# 保存路径到配置
|
||
config = load_config()
|
||
config['Last']['bin_path'] = filepath
|
||
save_config(config)
|
||
|
||
def g_WindowStart():
|
||
global win #界面实体
|
||
global CAN_Send_entry #发送数据的内容
|
||
global BIN_path_entry #BIN文件路径
|
||
global textLOG #textbox的数据
|
||
global scroll #滚动条实体
|
||
global thread
|
||
global thread2
|
||
global tx_id_var, rx_id_var
|
||
global status_can_label, status_tx_label, status_rx_label
|
||
global g_can_connected
|
||
|
||
# 读取配置
|
||
config = load_config()
|
||
try:
|
||
tx_id_default = config['CAN']['tx_id']
|
||
rx_id_default = config['CAN']['rx_id']
|
||
except KeyError:
|
||
tx_id_default, rx_id_default = '0x180', '0x181'
|
||
|
||
# 创建主窗口
|
||
win = tkinter.Tk()
|
||
win.title("BMS IAP 升级工具 v2.0 (CAN-UDS)")
|
||
win.geometry("%sx%s" % (config['UI'].get('window_width', '900'),
|
||
config['UI'].get('window_height', '600')))
|
||
win.minsize(800, 500)
|
||
win.columnconfigure(0, weight=1)
|
||
win.rowconfigure(4, weight=1) # 日志区随窗口拉伸
|
||
|
||
# ===== 配置区: CAN ID =====
|
||
config_frame = tkinter.LabelFrame(win, text=" CAN 配置 ", font=("微软雅黑", 10))
|
||
config_frame.grid(row=0, column=0, sticky="ew", padx=10, pady=5)
|
||
|
||
tkinter.Label(config_frame, text="CAN TX ID:", font=("微软雅黑", 10)).pack(side="left", padx=(10, 2))
|
||
tx_id_var = tkinter.StringVar(value=tx_id_default)
|
||
tkinter.Entry(config_frame, textvariable=tx_id_var, width=8, font=("Consolas", 11)).pack(side="left", padx=2)
|
||
|
||
tkinter.Label(config_frame, text="CAN RX ID:", font=("微软雅黑", 10)).pack(side="left", padx=(15, 2))
|
||
rx_id_var = tkinter.StringVar(value=rx_id_default)
|
||
tkinter.Entry(config_frame, textvariable=rx_id_var, width=8, font=("Consolas", 11)).pack(side="left", padx=2)
|
||
|
||
tkinter.Button(config_frame, text="应用", command=apply_can_id,
|
||
bg="lightblue", font=("微软雅黑", 10)).pack(side="left", padx=(15, 10))
|
||
|
||
tkinter.Label(config_frame, text="波特率: 500kbps", font=("微软雅黑", 9),
|
||
fg="gray").pack(side="right", padx=10)
|
||
|
||
# ===== 指令按钮区 =====
|
||
cmd_frame = tkinter.LabelFrame(win, text=" UDS 快捷指令 ", font=("微软雅黑", 10))
|
||
cmd_frame.grid(row=1, column=0, sticky="ew", padx=10, pady=5)
|
||
|
||
cmd_buttons = [
|
||
("会话控制", cmd_session, "0x10 0x02 进入编程会话"),
|
||
("读版本", cmd_read_version, "0x22 0xF1 0x00 读取Bootloader版本"),
|
||
("请求Seed", cmd_request_seed, "0x27 0x01 获取安全访问Seed"),
|
||
("发送Key", cmd_send_key, "0x27 0x02 发送Key(~Seed)解锁"),
|
||
("擦除Flash", cmd_erase, "0x31 0x01 0xFF 0x01 擦除App区"),
|
||
("请求下载", cmd_download, "0x34 + 固件大小 请求下载"),
|
||
("开始传输", cmd_transfer_start, "0x36 0x01 进入纯数据模式"),
|
||
("传输退出", cmd_transfer_exit, "0x37 结束传输"),
|
||
("CRC校验", cmd_crc_check, "0x31 0x01 0xFF 0x02 触发CRC校验"),
|
||
("ECU复位", cmd_ecu_reset, "0x11 0x01 复位跳转App"),
|
||
]
|
||
for i, (text, cmd, tip) in enumerate(cmd_buttons):
|
||
btn = tkinter.Button(cmd_frame, text=text, command=cmd,
|
||
font=("微软雅黑", 10), width=11,
|
||
bg="#e8e8e8", relief="raised")
|
||
btn.grid(row=i // 5, column=i % 5, padx=4, pady=3, sticky="ew")
|
||
create_ToolTip(btn, tip)
|
||
for col in range(5):
|
||
cmd_frame.columnconfigure(col, weight=1)
|
||
|
||
# ===== 手动指令区 =====
|
||
manual_frame = tkinter.LabelFrame(win, text=" 手动指令下发 ", font=("微软雅黑", 10))
|
||
manual_frame.grid(row=2, column=0, sticky="ew", padx=10, pady=5)
|
||
|
||
CAN_Send_entry = tkinter.StringVar()
|
||
CAN_Send_entry.set("0x10 0x02 0x00 0x00 0x00 0x00 0x00 0x00")
|
||
entry_cmd = tkinter.Entry(manual_frame, textvariable=CAN_Send_entry,
|
||
font=("Consolas", 12), width=40)
|
||
entry_cmd.pack(side="left", padx=10, pady=4, fill="x", expand=True)
|
||
create_ToolTip(entry_cmd, 'CAN指令8位(十六进制), 字节间用半角空格隔开')
|
||
tkinter.Button(manual_frame, text="发送指令", command=cmd_manual_send,
|
||
bg="pink", fg="red", font=("微软雅黑", 11),
|
||
width=10).pack(side="left", padx=10)
|
||
|
||
# ===== 固件区 =====
|
||
fw_frame = tkinter.LabelFrame(win, text=" BIN 固件 ", font=("微软雅黑", 10))
|
||
fw_frame.grid(row=3, column=0, sticky="ew", padx=10, pady=5)
|
||
|
||
BIN_path_entry = tkinter.StringVar()
|
||
BIN_path_entry.set(config['Last'].get('bin_path', ''))
|
||
entry_bin = tkinter.Entry(fw_frame, textvariable=BIN_path_entry,
|
||
font=("Consolas", 11), width=40)
|
||
entry_bin.pack(side="left", padx=10, pady=4, fill="x", expand=True)
|
||
tkinter.Button(fw_frame, text="浏览...", command=browse_bin,
|
||
font=("微软雅黑", 10), width=8).pack(side="left", padx=5)
|
||
tkinter.Button(fw_frame, text="IAP一键升级", command=cmd_iap_upgrade,
|
||
bg="palegreen", fg="darkgreen", font=("微软雅黑", 12),
|
||
width=14).pack(side="left", padx=10)
|
||
|
||
# ===== 日志区 (可拉伸) =====
|
||
log_frame = tkinter.LabelFrame(win, text=" 日志 ", font=("微软雅黑", 10))
|
||
log_frame.grid(row=4, column=0, sticky="nsew", padx=10, pady=5)
|
||
log_frame.rowconfigure(0, weight=1)
|
||
log_frame.columnconfigure(0, weight=1)
|
||
|
||
textLOG = tkinter.Text(log_frame, bg="black", fg="lime",
|
||
font=("Consolas", 11), state="disabled", wrap="word")
|
||
scroll = tkinter.Scrollbar(log_frame, orient="vertical", command=textLOG.yview)
|
||
textLOG.configure(yscrollcommand=scroll.set)
|
||
textLOG.grid(row=0, column=0, sticky="nsew", padx=(5, 0), pady=5)
|
||
scroll.grid(row=0, column=1, sticky="ns", pady=5)
|
||
textLOG.tag_config("normal", foreground="lime")
|
||
textLOG.tag_config("iap", foreground="cyan")
|
||
textLOG.tag_config("error", foreground="red")
|
||
|
||
# 绑定鼠标滚轮
|
||
def on_mousewheel(event):
|
||
textLOG.yview_scroll(int(-1 * (event.delta / 120)), "units")
|
||
textLOG.bind("<MouseWheel>", on_mousewheel)
|
||
|
||
# ===== 状态栏 =====
|
||
status_frame = tkinter.Frame(win)
|
||
status_frame.grid(row=5, column=0, sticky="ew", padx=10, pady=2)
|
||
status_can_label = tkinter.Label(status_frame, text="○ CAN未连接", fg="red",
|
||
font=("微软雅黑", 9))
|
||
status_can_label.pack(side="left", padx=10)
|
||
status_tx_label = tkinter.Label(status_frame, text="TX: 0", font=("微软雅黑", 9))
|
||
status_tx_label.pack(side="left", padx=10)
|
||
status_rx_label = tkinter.Label(status_frame, text="RX: 0", font=("微软雅黑", 9))
|
||
status_rx_label.pack(side="left", padx=10)
|
||
|
||
# 启动日志
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[PrgStart] BMS IAP 升级工具 v2.0\n"
|
||
textLOG.insert(tkinter.END, LOG_str)
|
||
Show_Update_log()
|
||
textLOG.config(state="disabled")
|
||
log_append("程序启动")
|
||
|
||
# 启动CAN
|
||
ret = connect()
|
||
g_can_connected = (ret == STATUS_OK)
|
||
CAN_Start()
|
||
update_status()
|
||
win.update()
|
||
|
||
# 线程接收
|
||
thread = threading.Thread(target=receive_by_thread, args=(CAN_INDEX_1,), daemon=True)
|
||
thread.start()
|
||
print(thread.is_alive())
|
||
print("threadREV.start()")
|
||
|
||
# 窗口关闭时保存配置
|
||
def on_close():
|
||
config = load_config()
|
||
config['UI']['window_width'] = str(win.winfo_width())
|
||
config['UI']['window_height'] = str(win.winfo_height())
|
||
config['Last']['bin_path'] = BIN_path_entry.get()
|
||
save_config(config)
|
||
win.destroy()
|
||
win.protocol("WM_DELETE_WINDOW", on_close)
|
||
|
||
# 进入消息循环，可以写控件
|
||
win.mainloop()
|
||
|
||
|
||
# ==================== 指令处理 ====================
|
||
|
||
def cmd_manual_send():
|
||
"""手动发送8位CAN指令"""
|
||
Sendstr = CAN_Send_entry.get()
|
||
Send_BUFF = str(Sendstr).strip().split(" ")
|
||
if len(Send_BUFF) < 1:
|
||
log_append("指令为空", "error")
|
||
return
|
||
Send_BUFFHEX = [0, 0, 0, 0, 0, 0, 0, 0]
|
||
try:
|
||
for i in range(0, len(Send_BUFF)):
|
||
if Send_BUFF[i] == "":
|
||
continue
|
||
Send_BUFFHEX[i] = int(str(Send_BUFF[i]), 16)
|
||
except Exception as e:
|
||
log_append("指令格式错误: %s" % str(e), "error")
|
||
return
|
||
transmitBIN(CAN_INDEX_1, Send_BUFFHEX, Sendstr)
|
||
win.update()
|
||
|
||
|
||
def cmd_iap_upgrade():
|
||
"""一键IAP升级"""
|
||
SendPath = str(BIN_path_entry.get()).strip()
|
||
if not SendPath:
|
||
log_append("请先选择BIN固件文件", "error")
|
||
return
|
||
if not os.path.exists(SendPath):
|
||
log_append("文件不存在: %s" % SendPath, "error")
|
||
return
|
||
IAP_Upgrade(SendPath)
|
||
print("IAP upgrade done")
|
||
|
||
|
||
# ==================== UDS 快捷指令 ====================
|
||
|
||
def cmd_session():
|
||
cmd_manual_send_set([0x10, 0x02, 0, 0, 0, 0, 0, 0], "0x10 0x02 进入编程会话")
|
||
|
||
def cmd_read_version():
|
||
cmd_manual_send_set([0x22, 0xF1, 0x00, 0, 0, 0, 0, 0], "0x22 0xF1 0x00 读版本")
|
||
|
||
def cmd_request_seed():
|
||
cmd_manual_send_set([0x27, 0x01, 0, 0, 0, 0, 0, 0], "0x27 0x01 请求Seed")
|
||
|
||
def cmd_send_key():
|
||
cmd_manual_send_set([0x27, 0x02, 0xED, 0xCB, 0xA9, 0x87, 0, 0], "0x27 0x02 发送Key")
|
||
|
||
def cmd_erase():
|
||
cmd_manual_send_set([0x31, 0x01, 0xFF, 0x01, 0, 0, 0, 0], "0x31 0x01 0xFF 0x01 擦除Flash")
|
||
|
||
def cmd_download():
|
||
"""请求下载: 0x34 + 固件大小(4字节大端)"""
|
||
fw_path = str(BIN_path_entry.get()).strip()
|
||
if not fw_path or not os.path.exists(fw_path):
|
||
log_append("请先选择BIN固件文件", "error")
|
||
return
|
||
fw_size = os.path.getsize(fw_path)
|
||
if not (0 < fw_size <= 36 * 1024):
|
||
log_append("固件大小超限(1-36864字节)", "error")
|
||
return
|
||
size_bytes = list(fw_size.to_bytes(4, 'big'))
|
||
data = [0x34] + size_bytes
|
||
desc = "0x34 固件大小=%d字节" % fw_size
|
||
cmd_manual_send_set(data, desc)
|
||
|
||
def cmd_transfer_start():
|
||
cmd_manual_send_set([0x36, 0x01, 0, 0, 0, 0, 0, 0], "0x36 0x01 开始传输")
|
||
|
||
def cmd_transfer_exit():
|
||
cmd_manual_send_set([0x37, 0, 0, 0, 0, 0, 0, 0], "0x37 传输退出")
|
||
|
||
def cmd_crc_check():
|
||
cmd_manual_send_set([0x31, 0x01, 0xFF, 0x02, 0, 0, 0, 0], "0x31 0x01 0xFF 0x02 CRC校验")
|
||
|
||
def cmd_ecu_reset():
|
||
cmd_manual_send_set([0x11, 0x01, 0, 0, 0, 0, 0, 0], "0x11 0x01 ECU复位")
|
||
|
||
def cmd_manual_send_set(data, desc):
|
||
"""发送一组指令(填充到8字节)"""
|
||
while len(data) < 8:
|
||
data.append(0)
|
||
Sendstr = " ".join("0x%02X" % b for b in data)
|
||
transmitBIN(CAN_INDEX_1, data, Sendstr + " (" + desc + ")")
|
||
win.update()
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# VCI_OpenDevice 打开设备：
|
||
# CAN卡类别为 USBCAN-2A, USBCAN-2C, CANalyst-II
|
||
VCI_USB_CAN_2 = 4
|
||
# CAN卡下标索引, 比如当只有一个USB-CAN适配器时, 索引号为0, 这时再插入一个USB-CAN适配器那么后面插入的这个设备索引号就是1, 以此类推
|
||
DEV_INDEX = 0
|
||
# 打开设备, 一个设备只能打开一次
|
||
STATUS_OK = 1
|
||
# return: 1=OK 0=ERROR
|
||
def connect():
|
||
# VCI_USB_CAN_2: 设备类型
|
||
# DEV_INDEX: 设备索引
|
||
# RESERVED: 保留参数
|
||
ret = Can_DLL.VCI_OpenDevice(VCI_USB_CAN_2, DEV_INDEX, RESERVED)
|
||
if ret == STATUS_OK:
|
||
print('VCI_OpenDevice: 设备开启成功')
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "] " + "VCI_OpenDevice: 设备开启成功\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
else:
|
||
print('VCI_OpenDevice: 设备开启失败')
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "] " + "VCI_OpenDevice: 设备开启失败\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
return ret
|
||
|
||
|
||
#VCI_InitCAN 初始化指定CAN通道：
|
||
# 通道初始化参数结构
|
||
# AccCode: 过滤验收码
|
||
# AccMask: 过滤屏蔽码
|
||
# Reserved: 保留字段
|
||
# Filter: 滤波模式 0/1=接收所有类型 2=只接收标准帧 3=只接收扩展帧
|
||
# Timing0: 波特率 T0
|
||
# Timing1: 波特率 T1
|
||
# Mode: 工作模式 0=正常工作 1=仅监听模式 2=自发自收测试模式
|
||
class VCI_CAN_INIT_CONFIG(Structure):
|
||
_fields_ = [
|
||
("AccCode", c_uint),
|
||
("AccMask", c_uint),
|
||
("Reserved", c_uint),
|
||
("Filter", c_ubyte),
|
||
("Timing0", c_ubyte),
|
||
("Timing1", c_ubyte),
|
||
("Mode", c_ubyte)
|
||
]
|
||
# 过滤验收码
|
||
ACC_CODE = 0x80000000
|
||
# 过滤屏蔽码
|
||
ACC_MASK = 0xFFFFFFFF
|
||
# 保留字段
|
||
RESERVED = 0
|
||
# 滤波模式 0/1=接收所有类型
|
||
FILTER = 0
|
||
# 波特率 T0
|
||
TIMING_0 = 0x00 #500k
|
||
# 波特率 T1
|
||
TIMING_1 = 0x1C
|
||
# 工作模式 0=正常工作
|
||
MODE = 0
|
||
# 初始化通道
|
||
# return: 1=OK 0=ERROR
|
||
def init(can_index):
|
||
init_config = VCI_CAN_INIT_CONFIG(ACC_CODE, ACC_MASK, RESERVED, FILTER, TIMING_0, TIMING_1, MODE)
|
||
# VCI_USB_CAN_2: 设备类型
|
||
# DEV_INDEX: 设备索引
|
||
# can_index: CAN通道索引
|
||
# init_config: 请求参数体
|
||
ret = Can_DLL.VCI_InitCAN(VCI_USB_CAN_2, DEV_INDEX, can_index, byref(init_config))
|
||
if ret == STATUS_OK:
|
||
print('VCI_InitCAN: 通道 ' + str(can_index + 1) + ' 初始化成功')
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "] " + "VCI_InitCAN: 通道 " \
|
||
+ str(can_index + 1) + " 初始化成功\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
else:
|
||
print('VCI_InitCAN: 通道 ' + str(can_index + 1) + ' 初始化失败')
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "] " + "VCI_InitCAN: 通道 "\
|
||
+ str(can_index + 1) + " 初始化失败\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
return ret
|
||
|
||
|
||
# 打开通道 VCI_StartCAN 打开指定CAN通道：
|
||
# return: 1=OK 0=ERROR
|
||
def start(can_index):
|
||
# VCI_USB_CAN_2: 设备类型
|
||
# DEV_INDEX: 设备索引
|
||
# can_index: CAN通道索引
|
||
ret = Can_DLL.VCI_StartCAN(VCI_USB_CAN_2, DEV_INDEX, can_index)
|
||
if ret == STATUS_OK:
|
||
print('VCI_StartCAN: 通道 ' + str(can_index + 1) + ' 打开成功')
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "] " + "VCI_StartCAN: 通道 " \
|
||
+ str(can_index + 1) + " 打开成功\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
else:
|
||
print('VCI_StartCAN: 通道 ' + str(can_index + 1) + ' 打开失败')
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "] " + "VCI_StartCAN: 通道 " \
|
||
+ str(can_index + 1) + " 打开失败\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
return ret
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# CAN帧结构体 VCI_Transmit 发送数据：
|
||
# ID: 帧ID, 32位变量, 数据格式为靠右对齐
|
||
# TimeStamp: 设备接收到某一帧的时间标识, 时间标示从CAN卡上电开始计时, 计时单位为0.1ms
|
||
# TimeFlag: 是否使用时间标识, 为1时TimeStamp有效, TimeFlag和TimeStamp只在此帧为接收帧时才有意义
|
||
# SendType: 发送帧类型 0=正常发送(发送失败会自动重发, 重发时间为4秒, 4秒内没有发出则取消) 1=单次发送(只发送一次, 发送失败不会自动重发, 总线只产生一帧数据)[二次开发, 建议1, 提高发送的响应速度]
|
||
# RemoteFlag: 是否是远程帧 0=数据帧 1=远程帧(数据段空)
|
||
# ExternFlag: 是否是扩展帧 0=标准帧(11位ID) 1=扩展帧(29位ID)
|
||
# DataLen: 数据长度DLC(<=8), 即CAN帧Data有几个字节, 约束了后面Data[8]中的有效字节
|
||
# Data: CAN帧的数据, 由于CAN规定了最大是8个字节, 所以这里预留了8个字节的空间, 受DataLen约束, 如DataLen定义为3, 即Data[0]、Data[1]、Data[2]是有效的
|
||
# Reserved: 保留字段
|
||
class VCI_CAN_OBJ(Structure):
|
||
_fields_ = [
|
||
("ID", c_uint),
|
||
("TimeStamp", c_uint),
|
||
("TimeFlag", c_ubyte),
|
||
("SendType", c_ubyte),
|
||
("RemoteFlag", c_ubyte),
|
||
("ExternFlag", c_ubyte),
|
||
("DataLen", c_ubyte),
|
||
("Data", c_ubyte * 8),
|
||
("Reserved", c_ubyte * 3)
|
||
]
|
||
# 要发送的参数
|
||
TRANSMIT_DATA = 5
|
||
# 保留字段
|
||
RESERVED = 0
|
||
# 发送帧ID (上位机→板子)
|
||
TRANSMIT_ID = 0x180
|
||
# 接收帧ID (板子→上位机)
|
||
RECEIVE_ID = 0x181
|
||
# 时间标识
|
||
TIME_STAMP = 0
|
||
# 是否使用时间标识
|
||
TIME_FLAG = 0
|
||
# 发送帧类型
|
||
TRANSMIT_SEND_TYPE = 1
|
||
# 接收帧类型
|
||
RECEIVE_SEND_TYPE = 0
|
||
# 是否是远程帧
|
||
REMOTE_FLAG = 0
|
||
# 是否是扩展帧
|
||
EXTERN_FLAG = 0
|
||
# 数据长度DLC
|
||
DATA_LEN = 8
|
||
# 用来接收的帧结构体数组的长度, 适配器中为每个通道设置了2000帧左右的接收缓存区
|
||
RECEIVE_LEN = 2500
|
||
# 接收保留字段
|
||
WAIT_TIME = 0
|
||
# 要发送的参数
|
||
TRANSMIT_DATA = 5
|
||
# 要发送的帧结构体数组的长度(发送的帧数量), 最大为1000, 建议设为1, 每次发送单帧, 以提高发送效率
|
||
TRANSMIT_LEN = 1
|
||
# 发送数据
|
||
# return: 1=OK 0=ERROR
|
||
def transmitBIN(can_index, BUFF, Sendstr):
|
||
ubyte_array_8 = c_ubyte * 8
|
||
DATA = ubyte_array_8(BUFF[0], BUFF[1], BUFF[2], BUFF[3], BUFF[4], BUFF[5],
|
||
BUFF[6], BUFF[7])
|
||
ubyte_array_3 = c_ubyte * 3
|
||
RESERVED_3 = ubyte_array_3(RESERVED, RESERVED, RESERVED)
|
||
can_obj = VCI_CAN_OBJ(TRANSMIT_ID, TIME_STAMP, TIME_FLAG, TRANSMIT_SEND_TYPE, REMOTE_FLAG, EXTERN_FLAG, DATA_LEN,
|
||
DATA, RESERVED_3)
|
||
# VCI_USB_CAN_2: 设备类型
|
||
# DEV_INDEX: 设备索引
|
||
# can_index: CAN通道索引
|
||
# can_obj: 请求参数体
|
||
# TRANSMIT_LEN: 发送的帧数量
|
||
ret = Can_DLL.VCI_Transmit(VCI_USB_CAN_2, DEV_INDEX, can_index, byref(can_obj), TRANSMIT_LEN)
|
||
if ret == STATUS_OK:
|
||
global g_tx_count
|
||
g_tx_count += 1
|
||
update_status()
|
||
print('VCI_Transmit: 通道 ' + str(can_index + 1) + ' 发送数据成功')
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "] " + "VCI_Transmit: 通道 " \
|
||
+ str(can_index + 1) + "发送数据成功\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
LOG_str = "CAN指令8位:" + Sendstr + "\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
else:
|
||
print('VCI_Transmit: 通道 ' + str(can_index + 1) + ' 发送数据失败')
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "] " + "VCI_Transmit: 通道 " \
|
||
+ str(can_index + 1) + " 发送数据失败\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
LOG_str = "CAN指令8位:" + Sendstr + "\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
|
||
def transmit(can_index):
|
||
ubyte_array_8 = c_ubyte * 8
|
||
DATA = ubyte_array_8(TRANSMIT_DATA, TRANSMIT_DATA, TRANSMIT_DATA, TRANSMIT_DATA, TRANSMIT_DATA, TRANSMIT_DATA,
|
||
TRANSMIT_DATA, TRANSMIT_DATA)
|
||
ubyte_array_3 = c_ubyte * 3
|
||
RESERVED_3 = ubyte_array_3(RESERVED, RESERVED, RESERVED)
|
||
can_obj = VCI_CAN_OBJ(TRANSMIT_ID, TIME_STAMP, TIME_FLAG, TRANSMIT_SEND_TYPE, REMOTE_FLAG, EXTERN_FLAG, DATA_LEN,
|
||
DATA, RESERVED_3)
|
||
# VCI_USB_CAN_2: 设备类型
|
||
# DEV_INDEX: 设备索引
|
||
# can_index: CAN通道索引
|
||
# can_obj: 请求参数体
|
||
# TRANSMIT_LEN: 发送的帧数量
|
||
ret = Can_DLL.VCI_Transmit(VCI_USB_CAN_2, DEV_INDEX, can_index, byref(can_obj), TRANSMIT_LEN)
|
||
if ret == STATUS_OK:
|
||
print('VCI_Transmit: 通道 ' + str(can_index + 1) + ' 发送数据成功')
|
||
else:
|
||
print('VCI_Transmit: 通道 ' + str(can_index + 1) + ' 发送数据失败')
|
||
|
||
|
||
def receive_by_thread(can_index):
|
||
t_ret = 0
|
||
t_end = 1
|
||
|
||
while(t_end):
|
||
time.sleep(0.001)
|
||
# IAP升级模式: 暂停异步接收, 由send_and_wait同步处理
|
||
if g_iap_mode:
|
||
continue
|
||
try:
|
||
# win.update()
|
||
t_ret = receive(CAN_INDEX_1)
|
||
except:
|
||
print("receive_by_thread end")
|
||
t_end = 0
|
||
#返回值不使用
|
||
|
||
|
||
# 接收数据 VCI_Receive 接收数据：
|
||
# return: 1=OK 0=ERROR
|
||
def receive(can_index):
|
||
t_over = 0
|
||
ubyte_array_8 = c_ubyte * 8
|
||
DATA = ubyte_array_8(RESERVED, RESERVED, RESERVED, RESERVED, RESERVED, RESERVED, RESERVED, RESERVED)
|
||
ubyte_array_3 = c_ubyte * 3
|
||
RESERVED_3 = ubyte_array_3(RESERVED, RESERVED, RESERVED)
|
||
# 参数结构参考122行
|
||
can_obj = VCI_CAN_OBJ(RECEIVE_ID, TIME_STAMP, TIME_FLAG, RECEIVE_SEND_TYPE, REMOTE_FLAG, EXTERN_FLAG, DATA_LEN, DATA, RESERVED_3)
|
||
# VCI_USB_CAN_2: 设备类型
|
||
# DEV_INDEX: 设备索引
|
||
# can_index: CAN通道索引
|
||
# can_obj: 请求参数体
|
||
# RECEIVE_LEN: 用来接收帧结构体数组的长度
|
||
# WAIT_TIME: 保留参数
|
||
ret = Can_DLL.VCI_Receive(VCI_USB_CAN_2, DEV_INDEX, can_index, byref(can_obj), RECEIVE_LEN, WAIT_TIME)
|
||
while ret != STATUS_OK:
|
||
# print('VCI_Receive: 通道 ' + str(can_index + 1) + ' 接收数据失败, 正在重试')
|
||
ret = Can_DLL.VCI_Receive(VCI_USB_CAN_2, DEV_INDEX, can_index, byref(can_obj), RECEIVE_LEN, WAIT_TIME)
|
||
time.sleep(0.001)
|
||
t_over = t_over + 1
|
||
try:
|
||
win.update()
|
||
except:
|
||
print("receive end")
|
||
return ret
|
||
|
||
#未接收到信息，子线程继续尝试接收
|
||
# print(t_over)
|
||
# if t_over == 500:
|
||
# print("接收超时")
|
||
# textLOG.config(state="normal")
|
||
# LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_receiveERR] " + "接收超时(500回读取)\n"
|
||
# textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# # 移动到最新一行
|
||
# textLOG.yview_moveto(1)
|
||
# textLOG.config(state="disabled")
|
||
# return ret
|
||
else:
|
||
t_over = 0
|
||
global g_rx_count
|
||
g_rx_count += 1
|
||
update_status()
|
||
print('VCI_Receive: 通道 ' + str(can_index + 1) + ' 接收数据成功')
|
||
print('ID: ', can_obj.ID)
|
||
print('DataLen: ', can_obj.DataLen)
|
||
print('Data: ', list(can_obj.Data))
|
||
t_ReceiveData = list(can_obj.Data)
|
||
|
||
Str_Receive = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Receive] "
|
||
Str_Receive = Str_Receive + "0x" + str('%02x' % t_ReceiveData[0]) + " "
|
||
Str_Receive = Str_Receive + "0x" + str('%02x' % t_ReceiveData[1]) + " "
|
||
Str_Receive = Str_Receive + "0x" + str('%02x' % t_ReceiveData[2]) + " "
|
||
Str_Receive = Str_Receive + "0x" + str('%02x' % t_ReceiveData[3]) + " "
|
||
Str_Receive = Str_Receive + "0x" + str('%02x' % t_ReceiveData[4]) + " "
|
||
Str_Receive = Str_Receive + "0x" + str('%02x' % t_ReceiveData[5]) + " "
|
||
Str_Receive = Str_Receive + "0x" + str('%02x' % t_ReceiveData[6]) + " "
|
||
Str_Receive = Str_Receive + "0x" + str('%02x' % t_ReceiveData[7]) + "\n"
|
||
textLOG.config(state="normal")
|
||
textLOG.insert(tkinter.INSERT, Str_Receive)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
|
||
# print(chr(int("0x4f", 16)))
|
||
# print(chr(int("0x4b", 16)))
|
||
# print(chr(int(hex(0x4f), 16)))
|
||
# print(chr(int(hex(0x4b), 16)))
|
||
|
||
# 应用程序指令集
|
||
if t_ReceiveData[0] == 0x7f:
|
||
print("不支持的指令")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "不支持的指令\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x50 and t_ReceiveData[1] == 0x03:
|
||
print("指令成功，进入扩展会话模式")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "接收指令成功，进入扩展会话模式\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x50 and t_ReceiveData[1] == 0x01:
|
||
print("指令成功，进入01默认会话模式")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "进入01默认会话模式\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0xc5 and t_ReceiveData[1] == 0x02:
|
||
print("指令成功，关闭DTC")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "接收指令成功，关闭DTC\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0xc5 and t_ReceiveData[1] == 0x01:
|
||
print("指令成功，开启DTC")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "接收指令成功，开启DTC\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x68 and t_ReceiveData[1] == 0x03:
|
||
print("指令成功，禁止非诊断报文的发送和接收")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "接收指令成功，禁止非诊断报文的发送和接收\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x68 and t_ReceiveData[1] == 0x01:
|
||
print("指令成功，允许非诊断报文的发送和接收")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "接收指令成功，允许非诊断报文的发送和接收\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x54:
|
||
print("指令成功，清除诊断信息")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "接收指令成功，清除诊断信息\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
|
||
|
||
# boot指令集
|
||
if t_ReceiveData[0] == 0x50 and t_ReceiveData[1] == 0x02:
|
||
print("进入编程会话模式")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "进入编程会话模式，请在30S内执行相关指令，超时后将执行旧APP程序\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x62:
|
||
print("读取DID数据成功")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "读取DID"\
|
||
+ str('%02x' % t_ReceiveData[1]) + str('%02x' % t_ReceiveData[2]) + "数据成功："\
|
||
+ hex(t_ReceiveData[3]) + "_" + chr(int(hex(t_ReceiveData[3]), 16)) + " " \
|
||
+ hex(t_ReceiveData[4]) + "_" + chr(int(hex(t_ReceiveData[4]), 16)) + "\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x67 and t_ReceiveData[1] == 0x01:
|
||
print("获取seed成功")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "安全访问_获取seed成功，请验证key："\
|
||
+ hex(t_ReceiveData[2]) + " " \
|
||
+ hex(t_ReceiveData[3]) + "\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x67 and t_ReceiveData[1] == 0x02:
|
||
print("验证key成功")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "验证key成功，解锁成功\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x6e:
|
||
print("写入DID数据成功")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "写入DID"+\
|
||
str('%02x' % t_ReceiveData[1]) + str('%02x' % t_ReceiveData[2]) + "数据成功\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x71 and t_ReceiveData[1] == 0x01:
|
||
print("执行RID成功")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "执行RID"+\
|
||
str('%02x' % t_ReceiveData[2]) + str('%02x' % t_ReceiveData[3]) + "成功"
|
||
if t_ReceiveData[2] == 0x10 and t_ReceiveData[3] == 0x05:
|
||
LOG_str = LOG_str + "(擦除Memory)\n"
|
||
elif t_ReceiveData[2] == 0x15 and t_ReceiveData[3] == 0x55 and t_ReceiveData[3] == 0x77:
|
||
LOG_str = LOG_str + "(检查完整性,简单校验OK)\n"
|
||
elif t_ReceiveData[2] == 0x15 and t_ReceiveData[3] == 0x55 and t_ReceiveData[3] == 0x66:
|
||
LOG_str = LOG_str + "(检查完整性,简单校验ERR)\n"
|
||
else:
|
||
LOG_str = LOG_str + "\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x74:
|
||
print("请求下载成功")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "请求下载成功，最大数据块:"
|
||
if t_ReceiveData[1] != 0x00:
|
||
t_len = t_ReceiveData[1]>>4
|
||
t_lenSize = t_len
|
||
for len_i in range(2, 2 + t_lenSize):
|
||
LOG_str = LOG_str + str('%02x' % t_ReceiveData[len_i])
|
||
LOG_str = LOG_str + "，请开始传输BIN文件\n"
|
||
else:
|
||
LOG_str = LOG_str + "，请开始传输BIN文件\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x76 and t_ReceiveData[1] == 0x01:
|
||
print("传输完成")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "传输完成\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x77:
|
||
print("退出传输完成")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "退出传输完成\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
if t_ReceiveData[0] == 0x51 and t_ReceiveData[1] == 0x55:
|
||
print("跳转APP")
|
||
textLOG.config(state="normal")
|
||
LOG_str = "[" + time.strftime('%H:%M:%S', time.localtime(time.time())) + "][CAN_Analyse] " + "程序即将跳转APP\n"
|
||
textLOG.insert(tkinter.INSERT, LOG_str)
|
||
# 移动到最新一行
|
||
textLOG.yview_moveto(1)
|
||
textLOG.config(state="disabled")
|
||
win.update()
|
||
return ret
|
||
|
||
|
||
# win.update()
|
||
return ret
|
||
|
||
|
||
def CAN_Start():
|
||
# 初始化CAN1
|
||
init(CAN_INDEX_1)
|
||
# 启动CAN1
|
||
start(CAN_INDEX_1)
|
||
|
||
|
||
# ==================== IAP 升级相关 ====================
|
||
|
||
# IAP模式标志: True时暂停异步接收线程, 由send_and_wait同步收发
|
||
g_iap_mode = False
|
||
|
||
def iap_log(msg):
|
||
"""IAP日志输出到界面(青色)"""
|
||
log_append(msg, "iap")
|
||
|
||
def can_send(data_list):
|
||
"""发送一帧CAN数据(8字节, 不更新日志, 用于IAP流程)"""
|
||
global g_tx_count
|
||
ubyte_array_8 = c_ubyte * 8
|
||
DATA = ubyte_array_8(*data_list[:8])
|
||
ubyte_array_3 = c_ubyte * 3
|
||
RESERVED_3 = ubyte_array_3(0, 0, 0)
|
||
can_obj = VCI_CAN_OBJ(TRANSMIT_ID, 0, 0, 1, 0, 0, 8, DATA, RESERVED_3)
|
||
ret = Can_DLL.VCI_Transmit(VCI_USB_CAN_2, DEV_INDEX, CAN_INDEX_1, byref(can_obj), 1)
|
||
if ret == STATUS_OK:
|
||
g_tx_count += 1
|
||
return ret
|
||
|
||
def can_recv(timeout_ms=2000):
|
||
"""同步接收一帧CAN数据(阻塞等待, 超时返回None)"""
|
||
global g_rx_count
|
||
ubyte_array_8 = c_ubyte * 8
|
||
DATA = ubyte_array_8(0, 0, 0, 0, 0, 0, 0, 0)
|
||
ubyte_array_3 = c_ubyte * 3
|
||
RESERVED_3 = ubyte_array_3(0, 0, 0)
|
||
can_obj = VCI_CAN_OBJ(0, 0, 0, 0, 0, 0, 8, DATA, RESERVED_3)
|
||
|
||
start_time = time.time()
|
||
while True:
|
||
ret = Can_DLL.VCI_Receive(VCI_USB_CAN_2, DEV_INDEX, CAN_INDEX_1,
|
||
byref(can_obj), RECEIVE_LEN, 0)
|
||
if ret == STATUS_OK:
|
||
g_rx_count += 1
|
||
return list(can_obj.Data)
|
||
if (time.time() - start_time) * 1000 > timeout_ms:
|
||
return None
|
||
time.sleep(0.001)
|
||
|
||
def send_and_wait(data, timeout_ms=2000):
|
||
"""发送数据并同步等待响应帧"""
|
||
can_send(data)
|
||
return can_recv(timeout_ms)
|
||
|
||
def IAP_Upgrade(bin_path):
|
||
"""IAP完整升级流程: 握手→擦除→下载→传输→校验→复位"""
|
||
global g_iap_mode
|
||
|
||
# 读取固件文件
|
||
try:
|
||
with open(bin_path, 'rb') as f:
|
||
fw_data = f.read()
|
||
except Exception as e:
|
||
iap_log("错误: 无法读取固件文件 - %s" % str(e))
|
||
return
|
||
|
||
fw_size = len(fw_data)
|
||
iap_log("=" * 40)
|
||
iap_log("开始IAP升级")
|
||
iap_log("固件路径: %s" % bin_path)
|
||
iap_log("固件大小: %d 字节 (%.1f KB)" % (fw_size, fw_size / 1024))
|
||
|
||
# 进入IAP模式(暂停异步接收线程)
|
||
g_iap_mode = True
|
||
time.sleep(0.02) # 等待接收线程退出当前循环
|
||
|
||
try:
|
||
# ===== 阶段1: 握手 =====
|
||
iap_log("--- 阶段1: 握手 ---")
|
||
|
||
# 0x10 进入编程会话
|
||
resp = send_and_wait([0x10, 0x02, 0, 0, 0, 0, 0, 0])
|
||
if not resp or resp[0] != 0x50:
|
||
iap_log("错误: 进入编程会话失败 resp=%s" % resp)
|
||
return
|
||
iap_log("进入编程会话成功")
|
||
|
||
# 0x22 读版本
|
||
resp = send_and_wait([0x22, 0xF1, 0x00, 0, 0, 0, 0, 0])
|
||
if not resp or resp[0] != 0x62:
|
||
iap_log("错误: 读版本失败 resp=%s" % resp)
|
||
return
|
||
ver = "".join(chr(b) for b in resp[3:8] if 32 <= b < 127)
|
||
iap_log("Bootloader版本: %s" % ver)
|
||
|
||
# 0x27 安全访问 (Seed/Key)
|
||
resp = send_and_wait([0x27, 0x01, 0, 0, 0, 0, 0, 0])
|
||
if not resp or resp[0] != 0x67:
|
||
iap_log("错误: 请求Seed失败 resp=%s" % resp)
|
||
return
|
||
seed = resp[2:6]
|
||
key = [~b & 0xFF for b in seed] # Key = ~Seed
|
||
iap_log("Seed: %s" % " ".join("%02X" % b for b in seed))
|
||
|
||
resp = send_and_wait([0x27, 0x02] + key, 2000)
|
||
if not resp or resp[0] != 0x67:
|
||
iap_log("错误: Key验证失败 resp=%s" % resp)
|
||
return
|
||
iap_log("安全解锁成功")
|
||
|
||
# ===== 阶段2: 数据传输 =====
|
||
iap_log("--- 阶段2: 数据传输 ---")
|
||
|
||
# 0x31 擦除App区
|
||
resp = send_and_wait([0x31, 0x01, 0xFF, 0x01, 0, 0, 0, 0], 5000)
|
||
if not resp or resp[0] != 0x71:
|
||
iap_log("错误: 擦除App区失败 resp=%s" % resp)
|
||
return
|
||
iap_log("擦除App区成功")
|
||
|
||
# 0x34 请求下载(声明固件大小, 支持断电续传)
|
||
size_bytes = list(fw_size.to_bytes(4, 'big'))
|
||
resp = send_and_wait([0x34] + size_bytes, 5000)
|
||
if not resp or resp[0] != 0x74:
|
||
iap_log("错误: 请求下载失败 resp=%s" % resp)
|
||
return
|
||
|
||
# 解析续传偏移量(resp[2:6]为offset大端4字节)
|
||
offset = int.from_bytes(bytes(resp[2:6]), 'big')
|
||
if offset > 0:
|
||
iap_log("断电续传: 从偏移 %d 字节 (%.1f KB) 处继续" % (offset, offset / 1024))
|
||
else:
|
||
iap_log("全新升级, 从头开始传输")
|
||
|
||
# 0x36 01 开始数据传输(进入纯数据模式)
|
||
resp = send_and_wait([0x36, 0x01, 0, 0, 0, 0, 0, 0])
|
||
if not resp or resp[0] != 0x76:
|
||
iap_log("错误: 开始传输失败 resp=%s" % resp)
|
||
return
|
||
iap_log("进入纯数据模式, 开始传输固件...")
|
||
|
||
# 纯数据模式: 8字节/帧, 无响应, 从offset处开始发送
|
||
total = fw_size - offset
|
||
sent = 0
|
||
last_progress = 0
|
||
for i in range(offset, fw_size, 8):
|
||
chunk = list(fw_data[i:i + 8])
|
||
# 不足8字节补0xFF
|
||
while len(chunk) < 8:
|
||
chunk.append(0xFF)
|
||
can_send(chunk)
|
||
sent += min(8, fw_size - i)
|
||
|
||
# 每1KB显示一次进度
|
||
if sent - last_progress >= 1024:
|
||
pct = sent * 100 // total
|
||
iap_log("传输进度: %d/%d 字节 (%d%%)" % (sent, total, pct))
|
||
last_progress = sent
|
||
|
||
time.sleep(0.001) # 1ms间隔, 防止总线拥塞
|
||
|
||
iap_log("传输进度: %d/%d 字节 (100%%)" % (total, total))
|
||
|
||
# 等待传输完成响应(板子收够fw_size后自动发0x76)
|
||
resp = can_recv(3000)
|
||
if not resp or resp[0] != 0x76:
|
||
iap_log("错误: 等待传输完成响应超时 resp=%s" % resp)
|
||
return
|
||
iap_log("数据传输完成")
|
||
|
||
# 0x37 传输退出
|
||
resp = send_and_wait([0x37, 0, 0, 0, 0, 0, 0, 0])
|
||
if not resp or resp[0] != 0x77:
|
||
iap_log("错误: 传输退出失败 resp=%s" % resp)
|
||
return
|
||
iap_log("传输退出成功")
|
||
|
||
# ===== 阶段3: 校验烧写 =====
|
||
iap_log("--- 阶段3: 校验烧写 ---")
|
||
|
||
# 0x31 CRC校验
|
||
resp = send_and_wait([0x31, 0x01, 0xFF, 0x02, 0, 0, 0, 0], 5000)
|
||
if not resp or resp[0] != 0x71:
|
||
iap_log("错误: CRC校验失败 resp=%s" % resp)
|
||
return
|
||
|
||
# 读取板子计算的CRC32
|
||
board_crc = int.from_bytes(bytes(resp[4:8]), 'big')
|
||
# 计算本地CRC32 (与板子一致: IEEE 802.3)
|
||
local_crc = binascii.crc32(fw_data) & 0xFFFFFFFF
|
||
iap_log("板子CRC: 0x%08X, 本地CRC: 0x%08X" % (board_crc, local_crc))
|
||
|
||
if board_crc != local_crc:
|
||
iap_log("错误: CRC校验不匹配! 升级失败")
|
||
return
|
||
iap_log("CRC校验通过")
|
||
|
||
# 0x11 复位跳转App
|
||
resp = send_and_wait([0x11, 0x01, 0, 0, 0, 0, 0, 0])
|
||
if not resp or resp[0] != 0x51:
|
||
iap_log("错误: 复位失败 resp=%s" % resp)
|
||
return
|
||
iap_log("复位成功, 等待跳转App...")
|
||
|
||
iap_log("=" * 40)
|
||
iap_log("IAP升级成功! 固件已烧写并校验通过")
|
||
|
||
except Exception as e:
|
||
iap_log("异常: %s" % str(e))
|
||
finally:
|
||
g_iap_mode = False # 恢复异步接收线程
|
||
|
||
|
||
def ReadFile(BIN_PATH):
|
||
# filepath='C:\\Users\\\HenchYoung\\Desktop\\LED2000.bin'
|
||
filepath = BIN_PATH
|
||
binfile = open(filepath, 'rb') #打开二进制文件
|
||
size = os.path.getsize(filepath) #获得文件大小
|
||
# for i in range(size):
|
||
# data = binfile.read(1) #每次输出一个字节
|
||
# print(data)
|
||
# print(i)
|
||
i = 0
|
||
Send_BUFF = [0,0,0,0,0,0,0,0]
|
||
# Send_BUFF = [b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00', b'\x00']
|
||
|
||
# with open("C:\\Users\\HenchYoung\\Desktop\\hex.c", "w") as f:
|
||
# f.write("NEW ")
|
||
while 1:
|
||
c = binfile.read(1)
|
||
# 将字节转换成16进制；
|
||
ssss = str(binascii.b2a_hex(c))[2:-1]
|
||
# print(str(binascii.b2a_hex(c))[2:-1])
|
||
if not c:
|
||
t_sendstr = ""
|
||
for sdata in Send_BUFF:
|
||
t_sendstr = t_sendstr + " " + hex(int(str(sdata),16))
|
||
t_sendstr = t_sendstr + "\n" + str(BIN_PATH)
|
||
t_sendstr = t_sendstr + " 固件发送完成，共计：" + str(i) + "字节"
|
||
transmitBIN(CAN_INDEX_1, Send_BUFF, t_sendstr)
|
||
# with open("C:\\Users\\HenchYoung\\Desktop\\hex.c", "a") as f:
|
||
# for se in Send_BUFF:
|
||
# f.write(" ")
|
||
# f.write(str(se))
|
||
# f.write(" ")
|
||
# f.write("\n")
|
||
Send_BUFF = [0,0,0,0,0,0,0,0]
|
||
time.sleep(0.001)
|
||
break
|
||
# Send_BUFF[i%8] = int(str(ssss),16)
|
||
Send_BUFF[i % 8] = int(str(ssss),16)
|
||
i = i + 1
|
||
if i % 8 == 0:
|
||
t_sendstr = ""
|
||
for sdata in Send_BUFF:
|
||
t_sendstr = t_sendstr + " " + hex(int(str(sdata), 16))
|
||
transmitBIN(CAN_INDEX_1, Send_BUFF, t_sendstr)
|
||
# with open("C:\\Users\\HenchYoung\\Desktop\\hex.c", "a") as f:
|
||
# for se in Send_BUFF:
|
||
# f.write(" ")
|
||
# f.write(str(se))
|
||
# f.write(" ")
|
||
# f.write("\n")
|
||
Send_BUFF = [0,0,0,0,0,0,0,0]
|
||
time.sleep(0.001)
|
||
if i == size:
|
||
break
|
||
# print(ssss)
|
||
# print(int(str(ssss),16))
|
||
# print(c)
|
||
# print(bytes().fromhex(ssss)) # 将16进制转换为字节
|
||
# if i % 16 == 0:
|
||
# time.sleep(0.001)
|
||
# 写每一行等待的时间
|
||
print(i)
|
||
|
||
binfile.close()
|
||
print(size)
|
||
|
||
|
||
|
||
|
||
if __name__ == '__main__':
|
||
global win #界面实体
|
||
global CAN_Send_entry #发送数据的内容
|
||
global BIN_path_entry #BIN文件路径
|
||
global textLOG #textbox的数据
|
||
global scroll #滚动条实体
|
||
global CAN_INDEX_1
|
||
global CAN_INDEX_2
|
||
global g_winOVER
|
||
global threadREV
|
||
|
||
g_winOVER = 0
|
||
|
||
CAN_INDEX_1 = 0
|
||
CAN_INDEX_2 = 1
|
||
time1 = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
|
||
|
||
g_WindowStart()
|
||
|
||
#关闭
|
||
Can_DLL.VCI_CloseDevice(VCI_USB_CAN_2, 0)
|
||
print("CAN end")
|
||
|
||
print(time1)
|
||
print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
|
||
# print(time.strftime('%H:%M:%S', time.localtime(time.time())))
|
||
|
||
|