IoTのためのクラウドシステム構築⑤ | 作業安全はAIの様々な情報を発信するAIMOS | スタッフブログ

サブスクリプションの作成

2.Lambda関数の作成

3.IoTCoreのルールを設定する

1. 左のメニューを開いて”メッセージのルーティング”＞”ルール”を押下
2. ”ルールを作成”ボタンを押下
3. ”ルール名”を適当に入力し、”次へ”を押下
4. ”SQLステートメント”に「SELECT * FROM ’トピック名’」を記入
5. ”ルールアクション”で”Lambda”を選択
6. 関数名リストから自分が作成したLambda関数を選択し、”次へ”を押下
7. “ルールの作成”ボタンを押下

 

Lambda関数プログラム実装

■参考コード

import boto3
import os
from botocore.exceptions import ClientError

TOPIC_ARN = u'arn:aws:sns:us-west-2:261265052204:IoT_train_topic'#自分のSNSTopicを代入する
sns = boto3.client('sns')
iot = boto3.client('iot-data')
temp_threshold =29

# センサデータがしきい値を超え続けた時の、メール連投防止トリガー
# Lambdaコンソール画面の設定から環境変数でトリガーを設定しておく。Alert_Triggerはstring
alert_trigger = os.environ["Alert_Trigger"]

# メールを設定し、送信する
def send_mail(arn,temp,time):
# 本文
msg = (u'室温が'+str(temp_threshold)+u'℃を超えています。\n'+str(time) +'現在の室温は'+str(temp)+u'℃です。')
# 件名
subject = u'暑い！！！'

request = {
'TopicArn': arn,
'Message': msg,
'Subject': subject
}
sns.publish(**request)

def lambda_handler(event, context):
global alert_trigger
print(alert_trigger)
print(event)
sensor = event['sensor']
temp = sensor['temp']
time_stamp = event['timestamp']

if temp >= temp_threshold:
# トリガーがTrueの場合のみ、メールを送る
if alert_trigger == "True":
try:
send_mail(TOPIC_ARN,temp,time_stamp)
alert_trigger = "False"
os.environ["Alert_Trigger"] = alert_trigger
except ClientError as e:
print(e.response['Error']['Message'])
# しきい値を下回った時、トリガーを戻す
else:
alert_trigger = "True"
os.environ["Alert_Trigger"] = alert_trigger

5.センサデータをAWSに送るプログラムを作成し、実行する

■ 実行プログラムの作成

#coding: utf-8

from bme280_get_data import BME280
import clientConfig as client
import time
from awsiot import mqtt5_client_builder
from awscrt import mqtt5, http
from concurrent.futures import Future

sensor = BME280()

# MQTT5 クライアントの設定
TIMEOUT = 100
clientName = client.clientName
endpoint_path = client.endpoint_path
amazonroot_path = client.amazonroot_path
privatekey_path = client.privatekey_path
certificate_path = client.certificate_path

# サブスクライブしたメッセージを受信したときのコールバック
def on_publish_received(publish_packet_data):
publish_packet = publish_packet_data.publish_packet
assert isinstance(publish_packet, mqtt5.PublishPacket)
print("Received message from topic '{}': {}".format(publish_packet.topic, publish_packet.payload))

# ライフサイクルイベント: 停止
def on_lifecycle_stopped(lifecycle_stopped_data: mqtt5.LifecycleStoppedData):
print("Lifecycle Stopped")

# ライフサイクルイベント: 接続成功
def on_lifecycle_connection_success(lifecycle_connect_success_data: mqtt5.LifecycleConnectSuccessData):
print("Lifecycle Connection Success")

# ライフサイクルイベント: 接続失敗
def on_lifecycle_connection_failure(lifecycle_connection_failure: mqtt5.LifecycleConnectFailureData):
print("Lifecycle Connection Failure")
print("Connection failed with exception: {}".format(lifecycle_connection_failure.exception))

#クライアント情報を設定する
def set_mqttclient(name=clientName,endpoint=endpoint_path,root=amazonroot_path,key=privatekey_path,certificate=certificate_path):
mqtt5_client = mqtt5_client_builder.mtls_from_path(
endpoint=endpoint_path,
cert_filepath=certificate_path,
pri_key_filepath=privatekey_path,
ca_filepath=amazonroot_path,
on_publish_received=on_publish_received,
on_lifecycle_stopped=on_lifecycle_stopped,
on_lifecycle_connection_success=on_lifecycle_connection_success,
on_lifecycle_connection_failure=on_lifecycle_connection_failure,
client_id=clientName,
clean_start=True,
keep_alive_secs=30
)
return mqtt5_client

if __name__ == '__main__':
mqtt5_client = set_mqttclient()
try:
print("Connecting to endpoint...")
mqtt5_client.start()

try:
while True:
messageJson = sensor.getMessage()
print(f"Publishing message: {messageJson}")

publish_future = mqtt5_client.publish(mqtt5.PublishPacket(
topic="your topic",
payload=messageJson.encode(),
qos=mqtt5.QoS.AT_LEAST_ONCE
))
publish_completion_data = publish_future.result(TIMEOUT)
print("Publish successful. PUBACK: {}".format(publish_completion_data.puback.reason_code)) #pubbackは0が返れば成功
time.sleep(5)
except Exception as e:
print(f"Publish failed: {e}")
except KeyboardInterrupt:
pass

※ インポートされている”clientConfig”は前々回記事の物と同じです。

このようなメールが届くはずです！！

以上、AWS LambdaとAmazon Simple Notification Serviceを組み合わせて、センサデータに基づくメール通知アプリケーションを作ってみました！！
センサーデータからメールで遠隔地にアラート、最近浸透してきているシステムっぽいですよね！

次回はいよいよ、センサデータを使ってアクチュエータ制御をしていきます！