如何不使用MQTT连接器，通过Python连接ThingsBoard Cloud？

ThingsBoard除了MQTT协议，还提供了完整的REST API接口，完全可以用Python通过HTTP请求实现设备连接、数据上报、遥测查询等核心功能，不需要依赖任何MQTT库。以下是具体实现方案：

1. 前置准备 #

• 你的ThingsBoard Cloud实例地址（比如https://demo.thingsboard.io，替换成你自己的实例地址）
• 设备的访问令牌（在ThingsBoard设备详情页的「凭证」标签下获取，或通过租户令牌动态生成）
• 安装Python的requests库：pip install requests

2. 核心功能代码实现 #

上报遥测数据 #

这是设备端最常用的场景，直接通过POST请求把数据发送到ThingsBoard：

import requests
import json
import time

# 配置项，替换成你自己的信息
THINGSBOARD_URL = "https://your-thingsboard-instance.com"
DEVICE_TOKEN = "your-device-access-token-here"

def send_telemetry(telemetry_data):
    """上报遥测数据到ThingsBoard"""
    api_url = f"{THINGSBOARD_URL}/api/v1/{DEVICE_TOKEN}/telemetry"
    headers = {"Content-Type": "application/json"}
    
    try:
        response = requests.post(api_url, data=json.dumps(telemetry_data), headers=headers, timeout=10)
        response.raise_for_status()  # 主动抛出HTTP错误
        print("遥测数据上报成功")
    except requests.exceptions.RequestException as e:
        print(f"上报失败: {str(e)}")

# 测试代码
if __name__ == "__main__":
    # 上报单条实时数据
    single_data = {
        "temperature": 26.3,
        "humidity": 58.7,
        "timestamp": int(time.time() * 1000)  # 可选，毫秒级时间戳，不填则用服务器时间
    }
    send_telemetry(single_data)

    # 批量上报多条历史数据
    batch_data = {
        "values": {
            "temperature": [
                {"ts": 1718000000000, "value": 25.9},
                {"ts": 1718000600000, "value": 26.1}
            ],
            "humidity": [
                {"ts": 1718000000000, "value": 58.5},
                {"ts": 1718000600000, "value": 58.6}
            ]
        }
    }
    send_telemetry(batch_data)

动态注册设备（如需新增设备） #

如果需要在代码中创建新设备，用租户令牌调用设备注册接口：

def register_device(tenant_token, device_name):
    """通过租户令牌注册新设备"""
    api_url = f"{THINGSBOARD_URL}/api/device"
    headers = {
        "Content-Type": "application/json",
        "X-Authorization": f"Bearer {tenant_token}"
    }
    payload = {
        "name": device_name,
        "type": "sensor"  # 自定义设备类型，比如sensor、gateway等
    }

    try:
        response = requests.post(api_url, data=json.dumps(payload), headers=headers, timeout=10)
        response.raise_for_status()
        device_info = response.json()
        print(f"设备注册成功，ID: {device_info['id']['id']}")

        # 获取新设备的访问令牌
        token_url = f"{THINGSBOARD_URL}/api/device/{device_info['id']['id']}/credentials"
        token_response = requests.get(token_url, headers=headers, timeout=10)
        token_response.raise_for_status()
        device_token = token_response.json()["credentialsId"]
        print(f"设备访问令牌: {device_token}")
        return device_info, device_token
    except requests.exceptions.RequestException as e:
        print(f"设备注册失败: {str(e)}")
        return None, None

# 调用示例（需替换租户令牌）
# tenant_token = "your-tenant-admin-token"
# register_device(tenant_token, "new-sensor-device-001")

查询设备遥测数据 #

通过GET请求获取指定时间范围的设备历史遥测：

def get_telemetry(device_token, data_keys, start_timestamp, end_timestamp):
    """查询指定时间范围的遥测数据"""
    api_url = f"{THINGSBOARD_URL}/api/v1/{DEVICE_TOKEN}/telemetry/values/timeseries"
    params = {
        "keys": ",".join(data_keys),
        "startTs": start_timestamp,
        "endTs": end_timestamp
    }

    try:
        response = requests.get(api_url, params=params, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"获取遥测失败: {str(e)}")
        return None

# 调用示例
# start_ts = 1717000000000  # 起始时间戳（毫秒）
# end_ts = 1718000000000    # 结束时间戳（毫秒）
# telemetry = get_telemetry(DEVICE_TOKEN, ["temperature", "humidity"], start_ts, end_ts)
# print(json.dumps(telemetry, indent=2))

3. 注意事项 #

• 确保你的设备/服务器能访问ThingsBoard Cloud的HTTPS端口（默认443），防火墙需放行对应端口
• 访问令牌、租户令牌属于敏感信息，不要硬编码在代码中，建议通过环境变量读取
• 可添加重试机制处理网络波动，比如用tenacity库实现自动重试逻辑
• 更多API接口细节可参考ThingsBoard官方REST API文档

内容的提问来源于stack exchange，提问作者ts tejaswi14