IoT機器でMachbaseを使う：Raspberry PiとDHT11センサーの実践ガイド
DHT11温湿度センサーのデータをMachbaseへ保存して検索する
概要
Machbaseは、IoT向けに最適化された高速な時系列DBMSです。産業現場の運用改善や予測分析を支え、Raspberry Piのような小型コンピューターでも高速に取り込み・分析できます。一方、データベースやSQLにまだ慣れていない開発者には、時系列データベースという言葉自体が難しく感じられるかもしれません。私も当初は初心者でした。そこで、身近なセンサーを使い、Machbaseを試すための簡単なガイドをまとめます。
私はMachbaseに加えて、時系列DBMSのInfluxDBも試しました。今回の利用範囲では、Machbaseの性能と使いやすさは十分で、特にリソースの限られたRaspberry Piでは、取り込みや検索の速度が印象的でした。ぜひ実際に試していただきたいと思います。
まずMachbase Neoをインストールし、DHT11温湿度センサーをRaspberry Piへ接続します。固定のサンプルではなく、実際に発生する温度・湿度データを使い、収集から保存、検索までの流れを体験します。本稿は元記事のRaspberry Pi 4と旧Adafruit_DHTライブラリを使った実験を紹介するものです。新しいOSやライブラリでは手順・APIが異なる場合があります。
Machbase
Machbase Neoのインストール
Machbase Neoは、Raspberry Piのような小型機器から高性能サーバーまで利用でき、SQLに近いMachSQLを使えます。HTTP、MQTT、gRPCなどを通じてデータの書き込みや検索ができるため、導入を始めやすい構成です。
インストール方法はNeoのWebサイトを参照してください。基本的には配布ファイルをダウンロードして展開し、サーバーを起動します。
この実験では、Linux ARM64向けの配布ファイルを使います。利用するRaspberry Pi OSのアーキテクチャに合うものを選んでください。画面に示す版は、元記事で使った当時のものです。
任意のディレクトリへ展開します。
展開先の machbase-neo 実行ファイルを、serve コマンドで起動します。
初回起動時にデータと設定を保存するディレクトリが作成されます。更新時は、その内容を失わないようバックアップし、使用する版の移行手順に従ってください。起動メッセージでは、各サービスが使うポートも確認できます。
図のコマンドでSQLシェルを開き、テーブルの作成・削除やクエリを実行できます。
DHT11センサーとは
まず、DHT11について簡単に説明します。
DHT11は、温度と湿度をデジタル信号で読み取るセンサーです。ArduinoやRaspberry Piなどで使われます。元記事のモジュールはVCC、DAT、GNDの3端子を持ち、温度0〜50℃、湿度20〜90%の範囲を扱うものです。裸のセンサーとモジュールでは端子構成が異なる場合があるため、使用する製品の仕様を確認してください。
Raspberry Pi 4への接続
電源を切った状態で配線します。3.3V動作に対応するDHT11モジュールなら、VCCを3.3V（物理ピン1）、DATをBCM GPIO2（物理ピン3）、GNDを物理ピン6へ接続します。元記事の「GPIO2＝物理ピン4」は誤記で、物理ピン4は5V電源です。Raspberry PiのGPIOへ5Vを直接入力しないでください。モジュールのプルアップ回路も確認し、データ信号が3.3Vを超えない構成にします。以下の写真は元記事の実験写真です。配線は、この訂正と公式GPIO資料を基に確認してください。
DHT11の値をCSVとして表示する
最初に、取得した値をCSVでコンソールへ出力します。CSVはカンマで値を区切る形式で、保存や変換が容易です。今回は、センサー用のライブラリと資料がそろっているPythonを使います。
元記事のコードは旧Adafruit_DHTライブラリを使います。現在のCircuitPython用 adafruit_dht とはAPIが異なります。再現する環境に対応したAdafruitのセットアップ資料を確認してください。
基本的な流れは次のとおりです。
この例で使用する Adafruit_DHT モジュールをインポートします。
DHT11を指定し、温度と湿度を取得します。コードのピン番号2はBCM GPIO2を意味し、物理ピン番号ではありません。
約3秒ごとに値を読み取り、測定時刻とともにCSVで表示します。Unix時刻は、1970年1月1日00:00:00 UTCからの経過秒数です。データベースではナノ秒単位を使う場合もありますが、最初の表示例は秒単位です。
PythonのUTC日時から timestamp() を呼び出すと、Unix時刻を取得できます。
旧Adafruit_DHTの read_retry() は、湿度と温度をこの順番で返します。
CSV表示用のPythonスクリプトとして保存して実行すると、次のような出力になります。
これがCSV形式です。見た目は簡素ですが、多数の測定値を保存し、他の形式へ変換しやすい利点があります。
温湿度データをリアルタイムにMachbaseへ保存する
次は、約3秒ごとに取得したデータをREST APIで保存します。Pythonの requests モジュールを使えば、HTTP POSTを簡単に送信できます。事前に、コードで指定する名前のタグテーブルを、NAME、TIME、VALUEの列を持つ構成で作成してください。
JSONへの変換とHTTP送信を行うメソッドを持つクラスを用意します。HTTPのステータスだけでなく、返されたJSONの success も確認して取り込み結果を判定します。以下では、元記事で崩れていたPythonの初期化メソッドと起動条件も修正しています。
センサーの値はリストとして受け取ります。CSVファイルを介さず、同じプログラムで取得してすぐに送信するためです。ループを繰り返してデータベースへ追加します。
当初は、保存処理が収集を妨げないよう、スレッドやプロセスを分けることも試しました。しかし、この実験ではかえって待ち時間が増え、単純なループのほうが安定しました。これは今回の実装で得られた結果です。
各ループで取得と送信にかかった時間を測り、3秒からその時間を引いた分だけ待機します。処理自体が3秒を超えた場合には、厳密な3秒周期にはなりません。構成例は次のとおりです。
import datetime, time
import Adafruit_DHT
import json
import requests
class machbase():
def __init__(self, url :str, table :str):
self._url = url
self._tablename = table
self._header = {'Content-Type': 'application/json; charset=utf-8'}
def to_jsonmessage(self, dataset):
table_dict = dict()
table_dict["table"] = self._tablename
data_dict = dict()
data_dict["columns"] = ["name", "time", "value"]
row_list = list()
row_list.append(["temperature", dataset[2], dataset[0]])
row_list.append(["humidity", dataset[2], dataset[1]])
data_dict["rows"] = row_list
table_dict["data"] = data_dict
return table_dict
def write_data_machbase(self, record):
table_dict = self.to_jsonmessage(record)
table_json = json.dumps(table_dict)
response = requests.post(url = self._url, data = table_json, headers = self._header, timeout = 10)
if response.status_code == 200 and response.json().get("success", False):
print("Http post success!")
return 0
else:
print("Http post failed!")
return -1
if __name__ == "__main__":
machbase_url = "http://127.0.0.1:5654/db/write"
machbase_table = "TEMP_HUM_EXAMPLE"
pinnum = 2
machbase_client = machbase(url = machbase_url, table = machbase_table)
while True:
start = float(time.time())
utctime = int(datetime.datetime.now(datetime.timezone.utc).timestamp())*1000000000
hum, temp = Adafruit_DHT.read_retry(Adafruit_DHT.DHT11 ,pinnum)
if hum is not None and temp is not None:
datalist = list()
datalist.append(float(temp))
datalist.append(float(hum))
datalist.append(utctime)
machbase_client.write_data_machbase(datalist)
stop = float(time.time())
elapsed = stop - start
sleep_t = 3 - elapsed
if sleep_t >= 0:
time.sleep(sleep_t)
MachbaseのHTTP書き込みAPIでは、次のようなJSON形式を使います。
NAMEはタグの識別子、TIMEは時刻、VALUEは測定値です。列名は作成したテーブルと一致させてください。
Pythonプログラムを起動すると、取り込みが始まります。
30分ほど経ったところで、SQLシェルからテーブルの内容を確認しました。
gon@raspberrypi:~/Desktop/exer ./machbase shell sql "select * from TEMP_HUM_EXAMPLE"
║38 temperature 2023-01-26 04:35:09.000000 26.000000
║39 humidity 2023-01-26 04:35:06.000000 19.000000
║40 temperature 2023-01-26 04:35:06.000000 26.000000
║41 temperature 2023-01-26 04:35:03.000000 26.000000
║42 humidity 2023-01-26 04:35:03.000000 19.000000
║43 temperature 2023-01-26 04:35:00.000000 26.000000
║44 humidity 2023-01-26 04:35:00.000000 19.000000
║45 humidity 2023-01-26 04:34:57.000000 19.000000
║46 temperature 2023-01-26 04:34:57.000000 26.000000値が正常に保存されていたので、その後も約1週間、プログラムを動かして収集しました。
簡単なクエリを試す
保存したデータをすべて表示するだけでなく、条件や集計によって必要な結果を取り出してみましょう。ここでは2つの例を紹介します。
1. 温度と湿度の最大値を求める
MachSQLは、一般的なSQLに近い構文を提供します。
MAX関数とGROUP BYで、タグごとの最大値を求められます。
2. 温度と湿度の1時間ごとの平均を求める
DATE_TRUNC() で、時刻を指定した単位の区間にまとめられます。
時間単位、時刻列、区間の長さを指定します。例えば1時間ごとの集計では、DATE_TRUNC('hour', TIME, 1) を使います。
select name, avg(value), date_trunc('hour', TIME, 1) as _TIME from temp_hum_example
group by _TIME,
name order by _TIME;
データ量が多くても、今回の実験では短い時間で結果を取得できました。SQLに慣れていれば、条件や集計を変えて必要な結果を取り出せます。さまざまなクエリを試してみてください。
まとめ
Machbase Neoを起動し、DHT11から温度・湿度を取得して保存しました。元記事の実験では専用のPythonクライアントを使わず、requestsでHTTP POSTを送りました。所定のJSON形式を用意すれば、簡単に書き込めます。検索にはSQLに近いMachSQLを使い、コンソールのSQLシェルから操作できました。大量のデータでも、実験では1秒未満で結果が返る場面がありました。
データベースの知識がほとんどない状態から、約1か月Machbaseを試した感想として、始めやすい製品だと感じました。すべての作業をRaspberry Piで行い、限られたハードウェアでも取り込みと検索の性能を体験できました。多くの操作はSSH経由のコンソールで完結し、モニターを接続する必要もありませんでした。手頃なRaspberry Piから、ぜひ試してみてください。
