Planifier et diffuser des travaux (Python)

Article
10/01/2024

Utilisez Azure IoT Hub pour planifier et suivre des travaux qui mettent à jour des millions d’appareils. Utilisez les travaux pour :

Mettre à jour les propriétés souhaitées
Mettre à jour les balises
Appeler des méthodes directes

Sur le plan conceptuel, un travail encapsule l’une de ces actions et suit la progression de l’exécution par rapport à un ensemble d’appareils, défini par une requête de représentation d’appareil. Par exemple, à l’aide d’un travail, une application principale peut appeler une méthode de redémarrage sur 10 000 appareils, spécifiée par une requête de représentation d’appareil et planifiée dans l’avenir. Cette application peut ensuite suivre la progression à mesure que chacun de ces appareils reçoit et exécute la méthode de redémarrage.

Pour en savoir plus sur chacune de ces fonctionnalités, consultez les articles suivants :

Jumeau d’appareil et propriétés : Prise en main des représentations d’appareils (Node) et Comprendre et utiliser les jumeaux d’appareil IoT Hub
Méthodes directes : Guide du développeur IoT Hub - méthodes directes

Notes

Les fonctionnalités décrites dans cet article sont uniquement disponibles au niveau Standard d’IoT Hub. Pour plus d’informations sur les niveaux de base et standard/gratuit d’IoT Hub, consultez Choisir le niveau IoT Hub correspondant à votre solution.

Cet article vous montre comment créer deux applications Python :

Une application Python d’appareil simulé, simDevice.py, qui implémente une méthode directe appelée lockDoor, qui peut être appelée par l’application back-end.
Une application console Python, scheduleJobService.py, qui crée deux travaux. Un travail appelle la méthode directe lockDoor et un autre travail envoie les mises à jour de propriété souhaitées à plusieurs appareils.

Notes

Pour plus d’informations sur les outils SDK disponibles pour créer des applications d’appareil et de back-end, consultez les SDK Azure IoT.

Prérequis

Un compte Azure actif. (Si vous ne possédez pas de compte, vous pouvez créer un compte gratuit en quelques minutes seulement.)
Un hub IoT dans votre abonnement Azure. Si vous n’avez pas encore de hub, vous pouvez suivre les étapes décrites dans Créer un hub IoT.
Un appareil inscrit dans votre hub IoT. Si vous n’avez pas d’appareil dans votre hub IoT, suivez les étapes décrites dans Inscrire un appareil.
Python version 3.7 ou ultérieure est recommandé. Veillez à utiliser l’installation 32 bits ou 64 bits comme requis par votre programme d’installation. Lorsque vous y êtes invité pendant l’installation, veillez à ajouter Python à votre variable d’environnement propre à la plateforme.

Création d’une application de périphérique simulé

Dans cette section, vous allez créer une application console Python qui répond à une méthode directe appelée par le cloud, ce qui déclenche une méthode lockDoor simulée.

Important

Cet article comprend les étapes à suivre pour connecter un appareil en utilisant une signature d’accès partagé, également appelée « authentification par clé symétrique ». Cette méthode d’authentification est pratique pour les tests et les évaluations, mais l’authentification d’un appareil en utilisant des certificats X.509 est une approche plus sécurisée. Pour plus d’informations, consultez Meilleures pratiques de sécurité > Sécurité des connexions.

À partir de votre invite de commandes, exécutez la commande suivante pour installer le package azure-iot-device :
```
pip install azure-iot-device
```
À l’aide d’un éditeur de texte, créez un fichier simDevice.py dans votre répertoire de travail.
Ajoutez les instructions et variables import ci-dessous au début du fichier simDevice.py. Remplacez deviceConnectionString par la chaîne de connexion de l’appareil que vous avez créé plus haut :
```
import time
from azure.iot.device import IoTHubDeviceClient, MethodResponse

CONNECTION_STRING = "{deviceConnectionString}"
```

Définissez la fonction suivante, qui va instancier un client et le configurer pour répondre à la méthode lockDoor, ainsi que pour recevoir des mises à jour de jumeau d’appareil :

def create_client():
    # Instantiate the client
    client = IoTHubDeviceClient.create_from_connection_string(CONNECTION_STRING)

    # Define behavior for responding to the lockDoor direct method
    def method_request_handler(method_request):
        if method_request.name == "lockDoor":
            print("Locking Door!")

            resp_status = 200
            resp_payload = {"Response": "lockDoor called successfully"}
            method_response = MethodResponse.create_from_method_request(
                method_request=method_request,
                status=resp_status,
                payload=resp_payload
            )
            client.send_method_response(method_response)

    # Define behavior for receiving a twin patch
    def twin_patch_handler(twin_patch):
        print("")
        print("Twin desired properties patch received:")
        print(twin_patch)

    # Set the handlers on the client
    try:
        print("Beginning to listen for 'lockDoor' direct method invocations...")
        client.on_method_request_received = method_request_handler
        print("Beginning to listen for updates to the Twin desired properties...")
        client.on_twin_desired_properties_patch_received = twin_patch_handler
    except:
        # If something goes wrong while setting the handlers, clean up the client
        client.shutdown()
        raise

