M5StackからAzure IoT HubにMQTTする

概要

Arduinoを使ってM5StackからAzure IoT HubにMQTT接続してみました。
MQTTとはなんぞや？というところから始めたので覚え書きです。

Arduinoのプログラミングも始めたばかりの初心者です。

MQTT接続について

IoTデバイスとクラウドの間で双方向にメッセージをやり取りするための軽量なプロトコルです。
特にクラウドからデバイスへのメッセージにリアルタイムに応答するためには、MQTT接続が必須です。

今回利用するAzure IoT Hubは、MQTTブローカーとして動作します。
MQTTにより、デバイスからのセンサーデータデータを高頻度でIoT Hubに送信(パブリッシュ)したり、IoT Hubからの要求を購読(サブスクライブ)して、リアルタイムに処理や応答を行うことがでるようになります。

HTTPとの比較

HTTP
リクエスト/レスポンスモデル
デバイスがサーバーにデータを要求するたびにコネクションを開始し、応答を受け取ったらコネクションを終了

MQTT
パブリッシュ/サブスクライブモデル
デバイスが常にサーバーとの接続を保持してメッセージを受け取ることができるようにする。

手段の検討

今回、IoT Hubへの接続はArduino用のMQTTクライアントである「PubSubClient」を使用しました。
他には「azure-sdk-for-c-arduino」と呼ばれるMicrosoft公式のSDKを使った方法がありましたが、前提知識のない自分にはサンプルコードがさっぱり理解できませんでした。
そこで、MQTTの仕組みを理解するために自前で実装してみました。

後で確認すると、Arduino用のSDKは組み込みデバイス向けのものがベースとなっており、低レイヤーすぎて使うメリットが薄そうでした。
https://github.com/Azure/azure-sdk-for-c-arduino/tree/main
MicroPythonの方が圧倒的に短いコードでかけそうです。

開発環境

• VSCode

• Azure IoT Hub拡張機能

• PlatformIO拡張機能

• Azure IoT Hubリソースの作成とデバイスの作成

• M5Stackデバイス

Azure IoT HubとMQTT接続

MQTT接続を行うにあたり必要な接続パラメータの調査です。
ユーザー名、パスワード、トピックなどの情報が必要です。
基本的なことは全てドキュメントに書いてあります。

接続情報

ブローカーサーバー、デバイスId、ユーザー名、パスワードの情報が必要になります。
VSCodeでAzureIoT Hub拡張をインストールし、デバイスを右クリックすると簡単に取得できます。
パスワードはSAS Tokenとなります。これも拡張機能から取得するのが楽です。

下記にコードを載せていますが、パラメータ類はAzIoTConfig.hファイル内に記述します。
以下の部分はご自身の環境に合わせて設定します。

AzIoTConfig.h

#define CONFIG_MQTT_BROKER      "yout-host-name.azure-devices.net"
#define CONFIG_DEVICE_ID   "YourDeviceName"
#define CONFIG_USERNAME    "yout-host-name.azure-devices.net/YourDeviceName/?api-version=2021-04-12"
#define CONFIG_SAS_TOKEN   "Your-SAS-Token"

後述しますが、ルート証明書の登録も必要です。
DigiCert Global Root G2のPEMレコードを以下から取得します。
https://www.digicert.com/kb/digicert-root-certificates.htm

MQTT プロトコルを直接使用するには、クライアントは TLS または SSL 経由で接続する "必要があります"。 この手順をスキップすると、接続エラーで失敗します。

代表的な機能とエンドポイント

MQTTでメッセージをやり取りするにあたり、機能ごとのエンドポイントを整理しておきます。
MQTTではトピックと呼ばれる文字列でメッセージの種類の識別を行います。
それぞれの機能にメッセージをパブリッシュしたりサブスクライブするときにトピック名の設定が必要です。

以下はIoT Hubの主な機能とトピックです。

①メッセージのやり取り

トピックのDeviceIdはIoT HubのデバイスIdを設定します。

