Agendar e transmitir tarefas (Python)

Artigo
03/17/2023

Utilize Hub IoT do Azure para agendar e controlar tarefas que atualizam milhões de dispositivos. Utilizar tarefas para:

Atualizar as propriedades pretendidas
Atualizar etiquetas
Invocar métodos diretos

Conceptualmente, uma tarefa encapsula uma destas ações e controla o progresso da execução num conjunto de dispositivos, que é definido por uma consulta de dispositivo duplo. Por exemplo, uma aplicação de back-end pode utilizar uma tarefa para invocar um método de reinício em 10 000 dispositivos, especificado por uma consulta de dispositivo duplo e agendado numa hora futura. Essa aplicação pode, em seguida, controlar o progresso à medida que cada um desses dispositivos recebe e executa o método de reinício.

Saiba mais sobre cada uma destas capacidades nestes artigos:

Dispositivo duplo e propriedades: introdução aos dispositivos duplos e Compreender e utilizar dispositivos duplos no Hub IoT
Métodos diretos: Hub IoT guia para programadores – métodos diretos

Nota

As funcionalidades descritas neste artigo só estão disponíveis no escalão padrão do Hub IoT. Para obter mais informações sobre os escalões de Hub IoT básico e standard/gratuito, consulte Escolher o escalão de Hub IoT certo para a sua solução.

Este artigo mostra-lhe como criar duas aplicações Python:

Uma aplicação de dispositivo simulado python, simDevice.py, que implementa um método direto chamado lockDoor, que pode ser chamado pela aplicação de back-end.
Uma aplicação de consola Python, scheduleJobService.py, que cria duas tarefas. Uma tarefa chama o método direto lockDoor e outra envia atualizações de propriedade pretendidas para vários dispositivos.

Nota

Veja SDKs do Azure IoT para obter mais informações sobre as ferramentas do SDK disponíveis para criar aplicações de dispositivos e de back-end.

Pré-requisitos

Uma conta ativa do Azure. (Se não tiver uma conta, pode criar uma conta gratuita em apenas alguns minutos.)
Um hub IoT. Crie uma com a CLI ou a portal do Azure.
Um dispositivo registado. Registe um na portal do Azure.
Recomenda-se a versão 3.7 ou posterior do Python . Certifique-se de que utiliza a instalação de 32 ou 64 bits, conforme exigido pela sua configuração. Quando lhe for pedido durante a instalação, confirme que adiciona Python à variável de ambiente específica da sua plataforma.

Criar uma aplicação de dispositivo simulada

Nesta secção, vai criar uma aplicação de consola Python que responde a um método direto chamado pela cloud, que aciona um método lockDoor simulado.

Na linha de comandos, execute o seguinte comando para instalar o pacote azure-iot-device :
```
pip install azure-iot-device
```
Com um editor de texto, crie um novo ficheiro simDevice.py no seu diretório de trabalho.
Adicione as seguintes import instruções e variáveis no início do ficheiro simDevice.py . Substitua deviceConnectionString pela cadeia de ligação do dispositivo que criou acima:
```
import time
from azure.iot.device import IoTHubDeviceClient, MethodResponse

CONNECTION_STRING = "{deviceConnectionString}"
```

Defina a seguinte função, que irá instanciar um cliente e configurá-lo para responder ao método lockDoor , bem como receber atualizações de dispositivo duplo:

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

Adicione o seguinte código para executar o exemplo:

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 e feche o ficheiro simDevice.py .

Nota

Para manter as coisas simples, este artigo não implementa uma política de repetição. No código de produção, deve implementar políticas de repetição (como um recuo exponencial), conforme sugerido no artigo Processamento transitório de falhas.

Obter a cadeia de ligação Hub IoT

Neste artigo, vai criar um serviço de back-end que invoca um método direto num dispositivo e atualiza o dispositivo duplo. O serviço precisa da permissão de ligação do serviço para chamar um método direto num dispositivo. O serviço também precisa das permissões de leitura e escrita do registo para ler e escrever o registo de identidade. Não existe uma política de acesso partilhado predefinida que contenha apenas estas permissões, pelo que tem de criar uma.

Para criar uma política de acesso partilhado que conceda permissões de ligação ao serviço, leitura do registo e escrita do registo e para obter uma cadeia de ligação para esta política, siga estes passos:

Abra o hub IoT no portal do Azure. A forma mais fácil de aceder ao hub IoT é selecionar Grupos de recursos, selecionar o grupo de recursos onde está localizado o hub IoT e, em seguida, selecionar o hub IoT na lista de recursos.
No painel esquerdo do hub IoT, selecione Políticas de acesso partilhado.
No menu superior acima da lista de políticas, selecione Adicionar política de acesso partilhado.
No painel Adicionar política de acesso partilhado , introduza um nome descritivo para a sua política; por exemplo: serviceAndRegistryReadWrite. Em Permissões, selecione Escrita do Registo e Ligação do Serviço (a opção Leitura do Registo é selecionada automaticamente quando seleciona Escrita do Registo) e, em seguida, selecione Adicionar.
Novamente na página Políticas de acesso partilhado , selecione a nova política na lista de políticas.
No novo painel que é apresentado, selecione o ícone de cópia da cadeia de ligação Primária e guarde o valor.

Para obter mais informações sobre Hub IoT permissões e políticas de acesso partilhado, veja Controlo de acesso e permissões.

Agendar tarefas para chamar um método direto e atualizar as propriedades de um dispositivo duplo

Nesta secção, vai criar uma aplicação de consola Python que inicia um bloqueio remotoDoor num dispositivo através de um método direto e também atualiza as propriedades pretendidas do dispositivo duplo.

Na linha de comandos, execute o seguinte comando para instalar o pacote azure-iot-hub :
```
pip install azure-iot-hub
```
Com um editor de texto, crie um novo ficheiro scheduleJobService.py no seu diretório de trabalho.

Adicione as seguintes import instruções e variáveis no início do ficheiro scheduleJobService.py . Substitua o {IoTHubConnectionString} marcador de posição pela cadeia de ligação do hub IoT que copiou anteriormente em Obter a cadeia de ligação do hub IoT. Substitua o {deviceId} marcador de posição pelo ID do dispositivo (o nome) do dispositivo registado:

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)

Adicione os seguintes métodos para executar as tarefas que chamam o método direto e o dispositivo duplo:

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)

Adicione o seguinte código para agendar as tarefas e atualizar o estado da tarefa. Inclua também a main rotina:

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 e feche o ficheiro scheduleJobService.py .

Executar as aplicações

Pode agora executar as aplicações.

Na linha de comandos do diretório de trabalho, execute o seguinte comando para começar a escutar o método direto de reinício:
```
python simDevice.py
```
Noutra linha de comandos do diretório de trabalho, execute o seguinte comando para acionar as tarefas para bloquear a porta e atualizar o duplo:
```
python scheduleJobService.py
```
Verá as respostas do dispositivo ao método direto e à atualização dos dispositivos duplos na consola do .

Passos seguintes

Neste artigo, agendou trabalhos para executar um método direto e atualizar as propriedades do dispositivo duplo.

Para continuar a explorar padrões de gestão de dispositivos e Hub IoT, atualize uma imagem no tutorial Atualização de Dispositivos para Hub IoT do Azure com a Imagem de Referência raspberry Pi 3 B+.

Share via

Agendar e transmitir tarefas (Python)

Pré-requisitos

Criar uma aplicação de dispositivo simulada

Obter a cadeia de ligação Hub IoT

Agendar tarefas para chamar um método direto e atualizar as propriedades de um dispositivo duplo

Executar as aplicações

Passos seguintes

Recursos adicionais