分布式光伏电站为什么需要统一运营平台?

分布式光伏 EMS 的数据链路是核心。Modbus → MQTT → TimescaleDB,完整 ETL 实战。

一、链路架构

graph LR
    A["逆变器"] -->|Modbus| B["边缘网关"]
    B -->|MQTT| C["云端 Broker"]
    C --> D["数据消费者"]
    D --> E["TimescaleDB"]
    D --> F["Kafka(备份)"]
    E --> G["Grafana / 大屏"]

二、边缘网关采集(Python)

import asyncio
import paho.mqtt.client as mqtt
import json
from pymodbus.client import AsyncModbusSerialClient
import struct
from datetime import datetime

INVERTERS = [
    {"id": "INV001", "slave_id": 1},
    {"id": "INV002", "slave_id": 2},
]

class EdgeGateway:
    def __init__(self, modbus_port: str, mqtt_broker: str):
        self.modbus_client = AsyncModbusSerialClient(
            port=modbus_port, baudrate=9600
        )
        self.mqtt_client = mqtt.Client(client_id="edge_gateway_001")
        self.mqtt_client.connect(mqtt_broker, 1883)
        self.mqtt_client.loop_start()

    async def read_inverter(self, inv: dict) -> dict | None:
        await self.modbus_client.connect()
        try:
            rr = await self.modbus_client.read_holding_registers(
                address=0x0010, count=16, slave=inv["slave_id"]
            )
            if rr.isError():
                return None
            regs = rr.registers
            return {
                "id": inv["id"],
                "timestamp": datetime.utcnow().isoformat(),
                "power_kw": struct.unpack(">f", struct.pack(">HH", regs[0], regs[1]))[0],
                "today_kwh": struct.unpack(">f", struct.pack(">HH", regs[2], regs[3]))[0],
                "voltage_v": regs[8] / 10.0,
                "current_a": regs[9] / 10.0,
                "temperature_c": regs[10] / 10.0,
                "status": regs[14],
            }
        finally:
            pass

    def publish(self, data: dict):
        topic = f"solar/inverters/{data['id']}/data"
        self.mqtt_client.publish(topic, json.dumps(data), qos=1)

    async def run(self):
        while True:
            for inv in INVERTERS:
                try:
                    data = await self.read_inverter(inv)
                    if data:
                        self.publish(data)
                except Exception as e:
                    print(f"Read failed {inv['id']}: {e}")
            await asyncio.sleep(60)  # 每分钟

# 启动
gw = EdgeGateway(modbus_port="/dev/ttyUSB0", mqtt_broker="mqtt.cloud.example.com")
asyncio.run(gw.run())

三、云端消费者(Python)

import paho.mqtt.client as mqtt
import asyncpg
import json
import asyncio

class MQTTtoDBConsumer:
    def __init__(self, mqtt_broker: str, db_dsn: str):
        self.db_pool = None
        self.db_dsn = db_dsn

        self.client = mqtt.Client(client_id="db_consumer_001")
        self.client.on_message = self._on_message
        self.client.connect(mqtt_broker, 1883)
        self.client.subscribe("solar/inverters/+/data")
        self.batch = []
        self.batch_size = 100

    async def init_db(self):
        self.db_pool = await asyncpg.create_pool(self.db_dsn, min_size=5, max_size=20)

    def _on_message(self, client, userdata, msg):
        try:
            data = json.loads(msg.payload.decode())
            self.batch.append(data)
            if len(self.batch) >= self.batch_size:
                asyncio.create_task(self._flush_batch())
        except Exception as e:
            print(f"Parse error: {e}")

    async def _flush_batch(self):
        if not self.batch:
            return
        async with self.db_pool.acquire() as conn:
            await conn.executemany(
                """
                INSERT INTO readings (ts, inverter_id, power_kw, energy_kwh, voltage_v, current_a, temperature_c)
                VALUES ($1, $2, $3, $4, $5, $6, $7)
                """,
                [
                    (
                        datetime.fromisoformat(d["timestamp"]),
                        d["id"], d["power_kw"], d["today_kwh"],
                        d["voltage_v"], d["current_a"], d["temperature_c"]
                    )
                    for d in self.batch
                ]
            )
        self.batch.clear()

    async def run(self):
        await self.init_db()
        self.client.loop_forever()

consumer = MQTTtoDBConsumer("mqtt.cloud.example.com", "postgresql://...")
asyncio.run(consumer.run())

四、TimescaleDB Schema

CREATE EXTENSION IF NOT EXISTS timescaledb;

CREATE TABLE readings (
    ts TIMESTAMPTZ NOT NULL,
    inverter_id VARCHAR(64) NOT NULL,
    power_kw DOUBLE PRECISION,
    energy_kwh DOUBLE PRECISION,
    voltage_v DOUBLE PRECISION,
    current_a DOUBLE PRECISION,
    temperature_c DOUBLE PRECISION,
    PRIMARY KEY (ts, inverter_id)
);

SELECT create_hypertable('readings', 'ts', chunk_time_interval => INTERVAL '1 day');

CREATE INDEX idx_readings_inverter_ts ON readings (inverter_id, ts DESC);

-- 30 天后压缩
ALTER TABLE readings SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'inverter_id'
);
SELECT add_compression_policy('readings', INTERVAL '30 days');

-- 实时聚合视图
CREATE MATERIALIZED VIEW readings_15min
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('15 minutes', ts) AS bucket,
    inverter_id,
    AVG(power_kw) AS avg_power,
    MAX(power_kw) AS max_power,
    LAST(energy_kwh, ts) - FIRST(energy_kwh, ts) AS energy_15min
FROM readings
GROUP BY bucket, inverter_id;

SELECT add_continuous_aggregate_policy('readings_15min',
    start_offset => INTERVAL '1 day',
    end_offset => INTERVAL '15 minutes',
    schedule_interval => INTERVAL '15 minutes');

五、Grafana SQL

-- 各逆变器今日发电
SELECT
    $__time(ts),
    inverter_id,
    energy_kwh
FROM readings
WHERE $__timeFilter(ts)
ORDER BY ts;

-- 实时总功率
SELECT
    inverter_id,
    LAST(power_kw, ts) as latest_power
FROM readings
WHERE ts > NOW() - INTERVAL '5 minutes'
GROUP BY inverter_id;

六、性能 benchmark

单实例:
- 边缘网关:100 逆变器 / 分钟
- MQTT broker:10万 客户端
- 消费者:100k events/min
- TimescaleDB:200k 行/秒
- 查询响应:< 100ms(预聚合)

七、ZenovaOS 实践

ZenovaOS 数据链路用 Modbus + MQTT + TimescaleDB,5000+ 客户跑稳。

总结

分布式光伏 EMS 数据链路:边缘 Modbus → MQTT → 云端 TimescaleDB → Grafana,Python 全栈。