AWS IoT Enterprise Buttonの入力結果をLambda経由でDynamoDBに保存し、API GatewayからM5Flowを使ってM5Stack Core2に表示する

この記事はメディカルハッカソンで作成されました。

積みガジェットや普段使わない技術を触ってみる目的で色々組み合わせてIoT数取器を作ってみた。

構成図

AWS IoT Enterprise Button - Lambda - DynamoDB

AWS IoT Enterprise Buttonの設定

iPhoneから初期設定を行います。AppleStoreからAWS IoT 1-Clickをインストールします。

AWS ルートユーザでログインした後、リージョンの設定を行います。今回は東京リージョン。

デバイスを登録します。

デバイスIDを登録でスキャンを選択し、AWS IoT Enterprise Buttonの外装に張られたバーコードを読み取ります。

インターネットに接続するためWi-Fiの設定もしておきます。2.4GHz帯を使います。

ブラウザでAWS IoT 1-Clickに接続し、登録状況を確認します。デバイスが無効になっている場合、有効にしてください。アプリからでも可能です。

また、動作確認のためボタンを押下（SINGLE、DOUBLE、LONG）してイベントログを確認します。LONGは1.5sec以上の押下で判定されるようです。アプリからでもブラウザからでも確認できます（画像はアプリ）。remainingLifeは電池の残量ですが、電池交換可能なデバイスだと意味が異なったりするようです。

DynamoDBの設定

マネジメントコンソールから、テーブルの作成を選択し、パーティションキーとソートキーを入力します。テーブルの名前は適当に設定します。

テーブルの詳細がこんな感じになっていればOK

Lambdaの設定

関数の作成で、一から作成を選び、ランタイムにPython3を選択します。適当な関数名を付けて、その他はデフォルトのまま作成完了します。

コードソースを以下のように編集してデプロイします。DynamoDBのテーブル名は先ほど作成したテーブル名を入力してください。

lambda

import json
import boto3
import datetime

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your table') # DynamoDBテーブル名を入力

def lambda_handler(event, context):
    clickType = event['deviceEvent']['buttonClicked']['clickType']
    timestamp = event['deviceEvent']['buttonClicked']['reportedTime']
    
    # クリックタイプとタイムスタンプをDynamoDBに保存
    response = table.put_item(
        Item={
            'clickType': clickType,
            'timestamp': timestamp
        }
    )
    
    return {
        'statusCode': 200,
        'body': json.dumps('ClickType and Timestamp saved successfully!')
    }

関数に割り当てたIAMロールのポリシーにDynamoDBへの書き込み権限を付与します。

IAMロールのポリシー設定

Lambdaの実行ロールのリンクからIAMのロールを表示し、許可ポリシーを選択します。

DynamoDBの書き込み権限を付与します。最小権限にするため、もっと絞ってもよさそうです。

AWS IoT 1-Clickの設定

プロジェクトを作成します。

デバイステンプレートのアクションでLambda関数を選択して、プロジェクトを作成します。

作成したプロジェクトにデバイスを割り当てます。プロジェクトの詳細からプレイスメントを作成します。

AWS IoT Enterprise Buttonを登録します。

動作確認

AWS IoT Enterprise Buttonを押下し、DynamoDBに値が保存されていることを確認します。

AWS IoT 1-Clickのプロジェクトからデバイスのアクティビティを確認できます。

DynamoDB - Lambda - API Gateway - M5Stack Core2

Lambda作成

同様に関数を作成します。

コードを設定。

lambda

import json
import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your table')

def lambda_handler(event, context):
    response = table.scan()
    items = response['Items']
    
    return {
        'statusCode': 200,
        'body': json.dumps(items)
    }

IAMロールのポリシー設定

Lambdaで使用中のロールの許可ポリシーに、DynamoDBの読み取り権限を付与します。

API Gateway設定

マネジメントコンソールからAPI Gatewayを開き、HTTP APIを構築します。

バックエンドに先ほど作成したLambdaを選択します

ルートを設定します。読み取りだけなのでGETだけでOK

ステージに対応したURLに設定したパスを付けて、ブラウザで読み込んでみましょう。DynamoDBのデータが表示されれば成功です。

M5Stack Core2 IoT設定

