InfluxDB Cloud にLTEモデムで送信する

はじめに

influxDBはデータ可視化ツールGrafanaとよく使われIoTセンサーの時系列データベースとしてよく使われています。
https://cloud2.influxdata.com/

クラウドサービスとしてのinfluxDB CloudにはESP32などからwifiでデータを送信できるように
InfluxDB Client for Arduino libraryが用意されています
ですが野外のセンサーなどwifiの無いところではこのライブラリは使えません。

そこでAPIの仕様を見ながら簡単なサンプルを
・ spresense+LTE拡張ボード
・ ESP32（M5stickC)+CatM UNIT(SIM7080G)
・IoT スターターキット for Arduino
で書いてみました
https://docs.influxdata.com/influxdb/cloud/api/#operation/PostWrite

証明書取得

まずTLS送信のために証明書をダウンロードします
今回は上のライブラリ内にあるヘッダーファイルをそのまま使います

ダウンロードして以下のスケッチと同じフォルダに入れておきます

InfluxDB 設定情報

・host
influxDB Cloudにログインした状態でのURLのホスト名
例：us-east-1-1.aws.cloud2.influxdata.com

・ bucketsを作る
Data (Load Data) > BUCKETS　>　 CREATE BUCKET
https://docs.influxdata.com/influxdb/v2/admin/buckets/create-bucket/
例：ltesensor

・APIトークンを取得
Data (Load Data) > API Tokens　>　 GENERATE API TOKEN
https://docs.influxdata.com/influxdb/v2/admin/tokens/create-token/

・　ラインプロトコル
送信データは以下のようにプロトコルが決められてます
https://docs.influxdata.com/influxdb/cloud/reference/syntax/line-protocol/

例:
sensor,device=SPRESENSE count=4u,temperature=25.0,humidity=80i

Measurement（テーブルのようなもの）に Point （レコードのようなもの）をテキスト1行で追加します

tag は省略可で検索時に高速に抽出されます

field の値はなにもないとfloat型　後ろに"i"を付けるとint64型 ,"u"はuint64型,ダブルクオーテーションで囲うとString型になります
空白の位置にも意味があるのでよく確認しておきます

timestamp　は送信時に省略するとクラウド側で自動的に付与されます（デフォルトではナノ秒のUNIX　time) 今回のサンプルでは略

参考　https://www.mikan-tech.net/entry/what-is-influxdb

サンプルスケッチ

spresense + LTE拡張ボード

・ライブラリ
ArduinoHttpClient
arduino-libraries/ArduinoHttpClient@^0.5.0
https://github.com/arduino-libraries/ArduinoHttpClient

influxdbHttpPostWrite.ino

#include <Arduino.h>
// libraries
#include <ArduinoHttpClient.h>
#include <RTC.h>
#include <LTE.h>
#include "InfluxDbCloud.h"

// LTE setting
#define APP_LTE_APN "povo.jp" // replace your APN
#define APP_LTE_USER_NAME "" // replace with your username
#define APP_LTE_PASSWORD  "" // replace with your password
#define APP_LTE_IP_TYPE (LTE_NET_IPTYPE_V4V6) // IP : IPv4v6
#define APP_LTE_AUTH_TYPE (LTE_NET_AUTHTYPE_CHAP) // Authentication : CHAP
#define APP_LTE_RAT (LTE_NET_RAT_CATM) // RAT : LTE-M (LTE Cat-M1)

// InfluxDB setting
// E.g. http://192.168.1.48:8086 (In InfluxDB 2 UI -> Load Data -> Client Libraries), 
#define INFLUXDB_URL "*********.aws.cloud2.influxdata.com"
// InfluxDB 2 server or cloud API authentication token (Use: InfluxDB UI -> Load Data -> Tokens -> <select token>)
#define INFLUXDB_TOKEN "***YOUR TOKEN****"

// InfluxDB 2 bucket name (Use: InfluxDB UI -> Load Data -> Buckets)
#define INFLUXDB_BUCKET "ltesensor"

char postWritePath[] = "/api/v2/write?bucket=" INFLUXDB_BUCKET;
int port = 443; // port 443 is the default for HTTPS

