光伏储能监控方案
方案概述
光伏储能监控系统通过移远模组实时采集光伏逆变器、储能电池、电网接入点的数据，实现发电量统计、故障预警、远程运维。
系统架构
┌─────────────────────────────────────────────────────┐
│ 现场设备层 │
│ 光伏组件 → 汇流箱 → 逆变器 → 储能电池 → 并网点 │
└─────────────────────────────────────────────────────┘
↓ RS485 / Modbus TCP
┌─────────────────────────────────────────────────────┐
│ 数据采集层 │
│ 光伏监控网关（EC25 + Linux） │
│ 采集：逆变器数据 / 气象数据 / 电网数据 │
└─────────────────────────────────────────────────────┘
↓ 4G/5G 网络
┌─────────────────────────────────────────────────────┐
│ 云端监控平台 │
│ 实时监控 / 发电统计 / 故障告警 / 收益分析 │
└─────────────────────────────────────────────────────┘逆变器数据采集
主流逆变器品牌与协议
| 品牌 | 通信协议 | 接口 |
|---|---|---|
| 华为 | Modbus TCP / SunSpec | 以太网 |
| 阳光电源（SMA） | SunSpec Modbus | RS485 / 以太网 |
| 古瑞瓦特 | Modbus RTU | RS485 |
| 固德威 | Modbus RTU | RS485 |
| 锦浪 | Modbus RTU | RS485 |
SunSpec 标准协议
SunSpec 是光伏行业的标准 Modbus 协议，主流逆变器均支持：
python
from pymodbus.client import ModbusTcpClient
import struct
class SunSpecInverter:
"""SunSpec 标准逆变器数据采集"""
# SunSpec 寄存器地址（Model 103：三相逆变器）
REGISTERS = {
'ac_power': 40083, # 交流功率（W）
'ac_energy': 40093, # 累计发电量（Wh）
'dc_voltage': 40098, # 直流电压（V）
'dc_current': 40099, # 直流电流（A）
'dc_power': 40100, # 直流功率（W）
'temperature': 40103, # 机箱温度（°C）
'status': 40107, # 运行状态
'fault_code': 40108, # 故障码
}
STATUS_MAP = {
1: "关机",
2: "待机",
3: "启动",
4: "MPPT",
5: "调节",
6: "关闭",
7: "故障",
8: "维护"
}
def __init__(self, host, port=502, unit_id=1):
self.client = ModbusTcpClient(host, port=port)
self.unit_id = unit_id
def connect(self):
return self.client.connect()
def read_all(self):
"""读取所有关键数据"""
data = {}
for name, address in self.REGISTERS.items():
result = self.client.read_holding_registers(
address - 1, # Modbus 地址从0开始
2,
slave=self.unit_id
)
if not result.isError():
# SunSpec 使用 32位浮点数（两个16位寄存器）
raw = (result.registers[0] << 16) | result.registers[1]
value = struct.unpack('>f', struct.pack('>I', raw))[0]
data[name] = round(value, 2)
# 解析状态
if 'status' in data:
data['status_text'] = self.STATUS_MAP.get(
int(data['status']), "未知"
)
return data
def get_efficiency(self, data):
"""计算逆变效率"""
if data.get('dc_power', 0) > 0:
return data.get('ac_power', 0) / data['dc_power'] * 100
return 0.0
# 使用示例
inverter = SunSpecInverter("192.168.1.100")
inverter.connect()
data = inverter.read_all()
print(f"交流功率: {data['ac_power']} W")
print(f"累计发电: {data['ac_energy']/1000:.1f} kWh")
print(f"直流电压: {data['dc_voltage']} V")
print(f"运行状态: {data['status_text']}")
print(f"逆变效率: {inverter.get_efficiency(data):.1f}%")储能电池监控
BMS 数据采集
python
class BatteryBMS:
"""储能电池管理系统数据采集"""
def __init__(self, modbus_client, slave_id):
self.client = modbus_client
self.slave_id = slave_id
def read_battery_status(self):
"""读取电池状态"""
# 读取电池基本参数（地址因厂商而异）
result = self.client.read_holding_registers(0, 20, slave=self.slave_id)
if result.isError():
return None
regs = result.registers
return {
"soc": regs[0] / 10.0, # 荷电状态（%）
"soh": regs[1] / 10.0, # 健康状态（%）
"voltage": regs[2] / 10.0, # 总电压（V）
"current": regs[3] / 10.0 - 3000, # 电流（A，偏移3000）
"temperature_max": regs[4] / 10.0, # 最高温度（°C）
"temperature_min": regs[5] / 10.0, # 最低温度（°C）
"charge_energy": regs[6] * 0.1, # 累计充电量（kWh）
"discharge_energy": regs[7] * 0.1, # 累计放电量（kWh）
"cycle_count": regs[8], # 循环次数
"status": self._parse_status(regs[9])
}
def _parse_status(self, status_code):
status_map = {
0: "待机",
1: "充电",
2: "放电",
3: "故障",
4: "均衡"
}
return status_map.get(status_code, "未知")
def check_alarms(self, data):
"""检查告警条件"""
alarms = []
if data['soc'] < 10:
alarms.append({"level": "critical", "msg": "电池电量严重不足（SOC < 10%）"})
elif data['soc'] < 20:
alarms.append({"level": "warning", "msg": "电池电量不足（SOC < 20%）"})
if data['temperature_max'] > 45:
alarms.append({"level": "critical", "msg": f"电池温度过高（{data['temperature_max']}°C）"})
if data['soh'] < 70:
alarms.append({"level": "warning", "msg": f"电池健康度下降（SOH = {data['soh']}%），建议检修"})
return alarms发电量统计与收益分析
python
class SolarAnalytics:
"""光伏发电统计与收益分析"""
def __init__(self, electricity_price=0.45, feed_in_tariff=0.35):
self.electricity_price = electricity_price # 用电价格（元/kWh）
self.feed_in_tariff = feed_in_tariff # 上网电价（元/kWh）
def daily_report(self, generation_kwh, consumption_kwh,
grid_export_kwh, grid_import_kwh):
"""日报告"""
self_consumption = generation_kwh - grid_export_kwh
self_consumption_rate = self_consumption / generation_kwh * 100 if generation_kwh > 0 else 0
# 收益计算
saved_electricity = self_consumption * self.electricity_price
feed_in_income = grid_export_kwh * self.feed_in_tariff
total_income = saved_electricity + feed_in_income
return {
"generation_kwh": round(generation_kwh, 2),
"consumption_kwh": round(consumption_kwh, 2),
"self_consumption_kwh": round(self_consumption, 2),
"self_consumption_rate": round(self_consumption_rate, 1),
"grid_export_kwh": round(grid_export_kwh, 2),
"grid_import_kwh": round(grid_import_kwh, 2),
"saved_electricity_yuan": round(saved_electricity, 2),
"feed_in_income_yuan": round(feed_in_income, 2),
"total_income_yuan": round(total_income, 2),
"co2_reduction_kg": round(generation_kwh * 0.785, 2) # 0.785 kg CO2/kWh
}
def payback_analysis(self, system_cost, annual_generation_kwh):
"""投资回收期分析"""
annual_income = annual_generation_kwh * (
self.electricity_price * 0.7 + # 假设70%自用
self.feed_in_tariff * 0.3 # 30%上网
)
payback_years = system_cost / annual_income
return {
"system_cost": system_cost,
"annual_income": round(annual_income, 0),
"payback_years": round(payback_years, 1),
"25year_income": round(annual_income * 25 - system_cost, 0)
}