Raspberry Pi でAWS IoT Core Lambda関数を未使用でPythonスクリプトを実行してみた！

f:id:iAtom:20201202142541j:plain

本記事では、コアデバイス（Raspberry Pi）にデプロイしたLambda関数ではないPython スクリプトを実行してAWS IoT Coreでメッセージ受信ができるかトライしてきます。

AWS構成図
Raspberry PiへSSH接続
環境構築
Pythonスクリプトファイル編集と実行
- pythonファイルの実行
AWS IoT テスト実行
最後に

Raspberry Piのインストールや証明書、ルートCA、キー情報は、「 Raspberry Pi でAWS IoT Greengrassを動かしてみた！（Part1）〜（Part5」をご確認ください。

AWS構成図

f:id:iAtom:20201203140450j:plain

Raspberry PiへSSH接続

MACターミナルからRaspberry PiへSSH接続をします。自分の環境に合わせてIPアドレスを変更してください。

    # ssh pi@192.168.10.10

環境構築

コアデバイス（Raspberry Pi）にMQTT通信ライブラリをインストールします。Lambda関数を使用した時は、SDKもAWSからデプロイしていましたが、今回はLambdaは使用しないため、MQTT通信するライブラリをインストールしておきます。

    $ pip3 install paho-mqtt

Python スクリプトファイル編集と実行

次のソースは下記サイトからコピーして証明書、ルートCA、endpointを変更しています。

aws.amazon.com

endpointはAWS IoTコンソールから確認ができます。

f:id:iAtom:20201130170539j:plain

それではPython ソースコード（mqtt_test.py)です。

証明書、ルートCA、キー情報は前回の設定をそのまま使用するので絶対パスで「/greengrass/certs/.....」としています。ここは個々の環境で変更してください。

from __future__ import print_function
import sys
import ssl
import time
import datetime
import logging, traceback
import paho.mqtt.client as mqtt

IoT_protocol_name = "x-amzn-mqtt-ca"
aws_iot_endpoint = "xxxxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com"  # 自分のAWS IoTのエンドポイントを設定。
url = "https://{}".format(aws_iot_endpoint)

ca = "/greengrass/certs/root.ca.pem"                          # -->  自分の環境に合わせたルートを設定。
cert = "/greengrass/certs/yyyyyyyyyy.cert.pem"          # -->  自分の環境に合わせた証明書を設定。
private = "/greengrass/certs/yyyyyyyyyy.private.key"  # -->  自分の環境に合わせたキー情報を設定。

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(sys.stdout)
log_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(log_format)
logger.addHandler(handler)

def ssl_alpn():
    try:
        #debug print opnessl version
        logger.info("open ssl version:{}".format(ssl.OPENSSL_VERSION))
        ssl_context = ssl.create_default_context()
        ssl_context.set_alpn_protocols([IoT_protocol_name])
        ssl_context.load_verify_locations(cafile=ca)
        ssl_context.load_cert_chain(certfile=cert, keyfile=private)

        return  ssl_context
    except Exception as e:
        print("exception ssl_alpn()")
        raise e

if __name__ == '__main__':
    topic = "test/date"
    try:
        mqttc = mqtt.Client()
        ssl_context= ssl_alpn()
        mqttc.tls_set_context(context=ssl_context)
        logger.info("start connect")
        mqttc.connect(aws_iot_endpoint, port=443)
        logger.info("connect success")
        mqttc.loop_start()

        while True:
            now = datetime.datetime.now().strftime('%Y-%m-%dT%H:%M:%S')
            logger.info("try to publish:{}".format(now))
            mqttc.publish(topic, now)
            time.sleep(1)

    except Exception as e:
        logger.error("exception main()")
        logger.error("e obj:{}".format(vars(e)))
        logger.error("message:{}".format(e.message))
        traceback.print_exc(file=sys.stdout)