Agendar e transmitir trabalhos (Python)

Artigo
10/01/2024

Use o Hub IoT do Azure para agendar e rastrear trabalhos que atualizam milhões de dispositivos. Use trabalhos para:

Atualizar as propriedades pretendidas
Atualizar tags
Invocar métodos diretos

Conceitualmente, um trabalho encapsula uma dessas ações e rastreia o progresso da execução em relação a um conjunto de dispositivos, que é definido por uma consulta gêmea de dispositivo. Por exemplo, um aplicativo back-end pode usar um trabalho para invocar um método de reinicialização em 10.000 dispositivos, especificado por uma consulta gêmea de dispositivo e agendado em um momento futuro. Esse aplicativo pode então acompanhar o progresso à medida que cada um desses dispositivos recebe e executa o método de reinicialização.

Saiba mais sobre cada um desses recursos nestes artigos:

Gêmeos e propriedades de dispositivos: Introdução aos gêmeos de dispositivo e Compreender e usar gêmeos de dispositivo no Hub IoT
Métodos diretos: Guia do desenvolvedor do Hub IoT - métodos diretos

Nota

Os recursos descritos neste artigo estão disponíveis somente na camada padrão do Hub IoT. Para obter mais informações sobre as camadas básica e padrão/gratuita do Hub IoT, consulte Escolha a camada certa do Hub IoT para sua solução.

Este artigo mostra como criar dois aplicativos Python:

Um aplicativo de dispositivo simulado Python, simDevice.py, que implementa um método direto chamado lockDoor, que pode ser chamado pelo aplicativo back-end.
Um aplicativo de console Python, scheduleJobService.py, que cria dois trabalhos. Um trabalho chama o método direto lockDoor e outro trabalho envia atualizações de propriedade desejadas para vários dispositivos.

Nota

Consulte SDKs do Azure IoT para obter mais informações sobre as ferramentas SDK disponíveis para criar aplicativos de dispositivo e 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 em sua assinatura do Azure. Se você ainda não tiver um hub, siga as etapas em Criar um hub IoT.
Um dispositivo registrado em seu hub IoT. Se você não tiver um dispositivo em seu hub IoT, siga as etapas em Registrar um dispositivo.
Python versão 3.7 ou posterior é recomendado. 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 seção, você cria um aplicativo de console Python que responde a um método direto chamado pela nuvem, que aciona um método lockDoor simulado.

Importante

Este artigo inclui etapas para conectar um dispositivo usando uma assinatura de acesso compartilhado, também chamada de autenticação de chave simétrica. Esse método de autenticação é conveniente para testes e avaliações, mas autenticar um dispositivo usando certificados X.509 é uma abordagem mais segura. Para saber mais, consulte Práticas > recomendadas de segurança Segurança de conexão.

No prompt de comando, execute o seguinte comando para instalar o pacote azure-iot-device :
```
pip install azure-iot-device
```
Usando um editor de texto, crie um novo arquivo de simDevice.py em seu diretório de trabalho.
Adicione as seguintes import instruções e variáveis no início do arquivo de simDevice.py . Substitua deviceConnectionString pela cadeia de conexão do dispositivo criado 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 gêmeas do dispositivo:

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

Salve e feche o arquivo simDevice.py .

Nota

Para simplificar, este artigo não implementa uma política de novas tentativas. No código de produção, você deve implementar políticas de repetição (como um backoff exponencial), conforme sugerido no artigo, Tratamento de falhas transitórias.

Obter a cadeia de conexão do Hub IoT

Neste artigo, você cria um serviço de back-end que invoca um método direto em um dispositivo e atualiza o gêmeo do dispositivo. O serviço precisa da permissão de conexão de serviço para chamar um método direto em um dispositivo. O serviço também precisa das permissões de leitura e gravação do registro para ler e gravar o registro de identidade. Não há nenhuma política de acesso compartilhado padrão que contenha apenas essas permissões, portanto, você precisa criar uma.

