Planowanie i emisja zadań (Python)

Artykuł
06/24/2023

Użyj Azure IoT Hub, aby zaplanować i śledzić zadania aktualizujące miliony urządzeń. Użyj zadań, aby:

Aktualizowanie żądanych właściwości
Aktualizowanie tagów
Wywoływanie metod bezpośrednich

Koncepcyjnie zadanie opakowuje jedną z tych akcji i śledzi postęp wykonywania względem zestawu urządzeń, który jest definiowany przez zapytanie bliźniaczej reprezentacji urządzenia. Na przykład aplikacja zaplecza może użyć zadania w celu wywołania metody ponownego rozruchu na 10 000 urządzeniach określonych przez zapytanie bliźniaczej reprezentacji urządzenia i zaplanowanych w przyszłości. Ta aplikacja może następnie śledzić postęp, gdy każde z tych urządzeń odbiera i wykonuje metodę ponownego rozruchu.

Dowiedz się więcej o każdej z tych funkcji w następujących artykułach:

Bliźniacze reprezentacje urządzenia i właściwości: Wprowadzenie do bliźniaczych reprezentacji urządzeń oraz Opis bliźniaczych reprezentacji urządzeń i korzystanie z nich w IoT Hub
Metody bezpośrednie: przewodnik dla deweloperów IoT Hub — metody bezpośrednie

Uwaga

Funkcje opisane w tym artykule są dostępne tylko w warstwie Standardowa IoT Hub. Aby uzyskać więcej informacji o warstwach podstawowa i Standardowa/Bezpłatna IoT Hub, zobacz Wybieranie odpowiedniej warstwy IoT Hub dla rozwiązania.

W tym artykule przedstawiono sposób tworzenia dwóch aplikacji w języku Python:

Aplikacja urządzenia symulowanego w języku Python , simDevice.py, która implementuje metodę bezpośrednią o nazwie lockDoor, którą można wywołać przez aplikację zaplecza.
Aplikacja konsolowa języka Python , scheduleJobService.py, która tworzy dwa zadania. Jedno zadanie wywołuje metodę bezpośrednią lockDoor , a drugie wysyła żądane aktualizacje właściwości do wielu urządzeń.

Uwaga

Zobacz Zestawy SDK usługi Azure IoT , aby uzyskać więcej informacji na temat narzędzi zestawu SDK dostępnych do tworzenia aplikacji zarówno urządzeń, jak i zaplecza.

Wymagania wstępne

Aktywne konto platformy Azure. (Jeśli nie masz konta, możesz utworzyć bezpłatne konto w ciągu zaledwie kilku minut).
Centrum IoT. Utwórz go przy użyciu interfejsu wiersza polecenia lub Azure Portal.
Zarejestrowane urządzenie. Zarejestruj go w Azure Portal.
Zalecane jest użycie języka Python w wersji 3.7 lub nowszej. Upewnij się, że używasz 32-bitowej lub 64-bitowej instalacji zgodnie z wymaganiami konfiguracji. Po wyświetleniu monitu podczas instalacji upewnij się, że język Python został dodany do zmiennej środowiskowej specyficznej dla platformy.

Tworzenie aplikacji symulowanego urządzenia

W tej sekcji utworzysz aplikację konsolową języka Python, która reaguje na metodę bezpośrednią wywoływaną przez chmurę, która wyzwala symulowaną metodę lockDoor .

W wierszu polecenia uruchom następujące polecenie, aby zainstalować pakiet azure-iot-device :
```
pip install azure-iot-device
```
Za pomocą edytora tekstów utwórz nowy plik simDevice.py w katalogu roboczym.
Dodaj następujące import instrukcje i zmienne na początku pliku simDevice.py . Zastąp deviceConnectionString ciąg parametrami połączenia utworzonego powyżej urządzenia:
```
import time
from azure.iot.device import IoTHubDeviceClient, MethodResponse

CONNECTION_STRING = "{deviceConnectionString}"
```

Zdefiniuj następującą funkcję, która utworzy wystąpienie klienta i skonfiguruje go do reagowania na metodę lockDoor , a także odbiera aktualizacje bliźniaczej reprezentacji urządzenia:

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

Dodaj następujący kod, aby uruchomić przykład:

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

Zapisz i zamknij plik simDevice.py .

Uwaga

Aby zachować prostotę, ten artykuł nie implementuje zasad ponawiania. W kodzie produkcyjnym należy zaimplementować zasady ponawiania (takie jak wycofywanie wykładnicze), zgodnie z sugestią w artykule Obsługa błędów przejściowych.

Pobieranie parametrów połączenia IoT Hub

