Planowanie i emisja zadań (Python)
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ąpdeviceConnectionString
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+ .