M5BurnerでM5Stack Core2の最新UIFlowをダウンロードします。

使用するWi-Fi情報を設定し、M5Stack Core2 IoTとM5Burnerを起動中のPCをUDBケーブルで接続し書き込みます。正常に書き込み完了し、起動すると「チーン！」と音がします。結構うるさい。

Wi-Fiの設定をします。SSIDとIPが表示されます。

スマホ等でネットワークに接続してブラウザにIPアドレスを入力すると、設定画面が表示されます。自宅のWi-Fi情報を入力してConfigureします。

設定が完了するとM5Stackが再起動します。Wi-Fi接続に成功するとAPI KEYが表示されます。

UIFlow設定

M5Flowのページから設定を行います。
M5Stackの画面に表示されたAPI KEYを入力します。

画面のレイアウトを決めてラベルを設定します。今回は数取器に今日のカウント機能と合計を表示できるようにしてみます。

Lambda応答をUIFlowで使いやすいように修正

作成済みのLambdaのコードを修正します。
ClickTypeがSINGLEなら+1、DOUBLEなら-1のようにカウントダウン機能も付けます。
本日分だけを返却してほしいのでフィルタ処理を書きました。
また、M5Flowのラベルにそのまま表示させたいので応答を0埋め4桁の文字列にしました。

today_count

import json
import boto3
from datetime import datetime, timezone

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your table')

def is_today(timestamp_str):
    timestamp = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
    today = datetime.now(timezone.utc).date()
    return timestamp.date() == today

def lambda_handler(event, context):
    response = table.scan()
    items = response['Items']
    
    today_data = [item for item in items if is_today(item['timestamp'])]
    
    single_count = sum(item['clickType'] == 'SINGLE' for item in today_data)
    double_count = sum(item['clickType'] == 'DOUBLE' for item in today_data)
    result = single_count - double_count
    result_str = f"{result:04d}"
    
    return {
        'statusCode': 200,
        'body': result_str
    }

API Gatewayのエンドポイントにアクセスし想定通りの応答が返ってくることを確認します。
また、IoT Button押下のSINGLE/DOUBLEによって増減することを確認します。

M5FlowでHTTP Requestの結果をラベルに表示させるブロックを組みRunで実行します。

上手く表示されていればOK。心なしか変更前のラベル値が薄っすら見えます。なぜ。

合計値を返すエンドポイントも作成しましょう。LambdaとAPI Gatewayをそれぞれ用意する必要があります。Lambdaの関数コピー機能は無さそうなので地道にやります。ロールを使い増せばIAM変更は省略できます。

total_count

import json
import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your table')

def lambda_handler(event, context):
    response = table.scan()
    items = response['Items']
    
    single_count = sum(item['clickType'] == 'SINGLE' for item in items)
    double_count = sum(item['clickType'] == 'DOUBLE' for item in items)
    result = single_count - double_count
    result_str = f"{result:04d}"
    
    return {
        'statusCode': 200,
        'body': result_str
    }

API Gatewayの作成も完了したらM5Flowに組み込みましょう。

カウント値を定期的に更新する

本当は更新あったときだけM5Stackから確認orデータ更新をM5Stackに通知するのが理想でMQTT等が必要ですが、今回は1秒に1回確認しに行くことで実装します。

API Gatewayの料金が3億リクエスト/1.29USDで、Lambdaが100万リクエスト/0.2USD。1秒に1回、2つのエンドポイントに確認しても18万リクエストなのでしばらくは無料枠で賄えそうです。

とりあえず完成！

拡張性

今回はNode-REDとMQTTを使ってみたかったのですが時間切れでした。
またenebulerについてもクラウド実行環境なら便利に使えそうなので次回試してみたいです。

enebular設定(未完)

最初、enebularを使ってAPI GatewayとM5Stackを連携させようとした時の資料。
(ChatGPT4がクラウド実行環境の存在を教えてくれなかった)

供養として載せておきます。

enebularにサインインし、プロジェクトを新規作成し、フローを追加します。

フローの編集画面でhttp requestとhttp responseのノードを作成し接続します。

http requestのURLに、先ほど作成したエンドポイントを設定します。TLS接続も有効化。

設定を終えたら保存してデプロイします。