### 使用 Python 从 OneNet 平台获取数据 要使用 Python 从 OneNet 平台获取数据，可以通过 HTTP API 或 MQTT 协议实现。以下是两种方法的具体说明： #### 方法一：通过 HTTP API 获取数据 OneNet 提供了 RESTful 风格的接口来查询设备数据。可以利用 `requests` 库发起请求并解析返回的结果。 ##### 示例代码 ```python import requests def get_device_data(device_id, datastream_id, api_key): url = f"http://api.heclouds.com/devices/{device_id}/datastreams/{datastream_id}" headers = { "api-key": api_key } response = requests.get(url, headers=headers) if response.status_code == 200: return response.json() else: return {"error": f"Failed to fetch data. Status Code: {response.status_code}"} # 替换为实际参数 device_id = "你的设备ID" datastream_id = "你的数据流ID" api_key = "你的API密钥" result = get_device_data(device_id, datastream_id, api_key) print(result) ``` 上述代码展示了如何通过设备 ID 和数据流 ID 查询特定数据[^1]。需要注意的是，`api-key` 是访问 OneNet 的授权凭证，需妥善保管。 --- #### 方法二：通过 MQTT 订阅实时数据 如果需要接收设备的实时数据更新，推荐使用 MQTT 协议订阅相关主题。以下是一个简单的示例： ##### 示例代码 ```python import paho.mqtt.client as mqtt def on_message(client, userdata, msg): print(f"Received message: {msg.payload.decode()} from topic: {msg.topic}") client = mqtt.Client() # 设置回调函数 client.on_message = on_message # 连接到 OneNet MQTT Broker broker_address = "mqtt.heclouds.com" port = 6002 username = "你的设备ID" password = "你的密码或Token" client.username_pw_set(username=username, password=password) try: client.connect(broker_address, port, keepalive=60) except Exception as e: print(f"Connection failed: {e}") else: # 订阅指定的主题 subscribe_topic = "{device_id}/{datastream_id}".format( device_id="你的设备ID", datastream_id="你的数据流ID" ) client.subscribe(subscribe_topic) client.loop_forever() ``` 此代码实现了与 OneNet 的 MQTT Broker 进行连接，并订阅某个具体的数据流主题以监听消息推送[^4]。 --- ### 注意事项 - **权限管理**：无论是通过 HTTP 请求还是 MQTT 接入，都需要提供有效的认证信息（如 `api-key` 或 Token），这些信息应严格保密。 - **错误处理**：在开发过程中建议加入异常捕获机制，以便更好地应对网络中断或其他不可预见的情况。 - **频率限制**：频繁调用 API 可能触发速率限制，请查阅官方文档确认配额详情[^3]。 ---