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

一、测试环境介绍

型号：IG502

固件版本：V2.2.0

SDK版本：1.4.5

APP版本：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_name=cloudName)

示例

期望格式：

{ "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_name=cloudName)

#实际输出 { "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_name=cloudName)

#实际输出 { "ID" : "001", "ts" : 1777433163, "data" : { "WSDJ" : { "WD" : 24.9, "SD" : 44.81 } } }

3、告警数据

（该部分不常用，这里不单独做解析说明）

三、总结

在这个脚本编写中可以实现多种格式对接，不仅限于json数据，还有加密，二进制字节等，并且设备支持安装第三方春Python库。可以实现高效编程，

如果需要更多帮助或者支持，请留言或者私信我。我看到后会尽快提供支持。