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

6분 분량

AWS IoT Core와 디바이스 또는 MQTT 클라이언트 간에 MQTT(MQ Telemetry Transport) 메시지를 주고받을 수 없습니다. MQTT 메시지를 AWS IoT Core에 게시하려면 어떻게 해야 하나요?

간략한 설명

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

해결 방법

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

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

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

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

pip 및 Python용 AWS IoT SDK 설치

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

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

pip install awsiotsdk

-또는-

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

pip install AWSIoTPythonSDK

자세한 내용은 GitHub의 Python v2용 AWS IoT SDK 또는 Python용 AWS IoT 디바이스 SDK를 참조합니다.

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

AWS IoT Core 정책 생성

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

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

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

4. 기존 AWS IoT Core 정책이 있는 경우 생성을 선택하여 새 정책을 생성합니다.
-또는-
기존 정책이 없는 경우 아직 정책 없음 페이지에서 정책 생성을 선택합니다.

5. 정책 생성 페이지에서 정책의 이름을 입력합니다. 예: 관리자.

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

7. 생성을 선택합니다.

자세한 내용은 AWS IoT Core 정책 생성 및 AWS IoT Core 정책을 참조합니다.

AWS IoT 사물 생성

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

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

2. 기존에 사물이 있는 경우 생성을 선택하여 새 사물을 생성합니다.
-또는-
기존에 사물이 없는 경우 아직 사물 없음 페이지에서 사물 등록을 선택합니다.

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

4. 사물 레지스트리에 디바이스 추가 페이지에서 다음을 수행합니다.
사물의 이름을 입력합니다. 예: 테스트-사물.
(선택 사항) 이 사물에 유형 추가에서 사물 유형을 선택하거나 생성합니다.
(선택 사항) 그룹에 이 사물 추가에서 그룹을 선택하거나 생성합니다. 그룹에 대한 자세한 내용은 정적 사물 그룹 및 동적 사물 그룹을 참조합니다.
(선택 사항) **검색 가능한 사물 속성 설정(선택 사항)**에서 속성을 키-값 쌍으로 추가합니다.
다음을 선택합니다.

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

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

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

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

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

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

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

12. 사물에 대한 정책 추가 페이지에서 다음을 수행합니다.
이전에 생성된 AWS IoT Core 정책을 선택합니다. 예: 관리자.
사물 등록을 선택합니다.

자세한 내용은 사물 객체 생성을 참조합니다.

AWS IoT Core 엔드포인트 URL 복사

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

2. 설정 페이지의 사용자 지정 엔드포인트에서 엔드포인트를 복사합니다. 이 AWS IoT Core 사용자 지정 엔드포인트 URL은 AWS 계정 및 리전 고유의 URL입니다.

Python 프로그램 파일 생성

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

이전에 Python v2용 AWS IoT SDK를 설치한 경우 다음의 예제 코드를 사용합니다.

중요:****customEndpointUrl을 AWS IoT Core 사용자 지정 엔드포인트 URL로 변경합니다. 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()

-또는-

Python용 AWS IoT 디바이스 SDK(이전 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 샘플을 사용하여 구독 및 WebSocket을 통한 연결과 같은 SDK의 다른 기능을 테스트할 수도 있습니다. 자세한 내용은 GitHub의 pubsub(Python v2용 AWS IoT SDK) 또는 BasicPubSub(Python용 AWS IoT 디바이스 SDK)를 참조하세요.

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

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