Ajoutez le code suivant pour exécuter l’exemple :

def main():
    print ("Starting the IoT Hub Python jobs sample...")
    client = create_client()

    print ("IoTHubDeviceClient waiting for commands, press Ctrl-C to exit")
    try:
        while True:
            time.sleep(100)
    except KeyboardInterrupt:
        print("IoTHubDeviceClient sample stopped!")
    finally:
        # Graceful exit
        print("Shutting down IoT Hub Client")
        client.shutdown()


if __name__ == '__main__':
    main()

Enregistrez et fermez le fichier simDevice.py.

Notes

Pour simplifier les choses, cet article n’implémente aucune stratégie de nouvelles tentatives. Dans le code de production, vous devez implémenter des stratégies de nouvelle tentative (par exemple, une interruption exponentielle), comme indiqué dans l’article Gestion des erreurs temporaires.

Obtention de la chaîne de connexion de l’IoT Hub

Dans cet article, vous allez créer un service principal qui appelle une méthode directe sur un appareil et met à jour le jumeau de celui-ci. Pour appeler une méthode directe sur un appareil, le service a besoin de l’autorisation de connexion de service. Le service a également besoin des autorisations de lecture du registre et d’écriture du registre pour pouvoir lire et écrire dans le registre des identités. Aucune stratégie d’accès partagé par défaut ne contient que ces autorisations. Vous devez donc en créer une.

Pour créer une stratégie d’accès partagé qui accorde des autorisations de connexion de service, de lecture du registre et d’écriture du registre, et obtenir une chaîne de connexion pour cette stratégie, procédez comme suit :

Ouvrez votre IoT Hub dans le portail Azure. La méthode la plus simple pour accéder à votre IoT Hub consiste à sélectionner Groupes de ressources, le groupe de ressources dans lequel se trouve votre IoT Hub, puis votre IoT Hub dans la liste des ressources.
Dans le volet de gauche de votre IoT Hub, sélectionnez Stratégies d’accès partagé.
Dans le menu supérieur, au-dessus de la liste des stratégies, sélectionnez Ajouter une stratégie d’accès partagé.
Dans le volet Ajouter une stratégie d’accès partagé, entrez un nom descriptif pour votre stratégie, par exemple : serviceAndRegistryReadWrite. Sous Autorisations, sélectionnez Écriture du registre et Connexion au service (Lecture du registre est automatiquement sélectionné lorsque vous sélectionnez Écriture du registre), puis Ajouter.
Dans la page Stratégies d’accès partagé, sélectionnez votre nouvelle stratégie dans la liste des stratégies.
Dans le nouveau volet qui s’affiche, sélectionnez l’icône de copie pour la Chaîne de connexion principale et enregistrez la valeur.

Pour plus d’informations sur les autorisations et les stratégies d’accès partagé IoT Hub, consultez Contrôle d’accès et autorisations.

Important

Cet article comprend les étapes à suivre pour se connecter à un service à l’aide d’une signature d’accès partagé. Cette méthode d’authentification est pratique pour les tests et les évaluations, mais l’authentification à un service avec Microsoft Entra ID ou des identités managées est une approche plus sécurisée. Pour plus d’informations, consultez Meilleures pratiques de sécurité > Sécurité du cloud.

Planifier des travaux pour appeler une méthode directe et mettre à jour les propriétés d’une représentation d’appareil

Dans cette section, vous créez une application console Python qui lance lockDoor à distance sur un appareil à l’aide d’une méthode directe et également met à jour les propriétés souhaitées du jumeau d’appareil.

À partir de votre invite de commandes, exécutez la commande suivante pour installer le package azure-iot-hub :
```
pip install azure-iot-hub
```
À l’aide d’un éditeur de texte, créez un fichier scheduleJobService.py dans votre répertoire de travail.

Ajoutez les instructions et variables import ci-dessous au début du fichier scheduleJobService.py. Remplacez la valeur de l’espace réservé {IoTHubConnectionString} par la chaîne de connexion du hub IoT que vous avez copiée à l’étape Obtenir la chaîne de connexion du hub IoT. Remplacez l’espace réservé {deviceId} par l’ID de l’appareil (le nom) de votre appareil inscrit :

import os
import sys
import datetime
import time
import threading
import uuid
import msrest

from azure.iot.hub import IoTHubJobManager, IoTHubRegistryManager
from azure.iot.hub.models import JobProperties, JobRequest, Twin, TwinProperties, CloudToDeviceMethod

CONNECTION_STRING = "{IoTHubConnectionString}"
DEVICE_ID = "{deviceId}"

