🚀 ThingsCloud MicroPython SDK

English

ThingsCloud 物联网平台的 MicroPython SDK，帮助物联网设备开发者快速将 ESP32/ESP8266/STM32/树莓派等设备接入 ThingsCloud 平台。

✨ 特性

• 📡 轻量级 MQTT 客户端，专为 ThingsCloud 优化
• 🔧 简单易用的 API，无需了解 MQTT 细节
• 📊 支持属性上报、命令接收、事件上报等完整功能
• 🔐 支持两种认证模式：直接连接和一型一密
• 📁 完整的示例代码，快速上手

🖥 支持的硬件

本 SDK 支持所有运行 MicroPython 的硬件平台，包括但不限于：

芯片/开发板	说明
ESP32	乐鑫 ESP32 系列芯片，支持 WiFi 和蓝牙
ESP8266	乐鑫 ESP8266 系列芯片，低成本 WiFi 方案
STM32	意法半导体 STM32 系列微控制器
RP2040	树莓派 Pico 系列（需配合网络模块）
RPi Pico W	树莓派 Pico W（内置 WiFi）
Pyboard	MicroPython 官方开发板

💡 提示： 使用非 ESP32/ESP8266 平台时，需要自行配置网络连接模块。

📥 安装

1. 将 thingscloud_mqtt_lib.py 复制到你的 MicroPython 设备中
2. 根据需要复制示例文件到设备中

使用 Thonny IDE 或 uPyCraft IDE 上传文件。

🚀 快速开始

1. 获取设备证书

在 ThingsCloud 控制台创建设备后，在设备详情页的「连接」页面获取：

• AccessToken: 设备访问令牌（MQTT 用户名）
• ProjectKey: 项目密钥（MQTT 密码）
• MQTT 主机: MQTT 服务器域名

2. 认证模式

SDK 支持两种认证模式：

模式一：直接连接（推荐用于开发调试）

使用 AccessToken 直接连接，适合开发调试阶段：

from thingscloud_mqtt_lib import ThingsCloudMQTTClient
import time

# 创建客户端
client = ThingsCloudMQTTClient(
    project_key="YOUR_PROJECT_KEY",
    server="YOUR_MQTT_HOST",
    port=1883,
    access_token="YOUR_ACCESS_TOKEN"
)

# 连接到 ThingsCloud
client.connect()

# 上报属性
client.report_attributes({
    "temperature": 25.5,
    "humidity": 60
})

# 保持连接并处理消息
while True:
    client.check_msg()
    time.sleep(1)

模式二：一型一密（推荐用于量产部署）

使用 DeviceKey 和 TypeKey 动态获取 AccessToken，适合量产部署：

from thingscloud_mqtt_lib import ThingsCloudMQTTClient

# 创建客户端（不需要 access_token）
client = ThingsCloudMQTTClient(
    project_key="YOUR_PROJECT_KEY",
    server="YOUR_MQTT_HOST",
    port=1883
)

# 使用一型一密模式连接
# 设备会自动在平台创建并获取 AccessToken
client.connect_with_device_cert(
    device_key="your_device_key",       # 设备唯一标识
    type_key="your_type_key",           # 设备类型标识
    api_endpoint="your_api_endpoint"
)

# 连接成功后即可正常使用
client.report_attributes({"temperature": 25.5})

3. 上报属性

设备连接后，可以随时向云端上报属性数据：

# 上报单个属性
client.report_attributes({"temperature": 25.5})

# 上报多个属性
client.report_attributes({
    "temperature": 25.5,
    "humidity": 60,
    "status": "online"
})

# 设置上报响应回调（可选）
def on_attributes_response(response):
    if response.get("result") == 1:
        print("上报成功")
    else:
        print("上报失败:", response)

client.on_attributes_response(on_attributes_response)

4. 接收属性下发

设备可以接收云端主动下发的属性，用于配置更新或远程控制：

def on_attributes_push(attributes):
    print("收到下发属性:", attributes)

    # 处理下发的属性
    if "switch" in attributes:
        switch_state = attributes["switch"]
        print(f"设置开关状态: {switch_state}")
        # 执行开关控制逻辑...

    if "report_interval" in attributes:
        interval = attributes["report_interval"]
        print(f"更新上报间隔: {interval}秒")

