Python实现Azure IoT Hub遥测数据导入Digital Twins报错求助

一、核心错误根源 #

你遇到的'NoneType' object has no attribute 'startswith'错误，本质是环境变量DIGITAL_TWINS_URL未正确加载，导致创建DigitalTwinsClient时传入了None值，触发SDK的URL格式校验失败。以下是分步骤的修复方案：

二、修复环境变量配置 #

本地运行Azure Functions时，环境变量必须写在local.settings.json文件中，而非系统环境变量：

1. 确保项目根目录下存在local.settings.json，内容示例：

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "DIGITAL_TWINS_URL": "https://你的ADT实例名称.区域.digitaltwins.azure.net"
  }
}

2. 替换DIGITAL_TWINS_URL为你的ADT实例实际URL，可在Azure门户ADT实例的"概述"页面复制。

三、修正事件数据解析逻辑 #

IoT Hub通过Event Grid推送的消息结构和教程中C#处理的结构有差异，需调整解析逻辑：

• 遥测消息的body是Base64编码字符串，需先解码再转JSON
• 系统属性的层级需对应Event Grid的消息格式

修改后的函数代码：

import os 
import logging
import json
import base64
from azure.identity import DefaultAzureCredential
from azure.digitaltwins.core import DigitalTwinsClient
from azure.core.exceptions import ResourceNotFoundError
import azure.functions as func

# 提前校验环境变量，避免后续报错
adt_instance_url = os.getenv("DIGITAL_TWINS_URL")
if not adt_instance_url:
    logging.error("请在local.settings.json中配置DIGITAL_TWINS_URL环境变量")
    raise ValueError("DIGITAL_TWINS_URL未设置")

async def main(event: func.EventGridEvent):
    try:
        # 初始化ADT客户端
        credential = DefaultAzureCredential()
        client = DigitalTwinsClient(adt_instance_url, credential)
        logging.info("ADT客户端连接成功")

        if event and event.data:
            logging.info(f"收到Event Grid事件: {json.dumps(event.data)}")
            
            # 1. 解码遥测消息体
            device_data = event.data
            body_str = base64.b64decode(device_data["body"]).decode("utf-8")
            telemetry = json.loads(body_str)
            
            # 2. 获取设备ID与遥测数据
            device_id = device_data["systemProperties"]["iothub-connection-device-id"]
            temperature = telemetry.get("temperature")
            humidity = telemetry.get("humidity")
            
            logging.info(f"设备 {device_id} 遥测：温度={temperature}，湿度={humidity}")

            # 3. 更新数字孪生
            twin_id = device_id
            patch = [
                {"op": "replace", "path": "/Temperature", "value": temperature},
                {"op": "replace", "path": "/Humidity", "value": humidity}
            ]
            await client.update_digital_twin(twin_id, patch)
            logging.info(f"孪生体 {twin_id} 更新成功")

    except ResourceNotFoundError:
        logging.error(f"孪生体 {twin_id} 不存在，请先创建对应孪生体")
    except Exception as ex:
        logging.error(f"处理出错: {str(ex)}", exc_info=True)  # 打印完整堆栈便于排查

四、本地身份验证配置 #

本地运行时，DefaultAzureCredential依赖Azure CLI的登录身份：

1. 打开终端执行az login，登录你的Azure账号
2. 确保该账号在ADT实例的IAM页面拥有数字孪生数据所有者或数字孪生数据参与者权限

五、事件订阅与孪生体检查 #

1. 事件订阅需选择IoT Hub的设备遥测事件类型，端点指向你的函数应用
2. 确保数字孪生体已创建，且ID与设备ID完全一致（大小写敏感）
3. 孪生体模型中必须包含Temperature和Humidity属性，属性路径需与patch中的/Temperature、/Humidity匹配

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