デバイスからクラウドへのメッセージの送信(パブリッシュ)

const String _mqttPubTopic = "devices/{DeviceId}/messages/events/";
mqttClient.publish(_mqttPubTopic.c_str(), payload);

クラウドからデバイスへのメッセージ受信(サブスクライブ)

const String _mqttSubTopic = "devices/{DeviceId}/messages/devicebound/#";
mqttClient.subscribe(_mqttSubTopic.c_str());

#はワイルドカードのようなもので、devicebound/に続けて要求本文が渡されるので、全てのメッセージを購読対象とします。
また、MQTTクライアント初期化時にコールバック関数を登録しておくことで、受信時の処理を実装できます。

②プロパティのやり取り(デバイスツイン)

クラウド側、デバイス側でプロパティを操作し、相手側で参照したり変更通知を受け取ったりする機能です。

デバイスツインプロパティをパブリッシュ

// デバイス側プロパティの更新
mqttClient.publish("$iothub/twin/PATCH/properties/reported/?$rid=1" , data_json);

③メソッドの実行(ダイレクトメソッド)

クラウド側からメソッド(任意な識別子)を実行し、デバイス側で受信、処理を行って応答する機能です。
→Rebootなどの命令を想定。

ダイレクトメソッドのサブスクライブ

mqttClient.subscribe("$iothub/methods/POST/#");

クラウド側でダイレクトメッセージを実行すると、$iothub/methods/POST/{method-name}/?$rid={request-id}に対して送信されます。
method-nameはクラウド側で設定する任意のメソッド名です。
ridはIoT Hub側で付加されます。

ダイレクトメソッドに応答

// 応答メッセージをメッセージをパブリッシュ
String resTopic = "$iothub/methods/res/200/?$rid=" + rid;
mqttClient.publish(resTopic.c_str(), data_json);

ridはサブスクライブ時のridと同じものを設定してパブリッシュします。
クラウド側からREST APIを使ってHTTPSでダイレクトメソッドを呼び出すと、
上記で設定する応答メッセージがそのまま応答本文になります。

プログラミング

ArduinoでM5Stack用のプログラムを作っていきます。
VSCode+PlatformIOです。

完成版のコードを以下のリポジトリにアップしました。
下記ではポイントを絞って解説します。

上記コードを使用する前に、「AzIoTConfig.h」にパラメータ設定が必要です。
また、M5Stack本体にENV IIIセンサーの取り付けが必要です。

関係するライブラリ

主なライブラリは以下です。

// MQTTクライアントです。
#include <PubSubClient.h>

// MQTTサーバーへの接続にルート証明書を使用するために必要
#include <WiFiClientSecure.h>

主なコード

１．Wi-Fi接続

割愛します。
まずはWi-Fiに接続します。

２．変数、パラメータの初期化

設定は「AzIoTConfig.h」ファイルに行い、パラメータを読み込みます。
SASトークンは、VSコードの拡張機能「Azure IoT Hub」からGenerate SAS Tokenから取得します。

root_ca[] はルート証明書pemレコードです。
ドキュメントにある通り、DigiCert Global Root G2のものを貼り付けています。

// MQTTの設定
// クライアントの初期化
WiFiClientSecure wifiClient;
PubSubClient mqttClient(wifiClient);

// 接続パラメータの取得
const char _mqttBroker[]      = CONFIG_MQTT_BROKER;
const char _mqttDeviceId[]  =  CONFIG_DEVICE_ID;
const int16_t _mqttPort = CONFIG_MQTT_PORT;
const char _mqttUsername[] = CONFIG_USERNAME;
const char _mqttPassword[] = CONFIG_SAS_TOKEN;
const String _mqttPubTopic = "devices/" + String(_mqttDeviceId) + "/messages/events/";
const String _mqttSubTopic = "devices/" + String(_mqttDeviceId) + "/messages/devicebound/#";

