suyu
/
suyu_two


			
							#!/usr/bin/env python3
"""XDH-60T4-E Modbus RTU 调试工具。

依赖：pyserial（pip install pyserial）
用途：通过 USB 转 RS-485 连接 PLC COM2，读写项目约定的 M0~M4000、D0~D4000。
"""

from __future__ import annotations

import queue
import struct
import threading
import time
import tkinter as tk
from dataclasses import dataclass
from tkinter import messagebox, ttk
from typing import Any

try:
    import serial
    from serial.tools import list_ports
except ImportError as error:
    raise SystemExit("缺少 pyserial。请执行：python -m pip install pyserial") from error


APP_TITLE = "XDH-60T4-E Modbus RTU 联机工具"
MAX_PROJECT_ADDRESS = 4000


class ModbusError(Exception):
    """Modbus RTU 通信或协议错误。"""


def crc16_modbus(payload: bytes) -> int:
    """计算 Modbus RTU CRC-16，返回未交换字节序的 16 位数。"""
    crc = 0xFFFF
    for byte in payload:
        crc ^= byte
        for _ in range(8):
            crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
    return crc


def append_crc(payload: bytes) -> bytes:
    return payload + struct.pack("<H", crc16_modbus(payload))


def validate_frame(frame: bytes, station: int, function_code: int) -> None:
    if len(frame) < 5:
        raise ModbusError("响应帧长度不足")
    received_crc = struct.unpack("<H", frame[-2:])[0]
    if crc16_modbus(frame[:-2]) != received_crc:
        raise ModbusError("响应 CRC 校验失败")
    if frame[0] != station:
        raise ModbusError(f"响应站号错误：期望 {station}，实际 {frame[0]}")
    if frame[1] == (function_code | 0x80):
        code = frame[2]
        raise ModbusError(f"PLC 返回 Modbus 异常 0x{code:02X}")
    if frame[1] != function_code:
        raise ModbusError(f"响应功能码错误：期望 0x{function_code:02X}，实际 0x{frame[1]:02X}")


