目次

はじめに
目的
環境
手順
　　1. AWS IoTでモノの登録
　　2. AWS IoT Core エンドポイントの確認
　　3. Jetsonの準備
　　4. JetsonからMQTTでデータが送信されることの確認
　　5. DeepStream Pythonアプリの作成
　　6. DynamoDB テーブルの作成
　　7. AWS IoT ルールの作成
動かしてみた
おわりに
参考

はじめに

これまでJetson、DeepStream SDKを使ってエッジAIアプリの開発をしてきましたが、AWSも勉強したいと思い始めました（AWSのサービスではEC2とS3を少し使ったことがある程度です）。そこで、まずエッジAIアプリと連携して使うことがあるAWS IoT, DynamoDBを使って簡単なアプリを作ることにしました。

目的

下図の構成のアプリを作ります。
Jetsonでは、カメラ映像から人を検出し、画面内の人数をカウントします。人数カウントの結果をMQTTでAWS Iotに送ります。AWS IoTではデータを受け取ったら、人数やタイムスタンプの情報をDynamoDBに登録します。

環境

Jetson Nano 4GB
JetPack 4.6.1
DeepStream SDK 6.0.1
deepstream_python_apps v1.1.1

手順

こちらのチュートリアルを参考にしました。
https://docs.aws.amazon.com/ja_jp/iot/latest/developerguide/using-laptop-as-device.html

1. AWS IoTでモノの登録

AWSマネジメントコンソールで作業します。
IoT Coreのページを開きます。

リージョンを選択します。今回は東京に設定します。
ナビゲーションペイン（左のウィンドウ）から、「モノ」を選択します。

「モノを作成」をクリックします。

「1つのモノを作成」を選択し、「次へ」をクリックします。

モノの名前を入力します。あとはデフォルトの設定のままで「次へ」をクリックします。

「新しい証明書を自動生成（推奨）」を選択し、「次へ」をクリックします。

「ポリシーを作成」をクリックします。

ポリシー名を入力し、「JSON」をクリックします。
「ポリシードキュメント」の欄にJSON形式のポリシーをコピペします。コピーする前に"region"を自分で設定したregionに、 "account"を自分のAWSアカウントに置き換えてください。
ここでは、すべてのクライアントと"/test/topic"というtopicをやりとりできるように設定しています。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "iot:Publish",
                "iot:Receive"
            ],
            "Resource": [
                "arn:aws:iot:region:account:topic/test/topic"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "iot:Subscribe"
            ],
            "Resource": [
                "arn:aws:iot:region:account:topicfilter/test/topic"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "iot:Connect"
            ],
            "Resource": [
                "arn:aws:iot:region:account:client/*"
            ]
        }
    ]
}

モノの作成ページに戻り、先ほど作成したポリシーを選択し、「モノを作成」をクリックします。

最後に証明書のダウンロード画面が出ますので、証明書をダウンロードします。

2. AWS IoT Core エンドポイントの確認

AWSマネジメントコンソールで作業します。
AWS IoT Coreのページを開きます。
ナビゲーションペインから、「設定」を選択します。

エンドポイントをコピーしてテキストファイルなどに保存してください。

3. Jetsonの準備

Jetsonで作業します。

python 3.7以降が必要なため、下記のコマンドでシンボリックリンクを作成します。

sudo ln -s /usr/bin/python3.6 /usr/bin/python3.7

AWS IoT Device SDK for Pythonをインストールします。

python3 -m pip install awsiotsdk

下記のコマンドでリポジトリをクローンします。

cd ~ 
git clone https://github.com/aws/aws-iot-device-sdk-python-v2.git

下記のコマンドを実行し、homeディレクトリ下に証明書などを保存するためのディレクトリを作成します。

mkdir ~/certs

手順1でダウンロードしたプライベートキー(xxxx-private.pem.key)、デバイス証明書(xxxx-certificate.pem.crt)、ルートCA証明書(AmazonRootCA1.pem)をJetsonのcertsディレクトリにコピーします。手順2でエンドポイントを保存したテキストファイルもJetsonにコピーします。

4. JetsonからMQTTでデータが送信されることの確認

Jetsonでサンプルプログラムを実行し、MQTTでデータを送信・受信します。AWS IoTのMQTTテストクライアントを使用してJetsonから送信されたメッセージを確認します。

4.1. MQTTテストクライアントの設定

AWSマネジメントコンソールで作業します。
AWS IoT Coreのページを開き、ナビゲーションペインから、「MQTTテストクライアント」を選択します。

「トピックをサブスクライブする」のタブを選択します。
ポリシーで設定した「test/topic」を入力し、「サブスクライブ」をクリックします。

Jetsonからデータが送られてくると、こちらのページでデータが表示されます。

4.2. Jetson側でサンプルプログラムを実行

Jetsonで作業します。

下記コマンドでサンプルプログラムがあるディレクトリに移動します。

cd ~/aws-iot-device-sdk-python-v2/samples

pubsub.pyを実行します。エンドポイント、プライベートキー、デバイス証明書、ルートCA証明書のファイルパスを設定する必要があります。

python3 pubsub.py --endpoint your-iot-endpoint --ca_file ~/certs/AmazonRootCA1.pem --cert ~/certs/xxxx-certificate.pem.crt --key ~/certs/xxxx-private.pem.key

