コネクテッドカー、自動運転、テレマティクス
農業のIoT
これまでの経験値と、体系知識の学びを目指して、春ごろから、IoTの技術資格習得を考えていました。2020年現在、国内には２つのIoT試験が存在します（【】は私の受験予定レベル）
最初、「IoTシステム技術検定」を受験することを考えていたのですが、コロナ禍の影響で今年度の試験が中止となってしまいました。鞍替えというわけではないのですが、停滞しているのもイヤだったので、８月～９月に「IoT検定」を受験する予定です。
勉強は、以下の書籍を基に進めていきます。
以下、勉強で知らなかった・分からなかった箇所をしっかりアウトプットして、記憶に定着させていきたいと思います。
バリューチェーン（価値連鎖）
ブルーオーシャン戦略
アンゾフの成長マトリクス
プロセス・イノベーション
スマート製品のケイパビリティ
キャストした釣り竿の「引き」をセンサーで検知し、一定の敷居値を超えた場合に、LINEにメッセージを通知する内容を実装しましたので、まとめてみました。
基本的な機器の実装と、加速度の取得は、以下の記事を参考にしてください。今回はx,y,z軸への移動距離を「振動のふり幅」とするので、取得した値のうち、Acceloretor値を使用します。 camelrush.hatenablog.com
今回の構成図は以下の通りです。MobilePC上でグラフを表示しつつ、スマホ(SmartPhone)のLINEにAPI通知をPushメッセージを通知します。
LINEのAPIのうち、今回は「Messaging API」を使用します。作成したチャネルのエントリポイント(WebHook)に対して、JSON形式でHTTPリクエストをPOSTすると、あらかじめチャネルを登録しているユーザに対してメッセージを発信することができます（発信先は、Toパラメータで個別指定ができますが、今回はブロードキャスト（全体）発信とします）詳細の仕様は以下に書かれています。
Messaging APIリファレンス | LINE Developers
LINEアカウントがあれば、LINE Developersにログインすることで、フリープランのAPIチャネルを使用することができます。
※アカウントの登録方法、チャネルの作成方法は省略します。
RaspberryPiには、以下のPythonプログラムを配置します。
from devices.accelerometer import AccelerometerMPU6050 as MPU6050 import paho.mqtt.client import ssl import asyncio import json import time import datetime import requests import random # Mqtt Define AWSIoT_ENDPOINT = "[AWS IoT Entry Point]" MQTT_PORT = 8883 MQTT_TOPIC_PUB = "mqttAccelerometer" MQTT_ROOTCA = "./awscert/AmazonRootCA1.pem" MQTT_CERT = "./awscert/xxxxxxxxxxx-certificate.pem.crt" MQTT_PRIKEY = "./awscert/xxxxxxxxxxx-private.pem.key" LINE_THRESHOLD = 5 MSG_LIST = ['ヒットしてない!?', '来てる来てるw', 'Fish on!', 'Gone!'] def mqtt_connect(client, userdata, flags, respons_code): print('mqtt connected.') # Entry Mqtt Subscribe. client.subscribe(MQTT_TOPIC_SUB) print('subscribe topic : ' + MQTT_TOPIC_SUB) def mqtt_message(client, userdata, msg): # Get Received Json Data json_dict = json.loads(msg.payload) # if use ... json_dict['xxx'] def line_post(asum,tmstr): access_token = [LINE API Access Token] headers = {'Authorization': 'Bearer {}'.format(access_token)} uri = "https://api.line.me/v2/bot/message/broadcast" msg = random.choice(MSG_LIST) + "\n" + tmstr + "(" + asum + ")" print(msg) item_data = { "messages":[ { "type":"text", "text":msg } ] } r_post = requests.post(uri, headers=headers, json=item_data) print(r_post) # Publish Loop async def pub_loop(): acmeter = MPU6050() sendtime = time.time() while True: # get Sensor data temp = acmeter.get_temp() ax, ay, az = acmeter.get_accel() gx, gy, gz = acmeter.get_gyro() tm = datetime.datetime.now() asum = abs(ax)+abs(ay)+abs(az) print('ax:{0:> .6f} ,ay:{1:> .6f} ,az:{2:> .6f} ,asum:{3:> .6f}'.format(ax,ay,az,asum)) json_msg = json.dumps({"dt":tm.strftime('%Y-%m-%d %H:%M:%S.%f'),"ax":ax,"ay":ay,"az":az,"asum":asum,"threshold":LINE_THRESHOLD,"gx":gx,"gy":gy,"gz":gz,"temp":temp}) # Line Message Send if (asum > LINE_THRESHOLD): if (time.time() - sendtime) > 1: sendtime = time.time() line_post('G:{0:>.2f}'.format(asum), tm.strftime('%H時').lstrip("0") + tm.strftime('%M分').lstrip("0") + tm.strftime('%S秒').lstrip("0")) # mqtt Publish client.publish(MQTT_TOPIC_PUB ,json_msg) time.sleep(0.01) # Main Procedure if __name__ == '__main__': # Mqtt Client Initialize client = paho.mqtt.client.Client() client.on_connect = mqtt_connect client.on_message = mqtt_message client.tls_set(MQTT_ROOTCA, certfile=MQTT_CERT, keyfile=MQTT_PRIKEY, cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2, ciphers=None) # Connect To Mqtt Broker(aws) client.connect(AWSIoT_ENDPOINT, port=MQTT_PORT, keepalive=60) # Start Mqtt Subscribe client.loop_start() # Start Publish Loop loop = asyncio.get_event_loop() loop.run_until_complete(pub_loop())
加速度センサー「MPU6050」を制御する処理は、以下のとおりとなります。
from abc import ABCMeta import smbus import math from time import sleep # MPU6050 Register Map DEV_ADDR = 0x68 ACCEL_XOUT = 0x3b ACCEL_YOUT = 0x3d ACCEL_ZOUT = 0x3f TEMP_OUT = 0x41 GYRO_XOUT = 0x43 GYRO_YOUT = 0x45 GYRO_ZOUT = 0x47 PWR_MGMT_1 = 0x6b PWR_MGMT_2 = 0x6c class AbstractAccelerometer(metaclass=ABCMeta): def __init__(self ,devadr): self._devadr = devadr self._bus = smbus.SMBus(1) acc_conf = self._bus.read_byte_data(self._devadr ,0x1C) acc_conf = acc_conf | 0x18 self._bus.write_byte_data(self._devadr, 0x1C, acc_conf) def startup(self ,pwr_mgmt_adr): self._bus.write_byte_data(self._devadr, pwr_mgmt_adr, 0) def _read_word(self ,adr): high = self._bus.read_byte_data(self._devadr, adr) low = self._bus.read_byte_data(self._devadr, adr+1) val = (high << 8) + low return val def read_word_sensor(self ,adr): val = self._read_word(adr) if (val >= 0x8000): # minus return -((65535 - val) + 1) else: # plus return val class AccelerometerMPU6050(AbstractAccelerometer): def __init__(self): super().__init__(DEV_ADDR) super().startup(PWR_MGMT_1) def get_temp(self): temp = super().read_word_sensor(TEMP_OUT) x = temp / 340 + 36.53 # data sheet(register map)記載の計算式. return x def get_gyro(self): x = super().read_word_sensor(GYRO_XOUT)/ 131.0 y = super().read_word_sensor(GYRO_YOUT)/ 131.0 z = super().read_word_sensor(GYRO_ZOUT)/ 131.0 return [x, y, z] def get_accel(self): #x = super().read_word_sensor(ACCEL_XOUT)/ 16384.0 #y = super().read_word_sensor(ACCEL_YOUT)/ 16384.0 #z = super().read_word_sensor(ACCEL_ZOUT)/ 16384.0 x = super().read_word_sensor(ACCEL_XOUT)/ 2048.0 y = super().read_word_sensor(ACCEL_YOUT)/ 2048.0 z = super().read_word_sensor(ACCEL_ZOUT)/ 2048.0 return [x, y, z]
以下のとおり、加速度センサーのふり幅（水色）が敷居値（灰色：5.0G）を超えた場合だけ、LINEに通知が送られてきました。
LINE API は、フリープランのうちは、60メッセージ/日、合計でも1.000メッセージが上限であり、それ以上を送ると通知してくれなくなります。センサーの通知メッセージをノーウエイトで送るとあっという間に上限に達し「429 Too Many Requests」を応答するようになりますので、注意してください。
前回、温湿度センサーのデータをOSSのElasticSearchとKibanaを使ってグラフ化しました。
しかし、このツールはグラフの描画間隔の最小単位で1秒となっており、リアルタイムな描画ができませんでした。
加速度センサーなど、プロッタのようなグラフ描画を行う方法を調べたところ、OSSとして、chart.jsの派生プラグインである「chartjs-plugin-streaming」がgitに公開されていました。
https://nagix.github.io/chartjs-plugin-streaming/ja/
上記を基にして、加速度センサーから取得した駆動データを、グラフにプロットすることにします。
今回、計測には以下の慣性センサーを使用します。製品仕様として、3軸加速度＋3軸角速度(ジャイロ)のセンサーを搭載した製品となっています。
慣性センサーとはセンサーの種類を示す大きな分類であり、具体的な計測対象に応じて「加速度」を計測する「加速度センサー」、「角速度」を計測する「ジャイロセンサー」が存在します。
加速度センサーとは、物体の加速度（単位時間当たりの速度の変化量）を計測する装置であり、機器の移動の変化を検知するものです。センサは、その方向に応じて「x」「y」「z」の3軸のベクトルへの移動値を出力します。単位は「」で表されます。一秒間でどの程度の距離を進んだかという表現になります。
ジャイロセンサーは、ジャイロスコープとも呼ばれ、角度（傾き）や角速度（回転の速さ）を測るセンサーです。単位は「」「
」で表されます。一秒間で何度角度が回転したかという表現になります。
構成としては、RaspberryPiで受信したセンサーデータをMQTTによって送信します。MQTTブローカーには、AWSのIoTCoreを使用しています。最終的に表示するWebページで「chartjs-plugin-streaming」を使用してリアルタイムグラフを表示します。
以下のGitHubにプログラムを掲載しています。
github.com
加速度センサーのデバイス制御は「devices/accelerometer.py」に記載しています。データシートをもとに、I2Cによってデータを取得します。
from abc import ABCMeta import smbus import math from time import sleep # MPU6050 Register Map DEV_ADDR = 0x68 ACCEL_XOUT = 0x3b ACCEL_YOUT = 0x3d ACCEL_ZOUT = 0x3f TEMP_OUT = 0x41 GYRO_XOUT = 0x43 GYRO_YOUT = 0x45 GYRO_ZOUT = 0x47 PWR_MGMT_1 = 0x6b PWR_MGMT_2 = 0x6c class AbstractAccelerometer(metaclass=ABCMeta): def __init__(self ,devadr): self._devadr = devadr self._bus = smbus.SMBus(1) def startup(self ,pwr_mgmt_adr): self._bus.write_byte_data(self._devadr, pwr_mgmt_adr, 0) def _read_word(self ,adr): high = self._bus.read_byte_data(self._devadr, adr) low = self._bus.read_byte_data(self._devadr, adr+1) val = (high << 8) + low return val def read_word_sensor(self ,adr): val = self._read_word(adr) if (val >= 0x8000): # minus return -((65535 - val) + 1) else: # plus return val class AccelerometerMPU6050(AbstractAccelerometer): def __init__(self): super().__init__(DEV_ADDR) super().startup(PWR_MGMT_1) def get_temp(self): temp = super().read_word_sensor(TEMP_OUT) x = temp / 340 + 36.53 # data sheet(register map)記載の計算式. return x def get_gyro(self): x = super().read_word_sensor(GYRO_XOUT)/ 131.0 y = super().read_word_sensor(GYRO_YOUT)/ 131.0 z = super().read_word_sensor(GYRO_ZOUT)/ 131.0 return [x, y, z] def get_accel(self): x = super().read_word_sensor(ACCEL_XOUT)/ 16384.0 y = super().read_word_sensor(ACCEL_YOUT)/ 16384.0 z = super().read_word_sensor(ACCEL_ZOUT)/ 16384.0 return [x, y, z]
データシートは以下を参照ください。
http://dlnmh9ip6v2uc.cloudfront.net/datasheets/Components/General%20IC/PS-MPU-6000A.pdf
配線は比較的シンプルです。データ取得にはI2Cバスを使用します。
数10ms程度のタイムラグはありますが、センサーデータのリアルタイムプロットができました。
youtu.be
RaspberryPiを活用する事例として、これまでモーターやカメラを使用したロボットやマニュピレーションを注意に勉強をしてきました。掌握とはいかないまでも、いくつかのプラットフォームやH/W制御を理解し、制御できるようになってきました。
しかし、一般的な産業用IoTとしてはセンターデータ収集（ビッグデータ）と活用も重要では？と思い、今回は温湿度センサー（DTH11）からの受信データの可視化を試してみたいと思います。
受信した温度・湿度データの可視化には、全文検索エンジンである「ElasticSearch」とBIツールの「Kibana」を使用します。このあたりは流行り廃りがありますが、割とメジャーどころを選んでみました。AWSにも同様のサービスがありますが、EC2前提のため費用がかかると思い、一旦オンプレで実践することにしました。
最終的に、表示されたイメージは以下のようになります。tempertureが温度、Humidityが湿度です。線がいびつなのは、途中でセンサーを手でつまんで、無理に温湿度を上げたためです。
事前に準備するハードウェアは、以下の２つです。
ちなみに、私はDHT11の導入で少しハマりまして、最初に購入したHiLetgoさんのセンサーキットでは、うまく動作しませんでした。詳しくは調べていませんが、ハンダの問題による初期不良かもしれません。上のリンクの製品を買いなおしたところ、簡単に接続できました。また（写真のような接続ボードがない）センサー部品だけのものも販売されており、この場合は、接続過程でプルアップ抵抗を付与する必要があるそうです。以下を参考にしてください。
今回の構成は以下の通り、比較的シンプルです。RaspberryPiに接続した、センサーDHT11から温湿度データをPythonで読み取り、ElasticSearchClientライブラリを介してHTTP RequestでElasticSearchのデータベース（index）にデータを格納します。格納されたデータはブラウザ経由でKibanaのポートにアクセスし、画面に表示します。
WindowsPCから、以下のElasticSearchページにアクセスします。
Download Elasticsearch Free | Get Started Now | Elastic | Elastic
遷移したページで、以下のリンクをクリックしてzipファイルをダウンロードします。
zipファイルをダウンロードしたら、適当なフォルダに解凍・展開します（ex. C:\ElasticSearch）
RaspberryPiからのリクエストデータを受信するため、解凍したフォルダにある「/config/elasticsearch.yml」ファイルを開き、以下の箇所を修正します。
# ---------------------------------- Network ----------------------------------- ： #network.host: 192.168.0.1 network.host: 0.0.0.0 ： # --------------------------------- Discovery ---------------------------------- ： #discovery.seed_hosts: ["host1", "host2"] discovery.seed_hosts: ["127.0.0.1"] ：
同様の目的で、ファイアウォールに、ElasticSearchの受信ポート（デフォルトは9200）に対する接続許可を設定します。
ElasticSearchの環境構築は以上です。
PowerShellを起動し、次のコマンドでElasticSearchサービスを開始します。
start (Install Folder)\bin\ElasticSearch.bat
起動にはしばらく時間がかかります。最後に以下のような表示が出ればOKです。
[2020-05-12T19:47:57,614][INFO ][o.e.n.Node ] [DESKTOP-L0AQN2J] started
RaspberryPi側で、以下のコマンドを実行します。
$ sudo curl https://(WindowsPC IPAddress):9200/
次のような結果が表示されれば、OKです。
{
"name" : "DESKTOP-L0AQN2J",
"cluster_name" : "elasticsearch",
"cluster_uuid" : "Yl0C61rVSV6qCa4Q294O6A",
"version" : {
"number" : "7.6.2",
"build_flavor" : "default",
"build_type" : "zip",
"build_hash" : "ef48eb35cf30adf4db14086e8aabd07ef6fb113f",
"build_date" : "2020-03-26T06:34:37.794943Z",
"build_snapshot" : false,
"lucene_version" : "8.4.0",
"minimum_wire_compatibility_version" : "6.8.0",
"minimum_index_compatibility_version" : "6.0.0-beta1"
},
"tagline" : "You Know, for Search"
}
WindowsPCから、以下のKibanaページにアクセスします。
https://www.elastic.co/jp/downloads/kibana
遷移したページで、以下のリンクをクリックしてzipファイルをダウンロードします。
zipファイルをダウンロードしたら、適当なフォルダに解凍・展開します（ex. C:\Kibana）
（任意）外部からKibanaページを参照させたい場合は、解凍したフォルダにある「/config/kibana.yml」ファイルを開き、以下の箇所を修正します。
#server.host: "localhost" server.host: "0.0.0.0"
PowerShellを起動し、次のコマンドでKibanaサービスを開始します。
start (Install Folder)\bin\Kibana.bat
こちらの起動には、ElasticSearchよりも、さらに時間がかかります。最後に以下のような表示が出ればOKです。
log [10:52:22.688] [info][server][Kibana][http] http server running at http://0.0.0.0:5601
WindowsPCで、ブラウザから以下のページへアクセスします。
http://localhost:5601
次のような結果が表示されれば、OKです。
以下の図を参考に配線してください。
今回のプログラムはPythonで開発しています。以下に開発の流れを示しますが、開発したプログラムはGitHubにも上げていますのでご確認ください。
$ sudo pip3 install elasticsearch
$ sudo git clone https://github.com/szazo/DHT11_Python.git
import RPi.GPIO as GPIO from DHT11_Python import dht11 import datetime import time from elasticsearch import Elasticsearch GPIO.setwarnings(False) GPIO.setmode(GPIO.BCM) GPIO.cleanup() # ElasticSearch Client作成（引数はWindowsPCのIPアドレス） es = Elasticsearch('192.168.3.11:9200') # DHT11センサーのインスタンス作成 instance = dht11.DHT11(pin=14) # ポーリング（１秒間隔） while True: # センサーからデータ取得 sensorData = instance.read() timestamp = datetime.datetime.now() if sensorData.is_valid(): es_body = { '@timestamp' : timestamp.isoformat(), 'temperture' : sensorData.temperature, 'humidity' : sensorData.humidity, } # ElasticSearchのデータ(index'raspi_accel_meter')に対して、 # センサーから取得したデータを登録 response = es.index(index='raspi_accel_meter',doc_type='doc',body=es_body) # 1秒ウエイトして次のループへ time.sleep(1) print(es_body)
$ sudo python3 elasticThermoHygroMeter.py
登録されたデータを可視化するため、WindowsPCでKibanaのページにアクセスし、グラフビューを設定します。 最初に、ブラウザから、アドレスhttp://localhost:5601にアクセスします。
左のメニューバーの上から３つ目あたりにある「visualize」を選択します。
「＋Create New Visualization」ボタンをクリックします。
今回は、一番左列にある「Line」アイコンをクリックします。
画面右に「Create Index pattern」が表示されます。ここで受信データ(index)から、データパターンを作成します。
「index pattern」に「raspi*」と入力し、RaspberryPiのPythonプログラムから登録したindex「raspi_accel_meter」が表示されることを確認し、「Next step」ボタンをクリックします。
「Time Filter field name Refresh」には、「I don't want to use Time Filter」を選択して、「Create index pattern」ボタンをクリックします。
以上でindex patternデータが作成されます。再度、左のメニューから「visualize」を選択し、「Line」をクリックしてください。
先ほど作成したindex patternが表示されますので、このリンクをクリックします。
index patternをベースにした、グラフ設定画面が表示されます。
最初に、X軸に「時間」経過を指定していきます。画面左側の「Buckets」の「+Add」をクリックし、「X-axis」を選択します。
追加した「X-axis」には、以下のとおり設定します。
次に、Y軸に「温度」(Temperture）を指定していきます。画面左側の「Metrics」の「Y-axis」に、以下の通り設定します。
ここまでの登録が済んだら、「Metrics」エリアの上にある▶マーク(Apply)をクリックしてください。登録しt温度の内容がグラフ表示されるはずです。
「Y-axis」で「+Add」をクリックして、表示内容に「湿度」(Humadity)を追加します。設定後、▶マークをクリックしてApplyすると、湿度グラフが表示されます。
最初の表示のままでは、累積したindexデータをすべて横軸に表示してしまうため、時間の経過とともに見づらくなっていきます。今回は設定を変更し、直近３分間だけを表示することにします。
上の「now+9h」は、UTCから日本の時差9時間を加算しています。そこから、3m（３分）前までの間（between）を表示するという指定になります。
上記の設定により、グラフのX軸の間隔が３分分になるはずです。なお、グラフの表記を日本時間に変更するには、左のメニューからManagement - Advanced Settings - TimeZone～ を設定する必要があります。
今回はさらに、定期的に画面を更新させましょう。上部の時計アイコン（Quick select）をクリックし、下部の「Refresh every」に「1 seconds」を指定して、右の「□ Start」ボタンをクリックします。
以上で、１秒ごとにセンサーデータをグラフ表示する画面が作成できました。
Kibana環境の構築が思ったより簡単で、データ可視化は簡単に行えました。最近ですと、Kibana Canvasを使用した素敵な表示画面が今度挑戦したいと思っています。
最後に実施した画面更新ですが、秒単位の周期しか指定できませんでした。温湿度変化程度であればこれで問題なさそうですが、よりリアルタイムなセンサーデータの表示（たとえば加速度センサによる振動計のプロッタのようなもの）には、向かないような気がします。こちらはMQTTによるインプットから、Chart.jsとかで作った方がいいのかもと思いました。
AWSのサービスの一つで、IoTから得たデータを分析、可視化できるサービスです。IoTCoreからの入力データに限らず、あらかじめ集計されたS3データや、RedShiftなどの登録済みデータも使用することができます。
こちらもサービスの一つで、AWSの各種データベースに蓄積されたデータを可視化するBIサービスです。
AWSコンソールからAWS IoT Analyticsを操作し、IoTからの受信データを蓄積する設定を行います。本来、温湿度計等、各種センサーデータを扱うところですが、今回は、以前の書き込みで作成したMQTT受信データ（ラジコンの操作ログ）を集計してくことにします。
AWS IoT Coreで、ルールの作成を行います。
次回はデータを加工して、QuickSightで蓄積した可視化します。
