Planowanie i emisja zadań (Python)

Artykuł
10/01/2024

Usługa Azure IoT Hub umożliwia planowanie i śledzenie zadań, które aktualizują 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żywać zadania do wywoływania metody ponownego uruchamiania na 10 000 urządzeniach określonych przez zapytanie bliźniaczej reprezentacji urządzenia i zaplanowane w przyszłości. Następnie aplikacja może ś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 możliwości w następujących artykułach:

Bliźniacze reprezentacje urządzenia i właściwości: Wprowadzenie do bliźniaczych reprezentacji urządzeń oraz Omówienie bliźniaczych reprezentacji urządzeń i korzystanie z nich w usłudze IoT Hub
Metody bezpośrednie: Przewodnik dewelopera usługi IoT Hub — metody bezpośrednie

Uwaga

Funkcje opisane w tym artykule są dostępne tylko w warstwie Standardowa usługi IoT Hub. Aby uzyskać więcej informacji na temat warstw podstawowej i standardowej/bezpłatnej usługi IoT Hub, zobacz Wybieranie odpowiedniej warstwy usługi IoT Hub dla rozwiązania.

W tym artykule pokazano, jak utworzyć dwie aplikacje języka 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 inne zadanie 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 kompilowania aplikacji urządzeń i zaplecza.

Wymagania wstępne

Aktywne konto platformy Azure. (Jeśli go nie masz, możesz utworzyć bezpłatne konto próbne w zaledwie kilka minut).
Centrum IoT w ramach subskrypcji platformy Azure. Jeśli nie masz jeszcze centrum, możesz wykonać kroki opisane w temacie Tworzenie centrum IoT Hub.
Urządzenie zarejestrowane w centrum IoT Hub. Jeśli nie masz urządzenia w centrum IoT Hub, wykonaj kroki opisane w temacie Rejestrowanie urządzenia.
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 .

Ważne

Ten artykuł zawiera kroki łączenia urządzenia przy użyciu sygnatury dostępu współdzielonego, nazywanej również uwierzytelnianiem klucza symetrycznego. Ta metoda uwierzytelniania jest wygodna do testowania i oceny, ale uwierzytelnianie urządzenia przy użyciu certyfikatów X.509 jest bardziej bezpieczne. Aby dowiedzieć się więcej, zobacz Zabezpieczenia najlepszych rozwiązań > zabezpieczeń Zabezpieczenia zabezpieczeń Zabezpieczenia zabezpieczeń.

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 element parametry 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 ją tak, aby odpowiadała na metodę lockDoor , a także odbierała 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 prób (takie jak wycofywanie wykładnicze), zgodnie z sugestią w artykule Obsługa przejściowych błędów.

Pobieranie parametry połączenia usługi IoT Hub

W tym artykule utworzysz usługę zaplecza, która wywołuje metodę bezpośrednią na urządzeniu i aktualizuje bliźniacze 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 musisz 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 uzyskania parametry połączenia dla tych zasad, wykonaj następujące kroki:

Otwórz centrum IoT Hub w witrynie 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, 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.
W górnym 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 z usługą (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 dla parametry połączenia Podstawowej i zapisz wartość.

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

Ważne

Ten artykuł zawiera kroki nawiązywania połączenia z usługą przy użyciu sygnatury dostępu współdzielonego. Ta metoda uwierzytelniania jest wygodna do testowania i oceny, ale uwierzytelnianie w usłudze przy użyciu identyfikatora Entra firmy Microsoft lub tożsamości zarządzanych jest bardziej bezpieczne. Aby dowiedzieć się więcej, zobacz Security best practices Cloud security (Najlepsze rozwiązania > dotyczące zabezpieczeń w chmurze).

Planowanie zadań 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 . {IoTHubConnectionString} Zastąp symbol zastępczy parametry połączenia centrum IoT, który został wcześniej skopiowany w sekcji Pobieranie parametry połączenia centrum IoT Hub. {deviceId} Zastąp symbol zastępczy identyfikatorem urządzenia (nazwą) z 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źniacze reprezentacje 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. Obejmują również rutynę 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 zadania uruchamiania metody bezpośredniej i aktualizowania właściwości bliźniaczej reprezentacji urządzenia.

Aby kontynuować eksplorowanie wzorców zarządzania usługą IoT Hub i urządzeniami, zaktualizuj obraz w samouczku Dotyczącym aktualizacji urządzenia dla usługi Azure IoT Hub przy użyciu obrazu referencyjnego urządzenia Raspberry Pi 3 B+ .

Udostępnij za pośrednictwem

Planowanie i emisja zadań (Python)

Wymagania wstępne

Tworzenie aplikacji symulowanego urządzenia

Pobieranie parametry połączenia usługi IoT Hub

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

Uruchamianie aplikacji

Następne kroki

Opinia

Dodatkowe zasoby