"""Tkinter control panel for the 2026 PLSR Modbus register map.""" from __future__ import annotations import argparse import queue import threading import time from datetime import datetime import tkinter as tk import tkinter.font as tkfont from tkinter import messagebox, ttk import serial from serial.tools import list_ports # Keep one RTU implementation for the board tests and this operator panel. from plsr_modbus_product_test import ( COMMAND_CLEAR, COMMAND_START, COMMAND_STOP, COMMON_WORDS, CONFIG_BASE, CONTROL, DIAGNOSTIC_CLEAR, DIAGNOSTIC_CONTROL, DIAG_COUNT_CHECKED, DIAG_COUNT_PASS, DIAG_CURVE_CHECKED, DIAG_CURVE_PASS, DIAG_FAULT_LATCHED, DIAG_FREQUENCY_CHECKED, DIAG_FREQUENCY_PASS, DIAG_MONITORING, ModbusException, OUTPUT_AB, OUTPUT_MODE, OUTPUT_PULSE_DIR, RtuClient, SEGMENT_1_BASE, SEGMENT_WORDS, TestFailure, read_diagnostic, read_status, split_i32, split_u32, ) BAUD_RATE = 9600 SERIAL_TIMEOUT_SECONDS = 0.8 DEFAULT_STATUS_POLL_MS = 250 MIN_STATUS_POLL_MS = 1 MAX_STATUS_POLL_MS = 3_600_000 SEGMENT_COUNT = 10 SEGMENT_STRIDE = 0x10 MAX_FREQUENCY_HZ = 100_000 OUTPUT_MODE_OPTIONS = ( ("PULSE/DIR (0)", OUTPUT_PULSE_DIR), ("AB 正交脉冲 (1)", OUTPUT_AB), ) PULSE_OPTIONS = ( ("Y0 (0)", 0), ("Y1 (1)", 1), ("Y2 (2)", 2), ("Y3 (3)", 3), ) DIRECTION_OPTIONS = ( ("Y12 (0)", 0), ("Y13 (1)", 1), ("Y14 (2)", 2), ("Y15 (3)", 3), ) INPUT_OPTIONS = (("X4 (0)", 0), ("X5 (1)", 1)) SEND_OPTIONS = (("脉冲发送完毕 (0)", 0), ("后续段 (1)", 1)) LOGIC_OPTIONS = (("正逻辑 (0)", 0), ("负逻辑 (1)", 1)) CURVE_OPTIONS = (("直线 (0)", 0), ("S 曲线 (1)", 1), ("正弦曲线 (2)", 2)) POSITION_OPTIONS = (("相对位置 (0)", 0), ("绝对位置 (1)", 1)) WAIT_OPTIONS = ( ("WAIT 时间 (0)", 0), ("WAIT 信号 (1)", 1), ("ACT 时间 (2)", 2), ("EXT 信号 (3)", 3), ("EXT 或完成 (4)", 4), ) STATUS_NAMES = { 0: "未初始化", 1: "空闲", 2: "加速", 3: "运行", 4: "减速", 5: "等待", 6: "暂停", 7: "完成", 8: "停止", 9: "错误", } ERROR_NAMES = { 0: "无错误", 1: "状态转换非法", 2: "资源冲突", 3: "资源非法", 4: "定时器错误", 5: "计数错误", 6: "正限位", 7: "负限位", 8: "急停", 9: "内部错误", } DIAGNOSTIC_REASON_NAMES = { 0: "无故障", 1: "计数", 2: "频率", 3: "曲线", 4: "无效采样", } def option_labels(options): return tuple(label for label, _value in options) def option_label(options, value): for label, candidate in options: if candidate == value: return label return str(value) def option_value(options, label, field_name): for candidate_label, value in options: if candidate_label == label: return value raise ValueError("%s 不是有效选项" % field_name) def parse_integer(text, field_name, minimum, maximum): raw = text.strip() if not raw: raise ValueError("%s 不能为空" % field_name) try: value = int(raw, 0) except ValueError: try: value = int(raw, 10) except ValueError as error: raise ValueError("%s 必须是整数" % field_name) from error if value < minimum or value > maximum: raise ValueError( "%s 必须在 %d 到 %d 之间" % (field_name, minimum, maximum) ) return value def write_common_configuration(client, words, output_mode): """Order writes so the intermediate configuration is always valid.""" if output_mode == OUTPUT_PULSE_DIR: client.write_single(OUTPUT_MODE, OUTPUT_PULSE_DIR) client.write_multiple(CONFIG_BASE, words) if output_mode == OUTPUT_AB: client.write_single(OUTPUT_MODE, OUTPUT_AB) class SerialWorker(threading.Thread): """Own the serial port and report all results through a queue.""" def __init__(self): super().__init__(name="plsr-serial", daemon=True) self.commands = queue.Queue() self.results = queue.Queue() self.stop_event = threading.Event() self.client = None self.next_status_poll = 0.0 self.polling_enabled = True self.poll_interval_seconds = DEFAULT_STATUS_POLL_MS / 1000.0 self.log_callback = None def submit(self, command, **payload): self.commands.put((command, payload)) def configure_polling(self, enabled, interval_ms): self.polling_enabled = bool(enabled) self.poll_interval_seconds = interval_ms / 1000.0 self.next_status_poll = time.monotonic() + self.poll_interval_seconds def set_log_callback(self, callback): self.log_callback = callback def _log(self, message): if self.log_callback is not None: self.log_callback("%s %s" % (datetime.now().strftime("%H:%M:%S.%f")[:-3], message)) def close(self): self.stop_event.set() self.commands.put(("shutdown", {})) def _emit(self, event, **payload): self.results.put((event, payload)) def _disconnect(self, notify=True, reason=""): client = self.client self.client = None if client is not None: try: client.__exit__(None, None, None) except (OSError, serial.SerialException): pass if notify: self._emit("connection", connected=False, reason=reason) def _require_client(self): if self.client is None: raise RuntimeError("串口尚未连接") return self.client def _read_configuration(self): client = self._require_client() common = client.read_holding(CONFIG_BASE, COMMON_WORDS) output_mode = client.read_holding(OUTPUT_MODE, 1)[0] segments = [] for index in range(SEGMENT_COUNT): address = SEGMENT_1_BASE + index * SEGMENT_STRIDE segments.append(client.read_holding(address, SEGMENT_WORDS)) self._emit( "configuration", common=common, output_mode=output_mode, segments=segments, ) def _handle_connection_error(self, operation, error): self._disconnect(notify=True, reason=str(error)) self._emit("error", operation=operation, message=str(error), modal=True) def _handle_command(self, command, payload): if command == "connect": self._disconnect(notify=False) try: client = RtuClient( payload["port"], BAUD_RATE, payload["slave"], SERIAL_TIMEOUT_SECONDS, ) client.__enter__() client.log_callback = self._log self.client = client self.next_status_poll = 0.0 self._emit( "connection", connected=True, port=payload["port"], slave=payload["slave"], ) self._read_configuration() except (OSError, serial.SerialException) as error: self._handle_connection_error("连接", error) except (TestFailure, ModbusException, RuntimeError) as error: self._disconnect(notify=True, reason=str(error)) self._emit( "error", operation="读取参数", message=str(error), modal=True ) return if command == "disconnect": self._disconnect(notify=True) return try: client = self._require_client() if command == "read_configuration": self._read_configuration() self._emit("operation", message="参数读取完成") elif command == "write_common": write_common_configuration( client, payload["words"], payload["output_mode"] ) common = client.read_holding(CONFIG_BASE, COMMON_WORDS) output_mode = client.read_holding(OUTPUT_MODE, 1)[0] self._emit("common", words=common, output_mode=output_mode) self._emit("operation", message="公共参数写入并回读完成") elif command == "write_segments": for index, words in enumerate(payload["segments"]): address = SEGMENT_1_BASE + index * SEGMENT_STRIDE client.write_multiple(address, words) segments = [] for index in range(SEGMENT_COUNT): address = SEGMENT_1_BASE + index * SEGMENT_STRIDE segments.append(client.read_holding(address, SEGMENT_WORDS)) self._emit("segments", words=segments) self._emit("operation", message="10 段参数写入并回读完成") elif command == "control": client.write_single(CONTROL, payload["value"]) self._emit("operation", message=payload["message"]) elif command == "clear_diagnostic": client.write_single(DIAGNOSTIC_CONTROL, DIAGNOSTIC_CLEAR) self._emit("diagnostic", diagnostic=read_diagnostic(client)) self._emit("operation", message="诊断锁存和统计已清除") else: raise RuntimeError("未知后台命令: %s" % command) except (OSError, serial.SerialException) as error: self._handle_connection_error(payload.get("operation", "通信"), error) except (TestFailure, ModbusException, RuntimeError) as error: self._emit( "error", operation=payload.get("operation", "通信"), message=str(error), modal=True, ) def _poll_status(self): try: status = read_status(self._require_client()) diagnostic = read_diagnostic(self._require_client()) self._emit("status", status=status) self._emit("diagnostic", diagnostic=diagnostic) self.next_status_poll = time.monotonic() + self.poll_interval_seconds except (OSError, serial.SerialException) as error: self._handle_connection_error("状态轮询", error) except (TestFailure, ModbusException, RuntimeError) as error: self._emit( "error", operation="状态轮询", message=str(error), modal=False ) self.next_status_poll = time.monotonic() + self.poll_interval_seconds def run(self): try: while not self.stop_event.is_set(): try: command, payload = self.commands.get(timeout=0.05) except queue.Empty: command = None payload = None if command == "shutdown": break if command is not None: self._handle_command(command, payload) if ( self.polling_enabled and self.client is not None and time.monotonic() >= self.next_status_poll ): self._poll_status() finally: self._disconnect(notify=False) class PlsrControlPanel: def __init__(self, root): self.root = root self.root.title("PLSR 控制面板") self.root.geometry("1180x720") self.root.minsize(980, 650) self.root.protocol("WM_DELETE_WINDOW", self._on_close) self.connected = False self.operation_pending = False self.worker = SerialWorker() self.worker.set_log_callback(self._queue_log) self.worker.start() self.log_window = None self.log_text = None self.port_var = tk.StringVar() self.slave_var = tk.StringVar(value="1") self.polling_enabled_var = tk.BooleanVar(value=True) self.poll_interval_var = tk.StringVar(value=str(DEFAULT_STATUS_POLL_MS)) self.connection_var = tk.StringVar(value="未连接") self.footer_var = tk.StringVar(value="请选择串口并连接") self.status_vars = { "position": tk.StringVar(value="--"), "frequency": tk.StringVar(value="--"), "state": tk.StringVar(value="--"), "segment": tk.StringVar(value="--"), "error": tk.StringVar(value="--"), } self.diagnostic_vars = { key: tk.StringVar(value="--") for key in ( "summary", "reason", "segment", "mode_direction", "expected_count", "actual_count", "count_error", "requested_hz", "expected_hz", "active_hz", "request_error", "sample_count", "mismatch_count", "maximum_error", "first_mismatch", ) } self.common_vars = {} self.segment_vars = [] self.connection_widgets = [] self.action_buttons = [] self._configure_style() self._build_ui() self.refresh_ports() self._set_connected(False) self.root.after(50, self._drain_results) def _configure_style(self): default_font = tkfont.nametofont("TkDefaultFont") default_font.configure(family="Microsoft YaHei UI", size=10) text_font = tkfont.nametofont("TkTextFont") text_font.configure(family="Microsoft YaHei UI", size=10) style = ttk.Style(self.root) style.configure("Connected.TLabel", foreground="#18794e") style.configure("Disconnected.TLabel", foreground="#5f6368") style.configure("Error.TLabel", foreground="#b42318") style.configure("StatusValue.TLabel", font=("Microsoft YaHei UI", 11, "bold")) def _build_ui(self): self.root.columnconfigure(0, weight=1) self.root.rowconfigure(2, weight=1) connection = ttk.LabelFrame(self.root, text="连接") connection.grid(row=0, column=0, padx=10, pady=(10, 6), sticky="ew") connection.columnconfigure(7, weight=1) ttk.Label(connection, text="串口").grid(row=0, column=0, padx=(8, 4), pady=8) self.port_box = ttk.Combobox( connection, textvariable=self.port_var, width=13, state="readonly" ) self.port_box.grid(row=0, column=1, padx=4, pady=8) refresh_button = ttk.Button(connection, text="刷新", command=self.refresh_ports) refresh_button.grid(row=0, column=2, padx=(0, 12), pady=8) ttk.Label(connection, text="串口参数").grid(row=0, column=3, padx=4, pady=8) ttk.Label(connection, text="9600, 8E1").grid(row=0, column=4, padx=(4, 12), pady=8) ttk.Label(connection, text="从站地址").grid(row=0, column=5, padx=4, pady=8) slave_entry = ttk.Entry(connection, textvariable=self.slave_var, width=7) slave_entry.grid(row=0, column=6, padx=(4, 12), pady=8) self.connect_button = ttk.Button( connection, text="连接", command=self._toggle_connection ) self.connect_button.grid(row=0, column=8, padx=8, pady=8) polling = ttk.LabelFrame(connection, text="状态轮询") polling.grid(row=1, column=0, columnspan=9, padx=8, pady=(0, 6), sticky="ew") ttk.Checkbutton( polling, text="启动轮询", variable=self.polling_enabled_var, command=self._apply_polling_settings, ).grid(row=0, column=0, padx=(8, 4), pady=6) ttk.Label(polling, text="轮询间隔").grid(row=0, column=1, padx=(12, 4), pady=6) ttk.Entry(polling, textvariable=self.poll_interval_var, width=9).grid( row=0, column=2, padx=4, pady=6 ) ttk.Label(polling, text="ms（最小 1 ms）").grid( row=0, column=3, padx=(4, 8), pady=6 ) self.connection_label = ttk.Label( connection, textvariable=self.connection_var, style="Disconnected.TLabel", width=28, anchor="e", ) self.connection_label.grid(row=0, column=7, padx=8, pady=8, sticky="e") self.connection_widgets = [self.port_box, refresh_button, slave_entry] ttk.Button(connection, text="通信日志", command=self._show_log_window).grid( row=0, column=9, padx=8, pady=8 ) self._build_runtime_status() notebook = ttk.Notebook(self.root) notebook.grid(row=2, column=0, padx=10, pady=6, sticky="nsew") common_tab = ttk.Frame(notebook, padding=12) segment_tab = ttk.Frame(notebook, padding=12) diagnostic_tab = ttk.Frame(notebook, padding=12) notebook.add(common_tab, text="公共参数 0x1000-0x1013") notebook.add(segment_tab, text="10 段参数 0x1100-0x1197") notebook.add(diagnostic_tab, text="输出诊断 0x2100-0x2119") self._build_common_tab(common_tab) self._build_segment_tab(segment_tab) self._build_diagnostic_tab(diagnostic_tab) footer = ttk.Frame(self.root) footer.grid(row=3, column=0, padx=10, pady=(4, 10), sticky="ew") footer.columnconfigure(4, weight=1) read_button = ttk.Button(footer, text="读取全部参数", command=self._read_all) read_button.grid(row=0, column=0, padx=(0, 6)) common_button = ttk.Button( footer, text="写入公共参数", command=self._write_common ) common_button.grid(row=0, column=1, padx=6) segment_button = ttk.Button( footer, text="写入 10 段参数", command=self._write_segments ) segment_button.grid(row=0, column=2, padx=6) self.action_buttons.extend([read_button, common_button, segment_button]) self.footer_label = ttk.Label( footer, textvariable=self.footer_var, anchor="e" ) self.footer_label.grid(row=0, column=4, padx=(12, 0), sticky="ew") def _build_runtime_status(self): frame = ttk.LabelFrame(self.root, text="实时状态 0x2000-0x2006") frame.grid(row=1, column=0, padx=10, pady=6, sticky="ew") for column in range(11): frame.columnconfigure(column, weight=1 if column % 2 else 0) fields = ( ("累计位置", "position"), ("当前频率", "frequency"), ("运行状态", "state"), ("当前段", "segment"), ("错误码", "error"), ) for index, (label, key) in enumerate(fields): ttk.Label(frame, text=label).grid( row=0, column=index * 2, padx=(8, 4), pady=9, sticky="e" ) ttk.Label( frame, textvariable=self.status_vars[key], style="StatusValue.TLabel" ).grid(row=0, column=index * 2 + 1, padx=(4, 12), pady=9, sticky="w") commands = ttk.Frame(frame) commands.grid(row=0, column=10, padx=8, pady=6, sticky="e") start_button = ttk.Button( commands, text="启动", command=lambda: self._send_control(COMMAND_START) ) stop_button = ttk.Button( commands, text="停止", command=lambda: self._send_control(COMMAND_STOP) ) clear_button = ttk.Button(commands, text="清零", command=self._clear_position) start_button.grid(row=0, column=0, padx=3) stop_button.grid(row=0, column=1, padx=3) clear_button.grid(row=0, column=2, padx=3) self.action_buttons.extend([start_button, stop_button, clear_button]) def _add_entry(self, parent, row, group, key, label, default="0", width=13): column = group * 2 ttk.Label(parent, text=label).grid( row=row, column=column, padx=(8, 4), pady=7, sticky="e" ) variable = tk.StringVar(value=default) ttk.Entry(parent, textvariable=variable, width=width).grid( row=row, column=column + 1, padx=(4, 18), pady=7, sticky="w" ) self.common_vars[key] = variable def _add_combo(self, parent, row, group, key, label, options): column = group * 2 ttk.Label(parent, text=label).grid( row=row, column=column, padx=(8, 4), pady=7, sticky="e" ) variable = tk.StringVar(value=options[0][0]) ttk.Combobox( parent, textvariable=variable, values=option_labels(options), state="readonly", width=18, ).grid(row=row, column=column + 1, padx=(4, 18), pady=7, sticky="w") self.common_vars[key] = variable def _build_common_tab(self, parent): for column in (1, 3, 5): parent.columnconfigure(column, weight=1) self._add_combo( parent, 0, 0, "pulse_output", "脉冲输出（单逻辑 PLSR）", PULSE_OPTIONS ) self._add_combo( parent, 0, 1, "direction_output", "方向输出", DIRECTION_OPTIONS ) self._add_entry(parent, 0, 2, "direction_delay", "方向延时 ms") self._add_combo(parent, 1, 0, "wait_input", "WAIT 输入", INPUT_OPTIONS) self._add_combo(parent, 1, 1, "ext_input", "EXT 输入", INPUT_OPTIONS) self._add_combo(parent, 1, 2, "send_mode", "发送模式", SEND_OPTIONS) self._add_combo(parent, 2, 0, "negative_logic", "方向逻辑", LOGIC_OPTIONS) self._add_combo(parent, 2, 1, "curve_mode", "曲线模式", CURVE_OPTIONS) self._add_combo(parent, 2, 2, "position_mode", "位置模式", POSITION_OPTIONS) self._add_entry(parent, 3, 0, "segment_count", "段数", default="1") self._add_entry(parent, 3, 1, "start_segment", "起始段", default="1") self._add_entry(parent, 3, 2, "default_speed", "默认速度 Hz", default="1000") self._add_entry(parent, 4, 0, "start_speed", "启动速度 Hz", default="100") self._add_entry(parent, 4, 1, "stop_speed", "停止速度 Hz", default="100") self._add_entry(parent, 4, 2, "acceleration", "加速时间 ms") self._add_entry(parent, 5, 0, "deceleration", "减速时间 ms") self._add_combo( parent, 5, 1, "output_mode", "输出模式 0x1200", OUTPUT_MODE_OPTIONS ) def _build_diagnostic_tab(self, parent): parent.columnconfigure(1, weight=1) parent.columnconfigure(3, weight=1) parent.columnconfigure(5, weight=1) fields = ( ("综合结果", "summary"), ("首个故障", "reason"), ("段号", "segment"), ("模式 / 方向", "mode_direction"), ("期望脉冲", "expected_count"), ("实测脉冲", "actual_count"), ("计数误差", "count_error"), ("请求频率", "requested_hz"), ("量化频率", "expected_hz"), ("活动频率", "active_hz"), ("请求量化误差", "request_error"), ("曲线采样数", "sample_count"), ("不匹配数", "mismatch_count"), ("最大频率误差", "maximum_error"), ("首个不匹配样本", "first_mismatch"), ) for index, (label, key) in enumerate(fields): row = index // 3 group = index % 3 column = group * 2 ttk.Label(parent, text=label).grid( row=row, column=column, padx=(8, 4), pady=9, sticky="e" ) ttk.Label( parent, textvariable=self.diagnostic_vars[key], style="StatusValue.TLabel" if key == "summary" else "TLabel", ).grid(row=row, column=column + 1, padx=(4, 18), pady=9, sticky="w") clear_button = ttk.Button( parent, text="清除诊断", command=self._clear_diagnostic ) clear_button.grid(row=5, column=5, padx=8, pady=(16, 4), sticky="e") self.action_buttons.append(clear_button) def _build_segment_tab(self, parent): headers = ( ("段 / 基址", 15), ("频率 Hz", 13), ("脉冲数", 15), ("等待类型", 19), ("WAIT ms", 11), ("ACT ms", 11), ("跳转段", 9), ) for column, (label, _width) in enumerate(headers): parent.columnconfigure(column, weight=1 if column in (1, 2, 3) else 0) ttk.Label(parent, text=label, anchor="center").grid( row=0, column=column, padx=4, pady=(0, 6), sticky="ew" ) for index in range(SEGMENT_COUNT): address = SEGMENT_1_BASE + index * SEGMENT_STRIDE ttk.Label(parent, text="%d / 0x%04X" % (index + 1, address)).grid( row=index + 1, column=0, padx=(2, 8), pady=4, sticky="e" ) variables = { "frequency": tk.StringVar(value="1000"), "pulses": tk.StringVar(value="0"), "wait_type": tk.StringVar(value=WAIT_OPTIONS[0][0]), "wait_ms": tk.StringVar(value="0"), "act_ms": tk.StringVar(value="0"), "jump": tk.StringVar(value="0"), } ttk.Entry(parent, textvariable=variables["frequency"], width=13).grid( row=index + 1, column=1, padx=4, pady=4, sticky="ew" ) ttk.Entry(parent, textvariable=variables["pulses"], width=15).grid( row=index + 1, column=2, padx=4, pady=4, sticky="ew" ) ttk.Combobox( parent, textvariable=variables["wait_type"], values=option_labels(WAIT_OPTIONS), state="readonly", width=19, ).grid(row=index + 1, column=3, padx=4, pady=4, sticky="ew") ttk.Entry(parent, textvariable=variables["wait_ms"], width=11).grid( row=index + 1, column=4, padx=4, pady=4 ) ttk.Entry(parent, textvariable=variables["act_ms"], width=11).grid( row=index + 1, column=5, padx=4, pady=4 ) ttk.Entry(parent, textvariable=variables["jump"], width=9).grid( row=index + 1, column=6, padx=4, pady=4 ) self.segment_vars.append(variables) def refresh_ports(self): ports = [item.device for item in list_ports.comports()] self.port_box["values"] = ports if ports and self.port_var.get() not in ports: self.port_var.set(ports[0]) elif not ports: self.port_var.set("") def _queue_log(self, message): self.worker.results.put(("log", {"message": message})) def _show_log_window(self): if self.log_window is not None and self.log_window.winfo_exists(): self.log_window.deiconify() self.log_window.lift() return self.log_window = tk.Toplevel(self.root) self.log_window.title("Modbus 通信日志") self.log_window.geometry("760x420") self.log_text = tk.Text(self.log_window, wrap="none", state="disabled") self.log_text.pack(fill="both", expand=True, padx=8, pady=8) ttk.Button(self.log_window, text="清空日志", command=self._clear_log).pack(pady=(0, 8)) def _clear_log(self): if self.log_text is not None: self.log_text.configure(state="normal") self.log_text.delete("1.0", "end") self.log_text.configure(state="disabled") def _polling_settings(self): interval_ms = parse_integer( self.poll_interval_var.get(), "轮询间隔", MIN_STATUS_POLL_MS, MAX_STATUS_POLL_MS ) return self.polling_enabled_var.get(), interval_ms def _apply_polling_settings(self): try: enabled, interval_ms = self._polling_settings() except ValueError as error: messagebox.showerror("轮询设置", str(error)) return self.worker.configure_polling(enabled, interval_ms) self.footer_var.set( "轮询已启动，间隔 %d ms" % interval_ms if enabled else "轮询已停止" ) def _set_connected(self, connected, description=""): self.connected = connected self.connect_button.configure( text="断开" if connected else "连接", state="normal" ) connection_state = "disabled" if connected else "normal" for widget in self.connection_widgets: if widget is self.port_box: widget.configure(state="disabled" if connected else "readonly") else: widget.configure(state=connection_state) if connected: self.connection_var.set(description or "已连接") self.connection_label.configure(style="Connected.TLabel") else: self.connection_var.set(description or "未连接") self.connection_label.configure( style="Error.TLabel" if description else "Disconnected.TLabel" ) for variable in self.status_vars.values(): variable.set("--") for variable in self.diagnostic_vars.values(): variable.set("--") self._update_action_state() def _update_action_state(self): state = "normal" if self.connected and not self.operation_pending else "disabled" for button in self.action_buttons: button.configure(state=state) def _start_operation(self, message): if not self.connected: messagebox.showerror("未连接", "请先连接 PLSR 从站") return False if self.operation_pending: return False self.operation_pending = True self.footer_var.set(message) self._update_action_state() return True def _finish_operation(self, message): self.operation_pending = False self.footer_var.set(message) self.footer_label.configure(style="TLabel") self._update_action_state() def _toggle_connection(self): if self.connected: self.operation_pending = False self.footer_var.set("正在断开") self.worker.submit("disconnect") return port = self.port_var.get().strip() if not port: messagebox.showerror("连接参数", "请选择可用串口") return try: slave = parse_integer(self.slave_var.get(), "从站地址", 1, 247) except ValueError as error: messagebox.showerror("连接参数", str(error)) return self.connect_button.configure(state="disabled") self.operation_pending = True self._update_action_state() self.connection_var.set("正在连接 %s" % port) self.footer_var.set("正在打开串口并读取参数") try: enabled, interval_ms = self._polling_settings() except ValueError as error: self.connect_button.configure(state="normal") self.operation_pending = False self._update_action_state() messagebox.showerror("轮询设置", str(error)) return self.worker.configure_polling(enabled, interval_ms) self.worker.submit("connect", port=port, slave=slave) def _common_words(self): values = self.common_vars segment_count = parse_integer(values["segment_count"].get(), "段数", 1, 10) start_segment = parse_integer( values["start_segment"].get(), "起始段", 1, 10 ) if start_segment > segment_count: raise ValueError("起始段不能大于段数") words = [0] * COMMON_WORDS words[0] = option_value( PULSE_OPTIONS, values["pulse_output"].get(), "脉冲输出" ) words[1] = option_value( DIRECTION_OPTIONS, values["direction_output"].get(), "方向输出" ) words[2] = option_value(INPUT_OPTIONS, values["wait_input"].get(), "WAIT 输入") words[3] = option_value(INPUT_OPTIONS, values["ext_input"].get(), "EXT 输入") words[4] = option_value(SEND_OPTIONS, values["send_mode"].get(), "发送模式") words[5] = parse_integer( values["direction_delay"].get(), "方向延时", 0, 0xFFFF ) words[6] = option_value( LOGIC_OPTIONS, values["negative_logic"].get(), "方向逻辑" ) words[7] = option_value(CURVE_OPTIONS, values["curve_mode"].get(), "曲线模式") words[8] = option_value( POSITION_OPTIONS, values["position_mode"].get(), "位置模式" ) words[9] = segment_count words[10] = start_segment words[11:13] = split_u32( parse_integer( values["default_speed"].get(), "默认速度", 1, MAX_FREQUENCY_HZ ) ) words[13:15] = split_u32( parse_integer( values["start_speed"].get(), "启动速度", 0, MAX_FREQUENCY_HZ ) ) words[15] = 0 words[16:18] = split_u32( parse_integer( values["stop_speed"].get(), "停止速度", 0, MAX_FREQUENCY_HZ ) ) words[18] = parse_integer(values["acceleration"].get(), "加速时间", 0, 0xFFFF) words[19] = parse_integer(values["deceleration"].get(), "减速时间", 0, 0xFFFF) return words def _segment_words(self): segment_count = parse_integer( self.common_vars["segment_count"].get(), "段数", 1, SEGMENT_COUNT ) encoded = [] for index, variables in enumerate(self.segment_vars): name = "第 %d 段" % (index + 1) frequency = parse_integer( variables["frequency"].get(), name + "频率", 0, MAX_FREQUENCY_HZ ) pulses = parse_integer( variables["pulses"].get(), name + "脉冲数", -(1 << 31), (1 << 31) - 1 ) wait_type = option_value( WAIT_OPTIONS, variables["wait_type"].get(), name + "等待类型" ) wait_ms = parse_integer( variables["wait_ms"].get(), name + " WAIT 时间", 0, 0xFFFF ) act_ms = parse_integer( variables["act_ms"].get(), name + " ACT 时间", 0, 0xFFFF ) jump = parse_integer(variables["jump"].get(), name + "跳转段", 0, 10) if index < segment_count and jump > segment_count: raise ValueError( "%s跳转段必须为 0 或不大于脉冲总段数 %d" % (name, segment_count) ) words = split_u32(frequency) + split_i32(pulses) words.extend((wait_type, wait_ms, act_ms, jump)) encoded.append(words) return encoded def _set_common_words(self, words): if len(words) != COMMON_WORDS: raise ValueError("公共参数回读长度错误") values = self.common_vars values["pulse_output"].set(option_label(PULSE_OPTIONS, words[0])) values["direction_output"].set(option_label(DIRECTION_OPTIONS, words[1])) values["wait_input"].set(option_label(INPUT_OPTIONS, words[2])) values["ext_input"].set(option_label(INPUT_OPTIONS, words[3])) values["send_mode"].set(option_label(SEND_OPTIONS, words[4])) values["direction_delay"].set(str(words[5])) values["negative_logic"].set(option_label(LOGIC_OPTIONS, words[6])) values["curve_mode"].set(option_label(CURVE_OPTIONS, words[7])) values["position_mode"].set(option_label(POSITION_OPTIONS, words[8])) values["segment_count"].set(str(words[9])) values["start_segment"].set(str(words[10])) values["default_speed"].set(str(words[11] | (words[12] << 16))) values["start_speed"].set(str(words[13] | (words[14] << 16))) values["stop_speed"].set(str(words[16] | (words[17] << 16))) values["acceleration"].set(str(words[18])) values["deceleration"].set(str(words[19])) def _set_segment_words(self, segments): if len(segments) != SEGMENT_COUNT: raise ValueError("段参数回读数量错误") for index, words in enumerate(segments): if len(words) != SEGMENT_WORDS: raise ValueError("第 %d 段回读长度错误" % (index + 1)) variables = self.segment_vars[index] frequency = words[0] | (words[1] << 16) pulse_bits = words[2] | (words[3] << 16) pulses = pulse_bits - (1 << 32) if pulse_bits & 0x80000000 else pulse_bits variables["frequency"].set(str(frequency)) variables["pulses"].set(str(pulses)) variables["wait_type"].set(option_label(WAIT_OPTIONS, words[4])) variables["wait_ms"].set(str(words[5])) variables["act_ms"].set(str(words[6])) variables["jump"].set(str(words[7])) def _read_all(self): if self._start_operation("正在读取公共参数和 10 段参数"): self.worker.submit("read_configuration", operation="读取参数") def _write_common(self): try: words = self._common_words() output_mode = option_value( OUTPUT_MODE_OPTIONS, self.common_vars["output_mode"].get(), "输出模式", ) if output_mode == OUTPUT_AB and words[0] not in (0, 2): raise ValueError("AB 模式的脉冲输出只能选择 Y0/Y1 或 Y2/Y3") except ValueError as error: messagebox.showerror("参数错误", str(error)) return if self._start_operation("正在写入公共参数"): self.worker.submit( "write_common", words=words, output_mode=output_mode, operation="写入公共参数", ) def _write_segments(self): try: segments = self._segment_words() except ValueError as error: messagebox.showerror("参数错误", str(error)) return if self._start_operation("正在写入 10 段参数"): self.worker.submit( "write_segments", segments=segments, operation="写入段参数" ) def _send_control(self, command): messages = { COMMAND_START: "启动命令已发送", COMMAND_STOP: "停止命令已发送", COMMAND_CLEAR: "清零命令已发送", } if self._start_operation("正在发送控制命令"): self.worker.submit( "control", value=command, message=messages[command], operation="控制命令", ) def _clear_position(self): if messagebox.askyesno("确认清零", "确认将累计位置清零？"): self._send_control(COMMAND_CLEAR) def _clear_diagnostic(self): if self._start_operation("正在清除诊断锁存和统计"): self.worker.submit( "clear_diagnostic", operation="清除诊断" ) def _show_status(self, status): self.status_vars["position"].set(str(status.position)) self.status_vars["frequency"].set("%d Hz" % status.frequency_hz) self.status_vars["state"].set( "%s (%d)" % (STATUS_NAMES.get(status.state, "未知"), status.state) ) self.status_vars["segment"].set(str(status.segment)) self.status_vars["error"].set( "%s (%d)" % (ERROR_NAMES.get(status.error, "未知"), status.error) ) def _show_diagnostic(self, diagnostic): def check_name(checked_bit, pass_bit): if not diagnostic.flags & checked_bit: return "未检查" return "通过" if diagnostic.flags & pass_bit else "失败" checks = "计数 %s / 频率 %s / 曲线 %s" % ( check_name(DIAG_COUNT_CHECKED, DIAG_COUNT_PASS), check_name(DIAG_FREQUENCY_CHECKED, DIAG_FREQUENCY_PASS), check_name(DIAG_CURVE_CHECKED, DIAG_CURVE_PASS), ) if diagnostic.flags & DIAG_MONITORING: checks = "监测中 / " + checks if diagnostic.flags & DIAG_FAULT_LATCHED: checks = "故障锁存 / " + checks direction = "正向" if diagnostic.direction_positive else "反向" mode = "AB" if diagnostic.output_mode == OUTPUT_AB else "PULSE/DIR" first_mismatch = ( "无" if diagnostic.first_mismatch_sample == 0xFFFFFFFF else str(diagnostic.first_mismatch_sample) ) values = self.diagnostic_vars values["summary"].set(checks) values["reason"].set( "%s (%d)" % ( DIAGNOSTIC_REASON_NAMES.get(diagnostic.reason, "未知"), diagnostic.reason, ) ) values["segment"].set(str(diagnostic.segment)) values["mode_direction"].set("%s / %s" % (mode, direction)) values["expected_count"].set(str(diagnostic.expected_pulses)) values["actual_count"].set(str(diagnostic.actual_pulses)) values["count_error"].set(str(diagnostic.count_error)) values["requested_hz"].set("%d Hz" % diagnostic.requested_hz) values["expected_hz"].set("%d Hz" % diagnostic.expected_timer_hz) values["active_hz"].set("%d Hz" % diagnostic.active_timer_hz) values["request_error"].set("%d Hz" % diagnostic.request_error_hz) values["sample_count"].set(str(diagnostic.sample_count)) values["mismatch_count"].set(str(diagnostic.mismatch_count)) values["maximum_error"].set( "%d Hz" % diagnostic.maximum_frequency_error_hz ) values["first_mismatch"].set(first_mismatch) def _handle_result(self, event, payload): if event == "connection": if payload["connected"]: description = "%s / 从站 %d" % (payload["port"], payload["slave"]) self._set_connected(True, description) self.footer_var.set("已连接，正在读取参数") else: reason = payload.get("reason", "") self.operation_pending = False self.connect_button.configure(state="normal") self._set_connected(False, reason) self.footer_var.set("连接已断开" if not reason else "通信错误") elif event == "configuration": self._set_common_words(payload["common"]) self.common_vars["output_mode"].set( option_label(OUTPUT_MODE_OPTIONS, payload["output_mode"]) ) self._set_segment_words(payload["segments"]) self._finish_operation("全部参数已读取") elif event == "common": self._set_common_words(payload["words"]) self.common_vars["output_mode"].set( option_label(OUTPUT_MODE_OPTIONS, payload["output_mode"]) ) elif event == "segments": self._set_segment_words(payload["words"]) elif event == "status": self._show_status(payload["status"]) elif event == "diagnostic": self._show_diagnostic(payload["diagnostic"]) elif event == "log": if self.log_text is not None and self.log_text.winfo_exists(): self.log_text.configure(state="normal") self.log_text.insert("end", payload["message"] + "\n") self.log_text.see("end") self.log_text.configure(state="disabled") elif event == "operation": self._finish_operation(payload["message"]) elif event == "error": message = "%s失败：%s" % (payload["operation"], payload["message"]) self._finish_operation(message) self.footer_label.configure(style="Error.TLabel") if payload.get("modal", False): messagebox.showerror("PLSR 通信错误", message) def _drain_results(self): try: while True: event, payload = self.worker.results.get_nowait() try: self._handle_result(event, payload) except (KeyError, ValueError) as error: self._finish_operation("界面数据错误：%s" % error) except queue.Empty: pass self.root.after(50, self._drain_results) def _on_close(self): self.worker.close() self.root.destroy() class _SelfTestVariable: def __init__(self, value): self.value = str(value) def get(self): return self.value def set(self, value): self.value = str(value) class _SelfTestClient: def __init__(self): self.calls = [] def write_single(self, address, value): self.calls.append(("single", address, value)) def write_multiple(self, address, values): self.calls.append(("multiple", address, list(values))) def self_test(): assert BAUD_RATE == 9600 assert INPUT_OPTIONS == (("X4 (0)", 0), ("X5 (1)", 1)) assert [SEGMENT_1_BASE + index * SEGMENT_STRIDE for index in range(SEGMENT_COUNT)] == [ 0x1100, 0x1110, 0x1120, 0x1130, 0x1140, 0x1150, 0x1160, 0x1170, 0x1180, 0x1190, ] assert parse_integer("0x10", "测试值", 0, 16) == 16 try: parse_integer("17", "测试值", 0, 16) except ValueError: pass else: raise AssertionError("range validation did not reject 17") panel = PlsrControlPanel.__new__(PlsrControlPanel) panel.common_vars = { "pulse_output": _SelfTestVariable(PULSE_OPTIONS[3][0]), "direction_output": _SelfTestVariable(DIRECTION_OPTIONS[3][0]), "wait_input": _SelfTestVariable(INPUT_OPTIONS[1][0]), "ext_input": _SelfTestVariable(INPUT_OPTIONS[0][0]), "send_mode": _SelfTestVariable(SEND_OPTIONS[1][0]), "direction_delay": _SelfTestVariable(25), "negative_logic": _SelfTestVariable(LOGIC_OPTIONS[1][0]), "curve_mode": _SelfTestVariable(CURVE_OPTIONS[2][0]), "position_mode": _SelfTestVariable(POSITION_OPTIONS[1][0]), "segment_count": _SelfTestVariable(10), "start_segment": _SelfTestVariable(2), "default_speed": _SelfTestVariable(100000), "start_speed": _SelfTestVariable(0), "stop_speed": _SelfTestVariable(65536), "acceleration": _SelfTestVariable(1000), "deceleration": _SelfTestVariable(2000), "output_mode": _SelfTestVariable(OUTPUT_MODE_OPTIONS[1][0]), } common = panel._common_words() assert common[:11] == [3, 3, 1, 0, 1, 25, 1, 2, 1, 10, 2] assert common[11:20] == [34464, 1, 0, 0, 0, 0, 1, 1000, 2000] assert option_value( OUTPUT_MODE_OPTIONS, panel.common_vars["output_mode"].get(), "输出模式", ) == OUTPUT_AB assert OUTPUT_MODE == 0x1200 assert DIAGNOSTIC_CONTROL == 0x3100 client = _SelfTestClient() write_common_configuration(client, common, OUTPUT_PULSE_DIR) assert client.calls[0] == ("single", OUTPUT_MODE, OUTPUT_PULSE_DIR) assert client.calls[1][0:2] == ("multiple", CONFIG_BASE) client = _SelfTestClient() write_common_configuration(client, common, OUTPUT_AB) assert client.calls[0][0:2] == ("multiple", CONFIG_BASE) assert client.calls[1] == ("single", OUTPUT_MODE, OUTPUT_AB) panel.segment_vars = [] for index in range(SEGMENT_COUNT): panel.segment_vars.append({ "frequency": _SelfTestVariable(1000 + index), "pulses": _SelfTestVariable(-1 if index == 0 else index), "wait_type": _SelfTestVariable(WAIT_OPTIONS[4][0]), "wait_ms": _SelfTestVariable(index), "act_ms": _SelfTestVariable(index + 1), "jump": _SelfTestVariable(0), }) segments = panel._segment_words() assert len(segments) == SEGMENT_COUNT assert all(len(words) == SEGMENT_WORDS for words in segments) assert segments[0] == [1000, 0, 0xFFFF, 0xFFFF, 4, 0, 1, 0] panel.segment_vars[0]["frequency"].set(0) segments = panel._segment_words() assert segments[0][0:2] == [0, 0] panel.common_vars["segment_count"].set(1) panel.segment_vars[0]["jump"].set(2) try: panel._segment_words() except ValueError: pass else: raise AssertionError("active segment jump exceeded segment count") print("PLSR control panel self-test passed; no GUI or serial port was opened") def parse_arguments(arguments=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--self-test", action="store_true", help="validate register encoding without opening the GUI or serial port", ) return parser.parse_args(arguments) def main(arguments=None): args = parse_arguments(arguments) if args.self_test: self_test() return 0 root = tk.Tk() PlsrControlPanel(root) root.mainloop() return 0 if __name__ == "__main__": raise SystemExit(main())