002.センサー値を取得して蓄積する

• シリアルポートでセンサー値を転送する
• 接続先PC側でシリアルポートに流れるデータを受信する
• PCからMQTTへパブリッシュする
• MQTTでサブスクライブしてPostgreSQLに書き込む
• 定期的にPostgreSQLから読み出してグラフを生成する

LCDTempture.ino

#include <LiquidCrystal.h>
#include <DHT.h>
 
#define DHTPIN  7
#define DHTTYPE DHT22
 
LiquidCrystal lcd(12, 11, 5, 4, 3, 2);
DHT dht(DHTPIN, DHTTYPE);
// 部屋の湿度計に合わせた補正値
float adj = 0.542;
 
void setup()
{
    lcd.begin(16, 2);
    dht.begin();
    Serial.begin(9600);
    lcd.print("DHT22 starting");
    delay(2000);
    lcd.clear();
}
 
void loop()
{
    float hum  = dht.readHumidity();
    float temp = dht.readTemperature();
 
    if (isnan(hum) || isnan(temp)) {
        lcd.setCursor(0, 0);
        lcd.print("Sensor Error!   ");
        lcd.setCursor(0, 1);
        lcd.print("                ");
        delay(2000);
        return;
    }
 
    lcd.setCursor(0, 0);
    lcd.print("Temp:");
    lcd.print(temp, 1);
    lcd.print(" C   ");
 
    lcd.setCursor(0, 1);
    lcd.print("Hum :");
    lcd.print(hum * adj, 1);
    lcd.print(" %   ");
 
    Serial.print("{\"device\":\"uno01\",\"temperature\":");
    Serial.print(temp, 1);
    Serial.print(",\"humidity\":");
    Serial.print(hum * adj, 1);
    Serial.println("}");
    delay(2000);
}

温湿度センサーをDHT11→DHT22へ変更したので記述が少し変わっている。DHT11を利用しているなら DHTTYPE を DHT11 、メッセージをDHT11 starting へ変更する。

内容としてはセンサーから取得した温度・湿度をプロパティ temperature・humidity に設定したJSONの文字列を作りシリアルポートに書き出している。このデータはPCのCOMポートから読み出しができる。

mqttgate.py

import serial
import paho.mqtt.client as mqtt
 
SERIAL_PORT = "COM3"
SERIAL_SPEED = 9600          # 調べて変更する事
 
MQTT_HOST = "192.168.1.20"   # MQTT BrokerのIP
MQTT_PORT = 1883
MQTT_TOPIC = "sensor/uno01/environment"
 
client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id="win-serial2mqtt"
)
 
client.connect(MQTT_HOST, MQTT_PORT, 60)
client.loop_start()
 
with serial.Serial(SERIAL_PORT, SERIAL_SPEED, timeout=5) as ser:
    while True:
        line = ser.readline()
 
        if not line:
            continue
 
        try:
            text = line.decode("utf-8").strip()
        except UnicodeDecodeError:
            continue
 
        if not text:
            continue
 
        print("serial:", text)
 
        client.publish(
            MQTT_TOPIC,
            text,
            qos=0,
            retain=False
        )

仮想COMポートをシリアルインタフェースとしてプログラムにアクセスさせればいい。読みだした結果は MQTTプロトコルで宅内ブローカーにパブリッシュする。
ChatGPTさんに仕様を話しすれば Python のコードをササっと作ってくれる。

宅内にはMosquittoが稼働しているサーバがあり、今回は UNO R3 から読みだした JSON をトピックsensor/uno01/environmentへパブリッシュする。
※HTTPで送るのもよいが、その場合は受信側（サーバ側）を作るのが面倒だったので MQTT 一択だった。

今回のスクリプトは2秒おきの送信だがHTTPでは処理遅延が起きそうだな、って感じたのもある。

$ psql mqtt mqtt
psql (18.6)
"help"でヘルプを表示します。
 
mqtt=> \d
                  リレーション一覧
 スキーマ |       名前        |   タイプ   | 所有者
----------+-------------------+------------+--------
 public   | sensor_log        | テーブル   | mqtt
 public   | sensor_log_id_seq | シーケンス | mqtt
(2 行)
 
mqtt=> \d sensor_log
                                        テーブル"public.sensor_log"
     列      |          タイプ          | 照合順序 | Null 値を許容 |               デフォルト
-------------+--------------------------+----------+---------------+----------------------------------------
 id          | bigint                   |          | not null      | nextval('sensor_log_id_seq'::regclass)
 received_at | timestamp with time zone |          | not null      | now()
 device      | text                     |          | not null      |
 temperature | numeric(5,2)             |          |               |
 humidity    | numeric(5,2)             |          |               |
 payload     | jsonb                    |          |               |
インデックス:
    "sensor_log_pkey" PRIMARY KEY, btree (id)
 
mqtt=>

単純にJSONパース後の各プロパティ値に対応するカラムと、ID、受信時タイムスタンプ、payloadがある。
タイムスタンプはテーブルへのINSERT時の時刻を適用。データの発生時刻ではなくてテーブルへの追加時刻。UNO R3からデータ受信した際のタイムスタンプは持っていないから。
payloadはサブスクライブした時のトピックがそのまま入る。
IDは今のところ使い道がないが念のため。

mqtt2pg.py

#!/usr/bin/env python3.12
 
import json
 
import paho.mqtt.client as mqtt
import psycopg
from psycopg.types.json import Jsonb
 
 
MQTT_HOST = "192.168.1.20"
MQTT_PORT = 1883
MQTT_TOPIC = "sensor/uno01/environment"
 
PG_DSN = (
    "host=192.168.1.23"
    "dbname=mqtt "
    "user=mqtt "
)
 
 
def on_connect(client, userdata, flags, reason_code, properties):
    print(f"MQTT connected: {reason_code}")
 
    client.subscribe(MQTT_TOPIC)
    print(f"subscribe: {MQTT_TOPIC}")
 
 
def on_message(client, userdata, msg):
    try:
        text = msg.payload.decode("utf-8")
        print(f"recv: topic={msg.topic} payload={text}")
 
        # JSONなら解析する
        try:
            data = json.loads(text)
        except json.JSONDecodeError:
            # 普通の文字列だった場合もJSONBへ保存できる形にする
            data = {
                "value": text
            }
 
        # JSONオブジェクトでなければ包む
        if not isinstance(data, dict):
            data = {
                "value": data
            }
 
        device = data.get("device", msg.topic)
        temperature = data.get("temperature")
        humidity = data.get("humidity")
 
        with psycopg.connect(PG_DSN) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO sensor_log
                        (device, temperature, humidity, payload)
                    VALUES
                        (%s, %s, %s, %s)
                    """,
                    (
                        device,
                        temperature,
                        humidity,
                        Jsonb(data)
                    )
                )
 
        print("inserted.")
 
    except Exception as e:
        print(f"ERROR: {e}")
 
 
client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id="mqtt2postgres"
)
 
client.on_connect = on_connect
client.on_message = on_message
 
client.connect(MQTT_HOST, MQTT_PORT, 60)
client.loop_forever()

ブローカーからトピックを取得してPostgreSQLのテーブルへ書き出すコード。もちろんChatGPTさん謹製。
ただし、このコードはある意味危険なので絶対本番的なところに使わない事。まぁ宅内でしか使わないような、セキュリティガバガバなコードなのは理解してもらっていると思う。

このスクリプトをバックグラウンドで走らせれば延々と計測値が記録され続ける。
このデータ量なら結構先になると思うけど、ストレージがいっぱいにならないように適宜古いデータを消す作業を実施すること。データ保存期間を決めておいた方が良い。