AWSマネジメントコンソールでMQTTテストクライアントのページを開き、データが送信されていることを確認します。
"Hello World! [10]"のようなメッセージが送信されています。

5. DeepStream Pythonアプリの作成

先ほどのサンプルプログラムpubsub.pyと、deepstream_python_appsのdeepstream-test1-usbcamアプリを参考にして、x秒間隔でカメラに映っている人の人数をMQTTで送信するアプリを作成しました。
コードはこちらにあります。
https://github.com/mihara-shoko/deepstream-python-aws-iot/tree/main

送信されるデータ
json形式でデータが送信されます。

{
    "time": "2024-02-10 18:03:03.620019",
    "device_id": "test-device",
    "person": 2
}

項目	単位	説明
time	-	データ送信時刻
device_id	-	デバイスID
person	人	カメラに映っている人数。

設定できるオプション
config/config.iniでオプションを設定できます。
endpoint, cert_filepath, pri_key_filepath, ca_filepathは設定必須となっています。

項目	単位	説明	デフォルト値
endpoint	-	AWS IoT Core エンドポイント	- (設定必須)
cert_filepath	-	デバイス証明書のファイルパス（絶対パス）	- (設定必須)
pri_key_filepath	-	プライベートキーのファイルパス（絶対パス）	- (設定必須)
ca_filepath	-	ルート CA 証明書のファイルパス（絶対パス）	- (設定必須)
device_id	-	デバイスID	test_jetson
send_interval_secs	秒	データ送信間隔	10
deveice	-	カメラのパス	/dev/video0
port	-	MQTT ブローカーに接続するポート番号	8883
client_id	-	クライアントID	test-123
keep_alive_secs	秒	keep alive値	30
message_topic	-	トピック	test/topic

deepstream-test1-usbcam-modifiedアプリを/opt/nvidia/deepstream/deepstream-6.0/sources/deepstream_python_apps/apps下に保存します。
下記コマンドを実行してアプリを動かします。

python3 deepstream_test_1_usb.py

MQTTテストクライアントを使って、test/topicをサブスクライブし、データが送信できていることを確認します。

6. DynamoDB テーブルの作成

AWSマネジメントコンソールで作業します。
DynamoDBのページを開きます。

ナビゲーションペインから「テーブル」を選択します。

リージョンは、AWS IoTでモノを作成した時のリージョンと同じに設定してください。
「テーブルの作成」をクリックします。

任意のテーブル名とパーティションキーを入力し、「テーブルの作成」をクリックします。　　
今回は、"time"をパーティションキーとしました。

作成したテーブルの状態が"アクティブ"になるまで待ちます。

7. AWS IoT ルールの作成

データが送られてきたらDynamoDBに登録するようにルールを作成します。
AWSマネジメントコンソールで作業します。
IoT Coreのページを開き、ナビゲーションペインから「メッセージのルーティング」を選択します。

「ルールを作成」をクリックします。

ルール名を入力します。

SQLでDynamoDBにデータをルーティングする条件を記入します。下記では、"test/topic"というtopicで送られたデータの中からtime, device_id, personというkeyを選択してDBに登録するように記載しています。

SELECT time, device_id, person FROM 'test/topic'

アクションのプルダウンから、「DynamoDBv2」を選択します。テーブル名のプルダウンからは、先ほど作成したテーブルを選択します。DynamoDBv2を選択すると、送信されたデータをテーブルの複数のカラムに保存することができます。

IAM ロールの項では「新しいロールを作成」をクリックします。

任意の名前を入力し、ロールを作成します。作成したロールをクリックします。
「許可を追加」をクリックし、さらに「インラインポリシーを作成」をクリックします。

下記のJSONのようなポリシーを作成し、DynamoDBにデータをPutできるようにします。region, account, table_nameをそれぞれ置き換えてください。

{
    "Version": "2012-10-17",
    "Statement": {
        "Effect": "Allow",
        "Action": "dynamodb:PutItem",
        "Resource": "arn:aws:dynamodb:region:account:table/table_name"
    }
}

メッセージのルーティング画面に戻り、作成した IAMポリシーを選択し、「次へ」をクリックします。　
確認画面に遷移します。内容に問題がなければ「作成」をクリックします。

MQTTテストクライアントからデータを送信し、DynamoDBに登録されるかを確認します。

AWS IoTのナビゲーションペインから「MQTTテストクライアント」を選択します。
「トピックに公開する」タブを選択します。
トピック名とメッセージペイロードを入力し、「発行」をクリックします。

DynamoDBのページを開き、作成したテーブルをクリックします。

「テーブルアイテムの探索」をクリックします。

「返された項目」の箇所に、先ほど発行したデータが入っていることを確認します。

動かしてみた

カメラで人が映っている動画を映し、DBにデータが保存されるかを確認しました。想定通りに動きました！

Jetsonのキャプチャ

DynamoDB PersonCountTable

おわりに

AWSでは多くのサービスが提供されており、Jetsonからデータを送信してDBに登録するやり方は他にもあります。今回は比較的シンプルな方法で行いました。今後はDBに登録されたデータをQuicksightで可視化したり、閾値以上の人数が検出されたらAmazon SNSで通知したりなども試してみたいです。

参考