LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@saurus12(サウルス１２)

ESP32でのmicropythonを使ったAWS IoTへの接続

Last updated at Posted at 2025-07-01

ESP32でのmicropythonを使ったAWS IoTへの接続について解説します。

参考サイト

Amazon Web Services ブログ MicroPython を使って AWS IoT Core を始める
このページを参考にとりあえずesp32からAWS IoTへの接続をやってみたのですが、
この手順をそのままやってもうまくいかなかったので、うまくいく方法を、忘備のため記事にします。

IoT ポリシーを作成する

Amazon Web Services ブログ MicroPython を使って AWS IoT Core を始める
ステップ1： IoT ポリシーを作成するのとおりポリシーを作成します。

👇ポリシードキュメントはこんな感じです。

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": "iot:Connect",
          "Resource": "arn:aws:iot:ap-northeast-1:0000000000000:client/BlogClient"
        },
        {
          "Effect": "Allow",
          "Action": "iot:Publish",
          "Resource": "arn:aws:iot:ap-northeast-1:0000000000000:topic/$aws/things/BlogThing/shadow/update"
        },
        {
          "Effect": "Allow",
          "Action": "iot:Subscribe",
          "Resource": "arn:aws:iot:ap-northeast-1:0000000000000:topicfilter/$aws/things/BlogThing/shadow/update/delta"
        },
        {
          "Effect": "Allow",
          "Action": "iot:Receive",
          "Resource": "arn:aws:iot:ap-northeast-1:0000000000000:topic/$aws/things/BlogThing/shadow/update/delta"
        }
      ]
}

AWS IoT のモノを作成する

Amazon Web Services ブログ MicroPython を使って AWS IoT Core を始める
ステップ2： AWS IoT のモノを作成するのとおりモノを作成します。
でもさ、「モノを作成する」って・・・

デバイス証明書とキーファイルをダウンロード

Amazon Web Services ブログ MicroPython を使って AWS IoT Core を始める
ステップ2： AWS IoT のモノを作成するの最後の手順「デバイス証明書とキーファイル」をダウンロードします。

マイクロコントローラ用のファイルを準備する

Amazon Web Services ブログ MicroPython を使って AWS IoT Core を始める
ステップ3： マイクロコントローラ用のファイルを準備する
ここは注意が必要です

1.デバイス証明書と鍵のファイルをダウンロードしたローカルディスクに移動します。
👆これはこのままでOK
2.デバイス証明書は、-certificate.pem.crt で終わるファイルです。この証明書のファイル名を cert.pem.crt に変更します。
👆これもこのままでOK
3.秘密鍵は、-private.pem.key で終わるファイルです。このファイル名を private.pem.key に変更します。
👆これもこのままでOK
4.次に、MQTT に必要なライブラリーをダウンロードします。GitHub のリポジトリの内容をローカルディスクにダウンロードします。
👆この手順は不要(MQTTは最新のmicropythonでは標準で組み込まれてます。)
5.ファイルを整理するために、getting-started-micropython-esp32 というフォルダを作成します。証明書と秘密鍵のファイルをこのフォルダーに移動します。
👆これもこのままでOK
6.micropython-lib GitHub リポジトリの micropython/umqtt.simple/umqtt/simple.py というファイルを、ローカルディスクの getting-started-micropython-esp32 フォルダの umqtt フォルダにコピーします。
👆この手順は不要(MQTTは最新のmicropythonでは標準で組み込まれてます。)
7.getting-started-micropython-esp32 フォルダには、以下のファイルがあるはずです。
※ umqtt/simple.pyは不要です。

MicroPython を使用して AWS IoT に接続する

Amazon Web Services ブログ MicroPython を使って AWS IoT Core を始める
ステップ4： MicroPython を使用して AWS IoT に接続する
ここがひっかけポイントです！

1.getting-started-micropython-esp32 フォルダに、main.py というファイルを新規に作成します。
👆これはこのままでOK
2.GitHub の aws-iot-core-getting-started-micropython リポジトリから main.py のコードをコピーします。
👆これもこのままでOK
3.以下のコードを入力します。　　

a. wifi_ssid をお使いのワイヤレスネットワーク名に置き換えてください。
b. wifi_password をお使いのワイヤレスネットワークのパスワードに置き換えてください。
c. aws_endpoint を お使いの AWS アカウントの AWS IoT のエンドポイントに置き換えてください。AWS IoT Core コンソールの設定ページで確認することができます。

👆これもこのままでOK

追加で以下の編集が必要です

デバイス証明書とキーファイルは、読み込んだ文字列ではなく、ファイルを指定するようにssl_paramsを編集する。