// initialize the library instance
LTE lteAccess;
LTETLSClient tlsClient;
HttpClient client = HttpClient(tlsClient, INFLUXDB_URL, port);
uint count = 0;

void printClock(RtcTime &rtc)
{
 printf("%04d/%02d/%02d %02d:%02d:%02d\n",
        rtc.year(), rtc.month(), rtc.day(),
        rtc.hour(), rtc.minute(), rtc.second());
}

void setup()
{
 char apn[LTE_NET_APN_MAXLEN] = APP_LTE_APN;
 LTENetworkAuthType authtype = APP_LTE_AUTH_TYPE;
 char user_name[LTE_NET_USER_MAXLEN] = APP_LTE_USER_NAME;
 char password[LTE_NET_PASSWORD_MAXLEN] = APP_LTE_PASSWORD;

 // initialize serial communications and wait for port to open:
 Serial.begin(115200);
 while (!Serial) {
     ; // wait for serial port to connect. Needed for native USB port only
 }

 Serial.println("Starting secure HTTP client.");
 Serial.println("=========== APN information ===========");
 Serial.print("Access Point Name  : ");
 Serial.println(apn);
 Serial.print("Authentication Type: ");
 Serial.println(authtype == LTE_NET_AUTHTYPE_CHAP ? "CHAP" :
                authtype == LTE_NET_AUTHTYPE_NONE ? "NONE" : "PAP");
 if (authtype != LTE_NET_AUTHTYPE_NONE) {
   Serial.print("User Name          : ");
   Serial.println(user_name);
   Serial.print("Password           : ");
   Serial.println(password);
 }

 while (true) {

   /* Power on the modem and Enable the radio function. */

   if (lteAccess.begin() != LTE_SEARCHING) {
     Serial.println("Could not transition to LTE_SEARCHING.");
     Serial.println("Please check the status of the LTE board.");
     for (;;) {
       sleep(1);
     }
   }

   /* The connection process to the APN will start.
    * If the synchronous parameter is false,
    * the return value will be returned when the connection process is started.
    */
   if (lteAccess.attach(APP_LTE_RAT,
                        apn,
                        user_name,
                        password,
                        authtype,
                        APP_LTE_IP_TYPE) == LTE_READY) {
     Serial.println("attach succeeded.");
     break;
   }

   /* If the following logs occur frequently, one of the following might be a cause:
    * - APN settings are incorrect
    * - SIM is not inserted correctly
    * - If you have specified LTE_NET_RAT_NBIOT for APP_LTE_RAT,
    *   your LTE board may not support it.
    * - Rejected from LTE network
    */
   Serial.println("An error has occurred. Shutdown and retry the network attach process after 1 second.");
   lteAccess.shutdown();
   sleep(1);
 }

 // Set local time (not UTC) obtained from the network to RTC.
 RTC.begin();
 unsigned long currentTime;
 while(0 == (currentTime = lteAccess.getTime())) {
   sleep(1);
 }
 RtcTime rtc(currentTime);
 printClock(rtc);
 RTC.setTime(rtc);

 tlsClient.setCACert(InfluxDbCloud2CACert);
}

void loop()
{
 // HTTP POST method
 Serial.println("making POST request");
 String measurement = "sensor";
 String tagSet = "device=SPRESENSE";
 String fieldSet ;
 float temp = 25.0;
 int humi = 80;
 float press = 1013.00;
 
 String contentType = "text/plain; charset=utf-8";
 fieldSet = "count=" + String(count) + "u";
 fieldSet += ",temperature=" + String(temp,1);
 fieldSet += ",humidity=" + String(humi) + "i";
 fieldSet += ",pressure=" + String(press,2);
 String postData = measurement + "," + tagSet + " " + fieldSet ;
 Serial.println(postData);
 client.beginRequest();
 client.post(postWritePath);
 client.sendHeader("Authorization: Token "INFLUXDB_TOKEN);
 client.sendHeader(HTTP_HEADER_CONTENT_TYPE, contentType);
 client.sendHeader(HTTP_HEADER_CONTENT_LENGTH,postData.length());
 client.endRequest();

 client.write((const byte*)postData.c_str(), postData.length());

 int statusCode = client.responseStatusCode();
 String response = client.responseBody();

 Serial.print("Status code: ");
 Serial.println(statusCode);
 Serial.print("Response: ");
 Serial.println(response);

 count ++;
 sleep(60);
}

