Programación y difusión de trabajos (Python)

Artículo
03/18/2023

Use Azure IoT Hub para programar y realizar el seguimiento de los trabajos que actualizan millones de dispositivos. Use los trabajos para:

Actualizar las propiedades deseadas
Actualizar etiquetas
Invocar métodos directos

Conceptualmente, un trabajo contiene una de estas acciones y realiza un seguimiento del progreso de ejecución en un conjunto de dispositivos, que define una consulta de dispositivo gemelo. Por ejemplo, una aplicación back-end puede usar un trabajo para invocar un método de reinicio en 10 000 dispositivos, especificado por una consulta de dispositivos gemelos y programada en el futuro. Esa aplicación puede después seguir el progreso cuando cada uno de estos dispositivos reciben y ejecutan el método de reinicio.

Más información sobre estas funcionalidades en estos artículos:

Dispositivo gemelo y propiedades: Introducción a los dispositivos gemelos y Tutorial: Uso de las propiedades deseadas para configurar dispositivos
Métodos directos: Guía para desarrolladores de IoT Hub: métodos directos

Nota

Las características descritas en este artículo solo están disponibles en el nivel estándar de IoT Hub. Para obtener más información sobre los niveles Básico y Estándar o Gratis de IoT Hub, consulte Elección del nivel adecuado de IoT Hub para la solución.

En este artículo se muestra cómo crear dos aplicaciones de Python:

Una aplicación de dispositivo simulado en Python, simDevice.py, que implementa un método directo llamado lockDoor, que puede ser llamado por la aplicación back-end.
Una aplicación de consola de Python, scheduleJobService.py, que crea dos trabajos. Un trabajo llama al método directo lockDoor y otro trabajo envía actualizaciones de propiedades deseadas a varios dispositivos.

Nota

Consulte los SDK de Azure IoT para obtener más información sobre las herramientas de SDK disponibles para compilar aplicaciones de dispositivo y back-end.

Requisitos previos

Una cuenta de Azure activa. (En caso de no tener ninguna, puede crear una cuenta gratuita en tan solo unos minutos).
Una instancia de IoT Hub. Cree uno con la CLI o el Azure Portal.
Dispositivo registrado. Registre uno en el Azure Portal.
Se recomienda usar Python versión 3.7 o posterior. Asegúrese de usar la instalación de 32 bits o 64 bits en función del programa de instalación. Cuando se le solicite durante la instalación, asegúrese de agregar Python a la variable de entorno específica de la plataforma.

Creación de una aplicación de dispositivo simulado

En esta sección, creará una aplicación de consola de Python que responda a un método directo al que la nube llama, que desencadena un método lockDoor simulado.

En el símbolo del sistema, ejecute el siguiente comando para instalar el paquete azure-iot-device:
```
pip install azure-iot-device
```
Con un editor de texto, cree un nuevo archivo simDevice.py en el directorio de trabajo.
Agregue las siguientes instrucciones y variables import al principio del archivo simDevice.py. Reemplace deviceConnectionString con la cadena de conexión del dispositivo que creó anteriormente:
```
import time
from azure.iot.device import IoTHubDeviceClient, MethodResponse

CONNECTION_STRING = "{deviceConnectionString}"
```

Defina la siguiente función, que crea una instancia de un cliente y la configura para responder al método lockDoor, así como para recibir actualizaciones de dispositivos gemelos:

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

Agregue el código siguiente para ejecutar el ejemplo:

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

Guarde y cierre el archivo simDevice.py.

Nota

Por simplificar, este artículo no implementa ninguna directiva de reintentos. En el código de producción, deberá implementar directivas de reintentos (por ejemplo, retroceso exponencial), tal y como se sugiere en el artículo Control de errores transitorios.

Obtener la cadena de conexión de IoT Hub

En este artículo, creará un servicio back-end que invoca un método directo en un dispositivo y actualiza el dispositivo gemelo. El servicio necesita el permiso Conexión del servicio para realizar una llamada al método directo de un dispositivo. El servicio también necesita los permisos Lectura del registro y Escribir en el registro para leer y escribir en el registro de identidad. No hay ninguna directiva de acceso compartido predeterminada que contenga solo estos permisos, por lo que tendrá que crearla.

