#!/usr/bin/env python3 """Tkinter upper-computer test application for the PLSR Modbus firmware.""" from __future__ import annotations import queue import threading import time import tkinter as tk from pathlib import Path from tkinter import filedialog, messagebox, ttk from typing import Any, Optional from plsr_test import ( COMMAND_CLEAR_COUNT, COMMAND_IMMEDIATE_STOP, COMMAND_RESTART, COMMAND_RESET, COMMAND_STOP, CommunicationError, ModbusRtuClient, PlsrError, Snapshot, MAP_SPEC, MAP_LEGACY, REG_SEGMENT_BASE, REG_MONITOR_TOTAL_PULSES, REG_CONTROL, PlsrConfiguration, SegmentParameters, validate_frequency, list_ports, ) APP_TITLE = "PLSR 单轴脉冲测试台" POLL_INTERVAL_MS = 500 MAX_LOG_LINES = 800 STATE_TEXT = { 0: "空闲", 1: "准备", 2: "运行中", 3: "等待", 4: "减速中", 5: "完成", 6: "已停止", 7: "错误", } STOP_REASON_TEXT = { 0: "0 · 无", 1: "1 · 受控停止", 2: "2 · 立即停止", } ERROR_TEXT = { 0: "0 · 无错误", 1: "1 · 参数无效", 2: "2 · 设备忙", 3: "3 · 定时器资源错误", } REGISTER_NAMES = { 0: "运行状态", 1: "状态标志", 3: "当前频率 Hz", 5: "停止原因", 6: "最后错误", 10: "累计脉冲高字", 11: "累计脉冲低字", 100: "命令", 102: "目标脉冲高字", 103: "目标脉冲低字", 104: "设定频率 Hz", 105: "方向 0正/1反", 106: "脉冲输出 Q 点", 107: "方向输出 Q 点", } SPEC_REGISTER_NAMES = { 0x1000: "脉冲发送端子 Y0..Y3", 0x1001: "脉冲方向端子 Y12..Y15", 0x1002: "WAIT 信号 X4/X5", 0x1003: "EXT 信号 X4/X5", 0x1004: "发送模式", 0x1005: "方向延时 ms", 0x1006: "方向逻辑", 0x1007: "加减速模式", 0x1008: "运行模式", 0x1009: "总段数", 0x100A: "起始执行段", 0x100B: "默认速度高字", 0x100C: "默认速度低字", 0x100D: "起始速度高字", 0x100E: "起始速度低字", 0x1010: "终止速度高字", 0x1011: "终止速度低字", 0x1012: "加速时间 ms", 0x1013: "减速时间 ms", 0x2000: "累计脉冲高字", 0x2001: "累计脉冲低字", 0x2002: "频率高字", 0x2003: "频率低字", 0x2004: "运行状态", 0x2005: "当前执行段", 0x2006: "错误码", 0x2007: "停止原因", 0x3000: "控制 bit0启动/bit1停止/bit2清零/bit3重启/bit4复位/bit5立即停止", } class GuiModbusClient(ModbusRtuClient): """Send raw-frame messages to Tk through a thread-safe queue.""" def __init__(self, *args: Any, event_queue: queue.Queue, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self._event_queue = event_queue def _log_frame(self, prefix: str, frame: bytes) -> None: now = time.time() timestamp = time.strftime("%H:%M:%S", time.localtime(now)) + f".{int(now % 1 * 1000):03d}" self._event_queue.put( { "type": "log", "line": f"{timestamp} {prefix} {frame.hex(' ').upper()}", } ) class SerialWorker(threading.Thread): """Own the serial object so only one request can be active at a time.""" def __init__(self, jobs: queue.Queue, events: queue.Queue) -> None: super().__init__(name="plsr-serial-worker", daemon=True) self.jobs = jobs self.events = events self.client: Optional[GuiModbusClient] = None def _require_client(self) -> GuiModbusClient: if self.client is None: raise CommunicationError("设备尚未连接") return self.client def run(self) -> None: while True: job = self.jobs.get() name = job["name"] if name == "shutdown": if self.client is not None: self.client.close() return try: result: Any = None if name == "connect": if self.client is not None: self.client.close() self.client = GuiModbusClient( port=job["port"], baud_rate=job["baud"], slave=job["slave"], timeout=0.6, retries=2, verbose=False, event_queue=self.events, protocol=job.get("protocol", MAP_SPEC), ) try: self.client.open() result = self.client.read_snapshot() except Exception: self.client.close() self.client = None raise elif name == "disconnect": if self.client is not None: self.client.close() self.client = None elif name == "snapshot": result = self._require_client().read_snapshot() elif name == "read_config": result = self._require_client().read_configuration() elif name == "write_config": client = self._require_client() client.write_configuration(job["config"]) result = client.read_configuration() elif name == "write_config_command": client = self._require_client() client.write_configuration(job["config"]) client.command(job["command"]) result = client.read_snapshot() elif name == "raw": client = self._require_client() if client.protocol == MAP_SPEC: result = {"config": client.read_holding(0x1000, 0x14), "segments": [client.read_holding(REG_SEGMENT_BASE + i * 0x10, 8) for i in range(10)], "monitor": client.read_holding(REG_MONITOR_TOTAL_PULSES, 8), "control": client.read_holding(REG_CONTROL, 1)} else: result = {"status": client.read_holding(0, 32), "control": client.read_holding(100, 8)} elif name == "configure": result = self._require_client().configure_outputs( job["pulse_output"], job["direction_output"], save=True ) elif name == "start": client = self._require_client() client.start_motion(job["pulses"], job["frequency"], job["direction"], job.get("restart", False)) result = client.read_snapshot() elif name == "frequency": result = self._require_client().set_running_frequency( job["frequency"] ) elif name == "command": client = self._require_client() client.command(job["command"]) result = client.read_snapshot() else: raise ValueError(f"未知后台任务: {name}") self.events.put( { "type": "result", "name": name, "purpose": job.get("purpose", ""), "result": result, } ) except Exception as exc: self.events.put( { "type": "error", "name": name, "purpose": job.get("purpose", ""), "error": str(exc), } ) class PlsrGui(tk.Tk): def __init__(self) -> None: super().__init__() self.title(f"{APP_TITLE} - Python") self.geometry("1120x820") self.minsize(960, 680) self.configure(background="#f2f5f8") self.jobs: queue.Queue = queue.Queue() self.events: queue.Queue = queue.Queue() self.worker = SerialWorker(self.jobs, self.events) self.worker.start() self.connected = False self.job_active = False self.poll_active = False self.deferred_job: Optional[dict[str, Any]] = None self.last_snapshot: Optional[Snapshot] = None self.communication_errors = 0 self.closing = False self.log_line_count = 0 self.auto_phase: Optional[str] = None self.run_baseline = 0 self.run_target = 0 self.motion_plan: Optional[tuple[int, int, int]] = None self.lamp_test_baseline = 0 self._create_variables() self._configure_style() self._build_ui() self._refresh_ports() self._update_controls() self.protocol("WM_DELETE_WINDOW", self._on_close) self.after(50, self._drain_events) self.after(POLL_INTERVAL_MS, self._poll_tick) def _create_variables(self) -> None: self.port_var = tk.StringVar(value="COM9") self.baud_var = tk.StringVar(value="115200") self.slave_var = tk.StringVar(value="1") self.connection_var = tk.StringVar(value="未连接") self.operation_var = tk.StringVar(value="等待连接") self.pulse_output_var = tk.StringVar(value="Y0") self.direction_output_var = tk.StringVar(value="Y12") self.pulses_var = tk.StringVar(value="10") self.frequency_var = tk.StringVar(value="2") self.direction_var = tk.IntVar(value=0) self.capture_frames_var = tk.BooleanVar(value=True) self.protocol_var = tk.StringVar(value=MAP_SPEC) self.wait_input_var = tk.StringVar(value="X4") self.ext_input_var = tk.StringVar(value="X4") self.send_mode_var = tk.StringVar(value="完成模式") self.direction_delay_var = tk.StringVar(value="0") self.direction_logic_var = tk.StringVar(value="正逻辑") self.accel_mode_var = tk.StringVar(value="直线") self.run_mode_var = tk.StringVar(value="相对") self.segment_count_var = tk.StringVar(value="1") self.start_segment_var = tk.StringVar(value="1") self.default_speed_var = tk.StringVar(value="1000") self.start_speed_var = tk.StringVar(value="1") self.end_speed_var = tk.StringVar(value="1") self.accel_time_var = tk.StringVar(value="0") self.decel_time_var = tk.StringVar(value="0") self.segment_index_var = tk.StringVar(value="1") self.segment_frequency_var = tk.StringVar(value="1000") self.segment_pulses_var = tk.StringVar(value="10") self.segment_wait_var = tk.StringVar(value="WAIT时间") self.segment_wait_time_var = tk.StringVar(value="0") self.segment_act_time_var = tk.StringVar(value="0") self.segment_jump_var = tk.StringVar(value="0") self.segment_models = [SegmentParameters() for _ in range(10)] self.state_var = tk.StringVar(value="--") self.flags_var = tk.StringVar(value="--") self.current_frequency_var = tk.StringVar(value="--") self.total_pulses_var = tk.StringVar(value="--") self.stop_reason_var = tk.StringVar(value="--") self.error_var = tk.StringVar(value="--") self.saved_outputs_var = tk.StringVar(value="--") self.progress_text_var = tk.StringVar(value="0 / 0") def _configure_style(self) -> None: style = ttk.Style(self) if "clam" in style.theme_names(): style.theme_use("clam") default_font = ("Microsoft YaHei UI", 9) self.option_add("*Font", default_font) style.configure("TFrame", background="#f2f5f8") style.configure("Surface.TFrame", background="#ffffff") style.configure("TLabel", background="#ffffff", foreground="#20364d") style.configure("TCheckbutton", background="#ffffff", foreground="#20364d") style.configure("TRadiobutton", background="#ffffff", foreground="#20364d") style.configure("Header.TLabel", background="#f2f5f8", foreground="#123354", font=("Microsoft YaHei UI", 20, "bold")) style.configure("Subtitle.TLabel", background="#f2f5f8", foreground="#61758a") style.configure("Section.TLabel", background="#ffffff", foreground="#173b61", font=("Microsoft YaHei UI", 10, "bold")) style.configure("Value.TLabel", background="#ffffff", foreground="#102f4f", font=("Microsoft YaHei UI", 10, "bold")) style.configure("StatusOff.TLabel", background="#fdebea", foreground="#b42318", padding=(12, 7), font=("Microsoft YaHei UI", 9, "bold")) style.configure("StatusOn.TLabel", background="#e7f6ec", foreground="#167044", padding=(12, 7), font=("Microsoft YaHei UI", 9, "bold")) style.configure("Primary.TButton", foreground="#ffffff", background="#1f6fa8", padding=(10, 6), font=("Microsoft YaHei UI", 9, "bold")) style.map("Primary.TButton", background=[("active", "#185d8e"), ("disabled", "#aab7c2")]) style.configure("Danger.TButton", foreground="#ffffff", background="#c93b3b", padding=(10, 6), font=("Microsoft YaHei UI", 9, "bold")) style.map("Danger.TButton", background=[("active", "#a92f2f"), ("disabled", "#c9b2b2")]) style.configure("TLabelframe", background="#ffffff", padding=10) style.configure("TLabelframe.Label", background="#ffffff", foreground="#173b61", font=("Microsoft YaHei UI", 10, "bold")) style.configure("TNotebook", background="#f2f5f8", borderwidth=0) style.configure("TNotebook.Tab", padding=(16, 7)) def _build_ui(self) -> None: outer = ttk.Frame(self, padding=(22, 14, 22, 12)) outer.grid(row=0, column=0, sticky="nsew") self.rowconfigure(0, weight=1) self.columnconfigure(0, weight=1) outer.columnconfigure(0, weight=1) outer.rowconfigure(3, weight=1) header = ttk.Frame(outer) header.grid(row=0, column=0, sticky="ew", pady=(0, 10)) header.columnconfigure(0, weight=1) ttk.Label(header, text=APP_TITLE, style="Header.TLabel").grid(row=0, column=0, sticky="w") ttk.Label( header, text="XDM-60T4-E / STM32 · Modbus RTU · 可配置并掉电保存 Q 点", style="Subtitle.TLabel", ).grid(row=1, column=0, sticky="w", pady=(2, 0)) self.connection_label = ttk.Label(header, textvariable=self.connection_var, style="StatusOff.TLabel") self.connection_label.grid(row=0, column=1, rowspan=2, sticky="e") connection = ttk.LabelFrame(outer, text="串口连接") connection.grid(row=1, column=0, sticky="ew", pady=(0, 10)) connection.columnconfigure(8, weight=1) ttk.Label(connection, text="端口").grid(row=0, column=0, padx=(0, 6)) self.port_combo = ttk.Combobox(connection, textvariable=self.port_var, width=12, state="readonly") self.port_combo.grid(row=0, column=1, padx=(0, 8)) self.refresh_button = ttk.Button(connection, text="刷新", command=self._refresh_ports, width=9) self.refresh_button.grid(row=0, column=2, padx=(0, 20)) ttk.Label(connection, text="波特率").grid(row=0, column=3, padx=(0, 6)) self.baud_combo = ttk.Combobox(connection, textvariable=self.baud_var, values=("9600", "19200", "38400", "57600", "115200"), width=10, state="readonly") self.baud_combo.grid(row=0, column=4, padx=(0, 18)) ttk.Label(connection, text="格式 8-E-1").grid(row=0, column=5, padx=(0, 18)) ttk.Label(connection, text="站号").grid(row=0, column=6, padx=(0, 6)) self.slave_spin = ttk.Spinbox(connection, textvariable=self.slave_var, from_=1, to=247, width=7) self.slave_spin.grid(row=0, column=7) ttk.Label(connection, text="映射").grid(row=0, column=8, padx=(16, 6)) self.protocol_combo = ttk.Combobox(connection, textvariable=self.protocol_var, values=(MAP_SPEC, MAP_LEGACY), width=9, state="readonly") self.protocol_combo.grid(row=0, column=9, padx=(0, 8)) self.connect_button = ttk.Button(connection, text="连接设备", command=self._toggle_connection, style="Primary.TButton", width=14) self.connect_button.grid(row=0, column=10, sticky="e", padx=(8, 0)) self.operation_label = ttk.Label(outer, textvariable=self.operation_var, style="Subtitle.TLabel") self.operation_label.grid(row=2, column=0, sticky="w", pady=(0, 6)) notebook = ttk.Notebook(outer) notebook.grid(row=3, column=0, sticky="nsew") control_tab = ttk.Frame(notebook, style="Surface.TFrame", padding=10) config_tab = ttk.Frame(notebook, style="Surface.TFrame", padding=10) register_tab = ttk.Frame(notebook, style="Surface.TFrame", padding=10) log_tab = ttk.Frame(notebook, style="Surface.TFrame", padding=10) notebook.add(control_tab, text="控制与状态") notebook.add(config_tab, text="完整参数与多段") notebook.add(register_tab, text="寄存器诊断") notebook.add(log_tab, text="通信日志") self._build_control_tab(control_tab) self._build_config_tab(config_tab) self._build_register_tab(register_tab) self._build_log_tab(log_tab) notebook.select(control_tab) ttk.Label( outer, text="安全提示：首次发脉冲前请断开机械负载，仅连接测试灯或示波器；先低频验证 Q 点和方向电平。", foreground="#a13b22", background="#f2f5f8", ).grid(row=4, column=0, sticky="w", pady=(9, 0)) def _build_control_tab(self, parent: ttk.Frame) -> None: parent.columnconfigure(0, weight=1, uniform="panel") parent.columnconfigure(1, weight=1, uniform="panel") parent.rowconfigure(0, weight=1) left = ttk.Frame(parent, style="Surface.TFrame") right = ttk.Frame(parent, style="Surface.TFrame") left.grid(row=0, column=0, sticky="nsew", padx=(0, 7)) right.grid(row=0, column=1, sticky="nsew", padx=(7, 0)) left.columnconfigure(0, weight=1) right.columnconfigure(0, weight=1) output_box = ttk.LabelFrame(left, text="模块输出映射（保存到设备 Flash）") output_box.grid(row=0, column=0, sticky="ew", pady=(0, 10)) output_box.columnconfigure(1, weight=1) output_box.columnconfigure(3, weight=1) ttk.Label(output_box, text="脉冲点").grid(row=0, column=0, sticky="w", padx=(0, 6)) self.pulse_output_combo = ttk.Combobox(output_box, textvariable=self.pulse_output_var, values=tuple(f"Y{i}" for i in range(4)), state="readonly", width=5) self.pulse_output_combo.grid(row=0, column=1, sticky="ew", padx=(0, 16)) ttk.Label(output_box, text="方向点").grid(row=0, column=2, sticky="w", padx=(0, 6)) self.direction_output_combo = ttk.Combobox(output_box, textvariable=self.direction_output_var, values=tuple(f"Y{12 + i}" for i in range(4)), state="readonly", width=5) self.direction_output_combo.grid(row=0, column=3, sticky="ew", padx=(0, 12)) self.save_outputs_button = ttk.Button(output_box, text="写入并保存", command=self._save_outputs, style="Primary.TButton") self.save_outputs_button.grid(row=0, column=4) ttk.Label(output_box, text="保存成功后断电不会丢失，下次上电自动恢复。", foreground="#61758a", background="#ffffff").grid(row=1, column=0, columnspan=5, sticky="w", pady=(9, 0)) motion_box = ttk.LabelFrame(left, text="运动参数") motion_box.grid(row=1, column=0, sticky="ew", pady=(0, 10)) motion_box.columnconfigure(1, weight=1) ttk.Label(motion_box, text="目标脉冲数").grid(row=0, column=0, sticky="w", padx=(0, 12), pady=5) self.pulses_spin = ttk.Spinbox(motion_box, textvariable=self.pulses_var, from_=-2147483648, to=2147483647) self.pulses_spin.grid(row=0, column=1, sticky="ew", pady=5) ttk.Label(motion_box, text="频率").grid(row=1, column=0, sticky="w", padx=(0, 12), pady=5) freq_line = ttk.Frame(motion_box, style="Surface.TFrame") freq_line.grid(row=1, column=1, sticky="ew", pady=5) freq_line.columnconfigure(0, weight=1) self.frequency_spin = ttk.Spinbox(freq_line, textvariable=self.frequency_var, from_=1, to=100000) self.frequency_spin.grid(row=0, column=0, sticky="ew") ttk.Label(freq_line, text="Hz（1–100000）", background="#ffffff").grid(row=0, column=1, padx=(8, 0)) self.apply_frequency_button = ttk.Button( freq_line, text="应用频率", command=self._apply_running_frequency ) self.apply_frequency_button.grid(row=0, column=2, padx=(10, 0)) ttk.Label(motion_box, text="运行方向").grid(row=2, column=0, sticky="w", padx=(0, 12), pady=5) direction_line = ttk.Frame(motion_box, style="Surface.TFrame") direction_line.grid(row=2, column=1, sticky="w", pady=5) self.forward_radio = ttk.Radiobutton(direction_line, text="正向（0）", variable=self.direction_var, value=0) self.reverse_radio = ttk.Radiobutton(direction_line, text="反向（1）", variable=self.direction_var, value=1) self.forward_radio.grid(row=0, column=0, padx=(0, 18)) self.reverse_radio.grid(row=0, column=1) actions = ttk.LabelFrame(left, text="运行控制") actions.grid(row=2, column=0, sticky="ew") for column in range(3): actions.columnconfigure(column, weight=1) self.start_button = ttk.Button(actions, text="启动/继续脉冲", command=self._start_motion, style="Primary.TButton") self.stop_button = ttk.Button(actions, text="受控停止", command=lambda: self._send_command(COMMAND_STOP, "stop")) self.emergency_button = ttk.Button(actions, text="重新运行脉冲", command=self._restart_motion, style="Danger.TButton") self.start_button.grid(row=0, column=0, sticky="ew", padx=(0, 6), pady=(0, 7)) self.stop_button.grid(row=0, column=1, sticky="ew", padx=6, pady=(0, 7)) self.emergency_button.grid(row=0, column=2, sticky="ew", padx=(6, 0), pady=(0, 7)) self.lamp_test_button = ttk.Button(actions, text="低速正反转灯测", command=self._start_lamp_test) self.reset_button = ttk.Button(actions, text="复位状态", command=lambda: self._send_command(COMMAND_RESET, "reset")) self.clear_button = ttk.Button(actions, text="清零累计脉冲", command=lambda: self._send_command(COMMAND_CLEAR_COUNT, "clear")) self.lamp_test_button.grid(row=1, column=0, sticky="ew", padx=(0, 6)) self.reset_button.grid(row=1, column=1, sticky="ew", padx=6) self.clear_button.grid(row=1, column=2, sticky="ew", padx=(6, 0)) status_box = ttk.LabelFrame(right, text="实时状态") status_box.grid(row=0, column=0, sticky="nsew") status_box.columnconfigure(1, weight=1) rows = ( ("运行状态", self.state_var), ("状态标志", self.flags_var), ("当前频率", self.current_frequency_var), ("累计脉冲", self.total_pulses_var), ("停止原因", self.stop_reason_var), ("最后错误", self.error_var), ("设备当前 Q 点", self.saved_outputs_var), ) for row, (label, variable) in enumerate(rows): ttk.Label(status_box, text=label).grid(row=row, column=0, sticky="w", padx=(0, 24), pady=5) ttk.Label(status_box, textvariable=variable, style="Value.TLabel").grid(row=row, column=1, sticky="e", pady=5) ttk.Separator(status_box).grid(row=len(rows), column=0, columnspan=2, sticky="ew", pady=(5, 8)) ttk.Label(status_box, text="本次运动进度").grid(row=len(rows) + 1, column=0, columnspan=2, sticky="w") self.progress = ttk.Progressbar(status_box, maximum=100, value=0) self.progress.grid(row=len(rows) + 2, column=0, columnspan=2, sticky="ew", pady=(8, 4)) ttk.Label(status_box, textvariable=self.progress_text_var, background="#ffffff", foreground="#61758a").grid(row=len(rows) + 3, column=0, columnspan=2, sticky="e") self.read_now_button = ttk.Button(status_box, text="立即读取状态", command=lambda: self._submit("snapshot", purpose="manual")) self.read_now_button.grid(row=len(rows) + 4, column=0, columnspan=2, sticky="ew", pady=(10, 0)) def _build_register_tab(self, parent: ttk.Frame) -> None: parent.rowconfigure(1, weight=1) parent.columnconfigure(0, weight=1) toolbar = ttk.Frame(parent, style="Surface.TFrame") toolbar.grid(row=0, column=0, sticky="ew", pady=(0, 8)) toolbar.columnconfigure(0, weight=1) ttk.Label(toolbar, text="需求版读取 1000/1100/2000/3000；legacy 读取 0/100。", background="#ffffff", foreground="#61758a").grid(row=0, column=0, sticky="w") self.raw_button = ttk.Button(toolbar, text="刷新寄存器", command=lambda: self._submit("raw", purpose="raw"), style="Primary.TButton") self.raw_button.grid(row=0, column=1, sticky="e") tree_frame = ttk.Frame(parent, style="Surface.TFrame") tree_frame.grid(row=1, column=0, sticky="nsew") tree_frame.rowconfigure(0, weight=1) tree_frame.columnconfigure(0, weight=1) self.register_tree = ttk.Treeview(tree_frame, columns=("address", "hex", "value", "meaning"), show="headings", selectmode="browse") self.register_tree.heading("address", text="地址") self.register_tree.heading("hex", text="十六进制") self.register_tree.heading("value", text="无符号值") self.register_tree.heading("meaning", text="项目定义") self.register_tree.column("address", width=90, anchor="center", stretch=False) self.register_tree.column("hex", width=130, anchor="center", stretch=False) self.register_tree.column("value", width=140, anchor="e", stretch=False) self.register_tree.column("meaning", width=340, anchor="w") scroll = ttk.Scrollbar(tree_frame, orient="vertical", command=self.register_tree.yview) self.register_tree.configure(yscrollcommand=scroll.set) self.register_tree.grid(row=0, column=0, sticky="nsew") scroll.grid(row=0, column=1, sticky="ns") def _build_config_tab(self, parent: ttk.Frame) -> None: parent.columnconfigure(0, weight=1) parent.rowconfigure(1, weight=1) global_box = ttk.LabelFrame(parent, text="系统与运动参数（0x1000..0x1013）") global_box.grid(row=0, column=0, sticky="ew", pady=(0, 10)) for column in (1, 3, 5, 7): global_box.columnconfigure(column, weight=1) fields = ( ("WAIT信号", self.wait_input_var, ("X4", "X5")), ("EXT信号", self.ext_input_var, ("X4", "X5")), ("发送模式", self.send_mode_var, ("完成模式", "后续模式")), ("方向逻辑", self.direction_logic_var, ("正逻辑", "负逻辑")), ("加减速模式", self.accel_mode_var, ("直线", "S曲线", "正弦曲线")), ("运行模式", self.run_mode_var, ("相对", "绝对")), ) for index, (label, variable, values) in enumerate(fields): row, pair = divmod(index, 3) column = pair * 2 ttk.Label(global_box, text=label).grid(row=row, column=column, sticky="w", padx=(0, 6), pady=4) ttk.Combobox(global_box, textvariable=variable, values=values, state="readonly", width=12).grid(row=row, column=column + 1, sticky="ew", padx=(0, 14), pady=4) numeric_fields = ( ("方向延时 ms", self.direction_delay_var, 0, 65535), ("总段数", self.segment_count_var, 1, 10), ("起始段", self.start_segment_var, 1, 10), ("默认速度 Hz", self.default_speed_var, 1, 100000), ("起始速度 Hz", self.start_speed_var, 0, 100000), ("终止速度 Hz", self.end_speed_var, 0, 100000), ("加速时间 ms", self.accel_time_var, 0, 65535), ("减速时间 ms", self.decel_time_var, 0, 65535), ) for index, (label, variable, minimum, maximum) in enumerate(numeric_fields): row, pair = divmod(index, 4) row += 2 column = pair * 2 ttk.Label(global_box, text=label).grid(row=row, column=column, sticky="w", padx=(0, 6), pady=4) ttk.Spinbox(global_box, textvariable=variable, from_=minimum, to=maximum, width=12).grid(row=row, column=column + 1, sticky="ew", padx=(0, 14), pady=4) body = ttk.Frame(parent, style="Surface.TFrame") body.grid(row=1, column=0, sticky="nsew") body.columnconfigure(0, weight=3) body.columnconfigure(1, weight=2) body.rowconfigure(0, weight=1) table_box = ttk.LabelFrame(body, text="1~10 段参数（0x1100..0x1197）") table_box.grid(row=0, column=0, sticky="nsew", padx=(0, 7)) table_box.rowconfigure(0, weight=1) table_box.columnconfigure(0, weight=1) self.segment_tree = ttk.Treeview(table_box, columns=("segment", "frequency", "pulses", "wait", "wait_ms", "act_ms", "jump"), show="headings", height=10) for key, title, width in (("segment", "段", 45), ("frequency", "频率 Hz", 85), ("pulses", "脉冲数", 100), ("wait", "等待条件", 130), ("wait_ms", "WAIT ms", 75), ("act_ms", "ACT ms", 75), ("jump", "跳转", 55)): self.segment_tree.heading(key, text=title) self.segment_tree.column(key, width=width, anchor="center", stretch=key == "wait") self.segment_tree.grid(row=0, column=0, sticky="nsew") self.segment_tree.bind("<>", self._select_segment) editor = ttk.LabelFrame(body, text="当前段编辑") editor.grid(row=0, column=1, sticky="nsew", padx=(7, 0)) editor.columnconfigure(1, weight=1) segment_fields = ( ("段号", self.segment_index_var, 1, 10), ("频率 Hz", self.segment_frequency_var, 0, 100000), ("脉冲数", self.segment_pulses_var, -2147483648, 2147483647), ("WAIT时间 ms", self.segment_wait_time_var, 0, 65535), ("ACT时间 ms", self.segment_act_time_var, 0, 65535), ("跳转段（0默认）", self.segment_jump_var, 0, 10), ) for row, (label, variable, minimum, maximum) in enumerate(segment_fields): ttk.Label(editor, text=label).grid(row=row, column=0, sticky="w", padx=(0, 8), pady=5) widget = ttk.Spinbox(editor, textvariable=variable, from_=minimum, to=maximum) if row == 0: widget.configure(state="readonly") widget.grid(row=row, column=1, sticky="ew", pady=5) ttk.Label(editor, text="等待条件").grid(row=6, column=0, sticky="w", padx=(0, 8), pady=5) ttk.Combobox(editor, textvariable=self.segment_wait_var, values=("WAIT时间", "WAIT信号", "ACT时间", "EXT信号", "EXT或完成"), state="readonly").grid(row=6, column=1, sticky="ew", pady=5) self.apply_segment_button = ttk.Button(editor, text="应用当前段", command=self._apply_segment_editor) self.apply_segment_button.grid(row=7, column=0, columnspan=2, sticky="ew", pady=(10, 5)) button_line = ttk.Frame(editor, style="Surface.TFrame") button_line.grid(row=8, column=0, columnspan=2, sticky="ew") button_line.columnconfigure((0, 1), weight=1) self.read_config_button = ttk.Button(button_line, text="从设备读取全部", command=self._read_full_config) self.write_config_button = ttk.Button(button_line, text="校验并写入全部", command=self._write_full_config, style="Primary.TButton") self.read_config_button.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.write_config_button.grid(row=0, column=1, sticky="ew", padx=(4, 0)) config_actions = ttk.Frame(editor, style="Surface.TFrame") config_actions.grid(row=9, column=0, columnspan=2, sticky="ew", pady=(8, 0)) config_actions.columnconfigure((0, 1, 2), weight=1) self.config_start_button = ttk.Button( config_actions, text="写入并启动/继续", command=lambda: self._start_config_sequence(False), style="Primary.TButton") self.config_restart_button = ttk.Button( config_actions, text="写入并重新启动", command=lambda: self._start_config_sequence(True), style="Danger.TButton") self.config_stop_button = ttk.Button( config_actions, text="停止", command=lambda: self._send_command(COMMAND_STOP, "stop")) self.config_start_button.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.config_restart_button.grid(row=0, column=1, sticky="ew", padx=4) self.config_stop_button.grid(row=0, column=2, sticky="ew", padx=(4, 0)) config_actions2 = ttk.Frame(editor, style="Surface.TFrame") config_actions2.grid(row=10, column=0, columnspan=2, sticky="ew", pady=(5, 0)) config_actions2.columnconfigure((0, 1, 2), weight=1) self.config_emergency_button = ttk.Button( config_actions2, text="立即停止", command=lambda: self._send_command(COMMAND_IMMEDIATE_STOP, "immediate_stop")) self.config_reset_button = ttk.Button( config_actions2, text="复位", command=lambda: self._send_command(COMMAND_RESET, "reset")) self.config_clear_button = ttk.Button( config_actions2, text="清零累计脉冲", command=lambda: self._send_command(COMMAND_CLEAR_COUNT, "clear")) self.config_emergency_button.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.config_reset_button.grid(row=0, column=1, sticky="ew", padx=4) self.config_clear_button.grid(row=0, column=2, sticky="ew", padx=(4, 0)) ttk.Label(editor, text="X4/X5、实际频率和曲线需配合示波器/人工触发验收。", wraplength=300, foreground="#61758a", background="#ffffff").grid(row=11, column=0, columnspan=2, sticky="w", pady=(10, 0)) self._refresh_segment_tree() def _build_log_tab(self, parent: ttk.Frame) -> None: parent.rowconfigure(1, weight=1) parent.columnconfigure(0, weight=1) toolbar = ttk.Frame(parent, style="Surface.TFrame") toolbar.grid(row=0, column=0, sticky="ew", pady=(0, 8)) toolbar.columnconfigure(0, weight=1) ttk.Checkbutton(toolbar, text="记录原始 TX / RX 帧", variable=self.capture_frames_var).grid(row=0, column=0, sticky="w") ttk.Label(toolbar, text=f"最多保留 {MAX_LOG_LINES} 行", background="#ffffff", foreground="#61758a").grid(row=0, column=1, padx=12) ttk.Button(toolbar, text="导出", command=self._export_log).grid(row=0, column=2, padx=(0, 6)) ttk.Button(toolbar, text="清空", command=self._clear_log).grid(row=0, column=3) text_frame = ttk.Frame(parent, style="Surface.TFrame") text_frame.grid(row=1, column=0, sticky="nsew") text_frame.rowconfigure(0, weight=1) text_frame.columnconfigure(0, weight=1) self.log_text = tk.Text(text_frame, wrap="none", state="disabled", background="#0e1b28", foreground="#dce7f0", insertbackground="#ffffff", font=("Cascadia Mono", 9), padx=10, pady=8, relief="flat") y_scroll = ttk.Scrollbar(text_frame, orient="vertical", command=self.log_text.yview) x_scroll = ttk.Scrollbar(text_frame, orient="horizontal", command=self.log_text.xview) self.log_text.configure(yscrollcommand=y_scroll.set, xscrollcommand=x_scroll.set) self.log_text.grid(row=0, column=0, sticky="nsew") y_scroll.grid(row=0, column=1, sticky="ns") x_scroll.grid(row=1, column=0, sticky="ew") def _refresh_ports(self) -> None: if list_ports is None: messagebox.showerror("缺少依赖", "未安装 pyserial，请先运行 start_plsr_test.bat。") return current = self.port_var.get() ports = sorted((item.device for item in list_ports.comports()), key=str.upper) self.port_combo["values"] = ports if current in ports: self.port_var.set(current) elif "COM9" in ports: self.port_var.set("COM9") elif ports: self.port_var.set(ports[0]) else: self.port_var.set("") self.operation_var.set(f"检测到 {len(ports)} 个串口" if ports else "未检测到串口，请检查 USB 转串口连接") def _toggle_connection(self) -> None: if self.job_active: return if self.connected: self.auto_phase = None self._submit("disconnect", purpose="disconnect") return try: port = self.port_var.get().strip() if not port: raise ValueError("请先选择串口") baud = int(self.baud_var.get()) slave = int(self.slave_var.get()) if not 1 <= slave <= 247: raise ValueError("站号必须为 1–247") except ValueError as exc: messagebox.showerror("连接参数错误", str(exc)) return self.connection_var.set("连接中…") self.operation_var.set(f"正在连接 {port}，{baud} / 8-E-1 / 站号 {slave}") self._submit("connect", purpose="connect", port=port, baud=baud, slave=slave, protocol=self.protocol_var.get()) def _submit(self, name: str, purpose: str = "", **data: Any) -> bool: job = {"name": name, "purpose": purpose, **data} if self.job_active: if self.poll_active and purpose != "poll" and self.deferred_job is None: self.deferred_job = job self._update_controls() return True return False if name not in ("connect", "disconnect") and not self.connected: messagebox.showwarning("设备未连接", "请先连接设备。") return False self._start_job(job) return True def _start_job(self, job: dict[str, Any]) -> None: self.job_active = True self.poll_active = job["name"] == "snapshot" and job.get("purpose") == "poll" self.jobs.put(job) if not self.poll_active: self._update_controls() def _start_deferred_job(self) -> None: if self.deferred_job is None or self.job_active: return job = self.deferred_job self.deferred_job = None self._start_job(job) def _parse_outputs(self) -> tuple[int, int]: if self.protocol_var.get() == MAP_SPEC: pulse = int(self.pulse_output_var.get().removeprefix("Y")) direction = int(self.direction_output_var.get().removeprefix("Y")) - 12 if not 0 <= pulse <= 3 or not 0 <= direction <= 3: raise ValueError("需求版输出必须为 Y0..Y3 与 Y12..Y15") else: pulse = int(self.pulse_output_var.get().removeprefix("Q")) direction = int(self.direction_output_var.get().removeprefix("Q")) if pulse == direction: raise ValueError("脉冲点与方向点不能相同") return pulse, direction def _parse_motion(self) -> tuple[int, int, int]: pulses = int(self.pulses_var.get()) frequency = int(self.frequency_var.get()) direction = int(self.direction_var.get()) if self.protocol_var.get() == MAP_SPEC: if pulses == 0 or not -2147483648 <= pulses <= 2147483647: raise ValueError("需求版脉冲个数必须为 -2147483648..2147483647 且不能为 0") validate_frequency(frequency) else: if not 1 <= pulses <= 0xFFFFFFFF: raise ValueError("legacy 目标脉冲数必须为 1–4294967295") if not 1 <= frequency <= 1000: raise ValueError("legacy 频率必须为 1–1000 Hz") return pulses, frequency, direction def _segment_from_editor(self) -> SegmentParameters: wait_values = {"WAIT时间": 0, "WAIT信号": 1, "ACT时间": 2, "EXT信号": 3, "EXT或完成": 4} segment = SegmentParameters( frequency_hz=int(self.segment_frequency_var.get()), pulses=int(self.segment_pulses_var.get()), wait_condition=wait_values[self.segment_wait_var.get()], wait_time_ms=int(self.segment_wait_time_var.get()), act_time_ms=int(self.segment_act_time_var.get()), jump_segment=int(self.segment_jump_var.get()), ) segment.validate(int(self.segment_index_var.get())) return segment def _apply_segment_editor(self) -> None: try: index = int(self.segment_index_var.get()) self.segment_models[index - 1] = self._segment_from_editor() except (KeyError, ValueError) as exc: messagebox.showerror("段参数错误", str(exc)) return self._refresh_segment_tree(select=index) self.operation_var.set(f"第 {index} 段参数已应用到待写配置") def _select_segment(self, _event: object = None) -> None: selection = self.segment_tree.selection() if not selection: return index = int(self.segment_tree.item(selection[0], "values")[0]) self._load_segment_editor(index) def _load_segment_editor(self, index: int) -> None: wait_names = ("WAIT时间", "WAIT信号", "ACT时间", "EXT信号", "EXT或完成") segment = self.segment_models[index - 1] self.segment_index_var.set(str(index)) self.segment_frequency_var.set(str(segment.frequency_hz)) self.segment_pulses_var.set(str(segment.pulses)) self.segment_wait_var.set(wait_names[segment.wait_condition] if 0 <= segment.wait_condition < 5 else "WAIT时间") self.segment_wait_time_var.set(str(segment.wait_time_ms)) self.segment_act_time_var.set(str(segment.act_time_ms)) self.segment_jump_var.set(str(segment.jump_segment)) def _refresh_segment_tree(self, select: int = 1) -> None: wait_names = ("WAIT时间", "WAIT信号", "ACT时间", "EXT信号", "EXT或完成") self.segment_tree.delete(*self.segment_tree.get_children()) selected = None for index, segment in enumerate(self.segment_models, 1): item = self.segment_tree.insert("", "end", values=(index, segment.frequency_hz, segment.pulses, wait_names[segment.wait_condition] if 0 <= segment.wait_condition < 5 else segment.wait_condition, segment.wait_time_ms, segment.act_time_ms, segment.jump_segment)) if index == select: selected = item if selected: self.segment_tree.selection_set(selected) self.segment_tree.focus(selected) def _configuration_from_ui(self) -> PlsrConfiguration: self.segment_models[int(self.segment_index_var.get()) - 1] = self._segment_from_editor() cfg = PlsrConfiguration( pulse_output=self._parse_outputs()[0], direction_output=self._parse_outputs()[1], wait_input=0 if self.wait_input_var.get() == "X4" else 1, ext_input=0 if self.ext_input_var.get() == "X4" else 1, send_mode=0 if self.send_mode_var.get() == "完成模式" else 1, direction_delay_ms=int(self.direction_delay_var.get()), direction_logic=0 if self.direction_logic_var.get() == "正逻辑" else 1, accel_mode={"直线": 0, "S曲线": 1, "正弦曲线": 2}[self.accel_mode_var.get()], run_mode=0 if self.run_mode_var.get() == "相对" else 1, segment_count=int(self.segment_count_var.get()), start_segment=int(self.start_segment_var.get()), default_speed_hz=int(self.default_speed_var.get()), start_speed_hz=int(self.start_speed_var.get()), end_speed_hz=int(self.end_speed_var.get()), accel_time_ms=int(self.accel_time_var.get()), decel_time_ms=int(self.decel_time_var.get()), segments=list(self.segment_models), ) cfg.validate() return cfg def _apply_configuration(self, cfg: PlsrConfiguration) -> None: self.pulse_output_var.set(f"Y{cfg.pulse_output}") self.direction_output_var.set(f"Y{cfg.direction_output + 12}") self.wait_input_var.set(f"X{cfg.wait_input + 4}") self.ext_input_var.set(f"X{cfg.ext_input + 4}") self.send_mode_var.set(("完成模式", "后续模式")[cfg.send_mode]) self.direction_delay_var.set(str(cfg.direction_delay_ms)) self.direction_logic_var.set(("正逻辑", "负逻辑")[cfg.direction_logic]) self.accel_mode_var.set(("直线", "S曲线", "正弦曲线")[cfg.accel_mode]) self.run_mode_var.set(("相对", "绝对")[cfg.run_mode]) self.segment_count_var.set(str(cfg.segment_count)) self.start_segment_var.set(str(cfg.start_segment)) self.default_speed_var.set(str(cfg.default_speed_hz)) self.start_speed_var.set(str(cfg.start_speed_hz)) self.end_speed_var.set(str(cfg.end_speed_hz)) self.accel_time_var.set(str(cfg.accel_time_ms)) self.decel_time_var.set(str(cfg.decel_time_ms)) self.segment_models = list(cfg.segments) self._refresh_segment_tree(select=cfg.start_segment) self._load_segment_editor(cfg.start_segment) def _read_full_config(self) -> None: if self.protocol_var.get() != MAP_SPEC: messagebox.showwarning("映射不支持", "完整参数页仅适用于 spec 参数表映射。") return self.operation_var.set("正在读取系统参数和 10 段路径") self._submit("read_config", purpose="read_config") def _write_full_config(self) -> None: if self.protocol_var.get() != MAP_SPEC: messagebox.showwarning("映射不支持", "完整参数页仅适用于 spec 参数表映射。") return try: config = self._configuration_from_ui() except (KeyError, ValueError) as exc: messagebox.showerror("完整参数校验失败", str(exc)) return self._refresh_segment_tree(select=config.start_segment) self.operation_var.set(f"正在写入完整配置：{config.segment_count} 段，从第 {config.start_segment} 段执行") self._submit("write_config", purpose="write_config", config=config) def _confirm_motion(self, title: str, detail: str) -> bool: pulse = self.pulse_output_var.get() direction = self.direction_output_var.get() return messagebox.askyesno( title, "即将产生真实硬件输出。\n\n" f"脉冲点：{pulse}\n方向点：{direction}\n{detail}\n\n" "请确认机械负载已断开，当前只连接测试灯或示波器，且现场允许输出。", icon="warning", ) def _save_outputs(self) -> None: try: pulse, direction = self._parse_outputs() except ValueError as exc: messagebox.showerror("Q 点设置错误", str(exc)) return output_text = (f"脉冲 Y{pulse}，方向 Y{direction + 12}" if self.protocol_var.get() == MAP_SPEC else f"脉冲 Q{pulse}，方向 Q{direction}") self.operation_var.set(f"正在写入输出映射：{output_text}") self._submit("configure", purpose="save_outputs", pulse_output=pulse, direction_output=direction) def _start_motion(self) -> None: self._start_or_restart_motion(False) def _restart_motion(self) -> None: self._start_or_restart_motion(True) def _apply_running_frequency(self) -> None: try: frequency = int(self.frequency_var.get()) validate_frequency(frequency) except ValueError as exc: messagebox.showerror("频率参数错误", str(exc)) return self.operation_var.set(f"正在将运行频率修改为 {frequency} Hz") self._submit( "frequency", purpose="update_frequency", frequency=frequency ) def _start_or_restart_motion(self, restart: bool) -> None: try: pulses, frequency, direction = self._parse_motion() pulse, direction_output = self._parse_outputs() if self.last_snapshot is None or ( pulse != self.last_snapshot.pulse_output or direction_output != self.last_snapshot.direction_output ): raise ValueError("所选 Q 点尚未写入设备，请先点击“写入并保存”") except ValueError as exc: messagebox.showerror("运动参数错误", str(exc)) return direction_text = "反向" if direction else "正向" action = "重新运行" if restart else "启动/继续" if not self._confirm_motion(f"确认{action}脉冲", f"{direction_text}，{pulses} 脉冲，{frequency} Hz"): return self.auto_phase = None plan = (abs(pulses), frequency, direction) continuing = (not restart and self.last_snapshot is not None and self.last_snapshot.state == 6 and self.motion_plan == plan) if not continuing: self.run_baseline = self.last_snapshot.total_pulses if self.last_snapshot else 0 self.run_target = abs(pulses) self.motion_plan = plan self._reset_progress() self.operation_var.set(f"正在{action}：{direction_text} / {pulses} 脉冲 / {frequency} Hz") self._submit("start", purpose="normal_restart" if restart else "normal_start", pulses=pulses, frequency=frequency, direction=direction, restart=restart) def _start_config_sequence(self, restart: bool) -> None: if self.protocol_var.get() != MAP_SPEC: messagebox.showwarning("映射不支持", "完整参数与多段启动仅适用于 spec 参数表映射。") return try: config = self._configuration_from_ui() except (KeyError, ValueError) as exc: messagebox.showerror("完整参数校验失败", str(exc)) return action = "重新启动" if restart else "启动/继续" detail = ( f"{action}完整多段任务\n" f"起始段：{config.start_segment} / 共 {config.segment_count} 段\n" f"发送模式：{'完成模式' if config.send_mode == 0 else '后续模式'}\n" "设备将先写入当前页全部参数，再执行启动命令。" ) if not self._confirm_motion(f"确认{action}多段脉冲", detail): return self._refresh_segment_tree(select=config.start_segment) self.auto_phase = None purpose = "config_restart" if restart else "config_start" self.operation_var.set(f"正在写入并{action}完整多段任务") self._submit( "write_config_command", purpose=purpose, config=config, command=COMMAND_RESTART if restart else COMMAND_START, ) def _start_lamp_test(self) -> None: try: pulse, direction = self._parse_outputs() except ValueError as exc: messagebox.showerror("Q 点设置错误", str(exc)) return if not self._confirm_motion("确认低速正反转灯测", "正向 6 脉冲 / 2 Hz，然后反向 6 脉冲 / 2 Hz"): return self.auto_phase = "config" self.lamp_test_baseline = self.last_snapshot.total_pulses if self.last_snapshot else 0 self.run_baseline = self.lamp_test_baseline self.run_target = 12 self._reset_progress() self.operation_var.set("灯测 1/4：保存 Q 点配置") self._submit("configure", purpose="auto_config", pulse_output=pulse, direction_output=direction) def _send_command(self, command: int, purpose: str) -> None: if purpose in ("stop", "immediate_stop"): self.auto_phase = None labels = { "stop": "正在发送受控停止命令", "immediate_stop": "正在发送立即停止命令", "reset": "正在复位状态", "clear": "正在清零累计脉冲", } self.operation_var.set(labels[purpose]) self._submit("command", purpose=purpose, command=command) def _drain_events(self) -> None: try: while True: event = self.events.get_nowait() if event["type"] == "log": if self.capture_frames_var.get(): self._append_log(event["line"]) elif event["type"] == "result": self._handle_result(event) elif event["type"] == "error": self._handle_error(event) except queue.Empty: pass if not self.closing: self.after(50, self._drain_events) def _handle_result(self, event: dict[str, Any]) -> None: self.job_active = False self.poll_active = False self.communication_errors = 0 name = event["name"] purpose = event["purpose"] result = event["result"] if name == "connect": self.connected = True self.connection_var.set("已连接") self.connection_label.configure(style="StatusOn.TLabel") self.connect_button.configure(text="断开连接") self.operation_var.set(f"连接成功：{self.port_var.get()} / {self.baud_var.get()} / 8-E-1 / 站号 {self.slave_var.get()}") self._append_log(f"{time.strftime('%H:%M:%S')} INFO 已连接 {self.port_var.get()}") self._apply_snapshot(result) self._sync_output_selectors(result) elif name == "disconnect": self._set_disconnected("已断开连接") elif name == "raw": self._show_raw_registers(result) self.operation_var.set("寄存器读取完成") elif isinstance(result, PlsrConfiguration): self._apply_configuration(result) self.operation_var.set("完整参数已从设备回读" if purpose == "read_config" else "完整参数写入并回读校验完成") elif isinstance(result, Snapshot): self._apply_snapshot(result) if name == "configure": self._sync_output_selectors(result) if purpose == "save_outputs": output_text = (f"脉冲 Y{result.pulse_output}，方向 Y{result.direction_output + 12}" if result.protocol == MAP_SPEC else f"脉冲 Q{result.pulse_output}，方向 Q{result.direction_output}") self.operation_var.set(f"输出映射已写入：{output_text}") elif purpose == "auto_config": self.auto_phase = "forward" self.run_baseline = result.total_pulses self.lamp_test_baseline = result.total_pulses self.operation_var.set("灯测 2/3：正向 6 脉冲 / 2 Hz") self._submit("start", purpose="auto_forward", pulses=6, frequency=2, direction=0, restart=True) elif purpose == "auto_forward": self.operation_var.set("灯测正向运行中：观察脉冲灯亮灭") elif purpose == "auto_reverse": self.operation_var.set("灯测反向运行中：脉冲灯仍应亮灭，方向点电平应改变") elif purpose == "normal_start": self.operation_var.set("运动已启动/继续，正在轮询状态") elif purpose == "normal_restart": self.operation_var.set("运动已从本次任务起点重新运行，正在轮询状态") elif purpose == "config_start": self.operation_var.set("完整多段参数已写入，任务已启动/继续") elif purpose == "config_restart": self.operation_var.set("完整多段参数已写入，任务已重新启动") elif purpose == "update_frequency": self.operation_var.set( f"运行频率已修改为 {result.current_frequency} Hz" ) elif purpose == "clear": self.run_baseline = result.total_pulses self.run_target = 0 self._reset_progress() self.operation_var.set("累计脉冲已清零") elif purpose == "reset": self.operation_var.set("状态已复位") elif purpose == "stop": self.operation_var.set("受控停止命令已执行") elif purpose == "immediate_stop": self.operation_var.set("立即停止命令已执行") elif name == "snapshot": if purpose == "manual": self.operation_var.set("状态读取完成") self._advance_lamp_test(result) self._start_deferred_job() self._update_controls() def _handle_error(self, event: dict[str, Any]) -> None: self.job_active = False self.poll_active = False name = event["name"] error = event["error"] purpose = event["purpose"] self._append_log(f"{time.strftime('%H:%M:%S')} ERROR {error}") self.operation_var.set(f"通信/操作失败：{error}") if name == "connect": self._set_disconnected("连接失败") messagebox.showerror("连接失败", error) else: if purpose.startswith("auto_"): self.auto_phase = None messagebox.showerror("灯测中止", error) if name == "snapshot": self.communication_errors += 1 if self.communication_errors >= 3: self._set_disconnected("通信中断") messagebox.showerror("通信中断", "连续 3 次读取失败，已停止自动轮询。\n请检查串口占用、接线和设备供电后重新连接。") self._start_deferred_job() self._update_controls() def _apply_snapshot(self, snapshot: Snapshot) -> None: self.last_snapshot = snapshot state = STATE_TEXT.get(snapshot.state, f"未知（{snapshot.state}）") self.state_var.set(f"{state}（{snapshot.state}）") flag_names = [] if snapshot.flags & 0x0001: flag_names.append("运行") if snapshot.flags & 0x0002: flag_names.append("完成") if snapshot.flags & 0x0004: flag_names.append("错误") suffix = " / ".join(flag_names) if flag_names else "无" self.flags_var.set(f"0x{snapshot.flags:04X} · {suffix}") self.current_frequency_var.set(f"{snapshot.current_frequency} Hz") self.total_pulses_var.set(f"{snapshot.total_pulses:,}") self.stop_reason_var.set(STOP_REASON_TEXT.get(snapshot.stop_reason, str(snapshot.stop_reason))) self.error_var.set(ERROR_TEXT.get(snapshot.last_error, f"{snapshot.last_error} · 未知错误")) self.saved_outputs_var.set((f"脉冲 Y{snapshot.pulse_output} / 方向 Y{snapshot.direction_output + 12}") if snapshot.protocol == MAP_SPEC else f"脉冲 Q{snapshot.pulse_output} / 方向 Q{snapshot.direction_output}") self._update_progress(snapshot) def _sync_output_selectors(self, snapshot: Snapshot) -> None: if snapshot.protocol == MAP_SPEC: self.pulse_output_var.set(f"Y{snapshot.pulse_output}") self.direction_output_var.set(f"Y{snapshot.direction_output + 12}") else: self.pulse_output_var.set(f"Q{snapshot.pulse_output}") self.direction_output_var.set(f"Q{snapshot.direction_output}") def _advance_lamp_test(self, snapshot: Snapshot) -> None: if self.auto_phase == "forward" and snapshot.state in (5, 6, 7): actual = snapshot.total_pulses - self.run_baseline if snapshot.state != 5 or actual != 6: self.auto_phase = None messagebox.showerror("正向灯测失败", f"状态：{STATE_TEXT.get(snapshot.state, snapshot.state)}\n本阶段累计：{actual}，期望：6") return self.auto_phase = "reverse" self.run_baseline = snapshot.total_pulses self.operation_var.set("灯测 3/3：反向 6 脉冲 / 2 Hz") self._submit("start", purpose="auto_reverse", pulses=6, frequency=2, direction=1, restart=True) elif self.auto_phase == "reverse" and snapshot.state in (5, 6, 7): actual = abs(snapshot.total_pulses - self.run_baseline) self.auto_phase = None if snapshot.state == 5 and actual == 6 and snapshot.total_pulses == self.lamp_test_baseline: self.operation_var.set("低速正反转灯测通过：正向和反向各 6 脉冲") messagebox.showinfo("灯测完成", "软件计数验证通过：正向 6 脉冲，反向 6 脉冲。\n请同时确认反向时脉冲灯仍亮灭，方向点电平已改变。") else: messagebox.showerror("反向灯测失败", f"状态：{STATE_TEXT.get(snapshot.state, snapshot.state)}\n反向阶段脉冲：{actual}，期望：6\n灯测前累计：{self.lamp_test_baseline}，灯测后累计：{snapshot.total_pulses}") def _update_progress(self, snapshot: Snapshot) -> None: if self.run_target <= 0: return done = abs(snapshot.total_pulses - self.run_baseline) percent = min(100.0, done * 100.0 / self.run_target) self.progress.configure(value=percent) self.progress_text_var.set(f"{min(done, self.run_target):,} / {self.run_target:,}（{percent:.0f}%）") def _reset_progress(self) -> None: self.progress.configure(value=0) self.progress_text_var.set(f"0 / {self.run_target:,}" if self.run_target else "0 / 0") def _show_raw_registers(self, values: dict[str, list[int]]) -> None: self.register_tree.delete(*self.register_tree.get_children()) if "status" in values: for address, value in enumerate(values["status"]): self.register_tree.insert("", "end", values=(address, f"0x{value:04X}", value, REGISTER_NAMES.get(address, "保留/未使用"))) for offset, value in enumerate(values["control"]): address = 100 + offset self.register_tree.insert("", "end", values=(address, f"0x{value:04X}", value, REGISTER_NAMES.get(address, "保留/未使用"))) return for offset, value in enumerate(values["config"]): address = 0x1000 + offset self.register_tree.insert("", "end", values=(f"0x{address:04X}", f"0x{value:04X}", value, SPEC_REGISTER_NAMES.get(address, "配置/保留"))) for index, segment in enumerate(values["segments"], 1): base = REG_SEGMENT_BASE + (index - 1) * 0x10 for offset, value in enumerate(segment): address = base + offset self.register_tree.insert("", "end", values=(f"0x{address:04X}", f"0x{value:04X}", value, f"第{index}段 / 偏移 {offset}")) for offset, value in enumerate(values["monitor"]): address = REG_MONITOR_TOTAL_PULSES + offset self.register_tree.insert("", "end", values=(f"0x{address:04X}", f"0x{value:04X}", value, SPEC_REGISTER_NAMES.get(address, "监控/保留"))) value = values["control"][0] self.register_tree.insert("", "end", values=("0x3000", f"0x{value:04X}", value, SPEC_REGISTER_NAMES[0x3000])) def _poll_tick(self) -> None: if not self.closing and self.connected and not self.job_active: self._submit("snapshot", purpose="poll") if not self.closing: self.after(POLL_INTERVAL_MS, self._poll_tick) def _update_controls(self) -> None: ui_busy = (self.job_active and not self.poll_active) or self.deferred_job is not None connected_idle = self.connected and not ui_busy device_running = self.last_snapshot is not None and self.last_snapshot.state in (1, 2, 3, 4) can_start = connected_idle and not device_running can_stop = connected_idle and device_running self._set_widget_state(self.connect_button, "disabled" if ui_busy else "normal") serial_state = "disabled" if self.connected or ui_busy else "readonly" self._set_widget_state(self.port_combo, serial_state) self._set_widget_state(self.baud_combo, serial_state) self._set_widget_state(self.slave_spin, "disabled" if self.connected or ui_busy else "normal") self._set_widget_state(self.refresh_button, "disabled" if self.connected or ui_busy else "normal") for widget in (self.save_outputs_button, self.read_now_button, self.raw_button, self.reset_button, self.clear_button): self._set_widget_state(widget, "normal" if connected_idle and not device_running else "disabled") for widget in (self.read_config_button, self.write_config_button, self.apply_segment_button): allowed = connected_idle and not device_running and self.protocol_var.get() == MAP_SPEC self._set_widget_state(widget, "normal" if allowed else "disabled") config_start_allowed = connected_idle and not device_running and self.protocol_var.get() == MAP_SPEC config_stop_allowed = connected_idle and device_running and self.protocol_var.get() == MAP_SPEC config_idle_allowed = connected_idle and not device_running and self.protocol_var.get() == MAP_SPEC for widget in (self.config_start_button, self.config_restart_button): self._set_widget_state(widget, "normal" if config_start_allowed else "disabled") self._set_widget_state(self.config_stop_button, "normal" if config_stop_allowed else "disabled") for widget in (self.config_emergency_button,): self._set_widget_state(widget, "normal" if config_stop_allowed else "disabled") for widget in (self.config_reset_button, self.config_clear_button): self._set_widget_state(widget, "normal" if config_idle_allowed else "disabled") self._set_widget_state(self.start_button, "normal" if can_start else "disabled") self._set_widget_state(self.lamp_test_button, "normal" if can_start else "disabled") self._set_widget_state(self.stop_button, "normal" if can_stop else "disabled") self._set_widget_state(self.emergency_button, "normal" if can_start else "disabled") config_state = "readonly" if connected_idle and not device_running else "disabled" self._set_widget_state(self.pulse_output_combo, config_state) self._set_widget_state(self.direction_output_combo, config_state) motion_state = "normal" if can_start else "disabled" self._set_widget_state(self.pulses_spin, motion_state) self._set_widget_state( self.frequency_spin, "normal" if connected_idle else "disabled" ) self._set_widget_state( self.apply_frequency_button, "normal" if connected_idle and device_running else "disabled", ) self._set_widget_state(self.forward_radio, motion_state) self._set_widget_state(self.reverse_radio, motion_state) @staticmethod def _set_widget_state(widget: ttk.Widget, state: str) -> None: if str(widget.cget("state")) != state: widget.configure(state=state) def _set_disconnected(self, message: str) -> None: self.connected = False self.job_active = False self.poll_active = False self.deferred_job = None self.auto_phase = None self.last_snapshot = None self.connection_var.set(message) self.connection_label.configure(style="StatusOff.TLabel") self.connect_button.configure(text="连接设备") self.operation_var.set(message) self._update_controls() def _append_log(self, line: str) -> None: self.log_text.configure(state="normal") self.log_text.insert("end", line + "\n") self.log_line_count += 1 if self.log_line_count > MAX_LOG_LINES: remove_count = min(100, self.log_line_count - MAX_LOG_LINES) self.log_text.delete("1.0", f"{remove_count + 1}.0") self.log_line_count -= remove_count self.log_text.see("end") self.log_text.configure(state="disabled") def _clear_log(self) -> None: self.log_text.configure(state="normal") self.log_text.delete("1.0", "end") self.log_text.configure(state="disabled") self.log_line_count = 0 def _export_log(self) -> None: path = filedialog.asksaveasfilename( title="导出通信日志", defaultextension=".log", filetypes=(("日志文件", "*.log"), ("文本文件", "*.txt"), ("所有文件", "*.*")), initialfile=f"plsr-{time.strftime('%Y%m%d-%H%M%S')}.log", ) if not path: return Path(path).write_text(self.log_text.get("1.0", "end-1c"), encoding="utf-8") self.operation_var.set(f"日志已导出：{path}") def _on_close(self) -> None: if self.closing: return self.closing = True self.jobs.put({"name": "shutdown"}) self.after(150, self.destroy) def main() -> None: app = PlsrGui() app.mainloop() if __name__ == "__main__": main()