Schemalägga och sända jobb (Python)

Artikel
10/01/2024

Använd Azure IoT Hub för att schemalägga och spåra jobb som uppdaterar miljontals enheter. Använd jobb för att:

Uppdatera önskade egenskaper
Uppdatera taggar
Anropa direktmetoder

Konceptuellt omsluter ett jobb en av dessa åtgärder och spårar körningens förlopp mot en uppsättning enheter, som definieras av en enhetstvillingfråga. En serverdelsapp kan till exempel använda ett jobb för att anropa en omstartsmetod på 10 000 enheter, som anges av en enhetstvillingfråga och schemaläggs vid en framtida tidpunkt. Programmet kan sedan spåra förloppet när var och en av dessa enheter tar emot och kör omstartsmetoden.

Läs mer om var och en av dessa funktioner i de här artiklarna:

Enhetstvillingar och egenskaper: Kom igång med enhetstvillingar och Förstå och använda enhetstvillingar i IoT Hub
Direktmetoder: Utvecklarguide för IoT Hub – direktmetoder

Kommentar

De funktioner som beskrivs i den här artikeln är endast tillgängliga på standardnivån för IoT Hub. Mer information om de grundläggande och standard-/kostnadsfria IoT Hub-nivåerna finns i Välj rätt IoT Hub-nivå för din lösning.

Den här artikeln visar hur du skapar två Python-appar:

En Python-simulerad enhetsapp, simDevice.py, som implementerar en direktmetod som kallas lockDoor, som kan anropas av serverdelsappen.
En Python-konsolapp, scheduleJobService.py, som skapar två jobb. Ett jobb anropar lockDoor-direktmetoden och ett annat jobb skickar önskade egenskapsuppdateringar till flera enheter.

Kommentar

Mer information om de SDK-verktyg som är tillgängliga för att skapa både enhets- och serverdelsappar finns i Azure IoT SDK:er .

Förutsättningar

Ett aktivt Azure-konto. (Om du inte har något konto kan du skapa ett kostnadsfritt konto på bara några minuter.)
En IoT-hubb i din Azure-prenumeration. Om du inte har någon hubb ännu kan du följa stegen i Skapa en IoT-hubb.
En enhet som är registrerad i din IoT-hubb. Om du inte har en enhet i din IoT-hubb följer du stegen i Registrera en enhet.
Python version 3.7 eller senare rekommenderas. Se till att använda en 32-bitars eller 64-bitars installation beroende på vad som krävs för din konfiguration. Se till att du lägger till Python i den plattformsspecifika miljövariabeln när du uppmanas att göra det under installationen.

Skapa en simulerad enhetsapp

I det här avsnittet skapar du en Python-konsolapp som svarar på en direktmetod som anropas av molnet, vilket utlöser en simulerad lockDoor-metod .

Viktigt!

Den här artikeln innehåller steg för att ansluta en enhet med en signatur för delad åtkomst, även kallad symmetrisk nyckelautentisering. Den här autentiseringsmetoden är praktisk för testning och utvärdering, men att autentisera en enhet med X.509-certifikat är en säkrare metod. Mer information finns i Metodtips > för säkerhet Anslutningssäkerhet.

Kör följande kommando i kommandotolken för att installera paketet azure-iot-device :
```
pip install azure-iot-device
```
Skapa en ny simDevice.py fil i arbetskatalogen med hjälp av en textredigerare.
Lägg till följande import instruktioner och variabler i början av filen simDevice.py . Ersätt deviceConnectionString med anslutningssträng för enheten som du skapade ovan:
```
import time
from azure.iot.device import IoTHubDeviceClient, MethodResponse

CONNECTION_STRING = "{deviceConnectionString}"
```

Definiera följande funktion, som instansierar en klient och konfigurerar den för att svara på metoden lockDoor , samt ta emot uppdateringar av enhetstvillingar:

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

Lägg till följande kod för att köra exemplet:

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

Spara och stäng filen simDevice.py .

Kommentar

För att hålla det enkelt implementerar den här artikeln inte en återförsöksprincip. I produktionskoden bör du implementera återförsöksprinciper (till exempel en exponentiell backoff), enligt vad som föreslås i artikeln Tillfälliga felhantering.

Hämta IoT Hub-anslutningssträng

