From 192db942d775c1ac27e38f02ae08aa74cf926edb Mon Sep 17 00:00:00 2001 From: YuanHongbin <975559679@qq.com> Date: Thu, 25 Jun 2026 09:49:14 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=BF=80=E5=85=89=E7=A4=BA?= =?UTF-8?q?=E8=B8=AA=E9=85=8D=E7=BD=AE=E5=B7=A5=E5=85=B7=E7=95=8C=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- laser_tracing_config.py | 385 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 385 insertions(+) create mode 100644 laser_tracing_config.py diff --git a/laser_tracing_config.py b/laser_tracing_config.py new file mode 100644 index 0000000..e17ff21 --- /dev/null +++ b/laser_tracing_config.py @@ -0,0 +1,385 @@ +""" +激光示踪设备网关配置工具 +通过串口(RS485)直接修改和查看网关参数 +""" + +import tkinter as tk +from tkinter import ttk, scrolledtext, messagebox +import serial +import serial.tools.list_ports +import threading +import time +import re +from datetime import datetime + + +class LaserTracingConfigApp: + def __init__(self, root): + self.root = root + self.root.title("激光示踪配置工具") + self.root.geometry("960x720") + self.root.resizable(True, True) + + self.serial_port = None + self.serial_thread = None + self.running = False + self.recv_buffer = "" + self._port_details = {} + + self._build_ui() + self._refresh_ports() + + # ─────────────────── 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=(12, 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=(12, 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=(12, 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=(12, 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 < 525000000: + 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) + + # 查询按钮行 + 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.Label(query_frame, text="配置后可通过para命令验证是否成功", + foreground="gray").pack(side=tk.LEFT, padx=8) + + # 参数显示区域 + para_frame = ttk.LabelFrame(self.root, text="设备参数信息", padding=8) + para_frame.pack(fill=tk.BOTH, expand=True, padx=8, pady=4) + + # 使用Treeview展示解析后的参数（双列无表头，不可选定） + columns = ("参数", "值") + self.para_tree = ttk.Treeview(para_frame, columns=columns, show="tree headings", + height=10, selectmode="none") + self.para_tree.heading("#0", text="") + self.para_tree.heading("参数", text="") + self.para_tree.heading("值", text="") + self.para_tree.column("#0", width=0, stretch=False) + self.para_tree.column("参数", width=180, anchor=tk.W) + self.para_tree.column("值", width=380, anchor=tk.W) + tree_scroll = ttk.Scrollbar(para_frame, orient=tk.VERTICAL, command=self.para_tree.yview) + self.para_tree.configure(yscrollcommand=tree_scroll.set) + self.para_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) + tree_scroll.pack(side=tk.RIGHT, fill=tk.Y) + + # 预填充参数表格（值为"-"待读取后更新） + self._para_order = ["软件版本", "硬件版本", "本机MAC", "电池电量", "设备类型", + "通信方式", "设备标识", "网关MAC", "注册状态", + "RS485_1", "RS485_2", "Lora参数", "低功耗配置", + "设备时间", "错误信息"] + self._para_value_items = {} + for key in self._para_order: + item_id = self.para_tree.insert("", tk.END, values=(key, "-")) + self._para_value_items[key] = item_id + + # 通讯日志区域 + 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.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回应（以***结尾） + if "**********************************************************" in self.recv_buffer: + 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 _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): + # 清空Treeview + for item in self.para_tree.get_children(): + self.para_tree.delete(item) + + lines = text.replace("\r\n", "\n").replace("\r", "\n").split("\n") + + key_value_patterns = [ + (r"SoftWare V:(.*)", "软件版本"), + (r"HardWare V:(.*)", "硬件版本"), + (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"RS485h1:\s*(.*)", "RS485_1"), + (r"RS485h2:\s*(.*)", "RS485_2"), + (r"Lora:\s*(.*)", "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() + + # 解析日期时间 + dt_match = re.search(r"(\d{4}/\d{1,2}/\d{1,2}-\d{1,2}:\d{2}:\d{2})", text) + if dt_match: + found["设备时间"] = dt_match.group(1) + + # 检查Error + err_match = re.search(r"Error:\s*(.*)", text) + if err_match: + found["错误信息"] = err_match.group(1).strip() + + # 重建参数表格（每次收到完整响应时） + for item in self.para_tree.get_children(): + self.para_tree.delete(item) + self._para_value_items = {} + for key in self._para_order: + value = found.get(key, "-") + item_id = self.para_tree.insert("", tk.END, values=(key, value)) + self._para_value_items[key] = item_id + + # 解析Lora详细参数并更新配置界面 + if "Lora参数" in found: + lora_str = found["Lora参数"] + fc_match = re.search(r"fc\s+(\d+)", lora_str) + if fc_match: + fc_val = fc_match.group(1) + 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) + + # 解析RS485波特率并提示 + if "RS485_1" in found: + baud_match = re.search(r"Baudrate\s+(\d+)", found["RS485_1"]) + if baud_match: + self.baudrate_var.set(baud_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): + 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) + else: + self.log_text.insert(tk.END, f"[{timestamp}] {message}", tag) + self.log_text.see(tk.END) + self.log_text.configure(state=tk.DISABLED) + + +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()