- 投稿日：2020-07-21T18:48:54+09:00
Greengrassでエッジ推論 - ビデオからの物体検知〜MQTT通知
AWS Greengrass でエッジ推論をやってみたかったので、TensorflowのObject Detection APIで、カメラに人が写ったらMQTTトピックで通知し、映像はKinesis Video Streamsに配信するという習作と実験を行いました。
作業はほぼ環境構築に終始してしまいましたが、Greengrassを使ってクラウド側からの設定によって、エッジ側にMLモデルと処理を同期して実行する一連の流れを理解することができました。
また、Greengrassの設定の要所も確認することができました。全体のイメージ
- AWS側は、Greengrassグループを作成し、エッジでTensorflow Object Detectionを行うLambda関数と、使用するモデルデータを関連付けします。また、ビデオをリモートで確認できるようにKinesis Video Streamsを準備します。
- エッジ側は、Greengrassコアを動作させますが、今回は練習なので、GreengrassCoreはDockerで作成します。Dockerfileがここからダウンロード出来ます。
- Lambdaの実行環境はPython3.7で、Gstreamerを使ってカメラの映像から、Tensorflowでの物体検知、Kinesis Video Streamsへの配信をパイプライン処理します。Gstreamerを使うのはビデオの処理をプラガブルに変更できるからで、ここで詳しい使い方は説明しませんが、情報は検索すればたくさん出てきます。
- カメラは、Gstreamerで処理するので何でも良いです。私は Tapo C200というAmazonで投げ売りされていたのを使っています。このカメラだとRTSPで覗けます。ラズパイやPCのカメラならV4L2を使うと良いです。
- 物体検知は、SSD Mobilenet ver.2のcocoデータセットのモデルをダウンロードして使用しました。検知するのは "person" だけとします。 検知できたらMQTTトピックで通知します。
- 図にはトピックをサブスクライブしてアラームする側のデバイスを描いていますが、これは次回にします。Echoに喋らせるか、AWS Chatbotでslackに通知したりする方が良いかもしれません。
作成したものは https://github.com/sabmeua/gg_gst このリポジトリにあります。以降の設定などは、このリポジトリを使って説明します。
ちなみに、私は普段Pythonを書きません。またTensorflowも今回初めて使用しました。その点ご容赦いただき、変な用法や実装がありましたらご指摘いただけるとありがたいです。
設定手順
Greengrassコアの準備
今回はDockerなのでイメージをビルドするだけです。Gstreamerと必要なプラグイン、Tensorflowなどをインストールしています。
docker-compose でビルドします。環境によってはけっこう時間がかかるかもしれません。Dockerボリュームとして証明書や設定ファイルをマウントするようになっていますが、これはGreengrassグループ作成時に準備します。$ docker-compose build物体検知処理
Gstreamerのプラグインとして、イメージ中の/myplugins/python/ に実装しています。
- /myplugins/python/object_detection.py : Tensorflowで映像からpersonを検知して、Gstreamerメタデータに入れるプラグイン
- /myplugins/python/awsiot_notify.py : Gstreamerメタデータから検知した情報をMQTTトピック
object/detectionにパブリッシュするプラグインGreengrass環境で特別なことと言うと、モデルのファイルは別段ダウンロードの処理などを書かなくても、設定したファイルをGreengrassコアがデプロイ時に配置してくれるので、
AWS_GG_RESOURCE_PREFIXという環境変数のパスでアクセスできることです。他には特別なことはしていません。PythonでのTensorflowの扱い方などは、こちらに詳しく解説されているので参考にしました。
Lambda関数の作成
LambdaでPython3.7の関数を作成します。アクセス権限など、その他の設定はGreengrass側の設定が採用されるので、ここでは名前以外は気にする必要はありません。
作成したら、以下のようにmakeでソースをfunction.zipにまとめてアップロードして関数を更新します。この src/lambda_function.pyが関数の本体です。
$ make lambda-archive cd src;\ zip -r ../function.zip greengrasssdk lambda_function.py adding: greengrasssdk/ (stored 0%) adding: greengrasssdk/__init__.py (deflated 32%) adding: greengrasssdk/utils/ (stored 0%) adding: greengrasssdk/utils/__init__.py (stored 0%) adding: greengrasssdk/utils/testing.py (deflated 62%) adding: greengrasssdk/client.py (deflated 53%) adding: greengrasssdk/Lambda.py (deflated 65%) adding: greengrasssdk/SecretsManager.py (deflated 67%) adding: greengrasssdk/IoTDataPlane.py (deflated 74%) adding: greengrasssdk/stream_manager/ (stored 0%) adding: greengrasssdk/stream_manager/__init__.py (deflated 40%) adding: greengrasssdk/stream_manager/util.py (deflated 81%) adding: greengrasssdk/stream_manager/data/ (stored 0%) adding: greengrasssdk/stream_manager/data/__init__.py (deflated 90%) adding: greengrasssdk/stream_manager/streammanagerclient.py (deflated 79%) adding: greengrasssdk/stream_manager/exceptions.py (deflated 72%) adding: lambda_function.py (deflated 56%)実装について簡単に説明します。
通常、Lambdaをデマンドで実行する場合は、イベント発生時に、設定したハンドラが呼び出されるわけですが、今回はGreengrassコアが起動したタイミングで実行し、制限時間でタイムアウトすることなく実行し続ける 長い存続期間の Lambda 関数というライフサイクルの設定で実行します。この場合、ハンドラとして設定されているlambda_handler(event, context)に処理を記述せず、ソースがロードされた時点で処理が開始されるように無限ループなり、スレッドを起動するように実装します。
今回はmain()に処理を記述しており、ここでGstreamerのイベントループを開始しています。
lambda_function.pyで行っているのはGstreamerパイプラインの起動のみです。物体検知やKVSへの送信といった処理のパイプラインは環境変数で設定できるようにしていて、Greengrass側のLambdaの実行設定の際に設定します。src/lambda_function.py#!/usr/bin/env python3 import os import sys import traceback import gi gi.require_version('Gst', '1.0') gi.require_version('GLib', '2.0') gi.require_version('GObject', '2.0') from gi.repository import GLib, GObject, Gst import logging logger = logging.getLogger(__name__) logging.basicConfig(stream=sys.stdout, level=logging.DEBUG) STREAM_NAME = os.environ.get('STREAM_NAME') IOT_CREDENTIAL_ENDPOINT = os.environ.get('IOT_CREDENTIAL_ENDPOINT') CERT_ID = os.environ.get('CERT_ID') KVS_CMD = f'kvssink stream-name={STREAM_NAME} framerate=15'\ ' aws-region=ap-northeast-1'\ ' log-config=/kvs_log_configuration'\ ' iot-certificate="iot-certificate,'\ f'endpoint={IOT_CREDENTIAL_ENDPOINT},'\ f'cert-path=/greengrass/certs/{CERT_ID}.cert.pem,'\ f'key-path=/greengrass/certs/{CERT_ID}.private.key,'\ 'ca-path=/greengrass/certs/root.ca.pem,'\ 'role-aliases=KvsCameraIoTRoleAlias"' DEFAULT_PIPELINE = 'videotestsrc ! clockoverlay auto-resize=false'\ ' ! videorate ! video/x-raw,format=I420,framerate=15/1'\ ' ! x264enc tune=zerolatency ! h264parse' def on_message(bus: Gst.Bus, message: Gst.Message, loop: GLib.MainLoop): mtype = message.type if mtype == Gst.MessageType.EOS: logger.info("End of stream") loop.quit() elif mtype == Gst.MessageType.ERROR: err, debug = message.parse_error() logger.error(err, debug) loop.quit() elif mtype == Gst.MessageType.WARNING: err, debug = message.parse_warning() logger.warning(err, debug) return True def main(): logger.info('Start gstream pipeline') Gst.init(sys.argv) cmd = os.environ.get('PIPELINE', DEFAULT_PIPELINE) if os.environ.get('USE_KVS', True): cmd += f' ! {KVS_CMD}' logger.info('Pipeline : %s', cmd) pipeline = Gst.parse_launch(cmd) bus = pipeline.get_bus() bus.add_signal_watch() pipeline.set_state(Gst.State.PLAYING) loop = GLib.MainLoop() bus.connect("message", on_message, loop) try: loop.run() except Exception: traceback.print_exc() loop.quit() pipeline.set_state(Gst.State.NULL) main() # This is a dummy handler and will not be invoked # Instead the code above will be executed in an infinite loop for our example def lambda_handler(event, context): returnGreengrassにLambda関数を設定する際、バージョンまたはエイリアスで指定する必要があるため、作成した関数のバージョンを発行して、これにエイリアス
devを設定します(キャプチャでは ver.2 になっています)。
Greengrassグループの設定
Greengrassグループの作成はAWSコンソールでAWS IoTのウィザードに沿って設定するだけですが、設定にはいくつかポイントがあります。
Greengrassグループ作成で
デフォルト作成を使用を選択します。
グループ名は
testとしました。Greengrassコアの名前はtest_Coreとなりますが、これもそのままOKしました。もう少し気の利いた名前にした方が良かったです。
証明書の配置
作成を実行すると、コア用の証明書、プライベートキー、パブリックキー、設定ファイルのtarballがダウンロードできます。これをdocker-composeのディレクトリの
certsとconfigに展開します。またCA 証明書も必要なので、ダウンロードして certs/root.ca.pem に保存します。
プライベートキーはLambdaを実行する ggc_user/ggc_group から参照されるので644にしておきます。$ ls -l certs/ config/ certs/: 合計 16 -rw-r--r-- 1 sabmeua sabmeua 1220 7月 13 11:35 0f0c0953bd.cert.pem -rw-r--r-- 1 sabmeua sabmeua 1679 7月 13 11:35 0f0c0953bd.private.key -rw-r--r-- 1 sabmeua sabmeua 451 7月 13 11:35 0f0c0953bd.public.key -rw-r--r-- 1 sabmeua sabmeua 1188 6月 22 14:03 root.ca.pem config/: 合計 4 -rw-r--r-- 1 sabmeua sabmeua 875 7月 13 11:35 config.jsonエッジ環境のローカルファイルシステムへのロギングの設定
コンソールで作成したGreengrassグループの
設定画面からローカルログ設定をinfoレベルで行うようにします。これによりLambdaの出力をファイルにロギングできます。
Lambda実行環境の設定
今回Lambda環境をDockerで実行するので、同じく
設定画面にて、デフォルトの Lambda 関数コンテナ化の設定をコンテナなしに変更します。実行ユーザはデフォルトのままggc_user/ggc_groupにしておきます。
Lambdaの設定
Greengrassグループの
Lambdaの画面で、Lambdaの追加->既存のLambdaの使用を選択して、先ほど作成した関数gg-gst-testを、バージョンにエイリアスdevを選択して追加します。
追加した関数の設定を編集します。
ここで、設定するのは2箇所です。まず、Lambdaのライフサイクルを存続期間が長く無制限に稼働する関数にするに変更します。もう一つは、タイムアウトを25に変更します。無制限と言いながらタイムアウトを設定するのは変な気がしますが、ここにかかれている通り、Lambda関数の呼び出し前に20秒のスリープがあるので、その間に終了しないようにするためです。
また、このLambdaの設定画面で、実行時の環境変数を設定します。
キー 説明 値 GST_PLUGIN_PATH Gstreamerのプラグインの所在を指定する /usr/local/lib/gstreamer-1.0:/myplugins STREAM_NAME KVSのストリーム名 test_Core IOT_CREDENTIAL_ENDPOINT kVSにiot-credentialsでアクセスするための認証エンドポイント ※自身の環境を参照 CERT_ID 証明書、キーのファイルのbasename ※自身の環境を参照 PIPELINE Gstreamerの処理パイプライン ※後述 IOT_CREDENTIAL_ENDPOINT はaws cliでこのように確認できます。
$ aws iot describe-endpoint --endpoint-type iot:CredentialProvider { "endpointAddress": "XXXXXXXXXXXXXX.credentials.iot.ap-northeast-1.amazonaws.com" }PIPELINE にはビデオソースにあわせたパイプラインを記述します。RTSPのカメラからの場合、私の環境では前処理など含めて以下のようになります。KVSに配信する kvssink のパイプラインは、Lambda関数内に記述しています。
rtspsrc location=rtsp://192.168.10.115:554/stream2 user-id=hogehoge user-pw=fugafuga ! rtph264_depay ! videorate ! video/x-h264,framerate=15/1 ! videoconvert ! object_detection model=models/ssd_mobilenet_v2_coco_2018_03_29/frozen_inference_graph.pb ! awsiot_notify ! videoconvert ! x264enc tune=zerolatency bframes=0 key-int-max=45 bitrate=512 ! video/x-h264,profile=baseline,stream-format=avc,alignment=au,framerate=15/1
object_detectionが物体検知のプラグインで、オプションでモデルファイルを指定します。ここで指定したモデルファイルは、デプロイ時に展開されるよう後ほど設定します。
続くawsiot_notifyが検知した情報を MQTT パブリッシュするプラグインです、それ以降はKVSに配信するkvssinkの入力に合わせるため、h.264エンコードしています。MLモデルリソースの設定
TensorflowのObject Detectionで使用するモデルをGreengrassに設定して、Lambda関数と関連付けます。
SageMakerのモデルを使用できるようですが、今回はこちらからダウンロードして使用します。http://download.tensorflow.org/models/object_detection/ssd_mobilenet_v2_coco_2018_03_29.tar.gz
ダウンロードしたファイルをS3にアップロードして、Greengrassグループの
リソースの画面から選択します。
Lambdaの実行をコンテナなしで行うため、パーミッションを以下のように設定する必要があります。OSグループIDの
997は ggc_group の番号です。Lambda関数gg-gst-testと関連付けて、実行時に参照できるようにします。
以上の設定で、Greengrassコアのデプロイ時にエッジ側にリソースが同期されます。
MQTTトピックの設定
サブスクリプションの画面で、物体検知を通知するためのMQTTトピックの設定を行います。
ソースをLambda関数 gg-gst-test に、ターゲットを Iot Cloud にします。トピックフィルタは指定しないと、ワイルドカード指定されすべてキャプチャされるようになります。
この設定により、ローカルMQTTブローカーではなく、AWS IoT側に通知できます。
Kinesis Video Streamsの設定
ストリームの作成
最後に、Kinesis Video Streams(以下 KVS)のストリームを作成します。ストリーム名は、Greengrassコアの名前と同じにする必要があります。その他はデフォルトのままで問題ありません。
ストリーミング用ポリシー、ロールの作成
KVSへのストリーミングはGstreamerのkvssinkで行いますが、kvssinkはオプションで、AWSIoTの証明書を利用してKVSにアクセスすることができます。
そのために必要となるロールとポリシーの設定をこの説明に従って、以下のように実行します。$ cat <<JSON > iam-policy-document.json { "Version":"2012-10-17", "Statement":[ { "Effect":"Allow", "Principal":{ "Service":"credentials.iot.amazonaws.com" }, "Action":"sts:AssumeRole" } ] } JSON $ aws iam create-role --role-name KVSCameraCertificateBasedIAMRole --assume-role-policy-document 'file://iam-policy-document.json' | tee iam-role.json { "Role": { "Path": "/", "RoleName": "KVSCameraCertificateBasedIAMRole", "RoleId": "AROAXGR6GGYYWJPW32Z5L", "Arn": "arn:aws:iam::XXXXXXXXXXXX:role/KVSCameraCertificateBasedIAMRole", "CreateDate": "2020-07-13T04:50:07Z", "AssumeRolePolicyDocument": { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "credentials.iot.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } } } $ cat <<'JSON' > iam-permission-document.json { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kinesisvideo:DescribeStream", "kinesisvideo:PutMedia", "kinesisvideo:TagStream", "kinesisvideo:GetDataEndpoint" ], "Resource": "arn:aws:kinesisvideo:*:*:stream/${credentials-iot:ThingName}/*" } ] } JSON $ aws iam put-role-policy --role-name KVSCameraCertificateBasedIAMRole --policy-name KVSCameraIAMPolicy --policy-document 'file://iam-permission-document.json' $ aws iot create-role-alias --role-alias KvsCameraIoTRoleAlias --role-arn $(jq --raw-output '.Role.Arn' iam-role.json) --credential-duration-seconds 3600 | tee iot-role-alias.json { "roleAlias": "KvsCameraIoTRoleAlias", "roleAliasArn": "arn:aws:iot:ap-northeast-1:XXXXXXXXXXXX:rolealias/KvsCameraIoTRoleAlias" } $ cat <<JSON > iot-policy-document.json { "Version":"2012-10-17", "Statement":[ { "Effect":"Allow", "Action":[ "iot:Connect" ], "Resource":"$(jq --raw-output '.roleAliasArn' iot-role-alias.json)" }, { "Effect":"Allow", "Action":[ "iot:AssumeRoleWithCertificate" ], "Resource":"$(jq --raw-output '.roleAliasArn' iot-role-alias.json)" } ] } JSON $ aws iot create-policy --policy-name KvsCameraIoTPolicy --policy-document 'file://iot-policy-document.json' $ aws iot attach-policy --policy-name KvsCameraIoTPolicy --target 'arn:aws:iot:ap-northeast-1:XXXXXXXXXXXX:cert/0f0c0953bd4a248aa210a2411328e9659c9dc0bc90e64c9625c08e807c0819c0'正しく作成出来ているかどうかは、以下のように確認できます。describe-streamが成功すれば問題ないです。
$ export IOT_GET_CREDENTIAL_ENDPOINT=$(aws iot describe-endpoint --endpoint-type iot:CredentialProvider --output text) $ curl -v -H "x-amzn-iot-thingname:test_Core" --cert certs/0f0c0953bd.cert.pem --key certs/0f0c0953bd.private.key https://${IOT_GET_CREDENTIAL_ENDPOINT}/role-aliases/KvsCameraIoTRoleAlias/credentials --cacert certs/0f0c0953bd.cert.pem > token.json $ AWS_ACCESS_KEY_ID=$(jq --raw-output '.credentials.accessKeyId' token.json) AWS_SECRET_ACCESS_KEY=$(jq --raw-output '.credentials.secretAccessKey' token.json) AWS_SESSION_TOKEN=$(jq --raw-output '.credentials.sessionToken' token.json) aws kinesisvideo describe-stream --stream-name test_Core以上で準備が整いました。
動作確認
結果がわかりやすいように、カメラ画像ではなく、MP4のファイルからの物体検知で試します。
ビデオはこれを使いました。
https://www2.nhk.or.jp/archives/creative/material/view.cgi?m=D0002161315_00000
CPUだけだとけっこう処理が厳しいので、予め640x360、15fpsに落としています。Lambdaの環境変数をこのように変更しました。ビデオをdemux, decodeして、確認ようの時刻を入れています。検出、通知処理を行い、再度h.264エンコードしています。
filesrc location=/data/D0002161315_00000_V_000_640x360_15fps.mp4 ! decodebin ! clockoverlay ! videoconvert ! object_detection model=models/ssd_mobilenet_v2_coco_2018_03_29/frozen_inference_graph.pb ! awsiot_notify ! videoconvert ! x264enc tune=zerolatency bframes=0 key-int-max=45 bitrate=512 ! video/x-h264,profile=baseline,stream-format=avc,alignment=au,framerate=15/1まず、Greengrassコアのコンテナを
docker-compose upして起動します。$ docker-compose up Starting aws-iot-greengrass ... done Attaching to aws-iot-greengrass aws-iot-greengrass | Process with pid 13 does not exist already aws-iot-greengrass | Setting up greengrass daemon aws-iot-greengrass | Validating hardlink/softlink protection aws-iot-greengrass | Waiting for up to 1m10s for Daemon to start aws-iot-greengrass | aws-iot-greengrass | Greengrass successfully started with PID: 13このようにコアのプロセスが起動したら、AWS IoTのコンソールのデプロイの画面から
デプロイを選択します。上手くいけば正常に完了しましたとなり、Lambda関数や、リソースがAWS側と同期され、関数が起動します。
関数の出力は
log/user/ap-northeast-1/XXXXXXXXXXXX/gg-gst-test.logというログファイルに出力されるので、これを tail すると、このような感じでログが流れました。[2020-07-21T07:18:57.327Z][INFO]-lambda_function.py:53,Start gstream pipeline [2020-07-21T07:18:58.455Z][INFO]-lambda_function.py:60,Pipeline : filesrc location=/data/D0002161315_00000_V_000_640x360_15fps.mp4 ! decodebin ! clockoverlay ! videoconvert ! object_detection model=models/ssd_mobilenet_v2_coco_2018_03_29/frozen_inference_graph.pb ! awsiot_notify ! videoconvert ! x264enc tune=zerolatency bframes=0 key-int-max=45 bitrate=512 ! video/x-h264,profile=baseline,stream-format=avc,alignment=au,framerate=15/1 ! kvssink stream-name=test_Core framerate=15 aws-region=ap-northeast-1 log-config=/kvs_log_configuration iot-certificate="iot-certificate,endpoint=c1m4lov0fgg33l.credentials.iot.ap-northeast-1.amazonaws.com,cert-path=/greengrass/certs/0f0c0953bd.cert.pem,key-path=/greengrass/certs/0f0c0953bd.private.key,ca-path=/greengrass/certs/root.ca.pem,role-aliases=KvsCameraIoTRoleAlias" (略) [2020-07-21T07:19:45.151Z][ERROR]-__init__.py:1028,DEBUG | root | 21.07 07:19:45.150 | Dummy-1 | Detect score=0.9208017587661743 [2020-07-21T07:19:45.155Z][ERROR]-__init__.py:1028,DEBUG | greengrasssdk.IoTDataPlane | 21.07 07:19:45.151 | Dummy-1 | Publishing message on topic "object/detection" with Payload "{"bounding_box": [0, 0, 0, 0], "confidence": 0.9208017587661743, "class_name": "", "track_id": 0, "pts": "0:00:53.866666666"}" [2020-07-21T07:19:45.208Z][ERROR]-__init__.py:1028,DEBUG | root | 21.07 07:19:45.207 | Dummy-1 | Detect score=0.9566311836242676 [2020-07-21T07:19:45.257Z][ERROR]-__init__.py:1028,DEBUG | root | 21.07 07:19:45.256 | Dummy-1 | Detect score=0.9690361022949219 [2020-07-21T07:19:45.305Z][ERROR]-__init__.py:1028,DEBUG | root | 21.07 07:19:45.305 | Dummy-1 | Detect score=0.952594518661499 [2020-07-21T07:19:45.353Z][ERROR]-__init__.py:1028,DEBUG | root | 21.07 07:19:45.353 | Dummy-1 | Detect score=0.9480921030044556 [2020-07-21T07:19:45.402Z][ERROR]-__init__.py:1028,DEBUG | root | 21.07 07:19:45.401 | Dummy-1 | Detect score=0.9473453164100647 [2020-07-21T07:19:45.452Z][ERROR]-__init__.py:1028,DEBUG | root | 21.07 07:19:45.451 | Dummy-1 | Detect score=0.9678215384483337 [2020-07-21T07:19:45.504Z][ERROR]-__init__.py:1028,DEBUG | root | 21.07 07:19:45.503 | Dummy-1 | Detect score=0.9282958507537842 [2020-07-21T07:19:46.195Z][ERROR]-__init__.py:1028,DEBUG | root | 21.07 07:19:46.195 | Dummy-1 | Detect score=0.9128057956695557 [2020-07-21T07:19:46.196Z][ERROR]-__init__.py:1028,DEBUG | greengrasssdk.IoTDataPlane | 21.07 07:19:46.195 | Dummy-1 | Publishing message on topic "object/detection" with Payload "{"bounding_box": [0, 0, 0, 0], "confidence": 0.9128057956695557, "class_name": "", "track_id": 0, "pts": "0:00:55.266666666"}" (略)
07:19:45に人を検知して、トピックobject/detectionにパブリッシュされています。MQTTテストクライアントでサブスクライブすると、このように確認できました。
また、Kinesis Video Streamsで該当時刻の映像をみると、
07:19:45に人が映っていました。
実験は成功です。
まとめ
- Greengrassでエッジ推論での物体検知の実装方法を知ることができました。
- Lambdaの実装や設定はかなり自由度が高いことがわかりました。
- CPUだけでは処理が厳しいことがわかったので、ラズパイでH/Wでenc/decするとか、JetsonNanoやEdgeTPUなどを導入したほうが良いと思いました。
- Greengrassのドキュメントはけっこう充実していることがわかりました。ドキュメントを読んでいて、今回は使用しなかった
コネクタを使うと、色々できそうなことがわかりました。- GreengrassコアをDockerで動かす(Lambdaをコンテナなしで実行する)ために必要な設定が多いです。コンテナで実行したほうが良いと思いました。