Para crear una directiva de acceso compartido que conceda los permisos Conexión del servicio, Lectura del registro y Escribir en el registro y obtener una cadena de conexión para esta directiva, siga estos pasos:

En Azure Portal, abra IoT Hub. La manera más fácil de acceder a IoT Hub es seleccionar Grupos de recursos, seleccionar el grupo de recursos donde se encuentra IoT Hub y después seleccionarlo en la lista de recursos.
En el panel de la izquierda de IoT Hub, seleccione Directivas de acceso compartido.
En el menú superior situado encima de la lista de directivas, seleccione Agregar una directiva de acceso compartida.
En el panel Agregar directiva de acceso compartida, escriba un nombre descriptivo para la directiva, por ejemplo serviceAndRegistryReadWrite. En Permisos, seleccione Escritura del registro y Conexión del servicio (Lectura del registro se selecciona automáticamente al seleccionar Escritura del Registro) y, a continuación, seleccione Agregar.
En la página Directivas de acceso compartido, seleccione la nueva directiva en la lista de directivas.
En el nuevo panel que aparece, seleccione el icono de copia de la cadena de conexión principal y guarde el valor.

Para obtener más información sobre las directivas de acceso compartido y los permisos de IoT Hub, consulte Permisos y control del acceso.

Programación de trabajos para llamar un método directo y actualización de las propiedades de un dispositivo gemelo

En esta sección, creará una aplicación de consola de Python que inicia un lockDoor remoto en un dispositivo con un método directo y actualiza las propiedades seleccionadas del dispositivo gemelo.

En el símbolo del sistema, ejecute el siguiente comando para instalar el paquete azure-iot-hub:
```
pip install azure-iot-hub
```
Con un editor de texto, cree un nuevo archivo scheduleJobService.py en el directorio de trabajo.

Agregue las siguientes instrucciones y variables import al principio del archivo scheduleJobService.py. Reemplace el marcador de posición {IoTHubConnectionString} por la cadena de conexión de IoT Hub que copió anteriormente en Obtención de la cadena de conexión de IoT Hub. Reemplace el {deviceId} marcador de posición por el identificador de dispositivo (el nombre) del dispositivo registrado:

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)

Agregue los métodos siguientes para ejecutar los trabajos que llaman al método directo y al dispositivo gemelo:

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)

Agregue el código siguiente para programar los trabajos y actualizar sus estados. Incluya también la rutina 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()

Guarde y cierre el archivo scheduleJobService.py.

Ejecución de las aplicaciones

Ahora está preparado para ejecutar las aplicaciones.

En el símbolo del sistema, en el directorio de trabajo, ejecute el siguiente comando para iniciar la escucha del método directo de reinicio:
```
python simDevice.py
```
En otro símbolo del sistema, en el directorio de trabajo, ejecute el siguiente comando para desencadenar los trabajos a fin de bloquear la puerta y actualizar el dispositivo gemelo:
```
python scheduleJobService.py
```
Verá las respuestas del dispositivo al método directo y la actualización de los dispositivos gemelos en la consola.

Pasos siguientes

En este artículo, ha programado trabajos para ejecutar un método directo y actualizar las propiedades del dispositivo gemelo.

Para continuar explorando el IoT Hub y los patrones de administración de dispositivos, actualice una imagen en el tutorial de Actualización de Dispositivos para Azure IoT Hub mediante la Imagen de Referencia de Raspberry Pi 3 B+.

Compartir a través de

Programación y difusión de trabajos (Python)

Requisitos previos

Creación de una aplicación de dispositivo simulado

Obtener la cadena de conexión de IoT Hub

Programación de trabajos para llamar un método directo y actualización de las propiedades de un dispositivo gemelo

Ejecución de las aplicaciones

Pasos siguientes

Comentarios

Comentarios

Recursos adicionales