Python Azure IoT SDK: How to invoke direct methods in backend app

Article
1/17/2024

Introduction

Recently, Microsoft has released the new version of Python Azure IoT SDK (V2.0) (refer to this page on IoT blog: New version of the Python SDK released). According to the release announcement, we should upgrade SDK from V1 to V2 since the v2 SDK aims to provide a simplified, more natural experience for developers. It’s designed in native Python.
In the previous tutorial “Python Azure IoT SDK: How to receive direct methods from IoT Hub”, we have showed the steps required to build a sample application for receiving direct methods from IoT Hub with Python Azure IoT SDK. In this article, we will learn how to invoke direct methods in backend app.
Direct methods represent a request-reply interaction with a device similar to an HTTP call in that they succeed or fail immediately (after a user-specified timeout). This approach is very useful for device management. Please refer to the following two articles for more details: “Understand and invoke direct methods from IoT Hub” and “Overview of device management with IoT Hub”. Take remote reboot control for example, as shown in Fig. 1, the back-end app informs the device through a direct method that it has initiated a reboot. The device uses the reported properties to update the reboot status of the device.

Fig. 1 Direct method call for reboot control

Prerequisites

Windows 10 with Visual Studio 2019 Community (“Python development” workload required)
Python Azure IoT device SDK: https://github.com/Azure/azure-iot-sdk-python/tree/master/azure-iot-device/samples
Python Azure IoT Hub Service SDK: https://github.com/Azure/azure-iot-sdk-python/tree/master/azure-iot-hub

Create Device using Device Explore

As we demonstrated in the last article, we will create a device and save the connection string for further use. Please refer to the session “Create Device using Device Explore” in this article “How to use Python Azure IoT SDK with Visual Studio”.

Invoke direct methods in backend app

Create a Python project with “Python Application” project temple, give a name such as “PythonIoTBackendDemo”. Copy and paste the following code to “PythonIoTBackendDemo.py”. Make sure that you substitute with your own IoTHub connection string and device ID in the code.

# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
 
import sys
import os
import time
from azure.iot.hub import IoTHubRegistryManager
from azure.iot.hub.models import CloudToDeviceMethod
 
 
iothub_connection_str =  "HostName=***.azure-devices.net;SharedAccessKeyName=iothubowner;SharedAccessKey=***"
device_id =  "MyRPi"
method_name1 =  "method1"
method_payload1 =  "lockdoor"
method_name2 =  "method2"
method_payload2 =  "lighton"
 
try:
    # Create IoTHubRegistryManager
    registry_manager =  IoTHubRegistryManager(iothub_connection_str)
 
    deviceMethod =  CloudToDeviceMethod(method_name=method_name1, payload=method_payload1)
    registry_manager.invoke_device_method(device_id, deviceMethod)
    print("executed method: " +  method_name1+" "+method_payload1)
    time.sleep(3)
    deviceMethod =  CloudToDeviceMethod(method_name=method_name2, payload=method_payload2)
    registry_manager.invoke_device_method(device_id, deviceMethod)
    print("executed method: " +  method_name2+" " +method_payload2)
 
except Exception as ex:
    print("Unexpected error {0}".format(ex))
except KeyboardInterrupt:
    print("iothub_registry_manager_sample stopped")

In this application, we will make use of Azure IoT Hub Service SDK to invoke “method 1” and “method 2” on the device “MyRPi” respectively, with 3 seconds interval. The output of the debug window will be shown as follows.

Fig. 2 Debug information on output window

Receive direct methods in device app

Create a Python project with “Python Application” project temple, give a name such as “PythonIoTDeviceDemo”.Copy and paste the following code to “PythonIoTDeviceDemo.py”. Make sure that you substitute with your own connection string in the code.

# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
 
import os
import asyncio
import threading
from six.moves import input
from azure.iot.device.aio import IoTHubDeviceClient
from azure.iot.device import MethodResponse
 
 
async def  main():
    # The connection string for your device.
    conn_str =  "HostName=***.azure-devices.net;DeviceId=***;SharedAccessKey=***"
    # The client object is used to interact with your Azure IoT hub.
    device_client =  IoTHubDeviceClient.create_from_connection_string(conn_str)
 
    # connect the client.
    await device_client.connect()
 
    # define behavior for handling methods
    async def  method1_listener(device_client):
        while True:
            method_request =  await device_client.receive_method_request(
                "method1"
            )  # Wait for method1 calls
            payload =  {"result":  True, "data": "execute successfully"}  # set response payload
            status =  200  # set return status code
            print("executed method1")
            method_response =  MethodResponse.create_from_method_request(
                method_request, status, payload
            )
            await device_client.send_method_response(method_response)  # send response
 
    async def  method2_listener(device_client):
        while True:
            method_request =  await device_client.receive_method_request(
                "method2"
            )  # Wait for method2 calls
            payload =  {"result":  True, "data": 1234}   # set response payload
            status =  200  # set return status code
            print("executed method2")
            method_response =  MethodResponse.create_from_method_request(
                method_request, status, payload
            )
            await device_client.send_method_response(method_response)  # send response
 
    async def  generic_method_listener(device_client):
        while True:
            method_request =  (
                await device_client.receive_method_request()
            )  # Wait for unknown method calls
            payload =  {"result":  False, "data": "unknown method"}  # set response payload
            status =  400  # set return status code
            print("executed unknown method: " +  method_request.name)
            method_response =  MethodResponse.create_from_method_request(
                method_request, status, payload
            )
            await device_client.send_method_response(method_response)  # send response
 
    # define behavior for halting the application
    def stdin_listener():
        while True:
            selection =  input("Press Q to quit\n")
            if selection == "Q"  or selection == "q":
                print("Quitting...")
                break
 
    # Schedule tasks for Method Listener
    listeners =  asyncio.gather(
        method1_listener(device_client),
        method2_listener(device_client),
        generic_method_listener(device_client),
    )
 
    # Run the stdin listener in the event loop
    loop =  asyncio.get_running_loop()
    user_finished =  loop.run_in_executor(None, stdin_listener)
 
    # Wait for user to indicate they are done listening for method calls
    await user_finished
 
    # Cancel listening
    listeners.cancel()
 
    # Finally, disconnect
    await device_client.disconnect()
 
 
if __name__ == "__main__":
    asyncio.run(main())
 
    # If using Python 3.6 or below, use the following code instead of asyncio.run(main()):
    # loop = asyncio.get_event_loop()
    # loop.run_until_complete(main())
# loop.close()

Then, press F5 to start debug. As soon as the local Python application runs, you will see “Press Q to quit” in the window.
Once the device receives the method call, it will execute the code that defined in “method1_listener”. It will do something locally and send the results back. Consequently, you will see “executed method1” and “executed method2” in the local Visual Studio debug output window, which is shown in Fig. 3.

Fig. 3 Visual Studio debug information

Summary

In this article, we showed the steps to invoke direct methods in backend app and receive direct methods in device app with Python Azure IoT SDK**.**

** **

Resources

IoT Blog: New version of the Python SDK released
Python Azure IoT device SDK: https://github.com/Azure/azure-iot-sdk-python/tree/master/azure-iot-device/samples
Python Azure IoT Hub Service SDK: https://github.com/Azure/azure-iot-sdk-python/tree/master/azure-iot-hub

Share via