4、数据采集实战：从风机PLC读取实时数据、数据缓存与本地存储、断线重连机制

好，咱们直接进入正题。这一章是实战中的硬骨头——从风机PLC里把实时数据拽出来，还得保证它不掉、不丢、不乱。我在风场调试那会儿，最怕的就是半夜接到电话说数据断了。嗯，今天就把这些坑都给你填上。

4.1 从风机PLC读取实时数据

风机PLC说白了就是个工业控制器，它里面跑着各种参数：风速、转速、桨距角、发电机温度……我们得用工业协议跟它对话。最常见的协议是Modbus TCP和OPC UA。

我个人习惯用Modbus TCP，因为它轻量、稳定，大部分风机都支持。你想想看，一个风场几十台风机，每台都要轮询，协议太重了扛不住。

下面是一个简单的Python示例，用pymodbus库读取风机PLC的寄存器数据：

from pymodbus.client import ModbusTcpClient
import time

# 风机PLC的IP和端口（通常Modbus TCP端口是502）
PLC_HOST = "192.168.1.100"
PLC_PORT = 502

client = ModbusTcpClient(PLC_HOST, port=PLC_PORT)

def read_wind_turbine_data():
    """读取风机关键参数"""
    if not client.connect():
        print("连接PLC失败，请检查网络")
        return None
    
    try:
        # 读取保持寄存器，起始地址0，读取10个寄存器
        # 地址映射：0=风速，1=转速，2=桨距角，3=有功功率...
        result = client.read_holding_registers(0, 10, unit=1)
        
        if result.isError():
            print(f"读取失败: {result}")
            return None
        
        data = {
            "wind_speed": result.registers[0] / 10.0,  # 风速，单位m/s
            "rotor_speed": result.registers[1] / 100.0, # 转速，单位rpm
            "pitch_angle": result.registers[2] / 10.0,  # 桨距角，单位度
            "active_power": result.registers[3],         # 有功功率，单位kW
            "timestamp": time.time()
        }
        return data
    
    except Exception as e:
        print(f"读取异常: {e}")
        return None
    finally:
        client.close()

# 调用一次试试
data = read_wind_turbine_data()
if data:
    print(f"风速: {data['wind_speed']} m/s, 功率: {data['active_power']} kW")

4.2 数据缓存与本地存储

数据读出来了，然后呢？直接往数据库里写？别急。风场网络不稳定，数据库也可能挂。所以本地缓存是必须的。

我一般用两种缓存策略：

• 内存缓存：用队列暂存最近N条数据，防止突发写入压力
• 文件缓存：写入本地CSV或JSON文件，作为持久化备份

下面是一个带缓存的数据采集器示例：

import csv
import os
from collections import deque
from datetime import datetime

class WindDataCollector:
    def __init__(self, max_cache=100, file_path="./wind_data.csv"):
        self.cache = deque(maxlen=max_cache)  # 内存缓存，最多100条
        self.file_path = file_path
        self._init_file()
    
    def _init_file(self):
        """如果文件不存在，创建并写入表头"""
        if not os.path.exists(self.file_path):
            with open(self.file_path, 'w', newline='') as f:
                writer = csv.writer(f)
                writer.writerow(["timestamp", "wind_speed", "rotor_speed", 
                                 "pitch_angle", "active_power"])
    
    def add_data(self, data):
        """添加一条数据到缓存，并写入文件"""
        # 先写入内存缓存
        self.cache.append(data)
        
        # 再写入本地文件（追加模式）
        with open(self.file_path, 'a', newline='') as f:
            writer = csv.writer(f)
            writer.writerow([
                datetime.fromtimestamp(data['timestamp']).isoformat(),
                data['wind_speed'],
                data['rotor_speed'],
                data['pitch_angle'],
                data['active_power']
            ])
        
        print(f"已缓存并存储: {data['wind_speed']} m/s")
    
    def get_cached_data(self):
        """获取当前缓存的所有数据"""
        return list(self.cache)