I den här artikeln skapar du en serverdelstjänst som anropar en direktmetod på en enhet och uppdaterar enhetstvillingen. Tjänsten behöver tjänstens anslutningsbehörighet för att anropa en direktmetod på en enhet. Tjänsten behöver också behörighet att läsa och skriva register för att kunna läsa och skriva identitetsregistret. Det finns ingen standardprincip för delad åtkomst som endast innehåller dessa behörigheter, så du måste skapa en.

Följ dessa steg för att skapa en princip för delad åtkomst som beviljar behörigheter för tjänstanslutning, registerläsning och registerskrivning och för att få en anslutningssträng för den här principen:

Öppna din IoT-hubb i Azure Portal. Det enklaste sättet att komma till din IoT-hubb är att välja Resursgrupper, välja resursgruppen där din IoT-hubb finns och sedan välja din IoT-hubb i listan över resurser.
Välj Principer för delad åtkomst i den vänstra rutan i din IoT-hubb.
På den översta menyn ovanför listan med principer väljer du Lägg till princip för delad åtkomst.
I fönstret Lägg till princip för delad åtkomst anger du ett beskrivande namn för din princip, till exempel: serviceAndRegistryReadWrite. Under Behörigheter väljer du Registerskrivning och Service Connect (Registerläsning väljs automatiskt när du väljer Registerskrivning) och väljer sedan Lägg till.
På sidan Principer för delad åtkomst väljer du din nya princip i listan över principer.
I det nya fönstret som visas väljer du kopieringsikonen för den primära anslutningssträng och sparar värdet.

Mer information om principer och behörigheter för delad åtkomst i IoT Hub finns i Åtkomstkontroll och behörigheter.

Viktigt!

Den här artikeln innehåller steg för att ansluta till en tjänst med hjälp av en signatur för delad åtkomst. Den här autentiseringsmetoden är praktisk för testning och utvärdering, men autentisering till en tjänst med Microsoft Entra-ID eller hanterade identiteter är en säkrare metod. Mer information finns i Metodtips för > säkerhet Molnsäkerhet.

Schemalägga jobb för att anropa en direktmetod och uppdatera en enhetstvillings egenskaper

I det här avsnittet skapar du en Python-konsolapp som initierar ett fjärrlåsDoor på en enhet med hjälp av en direktmetod och uppdaterar även enhetstvillingens önskade egenskaper.

Kör följande kommando i kommandotolken för att installera paketet azure-iot-hub :
```
pip install azure-iot-hub
```
Skapa en ny scheduleJobService.py fil i arbetskatalogen med hjälp av en textredigerare.

Lägg till följande import instruktioner och variabler i början av filen scheduleJobService.py . {IoTHubConnectionString} Ersätt platshållaren med IoT-hubben anslutningssträng du kopierade tidigare i Hämta IoT Hub-anslutningssträng. {deviceId} Ersätt platshållaren med enhets-ID :t (namnet) från den registrerade enheten:

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)

Lägg till följande metoder för att köra jobben som anropar direktmetoden och enhetstvillingen:

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)

Lägg till följande kod för att schemalägga jobben och uppdatera jobbstatusen. Inkludera även rutinen 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()

Spara och stäng filen scheduleJobService.py .

Kör programmen

Nu är det dags att köra programmen.

I kommandotolken i arbetskatalogen kör du följande kommando för att börja lyssna efter direktmetoden för omstart:
```
python simDevice.py
```
I en annan kommandotolk i arbetskatalogen kör du följande kommando för att utlösa jobben för att låsa dörren och uppdatera tvillingen:
```
python scheduleJobService.py
```
Du ser enhetssvaren på direktmetoden och enhetstvillingarna uppdateras i konsolen.

Nästa steg

I den här artikeln har du schemalagt jobb för att köra en direktmetod och uppdatera enhetstvillingens egenskaper.

Om du vill fortsätta utforska IoT Hub- och enhetshanteringsmönster uppdaterar du en avbildning i självstudiekursen Enhetsuppdatering för Azure IoT Hub med hjälp av Raspberry Pi 3 B+-referensbilden.

Dela via

Schemalägga och sända jobb (Python)

Förutsättningar

Skapa en simulerad enhetsapp

Hämta IoT Hub-anslutningssträng

Schemalägga jobb för att anropa en direktmetod och uppdatera en enhetstvillings egenskaper

Kör programmen

Nästa steg

Feedback

Ytterligare resurser