第四章：数据采集模块——Modbus协议解析、PLC数据读取、传感器数据清洗、实时数据入库
数据采集，说白了就是风电系统的“耳朵”和“眼睛”。
风机转不转、风速多少、齿轮箱温度高不高——这些信息都得靠数据采集模块来拿。我做了这么多年风电软件，见过太多因为采集层没做好，导致上层分析全是垃圾数据的案例。嗯，这一章咱们就把这块硬骨头啃下来。
4.1 Modbus协议解析——工业通信的“通用语言”
Modbus协议在风电领域太常见了。PLC、变频器、传感器，几乎都支持它。我个人习惯把Modbus理解成一种“问-答”机制：主站发请求，从站回数据。
Modbus有两种主流模式：RTU（二进制）和TCP（网络）。风场里，RTU多见于串口通信，TCP则走以太网。
核心要点：解析Modbus，关键是搞清楚功能码和寄存器地址。
- 功能码03：读保持寄存器（最常用，读PLC数据）
- 功能码04：读输入寄存器（读传感器数据）
- 功能码06/16：写单个/多个寄存器（控制设备）
举个例子，读取风机齿轮箱油温，PLC里可能映射到寄存器地址40001。你发个03功能码，起始地址0x0000，读2个字节，回来的数据就是温度值。
// Modbus RTU 读取请求示例（16进制）
// 从站地址: 01
// 功能码: 03
// 起始地址: 0x0000
// 寄存器数量: 0x0002
// CRC校验: 0xC40B
01 03 00 00 00 02 C4 0B
// 响应示例
// 从站地址: 01
// 功能码: 03
// 字节数: 04
// 数据: 0x00 0xFA (250，即25.0°C)
// CRC校验: 0x79 0x84
01 03 04 00 FA 00 00 79 84避坑指南：我曾经遇到过一个问题——CRC校验算错了，导致通信时好时坏。排查了两天才发现是CRC算法里高低字节顺序搞反了。记住，Modbus RTU的CRC是低字节在前，高字节在后。
4.2 PLC数据读取——别让轮询变成“轮询灾难”
PLC是风机的“大脑”，里面存着所有运行参数。读取PLC数据，最直接的方式就是轮询。
但这里有个坑：轮询频率太高，PLC会扛不住；频率太低，数据实时性又差。我一般建议这样设计：
- 高频数据（100ms-500ms）：风速、转速、功率——这些变化快，需要快速采集
- 中频数据（1s-5s）：温度、压力、振动——变化相对平缓
- 低频数据（10s-60s）：累计发电量、运行状态——不需要频繁读取
你想想看，如果所有数据都用100ms去轮询，PLC光忙着响应请求了，还怎么控制风机？
注意：PLC的通信端口是有带宽限制的。我曾经在某个风场遇到过，因为轮询任务太多，导致PLC通信超时，风机直接报“通信故障”停机。后来我们把轮询分组，错开时间，问题就解决了。
代码实现上，我习惯用异步非阻塞的方式。Python里用pymodbus库，配合asyncio，可以轻松管理多个PLC的连接。
import asyncio
from pymodbus.client import AsyncModbusTcpClient
async def read_plc_data(ip, port=502):
client = AsyncModbusTcpClient(ip, port)
await client.connect()
# 读取保持寄存器，起始地址0，长度10
result = await client.read_holding_registers(0, 10, slave=1)
if result.isError():
print(f"读取失败: {result}")
else:
# 解析数据，比如第一个寄存器是风速
wind_speed = result.registers[0] / 10.0 # 假设精度是0.1
print(f"风速: {wind_speed} m/s")
client.close()
# 运行
asyncio.run(read_plc_data("192.168.1.100"))4.3 传感器数据清洗——垃圾进，垃圾出
传感器数据，说白了就是“脏活累活”。风场环境恶劣，传感器容易出问题：信号漂移、断线、干扰、毛刺……
我见过最离谱的一次，某个风速传感器连续三天输出恒定的12.5m/s，明明那天根本没风。如果不清洗，上层系统会以为风机一直在发电，算出来的发电量全是错的。
数据清洗，我一般分三步走：
- 格式校验：检查数据长度、类型、范围是否合法
- 异常值过滤：超出物理极限的值直接丢弃（比如风速不可能超过50m/s）
- 平滑处理：用中值滤波或移动平均，去除毛刺
实战经验：我曾经在项目中用了一个简单的“3σ原则”来剔除异常值。计算最近N个点的均值和标准差，如果当前值偏离均值超过3倍标准差，就认为是异常点，用前一个有效值替代。
def clean_sensor_data(data_window, new_value):
"""
简单的3σ异常值清洗
data_window: 最近N个有效数据点
new_value: 新采集的原始值
"""
if len(data_window) < 3:
# 数据太少，直接接受
return new_value
mean = sum(data_window) / len(data_window)
variance = sum((x - mean) ** 2 for x in data_window) / len(data_window)
std = variance ** 0.5
# 如果新值偏离均值超过3倍标准差，认为是异常
if abs(new_value - mean) > 3 * std:
print(f"异常值检测: {new_value}，使用均值 {mean} 替代")
return mean
else:
return new_value小技巧：对于温度这类变化缓慢的信号，我习惯用“变化率限制”。比如温度1秒内变化超过5°C，基本可以判定是传感器故障。这个阈值可以根据历史数据动态调整。
4.4 实时数据入库——别让数据库成为瓶颈
数据清洗完了，得存起来。风电数据的特点是：量大、写入频繁、查询以时间范围为主。
传统的关系型数据库（比如MySQL）在这种场景下很容易成为瓶颈。我推荐使用时序数据库，比如InfluxDB、TimescaleDB，或者干脆用TDengine——专门为物联网设计的。
入库策略上，我一般这样设计：
- 批量写入：不要一条一条插，攒够100条或1秒再批量写入，性能提升10倍以上
- 数据压缩：时序数据库自带压缩，但也可以在前端做一次简单压缩（比如只存变化值）
- 分片存储：按时间分片，比如一天一个表，查询和清理都方便
# 使用InfluxDB批量写入示例
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
def batch_write_to_influx(data_points):
"""
data_points: 列表，每个元素是(时间戳, 标签, 字段值)
"""
client = InfluxDBClient(url="http://localhost:8086", token="my-token", org="my-org")
write_api = client.write_api(write_options=SYNCHRONOUS)
points = []
for ts, tag, value in data_points:
point = Point("wind_turbine") \
.tag("turbine_id", tag) \
.field("value", value) \
.time(ts)
points.append(point)
# 批量写入
write_api.write(bucket="wind_data", record=points)
print(f"写入 {len(points)} 条数据")
client.close()注意：我曾经遇到过一个问题——写入速度跟不上采集速度，导致数据积压在内存里，最后OOM（内存溢出）。解决方案是加一个“背压机制”：当写入队列超过阈值时，主动丢弃一些不重要的数据（比如低频传感器数据），保证核心数据不丢失。
知识体系总览
这一章的内容，说白了就是一条数据从“采集”到“入库”的完整链路。我画了一张图，帮你理清整体逻辑：
这张图把四个核心模块串起来了。从传感器原始数据开始，经过Modbus协议解析，再到PLC读取，然后清洗掉脏数据，最后入库存储。每一步都有坑，每一步也都有技巧。
嗯，数据采集模块就讲到这里。记住一句话：采集层做扎实了，上层分析才能站得住脚。否则，再牛的算法也救不了垃圾数据。