3、数据采集软件栈：Python与pyModbus库、pylogix库、OPC UA客户端实现

聊到数据采集，很多新手第一反应就是「找个软件，连上设备，开始读数据」。

嗯，理论上没错。但实际项目里，你会发现现场的设备五花八门——有的走Modbus TCP，有的用PLC的CIP协议，还有的必须走OPC UA。你不可能为每种协议都买一套商业软件，那成本太高了。

所以，我的做法是：用Python搭一套自己的采集软件栈。说白了，就是针对不同协议，选对库，写几行代码，搞定。

3.1 为什么选Python？

我个人习惯用Python做数据采集，原因很简单：生态好，上手快。你想想看，风电场的设备通信协议就那么几种，而Python社区几乎都有对应的开源库。你不需要从零写协议栈，只需要调用API就行。

我在项目中遇到过最极端的情况：一个风场同时有3种不同品牌的PLC和2种传感器。如果用C++写，光调试协议就得两周。换成Python，一天搭完原型，三天上线跑数据。

3.2 pyModbus库：Modbus TCP/RTU采集

Modbus协议在风电行业太常见了。很多气象站、振动传感器、甚至部分老式PLC都支持Modbus。pyModbus是我用得最多的库，没有之一。

3.2.1 安装与基础用法

pip install pymodbus

安装完，直接上代码。我习惯先写一个简单的读取保持寄存器的函数：

from pymodbus.client import ModbusTcpClient

def read_modbus_data(ip, port, start_address, count):
    client = ModbusTcpClient(ip, port=port)
    client.connect()
    result = client.read_holding_registers(start_address, count, slave=1)
    client.close()
    return result.registers

# 示例：读取IP为192.168.1.100的风速仪数据
data = read_modbus_data('192.168.1.100', 502, 0, 10)
print(data)

这里要注意一点：slave ID 不是固定的。有些设备默认是1，有些是255。我曾经因为没确认这个参数，浪费了整整一个下午。

⚠️ 避坑指南： 我曾经遇到过一台老式风速仪，slave ID 写的是0，但文档里写的是1。结果怎么连都连不上。后来用Modbus扫描工具扫了一遍才发现。所以，拿到设备后，先用扫描工具确认slave ID，别信文档。

3.2.2 批量读取与异常处理

实际风场里，你不可能只读一个寄存器。我一般会批量读取，比如一次读50个寄存器，然后解析成风速、风向、温度等物理量。

def batch_read_modbus(devices):
    results = {}
    for dev in devices:
        try:
            raw = read_modbus_data(dev['ip'], dev['port'], dev['start'], dev['count'])
            # 解析原始数据为物理量
            wind_speed = raw[0] / 10.0  # 假设精度为0.1
            wind_direction = raw[1] / 10.0
            results[dev['name']] = {'speed': wind_speed, 'direction': wind_direction}
        except Exception as e:
            print(f"读取 {dev['name']} 失败: {e}")
            results[dev['name']] = None
    return results

为什么加try-except？因为现场网络不稳定。你想想看，风场里电磁干扰大，网线可能被老鼠咬断，设备可能突然重启。不加异常处理，程序跑着跑着就崩了。

3.3 pylogix库：PLC数据采集（CIP协议）

风电场的核心控制器，很多用的是罗克韦尔（AB）的PLC。这些PLC走的是CIP协议，Modbus搞不定。这时候，pylogix库就派上用场了。

3.3.1 连接与读取标签

pip install pylogix
from pylogix import PLC

def read_plc_tag(ip, tag_name):
    with PLC() as comm:
        comm.IPAddress = ip
        ret = comm.Read(tag_name)
        if ret.Status == 'Success':
            return ret.Value
        else:
            print(f"读取标签 {tag_name} 失败: {ret.Status}")
            return None

# 示例：读取PLC中的风速标签
speed = read_plc_tag('192.168.1.200', 'WindSpeed')
print(f"当前风速: {speed} m/s")

pylogix的API设计得很简洁。你只需要知道PLC的IP地址和标签名（Tag Name），就能直接读数据。不需要关心底层是CIP还是EtherNet/IP。

💡 个人经验： 我建议你在PLC程序里，把所有需要采集的数据都定义成全局标签（Global Tag），不要用局部标签。因为局部标签在PLC程序扫描周期里可能被优化掉，导致读取失败。我在一个项目中就踩过这个坑，后来让电气工程师改了PLC程序才解决。

3.3.2 批量读取与性能优化

如果你一次只读一个标签，那效率太低了。pylogix支持批量读取：

def batch_read_plc(ip, tag_list):
    with PLC() as comm:
        comm.IPAddress = ip
        results = comm.Read(tag_list)
        data = {}
        for res in results:
            if res.Status == 'Success':
                data[res.TagName] = res.Value
            else:
                data[res.TagName] = None
        return data

