MELSOFT MaiLabのデモ機を製作してみた①　～M5Stack ＋ InfluxDB v2でセンサデータを時系列保存する編～

#include <InfluxDbClient.h>

// 本プログラム動作に必要なライブラリは以下
//#include <M5Dial.h>
//#include <WiFi.h>
//#include <Wire.h>
//#include <HTTPClient.h>

// InfluxDB 設定 --------------------
#define INFLUXDB_URL "http://192.168.xxx.xxx:8086/"     // NTPサーバと同じIPアドレス
#define INFLUXDB_TOKEN "your-influxdb-token"
#define INFLUXDB_ORG "your-influxdb-org"
#define INFLUXDB_BUCKET "your-influxdb-bucket"
#define WRITE_PRECISION WritePrecision::S   // 精度：秒
#define MAX_BATCH_SIZE 15                  // 一度に書き込まれるポイントの数
#define WRITE_BUFFER_SIZE 30               // バッファー内の最大ポイント数(リトライ含む)
InfluxDBClient InfluxDB_client(INFLUXDB_URL, INFLUXDB_ORG, INFLUXDB_BUCKET, INFLUXDB_TOKEN);
Point status("influxdb-point-name");


void InfluxDB_setup() {  // InfluxDBの接続チェック
  while (!InfluxDB_client.validateConnection()) {
    Serial.print("InfluxDB connection failed: ");
    delay(500);
  }
  Serial.print("Connected to InfluxDB: ");
  Serial.println(InfluxDB_client.getServerUrl());
}

void write_InfluxDB(String st) {  // InfluxDBへ書き込む
  status.clearFields();           // ポイントを空にする
  status.addField("Status", st);  
  status.setTime();                                             // 現在時刻をセット
  InfluxDB_client.writePoint(status);                           // InfluxDBに送信
  Serial.println(InfluxDB_client.pointToLineProtocol(status));  // ポイントの内容表示
  // Write point
  if (!InfluxDB_client.writePoint(status)) {
    Serial.print("InfluxDB write failed: ");
    Serial.println(InfluxDB_client.getLastErrorMessage());
  }
}

void btn_process(int mode) {  // ボタン押下時の処理（※画面表示系の処理は割愛）
  int _responcecode = -1;
  switch (mode) {
    case 1:  // メイン→センサ待機(自動で測定中へ)
      write_InfluxDB("start");
      display_mode = 2;
      break;
    case 2:  // 測定中→終了
      write_InfluxDB("stop");
      display_mode = 3;
      break;
    case 3:  // 終了→メイン
      display_mode = 1;
      break;
  }
  delay(100);
}

void setup() {
  InfluxDB_setup()
　// その他セットアップ処理は割愛
}

long oldPosition = -999;
void loop() {
  M5Dial.update();  // 処理の更新
  if (M5Dial.BtnA.wasPressed()) {　// ボタン押下処理
    btn_process(display_mode);
  }
}

#include <InfluxDbClient.h>

// InfluxDB 設定 --------------------
#define INFLUXDB_URL "http://192.168.xxx.xxx:8086/"     // NTPサーバと同じIPアドレス
#define INFLUXDB_TOKEN "your-influxdb-token"
#define INFLUXDB_ORG "your-influxdb-org"
#define INFLUXDB_BUCKET "your-influxdb-bucket"
#define WRITE_PRECISION WritePrecision::MS  // 精度：ミリ秒
#define MAX_BATCH_SIZE 15                   // 一度に書き込まれるポイントの数
#define WRITE_BUFFER_SIZE 30                // バッファー内の最大ポイント数(リトライ含む)
InfluxDBClient InfluxDB_client(INFLUXDB_URL, INFLUXDB_ORG, INFLUXDB_BUCKET, INFLUXDB_TOKEN);
Point status("influxdb-point-name");

#include "Queue.h"

