
import tkinter as tk from tkinter import messagebox from pymodbus.client import ModbusSerialClient # # ️ 自定义配置弹窗类 (保持不变) # class ConfigDialog: def __init__(self, parent): self.result None self.top tk.Toplevel(parent) self.top.title(⚙️ Modbus 连接配置) self.top.resizable(False, False) fields [ (串口号 (如 COM1):, COM2), # 默认设为 COM2 (波特率:, 9600), (设备地址 (Slave ID):, 100), (读取 DI 数量:, 8) ] self.entries [] for i, (label_text, default_val) in enumerate(fields): frame tk.Frame(self.top) frame.pack(pady5, padx10, fillx) label tk.Label(frame, textlabel_text, width20, anchorw) label.pack(sideleft) entry tk.Entry(frame) entry.insert(0, str(default_val)) entry.pack(sideright, expandTrue, fillx) self.entries.append(entry) btn_frame tk.Frame(self.top) btn_frame.pack(pady15) tk.Button(btn_frame, text✅ 确认连接, commandself.confirm).pack(sideleft, padx5) tk.Button(btn_frame, text❌ 取消, commandself.cancel).pack(sideleft, padx5) self.top.wait_window() def confirm(self): try: port self.entries[0].get() baudrate int(self.entries[1].get()) slave_id int(self.entries[2].get()) di_count int(self.entries[3].get()) if not port: raise ValueError(端口号不能为空) self.result { port: port, baudrate: baudrate, slave_id: slave_id, di_count: di_count } self.top.destroy() except ValueError as e: messagebox.showerror(输入错误, f请检查输入格式\n{e}) def cancel(self): self.top.destroy() # # 主程序逻辑 (已修正为 read_discrete_inputs) # if __name__ __main__: root tk.Tk() root.withdraw() # 1. 弹出配置窗口获取参数 dialog ConfigDialog(root) config dialog.result if config is None: print( 用户取消操作程序退出。) else: print( * 40) print(f 配置已加载: {config[port]} | Baud: {config[baudrate]}) print(f 地址: {config[slave_id]} | DI数: {config[di_count]}) print( * 40) # 初始化客户端 client ModbusSerialClient( methodrtu, portconfig[port], baudrateconfig[baudrate], timeout1 ) # 尝试建立物理连接 if not client.connect(): print(f❌ 无法连接到 {config[port]}请检查端口是否被占用) else: print(f✅ 成功连接到 {config[port]}) # 2. 进入循环读取 while True: print(\n正在读取数据...) # 【关键修正】使用 read_discrete_inputs (功能码 02) result client.read_discrete_inputs( address0, countconfig[di_count], slaveconfig[slave_id] ) if not result.isError(): print( 读取结果解析) has_alarm False for i in range(config[di_count]): status result.bits[i] if status: print(f ⚠️ 第 {i 1} 路告警) has_alarm True else: print(f ✅ 第 {i 1} 路正常) if not has_alarm: print( (所有端口均正常)) else: print(f❌ 读取失败: {result}) # 3. 询问用户是否继续 print(- * 40) choice input(是否继续读取(输入 y 继续其他键退出): ).strip().lower() if choice ! y: print( 程序已停止连接已关闭。) break # 确保程序结束时关闭连接 try: client.close() except: pass root.destroy() 禁止商用哈