import tkinter as tk from tkinter import ttk, messagebox, filedialog import os import threading from .base_page import BasePage from protocol import DbgUpgradeData from protocol import DbgUpgradeResponseData from config import Cmd class UpgradePage(BasePage): def __init__(self, parent, tcp_client): # 升级类型映射字典 self.upgrade_types = { "CMU+FPGA": 0, "CMU": 1, "FPGA": 2, "其他": 3 } super().__init__(parent, tcp_client, "固件升级") self.upgrade_file = None self.upgrade_data = None self.current_index = 0 self.total_packets = 0 self.upgrading = False self.waiting_response = False tcp_client.register_callback(Cmd.UPGRADE_DATA, self.on_upgrade_response) # self.create_widgets() def create_widgets(self): """创建界面控件""" # 升级类型选择 type_frame = ttk.Frame(self) type_frame.pack(fill=tk.X, padx=10, pady=5) ttk.Label(type_frame, text="升级类型:").pack(side=tk.LEFT, padx=5) self.upgrade_type_var = tk.StringVar(value="CMU+FPGA") # 默认值 # 创建下拉框，显示文本但存储对应的数值 upgrade_combo = ttk.Combobox(type_frame, textvariable=self.upgrade_type_var, values=list(self.upgrade_types.keys()), state="readonly", width=12) upgrade_combo.pack(side=tk.LEFT, padx=5) # 文件选择 file_frame = ttk.Frame(self) file_frame.pack(fill=tk.X, padx=10, pady=5) ttk.Label(file_frame, text="升级文件:").pack(side=tk.LEFT, padx=5) self.file_path_var = tk.StringVar() ttk.Entry(file_frame, textvariable=self.file_path_var, width=50, state='readonly').pack(side=tk.LEFT, padx=5) ttk.Button(file_frame, text="浏览", command=self.browse_file).pack(side=tk.LEFT, padx=5) # 进度显示 progress_frame = ttk.Frame(self) progress_frame.pack(fill=tk.X, padx=10, pady=5) ttk.Label(progress_frame, text="进度:").pack(side=tk.LEFT, padx=5) self.progress_var = tk.DoubleVar() progress_bar = ttk.Progressbar(progress_frame, variable=self.progress_var, maximum=100) progress_bar.pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True) self.progress_label = ttk.Label(progress_frame, text="0/0") self.progress_label.pack(side=tk.RIGHT, padx=5) # 状态显示 status_frame = ttk.Frame(self) status_frame.pack(fill=tk.X, padx=10, pady=5) ttk.Label(status_frame, text="状态:").pack(side=tk.LEFT, padx=5) self.status_var = tk.StringVar(value="就绪") ttk.Label(status_frame, textvariable=self.status_var).pack(side=tk.LEFT, padx=5) # 按钮区域 button_frame = ttk.Frame(self) button_frame.pack(pady=10) self.start_button = ttk.Button(button_frame, text="开始升级", command=self.start_upgrade) self.start_button.pack(side=tk.LEFT, padx=5) self.stop_button = ttk.Button(button_frame, text="停止升级", command=self.stop_upgrade, state='disabled') self.stop_button.pack(side=tk.LEFT, padx=5) ttk.Button(button_frame, text="清除状态", command=self.clear_status).pack(side=tk.LEFT, padx=5) def get_upgrade_type_code(self): """获取当前选择的升级类型代码""" selected_text = self.upgrade_type_var.get() return self.upgrade_types.get(selected_text, 0) # 默认返回0 def get_upgrade_type_text(self, code=None): """根据代码获取升级类型文本，或获取当前选择的文本""" if code is not None: # 根据代码找文本 for text, value in self.upgrade_types.items(): if value == code: return text return "其他" else: # 返回当前选择的文本 return self.upgrade_type_var.get() def browse_file(self): """选择升级文件""" if self.upgrading: messagebox.showerror("错误", "升级进行中，无法选择文件") return file_path = filedialog.askopenfilename( title="选择升级文件", filetypes=[("二进制文件", "*.bin"), ("固件文件", "*.fw"), ("所有文件", "*.*")] ) if file_path: self.file_path_var.set(file_path) self.upgrade_file = file_path file_size = os.path.getsize(file_path) upgrade_type_text = self.get_upgrade_type_text() self.status_var.set(f"已选择{upgrade_type_text}文件: {os.path.basename(file_path)} ({file_size} 字节)") def start_upgrade(self): """开始升级流程""" if not self.tcp_client.connected: messagebox.showerror("错误", "未连接到设备") return if not self.upgrade_file or not os.path.exists(self.upgrade_file): messagebox.showerror("错误", "请选择有效的升级文件") return # 获取升级类型代码 upgrade_type_code = self.get_upgrade_type_code() upgrade_type_text = self.get_upgrade_type_text() # 读取升级文件 try: with open(self.upgrade_file, 'rb') as f: self.upgrade_data = f.read() except Exception as e: messagebox.showerror("错误", f"读取文件失败: {e}") return # 计算总包数 self.total_packets = (len(self.upgrade_data) + 1023) // 1024 self.current_index = 0 self.upgrading = True self.waiting_response = False # 更新界面状态 self.start_button.config(state='disabled') self.stop_button.config(state='normal') self.progress_var.set(0) self.progress_label.config(text=f"0/{self.total_packets}") self.status_var.set(f"开始{upgrade_type_text}升级...") # 发送第一个数据包 self.send_next_packet() def stop_upgrade(self): """停止升级""" if self.upgrading: self.upgrading = False self.waiting_response = False upgrade_type_text = self.get_upgrade_type_text() self.status_var.set(f"{upgrade_type_text}升级已停止") self.start_button.config(state='normal') self.stop_button.config(state='disabled') def clear_status(self): """清除状态""" if not self.upgrading: self.status_var.set("就绪") self.progress_var.set(0) self.progress_label.config(text="0/0") def send_next_packet(self): """发送下一个数据包""" if not self.upgrading or not self.upgrade_data or self.waiting_response: return if self.current_index >= self.total_packets: self.complete_upgrade() return # 计算当前数据包范围 packet_size = 1024 start = self.current_index * packet_size end = min(start + packet_size, len(self.upgrade_data)) # 准备升级数据包头 upgrade_header = DbgUpgradeData() upgrade_header.type = self.get_upgrade_type_code() # 设置升级类型代码 upgrade_header.index = self.current_index upgrade_header.sum = self.total_packets upgrade_header.len = end - start # 组合数据包头和数据 packet_data = upgrade_header.to_bytes() + self.upgrade_data[start:end] # 发送数据包 if self.tcp_client.send_packet(Cmd.UPGRADE_DATA, packet_data): self.waiting_response = True # 更新状态显示 upgrade_type_text = self.get_upgrade_type_text() self.status_var.set(f"{upgrade_type_text}升级: {self.current_index + 1}/{self.total_packets}") # 更新进度 progress = (self.current_index * packet_size) / len(self.upgrade_data) * 100 self.progress_var.set(min(progress, 100)) self.progress_label.config(text=f"{self.current_index}/{self.total_packets}") else: self.upgrading = False self.status_var.set("发送失败") self.start_button.config(state='normal') self.stop_button.config(state='disabled') def complete_upgrade(self): """完成升级""" self.upgrading = False self.waiting_response = False upgrade_type_text = self.get_upgrade_type_text() self.status_var.set(f"{upgrade_type_text}升级完成") self.progress_var.set(100) self.start_button.config(state='normal') self.stop_button.config(state='disabled') messagebox.showinfo("成功", f"{upgrade_type_text}升级完成") def on_upgrade_response(self, header, body): """处理升级响应""" if not self.upgrading or not self.waiting_response: return try: response = DbgUpgradeResponseData.from_bytes(body) # 检查响应是否正确 if response.index == self.current_index: # 当前包发送成功，继续下一个 self.current_index += 1 self.waiting_response = False # 更新进度显示 packet_size = 1024 progress = (self.current_index * packet_size) / len(self.upgrade_data) * 100 self.progress_var.set(min(progress, 100)) self.progress_label.config(text=f"{self.current_index}/{self.total_packets}") # 发送下一个包或完成升级 if self.current_index < self.total_packets: upgrade_type_text = self.get_upgrade_type_text() self.status_var.set(f"继续{upgrade_type_text}升级...") self.send_next_packet() else: self.complete_upgrade() else: self.status_var.set(f"响应包索引不匹配，重发") self.waiting_response = False self.send_next_packet() except Exception as e: self.status_var.set(f"解析响应失败: {e}") # 重发当前包 self.waiting_response = False self.send_next_packet()