tags = ['WindSpeed', 'WindDirection', 'RotorSpeed', 'GeneratorPower']
data = batch_read_plc('192.168.1.200', tags)
print(data)

批量读取的好处是：一次网络请求，拿到多个标签的值。对于风场这种需要高频采集的场景（比如1秒一次），能显著降低网络负载。

3.4 OPC UA客户端实现

OPC UA是工业4.0的核心协议。现在很多新风机都支持OPC UA，因为它比Modbus更安全，数据模型更丰富。

3.4.1 使用opcua库

pip install opcua
from opcua import Client

def read_opcua_data(endpoint_url, node_id):
    client = Client(endpoint_url)
    try:
        client.connect()
        node = client.get_node(node_id)
        value = node.get_value()
        return value
    finally:
        client.disconnect()

# 示例：连接OPC UA服务器，读取风速
url = "opc.tcp://192.168.1.100:4840"
node = "ns=2;s=WindTurbine.WindSpeed"
speed = read_opcua_data(url, node)
print(f"OPC UA 风速: {speed}")

OPC UA的节点ID（Node ID）格式比较灵活。常见的有两种：

  • 数值型：比如 ns=2;i=1001
  • 字符串型：比如 ns=2;s=WindTurbine.WindSpeed

我个人更推荐用字符串型，因为可读性强，调试时一眼就能看出是哪个变量。

🔑 关键点： OPC UA支持安全连接。生产环境中，一定要启用用户名/密码认证或证书认证。别问我为什么——我曾经在一个测试风场里，用匿名连接跑了三个月，后来被安全审计发现，差点背处分。

3.4.2 订阅模式（实时推送）

OPC UA最强大的功能是订阅（Subscription）。你不需要轮询，服务器会主动推送数据变化。

from opcua import Client, ua

def subscribe_opcua(endpoint_url, node_id, callback):
    client = Client(endpoint_url)
    client.connect()
    node = client.get_node(node_id)
    
    # 创建订阅
    sub = client.create_subscription(100, callback)  # 100ms 采样间隔
    handle = sub.subscribe_data_change(node)
    
    print("订阅成功，等待数据...")
    # 保持运行
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        sub.delete()
        client.disconnect()

def data_change_callback(node, val, data):
    print(f"数据变化: {node} = {val}")

# 使用
subscribe_opcua("opc.tcp://192.168.1.100:4840", 
               "ns=2;s=WindTurbine.WindSpeed", 
               data_change_callback)

订阅模式的好处是：数据变化时立即通知，延迟低。对于需要实时监控的变量（比如振动、转速），我强烈建议用订阅，而不是轮询。

3.5 三种协议的对比与选型

说了这么多，到底该用哪种？我整理了一张表，方便你选型：

协议 适用场景 优点 缺点 我的推荐
Modbus TCP 传感器、老式PLC、气象站 简单、兼容性好 安全性差、数据模型简单 适合低频采集（>1秒）
CIP (pylogix) 罗克韦尔PLC 原生支持、性能高 仅限AB设备 AB PLC首选
OPC UA 新风机、SCADA系统 安全、数据模型丰富、支持订阅 配置稍复杂 新项目首选

你想想看，如果现场全是新设备，我肯定首选OPC UA。但如果有一台老式风速仪，那就只能用Modbus。实际项目中，往往是三种协议混用。

3.6 统一采集框架

最后，我分享一个我自己在项目中用的统一采集框架思路。说白了，就是把三种协议封装成统一的接口：

class DataCollector:
    def __init__(self, config):
        self.config = config
        self.clients = {}
    
    def connect(self, device):
        if device['protocol'] == 'modbus':
            # 初始化Modbus客户端
            pass
        elif device['protocol'] == 'cip':
            # 初始化pylogix客户端
            pass
        elif device['protocol'] == 'opcua':
            # 初始化OPC UA客户端
            pass
    
    def read(self, device, tag):
        # 统一读取接口
        pass
    
    def subscribe(self, device, tag, callback):
        # 统一订阅接口
        pass

这样做的好处是：上层业务代码不需要关心底层协议。你只需要在配置文件中写清楚设备IP、协议类型、标签名，采集框架自动处理一切。

💡 我的习惯： 我会把设备配置写在YAML文件里，而不是硬编码在代码中。这样，现场运维人员改个IP地址，不需要动代码，改配置文件就行。你想想看，风场运维人员有几个会Python？但改YAML文件，培训10分钟就会了。

嗯，数据采集软件栈就聊到这里。三种协议，三个库，一个统一框架。掌握了这些，大部分风电数据采集场景你都能搞定。

专注资料整理
← 上一章 📖 返回目录 下一章 →