Ingest telemery data from Azure IoT Hub to Digital Twins using Python Error

Question

I would like to send the telemetry data from the IOT hub to Azure digital twins using Python instead of C#( I don't have experience with C#), but I can't get it.

I have followed the complete example from Microsoft (https://learn.microsoft.com/en-us/azure/digital-twins/how-to-ingest-iot-hub-data) without success.

This is my code for Simulated Device. I checked and it successfully send the message to IoT Hub:

from azure.iot.device import IoTHubDeviceClient
import json
import random
import time

CONNECTION_STRING = "<DeviceIDConnectionString>"


def simulate_telemetry(device_client):
    while True:
        temperature = random.randint(20, 30)
        humidity = random.randint(60, 80)

        telemetry = {
            "temperature": temperature,
            "humidity": humidity
        }

        message = json.dumps(telemetry)
        device_client.send_message(message)

        print(f"Sent message: {message}")
        time.sleep(1)

if __name__ == "__main__":
    device_client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING)
    device_client.connect()

    simulate_telemetry(device_client)

And this is the code for function app using python and event grid trigger ( I followed the C# code in the tutorial link):

import os 
import logging
import json
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 adt_instance_url is None:
    logging.error("Application setting 'DIGITAL_TWINS_URL' not set")

async def main(event: func.EventGridEvent):
    try:
        # Authenticate with Digital Twins
        credential = DefaultAzureCredential()
        client = DigitalTwinsClient(adt_instance_url, credential)
        logging.info("ADT service client connection created.")

        if event is not None and event.data is not None:
            logging.info(event.data)

            # Find device ID and temperature
            device_message = event.get_json()
            device_id = device_message["systemProperties"]["iothub-connection-device-id"]
            temperature = device_message["body"]["temperature"]
            humidity = device_message["body"]["humidity"]
            
            logging.info(f"Device: {device_id} Temperature is: {temperature} Humidity is : {humidity}")

            # Update twin with device temperature
            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)
    except ResourceNotFoundError:
        logging.error(f"Digital twin with ID {twin_id} not found.")
    except Exception as ex:
        logging.error(f"Error in ingest function: {ex}")

I also create the event subscription to connect IoT hub and Function App already. I follow all the step in tutorial.

When I run the function app locally on VS code ( Ubuntu 20.04), it executed successfully but have this error in between:

Error in ingest function: 'NoneType' object has no attribute 'startswith'

I only use these 2 files code for whole project.

I run the query on Azure Digital Twins explorer but I don't see the twin updated as expected.

Should I add something else (code) or what can I change? I think maybe the problem is missing the Digital Twins as output or IoT Hub as input.

Answer 1

I have tested the telemetry ingestion with Python Code and here is a sample that I have tested which works.

1 Sample for IoT Hub data simulation

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

CONNECTION_STRING = "<Device Connection String>"
TEMPERATURE = 35.0
HUMIDITY = 60
MSG_TXT = '{{"Temperature": {Temperature},"Humidity": {Humidity}}}'


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)                     
        msg_txt_formatted = MSG_TXT.format(
        Temperature=temperature, Humidity=humidity)
    
        message = Message(msg_txt_formatted, content_encoding="utf-8", content_type="application/json")
        if temperature > 30:
            message.custom_properties["temperatureAlert"] = "true"
        else:
            message.custom_properties["temperatureAlert"] = "false"
    
        print("Sending message: {}".format(message))
        client.send_message(message)
        print("Message successfully sent")
        time.sleep(30)


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()

Please make sure to provide the correct IoT Hub device connection string in the above code

2 Azure Event Grid trigger function

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

adt_instance_url = "<SDT instance URL>"
credential = DefaultAzureCredential()
client = DigitalTwinsClient(adt_instance_url, credential)

async def main(event: func.EventGridEvent):
    
    if not adt_instance_url:
        raise ValueError("Application setting \"ADT_SERVICE_URL\" not set")

    try:
        # Authenticate with Digital Twins

        logging.info("ADT service client connection created.")
        result = json.dumps({
            'id': event.id,
            'data': event.get_json(),
            'topic': event.topic,
            'subject': event.subject,
            'event_type': event.event_type,
        })
        data = json.loads(result)

        if result and data:                

            # <Find_device_ID_and_temperature>
            device_message = data["data"]
            logging.info('Python EventGrid trigger processed an event: %s', device_message)
            device_id = device_message["systemProperties"]["iothub-connection-device-id"]
            temperature = device_message["body"]["Temperature"]
            # </Find_device_ID_and_temperature>

            logging.info(f"Device:{device_id} Temperature is:{temperature}")

            # <Update_twin_with_device_temperature>
            update_twin_data = [{"op": "replace", "path": "/Temperature", "value": temperature}]
            await client.update_digital_twin(device_id, update_twin_data)
            # </Update_twin_with_device_temperature>
    except Exception as ex:
        logging.info("Error in ingest function: %s", str(ex))

Please provide the correct ADT instance URL to the connection string. Please note that I have provided the URL directly in the code. However, it is recommended to use the approach you are trying to follow and pass the URL to the code through Environment variable.

I have published the code to Azure function App and assigned "Azure Digital Twins Data Owner" role access to the function app. I have also configured the function app to receive telemetry event data from the IoT Hub. Please refer the sections Assign an access role and Connect the function to IoT Hub for steps needed to do this.

Once I have everything configured, I could see the events received to the cloud Azure function and could update the Azure digital twin. Please refer the below image for reference.

Please also ensure that your Azure Digital Twin has an initial value loaded to the temperature property as we are using replace method patch to update the Twin.