// ルート証明書の設定
PROGMEM const char root_ca[] = R"(-----BEGIN CERTIFICATE-----
省略します
-----END CERTIFICATE-----
)";

３．mqttClientの初期化

mqttClientの初期化を初期化し、メッセージ受信時に実行されるコールバックを登録します。
また、wifiClientSecureライブラリを使ってルート証明書を登録します。

mqttCallbackコールバック関数には、メッセージを受け取ったときの処理を記述します。

setupとmqttCallback

void setup() {
//省略

  // MQTT接続設定  
  // MQTTサーバーの設定
  mqttClient.setServer(_mqttBroker,_mqttPort);
  // コールバック関数の登録
  mqttClient.setCallback(mqttCallback);
  // ルート証明書登録
  wifiClient.setCACert(root_ca);

//省略
}

// コールバック関数
// サブスクライブしたトピックに一致するメッセージを受け取ったときに実行される関数
  void mqttCallback(char* topic, byte* payload, unsigned int length) {
    //省略
  }

４．MQTT接続とトピック登録

MQTTサーバーに接続します。
接続後に、サブスクライブするトピックを登録します。
複数ある場合は複数登録します。

  // MQTTに接続します
  void ConnectMQTT(){

    // 省略 //
    
    // MQTT接続
    if (mqttClient.connect(_mqttDeviceId ,_mqttUsername ,_mqttPassword)) {
        Serial.println("MQTTサーバーに接続しました。");
    }
    
    // 省略 //
    
    // サブスクライブするトピックの設定
    if(mqttClient.connected() == true){
      _topSubLabel->UpdateLabel("IoT Hubからのイベントを購読します。");
      // C2Dメッセージ
      mqttClient.subscribe(_mqttSubTopic.c_str());
      // ダイレクトメソッド
      mqttClient.subscribe("$iothub/methods/POST/#");
      // DeviceTwinのReportedプロパティ変更通知(デバイス側プロパティ)
      mqttClient.subscribe("$iothub/twin/res/#");
      // DeviceTwinのDesiredプロパティ変更通知(クラウド側プロパティ)
      mqttClient.subscribe("$iothub/twin/PATCH/properties/desired/#");
   
    }
    
    // 省略 //

  }

コールバック関数を登録しただけではメッセージを受け取れません。
必ずサブスクライブするトピックを設定する必要があります。
また、必ずサーバーへの接続後にトピックをサブスクライブします。

５．デバイスからクラウドへのメッセージ送信

IoT Hubへセンサーデータなどのテレメトリを送信します。
トピックdevices/{DeviceId}/messages/events/ で送信します。

  // IoT HubにD2Cメッセージを送信します。
  void SendD2CMessage()
  {

    DynamicJsonDocument json(200);
    char data_json[200];
    json["Message"] = "Message From M5Stack";
    size_t payloadSize = serializeJson(json, data_json);

    // D2Cメッセージをパブリッシュ
    mqttClient.publish(_mqttPubTopic.c_str(), data_json);
  }

６．クラウドからデバイスへのメッセージ受信

