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

所要時間4分

AWS IoT Core とデバイスまたは MQTT (Message Queuing Telemetry Transport) クライアントの間で MQTT メッセージを送受信できません。

簡単な説明

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

解決策

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

開発環境に作業ディレクトリを作成します。例えば、iot-test-publish という名前のディレクトリを作成します。
新しい作業ディレクトリに証明書用のサブディレクトリを作成します。例えば、certificates という名前のサブディレクトリを作成します。
コマンドラインから、ディレクトリを新しい作業ディレクトリに変更します。

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

注: 始める前に、Python 3 パッケージ用の pip をインストールしてください。詳細については、Python Packaging Authority (PyPA) ウェブサイトの「Installation」を参照してください。

Python v2 用 AWS IoT SDK をインストールするには、次のコマンドを使用します。
```
pip install awsiotsdk
```
または、

Python 用 AWS IoT Device SDK (以前の SDK バージョン、v1) をインストールするには、次のコマンドを使用します。
```
pip install AWSIoTPythonSDK
```

注: お使いの AWS IoT SDK for Python バージョンの最新のアップデートを使用するのがベストプラクティスです。詳細については、GitHub の「AWS IoT SDK for Python v2」または「AWS IoT Device SDK for Python」を参照してください。これらの SDK を使用して AWS IoT Core に接続することもベストプラクティスですが、これらの SDK は必須ではありません。準拠している他のサードパーティの MQTT クライアントを使用して AWS IoT Core に接続できます。

AWS IoT Core ポリシーを作成する

AWS IoT Core コンソールを開きます。
左側のナビゲーションペインで、[セキュア] を選択します。
[セキュア] で [ポリシー] を選択します。
既存の AWS IoT Core ポリシーがある場合は、[作成] を選択して新しいポリシーを作成します。
または、
既存のポリシーがない場合は、[ポリシーはまだ作成されていません。] ページで [ポリシーの作成] を選択します。
[ポリシーの作成] ページで、ポリシーの名前を入力します。例えば、admin という名前を入力します。
「[ステートメントの追加] で、次のアクションを実行します。
[アクション] に「iot:*」と入力します。
重要: すべての AWS IoT アクション (iot:\ *) をテスト目的で許可できます。ただし、本番環境のセットアップではセキュリティを強化するのがベストプラクティスです。より安全なポリシーの例については、AWS IoT Core ポリシーの例を参照してください。
[リソース ARN] に「*」と入力します。
[効果] で、[許可] チェックボックスを選択します。
[作成] を選択します。

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

AWS IoT モノを作成する

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

AWS IoT Core コンソールの左側のナビゲーションペインで、[管理] を選択します。
既存のモノがある場合は、[作成] を選択して新しいモノを作成します。
または、
既存のモノがない場合は、[まだモノがありません] ページで [モノの登録] を選択します。
[AWS IoT モノを作成する] ページで [1 つのモノの作成] を選択します。
「Thing Registry にデバイスを追加」ページで、次の操作を行います。
モノの名前を入力します。例えば、「Test-Thing」と入力します。
(オプション) [このモノにタイプを追加] でモノのタイプを選択または作成します。
(オプション) [グループにこのモノを追加] で、グループを選択または作成します。グループの詳細については、「Static thing groups」と「Dynamic thing groups」を参照してください。
(オプション) [検索可能なモノの属性の設定 (オプション)] で、属性をキーと値のペアとして追加します。
[次へ] を選択します。
注: モノの名前をクライアント ID と一致させるのがベストプラクティスです。名前が一致すれば、フリートインデックスや Thing ポリシー変数などの AWS IoT Core 機能を使用できます。
[モノに証明書を追加] ページで [証明書を作成] を選択します。モノとモノの証明書が作成されたことを確認する通知が表示されます。
[証明書が作成されました] ページで、次の操作を実行します。
[デバイスを接続するには、以下をダウンロードする必要があります] で、証明書、パブリックキー、およびプライベートキーの [ダウンロード] を選択します。
ダウンロードした各ファイルを、先に作成した certificates サブディレクトリに保存します。
[AWS IoT のルート CA もダウンロードする必要があります] で [ダウンロード] を選択します。[サーバー認証] ページが開き、サーバー認証用の CA 証明書が表示されます。
[Amazon トラストサービスエンドポイント (推奨)] で [Amazon ルート CA 1] を選択します。証明書がブラウザで開きます。
証明書 (-----BEGIN CERTIFICATE----- から -----END CERTIFICATE----- までのすべて) をコピーし、テキストエディタに貼り付けます。
証明書を root.pem という名前の .pem ファイルとして証明書サブディレクトリに保存します。
AWS IoT Core コンソールの [証明書が作成されました] ページで、[アクティブ化] を選択します。ボタンが [無効化] に変わります。
[ポリシーをアタッチ] を選択します。
[モノにポリシーを追加] ページで、次の操作を行います。
先に作成した AWS IoT Core ポリシーを選択します。例えば、admin を選択します。
[モノの登録] を選択します。

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

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

AWS IoT Core コンソールのナビゲーションペインで、[設定] を選択します。
[設定] ページの [カスタムエンドポイント] で、エンドポイントをコピーします。この 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 v1 用 AWS IoT デバイス 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()

注: Python 用 AWS IoT デバイス SDK の古いバージョンは Python 3.12 以降ではサポートされていません。必ず Python 用 AWS IoT デバイス SDK の最新バージョンを使用してください。

セットアップをテストする

AWS IoT Core コンソールの左側のナビゲーションペインで、[テスト] を選択します。
[MQTT クライアント] ページの **[サブスクリプショントピック] ** に「test/testing」と入力します。
[トピックをサブスクライブ] を選択します。test/testing という名前のテストトピックでテストメッセージを公開する準備ができました。詳細については、「AWS IoT MQTT クライアントで MQTT を表示する」を参照してください。
コマンドラインから以下のコマンドを実行します。
```
python3 publish.py
```

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

ヒント: 付属の pubsub サンプルを使用して、WebSockets のサブスクリプションや接続など、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」を参照してください。