AWS IoTのMQTTのテストまでの手順まとめ(2023/03)

AWS IoTでMQTTのテストクライアントを試すまでの基本的な操作手順まとめ

目次
1. ポリシーの作成
2. モノの作成
3. MQTTのテストクライアントの実行

1. ポリシーの作成

「AWS IoT Core」で
[セキュリティ]=> [ポリシー] => [ポリシーの作成]

PolicyName=testMQTT

policy用Jsonファイル

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iot:Publish",
        "iot:Subscribe",
        "iot:Connect",
        "iot:Receive"
      ],
      "Resource": "*"
    }
  ]
}

2. モノの作成

「AWS IoT Core」で
[すべてのデバイス]=> [モノ]=> [モノの作成]=> [1つのモノの作成]

ThingName=testMQTT

「デバイス証明書を自動生成」で

・AmazonRootCA1.pem（ルートキー）
・publicキー(〜public.pem.key)
・privateキー(〜private.pem.key)
・デバイス証明書(〜certificate.pem.crt)

をdownloadする。

さっき作ったpolicyをアタッチする。
「モノの作成」をクリックして完成

MQTTのテストクライアントの実行

# エンドポイントの確認
$ aws iot describe-endpoint --endpoint-type IoT:Data-ATS
>>>
{
    "endpointAddress": "a3phz77nfdgsr1-ats.iot.ap-northeast-1.amazonaws.com"
}

# モノの情報確認
$ aws iot describe-thing --thing-name testMQTT
>>>
{
    "defaultClientId": "testMQTT",
    "thingName": "testMQTT",
    "thingId": "12fec87d-＊＊＊＊＊＊＊＊-d358c278d5e3",
    "thingArn": "arn:aws:iot:ap-northeast-1:＊＊＊＊＊:thing/testMQTT",
    "attributes": {},
    "version": 1
}

git clone https://github.com/aws/aws-iot-device-sdk-python
cd aws-iot-device-sdk-python

・エンドポイント='a3phz77nfdgsr1-ats.iot.ap-northeast-1.amazonaws.com'
・モノの名前='testMQTT'
・トピック名='device/22/data'

testMQTT.py

import os
import os.path
import time
import json

ROOTDIR='../keys'
THING_NAME = 'testMQTT'
TOPIC_NAME = 'device/22/data'

class IotMqttClient:

    ROOT_CA_PATH = os.path.join(ROOTDIR, 'AmazonRootCA1.pem')
    CERTIFICAT_PATH = os.path.join(ROOTDIR, 'certificate.pem.crt')
    PRIVATE_KEY_PATH = os.path.join(ROOTDIR, 'private.pem.key')
    IOT_ENDPOINT = 'a3phz77nfdgsr1-ats.iot.ap-northeast-1.amazonaws.com'
    IOT_PORT = 8883
    __my_iot_mqtt_client = None
    __is_connected = False

    def __init__(self):
        from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient

        # setup client
        self.__my_iot_mqtt_client = AWSIoTMQTTClient(THING_NAME)
        self.__my_iot_mqtt_client.configureEndpoint(self.IOT_ENDPOINT, self.IOT_PORT)
        self.__my_iot_mqtt_client.configureCredentials(self.ROOT_CA_PATH, self.PRIVATE_KEY_PATH, self.CERTIFICAT_PATH)

        # setup access infomation
        self.__my_iot_mqtt_client.configureAutoReconnectBackoffTime(1, 32, 20)
        self.__my_iot_mqtt_client.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
        self.__my_iot_mqtt_client.configureDrainingFrequency(2)  # Draining: 2 Hz
        self.__my_iot_mqtt_client.configureConnectDisconnectTimeout(10)  # 10 sec
        self.__my_iot_mqtt_client.configureMQTTOperationTimeout(5)  # 5 sec

        def on_offline(): self.__is_connected = False
        def on_online(): self.__is_connected = True

        self.__my_iot_mqtt_client.onOffline = on_offline
        self.__my_iot_mqtt_client.onOnline = on_online

        # start access
        self.__my_iot_mqtt_client.connect()

    def publish(self, IOT_TOPIC, unpacked_data):
        if not self.__is_connected:
            self.__my_iot_mqtt_client.connect()
        self.__my_iot_mqtt_client.publishAsync(IOT_TOPIC, str(unpacked_data), 1, ackCallback=None)


def main():
    iot_mqtt_client = IotMqttClient()

    count = 1
    while True:
       print(count)
       count += 1
       time.sleep(1)
       send_data = {'key':count}
       json_data = json.dumps(send_data)
       iot_mqtt_client.publish(os.path.join(THING_NAME, TOPIC_NAME), json_data)

if __name__ == '__main__':
    main()

MQTTテストクライアントを実行

**topicの決まり**
トピックのフィルター	意味
`#`	任意のトピック名
`topic_1/level_2/#`	`topic_1/level_2/`で始まるトピック名
`topic_1/+/level_3`	`topic_1/`で始まり`/level_3`で終わりその間に任意の一つの要素があるで始まるトピック名