参考:SPRESENSE LTEチュートリアル
TLSプロトコルを使用したHTTP Clientの動作を確認する

M5stickC + M5Stack CAT-M UNIT（SIM7080G搭載）

・ライブラリ

• ArduinoHttpClient
  arduino-libraries/ArduinoHttpClient@^0.5.0
  https://github.com/arduino-libraries/ArduinoHttpClient

• TinyGSM
  vshymanskyy/TinyGSM@^0.11.7
  https://github.com/vshymanskyy/TinyGSM

influxdbHttpPostWrite.ino

#include <Arduino.h>
// libraries
#define TINY_GSM_MODEM_SIM7080
#include <TinyGsmClient.h>
#include <ArduinoHttpClient.h>
#include "InfluxDbCloud.h"
#define SerialAT Serial1

// LTE setting
#define APP_LTE_APN "povo.jp" // replace your APN
#define APP_LTE_USER_NAME "" // replace with your username
#define APP_LTE_PASSWORD  "" // replace with your password

// InfluxDB setting
// E.g. http://192.168.1.48:8086 (In InfluxDB 2 UI -> Load Data -> Client Libraries), 
#define INFLUXDB_URL "*********.aws.cloud2.influxdata.com"
// InfluxDB 2 server or cloud API authentication token (Use: InfluxDB UI -> Load Data -> Tokens -> <select token>)
#define INFLUXDB_TOKEN "***YOUR TOKEN****"

// InfluxDB 2 bucket name (Use: InfluxDB UI -> Load Data -> Buckets)
#define INFLUXDB_BUCKET "ltesensor"

char postWritePath[] = "/api/v2/write?bucket=" INFLUXDB_BUCKET;
int port = 443; // port 443 is the default for HTTPS

// initialize the library instance
TinyGsm        modem(SerialAT);
TinyGsmClientSecure tlsClient(modem);
HttpClient client = HttpClient(tlsClient, INFLUXDB_URL, port);

uint count = 0;

void writeCaFiles(int index, const char *filename, const char *data,
                 size_t lenght)
{
   modem.sendAT("+CFSTERM");
   modem.waitResponse();


   modem.sendAT("+CFSINIT");
   if (modem.waitResponse() != 1) {
       Serial.println("INITFS FAILED");
       return;
   }
   // AT+CFSWFILE=<index>,<filename>,<mode>,<filesize>,<input time>
   // <index>
   //      Directory of AP filesystem:
   //      0 "/custapp/" 1 "/fota/" 2 "/datatx/" 3 "/customer/"
   // <mode>
   //      0 If the file already existed, write the data at the beginning of the
   //      file. 1 If the file already existed, add the data at the end o
   // <file size>
   //      File size should be less than 10240 bytes. <input time> Millisecond,
   //      should send file during this period or you can’t send file when
   //      timeout. The value should be less
   // <input time> Millisecond, should send file during this period or you can’t
   // send file when timeout. The value should be less than 10000 ms.

   size_t payloadLenght = lenght;
   size_t totalSize     = payloadLenght;
   size_t alardyWrite   = 0;

   while (totalSize > 0) {
       size_t writeSize = totalSize > 10000 ? 10000 : totalSize;

       modem.sendAT("+CFSWFILE=", index, ",", "\"", filename, "\"", ",",
                    !(totalSize == payloadLenght), ",", writeSize, ",", 10000);
       modem.waitResponse(30000UL, "DOWNLOAD");
REWRITE:
       modem.stream.write(data + alardyWrite, writeSize);
       if (modem.waitResponse(30000UL) == 1) {
           alardyWrite += writeSize;
           totalSize -= writeSize;
           Serial.printf("Writing:%d overage:%d\n", writeSize, totalSize);
       } else {
           Serial.println("Write failed!");
           delay(1000);
           goto REWRITE;
       }
   }

   Serial.println("Wirte done!!!");

   modem.sendAT("+CFSTERM");
   if (modem.waitResponse() != 1) {
       Serial.println("CFSTERM FAILED");
       return;
   }
}

