[AWS IoT] RaspberryPiからMQTTで画像をS3にアップロードしてみました

1 はじめに

CX事業本部の平内（SIN）です。

今回は、AWS IoTのメッセージブローカーを経由して、MQTTでRaspberryPiからS3に画像を送信する仕組み試してみました。

この仕組みでは、デバイス側に、S3アクセスのためのクレデンシャル情報は必要ありません。

また、送信先となるS3のバケット名やファイル（プレフィックス）名は、デバイス側で指定する仕組みとしてみました。

想定する利用場面以下のとおりです。

• RaspberryPiに接続したカメラで動画を撮影
• 動画をフレーム単位で送信
• IoT Coreでは、ルールでLambdaを起動しS3に蓄積
• S3の画像ファイルを動画ファイルにマージする

なお、赤色の点線内は、今回、未実装です。RaspberryPiからは、予め用意された画像ファイルを読み込んで送信しています。

2 画像ファイル

送信に使用した画像ファイルです。

先日行ってきた、うちの近くにある犬のだがし屋さん(旧：犬の焼きいも屋さん)の写真です。

640 * 480 で 97KByte となっています。

3 ルール

定義されたルールです。

指定したトピック（ここでは、topic_1となっています）を受信した場合に、Lambda(BinaryUploadTest)に送っています。

4 Lambda

ルールから起動されるLambdaのコードです。

受信したPayloadから、S3のバケット名、キー名及び、エンコードされた画像を取得し、S3にUploadしています。

S3へのUpload権限は、別途、このLambda関数に付与されています。

import json
import os
import boto3
from datetime import datetime
from io import BytesIO
from base64 import b64decode

def lambda_handler(event, context):
    print("event",event)

    img = event["img"]
    bucket = event["bucket"]
    key = event["key"]

    # デコード base64 => binary
    img_binary = b64decode(img)

    s3 = boto3.resource('s3')
    obj = s3.Object(bucket, key)
    obj.put(Body=BytesIO(img_binary),ContentType="image/jpg")

    return 0

5 クライアント

RaspberryPi上で動作しているコードは、以下のとおりです。

payloadには、バケット名、キー名、及び、Base64でエンコードされた画像が含まれています。

コード側で、バケット名及び、キー名を指定していますが、S3への権限は、Lambdaに付与が必要です。

index.py

from mqtt import Mqtt
import time
import datetime
import json
import base64

endPoint = "xxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com"
port = 8883
rootCA = "./cert/root-CA.crt"
privateKey = "./cert/private.key"
certificate = "./cert/cert.pem"

clientId = "client_id"
topic = "topic_1"

bucket = "binary-upload-test-2020-08-05"

def main():
    mqtt = Mqtt(endPoint, port, rootCA, privateKey, certificate, clientId)

    # 送信する画像ファイルを読み込む
    img = open("./photo.jpg", 'rb').read()
    
    while True:
        # 現在時間を使用して、ファイル名を生成
        dt = datetime.datetime.fromtimestamp(time.time())
        key = "{}{}.jpg".format(dt.strftime('%H%M'), dt.strftime('%S'))
        
        # エンコード binary => base64
        uploadImg = base64.b64encode(img).decode("utf-8")
        
        # 送信
        mqtt.publish(topic, payload=json.dumps({
            "img": uploadImg,
            "bucket": bucket,
            "key": key
        }))
        
        print("publish: {}".format(key))
        time.sleep(5)

main()

mqtt.py

# MQTT pub/subの基本クラス

from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
from log import logger

class Mqtt():
    def __init__(self, endPoint, port, rootCA, privateKey, certificate, clientId):
        logger.debug("Mqtt.__init__()")
        self.__client = AWSIoTMQTTClient(clientId)
        self.__client.configureEndpoint(endPoint, port)
        self.__client.configureCredentials(rootCA, privateKey, certificate)
        self.__client.configureAutoReconnectBackoffTime(1, 32, 20)
        self.__client.configureOfflinePublishQueueing(-1)
        self.__client.configureDrainingFrequency(2)
        self.__client.configureConnectDisconnectTimeout(10)
        self.__client.configureMQTTOperationTimeout(5)
        self.__client.connect()
    
    def subscribe(self, topic, callback):
        logger.debug("Mqtt.subscribe()")
        self.__client.subscribe(topic, 1, callback)

    def publish(self, topic, payload):
        result = self.__client.publish(topic, payload, 1)

6 実行

先のコードを実行すると、指定したバケットにファイルが保存されることが確認できます。

$ python3 index.py
2020-08-05 09:15:31,764 - Sample - DEBUG - Mqtt.__init__()
publish: 091532.jpg
publish: 091537.jpg
publish: 091542.jpg
publish: 091547.jpg

ダウンロードして、内容を確認している様子です。

7 最後に

今回は、MQTTでS3への画像ファイル送る仕組みを試してみました。Payloadは、自由に設計できるので、現在は、バケット名やキー名のみが付加されていますが、他の情報を追加して制御に利用するカスタマイズは簡単だと思います。