#ssl_params = {"key":key, "cert":cert, "server_side":False}
ssl_params = {"key":private_key, "cert":private_cert, "server_side":False}

main.pyのサンプルコード

main.pyのサンプルコードを付けておきます。
とりあえずAWSとの接続が目的だったので、いらないところは全て削除しました。

import os
import time
import ujson
import machine
import network
from umqtt.simple import MQTTClient

#Enter your wifi SSID and password below.
wifi_ssid = "SSID*******"
wifi_password = "************"

#Enter your AWS IoT endpoint. You can find it in the Settings page of
#your AWS IoT Core console. 
#https://docs.aws.amazon.com/iot/latest/developerguide/iot-connect-devices.html 
aws_endpoint = b'********-ats.iot.ap-northeast-1.amazonaws.com'

#If you followed the blog, these names are already set.
thing_name = "BlogThing"
client_id = "BlogClient"
private_key = "private.pem.key"
private_cert = "cert.pem.crt"

#Read the files used to authenticate to AWS IoT Core
#with open(private_key, 'r') as f:
#    key = f.read()
#with open(private_cert, 'r') as f:
#    cert = f.read()

#These are the topics we will subscribe to. We will publish updates to /update.
#We will subscribe to the /update/delta topic to look for changes in the device shadow.
topic_pub = "$aws/things/" + thing_name + "/shadow/update"
topic_sub = "$aws/things/" + thing_name + "/shadow/update/delta"

#ssl_params = {"key":key, "cert":cert, "server_side":False}
ssl_params = {"key":private_key, "cert":private_cert, "server_side":False}

#Define pins for LED and light sensor. In this example we are using a FeatherS2.
#The sensor and LED are built into the board, and no external connections are required.
#light_sensor = machine.ADC(machine.Pin(4))
#light_sensor.atten(machine.ADC.ATTN_11DB)
#led = machine.Pin(13, machine.Pin.OUT)
info = os.uname()

#Connect to the wireless network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
if not wlan.isconnected():
    print('Connecting to network...')
    wlan.connect(wifi_ssid, wifi_password)
    while not wlan.isconnected():
        pass

    print('Connection successful')
    print('Network config:', wlan.ifconfig())

def mqtt_connect(client=client_id, endpoint=aws_endpoint, sslp=ssl_params):
    mqtt = MQTTClient(client_id=client, server=endpoint, port=8883, keepalive=1200, ssl=True, ssl_params=sslp)
    print("Connecting to AWS IoT...")
    mqtt.connect()
    print("Done")
    return mqtt

def mqtt_publish(client, topic=topic_pub, message=''):
    print("Publishing message...")
    client.publish(topic, message)
    print(message)

def mqtt_subscribe(topic, msg):
    print("Message received...")
    message = ujson.loads(msg)
    print(topic, message)
    # if message['state']['led']:
    #     led_state(message)
    print("Done")

#def led_state(message):
#    led.value(message['state']['led']['onboard'])

#We use our helper function to connect to AWS IoT Core.
#The callback function mqtt_subscribe is what will be called if we 
#get a new message on topic_sub.
try:
    mqtt = mqtt_connect()
    mqtt.set_callback(mqtt_subscribe)
    mqtt.subscribe(topic_sub)
except:
    print("Unable to connect to MQTT.")


while True:
    try:
        mqtt.check_msg()
    except:
        print("Unable to check for messages.")

    mesg = ujson.dumps({
        "state":{
            "reported": {
                "device": {
                    "client": client_id,
                    "uptime": time.ticks_ms(),
                    "hardware": info[0],
                    "firmware": info[2]
                },
#                "sensors": {
#                    "light": light_sensor.read()
#                },
#                "led": {
#                    "onboard": led.value()
#                }
            }
        }
    })

#Using the message above, the device shadow is updated.
    try:
        mqtt_publish(client=mqtt, message=mesg)
    except:
        print("Unable to publish message.")

#Wait for 10 seconds before checking for messages and publishing a new update.
    print("Sleep for 10 seconds")
    time.sleep(10)

マイクロコントローラにファイルをコピーする

私はTonnyを使ってファイルをコピーしました。

main.py実行

image

👆実行するとこんな感じになります。

メッセージのルーティング

image

👆SQL ステートメントでSELECT * FROM '#'を指定して、アクションにCloudWatchを指定すれば
CloudWatchログで出力を確認できます。

image

👇関連記事

👇参考URL

Amazon Web Services ブログ MicroPython を使って AWS IoT Core を始める

本記事へのリンク

https://docs.saurus12.com/device/esp32_mqtt

[keywords]
ESP32 micropython mqtt AWS IoT

ESP32でのmicropythonを使ったAWS IoTへの接続

更新日:2025年07月01日

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?