void setup()
{
 // Set GSM module baud rate
 Serial.begin(115200);
 SerialAT.begin(115200, SERIAL_8N1, 33, 32);//M5StickC
 delay(1000);
 Serial.println("start");

 // Restart takes quite some time
 // To skip it, call init() instead of restart()
 Serial.println("Initializing modem...");
 // modem.restart();
 modem.init();

 String modemInfo = modem.getModemInfo();
 Serial.print("Modem Info: ");
 Serial.println(modemInfo);
 modem.gprsConnect(APP_LTE_APN, APP_LTE_USER_NAME, APP_LTE_PASSWORD );
 Serial.print("waitForNetwork()");
 while (!modem.waitForNetwork()) Serial.print(".");
 Serial.println(" Ok.");

 Serial.print("gprsConnect");
 modem.gprsConnect(APP_LTE_APN, APP_LTE_USER_NAME, APP_LTE_PASSWORD );
 Serial.println(" done.");

 Serial.print(F("isNetworkConnected()"));
 while (!modem.isNetworkConnected()) Serial.print(".");
 Serial.println(F(" Ok."));

 Serial.print("My IP addr: ");
 IPAddress ipaddr = modem.localIP();
 Serial.println(ipaddr);
 //Set CACert  
 writeCaFiles(3, "root.crt", InfluxDbCloud2CACert, strlen(InfluxDbCloud2CACert));
 modem.sendAT("+CSSLCFG=\"SSLVERSION\",0,3");
 modem.waitResponse();
 modem.sendAT("+CSSLCFG=\"CONVERT\",2,\"root.crt\"");
 if (modem.waitResponse() != 1) {
     Serial.println("Convert root.crt failed!");
 }
}

void loop()
{
 // HTTP POST method
 Serial.println("making POST request");
 String measurement = "sensor";
 String tagSet = "device=SIM7080G";
 String fieldSet ;
 float temp = 25.0;
 int humi = 80;
 float press = 1013.00;
 
 String contentType = "text/plain; charset=utf-8";
 fieldSet = "count=" + String(count) + "u";
 fieldSet += ",temperature=" + String(temp,1);
 fieldSet += ",humidity=" + String(humi) + "i";
 fieldSet += ",pressure=" + String(press,2);
 String postData = measurement + "," + tagSet + " " + fieldSet ;
 Serial.println(postData);
 
 client.beginRequest();
 client.post(postWritePath);
 client.sendHeader("Authorization: Token "INFLUXDB_TOKEN);
 client.sendHeader(HTTP_HEADER_CONTENT_TYPE, contentType);
 client.sendHeader(HTTP_HEADER_CONTENT_LENGTH,postData.length());
 client.endRequest();

 client.write((const byte*)postData.c_str(), postData.length());

 int statusCode = client.responseStatusCode();
 String response = client.responseBody();

 Serial.print("Status code: ");
 Serial.println(statusCode);
 Serial.print("Response: ");
 Serial.println(response);

 count ++;
 sleep(60);
}

参考:LilyGo-T-SIM7080G　ModemMqttsExample

Response

204 が返ってくれば書き込み成功です

SORACOM BEAM を利用する

上記はモデム自身でHTTPSの認証をやっていましたが、soracom　のデータ転送支援サービス BEAM(https://soracom.jp/services/beam/) 　を使えばモデム側はHTTP（TCP,UDP）で送信したものをHTTPSに変換して転送します。センサー端末側の通信コストや省電力に役立ちます

BEAM設定

SIMグループ　-　基本設定　-　SORACOM BEAM設定　- HTTPエントリポイントを以下のように設定します
ホスト名、bucket名、Tokenは適宜入れ替えて

スケッチ

IoT スターターキット for Arduino（Quectel BG96搭載) を使って

