diff --git a/docs/XDH-60T4-E硬件与接线要点.md b/docs/XDH-60T4-E硬件与接线要点.md index 859b4ef..1e4b515 100644 --- a/docs/XDH-60T4-E硬件与接线要点.md +++ b/docs/XDH-60T4-E硬件与接线要点.md @@ -77,7 +77,22 @@ PLC 必须先使用信捷官方编程软件或 XINJEConfig 配置工具，将 CO 5. 先读取 `M0` 和 `D0` 验证通信，再接入完整 HMI 页面。 6. 真机模式下关闭 PC 端逻辑执行器，避免双重写入寄存器。 -## 6. 资料出处 +## 6. 已验证联机记录 + +### 6.1 通信安全冒烟 + +- 配置：`COM3 / 9600 / 8E1 / 站号 1`。 +- 读取 `M0`、`D0` 和 `D4000`，确认真实 PLC 能够正常响应。 +- 临时写入 `D4000=1`，读回确认后恢复为 `D4000=0`，再次读回确认恢复成功。 +- 本次安全冒烟没有下载、修改或生成 PLC 程序。 + +### 6.2 RUN 联动演示 + +2026-08-21，用户使用 `json/motor_forward_reverse.json` 在已连接的真实 PLC 上完成真机 RUN 联动手动验收。运行监控 HMI 的正转启动、反转启动和停止操作均可正常联动，正转运行和反转运行状态反馈正常。 + +该项结果依赖现场 PLC 内部程序、RS-485 接线和实际负载，自动化测试只验证 PC 软件的通信契约、缓存和运行模式隔离，不能替代现场 RUN 验收。真机模式下 PC 不执行本地梯形图，PLC 内部程序是唯一控制逻辑来源。 + +## 7. 资料出处 1. 《XD、XL系列可编程序控制器用户手册【硬件篇】》，PD 01 20260410 1.6，重点参见印刷页 28。 2. 《XD、XL系列可编程控制器用户手册【基本指令篇】》，PD05 20260610 1.6.1，重点参见印刷页 273-276。 diff --git a/docs/ai/handoff.md b/docs/ai/handoff.md index e6198eb..b652f1b 100644 --- a/docs/ai/handoff.md +++ b/docs/ai/handoff.md @@ -1,6 +1,6 @@ # 当前开发交接 -> 更新日期：2026-08-20。本文件只保留当前状态和下一步；完成事项合并成结论，不追加按日期排列的开发流水账。 +> 更新日期：2026-08-21。本文件只保留当前状态和下一步；完成事项合并成结论，不追加按日期排列的开发流水账。 ## 当前状态 @@ -8,7 +8,7 @@ - 核心软件闭环已经可用：HMI 编辑、结构化梯形图、离线仿真、工程 JSON、统一运行监控大屏、自由监控和 Modbus RTU 真机读写均已实现 - HMI 外观属性和梯形图十列可选插入网格均已按逻辑完成提交；开始工作前仍需先看 `git status` 和相关 diff，不要重置来源不明的改动 - `docs/二次开发/` 是已有未跟踪资料，不能因当前任务删除 -- 当前没有代码阻塞；下一里程碑是实际设备 RUN 联动和交付整理 +- 当前没有代码阻塞；实际设备 RUN 联动和离线演示均已完成，下一步是最终交付资料整理 ## 已实现能力 @@ -49,6 +49,8 @@ - 独立运行监控窗口改动已通过重新生成 qmake 构建后的 `main_window_tests`、`hmi_editor_service_tests`、`offline_simulation_service_tests`、`runtime_mode_service_tests` 和 `plc_runtime_tests`；主窗口测试覆盖最大化弹窗、唯一监控实例、主窗口布局不变、离线与模拟真机 HMI 写入、页面跳转、返回编辑、关闭弹窗、通信故障隐藏、重复进入刷新和应用退出 - 独立运行监控窗口改动后的 Qt Release 主程序已在 `build/runtime-monitor-window-release` 完整构建并链接成功 - 2026-08-20 使用 `COM3 / 9600 / 8E1 / 站号 1` 完成真实 PLC 安全读写：读取 `D4000=0`，临时写入并读回 `1`，随后恢复并读回 `0`；没有下载 PLC 程序，也没有执行切换 RUN 的动作 +- 2026-08-21 用户使用 `json/motor_forward_reverse.json` 在已连接的真实 PLC 上完成 RUN 联动手动验收：通过运行监控 HMI 操作正转启动、反转启动和停止，并确认对应运行状态反馈正常；该项验证覆盖实际 PLC 内部程序、接线和现场负载，不由自动化替身测试替代 +- 2026-08-21 在 Qt/MinGW 运行库 PATH 正确的独立环境中重新执行 10 个 Release 测试目标，全部通过：`alarm_service_tests`、`domain_tests`、`hmi_editor_service_tests`、`logic_editor_service_tests`、`main_window_tests`、`offline_simulation_service_tests`、`plc_runtime_tests`、`project_management_tests`、`register_monitor_service_tests`、`runtime_mode_service_tests` - CTU/CTD 显示修复已通过 `main_window_tests`（`QT_QPA_PLATFORM=offscreen`）和 Qt Release 主程序重新构建；改动仅涉及梯形图 UI 投影，因此未连接真实 PLC - 报警列表和分页改动已通过 `domain_tests`、`alarm_service_tests`、`hmi_editor_service_tests`、`project_management_tests` 和 `main_window_tests`；覆盖输入长度、默认五行尺寸、默认字号、运行态高度收缩、六条报警翻页和跨页确认 - 报警列表改动后的 Qt Release 主程序已重新构建并链接成功；改动不涉及 PLC 通信、Modbus 地址或运行模式，因此未连接真实 PLC @@ -70,8 +72,8 @@ ## 待完成 -1. 在确认接线、地址和负载安全后，将 PLC 切到 RUN，完成 HMI 与实际设备的完整联动测试 -2. 整理最终演示工程、构建说明和验收记录 +1. 整理最终演示工程、构建说明和验收记录 +2. 根据交付时间评估插件化、新设备适配等可选加分项，不影响当前核心验收闭环 ## 下一会话起点 diff --git a/docs/二次开发/信捷D寄存器与浮点扩展说明.md b/docs/二次开发/信捷D寄存器与浮点扩展说明.md new file mode 100644 index 0000000..80747d6 --- /dev/null +++ b/docs/二次开发/信捷D寄存器与浮点扩展说明.md @@ -0,0 +1,63 @@ +# 信捷 D 寄存器与浮点扩展说明 + +> 这份文档给后续二次开发使用，说明 D 区的基本规则和接入浮点时要注意的地方。当前版本仍按 16 位 D 字寄存器实现。 + +## 1. D 寄存器到底是什么 + +信捷的一个 D 寄存器本质上是 **16 位原始数据**，不是固定的“整数变量”或“浮点变量”。 + +- 按整数解释时，单个 D 的范围是 `-32768~32767` +- PLC 程序、HMI 或上位机决定这些位按什么类型解释 +- 当前软件的 `RegisterRepository` 按单个有符号 16 位字读写 D + +所以，浮点数可以放进 D 区，但不能只占一个 D。 + +## 2. 连续 D 的占用规则 + +| 数据类型 | 占用寄存器 | 规则 | +| --- | --- | --- | +| 16 位整数 | 1 个 | `D10` | +| 32 位整数或单精度浮点 `REAL` | 2 个 | `D10` 低字，`D11` 高字 | +| 64 位双精度浮点 `LREAL` | 4 个 | `D10` 最低字，依次到 `D13` | + +信捷手册把 32 位数据写成 `(D1,D0)`，其中 `D0` 是低字、`D1` 是高字。指定 `D10` 作为起始地址时，后续寄存器会占用 `D11`、`D12` 等地址。 + +以后增加类型支持时，必须检查连续地址是否越界或与其他变量重叠。例如 `Float32 D10` 已经占用 `D10~D11`，`D11` 不能再单独分配。 + +## 3. Modbus 读写要点 + +Modbus 把每个 D 当作一个 16 位保持寄存器： + +- 16 位值读写 1 个保持寄存器 +- `Float32` 读写连续 2 个保持寄存器 +- `Float64` 读写连续 4 个保持寄存器 +- 低地址 D 保存低字，高地址 D 保存高字 + +解析时要分清两件事：先按 Modbus 规则解析每个 16 位寄存器的字节，再按信捷规则组合 D 的低字和高字。不要只读一个 D，也不要把“字节序”和“字序”混为一谈。 + +例如 `Float32 D10` 的写入，应该一次写 `D10` 和 `D11`，不能用只写单个寄存器的请求完成。 + +## 4. 当前软件边界 + +当前版本只支持： + +- HMI 数值控件绑定单个 D 字 +- 自由监控按有符号 16 位显示和写入 +- 梯形图的比较、MOVE、ADD/SUB 按 16 位 D 值运行 +- PLC 轮询按单个 D 地址收集和合并 + +不要为了接入浮点，直接把底层寄存器仓库改成 `float`。底层应该继续保存原始 16 位字，类型转换放在上层的类型解释或编解码模块中。 + +## 5. 后续接入建议 + +如果以后确实需要温度、压力、流量等带小数的数据，建议按下面顺序做： + +1. 先支持 `Float32/REAL`，暂时不做 `Float64/LREAL` +2. 给绑定增加数据类型和占用字数，同时保留起始 D 地址 +3. 增加连续地址占用和重叠校验 +4. 让 HMI、自由监控、报警和离线仿真都使用同一个浮点编解码规则 +5. Modbus 一次读取或写入完整的两个 D，并补充字序、边界和断线测试 + +只有当浮点数据真的需要参与梯形图比较、报警或运算时，才继续扩展浮点逻辑指令。不要只做一个浮点显示控件，却让其他模块仍按 16 位整数处理同一组 D。 + +参考手册：《XD、XL系列可编程控制器用户手册（基本指令篇）》印刷页 44、122、157~158、483。 diff --git a/docs/开发顺序.md b/docs/开发顺序.md index 8cf4787..f6af860 100644 --- a/docs/开发顺序.md +++ b/docs/开发顺序.md @@ -4,7 +4,7 @@ ## 当前进度 -第 1~10 步的软件实现已完成，工程格式继续使用 `1.0`，不保留旧梯形图结构的兼容代码。当前重点是第 11 步中的真机安全验证和第 12 步交付资料。 +第 1~11 步的软件实现和现场验证已完成，工程格式继续使用 `1.0`，不保留旧梯形图结构的兼容代码。当前重点是第 12 步的交付资料和最终演示整理。 ## 1. 建立开发基线 @@ -109,7 +109,7 @@ 完成标准：核心流程可重复执行，异常有反馈，程序不会崩溃或误写 PLC。 -当前待办：自动化回归已覆盖核心领域、编辑、存储、仿真、运行模式、PLC 缓存和主界面；真实串口、接线、STOP 安全写入和 RUN 联动仍需连接 XDH-60T4-E 验证。 +完成情况：自动化回归已覆盖核心领域、编辑、存储、仿真、运行模式、PLC 缓存和主界面；真实串口和 STOP 安全读写已经完成，2026-08-21 由用户在已连接的 XDH-60T4-E 上完成 RUN 联动手动验收。自动化测试不能替代这项现场验证，因为它还依赖 PLC 内部程序、接线和实际负载。 ## 12. 完成交付与演示 @@ -117,6 +117,8 @@ - 准备工程创建、HMI 配置、逻辑配置、离线仿真、真机联机和状态反馈的演示流程。 - 最后再评估新增控件、设备适配或插件机制等可选加分项。 +完成情况：工程创建/保存/加载、HMI 配置、逻辑配置、离线仿真、编辑态与运行态切换、PLC 真机联机、RUN 联动和状态反馈均已有演示路径；剩余工作是整理最终交付资料和演示顺序。 + 完成标准：在干净环境中可以按文档构建，并完整演示验收流程。 ## 建议进度 diff --git a/docs/用户使用/数量边界确认方案.md b/docs/用户使用/数量边界确认方案.md index bcf17ed..609c363 100644 --- a/docs/用户使用/数量边界确认方案.md +++ b/docs/用户使用/数量边界确认方案.md @@ -97,3 +97,5 @@ - Release 主程序已经重新构建通过，数量边界相关的 10 组自动化测试全部通过 - 2026-08-19 使用真实 PLC 在 `COM3、9600、8E1、站号 1` 下完成读写验证：先读取 `D4000=0`，写入 `1` 后读回 `1`，再恢复并读回 `0` - 真机验证没有下载 PLC 程序，测试结束后 `D4000` 已恢复原值 +- 2026-08-21 用户使用 `json/motor_forward_reverse.json` 完成真实 PLC RUN 联动手动验收，确认 HMI 正转启动、反转启动、停止和运行状态反馈正常 +- 2026-08-21 重新执行 10 个 Release 自动化测试目标，全部通过；自动化测试覆盖软件逻辑、运行模式、PLC 缓存和 UI 工作流，现场 RUN 联动另由真实设备手动验收 diff --git a/scripts/plc_modbus_rtu_tool.py b/scripts/plc_modbus_rtu_tool.py deleted file mode 100644 index 90de91d..0000000 --- a/scripts/plc_modbus_rtu_tool.py +++ /dev/null @@ -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(" None: - if len(frame) < 5: - raise ModbusError("响应帧长度不足") - received_crc = struct.unpack(" 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()