# 设置属性下发回调
client.on_attributes_push(on_attributes_push)

5. 接收命令

设备可以接收云端下发的命令并执行，然后回复执行结果:

def on_command(command):
    method = command.get("method")
    params = command.get("params")
    cmd_id = command.get("id")

    print(f"收到命令: {method}, 参数: {params}")

    # 执行命令逻辑...

    # 回复命令
    client.reply_command(method, {"status": "success"}, reply_id=cmd_id)

# 设置命令回调
client.on_command_send(on_command)

📂 示例

目录结构

examples/
├── MQTT_Basics/                    # MQTT 基础示例
│   ├── 01_direct_connect/          # 直接连接
│   ├── 02_dynamic_cert/            # 一型一密连接
│   ├── 03_report_attributes/       # 上报属性
│   ├── 04_report_events/           # 上报事件
│   ├── 05_receive_pushed_attributes/ # 接收下发属性
│   ├── 06_receive_commands/        # 接收下发命令
│   ├── 07_custom_data_stream/      # 自定义数据流
│   └── README.md
└── IoT_Tutorials/                  # IoT 应用教程
    ├── dht_sensor/                 # DHT 温湿度传感器
    └── relay_control/              # 继电器控制

MQTT_Basics - MQTT 基础示例

涵盖 MQTT 通信的各种场景：

• 直接连接和一型一密连接
• 属性上报和事件上报
• 接收云端下发的属性和命令
• 自定义数据流通信

IoT_Tutorials - IoT 应用教程

DHT11/DHT22 温湿度传感器

• 定时读取 DHT11/DHT22 传感器数据
• 自动上报温度和湿度到 ThingsCloud
• 支持配置上报间隔
• 支持两种认证模式

继电器控制

• 接收 ThingsCloud 下发的开关命令
• 控制继电器开关
• 自动上报开关状态
• 支持心跳保活

📖 API 参考

ThingsCloudMQTTClient

初始化

client = ThingsCloudMQTTClient(
    project_key,      # 项目 ProjectKey（必需）
    server,           # MQTT 服务器地址（必需）
    port=1883,        # 端口号
    access_token=None # 设备 AccessToken（可选，用于一型一密模式）
)

连接方法

# 模式一：直接连接（需要先设置 access_token）
client.connect()

# 模式二：一型一密模式
client.connect_with_device_cert(
    device_key="your_device_type",
    type_key="your_type_key",
    api_endpoint="your_api_endpoint"
)

client.disconnect()        # 断开连接
client.is_connected()      # 检查连接状态
client.check_msg()         # 检查消息（非阻塞）
client.wait_msg()          # 等待消息（阻塞）

属性相关

# 上报属性
client.report_attributes({"temperature": 25.5})

# 获取属性
client.get_attributes(["temperature"])  # 获取指定属性
client.get_attributes()                  # 获取所有属性

# 设置属性推送回调
client.on_attributes_push(callback)
# callback(attributes_dict)

# 设置属性上报响应回调
client.on_attributes_response(callback)
# callback(response_dict)

命令相关

# 设置命令接收回调
client.on_command_send(callback)
# callback(command_dict)
# command_dict: {"method": "...", "params": {...}, "id": ...}

# 回复命令
client.reply_command(method, params, reply_id)

事件相关

# 上报事件
client.report_event("alert", {"level": "warning"})

证书获取回调

# 设置证书获取结果回调（一型一密模式）
client.on_fetch_cert(callback)
# callback(success, device_info)

📡 MQTT 主题说明

• MQTT 接入文档

❓ 常见问题

Q: 连接失败怎么办？

A: 检查以下几点：

1. WiFi 是否正常连接
2. AccessToken 和 ProjectKey 是否正确
3. 服务器地址和端口是否正确
4. 设备是否已在平台创建

Q: 一型一密模式如何使用？

A: 一型一密模式适合量产设备，只需配置 ProjectKey、TypeKey 和 DeviceKey，设备会自动获取 AccessToken。需要在控制台开启设备的动态注册功能。

Q: 如何调试？

A: 可以使用 ThingsCloud 控制台的设备调试功能，查看消息收发情况。

📄 许可证

MIT License

🔗 相关链接

• ThingsCloud 官网
• ThingsCloud 文档
• MQTT 接入文档