|
|
|
@@ -1,493 +0,0 @@ |
|
|
|
#!/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() |
suyu
4 周前