class ModbusRtuClient:
    """只提供本工具所需的 Modbus RTU 读写能力。"""

    def __init__(self) -> None:
        self._serial: serial.Serial | None = None
        self.station = 1

    @property
    def is_connected(self) -> bool:
        return self._serial is not None and self._serial.is_open

    def connect(self, config: dict[str, Any]) -> None:
        self.close()
        self.station = int(config["station"])
        self._serial = serial.Serial(
            port=config["port"],
            baudrate=int(config["baudrate"]),
            bytesize=int(config["bytesize"]),
            parity=config["parity"],
            stopbits=float(config["stopbits"]),
            timeout=float(config["timeout"]),
            write_timeout=float(config["timeout"]),
        )
        self._serial.reset_input_buffer()
        self._serial.reset_output_buffer()

    def close(self) -> None:
        if self._serial is not None:
            self._serial.close()
            self._serial = None

    def _read_exact(self, size: int) -> bytes:
        if self._serial is None:
            raise ModbusError("串口尚未连接")
        data = self._serial.read(size)
        if len(data) != size:
            raise ModbusError(f"通信超时：期望接收 {size} 字节，实际收到 {len(data)} 字节")
        return data

    def _transact(self, function_code: int, request_data: bytes, response_kind: str) -> bytes:
        if self._serial is None or not self._serial.is_open:
            raise ModbusError("串口尚未连接")

        request = append_crc(bytes((self.station, function_code)) + request_data)
        self._serial.reset_input_buffer()
        self._serial.write(request)
        self._serial.flush()

        header = self._read_exact(2)
        if header[1] == (function_code | 0x80):
            frame = header + self._read_exact(3)
        elif response_kind == "read":
            byte_count = self._read_exact(1)
            frame = header + byte_count + self._read_exact(byte_count[0] + 2)
        else:
            frame = header + self._read_exact(6)
        validate_frame(frame, self.station, function_code)
        return frame

    def read_coils(self, address: int, count: int) -> list[bool]:
        self._validate_range(address, count, maximum=2000)
        frame = self._transact(0x01, struct.pack(">HH", address, count), "read")
        byte_count = frame[2]
        expected_bytes = (count + 7) // 8
        if byte_count != expected_bytes:
            raise ModbusError(f"线圈响应字节数错误：期望 {expected_bytes}，实际 {byte_count}")
        values: list[bool] = []
        for index in range(count):
            values.append(bool(frame[3 + index // 8] & (1 << (index % 8))))
        return values

    def read_holding_registers(self, address: int, count: int) -> list[int]:
        self._validate_range(address, count, maximum=125)
        frame = self._transact(0x03, struct.pack(">HH", address, count), "read")
        byte_count = frame[2]
        if byte_count != count * 2:
            raise ModbusError(f"寄存器响应字节数错误：期望 {count * 2}，实际 {byte_count}")
        return [struct.unpack(">h", frame[3 + offset : 5 + offset])[0] for offset in range(0, byte_count, 2)]

    def write_coil(self, address: int, value: bool) -> None:
        self._validate_range(address, 1, maximum=1)
        raw_value = 0xFF00 if value else 0x0000
        request_data = struct.pack(">HH", address, raw_value)
        frame = self._transact(0x05, request_data, "write")
        if frame[2:6] != request_data:
            raise ModbusError("写线圈响应内容与请求不一致")

    def write_register(self, address: int, value: int) -> None:
        self._validate_range(address, 1, maximum=1)
        if not -32768 <= value <= 32767:
            raise ModbusError("D 寄存器值必须在 -32768 到 32767 之间")
        request_data = struct.pack(">Hh", address, value)
        frame = self._transact(0x06, request_data, "write")
        if frame[2:6] != request_data:
            raise ModbusError("写寄存器响应内容与请求不一致")

    @staticmethod
    def _validate_range(address: int, count: int, maximum: int) -> None:
        if not 0 <= address <= MAX_PROJECT_ADDRESS:
            raise ModbusError(f"地址必须在 0 到 {MAX_PROJECT_ADDRESS} 之间")
        if not 1 <= count <= maximum:
            raise ModbusError(f"读取数量必须在 1 到 {maximum} 之间")
        if address + count - 1 > MAX_PROJECT_ADDRESS:
            raise ModbusError(f"地址范围不能超过 {MAX_PROJECT_ADDRESS}")


@dataclass(frozen=True)
class Command:
    request_id: int
    action: str
    payload: dict[str, Any]


class CommunicationWorker:
    """独占串口的后台线程，保证 GUI 不因通信超时而冻结。"""

    def __init__(self) -> None:
        self.commands: queue.Queue[Command] = queue.Queue()
        self.results: queue.Queue[tuple[int, bool, str, Any]] = queue.Queue()
        self._stop_event = threading.Event()
        self._thread = threading.Thread(target=self._run, name="modbus-rtu-worker", daemon=True)
        self._thread.start()

    def submit(self, command: Command) -> None:
        self.commands.put(command)

    def shutdown(self) -> None:
        self._stop_event.set()
        self.commands.put(Command(0, "shutdown", {}))
        self._thread.join(timeout=1.5)

    def _run(self) -> None:
        client = ModbusRtuClient()
        while not self._stop_event.is_set():
            try:
                command = self.commands.get(timeout=0.2)
            except queue.Empty:
                continue
            try:
                if command.action == "shutdown":
                    client.close()
                    return
                if command.action == "connect":
                    client.connect(command.payload)
                    result = f"已连接 {command.payload['port']}，站号 {client.station}"
                elif command.action == "disconnect":
                    client.close()
                    result = "串口已断开"
                elif command.action == "read_m":
                    result = client.read_coils(command.payload["address"], command.payload["count"])
                elif command.action == "read_d":
                    result = client.read_holding_registers(command.payload["address"], command.payload["count"])
                elif command.action == "write_m":
                    client.write_coil(command.payload["address"], command.payload["value"])
                    result = None
                elif command.action == "write_d":
                    client.write_register(command.payload["address"], command.payload["value"])
                    result = None
                else:
                    raise ModbusError(f"未知操作：{command.action}")
                self.results.put((command.request_id, True, command.action, result))
            except (ModbusError, serial.SerialException, ValueError, struct.error) as error:
                self.results.put((command.request_id, False, command.action, str(error)))


class PlcToolApp(tk.Tk):
    def __init__(self) -> None:
        super().__init__()
        self.title(APP_TITLE)
        self.minsize(840, 600)
        self.geometry("920x680")
        self.worker = CommunicationWorker()
        self._next_request_id = 1
        self._request_labels: dict[int, str] = {}

        self.port_var = tk.StringVar()
        self.baudrate_var = tk.StringVar(value="9600")
        self.bytesize_var = tk.StringVar(value="8")
        self.parity_var = tk.StringVar(value="N")
        self.stopbits_var = tk.StringVar(value="1")
        self.station_var = tk.StringVar(value="1")
        self.timeout_var = tk.StringVar(value="1.0")
        self.status_var = tk.StringVar(value="未连接")

        self.read_m_address_var = tk.StringVar(value="0")
        self.read_m_count_var = tk.StringVar(value="1")
        self.read_d_address_var = tk.StringVar(value="0")
        self.read_d_count_var = tk.StringVar(value="1")
        self.write_m_address_var = tk.StringVar(value="0")
        self.write_m_value_var = tk.BooleanVar(value=False)
        self.write_d_address_var = tk.StringVar(value="0")
        self.write_d_value_var = tk.StringVar(value="0")

        self._build_ui()
        self.refresh_ports()
        self.protocol("WM_DELETE_WINDOW", self.on_close)
        self.after(80, self.process_results)

    def _build_ui(self) -> None:
        root = ttk.Frame(self, padding=12)
        root.grid(sticky="nsew")
        self.columnconfigure(0, weight=1)
        self.rowconfigure(0, weight=1)
        root.columnconfigure(0, weight=1)
        root.rowconfigure(3, weight=1)

        connection = ttk.LabelFrame(root, text="连接参数", padding=10)
        connection.grid(row=0, column=0, sticky="ew")
        for column in range(8):
            connection.columnconfigure(column, weight=1 if column in (1, 3, 5) else 0)

        self._add_label_entry(connection, "串口", self.port_var, 0, 0, width=14, readonly=True)
        self.port_combo = connection.grid_slaves(row=0, column=1)[0]
        ttk.Button(connection, text="刷新", command=self.refresh_ports).grid(row=0, column=2, padx=(6, 12))
        self._add_label_entry(connection, "波特率", self.baudrate_var, 0, 3, width=10, values=("9600", "19200", "38400", "57600", "115200"))
        self._add_label_entry(connection, "站号", self.station_var, 0, 5, width=6)
        ttk.Button(connection, text="连接", command=self.connect).grid(row=0, column=7, padx=(12, 4))

        self._add_label_entry(connection, "数据位", self.bytesize_var, 1, 0, width=8, values=("7", "8"))
        self._add_label_entry(connection, "校验", self.parity_var, 1, 2, width=8, values=("N", "E", "O"))
        self._add_label_entry(connection, "停止位", self.stopbits_var, 1, 4, width=8, values=("1", "1.5", "2"))
        self._add_label_entry(connection, "超时(秒)", self.timeout_var, 1, 6, width=8)
        ttk.Button(connection, text="断开", command=lambda: self.submit("disconnect", {}, "断开")).grid(row=1, column=7, padx=(12, 4))

        operations = ttk.Frame(root)
        operations.grid(row=1, column=0, sticky="ew", pady=(12, 0))
        operations.columnconfigure(0, weight=1)
        operations.columnconfigure(1, weight=1)

        self._build_read_panel(operations)
        self._build_write_panel(operations)

        result_frame = ttk.LabelFrame(root, text="读取结果", padding=8)
        result_frame.grid(row=2, column=0, sticky="nsew", pady=(12, 0))
        result_frame.columnconfigure(0, weight=1)
        self.result_text = tk.Text(result_frame, height=8, wrap="word", state="disabled", font=("Consolas", 10))
        self.result_text.grid(row=0, column=0, sticky="nsew")

        log_frame = ttk.LabelFrame(root, text="通信日志", padding=8)
        log_frame.grid(row=3, column=0, sticky="nsew", pady=(12, 0))
        log_frame.columnconfigure(0, weight=1)
        log_frame.rowconfigure(0, weight=1)
        self.log_text = tk.Text(log_frame, height=10, wrap="word", state="disabled", font=("Consolas", 10))
        self.log_text.grid(row=0, column=0, sticky="nsew")
        ttk.Button(log_frame, text="清空日志", command=lambda: self._set_text(self.log_text, "")).grid(row=1, column=0, sticky="e", pady=(6, 0))

        status = ttk.Label(root, textvariable=self.status_var, relief="sunken", anchor="w", padding=(7, 3))
        status.grid(row=4, column=0, sticky="ew", pady=(10, 0))

    @staticmethod
    def _add_label_entry(
        parent: ttk.Widget,
        label: str,
        variable: tk.StringVar,
        row: int,
        column: int,
        width: int,
        values: tuple[str, ...] | None = None,
        readonly: bool = False,
    ) -> None:
        ttk.Label(parent, text=label).grid(row=row, column=column, sticky="w", padx=(0 if column == 0 else 10, 4), pady=3)
        if values is not None or readonly:
            state = "readonly" if readonly else "normal"
            widget = ttk.Combobox(parent, textvariable=variable, values=values, width=width, state=state)
        else:
            widget = ttk.Entry(parent, textvariable=variable, width=width)
        widget.grid(row=row, column=column + 1, sticky="ew", pady=3)

    def _build_read_panel(self, parent: ttk.Frame) -> None:
        panel = ttk.LabelFrame(parent, text="读取 PLC", padding=10)
        panel.grid(row=0, column=0, sticky="nsew", padx=(0, 6))
        ttk.Label(panel, text="M 地址").grid(row=0, column=0, sticky="w")
        ttk.Entry(panel, textvariable=self.read_m_address_var, width=9).grid(row=0, column=1, padx=5)
        ttk.Label(panel, text="数量").grid(row=0, column=2, sticky="w")
        ttk.Entry(panel, textvariable=self.read_m_count_var, width=7).grid(row=0, column=3, padx=5)
        ttk.Button(panel, text="读取 M", command=self.read_m).grid(row=0, column=4, padx=(8, 0))

        ttk.Label(panel, text="D 地址").grid(row=1, column=0, sticky="w", pady=(8, 0))
        ttk.Entry(panel, textvariable=self.read_d_address_var, width=9).grid(row=1, column=1, padx=5, pady=(8, 0))
        ttk.Label(panel, text="数量").grid(row=1, column=2, sticky="w", pady=(8, 0))
        ttk.Entry(panel, textvariable=self.read_d_count_var, width=7).grid(row=1, column=3, padx=5, pady=(8, 0))
        ttk.Button(panel, text="读取 D", command=self.read_d).grid(row=1, column=4, padx=(8, 0), pady=(8, 0))

    def _build_write_panel(self, parent: ttk.Frame) -> None:
        panel = ttk.LabelFrame(parent, text="写入 PLC（请先确认测试程序安全）", padding=10)
        panel.grid(row=0, column=1, sticky="nsew", padx=(6, 0))
        ttk.Label(panel, text="M 地址").grid(row=0, column=0, sticky="w")
        ttk.Entry(panel, textvariable=self.write_m_address_var, width=9).grid(row=0, column=1, padx=5)
        ttk.Checkbutton(panel, text="写入 ON", variable=self.write_m_value_var).grid(row=0, column=2, padx=5)
        ttk.Button(panel, text="写 M", command=self.write_m).grid(row=0, column=3, padx=(8, 0))

        ttk.Label(panel, text="D 地址").grid(row=1, column=0, sticky="w", pady=(8, 0))
        ttk.Entry(panel, textvariable=self.write_d_address_var, width=9).grid(row=1, column=1, padx=5, pady=(8, 0))
        ttk.Entry(panel, textvariable=self.write_d_value_var, width=10).grid(row=1, column=2, padx=5, pady=(8, 0))
        ttk.Button(panel, text="写 D", command=self.write_d).grid(row=1, column=3, padx=(8, 0), pady=(8, 0))

    def refresh_ports(self) -> None:
        ports = [port.device for port in list_ports.comports()]
        self.port_combo["values"] = ports
        if ports and self.port_var.get() not in ports:
            self.port_var.set(ports[0])
        self.log(f"检测到串口：{', '.join(ports) if ports else '无'}")

    def connect(self) -> None:
        try:
            config = {
                "port": self.port_var.get().strip(),
                "baudrate": int(self.baudrate_var.get()),
                "bytesize": int(self.bytesize_var.get()),
                "parity": self.parity_var.get().strip().upper(),
                "stopbits": float(self.stopbits_var.get()),
                "station": int(self.station_var.get()),
                "timeout": float(self.timeout_var.get()),
            }
            if not config["port"]:
                raise ValueError("请选择串口")
            if not 1 <= config["station"] <= 247:
                raise ValueError("站号必须在 1 到 247 之间")
            if config["timeout"] <= 0:
                raise ValueError("超时必须大于 0")
            if config["parity"] not in ("N", "E", "O"):
                raise ValueError("校验仅支持 N、E 或 O")
        except ValueError as error:
            self.show_input_error(str(error))
            return
        self.submit("connect", config, "连接")

    def read_m(self) -> None:
        payload = self.parse_address_count(self.read_m_address_var, self.read_m_count_var)
        if payload is not None:
            self.submit("read_m", payload, f"读取 M{payload['address']} 起 {payload['count']} 个")

    def read_d(self) -> None:
        payload = self.parse_address_count(self.read_d_address_var, self.read_d_count_var)
        if payload is not None:
            self.submit("read_d", payload, f"读取 D{payload['address']} 起 {payload['count']} 个")

    def write_m(self) -> None:
        try:
            address = int(self.write_m_address_var.get())
        except ValueError:
            self.show_input_error("M 地址必须是整数")
            return
        self.submit("write_m", {"address": address, "value": self.write_m_value_var.get()}, f"写入 M{address}")

    def write_d(self) -> None:
        try:
            address = int(self.write_d_address_var.get())
            value = int(self.write_d_value_var.get())
        except ValueError:
            self.show_input_error("D 地址和值必须是整数")
            return
        self.submit("write_d", {"address": address, "value": value}, f"写入 D{address}={value}")

    def parse_address_count(self, address_var: tk.StringVar, count_var: tk.StringVar) -> dict[str, int] | None:
        try:
            return {"address": int(address_var.get()), "count": int(count_var.get())}
        except ValueError:
            self.show_input_error("地址和数量必须是整数")
            return None

    def submit(self, action: str, payload: dict[str, Any], label: str) -> None:
        request_id = self._next_request_id
        self._next_request_id += 1
        self._request_labels[request_id] = label
        self.worker.submit(Command(request_id, action, payload))
        self.status_var.set(f"处理中：{label}")
        self.log(f"发送请求：{label}")

    def process_results(self) -> None:
        while True:
            try:
                request_id, success, action, result = self.worker.results.get_nowait()
            except queue.Empty:
                break
            label = self._request_labels.pop(request_id, action)
            if success:
                self.status_var.set(f"成功：{label}")
                self.log(f"成功：{label}")
                if action == "connect":
                    self.status_var.set(result)
                    self.log(result)
                elif action == "read_m":
                    self.show_read_result("M", self.read_m_address_var.get(), result)
                elif action == "read_d":
                    self.show_read_result("D", self.read_d_address_var.get(), result)
                elif action == "write_m":
                    self.log("写入完成；请使用“读取 M”确认 PLC 实际状态。")
                elif action == "write_d":
                    self.log("写入完成；请使用“读取 D”确认 PLC 实际值。")
            else:
                self.status_var.set(f"失败：{label}")
                self.log(f"失败：{label}。原因：{result}")
        self.after(80, self.process_results)

    def show_read_result(self, area: str, address_text: str, values: list[bool] | list[int]) -> None:
        start_address = int(address_text)
        lines = [f"{area} 区读取结果，共 {len(values)} 项："]
        for index, value in enumerate(values):
            if area == "M":
                rendered = "ON" if value else "OFF"
            else:
                rendered = str(value)
            lines.append(f"{area}{start_address + index} = {rendered}")
        self._set_text(self.result_text, "\n".join(lines))

    def log(self, message: str) -> None:
        timestamp = time.strftime("%H:%M:%S")
        self.log_text.configure(state="normal")
        self.log_text.insert("end", f"[{timestamp}] {message}\n")
        self.log_text.see("end")
        self.log_text.configure(state="disabled")

    @staticmethod
    def _set_text(widget: tk.Text, value: str) -> None:
        widget.configure(state="normal")
        widget.delete("1.0", "end")
        widget.insert("1.0", value)
        widget.configure(state="disabled")

    def show_input_error(self, message: str) -> None:
        self.status_var.set(f"输入错误：{message}")
        messagebox.showerror(APP_TITLE, message, parent=self)

    def on_close(self) -> None:
        self.worker.shutdown()
        self.destroy()


if __name__ == "__main__":
    PlcToolApp().mainloop()