""" 激光示踪设备网关配置工具 通过串口(RS485)直接修改和查看网关参数 """ import tkinter as tk from tkinter import ttk, scrolledtext, messagebox, filedialog import struct import binascii import serial import serial.tools.list_ports import threading import time import re import sys import os import shutil import subprocess from datetime import datetime def resource_path(relative_path): if hasattr(sys, '_MEIPASS'): return os.path.join(sys._MEIPASS, relative_path) return os.path.join(os.path.abspath('.'), relative_path) class LaserTracingConfigApp: def __init__(self, root): self.root = root self.root.title("边坡变形监测-变形(位移)自动测量系统配置工具") self.root.resizable(True, True) icon_path = resource_path("传感器.ico") if os.path.exists(icon_path): self.root.iconbitmap(icon_path) self.serial_port = None self.serial_thread = None self.running = False self.upgrading = False self.recv_buffer = "" self._port_details = {} self._build_ui() self._refresh_ports() # 根据内容自动调整窗口大小，刚好完整显示参数 self.root.update_idletasks() self.root.geometry("") # ─────────────────── UI构建 ─────────────────── def _build_ui(self): # 串口连接区域 conn_frame = ttk.LabelFrame(self.root, text="串口连接", padding=8) conn_frame.pack(fill=tk.X, padx=8, pady=(8, 4)) # ── 串口: 选择 + 波特率 + 数据位 + 停止位 + 校验 + 打开按钮 ── ttk.Label(conn_frame, text="串口:").grid(row=0, column=0, sticky=tk.W, padx=2) self.port_combo = ttk.Combobox(conn_frame, width=35, state="readonly") self.port_combo.grid(row=0, column=1, padx=2) ttk.Button(conn_frame, text="刷新", width=5, command=self._refresh_ports).grid(row=0, column=2, padx=2) ttk.Label(conn_frame, text="波特率:").grid(row=0, column=3, sticky=tk.W, padx=2) self.baudrate_var = tk.StringVar(value="9600") ttk.Label(conn_frame, textvariable=self.baudrate_var, width=8).grid(row=0, column=4, padx=2) ttk.Label(conn_frame, text="数据位:").grid(row=0, column=5, sticky=tk.W, padx=2) self.databits_var = tk.StringVar(value="8") ttk.Label(conn_frame, textvariable=self.databits_var, width=4).grid(row=0, column=6, padx=2) ttk.Label(conn_frame, text="停止位:").grid(row=0, column=7, sticky=tk.W, padx=2) self.stopbits_var = tk.StringVar(value="1") ttk.Label(conn_frame, textvariable=self.stopbits_var, width=4).grid(row=0, column=8, padx=2) ttk.Label(conn_frame, text="校验:").grid(row=0, column=9, sticky=tk.W, padx=2) self.parity_var = tk.StringVar(value="None") ttk.Label(conn_frame, textvariable=self.parity_var, width=6).grid(row=0, column=10, padx=2) self.connect_btn = ttk.Button(conn_frame, text="打开串口", width=10, command=self._toggle_connection) self.connect_btn.grid(row=0, column=11, padx=(12, 2)) # 参数配置区域 config_frame = ttk.LabelFrame(self.root, text="参数配置", padding=8) config_frame.pack(fill=tk.X, padx=8, pady=4) # Lora参数 lora_frame = ttk.LabelFrame(config_frame, text="Lora 参数", padding=6) lora_frame.pack(fill=tk.X, pady=(0, 4)) ttk.Label(lora_frame, text="信道频率(fc):").grid(row=0, column=0, sticky=tk.W, padx=2) self._fc_display_to_value = {} self._fc_value_to_display = {} fc_displays = [] freq = 410100000 ch = 1 while freq <= 492900000: display = f"信道{ch}: {freq / 1000000:.1f}MHz" fc_displays.append(display) self._fc_display_to_value[display] = str(freq) self._fc_value_to_display[str(freq)] = display freq += 200000 ch += 1 self.lora_fc_display_var = tk.StringVar(value=self._fc_value_to_display.get("433100000", fc_displays[0] if fc_displays else "")) self.lora_fc_combo = ttk.Combobox(lora_frame, textvariable=self.lora_fc_display_var, values=fc_displays, width=20, state="readonly") self.lora_fc_combo.grid(row=0, column=1, padx=2) ttk.Button(lora_frame, text="发送", command=self._send_lora_config).grid(row=0, column=3, padx=8) ttk.Label(lora_frame, text="设置LoRa通信信道频率", foreground="gray").grid(row=0, column=4, padx=(0, 2), sticky=tk.W) # 全功率时段配置 fpo_cfg = ttk.LabelFrame(config_frame, text="全功率时段配置", padding=6) fpo_cfg.pack(fill=tk.X, pady=(0, 4)) ttk.Label(fpo_cfg, text="开始时间(h):").grid(row=0, column=0, sticky=tk.W, padx=2) self._fpo_start_var = tk.StringVar(value="0") ttk.Entry(fpo_cfg, textvariable=self._fpo_start_var, width=8).grid(row=0, column=1, padx=2) ttk.Label(fpo_cfg, text="持续时间(h):").grid(row=0, column=2, sticky=tk.W, padx=2) self._fpo_dur_var = tk.StringVar(value="0") ttk.Entry(fpo_cfg, textvariable=self._fpo_dur_var, width=8).grid(row=0, column=3, padx=2) ttk.Button(fpo_cfg, text="发送", command=self._send_fpo_cfg).grid(row=0, column=4, padx=8) ttk.Label(fpo_cfg, text="设置全功率时间段，开始0~23时，持续0~24时(0=关闭)", foreground="gray").grid( row=0, column=5, padx=(0, 2), sticky=tk.W) # 激光控制配置 laser_cfg = ttk.LabelFrame(config_frame, text="激光控制", padding=6) laser_cfg.pack(fill=tk.X, pady=(0, 4)) ttk.Button(laser_cfg, text="打 开", command=lambda: self._send_command("laser on")).grid(row=0, column=0, padx=2) ttk.Button(laser_cfg, text="关 闭", command=lambda: self._send_command("laser off")).grid(row=0, column=1, padx=2) ttk.Label(laser_cfg, text="自动关闭(min):").grid(row=0, column=2, sticky=tk.W, padx=2) self._laser_time_var = tk.StringVar(value="10") ttk.Entry(laser_cfg, textvariable=self._laser_time_var, width=8).grid(row=0, column=3, padx=2) ttk.Button(laser_cfg, text="设置", command=self._send_laser_time).grid(row=0, column=4, padx=8) ttk.Label(laser_cfg, text="打开/关闭激光，自动关闭时间(0=禁用)", foreground="gray").grid( row=0, column=5, padx=(0, 2), sticky=tk.W) # 查询按钮行 query_frame = ttk.Frame(config_frame) query_frame.pack(fill=tk.X, pady=4) ttk.Button(query_frame, text="查询当前配置 (para)", command=self._send_para).pack(side=tk.LEFT, padx=4) ttk.Button(query_frame, text="设备升级", command=self._launch_upgrade).pack(side=tk.LEFT, padx=4) ttk.Label(query_frame, text="配置后可通过para命令验证是否成功", foreground="gray").pack(side=tk.LEFT, padx=8) # 参数显示区域 - 改用grid布局每行两个参数 para_frame = ttk.LabelFrame(self.root, text="设备参数信息", padding=8) para_frame.pack(fill=tk.X, padx=8, pady=4) # 配置两列均匀分布 para_frame.columnconfigure(0, weight=1) para_frame.columnconfigure(1, weight=1) para_frame.columnconfigure(2, weight=1) para_frame.columnconfigure(3, weight=1) # 参数按顺序每两个一组 self._para_order = ["软件版本", "硬件版本", "本机MAC", "电池电量", "设备类型", "通信方式", "设备标识", "网关MAC", "注册状态", "Lora频率"] # 创建标签对 self._para_labels = {} for idx, key in enumerate(self._para_order): row = idx // 2 col = (idx % 2) * 2 # 参数名标签 name_label = ttk.Label(para_frame, text=f"{key}:", font=("SimSun", 9)) name_label.grid(row=row, column=col, sticky=tk.W, padx=(0, 2), pady=2) # 参数值标签 value_label = ttk.Label(para_frame, text="-", font=("SimSun", 9), foreground="green") value_label.grid(row=row, column=col + 1, sticky=tk.W + tk.E, padx=(0, 10), pady=2) self._para_labels[key] = value_label # 低功耗配置单独区块 - 布局与设备参数信息一致 lp_frame = ttk.LabelFrame(self.root, text="低功耗配置", padding=6) lp_frame.pack(fill=tk.X, padx=8, pady=4) lp_frame.columnconfigure(0, weight=1) lp_frame.columnconfigure(1, weight=1) lp_frame.columnconfigure(2, weight=1) lp_frame.columnconfigure(3, weight=1) self._lp_labels = {} lp_params = [("CI", "采集间隔"), ("RI", "上报时间"), ("EI", "紧急上报间隔"), ("ET", "紧急上报时长")] for idx, (key, label) in enumerate(lp_params): row = idx // 2 col = (idx % 2) * 2 ttk.Label(lp_frame, text=f"{label}:", font=("SimSun", 9)).grid(row=row, column=col, sticky=tk.W, padx=(0, 2), pady=2) val_label = ttk.Label(lp_frame, text="-", font=("SimSun", 9), foreground="green") val_label.grid(row=row, column=col + 1, sticky=tk.W + tk.E, padx=(0, 10), pady=2) self._lp_labels[key] = val_label # 功率与激光配置 pwr_frame = ttk.LabelFrame(self.root, text="功率与激光配置", padding=6) pwr_frame.pack(fill=tk.X, padx=8, pady=4) for col in range(4): pwr_frame.columnconfigure(col, weight=1) self._pwr_labels = {} pwr_params = [ ("fpo_start", "开始时间"), ("fpo_end", "结束时间"), ("fpo_dur", "全功率时长"), ("fpo_mode", "当前模式"), ] for idx, (key, label) in enumerate(pwr_params): row = idx // 2 col = (idx % 2) * 2 ttk.Label(pwr_frame, text=f"{label}:", font=("SimSun", 9)).grid(row=row, column=col, sticky=tk.W, padx=(0, 2), pady=2) val_label = ttk.Label(pwr_frame, text="-", font=("SimSun", 9), foreground="green") val_label.grid(row=row, column=col + 1, sticky=tk.W + tk.E, padx=(0, 10), pady=2) self._pwr_labels[key] = val_label # 激光另起一行 laser_row = 2 ttk.Label(pwr_frame, text="激光状态:", font=("SimSun", 9)).grid(row=laser_row, column=0, sticky=tk.W, padx=(0, 2), pady=2) self._laser_state_label = ttk.Label(pwr_frame, text="-", font=("SimSun", 9), foreground="green") self._laser_state_label.grid(row=laser_row, column=1, sticky=tk.W + tk.E, padx=(0, 10), pady=2) ttk.Label(pwr_frame, text="自动关闭:", font=("SimSun", 9)).grid(row=laser_row, column=2, sticky=tk.W, padx=(0, 2), pady=2) self._laser_autooff_label = ttk.Label(pwr_frame, text="-", font=("SimSun", 9), foreground="green") self._laser_autooff_label.grid(row=laser_row, column=3, sticky=tk.W + tk.E, padx=(0, 10), pady=2) # 通讯日志区域 log_frame = ttk.LabelFrame(self.root, text="通讯日志", padding=4) log_frame.pack(fill=tk.BOTH, expand=True, padx=8, pady=(4, 8)) log_inner = ttk.Frame(log_frame) log_inner.pack(fill=tk.BOTH, expand=True) # 左侧：日志文本 + 底部清空按钮 log_left = ttk.Frame(log_inner) log_left.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) self.log_text = scrolledtext.ScrolledText(log_left, height=10, font=("Consolas", 9), state=tk.DISABLED, wrap=tk.WORD, bg="#1E1E1E", fg="#00FF00", insertbackground="#00FF00") self.log_text.pack(fill=tk.BOTH, expand=True) ttk.Button(log_left, text="清空日志", command=self._clear_log).pack(anchor=tk.W, pady=(2, 0)) # 右侧：自定义指令 cmd_frame = ttk.LabelFrame(log_inner, text="自定义指令", padding=6) cmd_frame.pack(side=tk.RIGHT, fill=tk.Y, padx=(6, 0)) self.custom_cmd_var = tk.StringVar() cmd_entry = ttk.Entry(cmd_frame, textvariable=self.custom_cmd_var, width=20) cmd_entry.pack(pady=(0, 4)) cmd_entry.bind("", lambda e: self._send_custom_cmd()) ttk.Button(cmd_frame, text="发送", command=self._send_custom_cmd).pack() # 日志颜色标签 self.log_text.tag_configure("send", foreground="#00CCFF") self.log_text.tag_configure("recv", foreground="#00FF00") self.log_text.tag_configure("error", foreground="#FF4444") self.log_text.tag_configure("info", foreground="#88FF88") # ─────────────────── 串口操作 ─────────────────── def _refresh_ports(self): port_list = serial.tools.list_ports.comports() self._port_details = {} display_names = [] for p in sorted(port_list, key=lambda x: x.device): display = p.description # 如果多个端口有相同描述，追加设备名区分 if display in self._port_details: display = f"{display} ({p.device})" display_names.append(display) self._port_details[display] = p.device self.port_combo["values"] = display_names if display_names: self.port_combo.set(display_names[0]) def _toggle_connection(self): if self.serial_port and self.serial_port.is_open: self._disconnect() else: self._connect() def _connect(self): selected = self.port_combo.get() port = self._port_details.get(selected) if not port: messagebox.showwarning("警告", "请选择串口") return try: self.serial_port = serial.Serial( port=port, baudrate=int(self.baudrate_var.get()), bytesize=serial.EIGHTBITS, stopbits=serial.STOPBITS_ONE, parity=serial.PARITY_NONE, timeout=0.1 ) self.running = True self.recv_buffer = "" self.serial_thread = threading.Thread(target=self._recv_loop, daemon=True) self.serial_thread.start() self.connect_btn.configure(text="关闭串口") self.port_combo.configure(state=tk.DISABLED) self._log(f"已连接 {port} @ {self.baudrate_var.get()}", "info") except serial.SerialException as e: messagebox.showerror("连接失败", str(e)) self._log(f"连接失败: {e}", "error") def _disconnect(self): self.running = False if self.serial_thread: self.serial_thread.join(timeout=1) if self.serial_port and self.serial_port.is_open: self.serial_port.close() self.serial_port = None self.connect_btn.configure(text="打开串口") self.port_combo.configure(state="readonly") self._log("已断开连接", "info") def _recv_loop(self): while self.running: try: if self.upgrading: time.sleep(0.02) continue if self.serial_port and self.serial_port.is_open: n = self.serial_port.in_waiting if n > 0: data = self.serial_port.read(n) text = data.decode("utf-8", errors="replace") self.recv_buffer += text self.root.after(0, self._on_data_received, text) else: time.sleep(0.05) else: time.sleep(0.1) except Exception as e: if self.running: self.root.after(0, self._log, f"接收错误: {e}", "error") break def _on_data_received(self, text): self._log(text, "recv", newline=False) # 收到完整para回应：起始与结束各出现一次分隔符（****）后才解析 marker = "**********************************************************" if self.recv_buffer.count(marker) >= 2: self._parse_para_response(self.recv_buffer) self.recv_buffer = "" def _send_command(self, cmd): if not self.serial_port or not self.serial_port.is_open: messagebox.showwarning("警告", "请先连接串口") return try: self.recv_buffer = "" self.serial_port.write((cmd + "\r\n").encode("utf-8")) self._log(f">>> {cmd}", "send") except serial.SerialException as e: self._log(f"发送失败: {e}", "error") def _send_custom_cmd(self): cmd = self.custom_cmd_var.get().strip() if cmd: self._send_command(cmd) self.custom_cmd_var.set("") # ─────────────────── 指令发送 ─────────────────── def _send_para(self): self._send_command("para") def _launch_upgrade(self): """打开设备升级对话框。复用当前已连接的串口，不断开连接。""" if not self.serial_port or not self.serial_port.is_open: messagebox.showwarning("警告", "请先打开串口") return UpgradeDialog(self, self.serial_port) def _send_lora_config(self): display = self.lora_fc_display_var.get().strip() fc = self._fc_display_to_value.get(display) if not fc: messagebox.showwarning("参数错误", "请选择信道频率") return self._send_command(f"lora fc {fc}") def _send_fpo_cfg(self): try: start = int(self._fpo_start_var.get().strip()) dur = int(self._fpo_dur_var.get().strip()) except ValueError: messagebox.showwarning("参数错误", "请输入有效整数") return if not (0 <= start <= 23): messagebox.showwarning("参数错误", "开始小时范围 0~23") return if not (0 <= dur <= 24): messagebox.showwarning("参数错误", "时长范围 0~24 (0=关闭)") return self._send_command(f"fpocfg {start} {dur}") def _send_laser_time(self): try: mins = int(self._laser_time_var.get().strip()) except ValueError: messagebox.showwarning("参数错误", "请输入有效整数") return if mins < 0: messagebox.showwarning("参数错误", "关闭时间不能为负") return self._send_command(f"laser time {mins}") # ─────────────────── 回应解析 ─────────────────── def _parse_para_response(self, text): lines = text.replace("\r\n", "\n").replace("\r", "\n").split("\n") key_value_patterns = [ (r"[Ss]oft[Ww]are\s*[Vv]:?\s*(\d+(?:\.\d+)*)", "软件版本"), (r"[Hh]ard[Ww]are\s*[Vv]:?\s*(\d+(?:\.\d+)*)", "硬件版本"), (r"Local MAC:\s*(.*)", "本机MAC"), (r"Battery:\s*(.*)", "电池电量"), (r"LocalType:\s*(.*)", "设备类型"), (r"CommInf:\s*(.*)", "通信方式"), (r"DeviceSign:\s*(.*)", "设备标识"), (r"GateWay MAC:\s*(.*)", "网关MAC"), (r"Register Status:\s*(.*)", "注册状态"), (r"Lora.*?fc\s*(\d+)", "Lora频率"), (r"LpCfg:\s*(.*)", "低功耗配置"), (r"FpoCfg:\s*(.*)", "全功率工作时间段"), (r"Laser:\s*(.*)", "激光"), ] found = {} for line in lines: for pattern, label in key_value_patterns: m = re.search(pattern, line) if m: found[label] = m.group(1).strip() # 更新参数显示 for key in self._para_order: value = found.get(key, "-") if key in self._para_labels: self._para_labels[key].configure(text=value) # 更新Lora频率下拉框 if "Lora频率" in found: fc_val = found["Lora频率"] display = self._fc_value_to_display.get(fc_val) if display: self.lora_fc_display_var.set(display) else: new_display = f"自定义: {int(fc_val) / 1000000:.1f}MHz" self._fc_display_to_value[new_display] = fc_val self._fc_value_to_display[fc_val] = new_display self.lora_fc_combo["values"] = list(self.lora_fc_combo["values"]) + [new_display] self.lora_fc_display_var.set(new_display) # 解析低功耗配置参数 if "低功耗配置" in found: lp_str = found["低功耗配置"] ci_match = re.search(r"CI[:#\s]*(\d+)", lp_str) ri_match = re.search(r"RI[:#\s]*(\d+)", lp_str) ei_match = re.search(r"EI[:#\s]*(\d+)", lp_str) et_match = re.search(r"ET[:#\s]*(\d+)", lp_str) def format_sec(val): return f"{val}秒" def format_min(val): val = int(val) if val >= 60: hours = val // 60 mins = val % 60 if mins == 0: return f"{hours}时" else: return f"{hours}时{mins}分钟" else: return f"{val}分钟" if ci_match: self._lp_labels["CI"].configure(text=format_sec(ci_match.group(1))) if ri_match: self._lp_labels["RI"].configure(text=format_min(ri_match.group(1))) if ei_match: self._lp_labels["EI"].configure(text=format_min(ei_match.group(1))) if et_match: self._lp_labels["ET"].configure(text=format_min(et_match.group(1))) # 解析全功率工作时间段 if "全功率工作时间段" in found: fpo_str = found["全功率工作时间段"] m_start = re.search(r"Start:(\S+)", fpo_str) m_dur = re.search(r"Dur:(\S+)", fpo_str) m_end = re.search(r"End:(\S+)", fpo_str) if m_start: self._pwr_labels["fpo_start"].configure(text=m_start.group(1).replace(",", "")) if m_end: self._pwr_labels["fpo_end"].configure(text=m_end.group(1).replace(",", "")) if m_dur: self._pwr_labels["fpo_dur"].configure(text=m_dur.group(1).replace(",", "")) self._pwr_labels["fpo_mode"].configure(text="低功耗" if "LowPower" in fpo_str else "全功率") # 解析激光状态 if "激光" in found: laser_str = found["激光"] m_state = re.search(r"^([A-Za-z]+)", laser_str) m_autooff = re.search(r"AutoOff:\s*(\S+)", laser_str) if m_state: self._laser_state_label.configure(text=m_state.group(1).replace(",", "")) if m_autooff: self._laser_autooff_label.configure(text=m_autooff.group(1).replace(",", "")) # ─────────────────── 日志 ─────────────────── def _clear_log(self): self.log_text.configure(state=tk.NORMAL) self.log_text.delete("1.0", tk.END) self.log_text.configure(state=tk.DISABLED) def _log(self, message, tag="info", newline=True): self.log_text.configure(state=tk.NORMAL) if newline: self.log_text.insert(tk.END, f"{message}\n", tag) else: self.log_text.insert(tk.END, f"{message}", tag) self.log_text.see(tk.END) self.log_text.configure(state=tk.DISABLED) # ─────────────────── 通用串口升级协议(V1.1) ─────────────────── UPGRADE_HEADER = 0x7D UPGRADE_PKG_SIZE = 200 def crc16_modbus(data): """MODBUS CRC16，多项式0x8005(反射0xA001)，初值0xFFFF，低位在前。""" crc = 0xFFFF for b in data: crc ^= b for _ in range(8): if crc & 1: crc = (crc >> 1) ^ 0xA001 else: crc >>= 1 return crc def build_upgrade_frame(devmac, cmd, payload): """按协议组帧：帧头(0x7D)+DevMac(6)+Cmd(1)+PayloadLen(1)+Payload+Check(2, 小端)。""" body = bytes([UPGRADE_HEADER]) + devmac + bytes([cmd, len(payload)]) + payload crc = crc16_modbus(body) return body + bytes([crc & 0xFF, (crc >> 8) & 0xFF]) def parse_mac(mac_str): """从MAC字符串(如 09-00-01-7F-FF-FF)转为6字节bytes。""" parts = mac_str.strip().replace("-", ":").replace(" ", "").split(":") if len(parts) != 6: return None try: return bytes(int(p, 16) for p in parts) except ValueError: return None class UpgradeDialog: PACKAGE_SIZE_LOG = 8 def __init__(self, app, serial_port): self.app = app self.serial_port = serial_port self.port = serial_port.port self.top = tk.Toplevel(app.root) self.top.title("设备升级 - 通用串口升级协议V1.1") self.top.geometry("720x500") self.top.transient(app.root) self.top.protocol("WM_DELETE_WINDOW", self._on_close) self._orig_baud = None self._orig_timeout = None self._closed = False self.running = False # UI cfg = ttk.LabelFrame(self.top, text="升级设置", padding=8) cfg.pack(fill=tk.X, padx=8, pady=(8, 4)) ttk.Label(cfg, text="固件文件(.bin):").grid(row=0, column=0, sticky=tk.W, padx=2) self.fw_var = tk.StringVar() ttk.Entry(cfg, textvariable=self.fw_var, width=44).grid(row=0, column=1, padx=2) ttk.Button(cfg, text="浏览", command=self._choose_file).grid(row=0, column=2, padx=2) ttk.Label(cfg, text="设备MAC:").grid(row=1, column=0, sticky=tk.W, padx=2) default_mac = app._para_labels["本机MAC"].cget("text") self.mac_var = tk.StringVar(value=default_mac if default_mac != "-" else "") ttk.Entry(cfg, textvariable=self.mac_var, width=24).grid(row=1, column=1, sticky=tk.W, padx=2) ttk.Label(cfg, text="6字节从机MAC地址").grid(row=1, column=2, sticky=tk.W, padx=2) self.start_btn = ttk.Button(cfg, text="开始升级", command=self._start) self.start_btn.grid(row=2, column=1, sticky=tk.W, padx=2, pady=4) ttk.Button(cfg, text="关闭", command=self._on_close).grid(row=2, column=2, sticky=tk.W, padx=2, pady=4) self.progress = ttk.Progressbar(cfg, mode="determinate", maximum=100) self.progress.grid(row=3, column=0, columnspan=3, sticky=tk.W + tk.E, padx=2, pady=(4, 0)) self.prog_var = tk.StringVar(value="就绪") ttk.Label(cfg, textvariable=self.prog_var).grid(row=3, column=3, sticky=tk.W, padx=2) log_frame = ttk.LabelFrame(self.top, text="升级日志", padding=4) log_frame.pack(fill=tk.BOTH, expand=True, padx=8, pady=(4, 8)) self.log_text = scrolledtext.ScrolledText(log_frame, height=12, font=("Consolas", 8), state=tk.DISABLED) self.log_text.tag_configure("ok", foreground="green") self.log_text.tag_configure("err", foreground="red") self.log_text.pack(fill=tk.BOTH, expand=True) def _choose_file(self): path = filedialog.askopenfilename( title="选择固件文件", filetypes=[("固件文件", "*.bin"), ("所有文件", "*.*")]) if path: self.fw_var.set(path) def _log(self, msg, tag="info"): if self._closed: return tag = {"ok": "ok", "err": "err"}.get(tag, tag) self.log_text.configure(state=tk.NORMAL) self.log_text.insert(tk.END, f"{msg}\n", tag) self.log_text.see(tk.END) self.log_text.configure(state=tk.DISABLED) def _on_close(self): self.running = False self._closed = True self._restore_port() if self.top.winfo_exists(): self.top.destroy() def _restore_port(self): # 恢复波特率/超时，并恢复主程序接收 if self.serial_port: if self._orig_baud is not None: try: self.serial_port.baudrate = self._orig_baud except Exception: pass if self._orig_timeout is not None: try: self.serial_port.timeout = self._orig_timeout except Exception: pass self._orig_baud = None self._orig_timeout = None self.app.upgrading = False def _start(self): if self.running: return fw_path = self.fw_var.get().strip() mac_bytes = parse_mac(self.mac_var.get()) if not fw_path or not os.path.exists(fw_path): messagebox.showwarning("提示", "请先选择固件文件") return if not mac_bytes: messagebox.showwarning("提示", "设备MAC格式错误，应为 09-00-01-7F-FF-FF") return try: with open(fw_path, "rb") as f: self.fw_data = f.read() except Exception as e: messagebox.showerror("错误", f"读取固件失败: {e}") return if not self.fw_data: messagebox.showwarning("提示", "固件文件为空") return self.mac = mac_bytes self.fw_size = len(self.fw_data) self.package_num = (self.fw_size + UPGRADE_PKG_SIZE - 1) // UPGRADE_PKG_SIZE self.pkg_size = UPGRADE_PKG_SIZE self.running = True self.start_btn.configure(state=tk.DISABLED) self.progress.configure(maximum=100) self.progress["value"] = 0 # 暂停主程序接收，调整串口至升级参数，复用同一连接 self.app.upgrading = True self._orig_baud = self.serial_port.baudrate self._orig_timeout = self.serial_port.timeout self.serial_port.baudrate = 9600 self.serial_port.timeout = 0.2 threading.Thread(target=self._upgrade_worker, daemon=True).start() def _send_frame(self, cmd, payload): if not self.serial_port or not self.serial_port.is_open: return False frame = build_upgrade_frame(self.mac, cmd, payload) self.serial_port.write(frame) return True def _read_frame(self, timeout=0.5): """读取一帧；自动跳过前置的ASCII调试输出，按CRC校验成功才算完整帧。""" port = self.serial_port if not port or not port.is_open: return None buf = getattr(self, "_read_buf", b"") end = time.time() + timeout while time.time() < end: chunk = port.read(64) if chunk: buf += chunk while True: idx = buf.find(bytes([UPGRADE_HEADER])) if idx < 0: break hdr = idx + 1 if len(buf) < hdr + 8: # 需：mac(6)+cmd(1)+len(1) break cmd = buf[hdr + 6] plen = buf[hdr + 7] total = hdr + 8 + plen + 2 # +payload+check if len(buf) < total: break payload = buf[hdr + 8:hdr + 8 + plen] check = buf[hdr + 8 + plen:hdr + 8 + plen + 2] body = buf[idx:hdr + 8 + plen] got_crc = check[0] | (check[1] << 8) if got_crc == crc16_modbus(body): # 设备ASCII调试信息 -> 输出到主界面通讯日志 prefix = buf[0:idx] if any(0x20 <= b <= 0x7E for b in prefix): dbg = prefix.decode("utf-8", errors="replace").strip("\x00\r\n ") if dbg: self.app.root.after(0, self.app._log, f"[设备] {dbg}", "recv") self._read_buf = buf[total:] return cmd, payload # 误判的"帧头"，丢弃继续搜索 buf = buf[idx + 1:] if not chunk: time.sleep(0.02) self._read_buf = buf return None def _upgrade_worker(self): try: self._log(f"复用串口 {self.port} @9600,8,N,1 升级中...") # 1. 下发APP信息包 (Cmd 0x81) payload = struct.pack("= 6: break continue stall = 0 cmd, payload = frm if cmd == 0x02: if len(payload) < 4: continue idx, total = struct.unpack("= self.package_num: self._log("从机请求超出范围，视为完成", "ok") break start = idx * self.pkg_size data = self.fw_data[start:start + self.pkg_size] dl = len(data) p = struct.pack("= self.package_num: self._log(f"升级完成 (已发送全部 {sent_pkgs} 包)", "ok") self.prog_var.set("完成") self.progress["value"] = 100 else: self._log(f"升级未完成，已发送 {sent_pkgs}/{self.package_num} 包", "err") self.start_btn.configure(state=tk.NORMAL) def _error(self, msg): self.running = False self._log(msg, "err") self.start_btn.configure(state=tk.NORMAL) def main(): root = tk.Tk() app = LaserTracingConfigApp(root) def on_closing(): app._disconnect() root.destroy() root.protocol("WM_DELETE_WINDOW", on_closing) root.mainloop() if __name__ == "__main__": main()