【MQTT】AWS IoT CoreとSoracom Beamを使ってデバイスとお喋りしてみた

はじめに

このアーキテクチャで実装する理由ですが、すでにSMSを使ってデバイスに対して通知をする仕組みがあるのですが、通信が安定しないことや、他社のサービスを使っていることもあって調査がしにくいという問題がありました。そこでMQTTを使って管理できるようにして安定化を図る狙いがあります。

要件

• IoT Coreからpublishされるメッセージを元に、デバイス側で処理を実行させる
• セキュリティ面を考慮してSoracom Beamを利用する

利用技術

• AWS IoT Core(以下 IoT Core)
• SORACOM Beam(以下 Beam)
• MQTT
• Python
• AWS SDK(AWSIoTPythonSDK)

内容

• デバイス側のソースコード（AWS SDK）についてがメイン
• 実装内容の注意点

事前準備

1. IoT Coreと、Beamの設定

まずはじめにIoT CoreとBeamの設定を行います。

SORACOMのガイドが非常にわかりやすいので参考にしてください。
手順に沿って作業すれば間違えないと思います。
※必要なのはモノの登録と、Beamの設定です。SNSの設定は不要です。

https://dev.soracom.io/jp/docs/aws_iot_guide_console/

完了すれば、デバイス側とIoT Coreで通信ができる状態となります。

1. SIMの設定
   pythonコードのなかでSIMからimsiを取得し、トピック名として利用します。
   そのためには「メタデータサービス設定」をオンにする必要があります。

設定方法は以下を参考にしてください
https://dev.soracom.io/jp/start/metadata/

実装！！

1. IoTデバイスの中で以下のpythonを実行します
2. IoT Coreのテストからmessageをpublish
3. コンソールに出力されたり、再起動ができたら成功です。

mqtt_sub.py

# -*- coding: utf-8 -*-
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
import time
import json
import subprocess
import requests

SORACOM_METADATA_URL = 'http://metadata.soracom.io/v1/subscriber'

def get_imsi(self):
    """get IMSI from SORACOM API"""
    subscriber = json.loads(requests.get(SORACOM_METADATA_URL).text)
    return subscriber["imsi"]

# MQTTが届いたら実行されます
def customCallback(client, userdata, message):
    message = json.loads(message.payload)["message"]
    if message == "REBOOT":
        reboot = "sudo reboot"
        subprocess.call(reboot, shell=True)
    else:
        print(message)
        pass


def main():
    try:
        # clientの識別子です。今回はなんでもよいです。
        mqttClient = AWSIoTMQTTClient('aws_iot_sub')

        # 通信ホストの設定です。beamのエンドポイントを設定しています。
        mqttClient.configureEndpoint('beam.soracom.io', 1883)

        # オフラインになった場合の設定です。
        # configureOfflinePublishQueueing(queueSize, dropBehavior=DROP_NEWEST)
        # queueSize: 1以上 ... queue size が引数の値に設定される
        # queueSize: 0 ... queue無効
        # queueSize: -1 ... queue size無制限
        # dropBehavior: queueがfullになったときの設定 設定しない場合はDROP_NEWEST
        mqttClient.configureOfflinePublishQueueing(0)

        # 実行頻度
        mqttClient.configureDrainingFrequency(2)  # Hz

        # clientのタイムアウト時間
        mqttClient.configureConnectDisconnectTimeout(10)  # sec

        # QoS 1の場合のタイムアウト時間
        mqttClient.configureMQTTOperationTimeout(5)  # sec

        # IoT Coreとの接続
        # keepaliveの設定もできます。デフォルトは600秒
        mqttClient.connect()

        imsi = get_imsi()

    except Exception as err:
        print(err)

    # 電波が悪い場合subscribeがExceptionとなりpythonが落ちるためtry/exceptでループする
    # exceptのerrを出力していないのは、subcribeのExceptionでメッセージを投げてこないので、出力しても空となるため
    # トピック名は実行している環境のみsubscribeできるようimsiとする
    while True:
        try:
            mqttClient.subscribe(str(imsi), 1, customCallback)
            time.sleep(1)
        except Exception as err:
            pass

if __name__ == "__main__":
    main()

まとめ

今回は再起動させるだけの紹介を行いましたが、shを実行するなどすれば、さまざまな処理を実行することが可能です。
本記事が登録しているadvent Calendar 2019で5日目に投稿されていた記事が、面白いことをされていたので参考になるかと思います。
やってみたい...

AWSIoTPythonSDKを利用しましたが、pahoを使用してもBeamと通信は可能でした。
今回の要件ではSDKを使うメリットは少なく、よりSDKが活躍するのは、ブローカーとしてIoT Coreを使うのではなく、シャドウの操作をする場合などになってくるかと思います。
ただ、SDKは内部でpahoが動いているのでAWSに特化しているSDKを採用しました。

他にもいい方法がないかご存知の方いれば教えて欲しいです。