Para criar uma política de acesso compartilhado que conceda permissões de conexão de serviço, leitura do Registro e gravação do Registro e para obter uma cadeia de conexão para essa política, siga estas etapas:

Abra seu hub IoT no portal do Azure. A maneira mais fácil de chegar ao seu hub IoT é selecionar Grupos de recursos, selecionar o grupo de recursos onde o hub IoT está localizado e, em seguida, selecionar seu hub IoT na lista de recursos.
No painel esquerdo do hub IoT, selecione Políticas de acesso compartilhado.
No menu superior acima da lista de políticas, selecione Adicionar política de acesso compartilhado.
No painel Adicionar política de acesso compartilhado, insira um nome descritivo para sua política, por exemplo: serviceAndRegistryReadWrite. Em Permissões, selecione Gravação do Registro e Conexão de Serviço (a Leitura do Registro é selecionada automaticamente quando você seleciona Gravação do Registro) e selecione Adicionar.
De volta à página Políticas de acesso compartilhado, selecione sua nova política na lista de políticas.
No novo painel exibido, selecione o ícone de cópia para a cadeia de conexão principal e salve o valor.

Para obter mais informações sobre políticas e permissões de acesso compartilhado do Hub IoT, consulte Controle de acesso e permissões.

Importante

Este artigo inclui etapas para se conectar a um serviço usando uma assinatura de acesso compartilhado. Esse método de autenticação é conveniente para teste e avaliação, mas autenticar em um serviço com ID do Microsoft Entra ou identidades gerenciadas é uma abordagem mais segura. Para saber mais, consulte Práticas > recomendadas de segurança Segurança na nuvem.

Agendar trabalhos para chamar um método direto e atualizar as propriedades de um gêmeo de dispositivo

Nesta seção, você cria um aplicativo de console Python que inicia um lockDoor remoto em um dispositivo usando um método direto e também atualiza as propriedades desejadas do gêmeo do dispositivo.

No prompt de comando, execute o seguinte comando para instalar o pacote azure-iot-hub :
```
pip install azure-iot-hub
```
Usando um editor de texto, crie um novo arquivo de scheduleJobService.py em seu diretório de trabalho.

Adicione as seguintes import instruções e variáveis no início do arquivo de scheduleJobService.py . Substitua o espaço reservado {IoTHubConnectionString} pela cadeia de conexão do hub IoT copiada anteriormente em Obter a cadeia de conexão do hub IoT. Substitua o espaço reservado {deviceId} pelo ID do dispositivo (o nome) do 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)

Adicione os seguintes métodos para executar os trabalhos que chamam o método direto e o gêmeo de dispositivo:

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 os trabalhos e atualizar o status do trabalho. 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()

Salve e feche o arquivo scheduleJobService.py .

Executar as aplicações

Pode agora executar as aplicações.

No prompt de comando no diretório de trabalho, execute o seguinte comando para começar a ouvir o método direto de reinicialização:
```
python simDevice.py
```
Em outro prompt de comando em seu diretório de trabalho, execute o seguinte comando para acionar os trabalhos para bloquear a porta e atualizar o gêmeo:
```
python scheduleJobService.py
```
Você vê as respostas do dispositivo ao método direto e a atualização de gêmeos de dispositivo no console.

Próximos passos

Neste artigo, você agendou trabalhos para executar um método direto e atualizar as propriedades do gêmeo do dispositivo.

Para continuar explorando o Hub IoT e os padrões de gerenciamento de dispositivos, atualize uma imagem no tutorial Atualização de Dispositivo para o Hub IoT do Azure usando a Imagem de Referência do Raspberry Pi 3 B+.

Partilhar via

Agendar e transmitir trabalhos (Python)

Pré-requisitos

Criar uma aplicação de dispositivo simulada

Obter a cadeia de conexão do Hub IoT

Agendar trabalhos para chamar um método direto e atualizar as propriedades de um gêmeo de dispositivo

Executar as aplicações

Próximos passos

Comentários

Recursos adicionais