温度 (C) と湿度 (%) を 60 秒ごとに計測し JSON で Unified Endpoint へ送信する

このスケッチを参考に

send_temp_and_humi_with_soracom.ino

/*
 * send_temp_and_humi_with_soracom.ino
 * Ambient temperature and humidity send to Cloud via Unified Endpoint of SORACOM
 *
 * Copyright SORACOM
 * This software is released under the MIT License, and libraries used by these sketches 
 * are subject to their respective licenses.
 * See also: https://github.com/soracom-labs/arduino-dragino-unified/README.md
*/

#define CONSOLE Serial
#define INTERVAL_MS (60000)
#define ENDPOINT "uni.soracom.io"
#define SKETCH_NAME "send_temp_and_humi_with_soracom"
#define VERSION "1.1"

/* for LTE-M Shield for Arduino */
#define RX 10
#define TX 11
#define BAUDRATE 9600
#define BG96_RESET 15

#define TINY_GSM_MODEM_BG96
#include <TinyGsmClient.h>

#include <SoftwareSerial.h>
SoftwareSerial LTE_M_shieldUART(RX, TX);
TinyGsm modem(LTE_M_shieldUART);
TinyGsmClient ctx(modem);

#include <U8x8lib.h>
U8X8_SSD1306_128X64_NONAME_HW_I2C u8x8(/* clock=*/ SCL, /* data=*/ SDA, /* reset=*/ U8X8_PIN_NONE);
#define U8X8_ENABLE_180_DEGREE_ROTATION 1

#include <DHT.h>

#define USE_DHT11 // Use DHT11 (Blue)
//#define USE_DHT20 // Use DHT20 (Black)

#ifdef USE_DHT11
  #define dht11Pin 3
  DHT dht(dht11Pin, DHT11);
#endif
#ifdef USE_DHT20
  DHT dht(DHT20);
#endif

#define OLED_MAX_CHAR_LENGTH 16
void drawText(U8X8* u8x8, const char* in_str, int width = OLED_MAX_CHAR_LENGTH);
void drawText_P(U8X8* u8x8, const char* pgm_s, int width = OLED_MAX_CHAR_LENGTH);

#define RESET_DURATION 86400000UL // 1 day
void software_reset() {
  asm volatile ("  jmp 0");
}

void setup() {
  CONSOLE.begin(115200);
  LTE_M_shieldUART.begin(BAUDRATE);
  u8x8.begin();
  u8x8.setFlipMode(U8X8_ENABLE_180_DEGREE_ROTATION);
  u8x8.setFont(u8x8_font_victoriamedium8_r);

  CONSOLE.print(F("Welcome to ")); CONSOLE.print(SKETCH_NAME); CONSOLE.print(F(" ")); CONSOLE.println(VERSION);
  u8x8.clear();
  drawText_P(&u8x8, PSTR("Welcome to ")); drawText(&u8x8, SKETCH_NAME); drawText_P(&u8x8, PSTR(" ")); drawText(&u8x8, VERSION);
  delay(3000);

  CONSOLE.print(F("resetting module "));
  pinMode(BG96_RESET,OUTPUT);
  digitalWrite(BG96_RESET,LOW);
  delay(300);
  digitalWrite(BG96_RESET,HIGH);
  delay(300);
  digitalWrite(BG96_RESET,LOW);
  CONSOLE.println(F(" done."));

  u8x8.clear();
  drawText_P(&u8x8, PSTR("modem.restart()..."));
  modem.restart();
  drawText_P(&u8x8, PSTR("done."));
  delay(500);

  u8x8.clear();
  drawText_P(&u8x8, PSTR("modem.getModemInfo():"));
  String modemInfo = modem.getModemInfo();
  u8x8.println();
  int modem_info_len = modemInfo.length() + 1;
  char modem_info_buf[modem_info_len];
  modemInfo.toCharArray(modem_info_buf, modem_info_len);
  drawText(&u8x8, modem_info_buf);
  delay(2000);

  u8x8.clear();
  drawText_P(&u8x8, PSTR("waitForNetwork()..."));
  while (!modem.waitForNetwork());
  drawText_P(&u8x8, PSTR("Ok."));
  delay(500);

  u8x8.clear();
  drawText_P(&u8x8, PSTR("gprsConnect(soracom.io)..."));
  modem.gprsConnect("soracom.io", "sora", "sora");
  drawText_P(&u8x8, PSTR("done."));
  delay(500);

  u8x8.clear();
  drawText_P(&u8x8, PSTR("isNetworkConnected()..."));
  while (!modem.isNetworkConnected());
  drawText_P(&u8x8, PSTR("Ok."));
  delay(500);

  u8x8.clear();
  drawText_P(&u8x8, PSTR("My IP addr: "));
  u8x8.println();
  IPAddress ipaddr = modem.localIP();
  CONSOLE.println(ipaddr);
  char ip_addr_buf[20];
  sprintf_P(ip_addr_buf, PSTR("%d.%d.%d.%d"), ipaddr[0], ipaddr[1], ipaddr[2], ipaddr[3]);
  drawText(&u8x8, ip_addr_buf);
  delay(2000);

  dht.begin();
}

