如何实现Azure IoT设备孪生消息在本地Python终端的接收与显示

Hey there! Great job getting your Python code sending messages to Azure IoT Hub already—you're well on your way to accessing Device Twin data locally. Let's walk through exactly what you need to do to view Device Twin messages right in your terminal, covering both the Azure side (which is minimal) and the Python code additions.

Azure IoT Hub Side Configuration #

First, the good news: you don't need any extra special configuration for Device Twins beyond what you've already done! Device Twins are automatically created for every registered device in IoT Hub. Just double-check that:

• Your device is properly registered with IoT Hub using your X509 certificate (you mentioned you already set this up, so this should be good)
• The certificate you're using for the device is valid and associated with the device in IoT Hub

Python Code Implementation #

You already have the core libraries imported, so you don't need any extra packages—azure-iot-device has all the tools you need for Device Twin interactions. Here's how to extend your existing code to listen for Twin updates and fetch the current Twin state:

Step 1: Initialize the Device Client (with X509) #

You probably already have this, but let's confirm the setup for X509 authentication, since that's what you're using:

from s7agent.common import LOG
from json import dumps as json_dumps
from azure.iot.device import IoTHubDeviceClient, X509
import threading

# Configure your X509 credentials
x509 = X509(
    cert_file="/path/to/your/device-cert.pem",  # Replace with your cert path
    key_file="/path/to/your/device-key.pem",    # Replace with your key path
    pass_phrase="your-cert-passphrase"          # Omit if your key doesn't have a passphrase
)

# Create the device client
client = IoTHubDeviceClient.create_from_x509_certificate(
    x509=x509,
    hostname="your-iothub-name.azure-devices.net",  # Replace with your IoT Hub hostname
    device_id="your-registered-device-id"           # Replace with your device ID
)

Step 2: Add Functions to Handle Twin Data #

Add these functions to fetch the current Twin state and listen for updates to desired properties (the part of the Twin that IoT Hub sends to the device):

def fetch_current_twin(client):
    """Fetch and print the full current Device Twin state"""
    try:
        twin = client.get_twin()
        print("\n=== Current Device Twin ===")
        print(json_dumps(twin, indent=2))
    except Exception as e:
        LOG.error(f"Failed to fetch Device Twin: {str(e)}")

def listen_for_twin_updates(client):
    """Listen indefinitely for desired property patches from IoT Hub"""
    print("\nListening for Device Twin desired property updates...")
    while True:
        # This blocks until a desired property patch is received
        patch = client.receive_twin_desired_properties_patch()
        print("\n=== Received Desired Properties Patch ===")
        print(json_dumps(patch, indent=2))
        
        # Optional: Send a reported property update to confirm receipt
        reported_update = {"lastDesiredUpdateReceived": patch, "status": "processed"}
        client.patch_twin_reported_properties(reported_update)
        print("\nReported properties updated to confirm patch receipt.")

Step 3: Run the Client and Listen #

Add this main block to connect the client, fetch the initial Twin, and start listening for updates:

if __name__ == "__main__":
    try:
        # Connect to IoT Hub
        client.connect()
        print("Successfully connected to Azure IoT Hub.")
        
        # Fetch the initial Twin state
        fetch_current_twin(client)
        
        # Start the twin update listener in a background thread (so it doesn't block the main program)
        listener_thread = threading.Thread(target=listen_for_twin_updates, args=(client,))
        listener_thread.daemon = True
        listener_thread.start()
        
        # Keep the main program running
        input("\nPress Enter to exit the program...\n")
    except Exception as e:
        LOG.error(f"Connection or runtime error: {str(e)}")
    finally:
        # Clean up the connection
        client.disconnect()
        print("Disconnected from Azure IoT Hub.")

Key Notes: #

• Desired vs. Reported Properties: Desired properties are sent from IoT Hub to your device, while reported properties are sent from your device back to IoT Hub. The code above listens for desired property updates and optionally sends a reported update to confirm.
• Threading: We use a background thread for the listener because receive_twin_desired_properties_patch() is a blocking call—this lets the main program stay responsive.
• Updating the Library: Make sure you have the latest version of the Azure IoT Device SDK:

  pip install --upgrade azure-iot-device
  

Test this out by updating the desired properties of your device in Azure IoT Hub (you can do this via the Azure Portal, Azure CLI, or IoT Hub SDKs)—you'll see the patch show up instantly in your local terminal!

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