はじめに こんにちは。TIG/DXユニット所属の宮永です。
本記事は「GoでMQTT!! ~温湿度マイスターbotの作成~(前編) 」 の後半パートです。
前半パートの記事をご覧になっていない方はそちらもご参照ください。
今回はAWSサービスのうちの一つAWS IoTを使用してRaspberryPiとのMQTTによる通信を行います。
AWS IoTを使用したMQTTのチュートリアルはAWS公式からも詳細なハンズオン記事が出ています。
本記事はこちらのハンズオンを基にGo言語を使用してMQTTによる通信を行いました。(公式の記事はPythonで実装されています。)
最終的には室内の快適な温湿度を教えてくれる「温湿度マイスターbot」を作成します。
なお、本記事で作成したコードは
にて公開しています。
前回の振り返り 実装は以下の手順で進めます。 前回の記事では「3. DHT22の温湿度情報をAWS IoTへPublish」 までを行いました。
DHT22から温湿度情報を取得する
AWS IoTを使用してRaspberryPiからのPublish動作確認
DHT22の温湿度情報をAWS IoTへPublish 👈ココまでやりました。
AWS IoTで取得した温湿度情報をDynamoDBに連携
Boto3を使用してDynamoDBからデータをQuery、データ整形
取得データをmatplotlibで可視化
作成したプロット図をSlack APIで画像投稿
本記事では「4. AWS IoTで取得した温湿度情報をDynamoDBに連携」 から取り組みます。
4. AWS IoTで取得した温湿度情報をDynamoDBに連携 DynamoDBについては入門記事などが弊社ブログでも投稿されているため説明は割愛いたします。
AWS IoTで受信したデータをDynamoDBに登録する方法は公式に詳細に記載されています。
公式のドキュメントは非常に丁寧にまとめられているため、本記事では要点のみ記載します。
テーブルの新規作成 まずはテーブルの作成です。テーブル名は任意ですが、今回はmydht22
としました。Partition keyにdevice_id
をSort keyにtimestamp
を定義しています。
ルールの作成 作成したテーブルにデータを送信するため、AWS IoT ルールの作成を行います。
AWS IoTコンソール上Act>Ruleからルールの新規作成が行えます。SQLバージョンの使用は2016-03-23を使用しました。From句には 「2. AWS IoTを使用してRaspberryPiからのPublish動作確認」 の実装にてに定義したトピック名を記載してください。
今回の場合トピック名は「topic/to/publish」です。as句は通常のSQLと同じようにテーブル名となるため適宜定義してください。
1 2 3 4 5 6 SELECT '01' as device_id , Timestamp as timestamp , Temperature as temperature , Humidity as humidity FROM 'topic/to/publish'
アクション追加の際には下図「DynamoDBテーブル(DynamoDBv2)の複数列にメッセージを分割する」 を追加してください。
リソースにテーブルmydht22
選択してください。また、今回はmydht22
という名称でロールを新規作成しました。
疎通確認 最後にDynamoDBにデータが正しく登録されているか確認します。 DynamoDBコンソールにアクセスして、下図の様にデータが登録されていることを確認してください。
作成したルールに従ってデータが登録されていますね! 次の章では、Boto3を使用してDynamoDBからデータを取得したうえで、扱いやすいようにデータを整形します。
5. Boto3を使用してDynamoDBからデータをQuery、データ整形 PythonモジュールBoto3を使用してDynamoDBからデータをQuery、最新値15点ほどを抜き出してプロットします。
Boto3を使用したQueryは非常に簡単に行えますが、Queryで取得したデータはDecimal型を含む特殊な構造をしているため、JSONに整形する必要があります。
今回JSONへの整形にはこちらのモジュールを利用させていただきました。
それでは、以下のスクリプトによって作成したテーブルmydht22 よりデータを取得、JSONファイルとして保存します。
*※本環境ではAWS_PROFILEが設定してあることを前提としています。 AWS_PROFILEの設定方法は以下ををご参照ください。
query.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 from dynamodb_json import json_util as utilfrom boto3.session import Sessionfrom boto3.dynamodb.conditions import Keyimport datetimeimport boto3import jsondef query_table (executed_time ): """ 現在日時より過去7日間のデータをクエリ """ begin = executed_time - datetime.timedelta(days=7 ) response = dynamodb_table.query( KeyConditionExpression=Key('device_id' ).eq('01' ) & Key('timestamp' ).between( begin.isoformat(sep = "T" ,timespec="milliseconds" ), executed_time.isoformat(sep = "T" ,timespec="milliseconds" ))) data = response['Items' ] return data def unmarshall (dynamodb_json ): """ dynamodbで取得したデータを通常のJDONへ整形 """ regular_json = util.loads(dynamodb_json) return regular_json def main (): now = datetime.datetime.now(pytz.timezone('Asia/Tokyo' )) raw = query_table(now) data = unmarshall(raw) with open ('scan_data.json' , mode='wt' , encoding='utf-8' ) as file: json.dump(data, file, ensure_ascii=False , indent=2 ) if __name__ == "__main__" : session = Session(profile_name='default' , region_name='ap-northeast-1' ) dynamodb = session.resource('dynamodb' ) dynamodb_table = dynamodb.Table('mydht22' ) main()
上記スクリプトの実行により出力されたJSONは
query_data.json 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 { "device_id" : "01" , "humidity" : 42.5 , "temperature" : 26.9 , "timestamp" : "2021-09-23T17:50:42.522120086+09:00" } , { "device_id" : "01" , "humidity" : 42.7 , "temperature" : 27 , "timestamp" : "2021-09-23T17:51:45.55949297+09:00" } , { "device_id" : "01" , "humidity" : 42.7 , "temperature" : 26.9 , "timestamp" : "2021-09-23T17:52:46.600977247+09:00" } , ...(省略) ]
扱いやすいJSONファイルを取得することができました!!
それでは、次章で取得したデータをプロットしましょう。
6. 取得データをmatplotlibで可視化 先ほど取得したデータを使用してプロットします。
今回はQueryによって得たデータのうち最新データ15点をプロットするようにしました。ただし、取得したデータをプロットするだけでは面白くないので不快指数を可視化できるようにしました。 天気予報で頻繁に耳にする不快指数ですが、以下論文にて数式化が掲載されていました。
を気温(℃)、 を湿度(%)としたとき、
不 快 指 数
として不快指数を計算することができるそうです。
屋外に関する指標ですが、室内環境の調整には十分そうです。
また、屋内における適正温度は夏場において25 ~ 28℃、適正湿度は40 ~ 70%とのことから、該当領域に色付けをして一目で室内環境を把握できるようにしました。
不快指数などのメッセージは「7. 作成したプロット図をSlack APIで画像投稿」 にてSlackで通知する仕組みとなっています。
plot.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 def calc (temp, humid ): """ 不快指数の計算をする """ return np.round (0.81 * temp + 0.01 * humid * (0.99 * temp - 14.3 ) + 46.3 , 2 ) def fetch_latest_value (Timestamp, Temperature, Humidity ): """ データより最新値15点を取得 """ length = len (Timestamp) if length > 15 : Temperature = Temperature[length - 15 :] Humidity = Humidity[length - 15 :] Timestamp = Timestamp[length - 15 :] return Timestamp, Temperature, Humidity, def converter (json_data ): """ JSONから配列へ変換 """ Humidity = [] Temperature = [] Timestamp = [] temp = 0 humid = 0 for i in range (len (json_data)): temp = json_data[i]['temperature' ] humid = json_data[i]['humidity' ] times = json_data[i]['timestamp' ][5 :16 ] Humidity.append(humid) Temperature.append(temp) Timestamp.append(times) return Timestamp, Temperature, Humidity, def generate_figure (Timestamp, Temperature, Humidity ): """ プロット図の生成 """ plt.rcParams['font.family' ] = 'DejaVu Sans' plt.rcParams['mathtext.fontset' ] = 'stix' plt.rcParams["font.size" ] = 20 plt.rcParams['xtick.labelsize' ] = 10 plt.rcParams['ytick.labelsize' ] = 10 plt.rcParams['figure.figsize' ] = (8 , 6 ) fig = plt.figure() ax1 = fig.add_subplot(111 ) ln1 = ax1.plot( Timestamp, Temperature, marker='o' , markeredgewidth=1. , markeredgecolor='k' , color="orange" , label=r'$Temperature$' ) ax2 = ax1.twinx() ln2 = ax2.plot( Timestamp, Humidity, marker='o' , markeredgewidth=1. , markeredgecolor='k' , color="blue" , label=r'$Humidity$' ) h1, l1 = ax1.get_legend_handles_labels() h2, l2 = ax2.get_legend_handles_labels() ax1.legend(h1 + h2, l1 + l2, loc='upper right' ) ax1.set_ylim([20 , 32 ]) ax2.set_ylim([25 , 85 ]) ax1.axhspan(25 , 28 , color="olive" , alpha=0.3 ) ax2.axhspan(40 , 70 , color="royalblue" , alpha=0.2 ) ax1.set_xlabel(r'$Timestamp$' ) ax1.set_ylabel(r'$Temperature$' ) ax2.set_ylabel(r'$Humidity$' ) ax1.grid(True ) plt.gcf().autofmt_xdate() fig_path = "室内温湿度.jpg" plt.savefig(fig_path) return fig_path def worker (): """ メインとなる関数 """ now = datetime.datetime.now(pytz.timezone('Asia/Tokyo' )) executed_time = now.strftime('%Y-%m-%d %H:%M:%S' ) print ("Executed:" , executed_time) data = query_table(now) json_data = unmarshall(data) Timestamp, Temperature, Humidity = converter(json_data) Timestamp, Temperature, Humidity = fetch_latest_value( Timestamp, Temperature, Humidity) fukai = calc(Temperature[-1 ], Humidity[-1 ]) fukai_message = "不快指数は" + str (fukai) + "です。" fig = generate_figure(Timestamp, Temperature, Humidity)
出力結果
7. 作成したプロット図をSlack APIで画像投稿 それでは、上記で出力した画像をSlackに投稿します。
実装はこちらの準公式記事を参考にしました。
Slackbotの作成などは本記事の主旨ではないため、割愛いたします。
先ほど作成したworker関数に以下を追加します。
plot.py 1 2 3 4 5 6 7 8 9 10 from slack_sdk.web import WebClient...(省略) client = WebClient( token="xoxb-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" ) response1 = client.chat_postMessage(text=" Temp : " + str (Temperature[-1 ]) + "℃, Humid : " + str ( Humidity[-1 ]) + "%, message : " + fukai_message, channel="#home" ) response2 = client.files_upload(channels="#home" , file=fig, title="室内温湿度" )
たったこれだけでSlackに自由に投稿できるとは!便利ですね。
注意点が一つあります。テキストメッセージの送信の際はchannel=”#home”となっていますが、画像を投稿する際はchannels = “#home”です。
出力結果
それでは、最後にSlack投稿を定期実行するようにしましょう。 定期実行には以下のモジュールを使わせていただきました。
2時間ごとにworkerを実行するように設定しました。 最終的なPythonスクリプトはこちらです。
plot.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 from slack_sdk.web import WebClientfrom dynamodb_json import json_util as utilfrom boto3.session import Sessionfrom boto3.dynamodb.conditions import Keyimport matplotlib.pyplot as pltimport numpy as npimport boto3import datetimeimport pytzimport scheduleimport timesession = Session(profile_name='default' , region_name='ap-northeast-1' ) dynamodb = session.resource('dynamodb' ) dynamodb_table = dynamodb.Table('mydht22' ) def query_table (executed_time ): """ 現在日時より過去7日間のデータをクエリ """ begin = executed_time - datetime.timedelta(days=7 ) response = dynamodb_table.query( KeyConditionExpression=Key('device_id' ).eq('01' ) & Key('timestamp' ).between( begin.isoformat(sep = "T" ,timespec="milliseconds" ), executed_time.isoformat(sep = "T" ,timespec="milliseconds" ))) data = response['Items' ] return data def unmarshall (dynamodb_json ): """ dynamodbで取得したデータを通常のJDONへ整形 """ regular_json = util.loads(dynamodb_json) return regular_json def calc (temp, humid ): """ 不快指数の計算をする """ return np.round (0.81 * temp + 0.01 * humid * (0.99 * temp - 14.3 ) + 46.3 , 2 ) def fetch_latest_value (Timestamp, Temperature, Humidity ): """ データより最新値15点を取得 """ length = len (Timestamp) if length > 15 : Temperature = Temperature[length - 15 :] Humidity = Humidity[length - 15 :] Timestamp = Timestamp[length - 15 :] return Timestamp, Temperature, Humidity, def converter (json_data ): """ JSONから配列へ変換 """ Humidity = [] Temperature = [] Timestamp = [] temp = 0 humid = 0 for i in range (len (json_data)): temp = json_data[i]['temperature' ] humid = json_data[i]['humidity' ] times = json_data[i]['timestamp' ][5 :16 ] Humidity.append(humid) Temperature.append(temp) Timestamp.append(times) return Timestamp, Temperature, Humidity, def generate_figure (Timestamp, Temperature, Humidity ): """ プロット図の生成 """ plt.rcParams['font.family' ] = 'DejaVu Sans' plt.rcParams['mathtext.fontset' ] = 'stix' plt.rcParams["font.size" ] = 20 plt.rcParams['xtick.labelsize' ] = 10 plt.rcParams['ytick.labelsize' ] = 10 plt.rcParams['figure.figsize' ] = (8 , 6 ) fig = plt.figure() ax1 = fig.add_subplot(111 ) ln1 = ax1.plot( Timestamp, Temperature, marker='o' , markeredgewidth=1. , markeredgecolor='k' , color="orange" , label=r'$Temperature$' ) ax2 = ax1.twinx() ln2 = ax2.plot( Timestamp, Humidity, marker='o' , markeredgewidth=1. , markeredgecolor='k' , color="blue" , label=r'$Humidity$' ) h1, l1 = ax1.get_legend_handles_labels() h2, l2 = ax2.get_legend_handles_labels() ax1.legend(h1 + h2, l1 + l2, loc='upper right' ) ax1.set_ylim([20 , 32 ]) ax2.set_ylim([25 , 85 ]) ax1.axhspan(25 , 28 , color="olive" , alpha=0.3 ) ax2.axhspan(40 , 70 , color="royalblue" , alpha=0.2 ) ax1.set_xlabel(r'$Timestamp$' ) ax1.set_ylabel(r'$Temperature$' ) ax2.set_ylabel(r'$Humidity$' ) ax1.grid(True ) plt.gcf().autofmt_xdate() fig_path = "室内温湿度.jpg" plt.savefig(fig_path) return fig_path def worker (): """ データを取得し、Slack APIで投稿する """ now = datetime.datetime.now(pytz.timezone('Asia/Tokyo' )) executed_time = now.strftime('%Y-%m-%d %H:%M:%S' ) print ("Executed:" , executed_time) data = query_table(now) json_data = unmarshall(data) Timestamp, Temperature, Humidity = converter(json_data) Timestamp, Temperature, Humidity = fetch_latest_value( Timestamp, Temperature, Humidity) fukai = calc(Temperature[-1 ], Humidity[-1 ]) fukai_message = "不快指数は" + str (fukai) + "です。" fig = generate_figure(Timestamp, Temperature, Humidity) message = "室内温度は" + \ str (Temperature[-1 ]) + "度。湿度は" + str (Humidity[-1 ]) + "%です。" + fukai_message client = WebClient( token="xoxb-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" ) response1 = client.chat_postMessage(text=" Temp : " + str (Temperature[-1 ]) + "℃, Humid : " + str ( Humidity[-1 ]) + "%, message : " + fukai_message, channel="#home" ) response2 = client.files_upload(channels="#home" , file=fig, title="室内温湿度" ) return message if __name__ == "__main__" : session = Session(profile_name='default' , region_name='ap-northeast-1' ) dynamodb = session.resource('dynamodb' ) dynamodb_table = dynamodb.Table('mydht22' ) schedule.every(2 ).hours.do(worker) while True : schedule.run_pending() time.sleep(1 )
まとめ GoとAWS IoTを使用してMQTT通信を行いました。
AWS IoTを使用したのは初めてだったのですが、公式のドキュメントやチュートリアルが非常に丁寧であったため、簡単に実装できました。
AWS以外のクラウドサービスのMQTTサポートなどの使い勝手も気になってきたため、近々まとめられればと思います。
(2021.10.1)続編が公開されました