Raspberry Pi → AWS IoT Core 経由でDynamoDBへ保存する

## はじめに
前回の記事（[Raspberry Pi + DS18B20で水温測定](https://qiita.com/syo_engineer/items/bb306a0455835b320d68)）でRaspberry PiとDS18B20温度センサーを使用して水温の取得ができたので、今回は取得したデータをAWS IoT Core経由でDynamoDBに保存するところまでやってみる。

## 開発環境
- Python 3.11.2
- Raspberry Pi OS (64-bit)
- AWS IoT Core
- Amazon DynamoDB

## システム構成
水温計DS18B20から温度を取得し、Raspberry PiとIoT Coreを接続、Raspberry Piから送信されたデバイスデータをIoT CoreのルールでDynamoDBに保存する。


![IoTCore-1.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/fc967fd7-bd0e-0dda-5a38-3691de24cdfb.jpeg)


:::note info
※ DS18B20→Raspberry Piへの温度取得に必要な設定と配線図は、[前回の記事](https://qiita.com/syo_engineer/items/bb306a0455835b320d68)に記載しています。
:::



## AWS IoT Coreの設定

### マネージドコンソールからIoTCoreに入る


[デバイスを接続] からデバイスの設定画面に移動します。

![image-9.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/3d1bf2f8-4965-dd28-b823-aef8016baecb.jpeg)

### デバイスがIoTCoreに接続できるか確認する

Raspberry PiにSSH接続後、以下のコマンドを実行しIoTCoreに接続できるか確認する

```
ping hogefuga.iot.ap-northeast-1.amazonaws.com
```

接続できない場合は、RaspberryPiのWifi設定などインターネットに接続できる環境になっているか確認する

![image-1.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/fe245a8b-7ea6-6a09-79f2-b0f7698c62e6.jpeg)


上記コマンドを実行し、結果が返ってこればRaspberryPi側の設定はできている。
```
Reply from xx.xx.xxx.90: bytes=32 time=21ms TTL=248
Reply from xx.xx.xxx.90: bytes=32 time=16ms TTL=248
Reply from xx.xx.xxx.90: bytes=32 time=21ms TTL=248
```

### 接続するデバイスの設定

**1.　デバイス名を登録**

今回は **raspberry_pi_4** で設定

![image_1.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/ed56361e-2085-5878-1c9e-292ca36e9f26.jpeg)


**2.　プラットフォームとSDKを選択**

今回は、RaspberryPi OSを使用しているので、**Linux** を選択
SDKは **Python** を使用する

![image-2.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/955f728d-da4e-8b2f-1008-2aab86a58d74.jpeg)


**3.　デバイスにソフトウェアをインストール**

接続キットをダウンロード

![image-3.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/90c7184d-7758-920f-5262-8d869c250cca.jpeg)


**4.　SCPコマンドでダウンロードしたZipをRaspberry Pi に転送**

```
scp .\connect_device_package.zip [ユーザー名]@[IPアドレス]:/home/[ユーザー名]
```

Raspberry PiにSSH接続で入り、転送したZipフォルダを解凍する
```
unzip connect_device_package.zip
```

**5.　接続キットを実行**

解凍したファイルからstart.shに実行権限を追加

```
chmod +x start.sh
```

Pythonの仮想環境を作成する
```
python -m venv venv
```

Pythonの仮想環境へアクティベートする
```
source ./venv/bin/activate
```

:::note warn
Pythonの仮想環境へアクティベートした状態でないと、start.sh中にエラーになります
:::

インストールスクリプトを実行する
```
./start.sh
```

→ インストールが完了すると、アプリケーションが動作し始めます
```
Publishing message to topic 'sdk/test/python': Hello World!  [1]
Received message from topic 'sdk/test/python': b'"Hello World!  [1]"'
Publishing message to topic 'sdk/test/python': Hello World!  [2]
Received message from topic 'sdk/test/python': b'"Hello World!  [2]"'
Publishing message to topic 'sdk/test/python': Hello World!  [3]
Received message from topic 'sdk/test/python': b'"Hello World!  [3]"'
Publishing message to topic 'sdk/test/python': Hello World!  [4]
Received message from topic 'sdk/test/python': b'"Hello World!  [4]"'
Publishing message to topic 'sdk/test/python': Hello World!  [5]
Received message from topic 'sdk/test/python': b'"Hello World!  [5]"'
```

AWSのコンソールパネルで送信されたメッセージがサブスクリプションされていることが分かる。


![image-4.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/1407a8fc-44d0-71cd-66e6-e7aa764a60cf.png)


これでRaspberryPiをAWS IoT Coreの連携は一旦完了


## MQTTパブリッシュプログラム作成

ここからRaspberry Piに実装するスクリプトの作成に入る

### ファイル構成

今回作成するプログラムのファイル構成は以下。

```
├── mqtt_publish.py
├── utils
│   └── ds18B20.py
├── venv
└── .env
```

### mqtt_publish.py

今回のメインになるプログラム。

RaspberryPiからIoT CoreへMQTTプロトコルを使用してパブリッシュする。

```python
import os
from os.path import join, dirname
from dotenv import load_dotenv
from awscrt import mqtt, http
from awsiot import mqtt_connection_builder
import sys
import threading
import time
import json
from datetime import datetime, date
from utils import ds18b20

# 環境変数を取得
dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)
ENDPOINT = os.environ.get("ENDPOINT")
PORT = int(os.environ.get("PORT"))
CERT_PATH = os.environ.get("CERT_FILEPATH")
PRIVATE_KEY_PATH = os.environ.get("PRI_KEY_FILEPATH")
CA_PATH = os.environ.get("CA_FILEPATH")
CLIENT_ID = os.environ.get("CLIENT_ID")
MESSAGE_COUNT = int(os.environ.get("MESSAGE_COUNT"))
MESSAGE_TOPIC = os.environ.get("MESSAGE_TOPIC")

received_count = 0
received_all_event = threading.Event()

# 接続切断時コールバック
def on_connection_interrupted(connection, error, **kwargs):
    print("Connection interrupted. error: {}".format(error))


# コネクション接続が再確立コールバック
def on_connection_resumed(connection, return_code, session_present, **kwargs):
    print("Connection resumed. return_code: {} session_present: {}".format(return_code, session_present))

    if return_code == mqtt.ConnectReturnCode.ACCEPTED and not session_present:
        print("Session did not persist. Resubscribing to existing topics...")
        resubscribe_future, _ = connection.resubscribe_existing_topics()
        resubscribe_future.add_done_callback(on_resubscribe_complete)


def on_resubscribe_complete(resubscribe_future):
    resubscribe_results = resubscribe_future.result()
    print("Resubscribe results: {}".format(resubscribe_results))

    for topic, qos in resubscribe_results['topics']:
        if qos is None:
            sys.exit("Server rejected resubscribe to topic: {}".format(topic))


# サブスクライブされたトピックがメッセージを受信したときのコールバック
def on_message_received(topic, payload, dup, qos, retain, **kwargs):
    print("Received message from topic '{}': {}".format(topic, payload))
    global received_count
    received_count += 1
    if received_count == MESSAGE_COUNT:
        received_all_event.set()

# 接続が成功したときのコールバック
def on_connection_success(connection, callback_data):
    assert isinstance(callback_data, mqtt.OnConnectionSuccessData)
    print("Connection Successful with return code: {} session present: {}".format(callback_data.return_code, callback_data.session_present))

# 接続試行が失敗した場合のコールバック
def on_connection_failure(connection, callback_data):
    assert isinstance(callback_data, mqtt.OnConnectionFailureData)
    print("Connection failed with error code: {}".format(callback_data.error))

# 接続がクローズされたときのコールバック
def on_connection_closed(connection, callback_data):
    print("Connection closed")

def make_mqtt_connection():
    # MQTT接続を作成
    mqtt_connection = mqtt_connection_builder.mtls_from_path(
        endpoint=ENDPOINT,
        port=PORT,
        cert_filepath=CERT_PATH,
        pri_key_filepath=PRIVATE_KEY_PATH,
        ca_filepath=CA_PATH,
        on_connection_interrupted=on_connection_interrupted,
        on_connection_resumed=on_connection_resumed,
        client_id=CLIENT_ID,
        clean_session=False,
        keep_alive_secs=30,
        http_proxy_options=None,
        on_connection_success=on_connection_success,
        on_connection_failure=on_connection_failure,
        on_connection_closed=on_connection_closed
        )
    
    return mqtt_connection

def main():
    mqtt_connection = make_mqtt_connection()
    
    print(f"Connecting to {ENDPOINT} with client ID '{CLIENT_ID}'...")
    connect_future = mqtt_connection.connect()

    # コネクションが利用可能になるまで待機
    connect_future.result()
    print("Connected!")
    
    # 水温センサーから温度を取得
    temperature = ds18b20.measure()
    
    # JSONデータを作成
    datetime_now = datetime.now().strftime("%Y%m%d%H%M")
    message_json = json.dumps(
            {
                'device_id' : 'ds18b20',
                "temperature": temperature,
                "datetime"  : datetime_now,
            }
        )
    
    # MESSAGE_COUNTの数だけメッセージをパブリッシュする（0の場合は無限ループ）
    if MESSAGE_COUNT == 0:
        print("Sending messages until program killed")
    else:
        print("Sending {} message".format(MESSAGE_COUNT))

    publish_count = 1
    while (publish_count <= MESSAGE_COUNT) or (MESSAGE_COUNT == 0):
        print("Publishing message to topic '{}': {}".format(MESSAGE_TOPIC, message_json))
        
        mqtt_connection.publish(
            topic=MESSAGE_TOPIC,
            payload=message_json,
            qos=mqtt.QoS.AT_LEAST_ONCE)
        time.sleep(1)
        
        publish_count = publish_count + 1

    # コネクション切断
    print("Disconnecting...")
    disconnect_future = mqtt_connection.disconnect()
    disconnect_future.result()
    print("Disconnected!")

if __name__ == '__main__':
    main()

```


### utils/ds18B20.py
[前回の記事](https://qiita.com/syo_engineer/items/bb306a0455835b320d68) で作成したプログラムを参考に作成。

温度センサー（DS18B20）から温度を取得するプログラムをmqtt_publish.pyでインポートするライブラリとして使用する


```python
from w1thermsensor import W1ThermSensor

def measure():
    try:
        ds18b20_sensor = W1ThermSensor()
        temperature = round(ds18b20_sensor.get_temperature(),1)
    except Exception as e:
        temperature = -9999
        
    return temperature
```

### venv
Pythonの仮想環境

### .env 
環境変数の設定ファイル

```.env
ENDPOINT=hogehoge.iot.ap-northeast-1.amazonaws.com
PORT=8883
CERT_FILEPATH=raspberry_pi_4.cert.pem
PRI_KEY_FILEPATH=raspberry_pi_4.private.key
CA_FILEPATH=root-CA.crt
CLIENT_ID=raspberry_pi_4
MESSAGE_COUNT=1
MESSAGE_TOPIC=raspberry_pi_4/ds18b20/python
```

## トピック名とクライアントID変更
トピック名とクライアントIDがデフォルトのままになっているので、環境変数ファイル（.env）に記載した内容になるようコンソール画面で変更する

IoT Coreのコンソール画面から、[セキュリティ] > [ポリシー] > [raspberry_pi_4-Policy]に移動し、JSON記載画面に移動し変更を行う

![image-10.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/f070526c-0c48-66ce-313c-c5da52c0dbcb.jpeg)

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iot:Publish",
        "iot:Receive",
        "iot:PublishRetain"
      ],
      "Resource": [
        "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:topic/raspberry_pi_4/ds18b20/java",
        "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:topic/raspberry_pi_4/ds18b20/python",
        "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:topic/raspberry_pi_4/ds18b20/js"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": [
        "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:topicfilter/raspberry_pi_4/ds18b20/java",
        "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:topicfilter/raspberry_pi_4/ds18b20/python",
        "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:topicfilter/raspberry_pi_4/ds18b20/js"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": [
        "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:client/sdk-java",
        "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:client/raspberry_pi_4",
        "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:client/sdk-nodejs-*"
      ]
    }
  ]
}
```

## Amazon DynamoDBを作成する

ここまででRaspberry → IoT Coreの設定は完了したので、次はIoT Core → DynamoDB保存の設定を行う。

### トピックのJSON構造

DynamoDBの設定の前に一旦トピックの内容を確認

RaspberryPi → IoT CoreへのMQTTメッセージは、以下のJSON形式でパブリッシュされる

```json
{
    "device_id" : 'ds18b20',
    "temperature": temperature,
    "datetime"  : datetime_now,
}
```

### DynamoDBのテーブル構造

| 項目 | 値 | データ型 |
| ---- | ---- | ---- |
| テーブル名 | raspberry_pi_4_data |
| パーティションキー | device_id  | String |
| ソートキー | datetime  | String |

パーティションキーとソートキーで複合プライマリーキーとなるように設定する

（参考） [Amazon DynamoDB のコアコンポーネント](https://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/developerguide/HowItWorks.CoreComponents.html)


### DynamoDBテーブルの作成

DynamoDBのコンソール画面から [テーブル] > [テーブル作成] を開き、テーブル構造で決めた設定内容を入力する

![image-11.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/81a726f5-e11a-808a-2ef3-c3dbc233d54b.jpeg)


## IoT Core ルールを作成する

最後に、IoT Coreルールを作成し、IoT Coreにデバイスデータが飛んで来た時にDynamoDBへ保存する設定を行う

### ルールの設定

IoT Coreのコンソールから、[メッセージのルーティング] > [ルール] に移動し、[ルールを作成] する


![image-6.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/52ae0ff0-d02c-833b-6ffb-77b6c987cb37.jpeg)


ルール名： raspberry_pi_4_data

説明：IoT Coreで受信したraspberry pi からのJSONデータを DynamoDBに保存する


![image-7.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/b75ae9c2-f65a-6d9f-2cc1-2ca88f9bff61.jpeg)



### SQLステートメントの設定

受信したMQTTメッセージを、DynamoDBに保存する為のSQLを設定する

SQLのバージョン：2016-03-23

![image-12.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/383348ff-9f8b-074e-22ea-21f3df1d9871.jpeg)


SQLステートメント

```sql
SELECT
  device_id AS device_id,
  temperature AS temperature,
  datetime AS datetime,