void loop() {
  // HTTP POST method
  CONSOLE.println("making POST request");
  char measurement[] = "sensor";
  char tagSet[] = "device=SORACOM";
  String fieldSet ;
  float t = dht.readTemperature();
  char t_buf1[10];
  dtostrf(t, -1, 1, t_buf1); /*  左詰め*/
  int humi = (int) dht.readHumidity();
  char payload[40];
  sprintf_P(payload, PSTR("%s,%s temperature=%s,humidity=%di"),measurement,tagSet, t_buf1, humi);

  CONSOLE.println(payload);
  u8x8.clear();
  drawText(&u8x8, payload);
  u8x8.println();

  CONSOLE.print(F("Send..."));
  drawText_P(&u8x8, PSTR("Send..."));
  /* connect */
  if (!ctx.connect(ENDPOINT, 80)) {
    CONSOLE.println(F("failed."));
    drawText_P(&u8x8, PSTR("failed."));
    delay(3000);
    return;
  }
  /* send request */
  char hdr_buf[40];
  ctx.println(F("POST / HTTP/1.1"));
//  sprintf_P(hdr_buf, PSTR("Host: %s"), ENDPOINT);
//  ctx.println(hdr_buf);
//  ctx.println(F("Content-Type: text/plain; charset=utf-8"));
  sprintf_P(hdr_buf, PSTR("Content-Length: %d"), strlen(payload));
  ctx.println(hdr_buf);
  ctx.println();
  ctx.println(payload);
  /* receive response */
  while (ctx.connected()) {
    String line = ctx.readStringUntil('\n');
    CONSOLE.println(line);
    if (line == "\r") {
      CONSOLE.println(F("Response header received."));
      break;
    }
  }
  // NOTE: response body is ignore
  ctx.stop();
  CONSOLE.println(F("done."));
  drawText_P(&u8x8, PSTR("done."));

  delay(INTERVAL_MS);
  
#ifdef RESET_DURATION
  if(millis() > RESET_DURATION )
  {
    CONSOLE.println("Execute software reset...");
    delay(1000);
    software_reset();
  }
#endif
}

void drawText(U8X8* u8x8, const char* in_str, int width = OLED_MAX_CHAR_LENGTH) {
  size_t len = strlen(in_str);
  for (int i = 0 ; i < len ; i++) {
    if (u8x8->tx > width - 1) {
      u8x8->tx = 0; // CR
      u8x8->ty++;   // LF
    }
    u8x8->print(in_str[i]);
  }
}

void drawText_P(U8X8* u8x8, const char* pgm_s, int width = OLED_MAX_CHAR_LENGTH) {
  size_t len = strlen_P(pgm_s);
  for (int i = 0 ; i < len ; i++) {
    if (u8x8->tx > width - 1) {
      u8x8->tx = 0; // CR
      u8x8->ty++;   // LF
    }
    char c = pgm_read_byte(pgm_s++);
    u8x8->print(c);
  }
}