Python을 사용하는 경우 디바이스에서 AWS IoT Core로 MQTT 메시지를 게시하려면 어떻게 해야 합니까?

6분 분량

AWS IoT Core와 디바이스 또는 MQTT 클라이언트 사이에서 MQ Telemetry Transport(MQTT) 메시지를 보내거나 받을 수 없습니다. AWS IoT Core에 MQTT 메시지를 게시하려면 어떻게 해야 합니까?

간략한 설명

AWS IoT 사물이 올바르게 구성되었고 인증서가 올바르게 연결되었는지 확인합니다. 설정을 테스트하기 위해 AWS IoT MQTT 클라이언트와 이 문서에서 제공하는 Python 코드 예제를 사용할 수 있습니다.

해결 방법

MQTT 게시를 테스트할 디렉터리 설정

1. 개발 환경에서 작업 디렉터리를 생성합니다. 예: iot-test-publish

2. 새 작업 디렉터리에서 인증서를 넣을 하위 디렉터리를 생성합니다. 예: certificates

3. 명령줄에서 디렉터리를 새 작업 디렉터리로 변경합니다.

Python용 AWS IoT SDK 및 pip 설치

1. Python 3 패키징용 pip를 아직 설치하지 않았다면 설치합니다. 자세한 내용은 Python Packaging Authority(PyPA) 웹 사이트에서 Installation을 참조하세요.

2. 명령줄에서 다음을 실행하여 Python v2용 AWS IoT SDK를 설치합니다.

pip install awsiotsdk

-또는-

다음 명령을 실행하여 Python용 AWS IoT 디바이스 SDK(이전 SDK 버전)를 설치합니다.

pip install AWSIoTPythonSDK

자세한 내용은 GitHub에서 AWS IoT SDK for Python v2 또는 AWS IoT Device SDK for Python을 참조하세요.

참고: 이러한 SDK는 AWS IoT Core에 연결하는 것이 좋지만, 필수 사항은 아닙니다. 또한 호환되는 서드 파티 MQTT 클라이언트를 사용하여 연결할 수도 있습니다.

AWS IoT Core 정책 생성

1. AWS IoT Core 콘솔을 엽니다.

2. 왼쪽 탐색 창에서 [보안]을 선택합니다.

3. [보안]에서 [정책]을 선택합니다.