IoT Hubからのメッセージを受信すると、下記のコールバック関数が実行されます。
受信したトピック名はtopic引数に入ります。
全てのトピックで同じ関数が実行されるので、イベントごとに分岐処理が必要です。

  // MQTTコールバック関数
  //MQTT Brokerからメッセージを受信されたときのコールバック関数です。
  void mqttCallback(char* topic, byte* payload, unsigned int length) {

    _topLabel->UpdateLabel("IoT Hubからメッセージを取得しました。");

    // トピックを文字列に変換
    String topicName = topic;

    // ペイロードを文字列に変換
    String str = "";
    for (int i = 0; i < length; i++) {
        Serial.print((char)payload[i]);
        str += (char)payload[i];
    }
    Serial.print("\n");
    Serial.println("Received. topic=" + topicName);

    // トピックの種類ごとに分岐
    if(topicName.startsWith("$iothub/twin/res/204/")){
      // デバイス側プロパティを更新した場合

      _topSubLabel->UpdateLabel("デバイスプロパティが正常に更新されました。");
      _bottomLabel->UpdateLabel(topicName);
      return;      
    }else if(topicName.startsWith("$iothub/twin/PATCH/properties/desired/")){      
      // サービス側プロパティを更新した場合
      
      _topSubLabel->UpdateLabel("サービス側のプロパティが正常に更新されました。");
      _bottomLabel->UpdateLabel(topicName);
      return;

    }else if(topicName.startsWith("$iothub/methods/POST/")){
      // ダイレクトメソッドを受信した場合

      _topSubLabel->UpdateLabel("ダイレクトメソッドが呼び出されました。");
      _bottomLabel->UpdateLabel(topicName);

      // ここに温度などを返す
      RespondToDirectMethod(topicName, str);
      return;
    }
    
    // D2Cメッセージを取得した場合
    _topSubLabel->UpdateLabel("D2Cメッセージを取得しました。");
    _bottomLabel->UpdateLabel(str);

  }

７．ダイレクトメソッドへの応答

クラウド側からダイレクトメソッドが実行されたときの応答を行います。
下記のコードでは、「GetTelemetry」メソッド名で実行されると、応答メッセージとして現在のセンサーデータを返しています。
送信トピックにある$rid=は、ダイレクトメソッド受信時のトピックに含まれているridを抽出し、同じridで応答する必要があります。

  void RespondToDirectMethod(String topic, String &reqBody){
  
  // topic「$iothub/methods/POST/MethodName/$rid=1」からMethodNameを抜き出します。
  // MethodNameの抜き出し
  String methodName = topic;
  methodName.replace("$iothub/methods/POST/","");
  methodName = methodName.substring(0,methodName.indexOf("/"));
  Serial.println("Method Name=" + methodName);

  // ridの抜き出し
  String rid = topic.substring(topic.lastIndexOf("=")+1);
  Serial.println("rid=" + rid);

  String resTopic = "$iothub/methods/res/200/?$rid=" + rid;

  DynamicJsonDocument json(200);
  char data_json[200];  
  size_t payloadSize;

  if(methodName == "GetTelemetry"){
    String batteryLevel = String(M5.Power.getBatteryLevel());
    String temperature = GetTempertureFromSencsor();
    String humidity = GetHumidityFromSencsor();
    json["BatteryLevel"] = batteryLevel +"%";
    json["Temperture"] = temperature;
    json["Humidity"] = humidity;        
    payloadSize = serializeJson(json, data_json);
  }else{
    json["Message"] = "M5Stack is responding to direct methods.";

  }

  // 応答メッセージをメッセージをパブリッシュ
  mqttClient.publish(resTopic.c_str(), data_json);
}

ここでの応答メッセージは、
クライアント側からREST APIを使ってPOSTしたときのレスポンスボディとなります。
今回の例では以下のようになります。
Status:200はトピック内で任意なものを指定します。

{
  "status": 200,
  "payload": {
    "BatteryLevel": "100%",
    "Temperture": "18℃",
    "Humidity": "46 %"
  }
}

クラウド側のクライアント

IoT Hubを経由してC2Dメッセージを送ったり、ダイレクトメソッドを実行するにはクラウド側の処理の実装が必要です。
こちらはSDKを使用する方法と、REST APIを使用する方法があります。

以下に、Power AppsやPower Automateからローコードで操作きるようカスタムコネクタを作成しました。
カスタムコネクタの作成→OpenAPI定義から下記ファイルをアップします。

アプリ設定方法の参考

あとがき

デバイス側のプログラムはMicroPythonライブラリやUIFlowを使った方が圧倒的に楽です。
Arduinoの方がライブラリも豊富でカスタマイズ情報も多いのですが、デバイスの機能をセンサーデータの送信に限定すると割り切り、UIFlowでちゃちゃっと作る方が生産性は高そうです。