METHOD_NAME = "lockDoor"
METHOD_PAYLOAD = "{\"lockTime\":\"10m\"}"
UPDATE_PATCH = {"building":43,"floor":3}
TIMEOUT = 60
WAIT_COUNT = 5

# Create IoTHubJobManager
iothub_job_manager = IoTHubJobManager.from_connection_string(CONNECTION_STRING)

Ajoutez les méthodes suivantes pour exécuter les travaux qui appellent la méthode directe et le jumeau d’appareil :

def device_method_job(job_id, device_id, execution_time):
    print ( "" )
    print ( "Scheduling job: " + str(job_id) )

    job_request = JobRequest()
    job_request.job_id = job_id
    job_request.type = "scheduleDeviceMethod"
    job_request.start_time = datetime.datetime.utcnow().isoformat()
    job_request.cloud_to_device_method = CloudToDeviceMethod(method_name=METHOD_NAME, payload=METHOD_PAYLOAD)
    job_request.max_execution_time_in_seconds = execution_time
    job_request.query_condition = "DeviceId in ['{}']".format(device_id)

    new_job_response = iothub_job_manager.create_scheduled_job(job_id, job_request)

def device_twin_job(job_id, device_id, execution_time):
    print ( "" )
    print ( "Scheduling job " + str(job_id) )

    job_request = JobRequest()
    job_request.job_id = job_id
    job_request.type = "scheduleUpdateTwin"
    job_request.start_time = datetime.datetime.utcnow().isoformat()
    job_request.update_twin = Twin(etag="*", properties=TwinProperties(desired=UPDATE_PATCH))
    job_request.max_execution_time_in_seconds = execution_time
    job_request.query_condition = "DeviceId in ['{}']".format(device_id)

    new_job_response = iothub_job_manager.create_scheduled_job(job_id, job_request)

Ajoutez le code suivant pour planifier les travaux et mettre à jour leur statut. Ajoutez également la routine main :

def iothub_jobs_sample_run():
    try:
        method_job_id = uuid.uuid4()
        device_method_job(method_job_id, DEVICE_ID, TIMEOUT)

        print ( "" )
        print ( "Direct method called with Job Id: " + str(method_job_id) )

        twin_job_id = uuid.uuid4()
        device_twin_job(twin_job_id, DEVICE_ID, TIMEOUT)

        print ( "" )
        print ( "Device twin called with Job Id: " + str(twin_job_id) )

        while True:
            print ( "" )

            method_job_status = iothub_job_manager.get_scheduled_job(method_job_id)
            print ( "...job " + str(method_job_id) + " " + method_job_status.status )

            twin_job_status = iothub_job_manager.get_scheduled_job(twin_job_id)
            print ( "...job " + str(twin_job_id) + " " + twin_job_status.status )

            print ( "Job status posted, press Ctrl-C to exit" )
            time.sleep(WAIT_COUNT)

    except msrest.exceptions.HttpOperationError as ex:
        print ( "" )
        print ( "Http error {}".format(ex.response.text) )
        return
    except Exception as ex:
        print ( "" )
        print ( "Unexpected error {}".format(ex) )
        return
    except KeyboardInterrupt:
        print ( "" )
        print ( "IoTHubService sample stopped" )

if __name__ == '__main__':
    print ( "Starting the IoT Hub jobs Python sample..." )
    print ( "    Connection string = {0}".format(CONNECTION_STRING) )
    print ( "    Device ID         = {0}".format(DEVICE_ID) )

    iothub_jobs_sample_run()

Enregistrez et fermez le fichier scheduleJobService.py.

Exécution des applications

Vous êtes maintenant prêt à exécuter les applications.

À l’invite de commandes dans votre répertoire de travail, exécutez la commande suivante pour commencer à écouter la méthode directe de redémarrage :
```
python simDevice.py
```
À une autre invite de commandes dans votre répertoire de travail, exécutez la commande suivante pour déclencher les travaux afin de verrouiller la porte et de mettre à jour le jumeau :
```
python scheduleJobService.py
```
Vous voyez les réponses de l’appareil à la méthode directe et la mise à jour des jumeaux d’appareils dans la console.

Étapes suivantes

Dans cet article, vous avez planifié des travaux pour exécuter une méthode directe et mettre à jour les propriétés du jumeau d’appareil.

Pour continuer à explorer le hub IoT et les modèles de gestion des appareils, mettez à jour une image dans le Tutoriel sur le service Device Update pour IoT Hub : Utilisation de l’image de référence Raspberry Pi 3 B+.

Partager via

Planifier et diffuser des travaux (Python)

Prérequis

Création d’une application de périphérique simulé

Obtention de la chaîne de connexion de l’IoT Hub

Planifier des travaux pour appeler une méthode directe et mettre à jour les propriétés d’une représentation d’appareil

Exécution des applications

Étapes suivantes

Commentaires

Ressources supplémentaires