"""Tkinter control panel for the 2026 PLSR Modbus register map.""" from __future__ import annotations import argparse import queue import threading import time 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, ModbusException, RtuClient, SEGMENT_1_BASE, SEGMENT_WORDS, TestFailure, read_status, split_i32, split_u32, ) BAUD_RATE = 9600 SERIAL_TIMEOUT_SECONDS = 0.8 STATUS_POLL_SECONDS = 0.25 SEGMENT_COUNT = 10 SEGMENT_STRIDE = 0x10 MAX_FREQUENCY_HZ = 100_000 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: "内部错误", } 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 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 def submit(self, command, **payload): self.commands.put((command, payload)) 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) 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, 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__() 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": client.write_multiple(CONFIG_BASE, payload["words"]) common = client.read_holding(CONFIG_BASE, COMMON_WORDS) self._emit("common", words=common) 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"]) 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()) self._emit("status", status=status) self.next_status_poll = time.monotonic() + STATUS_POLL_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() + 1.0 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.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.start() self.port_var = tk.StringVar() self.slave_var = tk.StringVar(value="1") 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.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) 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] 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) notebook.add(common_tab, text="公共参数 0x1000-0x1013") notebook.add(segment_tab, text="10 段参数 0x1100-0x1197") self._build_common_tab(common_tab) self._build_segment_tab(segment_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") 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 _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("--") 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("正在打开串口并读取参数") 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 + "频率", 1, 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() except ValueError as error: messagebox.showerror("参数错误", str(error)) return if self._start_operation("正在写入公共参数"): self.worker.submit( "write_common", words=words, 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 _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 _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._set_segment_words(payload["segments"]) self._finish_operation("全部参数已读取") elif event == "common": self._set_common_words(payload["words"]) elif event == "segments": self._set_segment_words(payload["words"]) elif event == "status": self._show_status(payload["status"]) 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) 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), } 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] 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.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())