W tym artykule utworzysz usługę zaplecza, która wywołuje metodę bezpośrednią na urządzeniu i aktualizuje bliźniaczą reprezentację urządzenia. Usługa wymaga uprawnienia połączenia usługi , aby wywołać metodę bezpośrednią na urządzeniu. Usługa wymaga również uprawnień do odczytu rejestru i zapisu rejestru w celu odczytu i zapisu rejestru tożsamości. Nie ma domyślnych zasad dostępu współdzielonego, które zawierają tylko te uprawnienia, więc należy je utworzyć.

Aby utworzyć zasady dostępu współdzielonego, które udzielają uprawnień do nawiązywania połączenia z usługą, odczytu rejestru i zapisu rejestru oraz pobierania parametrów połączenia dla tych zasad, wykonaj następujące kroki:

Otwórz centrum IoT w Azure Portal. Najprostszym sposobem uzyskania dostępu do centrum IoT jest wybranie pozycji Grupy zasobów, wybranie grupy zasobów, w której znajduje się centrum IoT Hub, a następnie wybranie centrum IoT z listy zasobów.
W okienku po lewej stronie centrum IoT hub wybierz pozycję Zasady dostępu współdzielonego.
Z górnego menu nad listą zasad wybierz pozycję Dodaj zasady dostępu współdzielonego.
W okienku Dodawanie zasad dostępu współdzielonego wprowadź opisową nazwę zasad; na przykład : serviceAndRegistryReadWrite. W obszarze Uprawnienia wybierz pozycję Zapis rejestru i połączenie usługi (odczyt rejestru jest wybierany automatycznie po wybraniu pozycji Zapis rejestru), a następnie wybierz pozycję Dodaj.
Po powrocie na stronę Zasady dostępu współdzielonego wybierz nowe zasady z listy zasad.
W wyświetlonym nowym okienku wybierz ikonę kopiowania podstawowych parametrów połączenia i zapisz wartość.

Aby uzyskać więcej informacji na temat IoT Hub zasad dostępu współdzielonego i uprawnień, zobacz Kontrola dostępu i uprawnienia.

Planowanie zadań na potrzeby wywoływania metody bezpośredniej i aktualizowania właściwości bliźniaczej reprezentacji urządzenia

W tej sekcji utworzysz aplikację konsolową języka Python, która inicjuje zdalne blokowanieDoor na urządzeniu przy użyciu metody bezpośredniej, a także aktualizuje żądane właściwości bliźniaczej reprezentacji urządzenia.

W wierszu polecenia uruchom następujące polecenie, aby zainstalować pakiet azure-iot-hub :
```
pip install azure-iot-hub
```
Za pomocą edytora tekstów utwórz nowy plik scheduleJobService.py w katalogu roboczym.

Dodaj następujące import instrukcje i zmienne na początku pliku scheduleJobService.py . Zastąp {IoTHubConnectionString} symbol zastępczy parametrami połączenia centrum IoT Hub skopiowanymi wcześniej w sekcji Pobieranie parametrów połączenia centrum IoT. Zastąp {deviceId} symbol zastępczy identyfikatorem urządzenia (nazwą) zarejestrowanego urządzenia:

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)

Dodaj następujące metody, aby uruchomić zadania wywołujące metodę bezpośrednią i bliźniaczą reprezentację urządzenia:

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)

Dodaj następujący kod, aby zaplanować zadania i zaktualizować stan zadania. Należy również uwzględnić procedurę 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()

Zapisz i zamknij plik scheduleJobService.py .

Uruchamianie aplikacji

Teraz można uruchomić aplikacje.

W wierszu polecenia w katalogu roboczym uruchom następujące polecenie, aby rozpocząć nasłuchiwanie metody bezpośredniej ponownego rozruchu:
```
python simDevice.py
```
W innym wierszu polecenia w katalogu roboczym uruchom następujące polecenie, aby wyzwolić zadania w celu zablokowania drzwi i zaktualizowania bliźniaczej reprezentacji:
```
python scheduleJobService.py
```
W konsoli zobaczysz odpowiedzi urządzenia na metodę bezpośrednią i bliźniacze reprezentacje urządzeń.

Następne kroki

W tym artykule zaplanowano uruchamianie metody bezpośredniej i aktualizowanie właściwości bliźniaczej reprezentacji urządzenia.

Aby kontynuować eksplorowanie wzorców IoT Hub i zarządzania urządzeniami, zaktualizuj obraz w usłudze Device Update na potrzeby Azure IoT Hub samouczka przy użyciu obrazu referencyjnego urządzenia Raspberry Pi 3 B+.

Share via

Planowanie i emisja zadań (Python)

Wymagania wstępne

Tworzenie aplikacji symulowanego urządzenia

Pobieranie parametrów połączenia IoT Hub

Planowanie zadań na potrzeby wywoływania metody bezpośredniej i aktualizowania właściwości bliźniaczej reprezentacji urządzenia

Uruchamianie aplikacji

Następne kroki

Dodatkowe zasoby