diff --git a/laser_tracing_config.py b/laser_tracing_config.py index a578b7a..09ee02b 100644 --- a/laser_tracing_config.py +++ b/laser_tracing_config.py @@ -4,7 +4,9 @@ """ import tkinter as tk -from tkinter import ttk, scrolledtext, messagebox +from tkinter import ttk, scrolledtext, messagebox, filedialog +import struct +import binascii import serial import serial.tools.list_ports import threading @@ -12,6 +14,8 @@ import time import re import sys import os +import shutil +import subprocess from datetime import datetime @@ -25,7 +29,6 @@ class LaserTracingConfigApp: def __init__(self, root): self.root = root self.root.title("边坡变形监测-变形(位移)自动测量系统配置工具") - self.root.geometry("900x700") self.root.resizable(True, True) icon_path = resource_path("传感器.ico") @@ -35,12 +38,17 @@ class LaserTracingConfigApp: 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): @@ -87,7 +95,7 @@ class LaserTracingConfigApp: fc_displays = [] freq = 410100000 ch = 1 - while freq < 525000000: + while freq <= 492900000: display = f"信道{ch}: {freq / 1000000:.1f}MHz" fc_displays.append(display) self._fc_display_to_value[display] = str(freq) @@ -99,11 +107,38 @@ class LaserTracingConfigApp: 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) @@ -156,6 +191,37 @@ class LaserTracingConfigApp: 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)) @@ -257,6 +323,9 @@ class LaserTracingConfigApp: 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: @@ -275,8 +344,14 @@ class LaserTracingConfigApp: def _on_data_received(self, text): self._log(text, "recv", newline=False) - # 检查是否收到完整para回应（以***结尾） - if "**********************************************************" in self.recv_buffer: + # 等待 dch ch1 切换完成后再发送 para + if getattr(self, "_pending_dch", False) and "Debug channel switched to RS485_1" in self.recv_buffer: + self._pending_dch = False + self._send_command("para") + return + # 收到完整para回应：起始与结束各出现一次分隔符（****）后才解析 + marker = "**********************************************************" + if self.recv_buffer.count(marker) >= 2: self._parse_para_response(self.recv_buffer) self.recv_buffer = "" @@ -300,7 +375,15 @@ class LaserTracingConfigApp: # ─────────────────── 指令发送 ─────────────────── def _send_para(self): - self._send_command("para") + self._pending_dch = True + self._send_command("dch ch1") + + 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() @@ -310,14 +393,40 @@ class LaserTracingConfigApp: 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"[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*(.*)", "设备类型"), @@ -327,6 +436,8 @@ class LaserTracingConfigApp: (r"Register Status:\s*(.*)", "注册状态"), (r"Lora.*?fc\s*(\d+)", "Lora频率"), (r"LpCfg:\s*(.*)", "低功耗配置"), + (r"FpoCfg:\s*(.*)", "全功率工作时间段"), + (r"Laser:\s*(.*)", "激光"), ] found = {} @@ -388,6 +499,30 @@ class LaserTracingConfigApp: 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): @@ -396,16 +531,297 @@ class LaserTracingConfigApp: self.log_text.configure(state=tk.DISABLED) def _log(self, message, tag="info", newline=True): - timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3] self.log_text.configure(state=tk.NORMAL) if newline: - self.log_text.insert(tk.END, f"[{timestamp}] {message}\n", tag) + self.log_text.insert(tk.END, f"{message}\n", tag) else: - self.log_text.insert(tk.END, f"[{timestamp}] {message}", tag) + 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) diff --git a/传感器通用串口升级协议V1.1.xlsx b/传感器通用串口升级协议V1.1.xlsx new file mode 100644 index 0000000..899b38c Binary files /dev/null and b/传感器通用串口升级协议V1.1.xlsx differ diff --git a/通用升级程序.exe b/通用升级程序.exe new file mode 100644 index 0000000..804aedc Binary files /dev/null and b/通用升级程序.exe differ