はじめに
前回はエッジデバイス側のデータがクラウドに送信されるまでを確認した。
今回はクラウド側の構成を拡張し、エッジデバイス(Raspberry Pi)のイベントを検出して任意のアクション実行につなげるAWS IoT Eventsとの連携を試してみる。
目指すゴール
エッジデバイス側では、擬似的な「ステータス」を表現するための方法として、2つのスイッチと1つのLEDを接続した回路を使用する。「ステータス」は2つのスイッチのon/offの組み合わせで決まり、MQTTでクラウドに送信される。
「ステータス」とスイッチの状態、クラウド送信値の関係は以下となる。
Switch 2 の状態 | Switch 1 の状態 | クラウドに送信するステータス |
---|---|---|
off | off | 0 |
off | on | 1 |
on | off | 2 |
on | on | 3 |
クラウド側で用意したAWS IoT Eventsのステートマシン図は以下となる。
フローは以下の2種類しかないものとする。
- 0(システム起動中) -> 1(処理中) -> 2(処理完了) -> 0(システム起動中)...
- 0(システム起動中) -> 1(処理中) -> 3(エラー発生) -> 0(システム起動中)...
エラー発生時にはAmazon SNSを使った通知メールを送信する。
エッジデバイス側のステータス遷移はボタンのon/offで疑似的に表現しているため、上記の順番でステータスが変わるようにボタンを押す必要がある。
ただし、エッジデバイスに以下のような異常ステータス遷移が発生した場合、IoT Events がどのような挙動をするのかも確認してみる。
- 0(システム起動中) -> 2(処理完了)
構築手順
- Raspberry Piにスイッチ回路を組む。
- Raspberry PiにAWS IoT Greengrass V2をインストールする。
- AWS IoT Coreにステータス値が送信されていることを確認する。
- Raspberry PiでGreengrassコンポーネントを実装し、デプロイする。
- AWS IoT Eventsでステートマシンを構築する。
- クラウド側でエッジ側のデバイスのステータス遷移と、エラー時にはメール通知が送信されることを確認する。
ソフトウェアVer情報
- Raspberry Pi 4
- Raspbian Ver.11
- Python 3.9.2
- pip 20.3.4
- Java 11.0.13
- aws cli 1.22.23
Raspberry Piのセットアップ
前回を参照。
Raspberry Piにスイッチ回路を組む
前回の回路にスイッチを1つ加えた構成となっている。
この回路を動かすためのPythonスクリプトは以下となる。
switch_state.py
import time
import RPi.GPIO as GPIO
BIT_10_PLACE_PIN = 23
BIT_1_PLACE_PIN = 24
LED_PIN = 25
GPIO.setmode(GPIO.BCM)
GPIO.setup(BIT_10_PLACE_PIN, GPIO.IN)
GPIO.setup(BIT_1_PLACE_PIN, GPIO.IN)
GPIO.setup(LED_PIN, GPIO.OUT)
def convert_bin_2_int(pin2_status: int, pin1_status: int) -> int:
return int("%s%s" % (pin2_status, pin1_status), 2)
try:
while True:
GPIO.output(LED_PIN, GPIO.HIGH)
time.sleep(0.1)
GPIO.output(LED_PIN, GPIO.LOW)
switch_status = convert_bin_2_int(GPIO.input(BIT_10_PLACE_PIN), GPIO.input(BIT_1_PLACE_PIN))
print(f"switch_status: {switch_status}")
time.sleep(2.9)
except KeyboardInterrupt:
pass
GPIO.cleanup()
Raspberry Piでターミナルを開き、switch_state.py スクリプトを設置する。
スクリプトを実行すると、回路が動き始める。
$ python switch_state.py
この回路を簡単に説明する。
- スイッチ1を押すとGPIO 24ピンが電気信号の入力を検知し、
GPIO.input(BIT_1_PLACE_PIN)
にインプットされる電圧がHIGH(1)となる。 - スイッチ1を離すとGPIO 24ピンが電気信号のoffを検知し、
GPIO.input(BIT_1_PLACE_PIN)
にインプットされる電圧がLOW(0)となる。 - スイッチ2を押すとGPIO 23ピンが電気信号の入力を検知し、
GPIO.input(BIT_10_PLACE_PIN)
にインプットされる電圧がHIGH(1)となる。 - スイッチ2を離すとGPIO 23ピンが電気信号のoffを検知し、
GPIO.input(BIT_10_PLACE_PIN)
にインプットされる電圧がLOW(0)となる。 - LEDは3秒間隔で点灯する。
- LEDの点灯の瞬間に2つのスイッチのステータス(0 or 1)を読み取り、擬似的なデバイスステータス値(0〜3)に変換し、クラウドにMQTTで送信する。
AWS IoT Greengrass V2をインストール
ここでは、エッジデバイス側のソフトウェアの設定を行う。
Greengrassのインストールは前回を参照。
エラー時の通知用SNSを登録する
エッジ側でエラー(擬似的なステータス:3)の発生をクラウド側のステートマシンで検知した際の、メール通知用SNSトピックとサブスクリプションを登録する。
サブスクリプション登録時、入力したメールアドレスに確認メールが送信される。わすれずに確認リンクをクリックしておく。
Greengrassコンポーネントを実装し、デプロイする
前回同様、Greengrass V2のコンポーネントを実装する。
アーティファクト
以下にアーティファクトのスクリプトを設置する。
$ mkdir -p /home/pi/components/artifacts/com.example.PublishSwitchState/1.0.0/
$ touch /home/pi/components/artifacts/com.example.PublishSwitchState/1.0.0/publish_switch_state.py
publish_switch_state.py
import RPi.GPIO as GPIO
import time
import datetime
import json
import awsiot.greengrasscoreipc
from awsiot.greengrasscoreipc.model import (
QOS,
PublishToIoTCoreRequest
)
BIT_10_PLACE_PIN = 23
BIT_1_PLACE_PIN = 24
LED_PIN = 25
GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)
GPIO.setup(BIT_10_PLACE_PIN, GPIO.IN)
GPIO.setup(BIT_1_PLACE_PIN, GPIO.IN)
GPIO.setup(LED_PIN, GPIO.OUT)
DEVICE_ID = "RaspberryPi4"
SLEEP_TIME = 4
topic = "raspberrypi4/switches"
qos = QOS.AT_LEAST_ONCE
ipc_client = awsiot.greengrasscoreipc.connect()
def convert_bin_2_int(pin2_status: int, pin1_status: int) -> int:
return int("%s%s" % (pin2_status, pin1_status), 2)
while True:
GPIO.output(LED_PIN, GPIO.HIGH)
time.sleep(1)
GPIO.output(LED_PIN, GPIO.LOW)
datetime_now = str(datetime.datetime.now())
switch_status = convert_bin_2_int(GPIO.input(BIT_10_PLACE_PIN), GPIO.input(BIT_1_PLACE_PIN))
message = {
"device_id": DEVICE_ID,
"switch_status": switch_status,
"timestamp": datetime_now
}
message_json = json.dumps(message).encode('utf-8')
request = PublishToIoTCoreRequest()
request.topic_name = topic
request.payload = message_json
request.qos = qos
operation = ipc_client.new_publish_to_iot_core()
operation.activate(request)
future = operation.get_response()
future.result(10)
with open('/tmp/Greengrass_PublishSwitchState.log', 'a') as f:
print(f"{datetime_now} / switch_status: {switch_status}.", file=f)
time.sleep(SLEEP_TIME)
GPIO.cleanup()
このコードは、switch_status.py
を拡張したものとなっている。
レシピ
以下にアーティファクトのスクリプトを設置する。
$ touch /home/pi/components/recipes/com.example.PublishSwitchState-1.0.0.yaml
com.example.PublishSwitchState-1.0.0.yaml
---
RecipeFormatVersion: 2020-01-25
ComponentName: com.example.PublishSwitchState
ComponentVersion: '2.0.0'
ComponentDescription: A component that publishes messages.
ComponentPublisher: Amazon
ComponentConfiguration:
DefaultConfiguration:
accessControl:
aws.greengrass.ipc.mqttproxy:
'com.example.PublishSwitchState:mqttproxy:1':
operations:
- "aws.greengrass#PublishToIoTCore"
resources:
- "raspberrypi4/switches"
Manifests:
- Platform:
os: linux
Lifecycle:
Run: python3 {artifacts:path}/publish_switch_state.py
レシピでは、accessControl にてIoT CoreへのPublishを許可する設定が必要となる。
resources では、MQTTメッセージ送信先となる AWS IoT Core のトピック名を記載する。
アーティファクトとレシピが以下のように設置されていることを確認する。
/home/pi/components
├── artifacts
│ └── com.example.PublishSwitchState
│ └── 1.0.0
│ └── publish_switch_state.py
└── recipes
└── com.example.PublishSwitchState-1.0.0.yaml
動作確認
デプロイ
では、Greengrass CLI でデプロイを実行してみよう。
$ sudo /greengrass/v2/bin/greengrass-cli deployment create \
--recipeDir /home/pi/components/recipes \
--artifactDir /home/pi/components/artifacts \
--merge "com.example.PublishSwitchState=1.0.0"
スクリプトで出力しているログを確認。
$ tail -f /tmp/Greengrass_PublishSwitchState.log
エッジデバイスの疑似ステータスがクラウドに送信されていることを確認してみる。
LEDが点灯している瞬間のスイッチ2とスイッチ1のon/offのステータスが、AWS IoT Core の MQTTテストクライアントで受信できていることを確認する。
AWS IoT Eventsでステートマシン(状態遷移)を構築する
エッジ側のデータをリアルタイムにクラウド側で確認するという意味では、前回とやっていることは同じである。前置きが長くなったが、ここからが本稿のメインである。
AWS IoT Events で構築したステートマシンでエッジ側をモニタリングし、エラーが発生した場合にメール送信されるかどうかを確認してみる。
本稿は、AWS IoT Events初級ハンズオン / 4. AWS IoT Eventsの設定の内容を参考にさせて頂いた。
IoT Eventの探知機モデルの作成
構築するAWS IoT Events ステートマシン
このステートマシンで発生しうる遷移パターンについて確認する。
1. 遷移パターン: 0
システムが起動する直前の状態から、起動中へのステータス遷移する。
2. 遷移パターン: 0 -> 1
{
"device_id": "RaspberryPi4",
"switch_status": 1
}
入力値を受けて、ステータスは0から1に遷移する。
3. 遷移パターン: 1 -> 2
システムが処理中から処理完了に遷移する。
デバイスからの入力
{
"device_id": "RaspberryPi4",
"switch_status": 2
}
入力値を受けて、ステータスは1から2に遷移する。
4. 遷移パターン: 2 -> 0
システムが処理完了から起動中に遷移する。
デバイスからの入力
{
"device_id": "RaspberryPi4",
"switch_status": 0
}
入力値を受けて、ステータスは2から0に遷移する。
5. 遷移パターン: 1 -> 3
システムが処理中からエラー発生に遷移する。
デバイスからの入力
{
"device_id": "RaspberryPi4",
"switch_status": 3
}
入力値を受けて、ステータスは1から3に遷移する。
6. 遷移パターン: 3 -> 0
システムがエラー発生から起動中に遷移する。
デバイスからの入力
{
"device_id": "RaspberryPi4",
"switch_status": 0
}
入力値を受けて、ステータスは3から0に遷移する。
遷移パターンをリストアップしてみたが、この中には 0(システム起動中)
-> 2(処理完了)
のような異常ステータス遷移は含まれていない。異常ステータス遷移が発生した場合は、発生時のステータスで設定されているイベントトリガー(後述)の条件に合致しない限りはそのステータスに留まり続ける。
以降は、上記の遷移パターンを実現するためのAWS IoT Eventsのステートマシンを構築していく。
入力の作成
まず、AWS IoT Eventsのステートマシンが受け入れる入力データのフォーマット設定を登録していく。
以下を記載したファイルを作成し、raspberrypi4_switch_status_event_input.json
という名前でローカルに保存する。
{
"device_id": "",
"switch_status": ""
}
その後、入力フォームの「JSONファイルのアップロード」にて上記ファイルを指定してアップロードする。
作成ボタンで入力情報の登録が完了する。
探知機を作成
探知機とはステートマシンを指す。
ステートマシンには主に2つの要素が存在する。「状態」は円、「移行シーケンス」は矢印で表現されている。
「状態」には個別に以下のタイミング評価される複数のイベントを設定可能である。
評価タイミング | |
---|---|
OnEnter | 該当の「状態」にステータスが遷移したタイミングで評価されるイベントを設定。 |
OnInput | ステートマシンに入力が発生した時に評価されるイベントを設定。 |
OnExit | 該当の「状態」から別の「状態」にステータスが遷移したタイミングで評価されるイベントを設定。 |
「移行シーケンス」には個別にイベントを設定可能である。
では、具体的な設定内容を見ていこう。
状態: 0-running
項目 | 設定値 |
---|---|
状態名 | 0-running |
状態: 1-inprocess
項目 | 設定値 |
---|---|
状態名 | 1-inprocess |
状態: 2-processed
項目 | 設定値 |
---|---|
状態名 | 2-processed |
状態: 3-error
項目 | 設定値 |
---|---|
状態名 | 3-error |
OnEnter イベント1
項目 | 設定値 |
---|---|
イベント名 | init |
イベントの条件 | $input.raspberrypi4_switch_status_event.switch_status == 3 |
イベントアクション | SNSメッセージの送信 SNSトピック: arn:aws:sns:ap-northeast-1:************:トピック名 デフォルトペイロード |
移行シーケンス: to-1: 0->1
項目 | 設定値 |
---|---|
イベント名 | to-1 |
最初の状態 | 0-running |
目的の状態 | 1-inprocess |
イベントのトリガーロジック | $input.raspberrypi4_switch_status_event.switch_status == 1 |
イベントアクション
項目 | 設定値 |
---|---|
変数の設定 | |
変数オペレーション | 値の割当 |
変数名 | switch_status |
値の割り当て | 1 |
移行シーケンス: to-2: 1->2
項目 | 設定値 |
---|---|
イベント名 | to-2 |
最初の状態 | 1-inprocess |
目的の状態 | 2-processed |
イベントのトリガーロジック | $input.raspberrypi4_switch_status_event.switch_status == 2 |
イベントアクション
項目 | 設定値 |
---|---|
変数の設定 | |
変数オペレーション | 値の割当 |
変数名 | switch_status |
値の割り当て | 2 |
移行シーケンス: to-3: 1->3
項目 | 設定値 |
---|---|
イベント名 | to-3 |
最初の状態 | 1-inprocess |
目的の状態 | 3-error |
イベントのトリガーロジック | $input.raspberrypi4_switch_status_event.switch_status == 3 |
イベントアクション
項目 | 設定値 |
---|---|
変数の設定 | |
変数オペレーション | 値の割当 |
変数名 | switch_status |
値の割り当て | 3 |
移行シーケンス: to-0: 2->0
項目 | 設定値 |
---|---|
イベント名 | to-0 |
最初の状態 | 2-processed |
目的の状態 | 0-running |
イベントのトリガーロジック | $input.raspberrypi4_switch_status_event.switch_status == 0 |
イベントアクション
項目 | 設定値 |
---|---|
変数の設定 | |
変数オペレーション | 値の割当 |
変数名 | switch_status |
値の割り当て | 0 |
移行シーケンス: to-0: 3->0
項目 | 設定値 |
---|---|
イベント名 | to-0 |
最初の状態 | 3-error |
目的の状態 | 0-running |
イベントのトリガーロジック | $input.raspberrypi4_switch_status_event.switch_status == 0 |
イベントアクション
項目 | 設定値 |
---|---|
変数の設定 | |
変数オペレーション | 値の割当 |
変数名 | switch_status |
値の割り当て | 0 |
探知機の発行
要素を設定したら、探知機を発行する。
探知機の処理実行ロールを選択。
探知機生成メソッドは「一意のキー値ごとに探知機を作成する」を選択。
探知機作成キー「入力」で設定しておいた device_id
を指定する。これは、この探知機を生成・分類するための区分けとなるパラメータ項目である。探知機は「入力」の device_id
の値の種類ごとにステートマシンを作成する。
IAM Roleの修正
探知機の発行が終わったら、上記で指定したIAMロールがSNSで通知メールを送信できるように権限を追加しておこう。
IoT Core のルールエンジン設定
エッジからクラウドにデータを受信する入り口となるAWS IoT Core からは、AWSの様々なサービスにデータを渡すことができる。データを渡すための条件やルールの設定は、AWS IoT Core > ACT > ルール
で設定可能である。
ルールクエリステートメントでは、IoT Core で受信したデータから必要なデータだけを抽出するためのクエリを入力する。
以下は、IoT Core のトピック raspberrypi4/switches
でサブスクライブしたデータから、device_id
と switch_status
だけを送信指定したAWSサービスに送るクエリである。
SELECT device_id, switch_status FROM 'raspberrypi4/switches'
アクションには、「IoT Events 入力にメッセージを送信する」を選択。
IoT Events で登録した「入力」と、IoT Eventsにアクセス可能なIAMロールを作成指定して登録する。
動作確認
これで準備はすべて完了である。
最後にもう一度、エッジデバイスでGreengrassコンポーネントをデプロイしよう。
$ sudo /greengrass/v2/bin/greengrass-cli deployment create \
--recipeDir /home/pi/components/recipes \
--artifactDir /home/pi/components/artifacts \
--merge "com.example.PublishSwitchState=1.0.0"
最初のデプロイ時と同様に IoT Core の MQTTテストクライアントでデータが受信できていることを確認後、IoT Events の探知機モデルから、エッジデバイスのステータスを確認してみよう。
以下を順番に実行する。
- エッジデバイス(Raspberry Pi)で、なにもスイッチを押していない状態だと、「現在の状態」は
0-running
となる。 - エッジデバイス(Raspberry Pi)で、スイッチ1のみを押す。「現在の状態」は
1-inprocess
となる。 - エッジデバイス(Raspberry Pi)で、スイッチ2のみを押す。「現在の状態」は
2-processed
となる。 - エッジデバイス(Raspberry Pi)で、なにもスイッチを押していない状態だと、「現在の状態」は
0-running
となる。 - エッジデバイス(Raspberry Pi)で、スイッチ1のみを押す。「現在の状態」は
1-inprocess
となる。 - エッジデバイス(Raspberry Pi)で、スイッチ1とスイッチ2を押す。「現在の状態」は
3-error
となる。この時、状態3のOnEnterイベントの発火条件を満たしたため、通知メールが送信されることを確認しよう。通知メール確認
- エッジデバイス(Raspberry Pi)で、なにもスイッチを押していない状態だと、「現在の状態」は
0-running
となる。
最後に
Greengrassコンポーネントのコードについては以下で共有していますので、Star押してもらえたら嬉しいです!