开发笔记:EC600QuecPython接入第三方MQTT服务器，以阿里云物联网平台为例，可替换为自己创建的MQTT服务器

EC600 QuecPython 自带Aliyun、腾讯云物联网平台的接入库&＃xff0c;但重口难调&＃xff0c;产品项目中更多的是接入自己搭建的第三方服务器。

MQTT协议接入不同的服务器&＃xff0c;最大的区别在于connect报文。

connect报文的载荷中&＃xff0c;包含了设备的登录账号&＃xff0c;而每个平台都有自己的设备管理方式&＃xff0c;也就造成编码方式的不一致。不过好在MQTT也规定了 clientId、userName、passWord 这几个关键载荷&＃xff0c;用户只要保证这几个部分与平台匹配即可连接。

使用 EC600 QuecPython 接入第三方MQTT服务器&＃xff0c;需要使用 umqtt 模块。

这里为方便测试&＃xff0c;使用阿里云账号&＃xff0c;不过不用 aliyun 的MQTT接入库&＃xff0c;直接用 umqtt 模块。

• EC600 umqtt 使用帮助&＃xff1a;QuecPython - 在线API文档
• 本篇demo使用的测试设备账号、以及密码的合成方式&＃xff1a;网络调试助手接入阿里云MQTT物联网平台&＃xff0c;逐字节讲解各字段合成方式
• 阿里云物联网平台的设备添加&＃xff1a;阿里云物联网平台注册、添加产品、设备

上述的设备账号会一直保留&＃xff0c;方便测试&＃xff0c;各位可放心使用。

接入第三方MQTT服务器&＃xff0c;需要根据自己的设备及服务器&＃xff0c;自行计算并替换&＃xff1a;clientId、userName、passWord、brokerUrl 以及 topic。&＃xff08;阿里云的 clientI、userName、passWord 计算方式见上述链接&＃xff09;

Demo代码&＃xff1a;

from machine import Pin
import log
import checkNet
from umqtt import MQTTClient
# 使用提前计算好的密码&＃xff0c;可以登录连接。hamc库暂时使用有问题&＃xff0c;待官方的固件库更新
# 使用umqtt库接入阿里云物联网平台&＃xff0c;模拟接入第三方MQTT服务器进行测试
clientId &＃61; &＃39;co_0001|securemode&＃61;3,signmethod&＃61;hmacsha1|&＃39;
userName &＃61;&＃39;co_0001&a1wFylTxYeD&＃39;
passWord &＃61; &＃39;e782b5e55b37655c27812a60c307b0a7575d8f6d&＃39;
brokerUrl &＃61; &＃39;a1wFylTxYeD.iot-as-mqtt.cn-shanghai.aliyuncs.com&＃39;
topic_post &＃61; b&＃39;/sys/a1wFylTxYeD/co_0001/thing/event/property/post&＃39;
topic_set &＃61; b&＃39;/sys/a1wFylTxYeD/co_0001/thing/service/property/set&＃39;
PROJECT_NAME &＃61; "MQTT_example"
PROJECT_VERSION &＃61; "1.0.0"
# 检查网络状态&＃xff0c;创建checkNet对象
checknet &＃61; checkNet.CheckNetwork(PROJECT_NAME, PROJECT_VERSION)
# 设置日志输出级别&＃xff0c;仅输出等级高于INFO的日志结果
log.basicConfig(level&＃61;log.INFO)
log &＃61; log.getLogger("MQTT")
state &＃61; 0
# V1.2 Demo板上的led - D6
blink &＃61; False
led &＃61; Pin(Pin.GPIO24, Pin.OUT, Pin.PULL_DISABLE, 0)
# 设置MQTT接收消息回调
def sub_cb(topic, msg):
global state
log.info("Subscribe Recv: Topic&＃61;{},Msg&＃61;{}".format(topic.decode(), msg.decode()))
state &＃61; 1
# 收到下发的消息时&＃xff0c;LED亮灭状态会发生变化
global blink
blink &＃61; bool(1-blink)
led.write(blink)
if __name__ &＃61;&＃61; &＃39;__main__&＃39;:
# 上电运行后调试输出 项目名称、项目版本号、固件版本号、开机原因、SIM卡状态
checknet.poweron_print_once()
print(&＃39;Hardware PowerON!&＃39;)
# 开机后LED亮起&＃xff0c;指示状态。
led.write(1)
# 阻塞等待网络就绪&＃xff0c;超时等待30s
stagecode, subcode &＃61; checknet.wait_network_connected(30)
if stagecode &＃61;&＃61; 3 and subcode &＃61;&＃61; 1:
# 网络准备就绪&＃xff0c;开始执行用户代码
log.info(&＃39;Network connection successful! Then connect to Aliyun MQTT&＃39;)
# 指示LED灭掉&＃xff0c;提示网络连接正常。
led.write(0)
print(&＃39;username:&＃39;, userName)
print(&＃39;password:&＃39;, passWord)
# 创建一个mqtt实例
c &＃61; MQTTClient(
client_id&＃61;clientId,
server&＃61;brokerUrl,
port&＃61;1883,
user&＃61;userName,
password&＃61;passWord,
keepalive&＃61;60)
# 设置消息回调
c.set_callback(sub_cb)
# 建立连接
try:
c.connect()
except Exception as e:
print(&＃39;!!!,e&＃61;%s&＃39; % e)
# 正常连接后&＃xff0c;输出消息
log.info(&＃39;Aliyun MQTT connected , Then subscribe topic&＃39;)
# 订阅主题
c.subscribe(topic_post)
log.info(&＃39;subscribe topic: %s&＃39; % topic_post)
c.subscribe(topic_set)
log.info(&＃39;subscribe topic: %s&＃39; % topic_set)
# 发布消息
c.publish(topic_post, &＃39;test publish&＃39;)
log.info(&＃39;Publish topic: %s, Msg: %s&＃39; % (topic_post, &＃39;test publish&＃39;))
log.info(&＃39;listen&＃39;)
while True:
c.wait_msg() # 阻塞函数&＃xff0c;监听消息
else:
log.info(&＃39;Network connection failed! stagecode &＃61; {}, subcode &＃61; {}&＃39;.format(stagecode, subcode))


SIM插入、4G网络正常&＃xff0c;正常运行的现象。

log会显示正常连接、订阅topic 和 publish。最后在 publish 一条消息后&＃xff0c;一直处于监听状态。

• 连接后能在管理页面看到设备在线。
  
• 测试下发控制消息
  
• 观察 V1.2 Demo板上的led - D6&＃xff0c;收到下发的消息时的同时&＃xff0c;LED亮灭状态也会发生变化。&＃xff08;led亮灭是脚本中自己加入的代码&＃xff09;
  

• hmac 暂时有问题&＃xff0c;需要等待固件更新&＃xff0c;才可以让设备自行合成 passWord。
• 接入第三方MQTT服务器&＃xff0c;需要自行计算clientId、userName、passWord、brokerUrl 以及 topic&＃xff0c;并替换。
• umqtt的异常提醒很坑&＃xff0c;如设备账号填错&＃xff0c;会报错 bytes index out of range。而非重新连接。
• 代码中如有 while&＃xff0c;会导致与上位机的通信阻塞。重新下载脚本时&＃xff0c;需要断开连接、复位、再重新连接。&＃xff08;如带有while 的脚本文件名为 main.py&＃xff0c;则会上电自运行&＃xff0c;只能通过重刷固件解决&＃xff09;