IG系列网关和EC系列边缘计算机DSA数采程序中，MQTT发布消息脚本编写说明

一、测试环境介绍型号IG502固件版本V2.2.0SDK版本1.4.5APP版本V3.4.0二、来源数据获取按照官方手册完成数据采集配置和连接云平台的设置在消息管理里面新增发布。新增发布后有一些参数需要填写如下图上介绍参数的意义。在上图的来源数据里面我们可以看出三个数据源类型1、测点数据测点数据是最原始的按照分组的方式对数据进行打包JSOM然后传递到平台。使用MQTT.fx软件连接MQTT平台然后订阅这个主题可以看到发上来的数据如下# 原始数据格式 { timestamp : 1777430120, # 秒级时间戳 timestampMsec : 1777430120167, #毫秒级时间戳 group_name : default, # 分组名称 values : { #无定义 WSDJ : { # 控制器名称 SD : { # 变量名称 raw_data : 44.82, # 变量值 timestamp : 1777430117, #采集变量的时间戳秒级 status : 1, #变量采集状态 timestampMsec : 1777430117769 #采集变量时间戳毫秒级 }, WD : { raw_data : 24.79, timestamp : 1777430117, status : 1, timestampMsec : 1777430117769 } }, IO : { DI3 : { raw_data : 0, timestamp : 1777430118, status : 1, timestampMsec : 1777430118339 }, DI2 : { raw_data : 0, timestamp : 1777430118, status : 1, timestampMsec : 1777430118339 } } } }以上为原始变量是发布程序通过默认脚本发出来的数据# 默认脚本 import json from common.Logger import logger from quickfaas.remotebus import publish def main(message, wizard_api, cloudName): logger.debug(publish topic:%s, payload: %s, cloudName: %s % (__topic__, message, cloudName)) publish(__topic__, json.dumps(message), __qos__, cloud_namecloudName)示例期望格式{ ID:001, #固定字段 ts:XXXXXXXXX, # 秒级时间戳 data:{ d_name:{ #控制器名称 name:34.5 #name-变量名称34.5-变量值 } } }# 示例脚本 import json from common.Logger import logger from quickfaas.remotebus import publish def main(message, wizard_api, cloudName): msg{ ID:001, ts:message[timestamp], data:{} } for key,value in message[values].items(): msg[data][key]{} for key1,value1 in value.items(): msg[data][key][key1]value1[raw_data] logger.debug(publish topic:%s, payload: %s, cloudName: %s % (__topic__, msg, cloudName)) publish(__topic__, json.dumps(msg), __qos__, cloud_namecloudName)#实际输出 { ID : 001, ts : 1777431854, data : { WSDJ : { SD : 44.65, WD : 24.9 }, IO : { DI3 : 0, DI2 : 0 } } }2、控制器数据测点数据是按照单一控制器的数据进行打包JSOM然后传递到平台。使用MQTT.fx软件连接MQTT平台然后订阅这个主题可以看到发上来的数据如下#原始数据格式 { timestamp : 1777432145, #时间戳秒级 timestampMsec : 1777432145467,#时间戳毫秒级 ctrlName : WSDJ, # 控制器名称 measures : [ { #固定字段 ctrlName : WSDJ, # 控制器名称 name : WD, #变量名称 health : 1, #采集状态 timestamp : 1777432145, #变量采集时间戳秒级 timestampMsec : 1777432145465, #变量采集时间戳毫秒级 value : 24.9 #变量值 }, { ctrlName : WSDJ, name : SD, health : 1, timestamp : 1777432145, timestampMsec : 1777432145465, value : 44.69 } ] }#期望格式 { ID:001, #固定字段 ts:XXXXXXXXX, # 秒级时间戳 data:{ d_name:{ #控制器名称 name:34.5 #name-变量名称34.5-变量值 } } }# 示例脚本 import json from common.Logger import logger from quickfaas.remotebus import publish def main(message, wizard_api, cloudName): msg{ ID:001, ts:message[timestamp], data:{message[ctrlName]:{}} } for key in message[measures]: msg[data][key[ctrlName]][key[name]]key[value] logger.debug(publish topic:%s, payload: %s, cloudName: %s % (__topic__, msg, cloudName)) publish(__topic__, json.dumps(msg), __qos__, cloud_namecloudName)#实际输出 { ID : 001, ts : 1777433163, data : { WSDJ : { WD : 24.9, SD : 44.81 } } }3、告警数据该部分不常用这里不单独做解析说明三、总结在这个脚本编写中可以实现多种格式对接不仅限于json数据还有加密二进制字节等并且设备支持安装第三方春Python库。可以实现高效编程如果需要更多帮助或者支持请留言或者私信我。我看到后会尽快提供支持。