この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
1 はじめに
CX事業本部の平内（SIN）です。
パナソニック様の提供するVieureka(ビューレカ)は、カメラ内のアプリで画像の解析などを行い、クラウドと連携することで多様なサービスが作成可能です。
Vieurekaでアプリを作成するためのSDKでは、多数のサンプルが梱包されていますが、今回は、その中から、TensorFlow Lite で物体検出するサンプル(object_detection_app)を使用させて頂いて、AWS IoT CoreへのMQTT送信を試してみました。
最初に、動作している様子です。画面の左は、推論結果をブラウザで確認しているもの（サンプルに付属）で、右が、AWSコンソールで、到着しているMQTTトピックを確認している画面です。
2 構成
構成は、以下のようなイメージです。Vieureka内に、証明書・秘密鍵を置き、AWS IoT Coreへ接続しています。
3 AWSIoTPythonSDK
送信には、AWSIoTPythonSDKを使用しています。 （※ AWS IoT Device SDK v2 for Pythonは、試してみたのですが、awscrtのセットアップに失敗してしまいました。改めて、確認してみます）
Adam(Vieurekaカメラのアプリ追加制御機構)では、Pythonの追加モジュールは、python/site-packagesに配置することで利用可能になっています。
アプリのpythonディレクトリでインストールしています。
% pip3 install AWSIoTPythonSDK -t site-packages
Collecting AWSIoTPythonSDK
Using cached AWSIoTPythonSDK-1.4.9-py3-none-any.whl
Installing collected packages: AWSIoTPythonSDK
Successfully installed AWSIoTPythonSDK-1.4.9
4 証明書/秘密鍵
IoT Coreへの接続のための証明書等は、dataに配置しています。
dataに配置したファイルは、下記のように利用できます。
cert = adam.AppDataDirPath + "/9f247e44ad-certificate.pem.crt"
key = adam.AppDataDirPath + "/9f247e44ad-private.pem.key"
root_ca = adam.AppDataDirPath + "/root-CA.crt"5 コード
サンプルコードに追加したのは、以下のとおりです。
- グローバル変数にMQTTオブジェクトを追加しています。
pymain.py
#################################
# グローバル変数定義 #
#################################
image_callback_counter = 0 #画像取得コールバックカウンタ
loop = None #イベントループオブジェクト
mqtt = None #Mqttオブジェクト- MQTTオブジェクトは、start_processing()で初期化しています。
pymain.py
def start_processing():
global loop
global mqtt
cert = adam.AppDataDirPath + "/9f247e44ad-certificate.pem.crt"
key = adam.AppDataDirPath + "/9f247e44ad-private.pem.key"
root_ca = adam.AppDataDirPath + "/root-CA.crt"
endpoint = "xxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com"
client_id = "client_id"
topic = "topic"
mqtt = Mqtt(endpoint, cert, key, root_ca, client_id, topic)
# ・・・略- image_callback()で推論を行っている、object_detection.detect_object()にMQTTオブジェクトを送っています。
pymain.py
def image_callback(img, clipInfo, dateTime):
# ・・・略
#オブジェクト検出実行
object_detection.detect_object(buf, mqtt)- detect_object()での推論結果を、MQTTオブジェクトでpublishしています。
ObjectDetection.py
def detect_object(self, img, mqtt):
# ・・・略
#オブジェクト検出
results = self.detect_objects(self.m_interpreter, image, self.m_threshold)
if len(results) > 0:
#JSON形式に変換する
json_data = self.create_json_format(results)
# ファイル操作オブジェクト
filecontrol = FileControl()
# JSONファイル保存
filecontrol.save_file(json_data, ConstDefine.JSON_FILE_PATH)
# httpIFオブジェクトimport
from __main__ import httpif
#クラウド保存するか
if httpif.get_cloud_send_flag() :
#クラウドへ送信する
httpif.put(json_data)
# AWS Core へのMQTT送信
mqtt.publish(json_data)
else:
pass- MQTTオブジェクト（クラス）です。
mqtt.py
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
class Mqtt:
def __init__(self, endpoint, cert, key, root_ca, client_id, topic):
self.__topic = topic
self.client = AWSIoTMQTTClient(client_id)
self.client.configureEndpoint(endpoint, 8883)
self.client.configureCredentials(root_ca, key, cert)
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 publish(self, payload):
self.client.publish(self.__topic, payload, 1)6 最後に
今回は、パナソニック様の提供するVieureka(ビューレカ)で、AWS IoT CoreへMQTT送信してみました。
Vieurekaでは、Pythonでの開発環境が提供されており、AWS SDKで簡単にクラウドとの連携が可能です。高機能なエッジデバイスとクラウド連携で実現可能なプロダクトの妄想が広がります。
2
