작업 예약 및 브로드캐스트(Python)

아티클
10/01/2024

Azure IoT Hub를 사용하여 수백만 대의 디바이스를 업데이트하는 작업을 예약하고 추적합니다. 작업을 사용하여 다음을 수행합니다.

desired 속성 업데이트
tags 업데이트
직접 메서드 호출

개념적으로 작업(job)은 이러한 작업(action) 중 하나를 래핑하고 디바이스 쌍 쿼리로 정의된 디바이스 집합에 대해 실행 진행 상태를 추적합니다. 예를 들어 백 엔드 앱은 작업을 사용하여 디바이스 쌍 쿼리로 지정되고 향후에 예약된 10,000개 디바이스에서 다시 부팅 메서드를 호출할 수 있습니다. 그러면 이러한 각 디바이스에서 다시 부팅 메서드를 받고 실행할 때 해당 애플리케이션에서 진행 상황을 추적할 수 있습니다.

이러한 기능에 대한 자세한 내용은 다음 문서를 참조하세요.

디바이스 쌍 및 속성: 디바이스 쌍 시작 및 IoT Hub의 디바이스 쌍 이해 및 사용
직접 메서드: IoT Hub 개발자 가이드 - 직접 메서드

참고 항목

이 문서에서 설명하는 기능은 IoT Hub의 표준 계층에서만 사용할 수 있습니다. 기본 및 표준/무료 IoT Hub 계층에 대한 자세한 내용은 솔루션에 적합한 IoT Hub 계층 선택을 참조하세요.

이 문서에서는 다음 두 개의 Python 앱을 만드는 방법을 보여 줍니다.

simDevice.js - 백 엔드 앱에서 호출할 수 있는 lockDoor라는 직접 메서드를 구현하는 Python 시뮬레이션된 디바이스 앱입니다.
scheduleJobService.py - 두 개의 작업을 만드는 Python 콘솔 앱입니다. 한 작업은 lockDoor 직접 메서드를 호출하고 다른 작업은 원하는 속성 업데이트를 여러 디바이스로 보냅니다.

참고 항목

디바이스 및 백 엔드 앱을 빌드하는 데 사용할 수 있는 SDK 도구에 대한 자세한 내용은 Azure IoT SDK를 참조하세요.

필수 조건

활성 Azure 계정. 계정이 없는 경우 몇 분 만에 무료 계정을 만들 수 있습니다.
Azure 구독의 IoT Hub 아직 허브가 없는 경우 IoT Hub 만들기의 단계를 따를 수 있습니다.
IoT Hub에 등록된 디바이스. IoT 허브에 디바이스가 없으면 디바이스 등록의 단계를 따릅니다.
Python 버전 3.7 이상을 사용하는 것이 좋습니다. 설치 프로그램의 요구 사항에 따라 32비트 또는 64비트 설치를 사용해야 합니다. 설치하는 동안 메시지가 나타나면 플랫폼별 환경 변수에 Python을 추가해야 합니다.

시뮬레이션된 디바이스 앱 만들기

이 섹션에서는 클라우드에서 호출한 메서드에 응답하는 Python 콘솔 앱을 만듭니다. 이 메서드는 시뮬레이션된 LockDoor 메서드를 트리거합니다.

Important

이 문서에서는 공유 액세스 서명(대칭 키 인증이라고도 함)을 사용하여 디바이스를 연결하는 단계를 설명합니다. 이 인증 방법은 테스트와 평가에 편리하지만, X.509 인증서를 사용하여 디바이스를 인증하는 것이 더 안전한 방식입니다. 자세한 내용은 보안 모범 사례 > 연결 보안을 참조하세요.

명령 프롬프트에서 다음 명령을 실행하여 azure-iot-device 패키지를 설치합니다.
```
pip install azure-iot-device
```
텍스트 편집기를 사용하여 작업 디렉터리에 새 simDevice.py 파일을 만듭니다.
simDevice.py 파일의 시작 부분에 다음 import 문 및 변수를 추가합니다. deviceConnectionString을 위에서 만든 디바이스의 연결 문자열로 바꿉니다.
```
import time
from azure.iot.device import IoTHubDeviceClient, MethodResponse

CONNECTION_STRING = "{deviceConnectionString}"
```

클라이언트를 인스턴스화하고, lockDoor 메서드에 응답하며 디바이스 쌍 업데이트를 수신하도록 클라이언트를 구성하는 다음 함수를 정의합니다.

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

다음 코드를 추가하여 샘플을 실행합니다.

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

simDevice.py 파일을 저장하고 닫습니다.

참고 항목

간단히 하기 위해 이 문서에서는 다시 시도 정책을 구현하지 않습니다. 프로덕션 코드에서는 문서 일시적인 오류 처리에서 제시한 대로 다시 시도 정책(예: 지수 백오프)을 구현해야 합니다.

IoT Hub 연결 문자열 가져오기

