"""PLSR单轴脉冲测试上位机 - tkinter版本。""" from __future__ import annotations import json import queue import sys import time from datetime import datetime from pathlib import Path from typing import Dict, List import tkinter as tk from tkinter import messagebox, ttk from tkinter.scrolledtext import ScrolledText from modbus_rtu import ModbusClient, list_serial_ports from plsr_protocol import ( Command, DiagnosticStatus, DIAGNOSTIC_PHASE_TEXT, DirectionLogic, DIRECTION_LOGIC_TEXT, DEFAULT_BAUD_RATE, DEFAULT_SLAVE_ADDRESS, SECONDARY_POINTS, Error, MAX_FREQUENCY_HZ, MAX_RAMP_TIME_MS, MIN_FREQUENCY_HZ, MODE_TEXT, PRESETS, PROFILE_TEXT, OUTPUT_MODE_TEXT, PULSE_POINTS, Register, WaitCondition, WAIT_CONDITION_TEXT, INPUT_POINTS, ProfileType, OutputMode, RunMode, SendMode, SEND_MODE_TEXT, State, STATUS_REGISTER_COUNT, error_text, calculate_execution_total, join_i32, join_u32, parse_status, point_id_from_label, register_read_count, segment_base, signed_to_u32, split_u32, state_text, ) APP_DIR = Path(__file__).resolve().parent SETTINGS_FILE = APP_DIR / "settings.json" DIRECTION_TEXT = {0: "正向", 1: "反向"} class PlsrTesterApp: POLL_INTERVAL_MS = 250 EVENT_INTERVAL_MS = 40 def __init__(self, root: tk.Tk) -> None: self.root = root self.client = ModbusClient() self.closing = False self.expected_pulses = 0 self.expected_constant_frequency = 0 self.display_target = 0 self.test_active = False self.monitoring_active = False self.seen_running = False self.test_started_at = 0.0 self.last_state = -1 self.last_error = -1 self.pending_frequency = 0 self.pending_motion = None self.pending_segments = None self.written_segments = None self.segments_written_ready = False self.start_after_segment_write = False self.multi_test_active = False self.wait_signal_select = 0 self.ext_signal_select = 0 self._build_style() self._build_ui() self.refresh_ports() self.apply_preset(log_message=False) self._load_settings() self._set_connected(False) self.root.protocol("WM_DELETE_WINDOW", self.close) self.root.after(self.EVENT_INTERVAL_MS, self._process_client_events) self.root.after(self.POLL_INTERVAL_MS, self._poll_status) self.log("上位机已启动。请选择USART1对应的COM端口。") def _build_style(self) -> None: self.root.title("PLSR 单轴脉冲测试台 - Python") self.root.geometry("1120x900") self.root.minsize(980, 820) self.root.configure(bg="#F3F6FA") style = ttk.Style(self.root) if "clam" in style.theme_names(): style.theme_use("clam") style.configure("TFrame", background="#F3F6FA") style.configure("Card.TFrame", background="#FFFFFF") style.configure( "TLabel", background="#F3F6FA", foreground="#172033", font=("Microsoft YaHei UI", 10) ) style.configure( "Card.TLabel", background="#FFFFFF", foreground="#172033", font=("Microsoft YaHei UI", 10) ) style.configure( "TLabelframe", background="#FFFFFF", bordercolor="#D8E1EB", relief="solid" ) style.configure( "TLabelframe.Label", background="#FFFFFF", foreground="#29445F", font=("Microsoft YaHei UI", 10, "bold") ) style.configure("TButton", font=("Microsoft YaHei UI", 10), padding=(12, 7)) style.configure("Primary.TButton", background="#1769AA", foreground="#FFFFFF") style.map("Primary.TButton", background=[("active", "#0F5B98"), ("disabled", "#CBD4DE")]) style.configure("Danger.TButton", background="#D64545", foreground="#FFFFFF") style.map("Danger.TButton", background=[("active", "#BE3434"), ("disabled", "#CBD4DE")]) style.configure("Secondary.TButton", background="#E5ECF4", foreground="#29445F") style.map("Secondary.TButton", background=[("active", "#D6E1EC")]) style.configure("TNotebook", background="#F3F6FA", borderwidth=0) style.configure("TNotebook.Tab", padding=(16, 9), font=("Microsoft YaHei UI", 10)) style.map( "TNotebook.Tab", background=[("selected", "#FFFFFF"), ("!selected", "#E4EAF1")], foreground=[("selected", "#15558A"), ("!selected", "#53677C")], ) style.configure("Treeview", rowheight=54, font=("Microsoft YaHei UI", 9)) style.configure("Treeview.Heading", font=("Microsoft YaHei UI", 9, "bold"), background="#E8EEF5") style.configure("Horizontal.TProgressbar", troughcolor="#E7EDF3", background="#2B83D5") def _build_ui(self) -> None: container = ttk.Frame(self.root, padding=(22, 17, 22, 20)) container.pack(fill=tk.BOTH, expand=True) connection = ttk.LabelFrame(container, text="串口连接", padding=(14, 11)) connection.pack(fill=tk.X, pady=(0, 12)) connection.columnconfigure(8, weight=1) ttk.Label(connection, text="端口", style="Card.TLabel").grid(row=0, column=0, padx=(0, 6)) self.port_var = tk.StringVar() self.port_combo = ttk.Combobox(connection, textvariable=self.port_var, state="readonly", width=8) self.port_combo.grid(row=0, column=1, padx=(0, 6)) self.refresh_button = ttk.Button( connection, text="刷新", width=5, style="Secondary.TButton", command=self.refresh_ports ) self.refresh_button.grid(row=0, column=2, padx=(0, 10)) ttk.Label(connection, text="波特率", style="Card.TLabel").grid(row=0, column=3, padx=(0, 6)) self.baud_var = tk.StringVar(value=str(DEFAULT_BAUD_RATE)) self.baud_combo = ttk.Combobox( connection, textvariable=self.baud_var, values=("115200",), state="readonly", width=7, ) self.baud_combo.grid(row=0, column=4, padx=(0, 10)) ttk.Label(connection, text="格式 8-E-1", style="Card.TLabel").grid(row=0, column=5, padx=(0, 10)) ttk.Label(connection, text="从站", style="Card.TLabel").grid(row=0, column=6, padx=(0, 6)) self.slave_var = tk.IntVar(value=DEFAULT_SLAVE_ADDRESS) self.slave_spin = ttk.Spinbox(connection, from_=1, to=247, textvariable=self.slave_var, width=4) self.slave_spin.grid(row=0, column=7, padx=(0, 10)) connection_actions = ttk.Frame(connection, style="Card.TFrame") connection_actions.grid(row=0, column=8, sticky="e") self.connect_button = ttk.Button( connection_actions, text="连接设备", style="Primary.TButton", command=self.toggle_connection, ) self.connect_button.pack(side=tk.LEFT) self.connection_badge = tk.Label( connection_actions, text="未连接", bg="#FDECEC", fg="#A61B1B", font=("Microsoft YaHei UI", 10, "bold"), padx=10, pady=7, ) self.connection_badge.pack(side=tk.LEFT, padx=(10, 0)) notebook = ttk.Notebook(container) notebook.pack(fill=tk.BOTH, expand=True) self.control_page = ttk.Frame(notebook, padding=(0, 14, 0, 0)) self.diagnostic_page = ttk.Frame(notebook, padding=(0, 14, 0, 0)) self.log_page = ttk.Frame(notebook, padding=(0, 14, 0, 0)) notebook.add(self.control_page, text="控制、状态与多段脉冲") notebook.add(self.diagnostic_page, text="输出诊断") notebook.add(self.log_page, text="通信日志") # 单段控制、实时状态和多段参数统一放进同一个可滚动页面。 # 内容按实际需要的高度排布，避免10段表格挤压上方按钮或撑大窗口。 self.control_page.columnconfigure(0, weight=1) self.control_page.rowconfigure(0, weight=1) self.control_canvas = tk.Canvas( self.control_page, bg="#F3F6FA", highlightthickness=0, borderwidth=0, ) control_scrollbar = ttk.Scrollbar( self.control_page, orient=tk.VERTICAL, command=self.control_canvas.yview, ) self.control_canvas.configure(yscrollcommand=control_scrollbar.set) self.control_canvas.grid(row=0, column=0, sticky="nsew") control_scrollbar.grid(row=0, column=1, sticky="ns") self.control_content = ttk.Frame(self.control_canvas) self.control_window = self.control_canvas.create_window( (0, 0), window=self.control_content, anchor="nw" ) self.control_canvas.bind("", self._resize_control_content) self.control_content.bind("", self._update_control_scrollregion) # 鼠标进入整个合并页后启用滚轮；进入Canvas中的子控件时不会失效。 self.control_page.bind("", self._enable_control_mousewheel) self.control_page.bind("", self._disable_control_mousewheel) self._build_control_page() self._build_multi_page() self._build_diagnostic_page() self._build_log_page() def _resize_control_content(self, event) -> None: """让滚动页内容始终占满可见宽度。""" self.control_canvas.itemconfigure(self.control_window, width=event.width) def _update_control_scrollregion(self, _event=None) -> None: """单段或多段区域尺寸变化后刷新垂直滚动范围。""" self.control_canvas.configure(scrollregion=self.control_canvas.bbox("all")) def _enable_control_mousewheel(self, _event=None) -> None: self.root.bind_all("", self._scroll_control_page) def _disable_control_mousewheel(self, _event=None) -> None: self.root.unbind_all("") def _scroll_control_page(self, event) -> None: if event.delta != 0: self.control_canvas.yview_scroll(int(-event.delta / 120), "units") def _build_control_page(self) -> None: """建立公共参数、10段参数和实时状态组成的主操作页。""" self.control_content.columnconfigure(0, weight=3, uniform="motion") self.control_content.columnconfigure(1, weight=2, uniform="motion") # 左右区域位于同一可伸缩网格行，边框高度始终保持一致。 self.control_content.rowconfigure(1, weight=1) common = ttk.LabelFrame( self.control_content, text="公共参数", padding=(14, 10) ) common.grid( row=0, column=0, columnspan=2, sticky="nsew", pady=(0, 10) ) for column in (1, 3, 5): common.columnconfigure(column, weight=1) # 这些变量仍由既有通信、Flash同步和参数校验流程统一使用。 self.preset_var = tk.StringVar(value=PRESETS[0]["name"]) self.pulses_var = tk.StringVar(value="10") self.run_mode_var = tk.StringVar(value=MODE_TEXT[RunMode.RELATIVE]) self.target_caption_var = tk.StringVar(value="目标脉冲数") self.direction_var = tk.StringVar(value=DIRECTION_TEXT[0]) self.profile_var = tk.StringVar(value=PROFILE_TEXT[ProfileType.LINEAR]) self.direction_logic_var = tk.StringVar( value=DIRECTION_LOGIC_TEXT[DirectionLogic.POSITIVE] ) self.output_mode_var = tk.StringVar( value=OUTPUT_MODE_TEXT[OutputMode.PULSE_DIRECTION] ) self.send_mode_var = tk.StringVar(value=SEND_MODE_TEXT[SendMode.COMPLETE]) self.pulse_point_var = tk.StringVar(value=PULSE_POINTS[0]) self.secondary_point_var = tk.StringVar(value=SECONDARY_POINTS[0]) self.frequency_var = tk.IntVar(value=1) self.runtime_frequency_var = tk.IntVar(value=1) self.acceleration_var = tk.IntVar(value=1000) self.deceleration_var = tk.IntVar(value=1000) self.maximum_frequency_var = tk.IntVar(value=MAX_FREQUENCY_HZ) self.start_frequency_var = tk.IntVar(value=100) self.end_frequency_var = tk.IntVar(value=100) def add_label(row: int, column: int, text: str) -> None: ttk.Label(common, text=text, style="Card.TLabel").grid( row=row, column=column, sticky="w", padx=(0, 8), pady=5 ) add_label(0, 0, "脉冲方向逻辑") self.direction_logic_combo = ttk.Combobox( common, textvariable=self.direction_logic_var, values=tuple(DIRECTION_LOGIC_TEXT[item] for item in DirectionLogic), state="readonly", width=24, ) self.direction_logic_combo.grid(row=0, column=1, sticky="ew", padx=(0, 18), pady=5) add_label(0, 2, "脉冲类型") self.output_mode_combo = ttk.Combobox( common, textvariable=self.output_mode_var, values=tuple(OUTPUT_MODE_TEXT.values()), state="readonly", width=23, ) self.output_mode_combo.grid(row=0, column=3, sticky="ew", padx=(0, 18), pady=5) self.output_mode_combo.bind("<>", self._on_output_mode_changed) add_label(0, 4, "脉冲发送模式") self.send_mode_combo = ttk.Combobox( common, textvariable=self.send_mode_var, values=tuple(SEND_MODE_TEXT[item] for item in SendMode), state="readonly", width=24, ) self.send_mode_combo.grid(row=0, column=5, sticky="ew", pady=5) add_label(1, 0, "脉冲输出端子") self.pulse_point_combo = ttk.Combobox( common, textvariable=self.pulse_point_var, values=list(PULSE_POINTS.values()), state="readonly", width=24, ) self.pulse_point_combo.grid(row=1, column=1, sticky="ew", padx=(0, 18), pady=5) add_label(1, 2, "脉冲方向端子") self.direction_point_combo = ttk.Combobox( common, textvariable=self.secondary_point_var, values=list(SECONDARY_POINTS.values()), state="readonly", width=23, ) self.direction_point_combo.grid(row=1, column=3, sticky="ew", padx=(0, 18), pady=5) self.save_outputs_button = ttk.Button( common, text="保存公共参数", style="Primary.TButton", command=self.save_output_mapping, ) self.save_outputs_button.grid(row=1, column=4, columnspan=2, sticky="ew", pady=5) add_label(2, 0, "脉冲默认速度（Hz）") self.frequency_spin = ttk.Spinbox( common, from_=MIN_FREQUENCY_HZ, to=MAX_FREQUENCY_HZ, textvariable=self.frequency_var, ) self.frequency_spin.grid(row=2, column=1, sticky="ew", padx=(0, 18), pady=5) add_label(2, 2, "脉冲默认加速时间（ms）") self.acceleration_spin = ttk.Spinbox( common, from_=0, to=MAX_RAMP_TIME_MS, textvariable=self.acceleration_var ) self.acceleration_spin.grid(row=2, column=3, sticky="ew", padx=(0, 18), pady=5) add_label(2, 4, "脉冲默认减速时间（ms）") self.deceleration_spin = ttk.Spinbox( common, from_=0, to=MAX_RAMP_TIME_MS, textvariable=self.deceleration_var ) self.deceleration_spin.grid(row=2, column=5, sticky="ew", pady=5) add_label(3, 0, "最高速度（Hz）") self.maximum_frequency_spin = ttk.Spinbox( common, from_=MIN_FREQUENCY_HZ, to=MAX_FREQUENCY_HZ, textvariable=self.maximum_frequency_var, ) self.maximum_frequency_spin.grid(row=3, column=1, sticky="ew", padx=(0, 18), pady=5) add_label(3, 2, "起始速度（Hz）") self.start_frequency_spin = ttk.Spinbox( common, from_=0, to=MAX_FREQUENCY_HZ, textvariable=self.start_frequency_var, ) self.start_frequency_spin.grid(row=3, column=3, sticky="ew", padx=(0, 18), pady=5) add_label(3, 4, "终止速度（Hz）") self.end_frequency_spin = ttk.Spinbox( common, from_=0, to=MAX_FREQUENCY_HZ, textvariable=self.end_frequency_var, ) self.end_frequency_spin.grid(row=3, column=5, sticky="ew", pady=5) add_label(4, 0, "脉冲加减速模式") self.profile_combo = ttk.Combobox( common, textvariable=self.profile_var, values=tuple(PROFILE_TEXT[item] for item in ProfileType), state="readonly", width=24, ) self.profile_combo.grid(row=4, column=1, sticky="ew", padx=(0, 18), pady=5) add_label(5, 0, "运行中目标频率（Hz）") runtime_frequency_box = ttk.Frame(common, style="Card.TFrame") runtime_frequency_box.grid(row=5, column=1, sticky="ew", padx=(0, 18), pady=5) runtime_frequency_box.columnconfigure(0, weight=1) self.runtime_frequency_spin = ttk.Spinbox( runtime_frequency_box, from_=MIN_FREQUENCY_HZ, to=MAX_FREQUENCY_HZ, textvariable=self.runtime_frequency_var, width=11, ) self.runtime_frequency_spin.grid(row=0, column=0, sticky="ew") self.update_frequency_button = ttk.Button( runtime_frequency_box, text="立即修改当前段", style="Secondary.TButton", command=self.update_frequency, ) self.update_frequency_button.grid(row=0, column=1, padx=(6, 0)) ttk.Label( common, text="仅影响本次当前段，不修改默认速度、段表或Flash", style="Card.TLabel", foreground="#5B7083", ).grid(row=5, column=2, columnspan=4, sticky="w", pady=5) status = ttk.LabelFrame(self.control_content, text="实时状态", padding=16) status.grid(row=1, column=1, sticky="nsew", padx=(8, 0)) status.columnconfigure(1, weight=1) self.state_badge = tk.Label( status, text="空闲", bg="#E8EEF5", fg="#425466", font=("Microsoft YaHei UI", 10, "bold"), padx=13, pady=6, ) ttk.Label(status, text="运行状态", style="Card.TLabel").grid( row=0, column=0, sticky="w", pady=6 ) self.state_badge.grid(row=0, column=1, sticky="e", pady=6) self.error_var = tk.StringVar(value="0 · 无错误") self.direction_status_var = tk.StringVar(value=DIRECTION_TEXT[0]) self.current_pulses_var = tk.StringVar(value="0") self.current_frequency_var = tk.StringVar(value="0 Hz") self.actual_frequency_var = tk.StringVar(value="0 Hz") self.running_segment_var = tk.StringVar(value="未运行") self.cumulative_position_var = tk.StringVar(value="0") self.total_pulses_var = tk.StringVar(value="0") status_rows = [ ("错误码", self.error_var), ("方向", self.direction_status_var), ("当前设定频率", self.current_frequency_var), ("运行中修改后的实际频率", self.actual_frequency_var), ("当前运行段", self.running_segment_var), ("当前Y点累计位置", self.cumulative_position_var), ("总脉冲数（不计方向）", self.total_pulses_var), ] for index, (caption, variable) in enumerate(status_rows, start=1): ttk.Label(status, text=caption, style="Card.TLabel").grid( row=index, column=0, sticky="w", pady=7 ) ttk.Label( status, textvariable=variable, style="Card.TLabel", foreground="#183B56", font=("Microsoft YaHei UI", 10, "bold"), ).grid(row=index, column=1, sticky="e", pady=7) ttk.Label(status, text="本次运动进度", style="Card.TLabel").grid( row=8, column=0, columnspan=2, sticky="w", pady=(14, 6) ) self.progress_var = tk.DoubleVar(value=0.0) self.progress = ttk.Progressbar( status, maximum=100.0, variable=self.progress_var, style="Horizontal.TProgressbar", ) self.progress.grid(row=9, column=0, columnspan=2, sticky="ew") self.pulse_fraction_var = tk.StringVar(value="0 / 0") ttk.Label( status, textvariable=self.pulse_fraction_var, style="Card.TLabel", foreground="#6E7F91", ).grid(row=10, column=0, columnspan=2, sticky="e", pady=(4, 10)) status_actions = ttk.Frame(status, style="Card.TFrame") status_actions.grid(row=11, column=0, columnspan=2, sticky="ew") for column in range(3): status_actions.columnconfigure(column, weight=1) self.read_button = ttk.Button( status_actions, text="立即读取状态", style="Secondary.TButton", command=self.request_status, ) self.read_button.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.clear_position_button = ttk.Button( status_actions, text="清零当前Y点位置", style="Danger.TButton", command=self.clear_current_position, ) self.clear_position_button.grid(row=0, column=1, sticky="ew", padx=4) self.clear_total_button = ttk.Button( status_actions, text="清零总脉冲数", style="Danger.TButton", command=self.clear_total_count, ) self.clear_total_button.grid(row=0, column=2, sticky="ew", padx=(4, 0)) def _build_control_page_legacy(self) -> None: self.control_content.columnconfigure(0, weight=1, uniform="control") self.control_content.columnconfigure(1, weight=1, uniform="control") params = ttk.LabelFrame(self.control_content, text="测试参数", padding=(14, 10)) params.grid(row=0, column=0, sticky="nsew", padx=(0, 8)) params.columnconfigure(1, weight=1) mapping = tk.Frame(params, bg="#EAF4FF", padx=10, pady=7) mapping.grid(row=0, column=0, columnspan=3, sticky="ew", pady=(0, 8)) mapping.columnconfigure(1, weight=1) mapping.columnconfigure(3, weight=1) tk.Label( mapping, text="输出方式与端子映射（保存到设备Flash）", bg="#EAF4FF", fg="#15558A", anchor="w", font=("Microsoft YaHei UI", 10, "bold"), ).grid(row=0, column=0, columnspan=5, sticky="w", pady=(0, 7)) tk.Label(mapping, text="脉冲/A相", bg="#EAF4FF", fg="#29445F").grid( row=1, column=0, sticky="w", padx=(0, 5) ) self.pulse_point_var = tk.StringVar(value=PULSE_POINTS[0]) self.pulse_point_combo = ttk.Combobox( mapping, textvariable=self.pulse_point_var, values=list(PULSE_POINTS.values()), state="readonly", width=21, ) self.pulse_point_combo.grid(row=1, column=1, sticky="ew", padx=(0, 9)) tk.Label(mapping, text="方向/B相", bg="#EAF4FF", fg="#29445F").grid( row=1, column=2, sticky="w", padx=(0, 5) ) # 第二输出点在协议中仍使用0～3编码，分别对应Y12～Y15。 self.secondary_point_var = tk.StringVar(value=SECONDARY_POINTS[0]) self.direction_point_combo = ttk.Combobox( mapping, textvariable=self.secondary_point_var, values=list(SECONDARY_POINTS.values()), state="readonly", width=13, ) self.direction_point_combo.grid(row=1, column=3, sticky="ew", padx=(0, 9)) self.save_outputs_button = ttk.Button( mapping, text="应用并保存", style="Primary.TButton", command=self.save_output_mapping, ) self.save_outputs_button.grid(row=1, column=4, sticky="ew") tk.Label(mapping, text="输出方式", bg="#EAF4FF", fg="#29445F").grid( row=2, column=0, sticky="w", padx=(0, 5), pady=(7, 0) ) self.output_mode_var = tk.StringVar( value=OUTPUT_MODE_TEXT[OutputMode.PULSE_DIRECTION] ) self.output_mode_combo = ttk.Combobox( mapping, textvariable=self.output_mode_var, values=tuple(OUTPUT_MODE_TEXT.values()), state="readonly", ) self.output_mode_combo.grid( row=2, column=1, columnspan=2, sticky="ew", padx=(0, 9), pady=(7, 0) ) self.output_mode_combo.bind("<>", self._on_output_mode_changed) tk.Label(mapping, text="发送模式", bg="#EAF4FF", fg="#29445F").grid( row=2, column=3, sticky="e", padx=(0, 5), pady=(7, 0) ) self.send_mode_var = tk.StringVar(value=SEND_MODE_TEXT[SendMode.COMPLETE]) self.send_mode_combo = ttk.Combobox( mapping, textvariable=self.send_mode_var, values=tuple(SEND_MODE_TEXT[item] for item in SendMode), state="readonly", width=24, ) self.send_mode_combo.grid(row=2, column=4, sticky="ew", pady=(7, 0)) tk.Label(mapping, text="方向逻辑", bg="#EAF4FF", fg="#29445F").grid( row=3, column=0, sticky="w", padx=(0, 5), pady=(7, 0) ) self.direction_logic_var = tk.StringVar( value=DIRECTION_LOGIC_TEXT[DirectionLogic.POSITIVE] ) self.direction_logic_combo = ttk.Combobox( mapping, textvariable=self.direction_logic_var, values=tuple(DIRECTION_LOGIC_TEXT[item] for item in DirectionLogic), state="readonly", ) self.direction_logic_combo.grid( row=3, column=1, columnspan=4, sticky="ew", pady=(7, 0) ) self.output_mode_hint = tk.Label( mapping, text="", bg="#EAF4FF", fg="#476A89", anchor="w", ) # 详细说明通过选择项本身表达，不再单独占用一行，避免小屏幕裁掉运行按钮。 self._on_output_mode_changed() ttk.Label(params, text="测试预设", style="Card.TLabel").grid(row=1, column=0, sticky="w", pady=4) self.preset_var = tk.StringVar(value=PRESETS[0]["name"]) self.preset_combo = ttk.Combobox( params, textvariable=self.preset_var, values=[item["name"] for item in PRESETS], state="readonly", ) self.preset_combo.grid(row=1, column=1, sticky="ew", padx=10, pady=4) ttk.Button(params, text="应用", style="Secondary.TButton", command=self.apply_preset).grid( row=1, column=2, pady=4 ) ttk.Label(params, text="运行模式", style="Card.TLabel").grid(row=2, column=0, sticky="w", pady=4) self.run_mode_var = tk.StringVar(value=MODE_TEXT[RunMode.RELATIVE]) self.run_mode_combo = ttk.Combobox( params, textvariable=self.run_mode_var, values=(MODE_TEXT[RunMode.RELATIVE], MODE_TEXT[RunMode.ABSOLUTE]), state="readonly", ) self.run_mode_combo.grid(row=2, column=1, columnspan=2, sticky="ew", padx=(10, 0), pady=4) self.run_mode_combo.bind("<>", self._on_run_mode_changed) self.target_caption_var = tk.StringVar(value="目标脉冲数") ttk.Label(params, textvariable=self.target_caption_var, style="Card.TLabel").grid(row=3, column=0, sticky="w", pady=4) self.pulses_var = tk.StringVar(value="10") self.pulses_entry = ttk.Entry(params, textvariable=self.pulses_var) self.pulses_entry.grid(row=3, column=1, columnspan=2, sticky="ew", padx=(10, 0), pady=4) ttk.Label(params, text="默认速度/匀速（Hz）", style="Card.TLabel").grid(row=4, column=0, sticky="w", pady=4) self.frequency_var = tk.IntVar(value=1) self.frequency_spin = ttk.Spinbox( params, from_=MIN_FREQUENCY_HZ, to=MAX_FREQUENCY_HZ, textvariable=self.frequency_var, ) self.frequency_spin.grid(row=4, column=1, sticky="ew", padx=10, pady=4) self.update_frequency_button = ttk.Button( params, text="立即修改默认速度", style="Secondary.TButton", command=self.update_frequency, ) self.update_frequency_button.grid(row=4, column=2, sticky="ew", pady=4) ttk.Label(params, text="起始速度（Hz）", style="Card.TLabel").grid(row=5, column=0, sticky="w", pady=4) self.start_frequency_var = tk.IntVar(value=100) self.start_frequency_spin = ttk.Spinbox( params, from_=0, to=MAX_FREQUENCY_HZ, textvariable=self.start_frequency_var ) self.start_frequency_spin.grid(row=5, column=1, columnspan=2, sticky="ew", padx=(10, 0), pady=4) ttk.Label(params, text="终止速度（Hz）", style="Card.TLabel").grid(row=6, column=0, sticky="w", pady=4) self.end_frequency_var = tk.IntVar(value=100) self.end_frequency_spin = ttk.Spinbox( params, from_=0, to=MAX_FREQUENCY_HZ, textvariable=self.end_frequency_var ) self.end_frequency_spin.grid(row=6, column=1, columnspan=2, sticky="ew", padx=(10, 0), pady=4) ttk.Label(params, text="加减速模式", style="Card.TLabel").grid(row=7, column=0, sticky="w", pady=4) self.profile_var = tk.StringVar(value=PROFILE_TEXT[ProfileType.LINEAR]) self.profile_combo = ttk.Combobox( params, textvariable=self.profile_var, values=tuple(PROFILE_TEXT[item] for item in ProfileType), state="readonly", ) self.profile_combo.grid(row=7, column=1, columnspan=2, sticky="ew", padx=(10, 0), pady=4) ttk.Label(params, text="加速时间（ms）", style="Card.TLabel").grid(row=8, column=0, sticky="w", pady=4) self.acceleration_var = tk.IntVar(value=1000) self.acceleration_spin = ttk.Spinbox( params, from_=0, to=MAX_RAMP_TIME_MS, textvariable=self.acceleration_var ) self.acceleration_spin.grid(row=8, column=1, columnspan=2, sticky="ew", padx=(10, 0), pady=4) ttk.Label(params, text="减速时间（ms）", style="Card.TLabel").grid(row=9, column=0, sticky="w", pady=4) self.deceleration_var = tk.IntVar(value=1000) self.deceleration_spin = ttk.Spinbox( params, from_=0, to=MAX_RAMP_TIME_MS, textvariable=self.deceleration_var ) self.deceleration_spin.grid(row=9, column=1, columnspan=2, sticky="ew", padx=(10, 0), pady=4) ttk.Label(params, text="运动方向", style="Card.TLabel").grid(row=10, column=0, sticky="w", pady=4) self.direction_var = tk.StringVar(value=DIRECTION_TEXT[0]) self.direction_combo = ttk.Combobox( params, textvariable=self.direction_var, values=(DIRECTION_TEXT[0], DIRECTION_TEXT[1]), state="readonly", ) self.direction_combo.grid(row=10, column=1, columnspan=2, sticky="ew", padx=(10, 0), pady=4) button_box = ttk.Frame(params, style="Card.TFrame") button_box.grid(row=11, column=0, columnspan=3, sticky="ew", pady=(8, 0)) button_box.columnconfigure(0, weight=2) button_box.columnconfigure(1, weight=1) button_box.columnconfigure(2, weight=1) self.start_button = ttk.Button( button_box, text="写参数并启动", style="Primary.TButton", command=self.start_motion, ) self.start_button.grid(row=0, column=0, sticky="ew", padx=(0, 6)) self.stop_button = ttk.Button( button_box, text="暂停发送", style="Danger.TButton", command=self.stop_motion, ) self.stop_button.grid(row=0, column=1, sticky="ew", padx=6) self.resume_button = ttk.Button( button_box, text="继续当前脉冲", style="Secondary.TButton", command=self.resume_motion, ) self.resume_button.grid(row=0, column=2, sticky="ew", padx=(0, 0)) status = ttk.LabelFrame(self.control_content, text="实时状态", padding=16) status.grid(row=0, column=1, sticky="nsew", padx=(8, 0)) status.columnconfigure(1, weight=1) self.state_badge = tk.Label( status, text="空闲", bg="#E8EEF5", fg="#425466", font=("Microsoft YaHei UI", 10, "bold"), padx=13, pady=6, ) ttk.Label(status, text="运行状态", style="Card.TLabel").grid(row=0, column=0, sticky="w", pady=6) self.state_badge.grid(row=0, column=1, sticky="e", pady=6) self.error_var = tk.StringVar(value="0 · 无错误") self.direction_status_var = tk.StringVar(value=DIRECTION_TEXT[0]) self.current_pulses_var = tk.StringVar(value="0") self.current_frequency_var = tk.StringVar(value="0 Hz") self.actual_frequency_var = tk.StringVar(value="0 Hz") self.running_segment_var = tk.StringVar(value="未运行") self.cumulative_position_var = tk.StringVar(value="0") self.total_pulses_var = tk.StringVar(value="0") status_rows = [ ("错误码", self.error_var), ("方向", self.direction_status_var), ("已发送脉冲", self.current_pulses_var), ("当前设定频率", self.current_frequency_var), ("运行中修改后的实际频率", self.actual_frequency_var), ("当前运行段", self.running_segment_var), ("当前Y点累计位置", self.cumulative_position_var), ("总脉冲数（不计方向）", self.total_pulses_var), ] for index, (caption, variable) in enumerate(status_rows, start=1): ttk.Label(status, text=caption, style="Card.TLabel").grid(row=index, column=0, sticky="w", pady=7) ttk.Label( status, textvariable=variable, style="Card.TLabel", foreground="#183B56", font=("Microsoft YaHei UI", 10, "bold"), ).grid(row=index, column=1, sticky="e", pady=7) ttk.Label(status, text="本次运动进度", style="Card.TLabel").grid( row=9, column=0, columnspan=2, sticky="w", pady=(14, 6) ) self.progress_var = tk.DoubleVar(value=0.0) self.progress = ttk.Progressbar( status, maximum=100.0, variable=self.progress_var, style="Horizontal.TProgressbar", ) self.progress.grid(row=10, column=0, columnspan=2, sticky="ew") self.pulse_fraction_var = tk.StringVar(value="0 / 0") ttk.Label( status, textvariable=self.pulse_fraction_var, style="Card.TLabel", foreground="#6E7F91", ).grid(row=11, column=0, columnspan=2, sticky="e", pady=(4, 10)) self.test_result_label = tk.Label( status, text="等待测试", bg="#EDF1F5", fg="#526170", anchor="w", justify="left", wraplength=430, padx=11, pady=10, font=("Microsoft YaHei UI", 9), ) self.test_result_label.grid(row=12, column=0, columnspan=2, sticky="ew", pady=(3, 11)) status_actions = ttk.Frame(status, style="Card.TFrame") status_actions.grid(row=13, column=0, columnspan=2, sticky="ew") status_actions.columnconfigure(0, weight=1) status_actions.columnconfigure(1, weight=1) status_actions.columnconfigure(2, weight=1) self.read_button = ttk.Button( status_actions, text="立即读取状态", style="Secondary.TButton", command=self.request_status, ) self.read_button.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.clear_position_button = ttk.Button( status_actions, text="清零当前Y点位置", style="Danger.TButton", command=self.clear_current_position, ) self.clear_position_button.grid(row=0, column=1, sticky="ew", padx=4) self.clear_total_button = ttk.Button( status_actions, text="清零总脉冲数", style="Danger.TButton", command=self.clear_total_count, ) self.clear_total_button.grid(row=0, column=2, sticky="ew", padx=(4, 0)) def _build_multi_page(self) -> None: """建立1～10段脉冲表；脉冲数符号决定各段运动方向。""" self.multi_page = ttk.Frame(self.control_content) self.multi_page.grid(row=1, column=0, sticky="nsew", padx=(0, 8)) card = ttk.LabelFrame(self.multi_page, text="10段参数配置", padding=12) card.pack(fill=tk.BOTH, expand=True) card.columnconfigure(0, weight=1) header = ttk.Frame(card, style="Card.TFrame") header.grid(row=0, column=0, sticky="ew", pady=(0, 12)) ttk.Label(header, text="脉冲总段数", style="Card.TLabel").pack(side=tk.LEFT) self.total_segments_var = tk.IntVar(value=3) ttk.Spinbox(header, from_=1, to=10, width=7, textvariable=self.total_segments_var).pack(side=tk.LEFT, padx=(8, 24)) ttk.Label(header, text="起始执行段数", style="Card.TLabel").pack(side=tk.LEFT) self.start_segment_var = tk.IntVar(value=1) ttk.Spinbox(header, from_=1, to=10, width=7, textvariable=self.start_segment_var).pack(side=tk.LEFT, padx=(8, 24)) self.current_segment_var = tk.StringVar(value="0（未执行）") ttk.Label(header, text="当前段", style="Card.TLabel").pack(side=tk.LEFT) ttk.Label(header, textvariable=self.current_segment_var, style="Card.TLabel", foreground="#15558A").pack(side=tk.LEFT, padx=8) table = ttk.Frame(card, style="Card.TFrame") table.grid(row=1, column=0, sticky="nsew") headings = ("段", "有符号脉冲数", "频率(Hz)", "跳转段", "等待条件", "条件值(ms/X点)") for column, text in enumerate(headings): table.columnconfigure(column, weight=1 if column in (1, 2, 4, 5) else 0) ttk.Label(table, text=text, style="Card.TLabel", font=("Microsoft YaHei UI", 9, "bold")).grid( row=0, column=column, padx=4, pady=4, sticky="ew") self.segment_pulse_vars = [] self.segment_frequency_vars = [] self.segment_jump_vars = [] self.segment_wait_vars = [] self.segment_condition_vars = [] for index in range(10): pulses = tk.StringVar(value=str((index + 1) * 100)) frequency = tk.StringVar(value=str((index + 1) * 100)) self.segment_pulse_vars.append(pulses) self.segment_frequency_vars.append(frequency) jump = tk.IntVar(value=0) wait = tk.StringVar( value=WAIT_CONDITION_TEXT[WaitCondition.WAIT_TIME] ) condition = tk.StringVar(value="0") self.segment_jump_vars.append(jump) self.segment_wait_vars.append(wait) self.segment_condition_vars.append(condition) row = index + 1 ttk.Label(table, text=str(index + 1), style="Card.TLabel").grid( row=row, column=0, padx=4, pady=3) ttk.Entry(table, textvariable=pulses, width=11).grid( row=row, column=1, padx=4, pady=3, sticky="ew") ttk.Entry(table, textvariable=frequency, width=9).grid( row=row, column=2, padx=4, pady=3, sticky="ew") ttk.Spinbox(table, from_=0, to=10, textvariable=jump, width=5).grid( row=row, column=3, padx=4, pady=3, sticky="ew") ttk.Combobox(table, textvariable=wait, values=tuple(WAIT_CONDITION_TEXT.values()), state="readonly", width=13).grid(row=row, column=4, padx=4, pady=3, sticky="ew") ttk.Entry(table, textvariable=condition, width=10).grid( row=row, column=5, padx=4, pady=3, sticky="ew") segment_actions = ttk.Frame(card, style="Card.TFrame") segment_actions.grid(row=2, column=0, sticky="ew", pady=(14, 0)) segment_actions.columnconfigure(0, weight=1) segment_actions.columnconfigure(1, weight=1) segment_actions.columnconfigure(2, weight=1) segment_actions.columnconfigure(3, weight=1) self.write_segments_button = ttk.Button( segment_actions, text="写入段表", style="Secondary.TButton", command=self.write_segment_table, ) self.write_segments_button.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.start_segments_button = ttk.Button( segment_actions, text="启动", style="Primary.TButton", command=self.start_segment_motion, ) self.start_segments_button.grid(row=0, column=1, sticky="ew", padx=4) self.stop_button = ttk.Button( segment_actions, text="暂停", style="Danger.TButton", command=self.stop_motion, ) self.stop_button.grid(row=0, column=2, sticky="ew", padx=4) self.resume_button = ttk.Button( segment_actions, text="继续", style="Secondary.TButton", command=self.resume_motion, ) self.resume_button.grid(row=0, column=3, sticky="ew", padx=(4, 0)) def _build_diagnostic_page(self) -> None: """构建固件在线诊断结果页。""" card = ttk.LabelFrame( self.diagnostic_page, text="PLSR输出信号在线诊断", padding=(18, 14) ) card.pack(fill=tk.BOTH, expand=True) card.columnconfigure(1, weight=1) self.diag_overall_var = tk.StringVar(value="未运行诊断") self.diag_count_var = tk.StringVar(value="未诊断") self.diag_frequency_var = tk.StringVar(value="未诊断") self.diag_profile_var = tk.StringVar(value="未诊断") self.diag_phase_var = tk.StringVar(value="空闲") self.diag_pulses_var = tk.StringVar(value="0 / 0") self.diag_frequency_values_var = tk.StringVar(value="0 Hz / 0 Hz") self.diag_max_error_var = tk.StringVar(value="0 Hz") self.diag_error_counts_var = tk.StringVar(value="个数0，频率0，曲线0") rows = [ ("综合结果", self.diag_overall_var), ("脉冲个数", self.diag_count_var), ("脉冲频率", self.diag_frequency_var), ("加减速曲线", self.diag_profile_var), ("当前曲线阶段", self.diag_phase_var), ("实际/设定脉冲", self.diag_pulses_var), ("实际/设定频率", self.diag_frequency_values_var), ("最大频率偏差", self.diag_max_error_var), ("异常记录", self.diag_error_counts_var), ] for row, (caption, variable) in enumerate(rows): ttk.Label(card, text=caption).grid( row=row, column=0, sticky="w", padx=(0, 24), pady=7 ) ttk.Label(card, textvariable=variable).grid( row=row, column=1, sticky="w", pady=7 ) actions = ttk.Frame(card) actions.grid(row=len(rows), column=0, columnspan=2, sticky="ew", pady=(14, 8)) actions.columnconfigure(0, weight=1) actions.columnconfigure(1, weight=1) ttk.Button( actions, text="立即读取诊断", style="Secondary.TButton", command=self.request_diagnostic, ).grid(row=0, column=0, sticky="ew", padx=(0, 6)) ttk.Button( actions, text="清除诊断记录", style="Danger.TButton", command=self.clear_diagnostic, ).grid(row=0, column=1, sticky="ew", padx=(6, 0)) tk.Label( card, text=("说明：诊断依据硬件定时器实际分频参数和脉冲完成中断计数，" "可发现内部配置、频率及曲线执行异常；若要诊断端子断线或驱动器漏收，" "还需将输出接回计数输入或使用示波器。"), bg="#FFF3D8", fg="#865600", anchor="w", justify=tk.LEFT, padx=10, pady=9, wraplength=820, ).grid(row=len(rows) + 1, column=0, columnspan=2, sticky="ew", pady=(8, 0)) def _build_log_page(self) -> None: toolbar = ttk.Frame(self.log_page) toolbar.pack(fill=tk.X, pady=(0, 8)) ttk.Label( toolbar, text="显示应用事件以及实际发送、接收的Modbus RTU十六进制帧。", foreground="#6E7F91", ).pack(side=tk.LEFT) ttk.Button( toolbar, text="清空日志", style="Secondary.TButton", command=self.clear_log, ).pack(side=tk.RIGHT) self.log_text = ScrolledText( self.log_page, wrap=tk.WORD, bg="#101A26", fg="#D9E5F2", insertbackground="#FFFFFF", font=("Consolas", 10), relief=tk.FLAT, padx=10, pady=10, ) self.log_text.pack(fill=tk.BOTH, expand=True) self.log_text.configure(state=tk.DISABLED) def refresh_ports(self) -> None: previous = self.port_var.get() ports = list_serial_ports() self.port_combo["values"] = ports if previous in ports: self.port_var.set(previous) elif ports: self.port_var.set(ports[0]) else: self.port_var.set("") self.log(f"检测到串口：{', '.join(ports) if ports else '无'}") def toggle_connection(self) -> None: if self.client.is_open: self._disconnect("串口已手动断开。") return port = self.port_var.get().strip() if not port: messagebox.showwarning("未选择串口", "没有检测到可用串口，请连接设备后刷新。") return try: slave = int(self.slave_var.get()) baud = int(self.baud_var.get()) self.client.open(port, baud, slave) except Exception as exc: messagebox.showerror("连接失败", str(exc)) self.log(f"连接失败：{exc}") return self._set_connected(True) self.log(f"已连接 {port}，{baud}，8-E-1，从站{slave}。") self.client.read_holding( int(Register.PULSE_POINT), register_read_count(), "INITIAL_SYSTEM" ) def _disconnect(self, reason: str) -> None: """关闭串口并统一恢复界面状态。""" if self.client.is_open: self.client.close() self.pending_motion = None self.pending_segments = None self.written_segments = None self.segments_written_ready = False self.start_after_segment_write = False # 断开时丢弃已排队的旧事件，避免日志又出现一组过期TX/RX。 while True: try: self.client.events.get_nowait() except queue.Empty: break self.test_active = False self.monitoring_active = False self.seen_running = False self._set_connected(False) self.log(reason) def _set_connected(self, connected: bool) -> None: self.connection_badge.configure( text="已连接" if connected else "未连接", bg="#DDF7EC" if connected else "#FDECEC", fg="#087A55" if connected else "#A61B1B", ) self.connect_button.configure(text="断开连接" if connected else "连接设备") port_state = tk.DISABLED if connected else "readonly" self.port_combo.configure(state=port_state) self.baud_combo.configure(state=port_state) self.slave_spin.configure(state=tk.DISABLED if connected else tk.NORMAL) self.refresh_button.configure(state=tk.DISABLED if connected else tk.NORMAL) control_state = tk.NORMAL if connected else tk.DISABLED self.stop_button.configure(state=control_state) self.resume_button.configure(state=control_state) self.update_frequency_button.configure(state=control_state) self.save_outputs_button.configure(state=control_state) self.read_button.configure(state=control_state) self.clear_position_button.configure(state=control_state) self.clear_total_button.configure(state=control_state) self.write_segments_button.configure(state=control_state) # 未手动写入段表时也允许点击启动；启动流程会先 # 自动写入当前界面的默认参数，设备确认后再启动。 self.start_segments_button.configure( state=(tk.NORMAL if connected else tk.DISABLED) ) def apply_preset(self, log_message: bool = True) -> None: selected = self.preset_var.get() preset = next((item for item in PRESETS if item["name"] == selected), PRESETS[0]) self.pulses_var.set(str(preset["pulses"])) self.frequency_var.set(preset["frequency"]) self.runtime_frequency_var.set(preset["frequency"]) self.acceleration_var.set(preset["acceleration_ms"]) self.deceleration_var.set(preset["deceleration_ms"]) self.start_frequency_var.set(preset["start_frequency"]) self.end_frequency_var.set(preset["end_frequency"]) self.display_target = preset["pulses"] if log_message: self.log(f"已应用预设：{preset['name']}") def _selected_output_mode(self) -> OutputMode: return next( (mode for mode, text in OUTPUT_MODE_TEXT.items() if text == self.output_mode_var.get()), OutputMode.PULSE_DIRECTION, ) def _selected_send_mode(self) -> SendMode: return next( (mode for mode, text in SEND_MODE_TEXT.items() if text == self.send_mode_var.get()), SendMode.COMPLETE, ) def _selected_direction_logic(self) -> DirectionLogic: return next( (logic for logic, text in DIRECTION_LOGIC_TEXT.items() if text == self.direction_logic_var.get()), DirectionLogic.POSITIVE, ) def _direction_display_text(self, direction: int) -> str: if self._selected_output_mode() == OutputMode.AB_QUADRATURE: return ( "正向（A相超前B相90°）" if direction == 0 else "反向（B相超前A相90°）" ) return DIRECTION_TEXT[1 if direction else 0] def _on_output_mode_changed(self, _event=None) -> None: if hasattr(self, "direction_status_var"): direction = 1 if self.direction_var.get() == DIRECTION_TEXT[1] else 0 self.direction_status_var.set(self._direction_display_text(direction)) def _on_run_mode_changed(self, _event=None) -> None: """根据模式切换目标含义；绝对模式的实际方向由固件自动计算。""" absolute = self.run_mode_var.get() == MODE_TEXT[RunMode.ABSOLUTE] if hasattr(self, "target_caption_var"): self.target_caption_var.set("目标绝对位置" if absolute else "目标脉冲数") def _read_motion_inputs(self) -> Dict[str, int]: try: pulses = int(self.pulses_var.get().strip()) frequency = int(self.frequency_var.get()) acceleration_ms = int(self.acceleration_var.get()) deceleration_ms = int(self.deceleration_var.get()) maximum_frequency = int(self.maximum_frequency_var.get()) start_frequency = int(self.start_frequency_var.get()) end_frequency = int(self.end_frequency_var.get()) except ValueError as exc: raise ValueError("脉冲数、三项速度和加减速时间必须是整数") from exc run_mode = ( RunMode.ABSOLUTE if self.run_mode_var.get() == MODE_TEXT[RunMode.ABSOLUTE] else RunMode.RELATIVE ) direction = 1 if self.direction_var.get() == DIRECTION_TEXT[1] else 0 if run_mode == RunMode.RELATIVE: if not -0x80000000 <= pulses <= 0x7FFFFFFF: raise ValueError("相对脉冲数必须在-2147483648到2147483647之间") # 负数按需求直接表示反向；正数仍允许通过“运动方向”选择反向， # 兼容原有上位机操作。INT32_MIN只能表示为反向2147483648脉冲。 if pulses < 0: direction = 1 magnitude = abs(pulses) command_pulses = -magnitude if direction else magnitude target_u32 = signed_to_u32(command_pulses) else: target_u32 = signed_to_u32(pulses) if not MIN_FREQUENCY_HZ <= maximum_frequency <= MAX_FREQUENCY_HZ: raise ValueError("最高速度必须在1到100000Hz之间") if not MIN_FREQUENCY_HZ <= frequency <= maximum_frequency: raise ValueError(f"默认速度必须在1到最高速度{maximum_frequency}Hz之间") if not 0 <= acceleration_ms <= MAX_RAMP_TIME_MS: raise ValueError(f"加速时间必须在0到{MAX_RAMP_TIME_MS}ms之间") if not 0 <= deceleration_ms <= MAX_RAMP_TIME_MS: raise ValueError(f"减速时间必须在0到{MAX_RAMP_TIME_MS}ms之间") if not 0 <= start_frequency <= frequency: raise ValueError("起始速度必须在0Hz到默认速度之间") if not 0 <= end_frequency <= frequency: raise ValueError("终止速度必须在0Hz到默认速度之间") profile_type = next( (item for item, text in PROFILE_TEXT.items() if text == self.profile_var.get()), ProfileType.LINEAR, ) return { "pulses": pulses, "target_u32": target_u32, "frequency": frequency, "direction": direction, "run_mode": int(run_mode), "profile_type": int(profile_type), "acceleration_ms": acceleration_ms, "deceleration_ms": deceleration_ms, "start_frequency": start_frequency, "end_frequency": end_frequency, "maximum_frequency": maximum_frequency, } def start_motion(self) -> None: if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return try: params = self._read_motion_inputs() except ValueError as exc: messagebox.showwarning("参数错误", str(exc)) return if params["run_mode"] == int(RunMode.ABSOLUTE): current_position = int(self.cumulative_position_var.get()) self.expected_pulses = abs(params["pulses"] - current_position) else: self.expected_pulses = abs(params["pulses"]) self.display_target = self.expected_pulses self.pending_motion = params self.expected_constant_frequency = ( params["frequency"] if params["start_frequency"] == params["frequency"] and params["end_frequency"] == params["frequency"] and params["acceleration_ms"] == 0 and params["deceleration_ms"] == 0 else 0 ) pulse_point = point_id_from_label(PULSE_POINTS, self.pulse_point_var.get()) direction_point = point_id_from_label(SECONDARY_POINTS, self.secondary_point_var.get()) profile_values = [pulse_point, direction_point, 0, 0, int(self._selected_send_mode()), 20, int(self._selected_direction_logic()), params["profile_type"], params["run_mode"], 1, 1] profile_values += split_u32(params["frequency"]) profile_values += split_u32(params["start_frequency"]) profile_values += [int(self._selected_output_mode())] self.client.write_multiple( int(Register.PULSE_POINT), profile_values, "SET_SYSTEM_A" ) self.test_active = True self.multi_test_active = False # 参数写入链完成并收到START响应后再开始周期查询，避免查询帧插队。 self.monitoring_active = False self.seen_running = False self.test_started_at = time.monotonic() if params["run_mode"] == int(RunMode.ABSOLUTE): actual_direction = 1 if params["pulses"] < current_position else 0 else: actual_direction = params["direction"] self.direction_status_var.set(self._direction_display_text(actual_direction)) self._set_test_result("启动命令发送中，等待设备确认。", "running") self.log(f"提交单段启动：目标{params['pulses']}，频率{params['frequency']}Hz。") def write_segment_table(self, start_after_write: bool = False) -> None: """校验并写入公共曲线参数、段数、起始段和段表。 start_after_write为True时，用于“未写段表直接启动”的自动 流程：设备完成段表保存后再发送启动命令。 """ self.start_after_segment_write = False if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return try: total_segments = int(self.total_segments_var.get()) start_segment = int(self.start_segment_var.get()) start_frequency = int(self.start_frequency_var.get()) end_frequency = int(self.end_frequency_var.get()) acceleration_ms = int(self.acceleration_var.get()) deceleration_ms = int(self.deceleration_var.get()) default_frequency = int(self.frequency_var.get()) maximum_frequency = int(self.maximum_frequency_var.get()) if not 1 <= total_segments <= 10: raise ValueError("脉冲总段数必须在1到10之间") if not 1 <= start_segment <= total_segments: raise ValueError("起始执行段数必须在1到脉冲总段数之间") if not 0 <= acceleration_ms <= MAX_RAMP_TIME_MS: raise ValueError(f"加速时间必须在0到{MAX_RAMP_TIME_MS}ms之间") if not 0 <= deceleration_ms <= MAX_RAMP_TIME_MS: raise ValueError(f"减速时间必须在0到{MAX_RAMP_TIME_MS}ms之间") if not MIN_FREQUENCY_HZ <= maximum_frequency <= MAX_FREQUENCY_HZ: raise ValueError("最高速度必须在1到100000Hz之间") if not MIN_FREQUENCY_HZ <= default_frequency <= maximum_frequency: raise ValueError(f"默认速度必须在1到最高速度{maximum_frequency}Hz之间") if not 0 <= start_frequency <= maximum_frequency: raise ValueError(f"起始速度必须在0到最高速度{maximum_frequency}Hz之间") if not 0 <= end_frequency <= maximum_frequency: raise ValueError(f"终止速度必须在0到最高速度{maximum_frequency}Hz之间") segments = [] for index in range(total_segments): pulses = int(self.segment_pulse_vars[index].get().strip()) frequency = int(self.segment_frequency_vars[index].get().strip()) jump = int(self.segment_jump_vars[index].get()) condition_value = int(self.segment_condition_vars[index].get().strip()) wait_condition = next( (item for item, text in WAIT_CONDITION_TEXT.items() if text == self.segment_wait_vars[index].get()), WaitCondition.WAIT_TIME) if not -0x80000000 <= pulses <= 0x7FFFFFFF: raise ValueError(f"第{index + 1}段脉冲数必须在-2147483648到2147483647之间") if not 0 <= frequency <= maximum_frequency: raise ValueError( f"第{index + 1}段频率必须为0或1到最高速度{maximum_frequency}Hz" ) effective_frequency = default_frequency if frequency == 0 else frequency if start_frequency > effective_frequency or end_frequency > effective_frequency: raise ValueError(f"第{index + 1}段实际目标速度不能低于起始或终止速度") if not 0 <= jump <= total_segments: raise ValueError(f"第{index + 1}段跳转段必须是0到{total_segments}") if not 0 <= condition_value <= 0xFFFFFFFF: raise ValueError(f"第{index + 1}段条件值超出32位无符号范围") if wait_condition in (WaitCondition.WAIT_SIGNAL, WaitCondition.EXT_SIGNAL, WaitCondition.EXT_OR_COMPLETE) and condition_value not in INPUT_POINTS: raise ValueError(f"第{index + 1}段信号X点无效") segments.append((pulses, frequency, jump, int(wait_condition), condition_value)) except (ValueError, tk.TclError) as exc: messagebox.showwarning("多段参数错误", str(exc)) return execution_total, execution_has_cycle = calculate_execution_total( segments, start_segment ) profile_type = next( (item for item, text in PROFILE_TEXT.items() if text == self.profile_var.get()), ProfileType.LINEAR, ) direction = 1 if segments[start_segment - 1][0] < 0 else 0 self.pending_segments = { "total": total_segments, "start": start_segment, "segments": segments, "direction": direction, "profile_type": int(profile_type), "acceleration_ms": acceleration_ms, "deceleration_ms": deceleration_ms, "start_frequency": start_frequency, "end_frequency": end_frequency, "default_frequency": default_frequency, "maximum_frequency": maximum_frequency, "execution_total": execution_total, "execution_has_cycle": execution_has_cycle, } wait_points = { item[4] for item in segments if item[3] == int(WaitCondition.WAIT_SIGNAL) } ext_points = { item[4] for item in segments if item[3] in (int(WaitCondition.EXT_SIGNAL), int(WaitCondition.EXT_OR_COMPLETE)) } if any(point not in INPUT_POINTS for point in wait_points | ext_points): messagebox.showwarning("信号点无效", "参数表仅允许WAIT/EXT选择X4或X5。") self.pending_segments = None return if len(wait_points) > 1 or len(ext_points) > 1: messagebox.showwarning( "全局信号配置冲突", "参数表中WAIT信号和EXT信号分别是全局配置，同类条件必须统一使用X4或X5。", ) self.pending_segments = None return wait_select = next(iter(wait_points), 4) - 4 ext_select = next(iter(ext_points), 4) - 4 self.wait_signal_select = wait_select self.ext_signal_select = ext_select pulse_point = point_id_from_label(PULSE_POINTS, self.pulse_point_var.get()) direction_point = point_id_from_label(SECONDARY_POINTS, self.secondary_point_var.get()) profile_values = [pulse_point, direction_point, wait_select, ext_select, int(self._selected_send_mode()), 20, int(self._selected_direction_logic()), int(profile_type), int(RunMode.RELATIVE), total_segments, start_segment] # 0x100B保存公共默认速度；各段自己的运行速度仍来自段表。 profile_values += split_u32(default_frequency) profile_values += split_u32(start_frequency) profile_values += [int(self._selected_output_mode())] self.start_after_segment_write = bool(start_after_write) self.client.write_multiple(int(Register.PULSE_POINT), profile_values, "SET_SEGMENT_SYSTEM_A") self.written_segments = None self.segments_written_ready = False self.start_segments_button.configure(state=tk.DISABLED) self.test_active = False self.multi_test_active = False self.monitoring_active = False self._set_test_result("正在写入多段参数和段表，不会自动启动。", "running") if execution_has_cycle: self.log(f"提交段表写入：共{total_segments}段，从第{start_segment}段开始，" "检测到循环跳转。") else: self.log(f"提交段表写入：共{total_segments}段，从第{start_segment}段开始，" f"预计输出{execution_total}个脉冲。") def start_segment_motion(self) -> None: """启动多段运动；未写段表时先自动写入当前默认参数。""" if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return if self.last_state in ( int(State.RUNNING), int(State.WAITING), int(State.PAUSED) ): messagebox.showwarning("当前不可启动", "设备正在运行、等待或暂停，请先结束当前运动。") return if not self.segments_written_ready or self.written_segments is None: self.log("段表尚未写入，正在按当前默认参数自动写入后启动。") self.write_segment_table(start_after_write=True) return params = self.written_segments execution_has_cycle = bool(params["execution_has_cycle"]) execution_total = int(params["execution_total"]) self.expected_pulses = 0 if execution_has_cycle else execution_total self.display_target = 0 if execution_has_cycle else execution_total self.test_active = True self.multi_test_active = True self.expected_constant_frequency = 0 self.monitoring_active = False self.seen_running = False self.test_started_at = time.monotonic() self.direction_status_var.set( self._direction_display_text(int(params["direction"])) ) self._set_test_result("多段启动命令发送中，等待设备确认。", "running") self.client.write_single( int(Register.CONTROL), int(Command.START_SEGMENTS), "START_SEGMENTS" ) self.log( f"提交多段启动：共{params['total']}段，从第{params['start']}段开始。" ) def stop_motion(self) -> None: if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return self.client.write_single(int(Register.CONTROL), int(Command.STOP), "STOP") self.test_active = False self.monitoring_active = False self._set_test_result("暂停命令发送中。", "running") self.log("提交暂停命令，将保留当前脉冲位置。") def resume_motion(self) -> None: """继续STM32中已暂停的运动，不重写参数也不清零本次计数。""" if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return self.client.write_single( int(Register.CONTROL), int(Command.RESUME), "RESUME" ) self.expected_pulses = self.display_target self.test_active = True self.monitoring_active = True self.seen_running = False self.test_started_at = time.monotonic() self._set_test_result("继续命令发送中，等待设备确认。", "running") self.log("提交继续命令，将从暂停的脉冲位置继续。") def update_frequency(self) -> None: """写入独立的瞬时目标频率，不改变默认速度、段表或当前脉冲计数。""" if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return if self.last_state not in (int(State.RUNNING), int(State.PAUSED)): messagebox.showwarning("当前不可改频", "只有运行中或暂停状态可以修改频率。") return try: frequency = int(self.runtime_frequency_var.get()) maximum_frequency = int(self.maximum_frequency_var.get()) start_frequency = int(self.start_frequency_var.get()) end_frequency = int(self.end_frequency_var.get()) except (ValueError, tk.TclError): messagebox.showwarning("参数错误", "频率必须是整数。") return if not MIN_FREQUENCY_HZ <= maximum_frequency <= MAX_FREQUENCY_HZ: messagebox.showwarning("参数错误", "最高速度必须在1到100000Hz之间。") return if not MIN_FREQUENCY_HZ <= frequency <= maximum_frequency: messagebox.showwarning( "参数错误", f"频率必须在1到最高速度{maximum_frequency}Hz之间。" ) return if frequency < start_frequency or frequency < end_frequency: messagebox.showwarning( "参数错误", "运行中目标频率不能低于起始速度或终止速度。" ) return self.pending_frequency = frequency self.client.write_multiple( int(Register.RUNTIME_FREQ_LO), split_u32(frequency), "SET_RUNTIME_FREQ_VALUE", ) self._set_test_result(f"正在将频率修改为{frequency}Hz。", "running") self.log(f"提交动态改频：{frequency}Hz，不清零当前脉冲计数。") def save_output_mapping(self) -> None: """一次写入12项公共参数，并请求设备保存到Flash。""" if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return if self.last_state in ( int(State.RUNNING), int(State.PAUSED), int(State.WAITING) ): messagebox.showwarning( "当前不可配置", "请等待脉冲完成；暂停中的任务需要先完成或重新启动后再保存点位。", ) return try: pulse_point = point_id_from_label(PULSE_POINTS, self.pulse_point_var.get()) direction_point = point_id_from_label( SECONDARY_POINTS, self.secondary_point_var.get() ) default_frequency = int(self.frequency_var.get()) maximum_frequency = int(self.maximum_frequency_var.get()) start_frequency = int(self.start_frequency_var.get()) end_frequency = int(self.end_frequency_var.get()) acceleration_ms = int(self.acceleration_var.get()) deceleration_ms = int(self.deceleration_var.get()) total_segments = int(self.total_segments_var.get()) start_segment = int(self.start_segment_var.get()) if not MIN_FREQUENCY_HZ <= maximum_frequency <= MAX_FREQUENCY_HZ: raise ValueError("最高速度必须在1到100000Hz之间") if not MIN_FREQUENCY_HZ <= default_frequency <= maximum_frequency: raise ValueError("默认速度必须在1Hz到最高速度之间") if not 0 <= start_frequency <= maximum_frequency: raise ValueError("起始速度必须在0Hz到最高速度之间") if not 0 <= end_frequency <= maximum_frequency: raise ValueError("终止速度必须在0Hz到最高速度之间") if not 0 <= acceleration_ms <= MAX_RAMP_TIME_MS: raise ValueError(f"默认加速时间必须在0到{MAX_RAMP_TIME_MS}ms之间") if not 0 <= deceleration_ms <= MAX_RAMP_TIME_MS: raise ValueError(f"默认减速时间必须在0到{MAX_RAMP_TIME_MS}ms之间") if not 1 <= total_segments <= 10: raise ValueError("脉冲总段数必须在1到10之间") if not 1 <= start_segment <= total_segments: raise ValueError("起始执行段数必须在有效段范围内") for index in range(total_segments): segment_frequency = int(self.segment_frequency_vars[index].get()) if not 0 <= segment_frequency <= maximum_frequency: raise ValueError(f"第{index + 1}段频率必须为0或不超过公共最高速度") effective_frequency = ( default_frequency if segment_frequency == 0 else segment_frequency ) if start_frequency > effective_frequency or end_frequency > effective_frequency: raise ValueError(f"第{index + 1}段实际目标速度不能低于公共起始或终止速度") except (ValueError, tk.TclError) as exc: messagebox.showwarning("公共参数错误", str(exc)) return profile_type = next( (item for item, text in PROFILE_TEXT.items() if text == self.profile_var.get()), ProfileType.LINEAR, ) values = [ pulse_point, direction_point, int(getattr(self, "wait_signal_select", 0)), int(getattr(self, "ext_signal_select", 0)), int(self._selected_send_mode()), 20, int(self._selected_direction_logic()), int(profile_type), int(RunMode.RELATIVE), total_segments, start_segment, ] values += split_u32(default_frequency) values += split_u32(start_frequency) values += [int(self._selected_output_mode())] values += split_u32(end_frequency) values += [acceleration_ms, deceleration_ms] values += split_u32(maximum_frequency) self.client.write_multiple( int(Register.PULSE_POINT), values, "SET_PUBLIC_PARAMETERS", ) self._set_test_result( "正在应用并保存全部公共参数。", "running", ) self.log( f"提交公共参数：{PROFILE_TEXT[profile_type]}，" f"默认{default_frequency}Hz，最高{maximum_frequency}Hz，" f"起始{start_frequency}Hz，终止{end_frequency}Hz，" f"加速{acceleration_ms}ms，减速{deceleration_ms}ms，请求掉电保存。" ) def clear_current_position(self) -> None: """清零当前所选脉冲Y点的累计位置。""" if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return if not messagebox.askyesno( "确认清零", "确定将当前脉冲Y点的累计位置清零吗？其他脉冲Y点不受影响。" ): return self.client.write_single( int(Register.CONTROL), int(Command.CLEAR_POSITION), "CLEAR_POSITION" ) self.log("提交当前脉冲Y点累计位置清零命令。") def clear_total_count(self) -> None: """清零不计方向和输出点的全局总脉冲数。""" if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return if not messagebox.askyesno( "确认清零", "确定将掉电保持的总脉冲数清零吗？Y0～Y3累计位置不受影响。" ): return self.client.write_single( int(Register.CONTROL), int(Command.CLEAR_TOTAL), "CLEAR_TOTAL" ) self.log("提交总脉冲数清零命令，等待设备确认。") def request_diagnostic(self) -> None: if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return if not self.client.is_busy: self.client.read_holding( int(Register.CUMULATIVE_POSITION_LO), STATUS_REGISTER_COUNT, "DIAGNOSTIC_STATUS", ) def clear_diagnostic(self) -> None: if not self.client.is_open: messagebox.showwarning("尚未连接", "请先连接串口设备。") return self.client.write_single( int(Register.CONTROL), int(Command.CLEAR_DIAGNOSTIC), "CLEAR_DIAGNOSTIC", ) self.log("已提交诊断记录清零命令。") def request_status(self) -> None: if self.client.is_open and not self.client.is_busy: # 状态区保持轮询；AB实际方向在启动时或多段切换时同步到界面。 self.client.read_holding( int(Register.CUMULATIVE_POSITION_LO), STATUS_REGISTER_COUNT, "POLL", ) def _request_public_save_status(self) -> None: """稍作延时后读取错误码，确认Flash任务已完成。""" if self.closing: return if not self.client.is_open: message = "公共参数已写入，但连接已断开，无法确认Flash保存结果。" self._set_test_result(message, "fail") self.log(message) messagebox.showerror("无法确认保存结果", message) return if self.client.is_busy: self.root.after(50, self._request_public_save_status) return self.client.read_holding( int(Register.CUMULATIVE_POSITION_LO), STATUS_REGISTER_COUNT, "PUBLIC_SAVE_STATUS", ) def _request_segment_save_status(self) -> None: """等待设备完成段表Flash保存，再判定写入成功。""" if self.closing: return if not self.client.is_open: self._abort_segment_table_write( "段表已写入寄存器，但连接已断开，无法确认Flash保存结果。" ) return if self.client.is_busy: self.root.after(50, self._request_segment_save_status) return self.client.read_holding( int(Register.CUMULATIVE_POSITION_LO), STATUS_REGISTER_COUNT, "SEGMENT_SAVE_STATUS", ) def _poll_status(self) -> None: if self.closing: return if self.monitoring_active: self.request_status() self.root.after(self.POLL_INTERVAL_MS, self._poll_status) def _process_client_events(self) -> None: if self.closing: return while True: try: event = self.client.events.get_nowait() except queue.Empty: break kind = event[0] if kind == "tx": self.log(f"TX {event[1].hex(' ').upper()}") elif kind == "rx": self.log(f"RX {event[1].hex(' ').upper()}") elif kind == "read": try: self._handle_read(event[1], event[2], event[3]) except (IndexError, ValueError) as exc: # 状态回读异常不能终止Tkinter事件循环，否则后续启动写入 # 的确认事件也不会再被处理，界面会一直停在“等待确认”。 self.log(f"寄存器数据解析失败：{exc}") self._set_test_result(f"设备回读数据不完整：{exc}", "fail") self.test_active = False self.monitoring_active = False elif kind == "write": self._handle_write(event[1], event[2], event[3], event[4]) elif kind == "retry": _, context, attempt = event self.log(f"{context}未收到响应，正在进行第{attempt}次发送。") elif kind == "exception": _, function, code, context = event message = f"Modbus异常：功能码0x{function:02X}，异常码0x{code:02X}，操作{context}。" self.log(message) self._set_test_result(message, "fail") if ((isinstance(context, str) and context.startswith("SET_SEGMENT_")) or context in ("SAVE_SEGMENT_CONFIGURATION", "SEGMENT_SAVE_STATUS")): self._abort_segment_table_write(message) if context in ( "SET_PUBLIC_PARAMETERS", "SAVE_PUBLIC", "PUBLIC_CONFIG_STATUS", "PUBLIC_SAVE_STATUS", ): messagebox.showerror("公共参数保存失败", message) self.test_active = False self.monitoring_active = False elif kind == "error": _, context, message = event text = f"{context}：{message}" if context else message self.log(f"通信错误：{text}") if ((isinstance(context, str) and context.startswith("SET_SEGMENT_")) or context in ("SAVE_SEGMENT_CONFIGURATION", "SEGMENT_SAVE_STATUS")): self._abort_segment_table_write(text) if context in ( "SET_PUBLIC_PARAMETERS", "SAVE_PUBLIC", "PUBLIC_CONFIG_STATUS", "PUBLIC_SAVE_STATUS", ): messagebox.showerror("公共参数保存失败", text) if self.test_active: self._set_test_result(f"测试中断：{text}", "fail") self.test_active = False self.monitoring_active = False self.root.after(self.EVENT_INTERVAL_MS, self._process_client_events) def _handle_read(self, address: int, values: List[int], context: str) -> None: if address == int(Register.CUMULATIVE_POSITION_LO) and len(values) >= 11: status = parse_status(values) self._update_status(status) if context == "INITIAL_STATUS": self._set_test_result("设备参数和状态读取完成。", "neutral") self.log("设备初始化同步完成，可以写入参数并启动。") elif context == "PUBLIC_SAVE_STATUS": error_code = status["error"] if error_code == int(Error.NONE): message = "12项公共参数已成功保存到设备Flash。" self._set_test_result(message, "pass") self.log(message) messagebox.showinfo("保存成功", message) else: message = ( f"公共参数保存失败：错误码{error_code}，" f"{error_text(error_code)}。" ) self._set_test_result(message, "fail") self.log(message) messagebox.showerror("保存失败", message) elif context == "SEGMENT_SAVE_STATUS": error_code = status["error"] if error_code == int(Error.NONE): self._finish_segment_table_write() else: self._abort_segment_table_write( f"段表保存失败：错误码{error_code}，" f"{error_text(error_code)}。" ) return if address == int(Register.PULSE_POINT) and len(values) >= 22: pulse_point = values[0] direction_point = values[1] self.wait_signal_select = values[2] if values[2] in (0, 1) else 0 self.ext_signal_select = values[3] if values[3] in (0, 1) else 0 if pulse_point in PULSE_POINTS: self.pulse_point_var.set(PULSE_POINTS[pulse_point]) if direction_point in SECONDARY_POINTS: self.secondary_point_var.set(SECONDARY_POINTS[direction_point]) if values[4] in [int(item) for item in SendMode]: self.send_mode_var.set(SEND_MODE_TEXT[SendMode(values[4])]) if values[6] in [int(item) for item in DirectionLogic]: self.direction_logic_var.set( DIRECTION_LOGIC_TEXT[DirectionLogic(values[6])] ) self.profile_var.set(PROFILE_TEXT.get(ProfileType(values[7]), PROFILE_TEXT[ProfileType.LINEAR])) self.run_mode_var.set(MODE_TEXT[RunMode.ABSOLUTE if values[8] else RunMode.RELATIVE]) self.total_segments_var.set(values[9]) self.start_segment_var.set(values[10]) self.frequency_var.set(join_u32(values[11], values[12])) self.start_frequency_var.set(join_u32(values[13], values[14])) output_mode = values[15] if output_mode in [int(item) for item in OutputMode]: self.output_mode_var.set(OUTPUT_MODE_TEXT[OutputMode(output_mode)]) self._on_output_mode_changed() self.end_frequency_var.set(join_u32(values[16], values[17])) self.acceleration_var.set(values[18]) self.deceleration_var.set(values[19]) self.maximum_frequency_var.set(join_u32(values[20], values[21])) if context == "INITIAL_SYSTEM": self.client.read_holding( int(Register.CUMULATIVE_POSITION_LO), STATUS_REGISTER_COUNT, "INITIAL_STATUS", ) elif context == "PUBLIC_CONFIG_STATUS": self.log("公共参数回读完成，正在确认Flash保存结果。") self.root.after(200, self._request_public_save_status) return if address == int(Register.END_FREQ_LO) and len(values) >= 4: self.end_frequency_var.set(join_u32(values[0], values[1])) self.acceleration_var.set(values[2]) self.deceleration_var.set(values[3]) if context == "INITIAL_RAMP": self.client.read_holding( int(Register.CUMULATIVE_POSITION_LO), STATUS_REGISTER_COUNT, "INITIAL_STATUS" ) return if False and address == int(Register.COMMAND) and len(values) >= 28: run_mode = RunMode.ABSOLUTE if values[18] == 1 else RunMode.RELATIVE target_raw = join_u32(values[2], values[3]) target = join_i32(values[2], values[3]) if run_mode == RunMode.ABSOLUTE else target_raw frequency = join_u32(values[4], values[5]) self.pulses_var.set(str(target)) self.run_mode_var.set(MODE_TEXT[run_mode]) self._on_run_mode_changed() self.frequency_var.set(max(MIN_FREQUENCY_HZ, min(frequency, MAX_FREQUENCY_HZ))) profile_value = values[19] if profile_value in [int(item) for item in ProfileType]: self.profile_var.set(PROFILE_TEXT[ProfileType(profile_value)]) self.acceleration_var.set(join_u32(values[20], values[21])) self.deceleration_var.set(join_u32(values[22], values[23])) self.start_frequency_var.set(join_u32(values[24], values[25])) self.end_frequency_var.set(join_u32(values[26], values[27])) self.direction_var.set(DIRECTION_TEXT[1 if values[1] else 0]) self.direction_status_var.set(self._direction_display_text(1 if values[1] else 0)) current_position = join_i32(values[12], values[13]) self.display_target = ( abs(target - current_position) if run_mode == RunMode.ABSOLUTE else target ) self._update_status(parse_status(values[6:18])) pulse_point = values[14] direction_point = values[15] if pulse_point in PULSE_POINTS: self.pulse_point_var.set(PULSE_POINTS[pulse_point]) if direction_point in SECONDARY_POINTS: self.secondary_point_var.set(SECONDARY_POINTS[direction_point]) if len(values) >= register_read_count(Register.COMMAND): self._load_segment_registers(values, int(Register.COMMAND)) if context == "OUTPUT_CONFIG_STATUS": if values[7] == int(Error.NONE): self._set_test_result( f"输出映射已保存：第一端子Y{pulse_point}，第二端子Y{direction_point + 12}。", "pass", ) self.log("设备已确认映射生效并写入Flash；断电重启后会自动恢复。") else: self._set_test_result( f"模块映射保存失败：{error_text(values[7])}。", "fail" ) else: self.log( f"已同步设备参数：第一端子Y{pulse_point}，第二端子Y{direction_point + 12}，" f"{self.profile_var.get()}，加速{self.acceleration_var.get()}ms，" f"减速{self.deceleration_var.get()}ms，起始{self.start_frequency_var.get()}Hz，" f"默认{self.frequency_var.get()}Hz，终止{self.end_frequency_var.get()}Hz。" ) elif address == int(Register.DIRECTION) and len(values) >= 17: actual_direction = 1 if values[0] else 0 self.direction_status_var.set(self._direction_display_text(actual_direction)) self._update_status(parse_status(values[5:17])) if len(values) >= register_read_count(Register.DIRECTION): self._load_segment_registers(values, int(Register.DIRECTION)) if context == "STOP_STATUS": self.log("已回读一次暂停位置和实际方向，后续不再周期查询。") elif context == "FREQ_STATUS": self.log("已回读动态改频后的状态和实际方向。") elif address == int(Register.STATE) and len(values) >= 12: self._update_status(parse_status(values)) if context == "STOP_STATUS": self.log("已回读一次暂停位置，后续不再周期查询。") elif context == "FREQ_STATUS": self.log("已回读动态改频后的状态。") elif address == int(Register.CUMULATIVE_POSITION_HI) and len(values) >= 2: position = join_i32(values[0], values[1]) self.cumulative_position_var.set(str(position)) self.log(f"当前脉冲Y点累计位置已更新为 {position}。") elif address == int(Register.TOTAL_PULSES_LO) and len(values) >= 2: total = join_u32(values[0], values[1]) self.total_pulses_var.set(str(total)) self.log(f"总脉冲数已更新为 {total}。") _ = context def _load_segment_registers(self, values: List[int], base_address: int) -> None: """把一次连续回读中的多段配置和当前段进度同步到界面。""" def at(address: int) -> int: index = address - base_address if index < 0 or index >= len(values): raise ValueError( f"寄存器回读不完整：需要0x{address:04X}，" f"实际范围0x{base_address:04X}~" f"0x{base_address + len(values) - 1:04X}" ) return values[index] total = at(int(Register.TOTAL_SEGMENTS)) start = at(int(Register.START_SEGMENT)) current = at(int(Register.CURRENT_SEGMENT)) if 1 <= total <= 10: self.total_segments_var.set(total) if 1 <= start <= 10: self.start_segment_var.set(start) self.current_segment_var.set(str(current) if current else "0（未执行）") table_base = int(Register.SEGMENT_TABLE_BASE) for index in range(10): register = table_base + index * 4 pulses = join_i32(at(register), at(register + 1)) frequency = join_u32(at(register + 2), at(register + 3)) self.segment_pulse_vars[index].set(str(pulses)) self.segment_frequency_vars[index].set(str(frequency)) meta = int(Register.SEGMENT_META_BASE) + index * 4 jump = at(meta) wait_value = at(meta + 1) condition = join_u32(at(meta + 2), at(meta + 3)) self.segment_jump_vars[index].set(jump) if wait_value in [int(item) for item in WaitCondition]: self.segment_wait_vars[index].set( WAIT_CONDITION_TEXT[WaitCondition(wait_value)]) self.segment_condition_vars[index].set(str(condition)) def _handle_write( self, function: int, address: int, quantity_or_value: int, context: str ) -> None: if context in ("SET_SYSTEM_A", "SET_SEGMENT_SYSTEM_A"): params = self.pending_motion if context == "SET_SYSTEM_A" else self.pending_segments if params is None: self.monitoring_active = False return self.client.write_multiple( int(Register.END_FREQ_LO), split_u32(params["end_frequency"]) + [params["acceleration_ms"], params["deceleration_ms"]] + split_u32(params["maximum_frequency"]), "SET_SYSTEM_B" if context == "SET_SYSTEM_A" else "SET_SEGMENT_SYSTEM_B", ) return if context == "SET_SYSTEM_B": params = self.pending_motion if params is None: return segment_target = params["target_u32"] values = split_u32(params["frequency"]) values += split_u32(segment_target) values += [int(WaitCondition.WAIT_TIME), 0, 0, 0] self.client.write_multiple(segment_base(0), values, "SET_SINGLE_SEGMENT") return if context == "SET_SINGLE_SEGMENT": self.client.write_single(int(Register.CONTROL), int(Command.START), "START") self.pending_motion = None return if context == "SET_SEGMENT_SYSTEM_B": self._write_segment_at(0) return if context.startswith("SET_SEGMENT_") and context[12:].isdigit(): self._write_segment_at(int(context[12:]) + 1) return if context == "SET_PROFILE": params = self.pending_motion if params is None: self._set_test_result("启动参数已丢失，请重新启动。", "fail") self.monitoring_active = False return self.client.write_single( int(Register.RUN_MODE), params["run_mode"], "SET_RUN_MODE" ) self.log("加减速参数写入成功，正在写入运行模式。") elif context == "SET_SEGMENT_PROFILE": params = self.pending_segments if params is None: self._set_test_result("多段参数已丢失，请重新启动。", "fail") self.monitoring_active = False return self.client.write_multiple( int(Register.TOTAL_SEGMENTS), [params["total"], params["start"]], "SET_SEGMENT_RANGE", ) self.log("公共曲线参数写入成功，正在写入总段数和起始段。") elif context == "SET_SEGMENT_RANGE": params = self.pending_segments if params is None: return values = [] for index in range(10): segment = params["segments"][index] if index < params["total"] else (0, 0, 0, 0, 0) pulses, frequency = segment[0], segment[1] values += split_u32(pulses & 0xFFFFFFFF) + split_u32(frequency) self.client.write_multiple(int(Register.SEGMENT_TABLE_BASE), values, "SET_SEGMENT_TABLE") self.log("段范围写入成功，正在写入10段参数表。") elif context == "SET_SEGMENT_TABLE": params = self.pending_segments if params is None: return values = [] for index in range(10): segment = params["segments"][index] if index < params["total"] else (0, 0, 0, 0, 0) values += [segment[2], segment[3]] + split_u32(segment[4]) self.client.write_multiple(int(Register.SEGMENT_META_BASE), values, "SET_SEGMENT_META") self.log("10段脉冲/频率写入成功，正在写入跳转和等待条件。") elif context == "SET_SEGMENT_META": params = self.pending_segments if params is None: return self.client.write_single( int(Register.CONTROL), int(Command.SAVE_PUBLIC), "SAVE_SEGMENT_CONFIGURATION", ) elif context == "START_SEGMENTS": self.monitoring_active = True self._set_test_result("设备已确认多段启动命令，正在监控运行。", "running") self.log("多段启动命令写入成功。") elif context == "SET_RUN_MODE": params = self.pending_motion if params is None: self._set_test_result("启动参数已丢失，请重新启动。", "fail") self.monitoring_active = False return values = [int(Command.START), params["direction"]] values += split_u32(params["target_u32"]) values += split_u32(params["frequency"]) self.client.write_multiple(int(Register.COMMAND), values, "START") self.pending_motion = None self.log("运行模式写入成功，正在发送启动参数。") elif context == "START": self.monitoring_active = True self._set_test_result("设备已确认启动命令，正在监控运行状态。", "running") self.log("启动参数写入成功。") elif context == "STOP": self.monitoring_active = False self._set_test_result("设备已确认暂停，可点击“继续当前脉冲”。", "neutral") self.log("暂停命令写入成功，串口保持连接，已停止周期查询。") if self.client.is_open: self.client.read_holding( int(Register.CUMULATIVE_POSITION_LO), STATUS_REGISTER_COUNT, "STOP_STATUS", ) elif context == "RESUME": self.monitoring_active = True self._set_test_result("设备已确认继续命令，正在监控剩余脉冲。", "running") self.log("继续命令写入成功，已恢复周期查询。") elif context == "SET_RUNTIME_FREQ_VALUE": self.client.write_single( int(Register.CONTROL), int(Command.UPDATE_FREQUENCY), "APPLY_RUNTIME_FREQ", ) self.log("运行中目标频率写入成功，正在发送立即应用命令。") elif context == "APPLY_RUNTIME_FREQ": self.log("设备已接收运行中改频命令，将按公共加减速斜率调整当前段。") if self.client.is_open and not self.monitoring_active: self.client.read_holding( int(Register.CUMULATIVE_POSITION_LO), STATUS_REGISTER_COUNT, "FREQ_STATUS", ) elif context == "SET_PUBLIC_PARAMETERS": self.client.write_single( int(Register.CONTROL), int(Command.SAVE_PUBLIC), "SAVE_PUBLIC" ) self.log("公共参数寄存器写入成功，正在请求设备保存Flash。") elif context == "SAVE_PUBLIC": self.log("设备已接收公共参数保存命令，正在回读确认。") if self.client.is_open: self.client.read_holding( int(Register.PULSE_POINT), register_read_count(), "PUBLIC_CONFIG_STATUS", ) elif context == "SAVE_SEGMENT_CONFIGURATION": self.log("设备已接收段表保存命令，正在确认Flash保存结果。") self.root.after(200, self._request_segment_save_status) elif context == "SET_OUTPUT_POINTS": self.client.write_single( int(Register.OUTPUT_MODE), int(self._selected_output_mode()), "SET_OUTPUT_MODE", ) self.log("输出点寄存器写入成功，正在写入输出方式。") elif context == "SET_OUTPUT_MODE": self.client.write_single( int(Register.SEND_MODE), int(self._selected_send_mode()), "SET_SEND_MODE", ) self.log("输出方式写入成功，正在写入完成/后续发送模式。") elif context == "SET_SEND_MODE": self.client.write_single( int(Register.DIRECTION_LOGIC), int(self._selected_direction_logic()), "SET_DIRECTION_LOGIC", ) self.log("发送模式写入成功，正在写入正/负方向逻辑。") elif context == "SET_DIRECTION_LOGIC": self.client.write_single( int(Register.CONTROL), int(Command.SAVE_OUTPUTS), "SAVE_OUTPUTS" ) self.log("方向逻辑写入成功，正在发送应用并保存命令。") elif context == "SAVE_OUTPUTS": self.log("设备已接收保存命令，正在回读Flash保存结果。") if self.client.is_open: self.client.read_holding( int(Register.PULSE_POINT), register_read_count(), "OUTPUT_CONFIG_STATUS", ) elif context == "CLEAR_TOTAL": self.total_pulses_var.set("0") self.log("总脉冲数清零命令写入成功，正在回读确认。") if self.client.is_open: self.client.read_holding( int(Register.TOTAL_PULSES_LO), 2, "TOTAL_AFTER_CLEAR" ) elif context == "CLEAR_POSITION": self.cumulative_position_var.set("0") self.log("当前Y点位置清零命令写入成功，正在回读确认。") if self.client.is_open: self.client.read_holding( int(Register.CUMULATIVE_POSITION_LO), 2, "POSITION_AFTER_CLEAR", ) elif context == "CLEAR_DIAGNOSTIC": self.log("诊断记录已清零，正在回读确认。") self.request_diagnostic() _ = function, address, quantity_or_value def _finish_segment_table_write(self) -> None: """设备确认段表已写入Flash后，开放启动或自动启动。""" params = self.pending_segments if params is None: return start_after_write = self.start_after_segment_write self.start_after_segment_write = False self.written_segments = params self.pending_segments = None self.segments_written_ready = True if self.client.is_open: self.start_segments_button.configure(state=tk.NORMAL) self._set_test_result("段表写入完成，可以点击“启动”。", "pass") self.log("多段系统参数和段表写入成功，设备尚未启动。") messagebox.showinfo( "写入成功", f"段表写入成功：有效段数{params['total']}，" f"起始执行段{params['start']}。\n" "参数已保存到设备Flash。", ) if start_after_write: self.root.after(0, self.start_segment_motion) def _abort_segment_table_write(self, message: str) -> None: """统一结束失败的段表写入或自动启动流程。""" self.pending_segments = None self.written_segments = None self.segments_written_ready = False self.start_after_segment_write = False if self.client.is_open: # 失败后仍允许再次点击启动，重试自动写入。 self.start_segments_button.configure(state=tk.NORMAL) self._set_test_result(message, "fail") self.log(message) messagebox.showerror("段表写入失败", message) def _write_segment_at(self, index: int) -> None: """按参数表每段0x10步长依次写入8个有效寄存器。""" params = self.pending_segments if params is None: return if index >= params["total"]: # 段表写完后由设备统一校验公共参数和10段表并保存Flash。 self.client.write_single( int(Register.CONTROL), int(Command.SAVE_PUBLIC), "SAVE_SEGMENT_CONFIGURATION", ) return segment = params["segments"][index] pulses, frequency, jump, wait_condition, condition_value = segment wait_time = condition_value if wait_condition == int(WaitCondition.WAIT_TIME) else 0 act_time = condition_value if wait_condition == int(WaitCondition.ACT_TIME) else 0 values = split_u32(frequency) + split_u32(pulses & 0xFFFFFFFF) values += [wait_condition, wait_time, act_time, jump] self.client.write_multiple(segment_base(index), values, f"SET_SEGMENT_{index}") def _update_status(self, status: Dict[str, int]) -> None: state = status["state"] error = status["error"] current = status["current_pulses"] current_segment = status["current_segment"] display_segment = current_segment # 固件在总段数为1时复用单段执行器并上报段号0；只要本次由段表 # 启动，上位机仍按段表的起始段显示目标频率和段号。 if current_segment == 0 and self.multi_test_active: if self.written_segments is not None: display_segment = int(self.written_segments.get("start", 1)) else: display_segment = max(1, int(self.start_segment_var.get())) self._set_state_badge(state) self.error_var.set(f"{error} · {error_text(error)}") self.current_pulses_var.set(str(current)) self.actual_frequency_var.set(f"{status['current_frequency']} Hz") # “设定频率”显示当前段的目标值；“实际频率”直接使用设备状态区 # 上报的定时器生效值。加减速或运行中改频时，两者可能暂时不同。 configured_frequency = self.frequency_var.get() if 1 <= display_segment <= len(self.segment_frequency_vars): try: configured_frequency = int( self.segment_frequency_vars[display_segment - 1].get() ) if configured_frequency == 0: configured_frequency = int(self.frequency_var.get()) except ValueError: pass self.current_frequency_var.set(f"{configured_frequency} Hz") # 单段运行时固件按协议上报段号0；多段运行上报1～10。 self.current_segment_var.set( str(display_segment) if display_segment else "0（未执行）" ) if display_segment > 0: suffix = "（等待）" if state == int(State.WAITING) else "" self.running_segment_var.set(f"第 {display_segment} 段{suffix}") elif state in (int(State.RUNNING), int(State.PAUSED)): suffix = "（暂停）" if state == int(State.PAUSED) else "" self.running_segment_var.set(f"单段{suffix}") else: self.running_segment_var.set("未运行") self.cumulative_position_var.set(str(status["cumulative_position"])) self.total_pulses_var.set(str(status["total_pulses"])) self._update_diagnostic(status) # 多段运动可逐段改变正反向，当前段变化后同步显示实际方向。 if self.multi_test_active and 1 <= display_segment <= len(self.segment_pulse_vars): try: segment_direction = 1 if int(self.segment_pulse_vars[display_segment - 1].get()) < 0 else 0 self.direction_status_var.set(self._direction_display_text(segment_direction)) except ValueError: pass target = self.expected_pulses if self.test_active and self.expected_pulses else self.display_target self.pulse_fraction_var.set(f"{current} / {target}") progress = min(100.0, (current * 100.0 / target)) if target else 0.0 self.progress_var.set(progress) running = state == int(State.RUNNING) if ( self.test_active and not self.multi_test_active and running and self.expected_constant_frequency and status["current_frequency"] != self.expected_constant_frequency ): actual_frequency = status["current_frequency"] expected_frequency = self.expected_constant_frequency self.test_active = False self.monitoring_active = False self.client.write_single( int(Register.CONTROL), int(Command.STOP), "FREQUENCY_MISMATCH_STOP" ) self._set_test_result( f"测试失败：设定{expected_frequency}Hz，设备实际运行" f"{actual_frequency}Hz，已自动暂停输出。", "fail", ) self.log( f"频率一致性检查失败：设定{expected_frequency}Hz，" f"状态寄存器报告{actual_frequency}Hz，已发送暂停命令。" ) return frequency_applied = False if self.pending_frequency: if error != int(Error.NONE): self.log( f"动态改频失败：设备错误{error}，{error_text(error)}。" ) self.pending_frequency = 0 elif running: self.log( f"动态改频已生效：新目标{self.pending_frequency}Hz，" f"当前曲线频率{status['current_frequency']}Hz。" ) self._set_test_result( f"目标频率已修改为{self.pending_frequency}Hz，曲线立即重新计算。", "running", ) self.pending_frequency = 0 frequency_applied = True elif state == int(State.PAUSED): self.log( f"新频率{self.pending_frequency}Hz已保存，将在继续发送时生效。" ) self._set_test_result( f"频率{self.pending_frequency}Hz已保存，继续发送时生效。", "neutral", ) self.pending_frequency = 0 if running: self.seen_running = True if self.test_active and not frequency_applied: self._set_test_result( f"运行中：{state_text(state)}，已发送{current}个脉冲。", "running", ) elif self.test_active and state == int(State.WAITING): self._set_test_result("当前段已停止输出，正在等待设定条件。", "running") elif self.test_active and state == int(State.COMPLETE): elapsed_ok = (time.monotonic() - self.test_started_at) >= 0.05 diagnostic_flags = status.get("diagnostic_status", 0) diagnostic_ok = ( (diagnostic_flags & int(DiagnosticStatus.COMPLETE)) != 0 and (diagnostic_flags & ( int(DiagnosticStatus.COUNT_FAIL) | int(DiagnosticStatus.FREQUENCY_FAIL) | int(DiagnosticStatus.PROFILE_FAIL) )) == 0 ) if ( error == int(Error.NONE) and (self.multi_test_active or current == self.expected_pulses) and (self.seen_running or elapsed_ok) and diagnostic_ok ): if self._selected_output_mode() == OutputMode.AB_QUADRATURE: verification = "请用双通道示波器核对A/B相周期数及90°相位关系。" else: verification = "请核对脉冲端边沿数以及方向端的稳定电平。" self._set_test_result( f"测试通过：设备完成并报告{current}个脉冲，" f"个数、频率和加减速曲线诊断均通过。{verification}", "pass", ) self.log("测试通过：完成状态、脉冲计数、频率和曲线诊断均符合目标。") else: self._set_test_result( f"测试失败：目标{self.expected_pulses}，设备报告{current}，" f"错误码{error}；诊断异常数为个数" f"{status.get('diagnostic_count_errors', 0)}、频率" f"{status.get('diagnostic_frequency_errors', 0)}、曲线" f"{status.get('diagnostic_profile_errors', 0)}。", "fail", ) self.test_active = False self.multi_test_active = False self.monitoring_active = False self.log("脉冲输出完成，串口保持连接，已停止周期查询。") elif self.test_active and (state == int(State.ERROR) or error != int(Error.NONE)): self._set_test_result(f"测试失败：{error_text(error)}。", "fail") self.test_active = False self.monitoring_active = False if state != self.last_state or error != self.last_error: self.log(f"状态变化：{state_text(state)}，错误：{error_text(error)}。") self.last_state = state self.last_error = error def _update_diagnostic(self, status: Dict[str, int]) -> None: """把固件诊断位和采样值转换为易读结果。""" flags = status.get("diagnostic_status", 0) def result_text(pass_flag: DiagnosticStatus, fail_flag: DiagnosticStatus) -> str: if flags & int(fail_flag): return "异常" if flags & int(pass_flag): return "一致" if flags & int(DiagnosticStatus.ACTIVE): return "诊断中" return "未诊断" if flags & int(DiagnosticStatus.ACTIVE): overall = "诊断中" elif flags & (int(DiagnosticStatus.COUNT_FAIL) | int(DiagnosticStatus.FREQUENCY_FAIL) | int(DiagnosticStatus.PROFILE_FAIL)): overall = "诊断失败" elif flags & int(DiagnosticStatus.COMPLETE): overall = "诊断通过" else: overall = "未运行诊断" profile_value = status.get("diagnostic_profile_type", 0) try: profile_name = PROFILE_TEXT[ProfileType(profile_value)] except (ValueError, KeyError): profile_name = f"未知曲线{profile_value}" phase = DIAGNOSTIC_PHASE_TEXT.get( status.get("diagnostic_phase", 0), "未知阶段" ) self.diag_overall_var.set(overall) self.diag_count_var.set(result_text( DiagnosticStatus.COUNT_PASS, DiagnosticStatus.COUNT_FAIL )) self.diag_frequency_var.set(result_text( DiagnosticStatus.FREQUENCY_PASS, DiagnosticStatus.FREQUENCY_FAIL )) self.diag_profile_var.set( f"{result_text(DiagnosticStatus.PROFILE_PASS, DiagnosticStatus.PROFILE_FAIL)}" f"（{profile_name}）" ) self.diag_phase_var.set(phase) self.diag_pulses_var.set( f"{status.get('diagnostic_observed_pulses', 0)} / " f"{status.get('diagnostic_expected_pulses', 0)}" ) self.diag_frequency_values_var.set( f"{status.get('diagnostic_actual_frequency', 0)} Hz / " f"{status.get('diagnostic_expected_frequency', 0)} Hz" ) self.diag_max_error_var.set( f"{status.get('diagnostic_max_frequency_error', 0)} Hz" ) self.diag_error_counts_var.set( f"个数{status.get('diagnostic_count_errors', 0)}，" f"频率{status.get('diagnostic_frequency_errors', 0)}，" f"曲线{status.get('diagnostic_profile_errors', 0)}" ) def _set_state_badge(self, state: int) -> None: if state == int(State.RUNNING): colors = ("#DDEEFF", "#15558A") elif state == int(State.COMPLETE): colors = ("#DDF7EC", "#087A55") elif state == int(State.ERROR): colors = ("#FDECEC", "#B42318") elif state == int(State.PAUSED): colors = ("#FFF3D8", "#865600") elif state == int(State.WAITING): colors = ("#E8E2FF", "#5B35A5") else: colors = ("#E8EEF5", "#425466") self.state_badge.configure(text=state_text(state), bg=colors[0], fg=colors[1]) def _set_test_result(self, text: str, kind: str) -> None: # 主页面不再显示独立的运行提示条；保留最近结果供流程和调试使用。 self.last_result_text = text self.last_result_kind = kind def log(self, message: str) -> None: if not hasattr(self, "log_text"): return timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3] self.log_text.configure(state=tk.NORMAL) self.log_text.insert(tk.END, f"[{timestamp}] {message}\n") self.log_text.see(tk.END) self.log_text.configure(state=tk.DISABLED) def clear_log(self) -> None: self.log_text.configure(state=tk.NORMAL) self.log_text.delete("1.0", tk.END) self.log_text.configure(state=tk.DISABLED) self.log("日志已清空。") def _load_settings(self) -> None: try: settings = json.loads(SETTINGS_FILE.read_text(encoding="utf-8")) except (OSError, ValueError): return port = str(settings.get("port", "")) if port in self.port_combo["values"]: self.port_var.set(port) self.slave_var.set(int(settings.get("slave", DEFAULT_SLAVE_ADDRESS))) preset_index = max(0, min(int(settings.get("preset", 0)), len(PRESETS) - 1)) self.preset_var.set(PRESETS[preset_index]["name"]) self.pulses_var.set(str(settings.get("pulses", self.pulses_var.get()))) self.frequency_var.set(int(settings.get("frequency", self.frequency_var.get()))) self.runtime_frequency_var.set( int(settings.get("runtime_frequency", self.frequency_var.get())) ) self.maximum_frequency_var.set( int(settings.get("maximum_frequency", MAX_FREQUENCY_HZ)) ) saved_profile = int(settings.get("profile_type", int(ProfileType.LINEAR))) self.profile_var.set( PROFILE_TEXT[ProfileType(saved_profile)] if saved_profile in [int(item) for item in ProfileType] else PROFILE_TEXT[ProfileType.LINEAR] ) self.acceleration_var.set(int(settings.get("acceleration_ms", 1000))) self.deceleration_var.set(int(settings.get("deceleration_ms", 1000))) legacy_speed_default = min(100, max(1, int(self.frequency_var.get()))) self.start_frequency_var.set( int(settings.get("start_frequency", legacy_speed_default)) ) self.end_frequency_var.set( int(settings.get("end_frequency", legacy_speed_default)) ) self.direction_var.set( DIRECTION_TEXT[1 if int(settings.get("direction", 0)) else 0] ) saved_run_mode = int(settings.get("run_mode", int(RunMode.RELATIVE))) self.run_mode_var.set( MODE_TEXT[RunMode.ABSOLUTE] if saved_run_mode == int(RunMode.ABSOLUTE) else MODE_TEXT[RunMode.RELATIVE] ) self._on_run_mode_changed() saved_send_mode = int(settings.get("send_mode", int(SendMode.COMPLETE))) self.send_mode_var.set( SEND_MODE_TEXT[SendMode(saved_send_mode)] if saved_send_mode in [int(item) for item in SendMode] else SEND_MODE_TEXT[SendMode.COMPLETE] ) saved_direction_logic = int(settings.get( "direction_logic", int(DirectionLogic.POSITIVE) )) self.direction_logic_var.set( DIRECTION_LOGIC_TEXT[DirectionLogic(saved_direction_logic)] if saved_direction_logic in [int(item) for item in DirectionLogic] else DIRECTION_LOGIC_TEXT[DirectionLogic.POSITIVE] ) saved_pulse_point = int(settings.get("pulse_point", 0)) saved_direction_point = int(settings.get( "secondary_point", settings.get("b_phase_point", settings.get("direction_point", 0)), )) if saved_pulse_point in PULSE_POINTS: self.pulse_point_var.set(PULSE_POINTS[saved_pulse_point]) if saved_direction_point in SECONDARY_POINTS: self.secondary_point_var.set(SECONDARY_POINTS[saved_direction_point]) saved_output_mode = int(settings.get("output_mode", int(OutputMode.PULSE_DIRECTION))) if saved_output_mode in [int(item) for item in OutputMode]: self.output_mode_var.set(OUTPUT_MODE_TEXT[OutputMode(saved_output_mode)]) self._on_output_mode_changed() try: self.display_target = int(self.pulses_var.get()) except ValueError: self.display_target = 0 def _save_settings(self) -> None: preset_index = next( (index for index, item in enumerate(PRESETS) if item["name"] == self.preset_var.get()), 0, ) settings = { "port": self.port_var.get(), "slave": self.slave_var.get(), "preset": preset_index, "pulses": self.pulses_var.get(), "frequency": self.frequency_var.get(), "runtime_frequency": self.runtime_frequency_var.get(), "maximum_frequency": self.maximum_frequency_var.get(), "profile_type": next( (int(item) for item, text in PROFILE_TEXT.items() if text == self.profile_var.get()), int(ProfileType.LINEAR), ), "acceleration_ms": self.acceleration_var.get(), "deceleration_ms": self.deceleration_var.get(), "start_frequency": self.start_frequency_var.get(), "end_frequency": self.end_frequency_var.get(), "direction": 1 if self.direction_var.get() == DIRECTION_TEXT[1] else 0, "run_mode": ( int(RunMode.ABSOLUTE) if self.run_mode_var.get() == MODE_TEXT[RunMode.ABSOLUTE] else int(RunMode.RELATIVE) ), "send_mode": int(self._selected_send_mode()), "direction_logic": int(self._selected_direction_logic()), "pulse_point": point_id_from_label(PULSE_POINTS, self.pulse_point_var.get()), "secondary_point": point_id_from_label( SECONDARY_POINTS, self.secondary_point_var.get() ), "output_mode": int(self._selected_output_mode()), } try: SETTINGS_FILE.write_text( json.dumps(settings, ensure_ascii=False, indent=2), encoding="utf-8", ) except OSError as exc: self.log(f"保存界面设置失败：{exc}") def close(self) -> None: if self.closing: return self.closing = True self._save_settings() self.client.close() self.root.destroy() def main() -> int: if sys.platform != "win32": print("该工具使用Windows原生串口API，只支持Windows。") return 1 root = tk.Tk() PlsrTesterApp(root) root.mainloop() return 0 if __name__ == "__main__": raise SystemExit(main())