Python定时生成测试数据发送至Azure IoT Hub及设备控制指导请求

一、修改Python代码实现每6分钟定时发送测试数据 #

无需修改Azure IoT Python SDK本身，直接在业务代码外层添加定时逻辑即可，以下是两种实用实现方式：

方法1：用基础time模块 #

import time
from azure.iot.device import IoTHubDeviceClient, Message

# 替换为你的设备连接字符串
CONNECTION_STRING = "你的设备连接字符串"

def send_test_message(client):
    # 生成模拟测试数据，可按需调整字段
    test_data = {
        "temperature": 25.5 + (time.time() % 5),
        "humidity": 60.2 + (time.time() % 3),
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
    }
    message = Message(str(test_data))
    print(f"发送消息: {test_data}")
    client.send_message(message)

def main():
    client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING)
    client.connect()
    
    try:
        while True:
            send_test_message(client)
            # 每6分钟发送一次（6*60=360秒）
            time.sleep(360)
    except KeyboardInterrupt:
        print("停止客户端...")
        client.disconnect()

if __name__ == "__main__":
    main()

方法2：用schedule库（更灵活的任务调度） #

先安装依赖：pip install schedule

import schedule
import time
from azure.iot.device import IoTHubDeviceClient, Message

CONNECTION_STRING = "你的设备连接字符串"
client = None

def send_test_message():
    global client
    test_data = {
        "temperature": 25.5 + (time.time() % 5),
        "humidity": 60.2 + (time.time() % 3),
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
    }
    message = Message(str(test_data))
    print(f"发送消息: {test_data}")
    client.send_message(message)

def main():
    global client
    client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING)
    client.connect()
    
    # 配置每6分钟执行一次发送任务
    schedule.every(6).minutes.do(send_test_message)
    
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        print("停止客户端...")
        client.disconnect()

if __name__ == "__main__":
    main()

二、数据路由与分析监控 #

利用Azure IoT Hub内置的路由功能，可将设备数据转发至不同服务做存储或分析，同时支持实时监控：

1. 配置IoT Hub消息路由 #

• 登录Azure门户，进入你的IoT Hub，左侧菜单选择消息路由
• 点击添加，设置路由名称，数据源选择设备遥测消息
• 选择终点类型：比如选存储容器（存储历史数据用于离线分析），或事件中心（对接实时分析工具）
• 可选设置路由查询（如仅转发温度超过30的数据：$body.temperature > 30）
• 保存后，设备消息会自动转发至指定终点

2. 实时监控数据 #

使用Azure IoT Explorer工具：

• 连接你的IoT Hub后，选择设备，找到目标设备进入遥测页面，即可实时接收设备发送的消息
• 也可在IoT Hub的监控菜单查看设备连接状态、消息发送统计等数据

三、用云数据控制ESP8266/Arduino设备 #

Azure IoT Hub支持直接方法（即时控制）和设备孪生（配置同步）两种控制方式：

1. 直接方法（即时控制示例：远程开关LED） #

• 云侧Python代码（调用设备方法）：

from azure.iot.hub import IoTHubRegistryManager

# 替换为你的IoT Hub连接字符串和设备ID
IOTHUB_CONNECTION_STRING = "你的IoT Hub连接字符串"
DEVICE_ID = "你的设备ID"

def call_toggle_led():
    registry_manager = IoTHubRegistryManager(IOTHUB_CONNECTION_STRING)
    method_params = {
        "methodName": "toggleLED",
        "payload": {"state": "on"},
        "responseTimeoutInSeconds": 30,
        "connectTimeoutInSeconds": 5
    }
    response = registry_manager.invoke_device_method(DEVICE_ID, method_params)
    print(f"设备响应: {response.payload}")

if __name__ == "__main__":
    call_toggle_led()

• 设备侧ESP8266 Arduino代码（处理方法调用）：

#include <AzureIoTHub.h>
#include <AzureIoTProtocol_MQTT.h>

// 替换为你的设备连接字符串
static const char* connectionString = "你的设备连接字符串";
static IoTHubClientHandle iotHubClientHandle;

// 直接方法回调函数
int toggleLED(const char* methodName, const char* payload, size_t size, char** response, size_t* response_size) {
  String payloadStr(payload);
  if(payloadStr.indexOf("\"state\":\"on\"") != -1){
    digitalWrite(LED_BUILTIN, HIGH);
  } else {
    digitalWrite(LED_BUILTIN, LOW);
  }
  // 返回成功响应
  const char* resp = "{\"result\": \"success\"}";
  *response_size = strlen(resp);
  *response = (char*)malloc(*response_size);
  strncpy(*response, resp, *response_size);
  return 200;
}

void setup() {
  pinMode(LED_BUILTIN, OUTPUT);
  iotHubClientHandle = IoTHubClient_CreateFromConnectionString(connectionString, MQTT_Protocol);
  // 注册直接方法回调
  IoTHubClient_SetDeviceMethodCallback(iotHubClientHandle, toggleLED, NULL);
}

void loop() {
  IoTHubClient_DoWork(iotHubClientHandle);
  delay(100);
}

2. 设备孪生（配置同步示例：设置目标温度） #

• 云侧Python代码（更新设备孪生）：

from azure.iot.hub import IoTHubRegistryManager

IOTHUB_CONNECTION_STRING = "你的IoT Hub连接字符串"
DEVICE_ID = "你的设备ID"

def update_target_temperature():
    registry_manager = IoTHubRegistryManager(IOTHUB_CONNECTION_STRING)
    twin_patch = {
        "properties": {
            "desired": {
                "targetTemperature": 28.0
            }
        }
    }
    registry_manager.update_twin(DEVICE_ID, twin_patch, "")
    print("设备孪生已更新")

if __name__ == "__main__":
    update_target_temperature()

• 设备侧ESP8266代码（监听孪生变化）：

#include <AzureIoTHub.h>
#include <AzureIoTProtocol_MQTT.h>

static const char* connectionString = "你的设备连接字符串";
static IoTHubClientHandle iotHubClientHandle;

// 设备孪生回调函数
void onTwinReceived(DEVICE_TWIN_UPDATE_STATE updateState, const unsigned char* payLoad, size_t size) {
  String payloadStr((const char*)payLoad);
  int tempPos = payloadStr.indexOf("targetTemperature");
  if(tempPos != -1){
    float targetTemp = payloadStr.substring(tempPos + 20).toFloat();
    Serial.print("目标温度已设置为: ");
    Serial.println(targetTemp);
    // 此处添加根据目标温度执行的控制逻辑
  }
}

void setup() {
  Serial.begin(115200);
  iotHubClientHandle = IoTHubClient_CreateFromConnectionString(connectionString, MQTT_Protocol);
  // 注册孪生回调
  IoTHubClient_SetDeviceTwinCallback(iotHubClientHandle, onTwinReceived, NULL);
}

void loop() {
  IoTHubClient_DoWork(iotHubClientHandle);
  delay(100);
}

内容的提问来源于stack exchange，提问作者Prathamesh Parit