4、Python与SCADA通信：使用pyModbus库读取PLC数据、使用opcua-asyncio库连接OPC UA服务器、异常处理与重连机制

好，咱们进入正题。这一章聊的是SCADA系统里最核心的一环——通信。

说白了，你界面做得再漂亮，报表画得再花哨，如果底层的PLC数据读不上来，那一切都是白搭。我这些年做项目，见过太多“界面跑得飞起，数据全是NaN”的尴尬场面。嗯，这里咱们就把通信这块彻底讲透。

4.1 为什么是pyModbus和opcua-asyncio？

先说说我的选择逻辑。

工业现场最常见的两种协议，一个是Modbus，一个是OPC UA。Modbus老当益壮，新设备老设备都在用；OPC UA则是未来的趋势，安全性和跨平台能力都强。

Python这边，pyModbus是我个人用得最顺手的Modbus库。它支持RTU和TCP两种模式，API设计得比较清爽。opcua-asyncio则是异步版的OPC UA客户端库，性能好，适合咱们做实时数据采集。

你想想看，一个SCADA系统可能要同时采集几十上百个设备的数据。如果每个连接都阻塞着等回复，那系统早就卡死了。所以异步IO是必须的。

4.2 使用pyModbus读取PLC数据

先来个最简单的例子。假设你有一台西门子S7-1200，开启了Modbus TCP服务。咱们用Python把它里面的保持寄存器读出来。

from pymodbus.client import ModbusTcpClient

# 连接PLC
client = ModbusTcpClient('192.168.1.100', port=502)
client.connect()

# 读取保持寄存器，从地址0开始，读10个
result = client.read_holding_registers(0, 10, unit=1)

if not result.isError():
    print(f"读取成功: {result.registers}")
else:
    print(f"读取失败: {result}")

client.close()

这段代码看着简单，但我得提醒你几个坑。

注意：Modbus的地址偏移问题。很多PLC的寄存器地址是从1开始的，但pyModbus里是从0开始。比如PLC里地址是40001，你代码里就要写0。我曾经因为这个搞错过一整个批次的数据，排查了整整一个下午。

另外，unit参数是设备ID，默认是1。如果你的总线上挂了多个设备，记得改这个值。

4.3 批量读取与性能优化

实际项目中，你不会只读10个寄存器。一个设备可能有几百个数据点。我建议你这样做：

def read_plc_batch(client, start_addr, count, unit=1):
    """
    批量读取PLC数据，自动处理最大读取长度限制
    """
    max_read = 125  # Modbus协议限制一次最多读125个寄存器
    data = []
    
    for offset in range(0, count, max_read):
        chunk = min(max_read, count - offset)
        result = client.read_holding_registers(start_addr + offset, chunk, unit=unit)
        
        if result.isError():
            raise Exception(f"读取失败，地址: {start_addr + offset}")
        
        data.extend(result.registers)
    
    return data

为什么要分块？因为Modbus协议规定了一次请求最多读125个寄存器。你一次读200个，PLC会直接报错。这个限制我在第一次做项目时就踩过，当时还以为是PLC坏了。

4.4 使用opcua-asyncio连接OPC UA服务器

OPC UA比Modbus复杂一些，但功能也更强大。它支持数据订阅、历史数据查询、安全认证等。

咱们用opcua-asyncio来连接一个OPC UA服务器：

import asyncio
from opcua import Client

async def read_opcua_data():
    client = Client("opc.tcp://192.168.1.200:4840")
    
    try:
        await client.connect()
        print("连接OPC UA服务器成功")
        
        # 读取一个节点
        node = client.get_node("ns=2;s=Device1.Temperature")
        value = await node.read_value()
        print(f"温度值: {value}")
        
        # 读取多个节点
        nodes = [
            client.get_node("ns=2;s=Device1.Temperature"),
            client.get_node("ns=2;s=Device1.Pressure"),
            client.get_node("ns=2;s=Device1.Flow")
        ]
        values = await asyncio.gather(*[node.read_value() for node in nodes])
        print(f"批量读取: {values}")
        
    finally:
        await client.disconnect()

asyncio.run(read_opcua_data())

这里有个细节：asyncio.gather可以并发读取多个节点，效率比一个个读高很多。我在一个水处理项目中，用这个方法把300个节点的读取时间从15秒降到了3秒。

小技巧：OPC UA的节点ID格式通常是"ns=命名空间索引;s=节点名称"。你可以用UA Expert这个工具先浏览一下服务器上的节点树，找到你要读的节点ID。

4.5 异常处理——别让程序崩在半夜

做SCADA系统，最怕的就是半夜三点接到电话：“系统挂了！”