4. 기존 AWS IoT Core 정책이 있는 경우 [생성(Create)]을 선택하여 새 정책을 생성합니다.
-또는-
기존 정책이 없는 경우 [아직 정책 없음(You don't have any policies yet)] 페이지에서 [정책 생성(Create a policy)]을 선택합니다.

5. [정책 생성(Create a policy)] 페이지에서 정책의 [이름(Name)]을 입력합니다. 예를 들어 admin과 같습니다.

6. [설명문 추가]에서 다음을 수행합니다.
[작업(Action)]에 iot:*를 입력합니다.
중요: 모든 AWS IoT 작업을 허용하면(iot:*) 테스트에 유용합니다. 하지만 프로덕션 설정의 보안을 강화하는 것이 모범 사례입니다. 보다 안전한 정책 예제는 AWS IoT 정책 예제를 참조하세요.
[리소스 ARN]에 *를 입력합니다.
[효과]에서 [허용] 확인란을 선택합니다.

7. [생성]을 선택합니다.

자세한 내용은 AWS IoT Core 정책 생성 및 AWS IoT Core 정책을 참조하세요.

AWS IoT 사물 생성

참고: AWS IoT에 연결할 사물을 생성할 필요는 없습니다. 하지만 사물을 활용하면 추가 보안 제어 및 플릿 인덱싱, 작업, 또는 디바이스 섀도우와 같은 다른 AWS IoT 기능을 사용할 수 있습니다.

1. AWS IoT Core 콘솔의 왼쪽 탐색 창에서 [관리(Manage)]를 선택합니다.

2. 기존 사물이 있는 경우 [생성(Create)]을 선택하여 새 사물을 생성합니다.
-또는-
기존 사물이 없는 경우 [아직 사물 없음(You don't have any things yet)] 페이지에서 [사물 등록(Register a thing)]을 선택합니다.

3. [AWS IoT 사물 생성] 페이지에서 [단일 사물 생성]을 선택합니다.

4. [사물 레지스트리에 디바이스 추가] 페이지에서 다음을 수행합니다.
사물의 [이름(Name)]을 입력합니다. 예를 들어 Test-Thing과 같습니다.
(선택 사항) [이 사물에 유형 추가(Add a type to this thing)]에서 [사물 유형(thing type)]을 선택하거나 생성합니다.
(선택 사항) [그룹에 이 사물 추가]에서 그룹을 선택하거나 생성합니다. 그룹에 대한 자세한 내용은 정적 사물 그룹 및 동적 사물 그룹을 참조하세요.
(선택 사항) [검색 가능한 사물 속성 설정 (선택 사항)]에서 속성을 키–값 페어로 추가합니다.
[다음]을 선택합니다.

5. [사물에 인증서 추가] 페이지에서 [인증서 생성]을 선택합니다. 사물 및 사물에 대한 인증서가 생성되었음을 확인하는 알림이 표시됩니다.

6. [인증서 생성 완료] 페이지에서 다음을 수행합니다.
[디바이스에 연결하려면 다음을 다운로드해야 합니다.]에서 인증서, 퍼블릭 키 및 프라이빗 키에 대한 [다운로드]를 선택합니다.
다운로드한 각 파일을 앞서 생성한 certificates 하위 디렉터리에 저장합니다.
[AWS IoT의 루트 CA도 다운로드해야 합니다.]에서 [다운로드]를 선택합니다. [서버 인증(Server authentication)] 페이지의 [서버 인증을 위한 CA 인증서(CA certificates for server authentication)]가 열립니다.

7. [Amazon Trust Services 엔드포인트(기본 설정)(Amazon Trust Services Endpoints (preferred))]에서 [Amazon Root CA 1]을 선택합니다. 브라우저에서 인증서가 열립니다.

8. 인증서(**-----BEGIN CERTIFICATE-----**에서 **-----END CERTIFICATE-----**까지 모든 내용)를 복사하여 텍스트 편집기에 붙여넣습니다.

9. 인증서를 certificates 하위 디렉터리에 root.pem이라는 .pem 파일로 저장합니다.

10. AWS IoT Core 콘솔의 [인증서 생성 완료] 페이지에서 [활성화]를 선택합니다. 그러면 버튼이 [비활성화]로 변경됩니다.

11. [정책 연결]을 선택합니다.

12. [사물에 정책 추가] 페이지에서 다음을 수행합니다.
이전에 생성한 AWS IoT Core 정책을 선택합니다. 예를 들어 admin과 같습니다.
[사물 등록(Register Thing)]을 선택합니다.

자세한 내용은 사물 객체 생성을 참조하세요.

AWS IoT Core 엔드포인트 URL 복사

1. AWS IoT Core 콘솔의 왼쪽 탐색 창에서 [설정]을 선택합니다.

2. [설정] 페이지의 [사용자 지정 엔드포인트]에서 [엔드포인트]를 복사합니다. 이 AWS IoT Core 사용자 지정 엔드포인트 URL은 AWS 계정 및 리전에 대해 개인적으로 지정됩니다.

Python 프로그램 파일 생성

다음 Python 코드 예제 중 하나를 publish.py라는 Python 프로그램 파일로 저장합니다.

앞서 AWS IoT SDK for Python v2를 설치한 경우 다음 예제 코드를 사용합니다.

중요: customEndpointUrl을 AWS IoT Core 사용자 지정 엔드포인트 URL로 바꿉니다. certificates를 certificates 하위 디렉터리 이름으로 바꿉니다. a1b23cd45e-certificate.pem.crt를 클라이언트 .crt 이름으로 바꿉니다. a1b23cd45e-private.pem.key를 프라이빗 키 이름으로 바꿉니다.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0

from awscrt import io, mqtt, auth, http
from awsiot import mqtt_connection_builder
import time as t
import json

# Define ENDPOINT, CLIENT_ID, PATH_TO_CERTIFICATE, PATH_TO_PRIVATE_KEY, PATH_TO_AMAZON_ROOT_CA_1, MESSAGE, TOPIC, and RANGE
ENDPOINT = "customEndpointUrl"
CLIENT_ID = "testDevice"
PATH_TO_CERTIFICATE = "certificates/a1b23cd45e-certificate.pem.crt"
PATH_TO_PRIVATE_KEY = "certificates/a1b23cd45e-private.pem.key"
PATH_TO_AMAZON_ROOT_CA_1 = "certificates/root.pem"
MESSAGE = "Hello World"
TOPIC = "test/testing"
RANGE = 20

# Spin up resources
event_loop_group = io.EventLoopGroup(1)
host_resolver = io.DefaultHostResolver(event_loop_group)
client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
mqtt_connection = mqtt_connection_builder.mtls_from_path(
            endpoint=ENDPOINT,
            cert_filepath=PATH_TO_CERTIFICATE,
            pri_key_filepath=PATH_TO_PRIVATE_KEY,
            client_bootstrap=client_bootstrap,
            ca_filepath=PATH_TO_AMAZON_ROOT_CA_1,
            client_id=CLIENT_ID,
            clean_session=False,
            keep_alive_secs=6
            )
print("Connecting to {} with client ID '{}'...".format(
        ENDPOINT, CLIENT_ID))
# Make the connect() call
connect_future = mqtt_connection.connect()
# Future.result() waits until a result is available
connect_future.result()
print("Connected!")
# Publish message to server desired number of times.
print('Begin Publish')
for i in range (RANGE):
    data = "{} [{}]".format(MESSAGE, i+1)
    message = {"message" : data}
    mqtt_connection.publish(topic=TOPIC, payload=json.dumps(message), qos=mqtt.QoS.AT_LEAST_ONCE)
    print("Published: '" + json.dumps(message) + "' to the topic: " + "'test/testing'")
    t.sleep(0.1)
print('Publish End')
disconnect_future = mqtt_connection.disconnect()
disconnect_future.result()

-또는-

AWS IoT Device SDK for Python(이전 SDK 버전)을 설치한 경우 다음 예제 코드를 사용합니다.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0

import time as t
import json
import AWSIoTPythonSDK.MQTTLib as AWSIoTPyMQTT

# Define ENDPOINT, CLIENT_ID, PATH_TO_CERTIFICATE, PATH_TO_PRIVATE_KEY, PATH_TO_AMAZON_ROOT_CA_1, MESSAGE, TOPIC, and RANGE
ENDPOINT = "customEndpointUrl"
CLIENT_ID = "testDevice"
PATH_TO_CERTIFICATE = "certificates/a1b23cd45e-certificate.pem.crt"
PATH_TO_PRIVATE_KEY = "certificates/a1b23cd45e-private.pem.key"
PATH_TO_AMAZON_ROOT_CA_1 = "certificates/root.pem"
MESSAGE = "Hello World"
TOPIC = "test/testing"
RANGE = 20

myAWSIoTMQTTClient = AWSIoTPyMQTT.AWSIoTMQTTClient(CLIENT_ID)
myAWSIoTMQTTClient.configureEndpoint(ENDPOINT, 8883)
myAWSIoTMQTTClient.configureCredentials(PATH_TO_AMAZON_ROOT_CA_1, PATH_TO_PRIVATE_KEY, PATH_TO_CERTIFICATE)

myAWSIoTMQTTClient.connect()
print('Begin Publish')
for i in range (RANGE):
    data = "{} [{}]".format(MESSAGE, i+1)
    message = {"message" : data}
    myAWSIoTMQTTClient.publish(TOPIC, json.dumps(message), 1) 
    print("Published: '" + json.dumps(message) + "' to the topic: " + "'test/testing'")
    t.sleep(0.1)
print('Publish End')
myAWSIoTMQTTClient.disconnect()

설정 테스트

1. AWS IoT Core 콘솔의 왼쪽 탐색 창에서 [테스트]를 선택합니다.

2. [MQTT 클라이언트] 페이지의 [구독 주제]에 test/testing을 입력합니다.

3. [주제 구독]을 선택합니다. test/testing이라는 테스트 주제를 테스트 메시지 게시에 사용할 준비가 되었습니다. 자세한 내용은 AWS IoT MQTT 클라이언트를 사용하여 MQTT 메시지 보기를 참조하십시오.

4. 명령줄에서 다음을 실행합니다.

python3 publish.py

Python 프로그램은 AWS IoT Core 콘솔에서 생성한 주제 test/testing 주제에 20개의 테스트 메시지를 게시합니다. 콘솔에서 주제를 보고 게시된 메시지를 확인합니다.

팁: 포함된 pubsub 샘플을 사용하여 WebSockets를 통한 구독 및 연결과 같은 SDK의 다른 기능을 테스트할 수도 있습니다. 자세한 내용은 GitHub에서 pubsub(AWS IoT SDK for Python v2) 또는 BasicPubSub(AWS IoT Device SDK for Python)를 참조하세요.

(선택 사항) Amazon CloudWatch에 대한 AWS IoT 로깅 활성화

AWS IoT Core에 게시하는 MQTT 메시지에 대한 이벤트 로그를 모니터링할 수 있습니다. 설정 지침은 AWS IoT 로깅 구성 및 CloudWatch Logs를 사용하여 AWS IoT 모니터링을 참조하십시오.