FROM
  'raspberry_pi_4/ds18b20/python'
```

### ルールアクションをアタッチ

先ほど作成したDynamoDBテーブル設定をします。

- アクション：DynamoDBv2

- テーブル：raspberry_pi_4_data

- IAMロール：raspberry_pi_4_query_role（[新しいロールを作成] から新規作成します）

![image-8.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/50e7f6c4-3b57-5c96-347a-e6e33fcbe0c7.jpeg)

## 動作確認

### Raspberry Pi上でPythonスクリプトを実行する

```
python  ./mqtt_publish.py
```

IoT Coreへパブリッシュが正常に完了すれば、以下が出力される

```
Connecting to hogehoge.iot.ap-northeast-1.amazonaws.com with client ID 'raspberry_pi_4'...
Connection Successful with return code: 0 session present: False
Connected!
Sending 1 message
Publishing message to topic 'raspberry_pi_4/ds18b20/python': {"device_id": "ds18b20", "temperature": 27.6, "datetime": "202408150030"}
Disconnecting...
Connection closed
Disconnected!
```

### IoT Coreを確認する
IoT Coreのコンソール画面から。[MQTTクライアント]を開く。

トピックを指定し、サブスライブした状態でスクリプトを実行すると IoT Coreにパブリッシュされていることが分かる。

![image-13.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/6036c96f-fb09-dc62-1990-fbb81de1f086.jpeg)


### DynamoDBを確認する

DynamoDBのコンソール画面から、[項目を探索] 画面を開く。

[raspberry_pi_4_data] テーブルの画面を開くと、IoT Coreでサブスクリプションしたデータが保存されていることが分かる

![image-14.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3774903/90150b9c-5ea1-8304-3485-6bf2e6aee41d.jpeg)


## 最後に
これで水温測定 → Raspberry Piから、IoT Core → DynamoDBまでデータと送れるようになったので、cron設定や定期実行用スクリプトを作成すれば定期的にDynamoDBへデータを送信することができる。
そこまでできればWebアプリケーションとの連携準備が整うはずなので、今後はWebアプリケーションからDynamoDBのデータを取得して、データ可視化を目標にやっていきたいと思う。

## 参照記事
[Raspberry Pi + DS18B20で水温測定](https://qiita.com/syo_engineer/items/bb306a0455835b320d68)