AWS IOT にパブリッシュをするサンプル

次のページを参考にしました。
AWS IoT Core -> デベロッパーガイド -> HTTPS

エンドポイントは次のコマンドで調べます。

aws iot describe-endpoint --endpoint-type iot:Data-ATS

コンソールで調べるには
　AWS IoT -> 設定

トピックはポリシーで定められたものに限られます。
次のものが使えました。
デモを実行した時にこのポリシーが作成されました。

sdk/test/Python
sdk/test/java
topic_1
topic_2

rest_publish.sh

#
#	rest_publish.sh
#
#					Sep/09/2021
#
HOST="https://abcd6goq68zt4o-ats.iot.ap-northeast-1.amazonaws.com:8443"
TOPIC="sdk/test/Python"
URL=$HOST"/topics/"$TOPIC"?qos=1"
#
curl --tlsv1.2 \
	--cacert ../certs/root-CA.crt \
	--cert ../certs/test01.cert.pem \
	--key ../certs/test01.private.key \
	-X POST $URL \
	--data "{ \"message\": \"Good Morning AM 09:35\" }" 

rest_publish.py

#! /usr/bin/python
#
#	rest_publish.py
#
#					Sep/10/2021
# ------------------------------------------------------------------
import requests

endpoint = "abcd6goq68zt4o-ats.iot.ap-northeast-1.amazonaws.com"
topic = "sdk/test/Python"
cert = "../certs/test01.cert.pem"
key = "../certs/test01.private.key"
message = '{"message": "Good Morning AM 09:32" }'
#
# create and format values for HTTPS request
publish_url = 'https://' + endpoint + ':8443/topics/' + topic + '?qos=1'
publish_msg = message.encode('utf-8')

# make request
publish = requests.request('POST',
	publish_url,
	data=publish_msg,
	cert=[cert, key])

# print results
print("Response status: ", str(publish.status_code))
if publish.status_code == 200:
	print("Response body:", publish.text)
#
# ------------------------------------------------------------------

mosquitto を使うサンプルです。

mos_publish.sh

#
cafile="../certs/Amazon-root-CA-1.pem"
certfile="../certs/device.pem.crt"
keyfile="../certs/private.pem.key"
#
HOST="abcd6goq68zt4o-ats.iot.ap-northeast-1.amazonaws.com"
TOPIC="sdk/test/Python"
#
mosquitto_pub --cafile $cafile \
	--cert $certfile \
	--key $keyfile \
    -h $HOST \
    -p 8883 -q 1 -d -t $TOPIC \
	 -m "{ \"message\": \"Good Morning PM 14:50\" }"

SDK を使うサンプルです。

sdk_publish.py

#! /usr/bin/python
#
#	sdk_publish.py
#
#					Sep/10/2021
# ------------------------------------------------------------------
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
import time
import json

# ------------------------------------------------------------------
host = "abcd6goq68zt4o-ats.iot.ap-northeast-1.amazonaws.com"
rootCAPath = "../certs/root-CA.crt"
certificatePath = "../certs/test01.cert.pem"
privateKeyPath = "../certs/test01.private.key"
port = 443
useWebsocket = 0
clientId = "basicPubSub"
topic = "sdk/test/Python"

myAWSIoTMQTTClient = AWSIoTMQTTClient(clientId)
myAWSIoTMQTTClient.configureEndpoint(host, port)
myAWSIoTMQTTClient.configureCredentials(rootCAPath, privateKeyPath, certificatePath)

# AWSIoTMQTTClient connection configuration
myAWSIoTMQTTClient.configureAutoReconnectBackoffTime(1, 32, 20)
myAWSIoTMQTTClient.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
myAWSIoTMQTTClient.configureDrainingFrequency(2)  # Draining: 2 Hz
myAWSIoTMQTTClient.configureConnectDisconnectTimeout(10)  # 10 sec
myAWSIoTMQTTClient.configureMQTTOperationTimeout(5)  # 5 sec

# Connect and subscribe to AWS IoT
myAWSIoTMQTTClient.connect()


# Publish to the same topic in a loop forever
count = 0
while True:
	message = {}
	message['message'] = "こんにちは"
	message['count'] = count 
	messageJson = json.dumps(message)
	myAWSIoTMQTTClient.publish(topic, messageJson, 1)
	print('Published topic %s: %s\n' % (topic, messageJson))
	count += 1
	time.sleep(2)
# ------------------------------------------------------------------

次のバージョンで確認しました。

$ python
Python 3.10.7 (main, Nov  2 2022, 18:49:29) [GCC 12.2.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import AWSIoTPythonSDK
>>> print(AWSIoTPythonSDK.__version__)
1.4.9

注意

新規に証明書を作成した場合、
証明書に policy 　がアタッチされている必要があります。
AWS IoT -> セキュリティ -> 証明書