使用 Azure 函数将有效负载发送到 IoT 中心以在 Azure 数字孪生中使用-解网

问：

对于任何不正确的格式，我很抱歉，因为我在堆栈溢出上发布了很长时间。

我希望将数据的 json 有效负载发送到 Azure IoT 中心，然后我将使用 Azure 函数应用进行处理，以在 Azure 数字孪生中显示实时遥测数据。

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

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

“iothub-message-source”：“遥测”}，“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，但这会产生一个错误，即字典无法编码，或者仍然在函数应用日志流中加密正文无法破译它。

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

python json utf-8 azure-iot-hub azure-digital-twins

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 类，该类具有内容类型和内容编码的重载。以下是我如何在代码中传递重载message = Message(msg_txt_formatted, content_encoding="utf-8", content_type="application/json")

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

希望这有帮助！

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

Sending payload to IoT Hub for using in Azure Digital Twin using an Azure Function

评论

评论