// マルチスレッド 設定 --------------------
TaskHandle_t thp[2];              // マルチスレッドのタスクハンドル格納用
SemaphoreHandle_t taskSemaphore;  // セマフォ（排他制御）用
bool f_write_point = 0;           // 【フラグ】クエリ→ポイント
bool f_write_InfluxDB = 0;        // 【フラグ】ポイント→InfluxDB
// キュー 設定 --------------------
struct queue_struct { // 書き込むキューの構造
  uint16_t adc_result;
  String timestamp;
};
Queue<queue_struct> queue_sensordata(MAX_BATCH_SIZE);
Queue<queue_struct> copy_queue_sensordata(MAX_BATCH_SIZE);

void setup() {
  // 諸々設定（省略）

  // セマフォ&タスク立ち上げ
  taskSemaphore = xSemaphoreCreateBinary();
  xSemaphoreGive(taskSemaphore);
  xTaskCreatePinnedToCore(Core0a, "Core0a", 4096, taskSemaphore, 4, &thp[0], 0);  // データストア
  xTaskCreatePinnedToCore(Core0b, "Core0b", 4096, taskSemaphore, 3, &thp[1], 0);  // InfluxDBへまとめて書き込み
}

void loop() {
  // センサ値取得、画面描画処理（省略）

  if (isOrder == 1) {                                // APIでInfluxDBモードがONになった場合
    M5Dial.Display.drawString("!DB Connecting!", center_x, center_y + 30);
    if (count_datastock == MAX_BATCH_SIZE) {         // メモリ最大キュー数の場合
      f_write_point = 1;                             // 【フラグ】書き込み指示ON
      count_datastock = 0;
    } else {
      String unixtime = calc_timestamp();
      //取得情報をキュー
      xSemaphoreTake(taskSemaphore, portMAX_DELAY);  // セマフォ取得
      queue_sensordata.push(queue_struct{ adc_mA, unixtime });
      xSemaphoreGive(taskSemaphore);                 // セマフォ解放
      Serial.println(adc_mA);                        // キューした情報の表示
      count_datastock++;
    }
  } else {                                           // 残存データがある場合
    if (count_datastock > 0) {
      M5Dial.Display.drawString("!DB Connecting!", center_x, center_y + 30);
      f_write_point = 1;
      count_datastock = 0;
    }
  }
  delay(300);
}

void Core0a(void* args) {                        // [タスク]キューのコピー
  while (1) {
    delay(1);                                    // コアタスクがバッティングしないように
    if (f_write_point) {
      // Queueをコピー
      SemaphoreHandle_t semaphore = (SemaphoreHandle_t)args;
      xSemaphoreTake(semaphore, portMAX_DELAY);  // セマフォ取得
      Serial.println("query copy start.");
      copy_queue_sensordata = queue_sensordata;
      queue_sensordata.clear();
      xSemaphoreGive(semaphore);                 // セマフォ解放
      Serial.println("query copy end.");
      f_write_InfluxDB = 1;
      f_write_point = 0;
    }
  }
}

void Core0b(void* args) {                        // [タスク]InfluxDBへバッチ書き込み
  while (1) {
    delay(1);                                    // コアタスクがバッティングしないように
    if (f_write_InfluxDB) {
      // QueueをInfluxDB形式へ
      SemaphoreHandle_t semaphore = (SemaphoreHandle_t)args;
      xSemaphoreTake(semaphore, portMAX_DELAY);  // セマフォ取得
      Serial.println("Write InfluxDB start.");
      if (client.isBufferEmpty()) {
        int count = copy_queue_sensordata.count();
        for (int i = 0; i < count; i++) {
          queue_struct item = copy_queue_sensordata.pop();
          Point sensor(INFLUXDB_POINT);
          sensor.addField("ADC_mA", item.adc_result);
          sensor.setTime(item.timestamp);
          client.writePoint(sensor);
        }
        copy_queue_sensordata.clear();
        xSemaphoreGive(semaphore);              // セマフォ解放

        Serial.println("Write InfluxDB end.");
        M5Dial.Speaker.tone(8000, 20);
        f_write_InfluxDB = 0;
      }
    }
  }
}