Python の使用時に、デバイスから AWS IoT Core に MQTT メッセージを発行する方法を教えてください。

所要時間4分

AWS IoT Core とデバイスまたは MQTT クライアント間で MQTT (MQ Telemetry Transport) メッセージを送受信できません。MQTT メッセージを AWS IoT Core に発行する方法を教えてください。

簡単な説明

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

解決方法

ディレクトリをセットアップして MQTT 発行をテストする

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

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

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

pip と AWS IoT SDK for Python のインストール

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

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

pip install awsiotsdk

または -

次のコマンドを実行して AWS IoT Device SDK for Python (以前の 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:*) を許可すると、テストに役立ちます。ただし、本番稼働用セットアップのセキュリティを強化することがベストプラクティスです。より安全なポリシーの例については、「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. [Thing Registry にデバイスを追加] ページで、以下の操作を行います。
モノの [名前] を入力します。例えば、[Test-Thing] などです。
(オプション) [タイプをモノに追加] で、[モノのタイプ] を選択または作成します。
(オプション) [グループにこのモノを追加] で、グループを選択または作成します。グループの詳細については、「静的なモノのグループ」と「動的なモノのグループ」を参照してください。
(オプション) [検索可能なモノの属性の設定 (オプション)] で、キーと値のペアとして属性を追加します。
[次へ] をクリックします。

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

6. [作成された証明書] ページで、以下の操作を行います。
[デバイスを接続するには、次の情報をダウンロードします] で、証明書、パブリックキー、プライベートキーの [ダウンロード] を選択します。
ダウンロードした各ファイルを、先ほど作成した certificates サブディレクトリに保存します。
[また、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 ポリシーを選択します。例えば、「admin」などです。
[モノの登録] を選択します。

詳細については、「モノのオブジェクトを作成する」を参照してください。

AWS IoT Core エンドポイント URL をコピーする

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

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

Python プログラムファイルを作成する

次の Python コード例のいずれか 1 つを 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 のモニタリングを参照してください。