M5Stack用3G拡張ボードを使用したMQTT通信方式による画像データ分割送信方法

はじめに

M5Stack用3G拡張ボードを使用したMQTT通信方式でのデータ送信に関する記事が見つからなかったため，本記事でまとめました．
さらに，M5Stack用3G拡張ボードを使用して画像データを送信する際に，1度のMQTT.Publishで送信できない状況に遭遇したため（原因はWifi経由であれば1度で送信できることから画像データが大きく3G通信の問題と考えられるが詳細は不明），この問題の解決も記載しています．

ここでは，以下の環境で実行しております．

クライアント側
・M5Stack GRAY (ボード選択 → M5Stack-Core2）
・M5Stack用3G拡張ボード
・ArduinoIDE
・M5UnitV
・MaixPy

サーバ側
・さくらVPS
・Node.js
・MQTT.Broker

Node.jsを使用したMQTT設定は以下のリンクから確認してください．

また，MaixPyの初期設定や3G拡張ボードのセットアップ，M5StackとM5UnitVの通信については以下のリンクから確認してください．

ここでは，M5Stack用3G拡張ボードを使用したMQTT通信に焦点を当てており，M5StackとM5UnitVのシリアル通信などのプログラムの解説は行っていません．

目次

本記事では以下の流れで解説します．

1.はじめに
2.M5Stack用3G拡張ボードを使用したMQTT通信
3.M5UnitVから取得した画像データの加工
4.Node.jsの受信方法
5.おわりに

M5Stack用3G拡張ボードを使用したMQTT通信

3G回線を用いたMQTT通信を行う準備

M5Stack用3G拡張ボードの通信については以下のリンクから詳細を確認できます．

M5Stackの「PubSubClient」ライブラリ用いたMQTT通信については以下のリンクから詳細を確認できます．

基本的なプログラムは以下のようになります．M5Stack用3G拡張ボードを使用するためには ArduinoIDE で 「TinyGSM」ライブラリのインストール が必要です．またMQTT通信を行うために 「PubSubClient」ライブラリのインストール も行ってください．

#include <M5Stack.h>
#include <string.h>

/* 3Gボード設定 */
#define TINY_GSM_MODEM_UBLOX
#include <TinyGsmClient.h>
TinyGsm modem(Serial2); // Serial1はM5UnitVとのシリアル通信で使用するためSerial2としている
TinyGsmClient ctx(modem);

/* MQTT通信の設定 */
#include <PubSubClient.h>
PubSubClient mqttClient(ctx);

/*
MQTT関連
*/
const char* mqttHost = "****";   // MQTTサーバのアドレス
const int mqttPort = ****;       // MQTTのポート
const char* topic_sub = "****";  // 受信用のトピック名
const char* topic_pub = "****";  // 送信先のトピック名

/*
3G通信
*/
void init3G() {
  M5.Lcd.clear(BLACK);
  M5.Lcd.setTextColor(WHITE);
  M5.Lcd.println(F("M5Stack + 3G Module"));

  M5.Lcd.print(F("modem.restart()"));
  Serial2.begin(115200, SERIAL_8N1, 16, 17);
  modem.restart();
  M5.Lcd.println(F("done"));

  M5.Lcd.print(F("getModemInfo:"));
  String modemInfo = modem.getModemInfo();
  M5.Lcd.println(modemInfo);

  M5.Lcd.print(F("waitForNetwork()"));
  while (!modem.waitForNetwork()) M5.Lcd.print(".");
  M5.Lcd.println(F("Ok"));

  M5.Lcd.print(F("gprsConnect(soracom.io)"));
  modem.gprsConnect("soracom.io", "sora", "sora");
  M5.Lcd.println(F("done"));

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

  M5.Lcd.print(F("My IP addr: "));
  IPAddress ipaddr = modem.localIP();
  M5.Lcd.print(ipaddr);
  delay(2000);
}

/*
MQTT再接続用の関数
*/
void reConnect() {
  while (!mqttClient.connected()) {
    Serial.print("Attempting MQTT connection...");
    
    // MQTT broker に接続する
    mqttClient.setServer(mqttHost, mqttPort);

    if (mqttClient.connect("M5Stack")) {
      mqttClient.subscribe(topic_sub, 0);
      Serial.println("subscribe!");
    } else {
      Serial.print("failed, rc=");
      // http://pubsubclient.knolleary.net/api.html#state に state 一覧が書いてある
      Serial.print(mqttClient.state());
      Serial.println("try again in 1 seconds");
    }

    delay(1000);
  }
}
/*
MQTT受信した時のコールバック関数
*/
void callback(char* topic, byte* payload, unsigned int length) {
  String temp = "";
  for (int i = 0; i < length; i++) {
    temp += String((char)payload[i]);
  }
  int len = temp.length();
  char temp_c[len + 1];
  temp.toCharArray(temp_c, len + 1);
  char* rece_data = temp_c; // 受信データ
  Serial.println(rece_data);
}
/*
MQTT送信関数（例）
*/
void sendMqtt(String sendData) {
  Serial.println(sendData);
  char sendMoji[50];
  sendData.toCharArray(sendMoji, 50);
  mqttClient.publish(topic_pub, sendMoji);
}

void setup() {
  Serial.begin(115200);
  M5.begin();

  init3G(); // 3G通信

  // MQTT broker へ接続
  mqttClient.setServer(mqttHost, mqttPort);
  mqttClient.connect("M5Stack");
  mqttClient.subscribe(topic_sub, 0);
  mqttClient.setCallback(callback); // MQTT受信したときのコールバック関数
}

void loop() {
  M5.update();

  // MQTT接続が切れていないか確かめる
  if (!mqttClient.connected()) {
    reConnect(); // 切断されていた場合は再接続
  }
  mqttClient.loop();  //ループを継続的に呼び出して、サーバーへの接続を確立する．
}

上記のプログラムではinit3G()関数を作成し，setup()時に実行することで3G接続設定を行っています．
MQTT通信設定でPubSubClient mqttClient(ctx);にTinyGsmClient ctxをセットすることで3G回線を用いたMQTT通信ができます．
MQTTのコールバック関数(callback)は本マイコンのトピックにデータが送信された際に呼び出されます．
sendMqtt()関数はMQTT送信のための関数です．例として記載しています．

M5UnitVから取得した画像データの加工

3G回線を使用してデータを送信する場合，データ容量に制限があるため（実際の原因は不明）画像データ等の大きいデータは分割して送信する必要 があります．そのため，ここではM5UnitVから取得した画像データを分割し，送信するようなプログラムを示します．

また，M5StackによるM5UnitVの画像データ取得については以下のリンクから詳細を確認できます．

上記のプログラムに付け加える形でお願いします．
付け加えたプログラムは+++...で囲ってあります．

#include <M5Stack.h>
#include <string.h>

/* 3Gボード設定 */
...

/* MQTT通信の設定 */
...

/*
MQTT関連
*/
...

+++++++++++++++++++++++++++++++++
/*
画像関連
*/
HardwareSerial serial_ext(2); // M5UnitVと通信するため
static const int RX_BUF_SIZE = 20000;
static const uint8_t packet_begin[3] = { 0xFF, 0xD8, 0xEA };
typedef struct {
  int length;
  uint8_t* buf;
} jpeg_data_t;
jpeg_data_t jpeg_data; //写真データ
int waitPhotoShot = 0; //非同期関数PhotoShotを実行している間，void loop()を止めておく変数
+++++++++++++++++++++++++++++++++

/*
3G通信
*/
void init3G() {
  ...
}

/*
MQTT再接続用の関数
*/
void reConnect() {
  ...
}
/*
MQTT受信した時のコールバック関数
*/
void callback(char* topic, byte* payload, unsigned int length) {
  ...
}
/*
MQTT送信関数
*/
void sendMqtt(String sendData) {
  ...
}

+++++++++++++++++++++++++++++++++
/*
画像送信関数（MQTT経由）
*/
void photoShot(void* parameters){
  /*
    ・この関数は、静止画像を含む jpeg_data_t 構造体へのポインタを引数として受け取ります．
    ・jpeg_data_t 構造体から画像ファイルのサイズとデータを取得します．（引数）
    ・画像ファイルをメモリにコピーし，std::vector<uint8_t> データ型の配列に変換します．この際，配列は700バイト単位で分割されます．
    ・分割された各配列の要素を，別々の uint8_t 型の配列にコピーします．
    ・分割された各配列をMQTTブローカーに送信します。送信後、2秒待機します．
    ・すべての分割配列が送信された後，"end" というメッセージをMQTTブローカーに送信します．これにより，受信側がメッセージ受信が完了したことを知ることができます．
    ・waitPhotoShot 変数を0に設定し，メインループ関数が再開されます．
  */

  jpeg_data_t* jpegData = (jpeg_data_t*) parameters;
  int jpegLen = jpegData->length;
  Serial.println(jpegLen); //画像ファイルの大きさ

  //画像ファイルをコピー
  uint8_t Data[jpegLen];
  memcpy(Data, jpegData->buf, jpegLen);

  std::vector<uint8_t> v;
  for (int i = 0; i < sizeof(Data); i++) {
    v.push_back(Data[i]);
  }

  const int CHUNK_SIZE = 700;
  int numOfChunks = (v.size() - 1) / CHUNK_SIZE + 1;
  std::vector<uint8_t> chunk[numOfChunks];
  for (int k = 0; k < numOfChunks; ++k) {
    //次の`CHUNK_SIZE`要素のセットの範囲を取得します
    auto start_itr = std::next(v.cbegin(), k * CHUNK_SIZE);
    auto end_itr = std::next(v.cbegin(), k * CHUNK_SIZE + CHUNK_SIZE);

    //サブchunkにメモリを割り当てます
    chunk[k].resize(CHUNK_SIZE);

    //最後のサブchunkを処理するコード
    //含まれる要素が少ない
    if (k * CHUNK_SIZE + CHUNK_SIZE > v.size()) {
      end_itr = v.cend();
      chunk[k].resize(v.size() - k * CHUNK_SIZE);
    }

    //入力範囲からサブchunkに要素をコピーします
    std::copy(start_itr, end_itr, chunk[k].begin());
  }

  for (int i = 0; i < numOfChunks; i++) {
    Serial.println("sendFile");
    Serial.println(chunk[i].size()); //分割ファイルの大きさ
    uint8_t data[chunk[i].size()];
    for (int j = 0; j < sizeof(data); j++) {
      data[j] = chunk[i][j];
    }
    int result = mqttClient.publish(topic_pub_pic, data, sizeof(data), false);
    while(result == 0){
      reConnect();
      result = mqttClient.publish(topic_pub_pic, data, sizeof(data), false);
    }
    Serial.println(result);
    vTaskDelay(2000);
  }
  int result = mqttClient.publish(topic_pub_pic, "end", false);
  while(result == 0){
    reConnect();
    result = mqttClient.publish(topic_pub_pic, "end", false);
  }
  Serial.println(result);
  waitPhotoShot = 0; //停止していたメインLoop関数を再開
  vTaskDelete(NULL);
}
+++++++++++++++++++++++++++++++++


void setup() {
  ...

+++++++++++++++++++++++++++++++++
  jpeg_data.buf = (uint8_t*)malloc(sizeof(uint8_t) * RX_BUF_SIZE);
  jpeg_data.length = 0;
  Serial1.begin(115200, SERIAL_8N1, 21, 22);  // Grove
+++++++++++++++++++++++++++++++++
}

void loop() {
  ...

+++++++++++++++++++++++++++++++++
  // ボタンが押されたとき
  if (M5.BtnA.wasPressed()) {
    char* inByte = "photoShot"; // M5UnitVに送るメッセージ内容
    Serial1.write(inByte); // M5UnitVにメッセージを送信
  }

  // M5UnitVから画像データが送られたとき
  if (Serial1.available()) {
    uint8_t rx_buffer[10];  //unit8_tは符号なし8bit整数型が表現できる整数型
    int rx_size = Serial1.readBytes(rx_buffer, 10);
    if (rx_size == 10) {  //packet receive of packet_begin
      if ((rx_buffer[0] == packet_begin[0]) && (rx_buffer[1] == packet_begin[1]) && (rx_buffer[2] == packet_begin[2])) {
        //image size receive of packet_begin
        jpeg_data.length = (int)(rx_buffer[4] << 16) | (rx_buffer[5] << 8) | rx_buffer[6];
        int rx_size = Serial1.readBytes(jpeg_data.buf, jpeg_data.length);
        TaskHandle_t taskHandle;
        waitPhotoShot = 1;
        xTaskCreatePinnedToCore( // 非同期関数photoShotを実行
          photoShot,
          "photoShot",
          12000, // 処理に割り当てるメモリ．この値が画像データより小さいとエラーが出る
          &jpeg_data, // 非同期関数photoShotの引数になる値
          2,
          &taskHandle,
          1
        );
        while(waitPhotoShot == 1){ // 非同期関数photoShotが実行している間loop()を止める
          delay(5000);
        }
      }
    }
  }
+++++++++++++++++++++++++++++++++
}

上記のプログラムでは以下の流れで処理が行われています．

1．M5Stackのボタンを押すと，M5UnitVへ撮影するよう命令が送られる

2．命令を受け取ったM5UnitVが撮影し，M5Stackへ画像データを送る

3．M5Stackは受信した画像データを非同期関数photoShotの引数に乗せ，photoShotを実行する

4．photoShot関数では画像データを700byte単位で分割し，サーバへ順次送る

5．すべての画像データを送った後に文字列"end"のデータをサーバに送ることで，サーバはすべての画像データが送られたことを確認し，分割データを統合し画像データを復元して処理を終える

また，作成したphotoShot関数は非同期関数ですが，この関数を実行している間にloop()が実行されると予期しない動作が発生する可能性があるため，loop()はwaitPhotoShot変数を用いて制御しています．waitPhotoShot変数が1の間はloop()をdelay()で停止し、photoShot関数の実行が完了するとwaitPhotoShot変数が0になりloop()が再開されます．

非同期関数xTaskCreatePinnedToCoreの詳細については以下のリンクから確認できます．

Node.jsの受信方法

以下に，Node.jsを使用したMQTT受信のプログラムを示します．

class MQTT {
  _topic = '**********'  // マイコンチャネル
  _topic2 = '**********' // サーバチャネル
  mqttOpt = { port: ** } // ポート
  client = mqtt.connect('****', this.mqttOpt) // MQTTアドレス

  mqtt_connect(){ // MQTT接続
    this.client.on('connect', () => {
      this.client.subscribe(this._topic2, (error) => { //サーバチャネル
        if (error) {
            console.error(`subscribe ${this._topic2} error.`)
            this.client.end()
            return
        }
        console.log(`subscribe ${this._topic2} success.`)
      })
    })
  }

  mqtt_receve(){
    let fileCnt = 10 //tempfile名
    this.client.on('message', (topic, message) => {
      console.log("M5Atomからのの通信", topic, message);
        if(message.toString() == "end"){ //送信完了
          console.log("end")
          const folderPath = './public/images/tempFiles/'
          const fileNames = fs.readdirSync(folderPath)
          var file = ""

          // ファイル名を昇順でソート
          fileNames.sort()
          fileNames.forEach((fileName) => {
            const filePath = `${folderPath}/${fileName}`
            const data = fs.readFileSync(filePath, 'binary')
            console.log(Buffer.from(data))
            file += Buffer.from(data)
          })

          var td = new Date()
          const y = td.getFullYear()
          const m = td.getMonth() + 1
          const d = td.getDate()
          const h = td.getHours()
          const mi = td.getMinutes()
          const today = `${y}-${m}-${d}-${h}:${mi}`
          const savePath = `./public/images/${today}.jpg`
          fs.writeFileSync(savePath, file, { encoding: 'binary', flag: 'w', mode: 0o666 }, (err) => {
            if (err) {
              console.error('Failed to write file', err)
            } else {
              console.log('File saved:', savePath)
            }
          })

          // tempファイルの削除
          fileNames.forEach((fileName) => {
            const filePath = `${folderPath}/${fileName}`
            fs.unlinkSync(filePath)
          })

          fileCnt = 10
        }else{
          fs.writeFileSync('./public/images/tempFiles/'+fileCnt+'.txt', message, 'binary');
          fileCnt += 1
        }
    })
  }

}
const _mqtt = new MQTT()
_mqtt.mqtt_connect()
_mqtt.mqtt_receve()

this.client.on('message', (topic, message))により受信し，メッセージ内容が "end" でなければ一時ファイルを作成しバイナリデータを保存します．メッセージ内容が "end" であれば保存してある一時ファイルに格納されたバイナリデータをまとめてjpegファイルとして保存します．
また，ファイルパスなどは各自のディレクトリに合わせて適切に設定してください．

おわりに

正直，M5Stack用3G拡張ボードを用いて1度で画像データが送れない原因は他にあるかもしれません．私の場合は，700byteくらいを超えると送信できなくなったため，このような疑似的な分割送信の方法を取りました．
また，サーバ，M5UnitV，M5Stackと関連する機器が多くなってしまったため，プログラムを簡潔に説明することができませんでした．そのため，分からない点や不明な点がありましたら，どうぞコメントより質問してください．

おわり