如何实现Python(Tkinter)与倍福CX9020 PLC的ADS通信自动重连？

核心思路 #

放弃单纯依赖read_state()，改用实际执行轻量ADS操作+异常捕获的方式判断连接状态，结合Tkinter的线程+队列实现后台定时检测，在断开时自动触发重连。read_state()效果差的原因是它在部分断连场景下不会及时抛出错误，而实际读写/设备信息读取能更准确反映真实连接状态。

实现步骤与代码示例 #

1. 封装ADS连接管理类 #

这个类负责统一处理连接、断开、状态检测逻辑，管理ADS客户端实例：

import pyads
from pyads.pyads_ex import ADSError

class ADSConnection:
    def __init__(self, net_id, port):
        self.net_id = net_id
        self.port = port
        self.client = None
        self.connected = False

    def connect(self):
        try:
            if self.client:
                self.disconnect()
            # 创建新的ADS客户端实例
            self.client = pyads.Connection(self.net_id, self.port)
            self.client.open()
            # 读取设备信息验证连接有效性
            self.client.read_device_info()
            self.connected = True
            return True
        except ADSError as e:
            self.connected = False
            print(f"连接失败: {e}")
            return False

    def disconnect(self):
        if self.client and self.connected:
            try:
                self.client.close()
            except ADSError:
                pass
            self.connected = False
            self.client = None

    def check_connection(self):
        if not self.connected:
            return False
        try:
            # 执行轻量操作验证连接
            self.client.read_device_info()
            return True
        except ADSError as e:
            if e.error_code == 1861:  # 捕获超时错误，判定为断连
                self.connected = False
            print(f"连接异常: {e}")
            return False

2. 集成Tkinter GUI与后台检测 #

用线程执行定时检测，通过队列同步状态到GUI，避免阻塞主循环：

import tkinter as tk
from tkinter import ttk
import threading
import queue
import time

class PLC_GUI:
    def __init__(self, root):
        self.root = root
        self.root.title("PLC连接监控")
        
        # 替换为你的PLC实际NetID和端口
        self.plc_conn = ADSConnection("192.168.1.1.1.1", pyads.PORT_TC3PLC1)
        
        # 状态显示组件
        self.status_label = ttk.Label(root, text="连接状态: 未连接", font=("Arial", 12))
        self.status_label.pack(pady=20)
        
        # 线程与GUI通信队列
        self.status_queue = queue.Queue()
        
        # 启动后台检测线程
        self.monitor_thread = threading.Thread(target=self.connection_monitor, daemon=True)
        self.monitor_thread.start()
        
        # 启动GUI状态更新循环
        self.update_gui_status()

    def connection_monitor(self):
        while True:
            time.sleep(1)  # 每秒检测一次
            if not self.plc_conn.check_connection():
                # 尝试重连
                success = self.plc_conn.connect()
                self.status_queue.put("已连接" if success else "重连失败")
            else:
                self.status_queue.put("已连接")

    def update_gui_status(self):
        try:
            status = self.status_queue.get_nowait()
            self.status_label.config(text=f"连接状态: {status}")
            # 根据状态调整标签颜色
            self.status_label.config(foreground="green" if status == "已连接" else "red")
        except queue.Empty:
            pass
        # 每隔100ms更新一次GUI
        self.root.after(100, self.update_gui_status)

if __name__ == "__main__":
    root = tk.Tk()
    app = PLC_GUI(root)
    root.mainloop()

3. 关键细节说明 #

• 轻量检测操作：用read_device_info()替代read_state()，它能快速验证连接，断连时会立刻抛出1861超时错误。也可以换成读取PLC中存在的小型变量（比如read_by_name("Main.bConnected", pyads.PLCTYPE_BOOL)），更贴合业务场景。
• 线程与队列：后台线程负责检测和重连，避免阻塞Tkinter主循环；队列用于安全传递状态，防止线程直接操作GUI组件引发异常。
• 异常精准处理：捕获ADSError并判断错误码1861，精准识别超时断连；重连前先关闭旧连接，避免资源泄漏。

额外注意事项 #

• 确保PLC的ADS端口开放，网络配置（IP、NetID）与代码匹配。
• 若重连频繁失败，可延长重连间隔（比如3秒一次），减少无效重试。
• 实际读写变量时也要包裹异常捕获，避免单次读写失败导致程序崩溃。

内容的提问来源于stack exchange，提问作者עומר דגן