LaserTracing_ConfigureAPP/laser_tracing_config.py

"""
激光示踪设备网关配置工具
通过串口(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("900x700")
        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=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 < 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)

        # 参数显示区域 - 改用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")

    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):
        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 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):
        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()