Modbus-rtu/HostComputer/plsr_control_panel.py

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