Python を使用している場合に、デバイスから AWS IoT Core に MQTT メッセージをパブリッシュする方法を教えてください。

所要時間4分

AWS IoT Core とデバイスまたは MQTT クライアントの間で MQTT (MQ テレメトリトランスポート) メッセージを送受信できません。MQTT メッセージを AWS IoT Core にパブリッシュする方法を教えてください。

簡単な説明

AWS IoT モノが正しく設定され、証明書が正しく添付されていることを確認します。セットアップをテストするには、AWS IoT MQTT クライアントとこの記事で提供されているサンプル Python コードを使用できます。

解決策

MQTT パブリッシングをテストするためのディレクトリを設定する

1.開発環境に作業ディレクトリを作成します。例: iot-test-publish

2.新しい作業ディレクトリに証明書用のサブディレクトリを作成します。例:certificates

3.コマンドラインから、ディレクトリを新しい作業ディレクトリに変更します。

Python 用の pip と AWS IoT SDK をインストールする

1.Python 3 パッケージ用の pip をまだインストールしていない場合は、インストールしてください。詳細については、Python パッケージングオーソリティ (PyPA) のウェブサイトの「Installation」を参照してください。

2.コマンドラインから以下を実行して、Python v2 用 AWS IoT SDK をインストールします。

pip install awsiotsdk

または

次のコマンドを実行して、Python 用 AWS IoT Device 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 ポリシーがある場合は、[作成] を選択して新しいポリシーを作成します。
または、
既存のポリシーがない場合は、[ポリシーはまだ作成されていません。] ページで [ポリシーの作成] を選択します。

5.[ポリシーの作成] ページで、ポリシーの名前を入力します。例: admin

6.[ステートメントを追加] で次の操作を行います。
[アクション] に「iot:*」と入力します。
重要:すべての AWS IoT アクション (iot:*) を許可するとテストに役立ちます。ただし、本番環境のセットアップではセキュリティを強化するのがベストプラクティスです。より安全なポリシーの例については、「Example AWS IoT policies」を参照してください。
[リソース ARN] に「*」と入力します。
[効果]で、[許可] チェックボックスを選択します。

7.[作成] を選択します。

詳細については、「Create an AWS IoT Core policy」と「AWS IoT Core policies」を参照してください。

AWS IoT モノを作成する

**注:**AWS IoT に接続するためにモノを作成する必要はありません。ただし、作成することで追加のセキュリティコントロールや、フリートインデックス、ジョブ、デバイスシャドウなどの他の AWS IoT 機能を使用できるようになります。

1.AWS IoT Core コンソールの左側のナビゲーションペインで、[管理] を選択します。

2.既存のモノがある場合は、[作成] を選択して新しいモノを作成します。
または、
既存のモノがない場合は、[まだモノがありません] ページで [モノの登録] を選択します。

3.**[AWS IoT モノを作成する]**ページで [1 つのモノの作成] を選択します。

4.[Thing Registry にデバイスを追加] ページで、次の操作を行います。
モノの名前を入力します。例:Test-Thing
(オプション) [このモノにタイプを追加] でモノのタイプを選択または作成します。
(オプション) **[グループにこのモノを追加]**で、グループを選択または作成します。グループの詳細については、「Static thing groups」と「Dynamic thing groups」を参照してください。
(オプション) [検索可能なモノの属性の設定 (オプション)] で、属性をキーと値のペアとして追加します。
[次へ] を選択します。

5.[モノに証明書を追加] ページで [証明書を作成] を選択します。モノとモノの証明書が作成されたことを確認する通知が表示されます。

6.[証明書が作成されました] ページで次の操作を行います。
[デバイスを接続するには、以下をダウンロードする必要があります]で、証明書、パブリックキー、およびプライベートキーの [ダウンロード] を選択します。
ダウンロードした各ファイルを、前に作成した証明書サブディレクトリに保存します。
[AWS IoT のルート CA もダウンロードする必要があります] で [ダウンロード] を選択します。[サーバー認証] ページが開き、サーバー認証用の CA 証明書が表示されます。

7.[Amazon トラストサービスエンドポイント (推奨)] で [Amazon ルート CA 1] を選択します。証明書がブラウザで開きます。

8.証明書 (-----BEGIN CERTIFICATE----- から -----END CERTIFICATE----- までのすべて) をコピーし、テキストエディタに貼り付けます。

9.証明書を root.pem という名前の .pem ファイルとして証明書サブディレクトリに保存します。

10.AWS IoT Core コンソールの [証明書が作成されました] ページで、[アクティブ化] を選択します。ボタンが [無効化] に変わります。

11.[ポリシーをアタッチ] を選択します。

12.[モノにポリシーを追加] ページで、次の操作を行います。
以前に作成した AWS IoT Core ポリシーを選択します。例: admin
[モノの登録] を選択します。

詳細については、「Create a thing object」を参照してください。

AWS IoT Core エンドポイントの URL をコピーします。

1.AWS IoT Core コンソールの左側のナビゲーションペインで、[設定] を選択します。

2.[設定] ページの [カスタムエンドポイント] で、エンドポイントをコピーします。この AWS IoT Core カスタムエンドポイント URL は、お客様の AWS アカウントとリージョンに固有のものです。

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 Device 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 という名前のテストトピックでテストメッセージを公開する準備ができました。詳細については、「View device MQTT messages with the AWS IoT MQTT client」を参照してください。

4.コマンドラインから以下を実行します。

python3 publish.py

Python プログラムは、AWS IoT Core コンソールで作成した「test/testing」というトピックに 20 個のテストメッセージをパブリッシュします。コンソールでトピックを表示して、公開されたメッセージを確認します。

**ヒント：**付属の pubsub サンプルを使用して、WebSocket 経由のサブスクライブや接続など、SDK の他の機能をテストすることもできます。詳細については、GitHub の pubsub (Python 用 AWS IoT SDK v2) または BasicPubSub (Python 用 AWS IoT Device SDK) を参照してください。

(オプション) Amazon CloudWatch への AWS IoT ロギングを有効にする

AWS IoT Core に公開した MQTT メッセージのイベントログをモニタリングできます。セットアップ手順については、「Configure AWS IoT logging」と「Monitor AWS IoT using CloudWatch Logs」を参照してください。