使用 Azure 函数将负载发送到 IoT 中心以在 Azure 数字孪生中使用答案

【问题标题】：Sending payload to IoT Hub for using in Azure Digital Twin using an Azure Function使用 Azure 函数将负载发送到 IoT 中心以在 Azure 数字孪生中使用
【发布时间】：2023-02-01 03:44:37
【问题描述】：

对于任何不正确的格式表示歉意，很长一段时间以来我在堆栈溢出上发布了任何内容。

我希望将数据的 json 负载发送到 Azure IoT 中心，然后我将使用 Azure Function App 对其进行处理，以在 Azure Digital Twin 中显示实时遥测数据。

我可以将负载发布到 IoT 中心并使用资源管理器查看它，但是我的函数无法获取它并在 Azure 数字孪生中显示该遥测数据。通过谷歌搜索，我发现 json 文件需要进行 utf-8 加密并设置为 application/json，我认为这可能是我目前尝试修复此问题的问题。

我在下面包含了来自我的 azure 函数应用程序的日志流的片段，如图所示消息的“正文”部分被打乱，这就是为什么我认为这可能是有效负载编码方式的问题：

“iothub-message-source”：“Telemetry”}，“body”：“eyJwb3dlciI6ICIxLjciLCAid2luZF9zcGVlZCI6ICIxLjciLCAid2luZF9kaXJlY3Rpb24iOiAiMS43In0 =”} 2023-01-27T13:39:05Z [错误] 摄取函数出错：无法访问 Newtonsoft.Json.Linq.JValue 上的子值。

我当前的测试代码如下，用于将有效载荷发送到 IoT 中心，潜在的问题是我没有正确编码有效载荷。

import datetime, requests 
import json

deviceID = "JanTestDT"
IoTHubName = "IoTJanTest"
iotHubAPIVer = "2018-04-01"
iotHubRestURI = "https://" + IoTHubName + ".azure-devices.net/devices/" + deviceID +     "/messages/events?api-version=" + iotHubAPIVer
SASToken = 'SharedAccessSignature'

Headers = {}
Headers['Authorization'] = SASToken
Headers['Content-Type'] = "application/json"
Headers['charset'] = "utf-8"

datetime =  datetime.datetime.now()
payload = {
'power': "1.7",
'wind_speed': "1.7",
'wind_direction': "1.7"
}

payload2 = json.dumps(payload, ensure_ascii = False).encode("utf8")

resp = requests.post(iotHubRestURI, data=payload2, headers=Headers)

我尝试以几种不同的方式正确编码有效负载，包括 request.post 中的 utf-8，但这会产生一个错误，即无法对 dict 进行编码，或者仍然在 Function App 日志流中加密主体无法破译它。

感谢您对此提供的任何帮助和/或指导 - 很乐意进一步详细说明任何不清楚的地方。

【问题讨论】：

为什么不直接传递带有 requests.post() 的 json= 参数的字典，让 requests 处理所有序列化和内容类型/字符集标头？

标签： python json utf-8 azure-iot-hub azure-digital-twins

【解决方案1】：

为什么要使用 Azure IoT Hub Rest API 端点而不是使用 Python SDK 有什么特别的原因吗？此外，即使您在通过 Azure IoT Explorer 查看时看到的是 JSON 格式的值，但通过 blob 等存储端点查看时的消息格式会显示出与您指出的不同的格式。

我没有使用 REST API 测试 Python 代码，但我有一个适合我的 Python SDK。请参考下面的代码示例

import os
import random
import time
from datetime import date, datetime
from json import dumps
from azure.iot.device import IoTHubDeviceClient, Message


def json_serial(obj):
    """JSON serializer for objects not serializable by default json code"""

    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    raise TypeError("Type %s not serializable" % type(obj))


CONNECTION_STRING = "<AzureIoTHubDevicePrimaryConnectionString>"
TEMPERATURE = 45.0
HUMIDITY = 60
MSG_TXT = '{{"temperature": {temperature},"humidity": {humidity}, "timesent": {timesent}}}'


def run_telemetry_sample(client):
    print("IoT Hub device sending periodic messages")

    client.connect()

    while True:
        temperature = TEMPERATURE + (random.random() * 15)
        humidity = HUMIDITY + (random.random() * 20)
        x = datetime.now().isoformat()
        timesent = dumps(datetime.now(), default=json_serial)
        msg_txt_formatted = MSG_TXT.format(
            temperature=temperature, humidity=humidity, timesent=timesent)
        message = Message(msg_txt_formatted, content_encoding="utf-8", content_type="application/json")
        
        print("Sending message: {}".format(message))
        client.send_message(message)
        print("Message successfully sent")
        time.sleep(10)


def main():
    print("IoT Hub Quickstart #1 - Simulated device")
    print("Press Ctrl-C to exit")

    client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING)

    try:
        run_telemetry_sample(client)
    except KeyboardInterrupt:
        print("IoTHubClient sample stopped by user")
    finally:
        print("Shutting down IoTHubClient")
        client.shutdown()


if __name__ == '__main__':
    main()

您可以在代码中编辑 MSG_TXT 变量以匹配负载格式并传递值。请注意，SDK 使用来自 Azure IoT 设备库的 Message class，该库具有内容类型和内容编码的重载。这是我在代码message = Message(msg_txt_formatted, content_encoding="utf-8", content_type="application/json") 中传递重载的方式

我已通过路由到 Blob 存储容器来验证消息，并且可以看到 JSON 格式的遥测数据。请参考下面的图像截图，参考在终点捕获的数据。

希望这可以帮助！

【讨论】：