# 使用示例
collector = WindDataCollector(max_cache=50)
data = read_wind_turbine_data()
if data:
    collector.add_data(data)

4.3 断线重连机制

这是最关键的环节。风场环境恶劣，网络抖动、PLC重启、交换机掉电……什么情况都可能发生。没有断线重连，数据采集就是空中楼阁。

我设计断线重连时，遵循三个原则：

1. 指数退避：第一次重连等1秒，第二次等2秒，第三次等4秒……最多等60秒
2. 数据不丢：断线期间的数据要缓存起来，等恢复后补发
3. 状态可观测：记录重连日志，方便排查问题

来看一个完整的断线重连采集器：

import time
import random

class RobustCollector:
    def __init__(self, plc_host, plc_port, max_retry_interval=60):
        self.host = plc_host
        self.port = plc_port
        self.max_retry = max_retry_interval
        self.retry_count = 0
        self.is_connected = False
        self.client = None
        self.cache = []  # 断线期间的数据缓存
    
    def connect(self):
        """尝试连接PLC，带重试机制"""
        while True:
            try:
                self.client = ModbusTcpClient(self.host, port=self.port)
                if self.client.connect():
                    self.is_connected = True
                    self.retry_count = 0
                    print("连接成功！")
                    # 连接恢复后，补发缓存数据
                    self._flush_cache()
                    return True
                else:
                    raise Exception("连接失败")
            except Exception as e:
                self.is_connected = False
                self.retry_count += 1
                wait_time = min(2 ** self.retry_count, self.max_retry)
                print(f"连接失败，{wait_time}秒后重试... (第{self.retry_count}次)")
                time.sleep(wait_time)
    
    def _flush_cache(self):
        """补发缓存中的数据"""
        if self.cache:
            print(f"正在补发{len(self.cache)}条缓存数据...")
            for data in self.cache:
                # 这里可以写入数据库或发送到云端
                pass
            self.cache.clear()
    
    def collect_once(self):
        """单次采集，带断线重连"""
        if not self.is_connected:
            self.connect()
        
        try:
            data = read_wind_turbine_data()  # 复用之前的读取函数
            if data:
                # 正常采集，写入本地存储
                collector.add_data(data)
                return data
        except Exception as e:
            print(f"采集异常: {e}")
            self.is_connected = False
            # 缓存当前时间戳，表示这个时刻的数据丢失了
            self.cache.append({"timestamp": time.time(), "status": "lost"})
            return None
    
    def run_forever(self, interval=1):
        """持续采集，interval为采集间隔（秒）"""
        print("开始持续采集...")
        while True:
            self.collect_once()
            time.sleep(interval)

# 启动采集
collector = RobustCollector("192.168.1.100", 502)
# collector.run_forever(interval=5)  # 每5秒采集一次

4.4 整体架构与流程图

把上面三个环节串起来，就是一套完整的数据采集链路。我画了一张流程图，帮你理清思路：

这张图里，数据从PLC出发，经过采集模块进入内存缓存，同时写入本地文件。如果网络断了，断线重连模块会接管，等恢复后自动补发数据。说白了，这就是一套「不怕断、不怕丢」的采集方案。

4.5 避坑指南

最后，分享几个我踩过的坑：

• 寄存器地址别写死：不同型号的风机，地址映射可能不同。建议用配置文件管理。
• 采集间隔别太短：PLC的响应时间一般在几十毫秒到几百毫秒。如果你每秒采集100次，PLC会扛不住。我一般设3-5秒一次。
• 文件写入要加锁：如果多个线程同时写同一个文件，数据会乱掉。用队列串行化写入。
• 日志一定要打：断线重连的每个步骤都要记录日志。不然出了问题，你根本不知道是网络断了还是PLC挂了。

好了，这一章的内容就到这里。数据采集是AI预处理的第一步，也是最重要的一步。数据质量不行，后面再好的模型也是白搭。记住：稳比快重要，可靠比花哨重要。