当前位置：首页 > java >正文

QuecPython+aLiYun：快速连接阿里云IoT平台

java 2025/5/1 9:27:52

概述

随着物联网技术的快速发展，云平台已成为连接物理世界与数字世界的核心枢纽。QuecPython 深度整合阿里云物联网平台（aLiYun），推出"终端+云端"一体化开发解决方案，为物联网开发者提供开箱即用的云端连接能力。

核心优势

极简开发接口：通过 Python 化的 API 设计，将复杂的 MQTT 协议栈、安全认证、数据编解码等底层操作封装为简洁的方法调用，无需复杂的底层驱动编写。
文档齐全：每个 API 接口都有详细的使用介绍，跟随文档即可轻松进行开发设计。

功能概述

主要介绍如何使用 QuecPython 系列模组连接阿里云物联网平台，并进行数据的订阅与发布，以下为大致连接流程：

初始化阿里云 IoT 平台
设置相关功能和回调
订阅发布功能
连接启停

初始化阿里云 IoT 平台

该方法用于配置阿里 IoT 平台物联网套件的产品信息和设备信息，并返回 aLiYun 连接对象。

aLiYun(productKey, productSecret, DeviceName, DeviceSecret, MqttServer)

API 详情请参考 aLiYun。

设置相关功能和回调

MQTT 参数设置

该方法用于设置 MQTT 数据通道参数。

aLiYun.setMqtt(clientID, clean_session, keepAlive=300,reconn=True)

API 详情请参考 aLiYun.setMqtt。

回调注册

该方法用于注册回调函数。

aLiYun.setCallback(callback)

API 详情请参考 aLiYun.setCallback。

订阅发布

发布

该方法用于发布消息。

aLiYun.publish(topic,msg, qos=0)

API 详情请参考 aLiYun.publish。

启停功能

运行连接

该方法用于运行连接。

aLiYun.start()

API 详情请参考 aLiYun.start。

关闭连接

该方法用于关闭连接。

aLiYun.disconnect()

API 详情请参考 aLiYun.disconnect。

应用案例

import log
import utime
import checkNet
from aLiYun import aLiYun'''
下面两个全局变量是必须有的，用户可以根据自己的实际项目修改下面两个全局变量的值
'''
PROJECT_NAME = "QuecPython_AliYin_example"
PROJECT_VERSION = "1.0.0"
checknet = checkNet.CheckNetwork(PROJECT_NAME, PROJECT_VERSION)# 设置日志输出级别
log.basicConfig(level=log.INFO)
aliYun_log = log.getLogger("ALiYun")productKey = ""  # 产品标识(参照阿里 IoT 平台应用开发指导)
productSecret = None  # 产品密钥（使用一机一密认证时此参数传入None，参照阿里 IoT 平台应用开发指导)
DeviceName = ""  # 设备名称(参照阿里 IoT 平台应用开发指导)
DeviceSecret = ""  # 设备密钥（使用一型一密认证此参数传入None，免预注册暂不支持，需先在 IoT 平台创建设备，参照阿里 IoT 平台应用开发指导)state = 5# 回调函数
def sub_cb(topic, msg):global statealiYun_log.info("Subscribe Recv: Topic={},Msg={}".format(topic.decode(), msg.decode()))state -= 1if __name__ == '__main__':stagecode, subcode = checknet.wait_network_connected(30)if stagecode == 3 and subcode == 1:aliYun_log.info('Network connection successful!')# 创建aliyun连接对象ali = aLiYun(productKey, productSecret, DeviceName, DeviceSecret)# 设置mqtt连接属性clientID = ""  # 自定义字符（不超过64）ali.setMqtt(clientID, clean_session=False, keepAlive=300)# 设置回调函数ali.setCallback(sub_cb)topic = ""  # IoT 平台自定义或自拥有的Topic# 订阅主题ali.subscribe(topic)# 发布消息ali.publish(topic, "hello world")# 运行ali.start()while 1:if state:passelse:ali.disconnect()breakelse:aliYun_log.info('Network connection failed! stagecode = {}, subcode = {}'.format(stagecode, subcode))