이 문서에서는 디바이스에서 직접 메서드를 호출하고 디바이스 쌍을 업데이트하는 백 엔드 서비스를 만듭니다. 서비스는 디바이스에서 직접 메서드를 호출하려면 서비스 연결 권한이 필요합니다. 또한 서비스에는 ID 레지스트리를 읽고 쓰기 위한 레지스트리 읽기 및 레지스트리 쓰기 권한이 필요합니다. 이러한 권한만 포함하는 기본 공유 액세스 정책은 없으므로, 만들어야 합니다.

서비스 연결 및 레지스트리 읽기 및 레지스트리 쓰기 권한을 부여하는 공유 액세스 정책을 만들고, 이 정책에 대한 연결 문자열을 가져오려면 다음 단계를 수행합니다.

Azure Portal에서 IoT 허브를 엽니다. IoT 허브로 가져오는 가장 쉬운 방법은 리소스 그룹을 선택하고 IoT 허브가 위치한 리소스 그룹에서 리소스 그룹을 선택한 다음, 리소스 목록에서 IoT 허브를 선택하는 것입니다.
IoT Hub의 왼쪽 창에서 공유 액세스 정책을 선택합니다.
정책 목록의 위쪽 메뉴에서 공유 액세스 정책 추가를 선택합니다.
공유 액세스 정책 추가 창에서 정책을 설명하는 이름(예: serviceAndRegistryReadWrite)을 입력합니다. 권한아래에서 레지스트리 쓰기 및 서비스 연결을 선택한 다음(레지스트리 쓰기를 선택하면 레지스트리 읽기가 자동으로 선택됨), 추가를 선택합니다.
공유 액세스 정책 창으로 돌아가서 정책 목록에서 새 정책을 선택합니다.
표시되는 새 창에서 기본 연결 문자열의 복사 아이콘을 선택하고 값을 저장합니다.

IoT Hub 공유 액세스 정책 및 사용 권한에 대한 자세한 내용은 액세스 제어 및 권한을 참조하세요.

Important

이 문서에서는 공유 액세스 서명을 사용하여 서비스에 연결하는 단계를 설명합니다. 이 인증 방법은 테스트와 평가에 편리하지만, Microsoft Entra ID나 관리 ID를 사용하여 서비스를 인증하는 것이 더 안전한 방식입니다. 자세한 내용은 보안 모범 사례 > 클라우드 보안을 참조하세요.

직접 메서드를 호출하고 디바이스 쌍의 속성을 업데이트하기 위한 작업 예약

이 섹션에서는 직접 메서드를 사용하여 디바이스에서 원격 LockDoor를 시작하는 Python 콘솔 앱을 만들고 디바이스 쌍의 desired 속성을 업데이트합니다.

명령 프롬프트에서 다음 명령을 실행하여 azure-iot-hub 패키지를 설치합니다.
```
pip install azure-iot-hub
```
텍스트 편집기를 사용하여 작업 디렉터리에 새 scheduleJobService.py 파일을 만듭니다.

scheduleJobService.py 파일의 시작 부분에 다음 import 문 및 변수를 추가합니다. {IoTHubConnectionString} 자리 표시자 값을 이전에 IoT 허브 연결 문자열 가져오기에서 복사한 IoT 허브 연결 문자열로 바꿉니다. {deviceId} 자리 표시자를 등록된 디바이스의 디바이스 ID(이름)로 바꿉니다.

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)

직접 메서드와 디바이스 쌍을 호출하는 작업을 실행하는 다음 메서드를 추가합니다.

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)

작업을 예약하고 작업 상태를 업데이트하도록 다음 코드를 추가합니다. 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()

scheduleJobService.py 파일을 저장하고 닫습니다.

애플리케이션 실행

이제 애플리케이션을 실행할 준비가 되었습니다.

작업 디렉터리의 명령 프롬프트에서 다음 명령을 실행하여 다시 시작 직접 메서드에 대한 수신을 시작합니다.
```
python simDevice.py
```
작업 디렉터리의 명령 프롬프트에서 다음 명령을 실행하여 도어를 잠그고 쌍을 업데이트하는 작업을 트리거합니다.
```
python scheduleJobService.py
```
콘솔에서 직접 메서드와 디바이스 쌍 업데이트에 대한 디바이스 응답을 확인합니다.

다음 단계

이 문서에서는 직접 메서드를 실행하고 디바이스 쌍의 속성을 업데이트하는 작업을 예약했습니다.

IoT Hub 및 디바이스 관리 패턴을 계속 살펴보려면 Raspberry Pi 3 B+ 참조 이미지를 사용하는 Azure IoT Hub용 디바이스 업데이트 자습서에서 이미지를 업데이트합니다.

다음을 통해 공유

작업 예약 및 브로드캐스트(Python)

필수 조건

시뮬레이션된 디바이스 앱 만들기

IoT Hub 연결 문자열 가져오기

직접 메서드를 호출하고 디바이스 쌍의 속성을 업데이트하기 위한 작업 예약

애플리케이션 실행

다음 단계

피드백

추가 리소스