拓冰网站定制 
简介在光伏监控与能源管理系统中数据采集是打通设备与平台的关键环节。Modbus TCP作为工业领域广泛应用的通信协议通过标准化寄存器读写实现设备数据的可靠获取MQTT则以其轻量、解耦的特性成为物联网数据传递的理想选择。将两者结合能够构建一条从现场设备到云端的稳定数据链路既支持本地实时监控又便于对接时序数据库、可视化看板与告警服务。针对家庭光伏场景这种方式可摆脱对厂商云平台的依赖实现自主数据留存与多设备统一接入。本文从协议原理出发完整演示如何利用Modbus TCP读取华为逆变器运行参数并转换为MQTT消息发布同时涵盖轮询策略、断线重连、现场部署常见问题与多品牌扩展方法为能源数据采集提供一套可复用的工程实践方案。 如果家里装的是华为SUN2000L-KTL-L1这种户用逆变器想实时盯着发电数据官方App其实够用但一旦你想把这些数据接入自己的运维平台、做本地存储或联动告警就会撞上一堵墙。我最后走通的路是用Modbus TCP协议从逆变器本地接口读取运行数据统一转成MQTT消息发给Broker整套系统从采集到上抛完全由自己控制。这篇文章就是这套采集系统的完整实现记录从原理、代码到现场部署的坑一次说清楚。1. 为什么非得自己做一套数据采集官方App解决不了的问题1.1 从看数据到用数据中间隔着一道平台墙官方App解决的是看的问题不是用的问题。家里装了光伏以后我一开始也觉得App够用每天打开看一眼发了多少度电系统是否正常运行似乎没什么不满足。真正觉得不够用是当我准备把光伏数据并入自己的一套家庭能源管理平台时平台需要同时汇总光伏发电、电表数据、充电桩功率光伏侧如果只靠人工看App再抄数那这套平台就失去意义了。我需要程序化、自动化地拿到逆变器的实时功率、逐日发电量、累计发电量、机内温度这类数据而且最好能秒级更新。想从官方云平台拿这些数据基本走不通。一方面接口没有公开文档抓包逆向出来的接口既要处理token过期、签名算法还要冒着版本升级后失效的风险另一方面跨平台调云端接口的链路太长数据从逆变器到华为云再转发到自己的服务器延迟和稳定性都没保证。这时候我发现了一条更直接的路逆变器本身就提供本地Modbus接口。也就是说我完全可以不经过华为云直接在局域网内把数据读出来。1.2 SUN2000L-KTL-L1的本地数据接口Modbus TCP确实可用华为的逆变器在云平台侧走的是私有协议但在设备本地侧却保留了一个非常传统的工业协议口子Modbus。SUN2000L-KTL-L1这个系列在具备网络通信能力的版本上会开放Modbus TCP服务默认监听502端口通过标准的Modbus功能码就能读取寄存器内的运行参数。关于硬件我需要多说一句不是所有SUN2000L-KTL-L1都自带网口。有些型号需要搭配华为的Dongle或数据采集器才能联网也有部分场景只能用RS485走Modbus RTU。我这次做的是设备在局域网内可通过IP:502访问的场景也就是Modbus TCP。如果你的手头设备只能走RS485思路是一样的只是采集程序里的传输层从TCP换成串口寄存器地址和解析逻辑完全复用。方案确定之后整个项目的任务变得很清晰用Modbus TCP把寄存器里的原始数值读出来按协议文档换算成有意义的数据再通过MQTT发布到消息总线。这样做的好处是下游不管是写数据库、画仪表盘、发告警还是接智能家居都只关心MQTT消息完全不关心逆变器是什么品牌。1.3 这套采集方案适合哪些场景家庭光伏用户想摆脱厂商云平台自己留存数据或者接入Home Assistant等本地自动化系统。光伏运维/能源管理开发者需要把不同品牌逆变器统一接入同一个平台Modbus转MQTT是最通用的做法。边缘计算与物联网爱好者想理解工业协议和物联网消息协议之间怎么打通。如果你只是偶尔看一眼发电量那官方App确实够了不用折腾。但如果你发现自己开始在浏览器里反复刷新数据、想做一个月度报表或者家里已经有一套自动化平台那这篇文章适合你。2. 架构设计与协议原理先搭骨架再填肉2.1 一条完整的数据链路逆变器到MQTT Broker在做任何代码之前先把整体链路画出来逆变器Modbus TCP Server → 采集程序Modbus Client MQTT Publisher → MQTT Broker → 订阅端数据库/可视化/告警有些做法是让采集程序读完Modbus直接写InfluxDB省掉MQTT这一步。短期看确实更简单但我不建议这么干。我在第一版就是这么写的结果后面加告警服务时又得改采集程序把数据同时写数据库和发HTTP请求代码越来越脏。中间加一层MQTT Broker之后采集程序只负责读逆变器数据并发布到MQTT下游谁要数据谁自己订阅。采集端和消费端完全解耦。还有一个很实在的好处如果数据库暂时不可用MQTT Broker可以帮你把数据缓冲住消费端恢复后还能补收不会因为下游故障丢掉数据。2.2 Modbus TCP读取过程一次请求到底发生了什么Modbus TCP是一个请求/响应协议。逆变器是Server端你的采集程序是Client端。读保持寄存器用的是功能码0x03。如果用TCP工具手动发一条请求报文大致长这样事务ID0x0001用来匹配请求和响应协议ID0x0000这是Modbus TCP固定值长度0x0006表示后面还有6个字节单元ID0x01逆变器在Modbus总线上的地址功能码0x03起始地址例如0x7D00寄存器数量例如0x000A响应内容就是起始地址开始的一串寄存器值每个寄存器16位大端序排列。一条请求读20个寄存器一次往返就能拿到40字节数据效率很高。所以写采集程序的时候尽量批量读取而不是一个字段一个字段地读。一次读一组连续寄存器然后在程序里按偏移量拆字段这样对逆变器友好程序自己也省事。2.3 为什么选MQTT做中转解耦、缓冲、多端消费MQTT在这套系统里的角色类似一个数据总线。采集程序把数据发布到某个主题Topic所有需要数据的下游程序都去订阅这个主题。主题设计我推荐下面这种结构solar/{逆变器序列号}/telemetrytelemetry放实时遥测数据比如功率、温度、电压。如果想区分数据类型还可以拆成solar/{序列号}/status和solar/{序列号}/energy前者实时运行状态后者累计发电量每天变更频率低的数据可以设置retain消息这样新订阅者上线立刻能拿到最近一次数值不用等下一个采集周期。消息体我用的JSON结构大概是{ timestamp: 2024-12-20T10:30:0008:00, active_power: 5200.5, daily_yield: 23.4, total_yield: 12345.6, inverter_temp: 35.2, pv1_voltage: 220.1, pv1_current: 6.5, status: 1 }字段名在写入数据库的时候会直接对应列名所以建议一开始就定成统一的命名规范后面省很多事。我用的是snake_case实际项目里用camelCase也可以关键是别变来变去。2.4 轮询策略与数据分组别把所有数据打成一次请求实时功率这类变化快的数据我设置了5秒轮询一次日发电量、总发电量这类累计值其实10秒到1分钟刷一次都行我统一走60秒的批次。这么区分的原因有两个。第一Modbus请求会对逆变器造成一定负载虽然读寄存器很轻但也没必要每秒去轰炸它。第二不同数据对实时性的要求完全不同功率是给监控大屏和告警用的5秒足够了发电量是给报表用的1分钟采一次完全不会影响统计精度。如果一次把所有寄存器全读完虽然请求次数少了但报文会比较长而且有些寄存器在设备空闲时是不更新的读回来也是旧值。所以我把读取分成两批实时运行区功率、电压、电流、温度5秒一轮能量累计区日发电量、总发电量、并网时间等60秒一轮3. 采集程序逐步实现Modbus TCP MQTT完整代码3.1 开发环境与工程目录我用的是Python 3.9依赖库主要是pymodbus和paho-mqtt。pymodbus需要安装3.x版本因为2.x的API在3.x里有不少变化网上很多老教程直接跑会报错。安装命令pip install pymodbus paho-mqtt pyyamlpyyaml用来读配置文件。工程目录我习惯这样组织solar-dashboard/ ├── config.yaml ├── inverter.py ├── mqtt_publisher.py ├── parser.py └── main.py单文件也不是不行但把连接、解析、发布分开后面加新功能会舒服很多。3.2 配置文件与逆变器连接配置文件config.yamlinverter: host: 192.168.8.10 port: 502 unit: 1 timeout: 5 poll_interval_sec: 5 energy_interval_sec: 60 mqtt: broker: 192.168.8.100 port: 1883 username: password: topic_prefix: solar/华为SUN2000L-KTL-L1的Modbus单元ID一般是1但也有特殊设置过的最好在官方文档或调试工具里确认。连接超时我设5秒太短的话设备启动阶段会连不上太长的话故障时程序会卡很久。采集程序里封装一个逆变器连接类用长连接方式不要每次读取都创建新连接# inverter.py from pymodbus.client import ModbusTcpClient class HuaweiInverter: def __init__(self, host, port, unit1, timeout5): self.client ModbusTcpClient(host, portport, timeouttimeout) self.unit unit def connect(self): if self.client.connected: return True return self.client.connect() def close(self): self.client.close() def read_holding_registers(self, address, count): rr self.client.read_holding_registers(addressaddress, countcount, unitself.unit) if rr.isError(): raise RuntimeError(fModbus read error: {rr}) return rr.registers3.3 寄存器批量读取与字段解析华为的Modbus寄存器映射表在官方《SUN2000-XXXKTL Modbus协议》文档里有完整定义不同固件版本略有差异我这里不方便直接贴出绝对地址因为设备型号和固件版本不同地址表可能不一样。最稳妥的方式是先从官方文档拿到你要的那几个字段的寄存器地址然后用Modbus Poll这类工具连上去实测一遍确认数值范围合理后再写进配置文件。我建议把地址和字段映射关系单独放到配置里像这样runtime_registers: address: 32064 # 示例值请以官方文档为准 count: 40 fields: - name: active_power offset: 0 type: int16 scale: 1 - name: inverter_temp offset: 6 type: int16 scale: 0.1这样如果换了型号只需要改配置和字段表不用改解析代码。解析函数的核心逻辑是按偏移量和类型把原始寄存器数组拆成业务字段# parser.py import struct def parse_registers(registers, fields): data {} for f in fields: offset f[offset] ftype f[type] scale f.get(scale, 1) if ftype int16: raw registers[offset] if raw 0x8000: raw - 0x10000 # 转成有符号数 value raw * scale elif ftype uint32: raw (registers[offset] 16) | registers[offset 1] value raw * scale elif ftype float32: raw_bytes struct.pack(HH, registers[offset], registers[offset 1]) value struct.unpack(f, raw_bytes)[0] else: continue data[f[name]] round(value, 3) return data这个函数里我用了大端序处理这个很重要。有些设备字段是低字在前如果发现解析结果乱套把struct.pack里的换成或者把(registers[offset] 16) | registers[offset 1]换成(registers[offset 1] 16) | registers[offset]再验证一遍。3.4 MQTT发布与主循环MQTT发布我单独写了一个类# mqtt_publisher.py import json import paho.mqtt.client as mqtt class MqttPublisher: def __init__(self, broker, port, username, password): self.client mqtt.Client() if username: self.client.username_pw_set(username, password) self.client.on_connect self._on_connect self.client.connect(broker, port, keepalive60) self.client.loop_start() def _on_connect(self, client, userdata, flags, rc): if rc 0: print(MQTT connected) else: print(fMQTT connect failed: {rc}) def publish(self, topic, payload, retainFalse): data json.dumps(payload, ensure_asciiFalse) self.client.publish(topic, data, qos1, retainretain)主循环逻辑# main.py import time import yaml from inverter import HuaweiInverter from mqtt_publisher import MqttPublisher from parser import parse_registers with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) inv_cfg config[inverter] mqtt_cfg config[mqtt] inverter HuaweiInverter(inv_cfg[host], inv_cfg[port], inv_cfg[unit]) publisher MqttPublisher(mqtt_cfg[broker], mqtt_cfg[port], mqtt_cfg[username], mqtt_cfg[password]) sn SUN2000L-KTL-L1-1234 # 建议从设备信息寄存器读取 while True: try: if not inverter.connect(): time.sleep(5) continue registers inverter.read_holding_registers( inv_cfg[runtime_registers][address], inv_cfg[runtime_registers][count] ) data parse_registers(registers, inv_cfg[runtime_registers][fields]) data[timestamp] time.strftime(%Y-%m-%dT%H:%M:%S%z) topic f{mqtt_cfg[topic_prefix]}{sn}/telemetry publisher.publish(topic, data) except Exception as e: print(fpoll error: {e}) time.sleep(inv_cfg[poll_interval_sec])这个主循环看着简单实际部署时还需要考虑几个问题异常后重连间隔要递增、MQTT断了要检测、消息要加本地缓存。3.5 断线重连与本地缓存补发Modbus设备在弱电环境下偶尔超时是正常的关键是程序不能因为一次异常就挂掉。我的做法是记录连续失败次数失败越多重连间隔越大retry 0 while True: try: if not inverter.connect(): retry 1 time.sleep(min(30, retry * 5)) continue # 读数据 发布 ... retry 0 time.sleep(inv_cfg[poll_interval_sec]) except Exception: retry 1 time.sleep(min(30, retry * 5))MQTT Broker离线是另一个场景。paho默认在Broker不可用时会把消息丢掉这样数据库会缺一段数据。我在采集端加了一个SQLite本地缓存发布失败就写本地重连成功后补发。SQLite操作简单、单文件、适合这种低并发场景不需要额外起数据库服务。补发的逻辑大致是每次publish返回失败时把(topic, payload)插入本地表检测到MQTT恢复on_connect回调后读取未发送记录逐条重新发布发布成功再删除记录这个功能看起来不起眼但真遇到Broker重启或者断网几分钟你会发现它救了很多历史数据。4. 现场部署遇到的坑与排查过程4.1 问题一Modbus TCP连接频频超时现象是程序运行一会儿就报connection timeout然后又自己恢复。一开始我怀疑是逆变器性能问题后来排查发现是采集程序部署在Docker容器里而Docker默认使用bridge网络容器到宿主机局域网是经过了NAT的某些非标准端口或长连接会被iptables规则干扰。解决方法有两个一是启动容器时加network_mode: host让采集程序直接用宿主机网络访问局域网二是如果一定要用bridge在宿主机上写一条稳定的端口映射规则并确保防火墙没有清理长连接。我用的是host模式之后连接超时的问题再没出现过。另外要注意路由器或交换机上的端口隔离设置。有些带管理功能的交换机默认会隔离不同VLAN的广播域导致Modbus TCP回包能到但连接建立不稳定。如果Modbus直连正常、从采集服务器连就超时优先检查网络策略和防火墙。4.2 问题二读取出来的功率值明显不对这个坑我印象最深刻。读出来active_power是58432换算成瓦特显然不合理换另一台设备读甚至出现过负值。排查链路是这样的先用Modbus Poll手动读同一个寄存器发现Poll显示的原始值是0xE440确实是58432。对照官方文档发现该字段的寄存器是int16类型不是uint16。0xE440在无符号数里是58432但如果按有符号数解释0xE440的高位为1是负数实际是-7104。再结合该字段的系数0.1就是-710.4这就不对了。随后发现我连错了字段这个寄存器并不是active_power而是另一个温度补偿参数。真正有用的寄存器在相邻的另一块区域。这个排查过程告诉我别迷信任何一份网上贴出来的寄存器地址表必须用自己的设备实测并且要把读到的值和现场能看到的现象比如功率、温度对照起来验证。程序里只要一个字段的解析方式错了后面所有关联数据都会错。4.3 问题三采集程序一启动官方App就掉线程序能稳定读数据之后我发现一个奇怪现象只要我的采集程序连着逆变器官方App的实时数据刷新就会出现卡顿甚至提示设备离线。后来查资料、翻社区才搞明白部分SUN2000L-KTL-L1固件对Modbus TCP的并发连接数做了限制同一时间可能只允许一个或两个客户端连接。官方App或者Dongle本身会占用一条连接我的采集程序再挤进去就会导致有一条连接被踢掉。解决方式很直接采集程序长连接独占同时不要开着官方调试工具或者多个采集实例。如果需要同时读数据最好用RS485扩展串口而不是多个TCP客户端同时连。还有一点程序不需要每次读取都新建连接长连接就OK了我一开始为了安全反复断开重连反而更容易触发连接数限制。4.4 问题四跨天发电量突然变大或回跳日发电量这个字段在0点时会出现归零或者跳变。一开始我以为自己解析错了盯着数据看了好几天才明白这是正常现象。逆变器内部的日发电量在每天凌晨会清零重新累计如果采集服务器和逆变器的时间基准不一致或者采集时间戳用了逆变器本地时间就会出现凌晨那一段数据乱跳。处理方式是让采集程序统一使用服务器时间作为timestamp发电机内部的日电量寄存器值只作为当日累计值记录跨天判断和统计由下游数据库完成不依赖逆变器时间。另外建议给逆变器做一次时间同步很多型号支持NTP时间设置。4.5 关于轮询频率和Broker性能的一些建议轮询太频繁除了给逆变器增加负载还会给MQTT Broker带来消息压力。我的经验是每秒最多一次轮询到头了5秒一次已经足够感受到实时。MQTT Broker如果只是单机服务百来个设备压力很小不需要刻意优化。但如果消息体里带了大字符串、历史累积字段建议拆成两个topic发布低频高价值数据单独走retain消息高频遥测数据用非retain消息这样订阅方可以按需订阅避免接收一堆用不上的数据。5. 数据接到MQTT之后可视化、告警与多站扩展5.1 订阅MQTT写入时序数据库MQTT数据到了Broker之后需要一个订阅程序把消息写入时序数据库。我用的TDengine也可以换成InfluxDB思路一样。TDengine建表语句CREATE STABLE solar_data ( ts TIMESTAMP, active_power FLOAT, daily_yield FLOAT, total_yield FLOAT, inverter_temp FLOAT, status INT ) TAGS (device_id BINARY(32));订阅程序负责从MQTT收消息、解析JSON、执行insertimport json import paho.mqtt.client as mqtt from taos import connect def on_message(client, userdata, msg): data json.loads(msg.payload) device_id msg.topic.split(/)[1] ts data[timestamp] userdata[cursor].execute( INSERT INTO solar_data USING solar_data TAGS (%s) VALUES (?, ?, ?, ?, ?, ?), [device_id, ts, data.get(active_power), data.get(daily_yield), data.get(total_yield), data.get(inverter_temp), data.get(status)] )当然可以直接用EMQX这类Broker内置的规则引擎把消息转发到数据库省一个订阅程序。我这边更倾向于写独立的订阅程序因为后续要做数据清洗和补发逻辑程序方式更灵活。5.2 Grafana看板与告警规则数据进时序数据库之后用Grafana做可视化是最快的。选好数据源、写一条SQL就能得到实时功率曲线、日发电量柱状图和温度趋势。告警我用的是Grafana Alerting。举两个实用规则连续15分钟active_power为0但当天总有光照说明可能存在故障停机inverter_temp超过80度触发高温告警MQTT本身也可以接轻量级告警程序订阅telemetry主题判断数据异常后发Webhook或者邮件。这个程序是独立消费者不影响采集链路所以出问题也不会拖垮数据采集。5.3 扩展到多台逆变器与多品牌设备这套架构最大的好处是方便扩展。多台华为逆变器只需要在配置里增加连接信息每个设备用一个独立线程采集发布到不同topic。其他品牌的逆变器只要支持Modbus也能用同一套解析框架接入差异只体现在寄存器字段映射表上。如果现场既有光伏又有电表和充电桩可以都用Modbus转MQTT这个思路最后所有设备的数据都汇聚到同一个Broker平台层只面对MQTT一个协议这是很标准的多设备数据整合方式。5.4 个人在实际落地中的一点体会整套系统从开始做到稳定运行花了我周末两天时间。起初最耗时的不是写代码而是调寄存器解析。建议大家动手前一定先用Modbus工具把要读的寄存器手动验证一遍把字段清单、类型、系数整理成表格再写代码能省掉大量时间。关于MQTT Broker个人用的话直接Mosquitto就很稳追求管理界面可以用EMQX。我中间还踩了一个小坑Broker的keepalive参数没设置好低频数据场景下连接会被误判断开后来把keepalive调成60秒并加了定时ping才稳定下来。另外建议采集程序里所有日志都带时间戳和来源设备标识排查问题时能直接定位是哪台设备、哪一步出错尤其是多设备场景日志不规范化会非常痛苦。本文还有配套的精品资源点击获取 华为逆变器Modbus TCP数据采集实战：从寄存器到MQTT的完整链路
简介在光伏监控与能源管理系统中数据采集是打通设备与平台的关键环节。Modbus TCP作为工业领域广泛应用的通信协议通过标准化寄存器读写实现设备数据的可靠获取MQTT则以其轻量、解耦的特性成为物联网数据传递的理想选择。将两者结合能够构建一条从现场设备到云端的稳定数据链路既支持本地实时监控又便于对接时序数据库、可视化看板与告警服务。针对家庭光伏场景这种方式可摆脱对厂商云平台的依赖实现自主数据留存与多设备统一接入。本文从协议原理出发完整演示如何利用Modbus TCP读取华为逆变器运行参数并转换为MQTT消息发布同时涵盖轮询策略、断线重连、现场部署常见问题与多品牌扩展方法为能源数据采集提供一套可复用的工程实践方案。 如果家里装的是华为SUN2000L-KTL-L1这种户用逆变器想实时盯着发电数据官方App其实够用但一旦你想把这些数据接入自己的运维平台、做本地存储或联动告警就会撞上一堵墙。我最后走通的路是用Modbus TCP协议从逆变器本地接口读取运行数据统一转成MQTT消息发给Broker整套系统从采集到上抛完全由自己控制。这篇文章就是这套采集系统的完整实现记录从原理、代码到现场部署的坑一次说清楚。1. 为什么非得自己做一套数据采集官方App解决不了的问题1.1 从看数据到用数据中间隔着一道平台墙官方App解决的是看的问题不是用的问题。家里装了光伏以后我一开始也觉得App够用每天打开看一眼发了多少度电系统是否正常运行似乎没什么不满足。真正觉得不够用是当我准备把光伏数据并入自己的一套家庭能源管理平台时平台需要同时汇总光伏发电、电表数据、充电桩功率光伏侧如果只靠人工看App再抄数那这套平台就失去意义了。我需要程序化、自动化地拿到逆变器的实时功率、逐日发电量、累计发电量、机内温度这类数据而且最好能秒级更新。想从官方云平台拿这些数据基本走不通。一方面接口没有公开文档抓包逆向出来的接口既要处理token过期、签名算法还要冒着版本升级后失效的风险另一方面跨平台调云端接口的链路太长数据从逆变器到华为云再转发到自己的服务器延迟和稳定性都没保证。这时候我发现了一条更直接的路逆变器本身就提供本地Modbus接口。也就是说我完全可以不经过华为云直接在局域网内把数据读出来。1.2 SUN2000L-KTL-L1的本地数据接口Modbus TCP确实可用华为的逆变器在云平台侧走的是私有协议但在设备本地侧却保留了一个非常传统的工业协议口子Modbus。SUN2000L-KTL-L1这个系列在具备网络通信能力的版本上会开放Modbus TCP服务默认监听502端口通过标准的Modbus功能码就能读取寄存器内的运行参数。关于硬件我需要多说一句不是所有SUN2000L-KTL-L1都自带网口。有些型号需要搭配华为的Dongle或数据采集器才能联网也有部分场景只能用RS485走Modbus RTU。我这次做的是设备在局域网内可通过IP:502访问的场景也就是Modbus TCP。如果你的手头设备只能走RS485思路是一样的只是采集程序里的传输层从TCP换成串口寄存器地址和解析逻辑完全复用。方案确定之后整个项目的任务变得很清晰用Modbus TCP把寄存器里的原始数值读出来按协议文档换算成有意义的数据再通过MQTT发布到消息总线。这样做的好处是下游不管是写数据库、画仪表盘、发告警还是接智能家居都只关心MQTT消息完全不关心逆变器是什么品牌。1.3 这套采集方案适合哪些场景家庭光伏用户想摆脱厂商云平台自己留存数据或者接入Home Assistant等本地自动化系统。光伏运维/能源管理开发者需要把不同品牌逆变器统一接入同一个平台Modbus转MQTT是最通用的做法。边缘计算与物联网爱好者想理解工业协议和物联网消息协议之间怎么打通。如果你只是偶尔看一眼发电量那官方App确实够了不用折腾。但如果你发现自己开始在浏览器里反复刷新数据、想做一个月度报表或者家里已经有一套自动化平台那这篇文章适合你。2. 架构设计与协议原理先搭骨架再填肉2.1 一条完整的数据链路逆变器到MQTT Broker在做任何代码之前先把整体链路画出来逆变器Modbus TCP Server → 采集程序Modbus Client MQTT Publisher → MQTT Broker → 订阅端数据库/可视化/告警有些做法是让采集程序读完Modbus直接写InfluxDB省掉MQTT这一步。短期看确实更简单但我不建议这么干。我在第一版就是这么写的结果后面加告警服务时又得改采集程序把数据同时写数据库和发HTTP请求代码越来越脏。中间加一层MQTT Broker之后采集程序只负责读逆变器数据并发布到MQTT下游谁要数据谁自己订阅。采集端和消费端完全解耦。还有一个很实在的好处如果数据库暂时不可用MQTT Broker可以帮你把数据缓冲住消费端恢复后还能补收不会因为下游故障丢掉数据。2.2 Modbus TCP读取过程一次请求到底发生了什么Modbus TCP是一个请求/响应协议。逆变器是Server端你的采集程序是Client端。读保持寄存器用的是功能码0x03。如果用TCP工具手动发一条请求报文大致长这样事务ID0x0001用来匹配请求和响应协议ID0x0000这是Modbus TCP固定值长度0x0006表示后面还有6个字节单元ID0x01逆变器在Modbus总线上的地址功能码0x03起始地址例如0x7D00寄存器数量例如0x000A响应内容就是起始地址开始的一串寄存器值每个寄存器16位大端序排列。一条请求读20个寄存器一次往返就能拿到40字节数据效率很高。所以写采集程序的时候尽量批量读取而不是一个字段一个字段地读。一次读一组连续寄存器然后在程序里按偏移量拆字段这样对逆变器友好程序自己也省事。2.3 为什么选MQTT做中转解耦、缓冲、多端消费MQTT在这套系统里的角色类似一个数据总线。采集程序把数据发布到某个主题Topic所有需要数据的下游程序都去订阅这个主题。主题设计我推荐下面这种结构solar/{逆变器序列号}/telemetrytelemetry放实时遥测数据比如功率、温度、电压。如果想区分数据类型还可以拆成solar/{序列号}/status和solar/{序列号}/energy前者实时运行状态后者累计发电量每天变更频率低的数据可以设置retain消息这样新订阅者上线立刻能拿到最近一次数值不用等下一个采集周期。消息体我用的JSON结构大概是{ timestamp: 2024-12-20T10:30:0008:00, active_power: 5200.5, daily_yield: 23.4, total_yield: 12345.6, inverter_temp: 35.2, pv1_voltage: 220.1, pv1_current: 6.5, status: 1 }字段名在写入数据库的时候会直接对应列名所以建议一开始就定成统一的命名规范后面省很多事。我用的是snake_case实际项目里用camelCase也可以关键是别变来变去。2.4 轮询策略与数据分组别把所有数据打成一次请求实时功率这类变化快的数据我设置了5秒轮询一次日发电量、总发电量这类累计值其实10秒到1分钟刷一次都行我统一走60秒的批次。这么区分的原因有两个。第一Modbus请求会对逆变器造成一定负载虽然读寄存器很轻但也没必要每秒去轰炸它。第二不同数据对实时性的要求完全不同功率是给监控大屏和告警用的5秒足够了发电量是给报表用的1分钟采一次完全不会影响统计精度。如果一次把所有寄存器全读完虽然请求次数少了但报文会比较长而且有些寄存器在设备空闲时是不更新的读回来也是旧值。所以我把读取分成两批实时运行区功率、电压、电流、温度5秒一轮能量累计区日发电量、总发电量、并网时间等60秒一轮3. 采集程序逐步实现Modbus TCP MQTT完整代码3.1 开发环境与工程目录我用的是Python 3.9依赖库主要是pymodbus和paho-mqtt。pymodbus需要安装3.x版本因为2.x的API在3.x里有不少变化网上很多老教程直接跑会报错。安装命令pip install pymodbus paho-mqtt pyyamlpyyaml用来读配置文件。工程目录我习惯这样组织solar-dashboard/ ├── config.yaml ├── inverter.py ├── mqtt_publisher.py ├── parser.py └── main.py单文件也不是不行但把连接、解析、发布分开后面加新功能会舒服很多。3.2 配置文件与逆变器连接配置文件config.yamlinverter: host: 192.168.8.10 port: 502 unit: 1 timeout: 5 poll_interval_sec: 5 energy_interval_sec: 60 mqtt: broker: 192.168.8.100 port: 1883 username: password: topic_prefix: solar/华为SUN2000L-KTL-L1的Modbus单元ID一般是1但也有特殊设置过的最好在官方文档或调试工具里确认。连接超时我设5秒太短的话设备启动阶段会连不上太长的话故障时程序会卡很久。采集程序里封装一个逆变器连接类用长连接方式不要每次读取都创建新连接# inverter.py from pymodbus.client import ModbusTcpClient class HuaweiInverter: def __init__(self, host, port, unit1, timeout5): self.client ModbusTcpClient(host, portport, timeouttimeout) self.unit unit def connect(self): if self.client.connected: return True return self.client.connect() def close(self): self.client.close() def read_holding_registers(self, address, count): rr self.client.read_holding_registers(addressaddress, countcount, unitself.unit) if rr.isError(): raise RuntimeError(fModbus read error: {rr}) return rr.registers3.3 寄存器批量读取与字段解析华为的Modbus寄存器映射表在官方《SUN2000-XXXKTL Modbus协议》文档里有完整定义不同固件版本略有差异我这里不方便直接贴出绝对地址因为设备型号和固件版本不同地址表可能不一样。最稳妥的方式是先从官方文档拿到你要的那几个字段的寄存器地址然后用Modbus Poll这类工具连上去实测一遍确认数值范围合理后再写进配置文件。我建议把地址和字段映射关系单独放到配置里像这样runtime_registers: address: 32064 # 示例值请以官方文档为准 count: 40 fields: - name: active_power offset: 0 type: int16 scale: 1 - name: inverter_temp offset: 6 type: int16 scale: 0.1这样如果换了型号只需要改配置和字段表不用改解析代码。解析函数的核心逻辑是按偏移量和类型把原始寄存器数组拆成业务字段# parser.py import struct def parse_registers(registers, fields): data {} for f in fields: offset f[offset] ftype f[type] scale f.get(scale, 1) if ftype int16: raw registers[offset] if raw 0x8000: raw - 0x10000 # 转成有符号数 value raw * scale elif ftype uint32: raw (registers[offset] 16) | registers[offset 1] value raw * scale elif ftype float32: raw_bytes struct.pack(HH, registers[offset], registers[offset 1]) value struct.unpack(f, raw_bytes)[0] else: continue data[f[name]] round(value, 3) return data这个函数里我用了大端序处理这个很重要。有些设备字段是低字在前如果发现解析结果乱套把struct.pack里的换成或者把(registers[offset] 16) | registers[offset 1]换成(registers[offset 1] 16) | registers[offset]再验证一遍。3.4 MQTT发布与主循环MQTT发布我单独写了一个类# mqtt_publisher.py import json import paho.mqtt.client as mqtt class MqttPublisher: def __init__(self, broker, port, username, password): self.client mqtt.Client() if username: self.client.username_pw_set(username, password) self.client.on_connect self._on_connect self.client.connect(broker, port, keepalive60) self.client.loop_start() def _on_connect(self, client, userdata, flags, rc): if rc 0: print(MQTT connected) else: print(fMQTT connect failed: {rc}) def publish(self, topic, payload, retainFalse): data json.dumps(payload, ensure_asciiFalse) self.client.publish(topic, data, qos1, retainretain)主循环逻辑# main.py import time import yaml from inverter import HuaweiInverter from mqtt_publisher import MqttPublisher from parser import parse_registers with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) inv_cfg config[inverter] mqtt_cfg config[mqtt] inverter HuaweiInverter(inv_cfg[host], inv_cfg[port], inv_cfg[unit]) publisher MqttPublisher(mqtt_cfg[broker], mqtt_cfg[port], mqtt_cfg[username], mqtt_cfg[password]) sn SUN2000L-KTL-L1-1234 # 建议从设备信息寄存器读取 while True: try: if not inverter.connect(): time.sleep(5) continue registers inverter.read_holding_registers( inv_cfg[runtime_registers][address], inv_cfg[runtime_registers][count] ) data parse_registers(registers, inv_cfg[runtime_registers][fields]) data[timestamp] time.strftime(%Y-%m-%dT%H:%M:%S%z) topic f{mqtt_cfg[topic_prefix]}{sn}/telemetry publisher.publish(topic, data) except Exception as e: print(fpoll error: {e}) time.sleep(inv_cfg[poll_interval_sec])这个主循环看着简单实际部署时还需要考虑几个问题异常后重连间隔要递增、MQTT断了要检测、消息要加本地缓存。3.5 断线重连与本地缓存补发Modbus设备在弱电环境下偶尔超时是正常的关键是程序不能因为一次异常就挂掉。我的做法是记录连续失败次数失败越多重连间隔越大retry 0 while True: try: if not inverter.connect(): retry 1 time.sleep(min(30, retry * 5)) continue # 读数据 发布 ... retry 0 time.sleep(inv_cfg[poll_interval_sec]) except Exception: retry 1 time.sleep(min(30, retry * 5))MQTT Broker离线是另一个场景。paho默认在Broker不可用时会把消息丢掉这样数据库会缺一段数据。我在采集端加了一个SQLite本地缓存发布失败就写本地重连成功后补发。SQLite操作简单、单文件、适合这种低并发场景不需要额外起数据库服务。补发的逻辑大致是每次publish返回失败时把(topic, payload)插入本地表检测到MQTT恢复on_connect回调后读取未发送记录逐条重新发布发布成功再删除记录这个功能看起来不起眼但真遇到Broker重启或者断网几分钟你会发现它救了很多历史数据。4. 现场部署遇到的坑与排查过程4.1 问题一Modbus TCP连接频频超时现象是程序运行一会儿就报connection timeout然后又自己恢复。一开始我怀疑是逆变器性能问题后来排查发现是采集程序部署在Docker容器里而Docker默认使用bridge网络容器到宿主机局域网是经过了NAT的某些非标准端口或长连接会被iptables规则干扰。解决方法有两个一是启动容器时加network_mode: host让采集程序直接用宿主机网络访问局域网二是如果一定要用bridge在宿主机上写一条稳定的端口映射规则并确保防火墙没有清理长连接。我用的是host模式之后连接超时的问题再没出现过。另外要注意路由器或交换机上的端口隔离设置。有些带管理功能的交换机默认会隔离不同VLAN的广播域导致Modbus TCP回包能到但连接建立不稳定。如果Modbus直连正常、从采集服务器连就超时优先检查网络策略和防火墙。4.2 问题二读取出来的功率值明显不对这个坑我印象最深刻。读出来active_power是58432换算成瓦特显然不合理换另一台设备读甚至出现过负值。排查链路是这样的先用Modbus Poll手动读同一个寄存器发现Poll显示的原始值是0xE440确实是58432。对照官方文档发现该字段的寄存器是int16类型不是uint16。0xE440在无符号数里是58432但如果按有符号数解释0xE440的高位为1是负数实际是-7104。再结合该字段的系数0.1就是-710.4这就不对了。随后发现我连错了字段这个寄存器并不是active_power而是另一个温度补偿参数。真正有用的寄存器在相邻的另一块区域。这个排查过程告诉我别迷信任何一份网上贴出来的寄存器地址表必须用自己的设备实测并且要把读到的值和现场能看到的现象比如功率、温度对照起来验证。程序里只要一个字段的解析方式错了后面所有关联数据都会错。4.3 问题三采集程序一启动官方App就掉线程序能稳定读数据之后我发现一个奇怪现象只要我的采集程序连着逆变器官方App的实时数据刷新就会出现卡顿甚至提示设备离线。后来查资料、翻社区才搞明白部分SUN2000L-KTL-L1固件对Modbus TCP的并发连接数做了限制同一时间可能只允许一个或两个客户端连接。官方App或者Dongle本身会占用一条连接我的采集程序再挤进去就会导致有一条连接被踢掉。解决方式很直接采集程序长连接独占同时不要开着官方调试工具或者多个采集实例。如果需要同时读数据最好用RS485扩展串口而不是多个TCP客户端同时连。还有一点程序不需要每次读取都新建连接长连接就OK了我一开始为了安全反复断开重连反而更容易触发连接数限制。4.4 问题四跨天发电量突然变大或回跳日发电量这个字段在0点时会出现归零或者跳变。一开始我以为自己解析错了盯着数据看了好几天才明白这是正常现象。逆变器内部的日发电量在每天凌晨会清零重新累计如果采集服务器和逆变器的时间基准不一致或者采集时间戳用了逆变器本地时间就会出现凌晨那一段数据乱跳。处理方式是让采集程序统一使用服务器时间作为timestamp发电机内部的日电量寄存器值只作为当日累计值记录跨天判断和统计由下游数据库完成不依赖逆变器时间。另外建议给逆变器做一次时间同步很多型号支持NTP时间设置。4.5 关于轮询频率和Broker性能的一些建议轮询太频繁除了给逆变器增加负载还会给MQTT Broker带来消息压力。我的经验是每秒最多一次轮询到头了5秒一次已经足够感受到实时。MQTT Broker如果只是单机服务百来个设备压力很小不需要刻意优化。但如果消息体里带了大字符串、历史累积字段建议拆成两个topic发布低频高价值数据单独走retain消息高频遥测数据用非retain消息这样订阅方可以按需订阅避免接收一堆用不上的数据。5. 数据接到MQTT之后可视化、告警与多站扩展5.1 订阅MQTT写入时序数据库MQTT数据到了Broker之后需要一个订阅程序把消息写入时序数据库。我用的TDengine也可以换成InfluxDB思路一样。TDengine建表语句CREATE STABLE solar_data ( ts TIMESTAMP, active_power FLOAT, daily_yield FLOAT, total_yield FLOAT, inverter_temp FLOAT, status INT ) TAGS (device_id BINARY(32));订阅程序负责从MQTT收消息、解析JSON、执行insertimport json import paho.mqtt.client as mqtt from taos import connect def on_message(client, userdata, msg): data json.loads(msg.payload) device_id msg.topic.split(/)[1] ts data[timestamp] userdata[cursor].execute( INSERT INTO solar_data USING solar_data TAGS (%s) VALUES (?, ?, ?, ?, ?, ?), [device_id, ts, data.get(active_power), data.get(daily_yield), data.get(total_yield), data.get(inverter_temp), data.get(status)] )当然可以直接用EMQX这类Broker内置的规则引擎把消息转发到数据库省一个订阅程序。我这边更倾向于写独立的订阅程序因为后续要做数据清洗和补发逻辑程序方式更灵活。5.2 Grafana看板与告警规则数据进时序数据库之后用Grafana做可视化是最快的。选好数据源、写一条SQL就能得到实时功率曲线、日发电量柱状图和温度趋势。告警我用的是Grafana Alerting。举两个实用规则连续15分钟active_power为0但当天总有光照说明可能存在故障停机inverter_temp超过80度触发高温告警MQTT本身也可以接轻量级告警程序订阅telemetry主题判断数据异常后发Webhook或者邮件。这个程序是独立消费者不影响采集链路所以出问题也不会拖垮数据采集。5.3 扩展到多台逆变器与多品牌设备这套架构最大的好处是方便扩展。多台华为逆变器只需要在配置里增加连接信息每个设备用一个独立线程采集发布到不同topic。其他品牌的逆变器只要支持Modbus也能用同一套解析框架接入差异只体现在寄存器字段映射表上。如果现场既有光伏又有电表和充电桩可以都用Modbus转MQTT这个思路最后所有设备的数据都汇聚到同一个Broker平台层只面对MQTT一个协议这是很标准的多设备数据整合方式。5.4 个人在实际落地中的一点体会整套系统从开始做到稳定运行花了我周末两天时间。起初最耗时的不是写代码而是调寄存器解析。建议大家动手前一定先用Modbus工具把要读的寄存器手动验证一遍把字段清单、类型、系数整理成表格再写代码能省掉大量时间。关于MQTT Broker个人用的话直接Mosquitto就很稳追求管理界面可以用EMQX。我中间还踩了一个小坑Broker的keepalive参数没设置好低频数据场景下连接会被误判断开后来把keepalive调成60秒并加了定时ping才稳定下来。另外建议采集程序里所有日志都带时间戳和来源设备标识排查问题时能直接定位是哪台设备、哪一步出错尤其是多设备场景日志不规范化会非常痛苦。本文还有配套的精品资源点击获取