この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
こんにちは、CX事業本部MAD事業部の森茂です。
突然ですが、我が家の子どもたちにとってゲームの時間はとても重要で、1日の制限時間は1秒たりともムダにできません。
おとな:もう3時間はゲームしてるんじゃない?
子ども:まだ2時間しかやっていない。
おとな:もう3時間過ぎているでしょう?もう終わりにしなさい。
子ども:ぴえん
というわけで、親子間のムダな争いを避けそれぞれの権利・主張・平和のために子どもたちが今日どれくらいゲームをやったのかをRaspberry PiとAWSを利用して可視化してみました。
はじめに
つくってみた構成はこちら。
若干省略していますが最終的にこのような形になりました。
システム構成
まず、Raspberry Pi ZeroWHを使ってゲームの開始、終了時間を測るタイマーボタンを設置。ボタンを押すごとにAPI Gateway経由でLambdaを起動、ゲームの開始時間、終了時間をDynamoDBに保存します。
開始時、終了時にはおとなのLINEに通知を行い、統計情報はS3にホスティングしたReactアプリケーションから閲覧できるようにします。オマケ機能としてEventBridgeとLambdaで毎夜合計時間を集計して、直近一週間の合計時間も閲覧できるようにしてみます。
利用ユーザー想定
今回の利用ユーザーはLucyとMikeの2名を想定。それぞれがゲームの開始時と終了時、自分に割り当てられたボタンを押します。
開発環境
- M1 MacBook Pro(Big Sur)
- Raspberry Pi ZeroWH
- Python 3.9.7
- node.js v16.10.0
バックエンド環境構築
バックエンドの環境構築にはCDK v2を使っていきます。今回はRaspberry Piを使うのでいつもならTypeScriptでまとめるところ、あまり書く機会のなかったPythonで挑戦してみました。
なお、CDKの利用にあたってAWS CLIの用意、また認証の設定などをあらかじめ完了しておく必要があります。
またLINE Notifyで通知を行う設計のためLINE Notifyトークンが必要になります。下記を参考にトークンの取得をしておきます。
CDK v2のインストール
今回はまだRC版ではありますがv2系を利用していきます。(RC23まできており個人用途であれば問題は少ないのではと考えています。) ※v1系でもライブラリのimportなどが若干異なりますがほぼ同じような構成で組み上げること可能です。
$ npm install -g cdk@next
$ cdk --version
2.0.0-rc.23 (build 1e54fb9)
CDKでスタックの雛形を作成
cdk init app
スクリプトを利用してスタックの雛形を作ります。
$ cdk init app --language python
$ python3 -m venv .venv
$ source .venv/bin/activate # 以後は仮想環境内での作業
$ pip install -r requirements.txt
セットアップ後はこのような構成になりました。
├── README.md
├── app.py
├── cdk.json
├── game_counter
│ ├── __init__.py
│ ├── __pycache__
│ ├── game_counter.egg-info
│ └── game_counter_stack.py
├── requirements.txt
├── setup.py
└── source.bat
なお、CDKをはじめて利用する環境の場合、cdk bootstrap
でAWS上に受け入れ環境を用意しておく必要があります。(v1系とは異なる受け入れ環境が作成されるためv2系をはじめ利用の場合はこの作業が必要です)
$ cdk bootstrap
追加ライブラリのインストール
今回アプリケーションでは外部ライブラリrequests
を利用するためインストールしておきます。
その他にもライブラリをimportしていますがLambdaのPython3.9環境にはあらかじめ用意されているライブラリのためローカルで動作確認を行う必要がなければ不要でした。
$ pip install requests
CDKでインフラ環境を構築
├── game_counter
│ └── game_counter_stack.py
今回はシンプルな構成のためファイルは分割せずgame_counter_stack.py
にすべてのリソースを記載していきます。
手始めにDynamoDBのスキーマについては下記のようにしてみました。
ログ記録用テーブル(GameCounter)
Raspberry Piからusername
timestamp
flag
をAPI経由で渡す想定。
項目名 | データ型 | 種別 | 備考 |
---|---|---|---|
username | String | パーティションキー | 名前(Lucy、Mike) |
timestamp | Number | ソートキー | UNIXTIME(秒) |
date | String | 年月日(2021-10-10) | |
time | String | 時間(12:00:00) | |
flag | String | 開始、終了のフラグ(start | end) | |
uid | String | uuid(ID、機能拡張時に使うかも?) |
uid
については今後の拡張で個別ログの編集や削除することを考えて用意しています。
集計用テーブル(GameCounterLog)
EventBridgeを利用して毎日ログを集計するためのテーブル。
項目名 | データ型 | 種別 | 備考 |
---|---|---|---|
username | String | パーティションキー | 名前 |
date | String | ソートキー | 年月日(2021-10-10) |
totaltime | String | 一日の合計時間(4:00:00) |
DynamoDBの作成
さっそくテーブルをCDKで構築してみます。
game_counter_stack.py
from aws_cdk import (
Stack,
RemovalPolicy,
aws_dynamodb as dynamodb, # DynamoDBのライブラリをimport
)
from constructs import Construct
class GameCounterStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# The code that defines your stack goes here
# ここから下に追記していきます。
# DynamoDB ログデータ格納用
table = dynamodb.Table(
self, "GameCounterTable",
table_name="GameCounter",
partition_key=dynamodb.Attribute(
name="username", type=dynamodb.AttributeType.STRING),
sort_key=dynamodb.Attribute(
name="timestamp", type=dynamodb.AttributeType.NUMBER),
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
removal_policy=RemovalPolicy.DESTROY
)
# DynamoDB 統計データ格納用
log_table = dynamodb.Table(
self, "GameCounterLogTable",
table_name="GameCounterLog",
partition_key=dynamodb.Attribute(
name="username", type=dynamodb.AttributeType.STRING),
sort_key=dynamodb.Attribute(
name="date", type=dynamodb.AttributeType.STRING),
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
removal_policy=RemovalPolicy.DESTROY
)
この時点でいったんCDKの動作を確認してみます。
cdk synth
で想定される構成の確認、エラーがないようであればcdk deploy
でデプロイします。
$ cdk synth
$ cdk deploy
念の為CLIからテーブルの一覧を確認。
$ aws dynamodb list-tables
{
"TableNames": [
"GameCounter",
"GameCounterLog"
]
}
ここからは一気にいっちゃいます。 まずは必要なライブラリのimportまわりから。
game_counter_stack.py
from aws_cdk import (
Stack,
RemovalPolicy,
+ Duration,
+ CfnOutput,
aws_dynamodb as dynamodb,
+ aws_lambda as lambda_,
+ aws_apigateway as apigateway,
+ aws_events as events,
+ aws_events_targets as targets
)
from constructs import Construct
さきほど作成したDynamoDBの記載下から一気にリソース(Lambda、API Gateway、S3、EventBridge)を追加します。
game_counter_stack.py
class GameCounterStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
....
# Lambda用共通パラメーター
common_params = {
"runtime": lambda_.Runtime.PYTHON_3_9,
"timeout": Duration.seconds(10),
}
# Lambda データ一覧表示用
list_data = lambda_.Function(
self, "ListData",
code=lambda_.Code.from_asset("lambda/list_data"),
handler="index.handler",
function_name="GameCounter-ListData",
environment={
"TABLE_NAME": table.table_name,
},
**common_params
)
# DynamoDBへ権限を付与
table.grant_read_data(list_data)
# SSMへのアクセス用のIAMロールを設定
role_for_lambda = iam.Role(
self, "Role",
role_name="GameCounterSSMRole",
assumed_by=iam.ServicePrincipal(
"lambda.amazonaws.com"),
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name(
'service-role/AWSLambdaBasicExecutionRole'),
iam.ManagedPolicy.from_aws_managed_policy_name(
'AmazonSSMReadOnlyAccess')
]
)
# Lambda データ投稿用
push_data = lambda_.Function(
self, "PushData",
code=lambda_.Code.from_asset("lambda/push_data"),
handler="index.handler",
function_name="GameCounter-PushData",
# LINEトークンをSSMから取得するためにRoleを割り当て
role=role_for_lambda,
environment={
"TABLE_NAME": table.table_name,
},
**common_params
)
# DynamoDBへ権限を付与
table.grant_write_data(push_data)
# Lambda ログデータ統計用
create_log = lambda_.Function(
self, "CreateLog",
code=lambda_.Code.from_asset("lambda/create_log"),
handler="index.handler",
function_name="GameCounter-CreateLog",
environment={
"TABLE_NAME": table.table_name,
"LOG_TABLE_NAME": log_table.table_name,
},
**common_params
)
# DynamoDBへ権限を付与
table.grant_read_data(create_log)
log_table.grant_write_data(create_log)
# Lambda データ一覧表示用
list_log = lambda_.Function(
self, "ListLog",
code=lambda_.Code.from_asset("lambda/list_log"),
handler="index.handler",
function_name="GameCounter-ListLog",
environment={
"LOG_TABLE_NAME": log_table.table_name,
},
**common_params
)
# DynamoDBへ権限を付与
log_table.grant_read_data(list_log)
# ApiGatewayを作成
api = apigateway.RestApi(
self, "GameCounterApi",
default_cors_preflight_options={
"allow_origins": apigateway.Cors.ALL_ORIGINS,
"allow_methods": apigateway.Cors.ALL_METHODS
})
# ApiGatewayのURI/counterというエンドポイントを作成
counter_list = api.root.add_resource("counter")
# /counter にGETでアクセスの際には一覧表示のLambdaを割り当てる
counter_list.add_method(
"GET", apigateway.LambdaIntegration(list_data),
api_key_required=True
)
# /counter にPOSTでアクセスの際にはログ書き込みのLambdaを割り当てる
counter_list.add_method(
"POST", apigateway.LambdaIntegration(push_data),
api_key_required=True
)
# 統計データ取得用のエンドポイント作成(counter/log)
log_list = counter_list.add_resource("log")
# /counter/log にGETでアクセスの際にはログ一覧表示のLambdaを割り当てる
log_list.add_method(
"GET", apigateway.LambdaIntegration(list_log),
api_key_required=True
)
# API Keyを利用するために利用料プランとAPI KEYを作成
plan = api.add_usage_plan(
"UsagePlan",
name="GameCounterPlan",
throttle={
"rate_limit": 10,
"burst_limit": 2
},
api_stages=[{"api": api, "stage": api.deployment_stage}]
)
key = api.add_api_key(
"GameCounterApiKey",
api_key_name="GameCounterApiKey"
)
plan.add_api_key(key)
# API Keyを確認するためにAPI KeyのIDをコンソールに出力
CfnOutput(
self, "GameCounterApiKeyExport",
value=key.key_id,
export_name="GameCounterApiKeyId"
)
# EventBridge 毎日24:30に前日の集計を行うLambdaを実行
events.Rule(
self, "GameCounterEventRule",
rule_name="GameCounterEventRule",
schedule=events.Schedule.cron(
minute="30",
hour="15",
day="*"),
targets=[targets.LambdaFunction(create_log, retry_attempts=3)]
)
SSMにLINE Notify用トークンを格納
あらかじめ取得していたLINE Notify用のトークン情報をLambda上で利用するためSystems Managerのパラメーターストアに情報を入れておきます。
$ aws ssm put-parameter \
--type 'SecureString' \
--name '/GameCounter/LineToken' \
--value 'LINE Notifyトークン情報'
Lambda関数の用意
プロジェクトルートにlambda
ディレクトリを作成して用途別にlambda関数を用意します。create_log
はEventBridgeから定期実行され、それ以外はAPI Gateway経由で実行してDynamoDBを操作するLambdaになります。
※バリデート処理など各種エラー処理は簡素化もしくは未実装のためご留意ください。またデータ量が少なく1回のクエリで全データ取得できる想定で、仮に取りこぼしても致命的な問題になるようなシステムではないため処理を一部省略しています。
./lambda
├── create_log # 統計ログ書き込み用
│ └── index.py
├── list_log # 統計ログ一覧表示用
│ └── index.py
├── list_data # ログ一覧表示用
│ └── index.py
└── push_data # ログ書き込み用
└── index.py
ログ書き込み用Lambda(lambda/push_data/index.py)
Raspberry PiからAPI Gateway経由で情報を受け取るLambda。requests
ライブラリが別途必要なためあらかじめ同梱しておく必要があり。(今回はライブラリもそれを利用するLambdaも1つのみのためLayer化せずにそのまま入れています)
$ pip freeze | grep requests > lambda/push_data/requirements.txt
$ cd lambda/push_data
$ pip install -r requirements.txt -t .
lambda/push_data/index.py
import simplejson as json
import os
import uuid
import datetime
import boto3
from boto3.dynamodb.conditions import Key
import requests
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
headers = {
"Content-Type": "application/json",
"Access-Control-Allow-Origin": "*",
}
# SSMパラメーターストアから値を取得
def get_parameters(param_key, ssm=None):
if not ssm:
ssm = boto3.client('ssm')
response = ssm.get_parameter(
Name=param_key,
WithDecryption=True
)
return response['Parameter']['Value']
# UNIX時間をJSTのdatetimeに変換
def convert_timestamp(target):
result = datetime.datetime.fromtimestamp(
target, datetime.timezone(datetime.timedelta(hours=9)))
return result
def handler(event, context):
print("Received event: " + json.dumps(event))
# クエリーパラメータの存在確認
if event.get("body"):
parsed = json.loads(event["body"])
uid = str(uuid.uuid4())
username = parsed["username"]
timestamp = parsed["timestamp"]
date = convert_timestamp(parsed["timestamp"]).date().isoformat()
time = convert_timestamp(parsed["timestamp"]).time().isoformat()
flag = parsed["flag"]
else:
username = None
try:
# usernameがある場合はDynamoDBにデータを登録
if username:
response = table.put_item(
Item={
"uid": uid,
"username": username,
"timestamp": int(timestamp),
"date": date,
"time": time,
"flag": flag
}
)
if response["ResponseMetadata"]["HTTPStatusCode"] != 200:
raise Exception("Unable to connect to the database.")
else:
status_code = 200
res = {"message": f"Created item with id: {uid}"}
# LINEへ通知
line_token = get_parameters("/GameCounter/LineToken")
line_url = "https://notify-api.line.me/api/notify"
line_headers = {"Authorization": "Bearer " + line_token}
line_message = "flag未取得"
if flag == "start":
line_message = username + "がゲームを開始しました"
if flag == "end":
line_message = username + "がゲームを終了しました"
payload = {"message": line_message}
requests.post(
line_url, headers=line_headers,
params=payload
)
# usernameがない場合はエラーとして終了
else:
raise Exception("Parameter is missing.")
except Exception as e:
status_code = 500
res = {"message": f"Internal Server Error. {str(e)}"}
return {
"statusCode": status_code,
"headers": headers,
"body": json.dumps(res)
}
Raspberry Piからは以下のデータをJSON形式で受け取る想定です。
{
"username": "Lucy", // String
"timestamp: 1634113452, // Number unixtime(秒)
"flag": "start" // start | end
}
ログ一覧表示用Lambda(lambda/list_data/index.py)
フロントエンド側からユーザー別、日付別のデータを取得するためのLambda。
lambda/list_data/index.py
import simplejson as json
import os
import boto3
from boto3.dynamodb.conditions import Key
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
headers = {
"Content-Type": "application/json",
"Access-Control-Allow-Origin": "*",
}
def handler(event, context):
print("Received event: " + json.dumps(event))
# クエリーパラメータの存在確認
if event.get("queryStringParameters"):
username = event['queryStringParameters'].get("username")
starttime = event['queryStringParameters'].get("starttime")
endtime = event['queryStringParameters'].get("endtime")
else:
username = None
try:
# usernameの指定がある場合はusernameの指定した期間の一覧を出力
if username:
response = table.query(
KeyConditionExpression=Key("username").eq(username) & Key(
"timestamp").between(int(starttime), int(endtime))
)
# usernameの指定がない場合は全件一覧を出力
else:
response = table.scan()
if response["ResponseMetadata"]["HTTPStatusCode"] != 200:
raise Exception("Unable to connect to the database.")
else:
status_code = 200
res = response["Items"]
except Exception as e:
status_code = 500
res = {"message": f"Internal Server Error. {str(e)}"}
return {
"statusCode": status_code,
"headers": headers,
"body": json.dumps(res)
}
クエリパラメーターとして
- username…名前
- starttime…開始時間(unixtime)
- endtime…終了時間(unixtime)
を受け取ります。
統計ログ書き込み用Lambda(lambda/create_log/index.py)
EventBridgeから定期的に起動するLambda。GameCounterテーブルからデータを取得して合計時間をGameCounterLogテーブルに書き込みます。
lambda/create_log/index.py
import simplejson as json
import os
import boto3
from boto3.dynamodb.conditions import Key, Attr
import datetime
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
log_table = dynamodb.Table(os.environ["LOG_TABLE_NAME"])
def query_today(username, target_date):
response = table.query(
KeyConditionExpression=Key("username").eq(username),
FilterExpression=Attr("date").eq(target_date)
)
return response["Items"]
def create_total_time(log_data):
total = 0
for i, item in enumerate(log_data):
temp = 0
try:
if log_data[i + 1].get("flag") == 'end':
temp = int(log_data[i + 1]["timestamp"]) - \
int(item["timestamp"])
except:
print('done')
total += temp
return total
def handler(event, context):
# 24時過ぎの計測を想定するため1日前の日付に変換
today = str((datetime.datetime.now(
datetime.timezone(datetime.timedelta(hours=9))) - datetime.timedelta(days=1)).date())
try:
# 対象の今日のログを集計、DBへ格納する
targets = ["Lucy", "Mike"]
for target in targets:
res = query_today(target, today)
total_time = create_total_time(res)
data = str(datetime.timedelta(seconds=total_time or 0))
if data:
response = log_table.put_item(
Item={
"username": target,
"date": today,
"totaltime": data,
}
)
except Exception as e:
print(f"Internal Server Error. {str(e)}")
return "OK"
統計ログ一覧表示用Lambda(lambda/list_log/index.py)
フロントエンド側からユーザー別の統計情報を取得するためのLambda。
lambda/list_log/index.py
import simplejson as json
import os
import boto3
from boto3.dynamodb.conditions import Key
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["LOG_TABLE_NAME"])
headers = {
"Content-Type": "application/json",
"Access-Control-Allow-Origin": "*",
}
def handler(event, context):
print("Received event: " + json.dumps(event))
# クエリーパラメータの存在確認
if event.get("queryStringParameters"):
username = event['queryStringParameters'].get("username")
target_date = event['queryStringParameters'].get("date")
else:
username = None
try:
# usernameの指定がある場合はusernameの指定した期間の一覧を出力
if username:
response = table.query(
KeyConditionExpression=Key("username").eq(username) & Key(
"date").gte(target_date)
)
# usernameの指定がない場合は全件一覧を出力
else:
response = table.scan()
if response["ResponseMetadata"]["HTTPStatusCode"] != 200:
raise Exception("Unable to connect to the database.")
else:
status_code = 200
res = response["Items"]
except Exception as e:
status_code = 500
res = {"message": f"Internal Server Error. {str(e)}"}
return {
"statusCode": status_code,
"headers": headers,
"body": json.dumps(res)
}
クエリパラメーターとして
- username…名前
- date…日付
を受け取ります。
CDKのデプロイ
ここまで一気に作ったところで環境をデプロイします。
$ cdk synth
$ cdk deploy
APIエンドポイント、API Keyの確認
デプロイ完了後、ターミナルコンソールにAPIのエンドポイントとAPIキーのIDが出力されることを確認してみます。
GameCounterStack.GameCounterApiEndpoint**** = https://****.execute-api.ap-northeast-1.amazonaws.cazonaws.com/prod/
GameCounterStack.GameCounterApiKeyExport = ********
エンドポイント末尾にcounter
を追加したものが、一覧取得(GET)とログ登録(POST)に、末尾にcounter/log
を追加したものが統計情報の一覧取得(GET)になります。
- https://xxxx.execute-api.ap-northeast-1.amazonaws.cazonaws.com/prod/counter
- https://xxxx.execute-api.ap-northeast-1.amazonaws.cazonaws.com/prod/counter/log
またAPIキーの確認のためターミナルからAWS CLIを使い下記コマンドで確認しておきます。APIキーはフロントエンド、Raspberry Piからそれぞれ利用します。
$ aws apigateway get-api-key \
--api-key APIキーID \
--include-value
{
"id": "APIキーID",
"value": "**APIキー**", #APIキーはフロントエンド、Raspberry Piから利用
"name": "GameCounterApiKey",
"enabled": true,
"createdDate": "2021-10-10T15:18:17+09:00",
"lastUpdatedDate": "2021-10-10T15:18:17+09:00",
"stageKeys": [
"xxxxxxxx/prod"
],
"tags": {}
}
Raspberry Pi環境構築
やっとバックエンド周りが用意できました。次にボタンを押すとゲーム開始、さらに押すと終了の通知を送る装置をRaspberry Pi ZeroWHを使って構築します。今回はブレッドボードを利用してざっくりと装置をつくってみました。
装置のイメージ
タクトスイッチを押すとLEDが点灯し、もう一度押すとLEDが消灯する仕様。
用意したもの
- Raspberry Pi ZeroWH
- ブレッドボード
- LED x2個(Vf2.2Vくらい?)
- ジャンパーワイヤー(オス-オス、オス-メス)いくつか
- 抵抗(100Ω x2、 10kΩ x2)
- タクトスイッチ x2
10kΩの抵抗はプルアップ用に用意しています。Raspberry Piの本体側でも用意されているので、そちらを利用する場合は不要ですが今回は抵抗器を使って実装しています。
制御プログラム
装置側ができあがったところで、次に装置を制御するプログラムをRaspberry Piに用意していきます。
API Gatewayへボタンを押した際にリクエストを送るために外部ライブラリとしてrequests
をインストール。
$ pip install requests
app.py
import RPi.GPIO as GPIO
import time
import sys
import requests
# GPIOポートの設定
SwGpio = 24
Sw2Gpio = 6
LedGpio = 14
Led2Gpio = 5
lastStatus = 0
ledStatus = 0
last2Status = 0
led2Status = 0
# GPIOの設定
GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)
# 1つ目のボタン
GPIO.setup(SwGpio, GPIO.IN)
GPIO.setup(LedGpio, GPIO.OUT)
# 2つ目のボタン
GPIO.setup(Sw2Gpio, GPIO.IN)
GPIO.setup(Led2Gpio, GPIO.OUT)
# API情報 API GatewayのURLとAPI Keyを設定
API_URL = "https://xxxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod/counter"
API_KEY = "****************"
def pushCount(name, flag):
print("start")
# UNIX時間(秒)で現在時間を取得
timestamp = int(time.time())
print(timestamp)
url = API_URL
headers = {"x-api-key": API_KEY}
q = {'username': name, 'timestamp': timestamp, 'flag': flag}
r = requests.post(url, headers=headers, json=q)
print(r.json())
while True:
try:
btnStatus = not(GPIO.input(SwGpio))
btn2Status = not(GPIO.input(Sw2Gpio))
# 1つ目のボタンの動作を制御 今回は青いLEDをLucy用のボタンとします。
if btnStatus == 0 and lastStatus == 1:
ledStatus = not ledStatus
if ledStatus == True:
GPIO.output(LedGpio, True)
pushCount('Lucy', 'start')
print('button 1 ON')
else:
GPIO.output(LedGpio, False)
pushCount('Lucy', 'end')
print('button 1 OFF')
# 連続でボタンを誤認識しないように少し待つ
time.sleep(0.2)
# 2つ目のボタンの動作を制御 今回は赤いLEDをMike用のボタンとします。
if btn2Status == 0 and last2Status == 1:
led2Status = not led2Status
if led2Status == True:
GPIO.output(Led2Gpio, True)
pushCount('Mike', 'start')
print('button 2 ON')
else:
GPIO.output(Led2Gpio, False)
pushCount('Mike', 'end')
print('button 2 OFF')
# 連続でボタンを認識しないように少し待つ
time.sleep(0.2)
lastStatus = btnStatus
last2Status = btn2Status
except KeyboardInterrupt:
GPIO.cleanup()
sys.exit()
制御プログラムの用意ができたので、早速スクリプトを起動し、ボタンを押してデータを送信してみます。
$ python3 ./app.py
# ボタンを押すと開始
$ python3 game-counter/app.py
start
1634044409
{'message': 'Created item with id: a968f92a-7828-462a-b26a-bcbb3fab2810'}
button 2 ON
# もう一度ボタンを押すと終了
start
1634044423
{'message': 'Created item with id: 4c1c5fd5-e6b9-4bfc-b73b-11bcd2ae616d'}
button 2 OFF
# Ctrl + C で終了
これでボタンを押すたびにLEDが点灯してLINE Notify経由で通知が届くようになりました。
念の為、実際にログがDynamoDBに格納されているか確認してみます。
$ aws dynamodb scan --table-name GameCounter
{
"Items": [
{
"date": {
"S": "2021-10-12"
},
"flag": {
"S": "start"
},
"timestamp": {
"N": "1634044565"
},
"time": {
"S": "22:16:05"
},
"username": {
"S": "Lucy"
},
"uid": {
"S": "4c1c5fd5-e6b9-4bfc-b73b-11bcd2ae616d"
}
},
...
ちゃんと入っていそう。
システムサービスへの登録
もろもろ動作を確認できたところでRaspberry Pi起動時にスクリプトを自動起動できるようsystemd
にサービスとして登録します。
counter.service
[Unit]
Description=GameCounterService
After=syslog.target
[Service]
Type=simple
# app.pyの格納先ディレクトリを指定 /home/pi/scripts/game-counter/app.pyに配置とした場合
WorkingDirectory=/home/pi/scripts/game-counter
ExecStart=/usr/bin/python3 /home/pi/scripts/game-counter/app.py
TimeoutStopSec=5
StandardOutput=null
[Install]
WantedBy = multi-user.target
systemd
へ登録するために作成したcounter.service
を/etc/systemd/system
以下に格納してサービスの起動終了をテストしてみます。自動起動を設定することでRaspberry Piを起動するたびに自動的に動作するようになりました。
$ sudo cp counter.service /etc/systemd/system/
$ sudo systemctl daemon-reload
# 登録したサービスを起動
$ sudo systemctl start counter
# 念の為プロセスが動いているか確認
$ ps ax | grep app.py
263 ? Rs 00:55 /usr/bin/python3 /home/pi/scripts/game-counter/app.py
# 登録したサービスを終了
$ sudo systemctl stop counter
# 起動、終了に問題なければ自動起動を設定
$ sudo systemctl enable counter
フロントエンド構築
長丁場になってきましたが、最後にフロントエンドの構築。
今回はAPIとのやり取りにRedux ToolkitのRTK Query
が便利(と個人的に推しているだけ)なのでcreate-react-appのreduxのテンプレートを使って構築します。
なお、UIにはChakra UI
を、日時の計算用にmoment
、ページ遷移のためにReactRouter
を利用しました。
$ yarn create react-app frontend --template redux-typescript
$ cd frontend
$ yarn add @chakra-ui/react @emotion/react@^11 @emotion/styled@^11 framer-motion@^4
$ yarn add moment
$ yarn add react-router-dom
$ yarn add -D @types/react-router-dom
また、API情報については環境変数ファイルを用意してAPI GatewayのURLとAPI Keyを記載しておきます。
frontend/.env.local
REACT_APP_API_URL=https://********.execute-api.ap-northeast-1.amazonaws.com/prod/counter
REACT_APP_API_KEY=********
ページ構成
今回はホーム画面と詳細画面を実装していきます。
- ホーム画面…LucyとMikeの直近7日間の合計時間一覧を表示
- ユーザー別詳細画面…ユーザー別の当日のゲーム時間のログ一覧を表示
フロントエンド実装
Reduxのテンプレートを使用しているので追加、変更したファイルは以下のようになりました。(詳細のソースはGitHubより参照ください)
frontend/
+── .env.local
├── README.md
├── package.json
├── public
├── src
│ ├── App.css
│ ├── App.test.tsx
+ ├── App.tsx
│ ├── app
│ │ ├── hooks.ts
+ │ └── store.ts
+ ├── components
+ │ └── Header.tsx
│ ├── features
│ │ └── counter
│ ├── index.css
+ ├── index.tsx
│ ├── logo.svg
+ ├── pages
+ │ ├── Home.tsx
+ │ ├── Page404.tsx
+ │ └── UserPage.tsx
│ ├── react-app-env.d.ts
│ ├── serviceWorker.ts
+ ├── services
+ │ └── game-counter.ts
│ ├── setupTests.ts
+ └── types
+ └── index.ts
├── tsconfig.json
└── yarn.lock
RTK Queryの紹介
少し脇道に逸れますが、フロントエンド側でAPIからデータを取得するためにRedux ToolkitのRTK Queryを利用しています。RTK QueryはRedux Toolkit v1.5から導入された新機能でswr
やreact-query
と同じようなキャッシュ戦略を駆使したRedux Toolkitチーム謹製のQueryツールとなっています。
下記は統計ログ一覧の取得、ユーザーごとのログの取得のRTK Query部分になりますが、この記述のみで取得やエラーハンドリングなど複数の機能をもったカスタムフックを生成してくれます。
frontend/src/services/game-counter.ts
import { createApi, fetchBaseQuery } from '@reduxjs/toolkit/query/react';
import { GameCounterItem, GameCounterLogItem, GetQueryParams, LogQueryParams } from '../types';
// .env.localファイルからAPI URLとAPI KEYを取得
const API_URL = process.env.REACT_APP_API_URL!;
const API_KEY = process.env.REACT_APP_API_KEY!;
// RTK Queryを利用したAPIを作成
export const gameCounterApi = createApi({
reducerPath: 'gameCounterApi',
baseQuery: fetchBaseQuery({
baseUrl: API_URL,
// headerにAPI KEYを付与
prepareHeaders: (headers) => {
headers.set('x-api-key', API_KEY);
return headers;
},
}),
endpoints: (builder) => ({
// ユーザー詳細画面でユーザーごとの情報を取得するためのクエリー
// ユーザ名と期間(UNIXTIME)を送る
getCounterByName: builder.query<GameCounterItem[], GetQueryParams>({
query: (params) =>
`?username=${params.username}&starttime=${params.startTime}&endtime=${params.endTime}`,
}),
// ホーム画面でユーザーごとの統計情報を取得するためのクエリー
// ユーザ名と取得日(YYYY-MM-DD)を送る
getLogDataByName: builder.query<GameCounterLogItem[], LogQueryParams>({
query: (params) => `log?username=${params.username}&date=${params.date}`,
}),
}),
});
// RTK Queryが自動的に用意してくれる!カスタムフックをエクスポート
export const { useGetCounterByNameQuery, useGetLogDataByNameQuery } = gameCounterApi;
コンポーネント側からは、エクスポートされたuseGetCounterByNameQuery
というカスタムフックを受け取ることで、下記複数のプロパティを返してくれます。
property | |
---|---|
data | 結果を返す |
error | エラー結果を返す |
isUninitialized | 実行前にtrueを返す |
isLoading | 最初の読み込み中にtrueを返す |
isFetching | 最初だけでなく、以後再読み込み中のときもtrueを返す |
isSuccess | リクエストが成功時にtrueを返す |
isError | エラー時にtrueを返す |
refetch | キャッシュを使わずに再読込する関数 |
主に利用するのは、data
isLoading
isFetching
が多そうです。
frontend/src/pages/UserPage.tsx
...
const { data, isLoading, isFetching } = useGetCounterByNameQuery(params, {
// 30000秒ごとに自動更新とかもできます
pollingInterval: 30000,
});
...
// 最初の読み込み時はLoadingを表示
if (isLoading) return <Box p={4}>Loading...</Box>;
// データが取得できなかった場合にエラーを表示
if (!data) return <Box p={4}>Missing data!</Box>;
// 万事うまくいった場合にコンテンツを表示
return (
// 再読み込み中のときにSpinnerなどローディング状態を表示
{isFetching && (
<Box alignItems="center" justifyContent="center" textAlign="center">
<Spinner thickness="4px" speed="0.65s" emptyColor="gray.200" color="blue.500" size="xl" />
</Box>
)}
...コンテンツを表示
)
...
RTK Queryを利用することで機能の役割をシンプルにわけることができるのと、Redux DevToolsでデバッグできる点など開発のメリットも多数あるかと思います。swr
やreact-query
とはまた別の選択肢のひとつとしてぜひ試してみてください。
少し古いのですが以前RTK Queryのさわりを紹介した記事になりますのでよろしければこちらも参照ください。
フロントエンドのデプロイ
今回はフロントエンドのデプロイ先としてS3を利用することにしました。CDKを利用してS3バケットの作成とS3バケットへのデプロイを追記します。
game_counter_stack.py
from aws_cdk import (
Stack,
RemovalPolicy,
Duration,
CfnOutput,
aws_dynamodb as dynamodb,
aws_lambda as lambda_,
aws_iam as iam,
aws_apigateway as apigateway,
+ aws_s3 as s3,
+ aws_s3_deployment as s3deploy,
aws_events as events,
aws_events_targets as targets
)
from constructs import Construct
game_counter_stack.py
class GameCounterStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs
....
# S3 Bucket フロントエンド公開用
bucket = s3.Bucket(
self, "GameCounterBucket",
website_index_document="index.html",
website_error_document="index.html",
public_read_access=True,
removal_policy=RemovalPolicy.DESTROY
)
# S3 BucketにReactからビルドしたファイルをアップロード
s3deploy.BucketDeployment(
self, "GameCounterBucketDeployment",
destination_bucket=bucket,
sources=[s3deploy.Source.asset("frontend/build")],
retain_on_delete=False
)
# S3 Bucketの公開URLを出力
CfnOutput(
self, "GameCounterS3PublickUrl",
value=bucket.bucket_website_url,
export_name="GameCounterS3PublicUrl"
)
S3へはReactのbuildディレクトリをアップロードするため、CDKをデプロイする前にReactアプリケーションをビルドしておきます。
$ cd frontend/
$ yarn build
$ cd ../
$ cdk synth
$ cdk deploy
...
GameCounterStack.GameCounterS3PublickUrl = http://gamecounterstack-gamecounterbucket*********.s3-website-ap-northeast-1.amazonaws.com
デプロイ終了時に出力されたS3バケットの公開URLへアクセスすることで無事にアプリケーションのホーム画面が表示されました。
おかたづけ
リソースが不要になった場合に備えてリソースの削除も試しておきます。※S3バケット、DynamoDBのRemovalPolicyをDESTROYとしているためすべてのデータが消えるので注意。
$ cdk destroy
SSMへ登録したLINEトークンは下記で削除。
$ aws ssm delete-parameter --name "/GameCounter/LineToken"
さいごに
かなりの長文になってしまいましたが、ここまで読んでいただきありがとうございました。
今回はRaspberry PiとAWSでよく使われる技術をざっくりと組み合わせてこどもたちを巻き込めるアプリケーションをつくってみました。週末に家庭内でのネタから突然つくりあげたところ、子どもたちにも受けがよくそのまま採用され絶賛運用稼働中です。
また、最近のデジタルネイティブな子どもたちはPythonということばを知っていたようでRaspberry Piをつかった電子工作やプログラミングにも興味をもってくれたことが新たな発見でした。
Raspberry PiやAWSを使ってみたいけど何をしようか、という方にも何かをつくるきっかけになれば幸いです。
完成版のソースについてはGitHubに公開しています。ハードコーディングな部分を最適化したり、グラフ表示を入れたりと拡張すべきところはまだまだたくさんありそうです。(手抜きな部分は徐々にアップデートしようと思っています。。。)
ん、話と違って、なんか子どもたちのゲーム時間長すぎ?!