733 lines
30 KiB
Python
733 lines
30 KiB
Python
"""
|
|
激光示踪设备网关配置工具
|
|
通过串口(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")
|
|
self.baudrate_combo = ttk.Combobox(conn_frame, textvariable=self.baudrate_var, width=8,
|
|
values=["2400", "4800", "9600", "19200", "38400", "57600", "115200"])
|
|
self.baudrate_combo.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 = 470100000
|
|
ch = 0
|
|
while freq <= 492600000:
|
|
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 += 500000
|
|
ch += 1
|
|
self.lora_fc_display_var = tk.StringVar(value=self._fc_value_to_display.get("470100000", 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)
|
|
|
|
# 查询按钮行
|
|
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
|
|
|
|
# 通讯日志区域
|
|
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("<Return>", 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")
|
|
except ValueError:
|
|
messagebox.showwarning("参数错误", "波特率必须为数字")
|
|
self._log("波特率必须为数字", "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 _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*(.*)", "低功耗配置"),
|
|
]
|
|
|
|
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 == "注册状态" and value != "-":
|
|
raw = value.strip().lower()
|
|
value = "已注册" if raw in ("1", "registered", "reg", "ok", "true", "yes", "success") else "未注册"
|
|
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)))
|
|
|
|
# ─────────────────── 日志 ───────────────────
|
|
|
|
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("<HI", self.package_num & 0xFFFF, self.fw_size)
|
|
payload += struct.pack("<I", binascii.crc32(self.fw_data) & 0xFFFFFFFF)
|
|
self._log(f"下发APP信息: 包数={self.package_num}, 大小={self.fw_size}字节")
|
|
self._send_frame(0x81, payload)
|
|
# 传感器会先打印调试信息，随后直接以 0x02 请求数据包；无需单独的 0x01 应答
|
|
self._log("等待设备握手后接收数据请求...")
|
|
|
|
sent = set()
|
|
stall = 0
|
|
while self.running:
|
|
frm = self._read_frame(1.5)
|
|
if frm is None:
|
|
stall += 1
|
|
if stall >= 6:
|
|
break
|
|
continue
|
|
stall = 0
|
|
cmd, payload = frm
|
|
if cmd == 0x02:
|
|
if len(payload) < 4:
|
|
continue
|
|
idx, total = struct.unpack("<HH", payload[0:4])
|
|
if idx >= 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("<HHB", idx, self.package_num & 0xFFFF, dl) + data
|
|
self._send_frame(0x82, p)
|
|
if idx not in sent:
|
|
sent.add(idx)
|
|
self.app.root.after(0, self._update_progress, len(sent))
|
|
elif cmd == 0x01:
|
|
self._log("收到从机应答(0x01)", "ok")
|
|
else:
|
|
self._log(f"收到其他帧 Cmd={cmd:#04x}", "err")
|
|
self.app.root.after(0, self._finish, len(sent))
|
|
except Exception as e:
|
|
self.app.root.after(0, self._error, f"升级出错: {e}")
|
|
finally:
|
|
self._restore_port()
|
|
|
|
def _update_progress(self, sent_pkgs):
|
|
if self._closed:
|
|
return
|
|
pct = sent_pkgs * 100.0 / self.package_num if self.package_num else 0
|
|
self.progress["value"] = pct
|
|
self.prog_var.set(f"{sent_pkgs}/{self.package_num} 包")
|
|
|
|
def _finish(self, sent_pkgs):
|
|
self.running = False
|
|
if sent_pkgs >= 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()
|