逆变器 IGBT 温升预警:基于散热数据提前 30 天识别失效风险

光伏数采网关支持的逆变器几十款。自动维护兼容矩阵用 Python 爬虫 + DB + Web。

一、数据模型

from dataclasses import dataclass
from datetime import datetime

@dataclass
class Inverter:
    id: str
    manufacturer: str
    model: str
    power_kw: float
    voltage_range: str
    protocol: str

@dataclass
class CompatibilityRecord:
    gateway_model: str
    inverter_model: str
    is_compatible: bool
    requires_special_config: bool
    notes: str
    last_tested: datetime
    test_engineer: str

二、Schema

CREATE TABLE inverters (
    id UUID PRIMARY KEY,
    manufacturer VARCHAR(100),
    model VARCHAR(100),
    power_kw NUMERIC,
    voltage_range VARCHAR(50),
    protocol VARCHAR(50)
);

CREATE TABLE compatibility (
    gateway_model VARCHAR(50),
    inverter_id UUID REFERENCES inverters(id),
    is_compatible BOOLEAN,
    requires_special_config BOOLEAN,
    notes TEXT,
    last_tested TIMESTAMP,
    test_engineer VARCHAR(100),
    PRIMARY KEY (gateway_model, inverter_id)
);

三、自动爬取(部分厂商有公开 API)

import asyncio
import httpx

class InverterCatalogScraper:
    SOURCES = {
        "huawei": "https://api.huawei.com/sun2000/products",
        "growatt": "https://api.growatt.com/products",
        "sungrow": "https://api.sungrowpower.com/products",
        "ginlong": "https://api.solis-power.com/products",
    }

    async def scrape_all(self):
        async with httpx.AsyncClient() as client:
            tasks = [self.scrape_vendor(client, v, url) for v, url in self.SOURCES.items()]
            results = await asyncio.gather(*tasks, return_exceptions=True)

        # 汇总
        for vendor, products in zip(self.SOURCES.keys(), results):
            if isinstance(products, Exception):
                print(f"Failed {vendor}: {products}")
                continue
            await self.save_to_db(vendor, products)

四、自动测试

# 网关连接逆变器 + 测试
class CompatibilityTester:
    async def test_inverter(self, gateway_ip: str, inverter_brand: str, inverter_model: str) -> dict:
        """实际跑测试"""
        from modbus_client import ModbusClient
        client = ModbusClient(host=gateway_ip)

        # 探测协议
        try:
            # Modbus RTU
            result = await client.read(unit=1, register=40001)
            return {
                "is_compatible": True,
                "test_passed_at": datetime.utcnow(),
                "protocol": "Modbus RTU"
            }
        except:
            try:
                # SunSpec
                result = await client.read_sunspec()
                return {
                    "is_compatible": True,
                    "protocol": "SunSpec"
                }
            except:
                return {"is_compatible": False, "notes": "无法识别协议"}

五、定期更新

@cron.daily(hour=2)
async def daily_catalog_update():
    scraper = InverterCatalogScraper()
    await scraper.scrape_all()

    tester = CompatibilityTester()
    # 自动测试新发现的逆变器
    new_inverters = await get_inverters_without_compatibility_test()
    for inv in new_inverters:
        result = await tester.test_inverter("192.168.1.10", inv.manufacturer, inv.model)
        await save_test_result(inv, result)

六、Web 可视化(React)

// 兼容矩阵
function CompatibilityMatrix() {
  const [gateways, setGateways] = useState([]);
  const [inverters, setInverters] = useState([]);
  const [matrix, setMatrix] = useState({});

  useEffect(() => { loadData(); }, []);

  return (
    <table className="compatibility-matrix">
      <thead>
        <tr>
          <th></th>
          {gateways.map(g => <th key={g.id}>{g.model}</th>)}
        </tr>
      </thead>
      <tbody>
        {inverters.map(inv => (
          <tr key={inv.id}>
            <td><strong>{inv.manufacturer} {inv.model}</strong></td>
            {gateways.map(g => {
              const compat = matrix[`${g.model}_${inv.id}`];
              return (
                <td key={g.id} className={compat ? "ok" : "no"}>
                  {compat?.is_compatible ? "✓" : "×"}
                  {compat?.requires_special_config && " ⚠"}
                </td>
              );
            })}
          </tr>
        ))}
      </tbody>
    </table>
  );
}

七、查询 API

@app.get("/api/compatibility")
async def get_compatibility(gateway_model: str, inverter_brand: str = None):
    return await db.fetch(
        "SELECT * FROM compatibility WHERE gateway_model = $1",
        gateway_model
    )

@app.post("/api/compatibility/feedback")
async def report_compatibility(report: dict):
    """用户反馈"""
    await db.execute("""
        INSERT INTO user_feedback
        (gateway_model, inverter_brand, inverter_model, is_working, notes)
        VALUES ($1, $2, $3, $4, $5)
    """, ...)

八、ZenovaOS 视角

ZenovaOS 网关自动测试 + 维护兼容矩阵,业主选型一键查询。

总结

兼容矩阵 = 自动爬取 + 测试 + Web 可视化。告别 Excel 维护。