工业现场的网络环境很复杂。PLC可能重启、网线可能松动、OPC UA服务器可能维护。你的代码必须能优雅地处理这些异常。

我总结了一套异常处理模板：

import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class PLCReader:
    def __init__(self, host, port=502):
        self.host = host
        self.port = port
        self.client = None
        self.connected = False
    
    def connect(self):
        try:
            self.client = ModbusTcpClient(self.host, port=self.port)
            self.connected = self.client.connect()
            if self.connected:
                logger.info(f"连接成功: {self.host}:{self.port}")
            return self.connected
        except Exception as e:
            logger.error(f"连接失败: {e}")
            self.connected = False
            return False
    
    def read_data(self, address, count):
        if not self.connected:
            raise ConnectionError("未连接到PLC")
        
        try:
            result = self.client.read_holding_registers(address, count)
            if result.isError():
                raise Exception(f"读取错误: {result}")
            return result.registers
        except Exception as e:
            logger.error(f"读取数据失败: {e}")
            raise

你看，我把连接和读取分开了。这样在重连时，只需要重新调用connect()方法就行。

4.6 重连机制——让系统自己恢复

重连机制是SCADA系统的保命符。我的做法是：

  1. 检测到连接断开
  2. 等待一段时间（指数退避）
  3. 尝试重连
  4. 如果重连成功，恢复数据采集
  5. 如果重连失败，继续等待并重试

代码实现如下：

import asyncio
import random

class ReconnectingClient:
    def __init__(self, connect_func, max_retries=10, base_delay=1):
        self.connect_func = connect_func
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.retry_count = 0
    
    async def run_with_reconnect(self):
        while self.retry_count < self.max_retries:
            try:
                await self.connect_func()
                logger.info("连接成功，开始采集数据")
                self.retry_count = 0  # 重置重试计数
                return True
            except Exception as e:
                self.retry_count += 1
                delay = self.base_delay * (2 ** self.retry_count) + random.uniform(0, 1)
                logger.warning(f"连接失败，{delay:.1f}秒后重试 (第{self.retry_count}次)")
                await asyncio.sleep(delay)
        
        logger.error("重连次数耗尽，请人工干预")
        return False

这里用了指数退避策略。第一次失败等1秒，第二次等2秒，第三次等4秒……这样既不会在短暂故障时频繁重连，也不会在长时间故障时浪费资源。

核心要点：重连机制不是无限重试的。我一般设置最大重试次数为10次，如果10次都连不上，就报警让人工介入。否则系统会陷入死循环，浪费CPU资源。

4.7 实战：一个完整的采集循环

最后，咱们把上面所有内容整合成一个完整的采集循环：

async def data_collection_loop():
    plc_reader = PLCReader('192.168.1.100')
    opcua_reader = OPCUAReader('opc.tcp://192.168.1.200:4840')
    
    while True:
        try:
            # 读取PLC数据
            if not plc_reader.connected:
                plc_reader.connect()
            
            plc_data = plc_reader.read_data(0, 50)
            logger.info(f"PLC数据: {plc_data[:5]}...")  # 只打印前5个
            
            # 读取OPC UA数据
            if not opcua_reader.connected:
                await opcua_reader.connect()
            
            opcua_data = await opcua_reader.read_multiple_nodes([
                "ns=2;s=Device1.Temperature",
                "ns=2;s=Device1.Pressure"
            ])
            logger.info(f"OPC UA数据: {opcua_data}")
            
            # 等待下一个采集周期
            await asyncio.sleep(1)  # 1秒采集一次
            
        except Exception as e:
            logger.error(f"采集循环异常: {e}")
            await asyncio.sleep(5)  # 异常时等待5秒再重试

这个循环会一直运行，每秒采集一次数据。如果某个设备断开了，它会自动尝试重连。如果所有设备都挂了，它会等待5秒后重试。

嗯，这套机制我在好几个项目里都用过，稳定运行了几年没出过问题。你拿去用，基本不用改。

4.8 本章小结

这一章咱们聊了：

  • 用pyModbus读取PLC的Modbus数据，注意地址偏移和批量读取限制
  • 用opcua-asyncio连接OPC UA服务器，利用异步IO提高效率
  • 异常处理要分层，连接异常和读取异常分开处理
  • 重连机制用指数退避策略，避免频繁重连

下一章，咱们会把这些采集上来的数据存到数据库里，然后生成报表。到时候你就知道，数据采集只是第一步，真正的挑战在后面。

课后练习：试着写一个程序，同时连接3台PLC和2台OPC UA服务器，每2秒采集一次数据。如果某个设备断开，自动重连并记录日志。试试看你能不能做到连续运行24小时不出错。
← 上一章 📖 返回目录 下一章 →