こんにちは！
ソニーセミコンダクタソリューションズ の高田と申します。
新卒1年目でプログラミング歴も浅いですが、初めて投稿します！
この記事の要約
- AITRIOSから出てくるシリアライズ化されたデータを自動でデコード・デシリアライズして、Azure Blob Storageに保存することができます。
- この記事はAITRIOSのREST APIからEvent Hubs転送機能を使って、AzureのEvent HubsをトリガーにするところからStorageに自動で保存されるところまで説明します。
- AITRIOSの物体検知AIモデルの運用を始め、デコード・デシリアライズを手動で行うのが煩わしくなった方をメインにした記事です。
はじめに
この記事では、AITRIOSのDeveloper Editionで物体検知のAIモデルを作成し、エッジデバイスにエッジファームウェアやAIモデルをデプロイ後に、撮影+推論を実行している状態で、得られた推論結果を自動でデコード・デシリアライズし、Azure Blob Storageに保存する方法を紹介します。
システム構成図は下記の画像のようになっています。
もし、まだAITRIOSを運用したことがない場合は、下記の記事をご覧ください。まず、物体検知AIモデルを実行し、正常に撮影+推論が行えるようにしてください。（どちらかで実行できていたらOKです。）
- Qiitaの記事では6章まで完了しておいてください。7章を自動化し、Storageに保存していきます
- GetStartedでは、2.7章まで完了しておいてください。3.6.7章や4.9章で行っていることを自動化し、Storageにデータを保存します
※ 今回の実装には課金制のAzureが必要になります。
※ こちらの実装は基本的なプログラミング知識やAIの知識が必要になります。
本文
0. 今回の作業の流れ
1. AITRIOSのEvent Hubs転送機能を使って、AzureのEvent Hubsに推論結果を転送する。
2. 今回必要な実装を行い、ローカル環境で実行して試す。
3. Azure Functionsにデプロイし、自動で実行できるようにする。
0.5 実装の準備
Azureにて、Event HubsとストレージアカウントとAzure Functions（関数アプリ）のリソースを作成しておく必要があります。これらを同じリソースグループに入れておくことを推奨します。
またコードエディターとしてはVisual Studio Code(VSCode)を使用しており、拡張機能としてAzure ResourcesとAzure Functionsをインストールしています。
今回私は、Python 3.11.9で動作確認しています。仮想環境を構築して下記のrequirements.txtにあるモジュールをインストールしてください。
requirements.txt
flatbuffers>=2.0
azure-functions==1.11.2
azure-storage-blob==12.9.0
azure-eventhub
1. Event Hubs転送機能について
下記のREST APIを実行してEvent Hubsに推論結果を送信します。本記事ではPostmanで実行しました。
実行後、推論結果がEvent Hubsに転送されるようになります。これをトリガーに自動でデコード・デシリアライズします。
2. 今回の実装について
ディレクトリ構造
VSCodeのAzure FunctionsからV2のFunctionsテンプレートを選択して自動的に.venv/、.vscode/、funcignore、.gitignoreが入ったワークスペースを作れます。
FUNCTIONAPP_OBJECTDETECTION/
├── .venv/
├── .vscode/
├── SmartCamera/
│ ├── __init__.py
│ ├── BoundingBox2d.py
│ ├── BoundingBox.py
│ ├── GeneralObject.py
│ ├── ObjectDetectionData.py
│ └── ObjectDetectionTop.py
├── .funcignore
├── .gitignore
├── function_app.py
├── host.json
├── local.settings.json
└── requirements.txt
デシリアライズに必要な準備
SmartCameraのディレクトリには、物体検知のデシリアライズに必要なアイテムを入れます。
こちらの5章の"5.2. 推論結果のデシリアライズ"の準備から簡単に物体検知用のデコード・デシリアライズするコードを取得することができます
Python以外や他のAIモデルを使用する場合は、下記の記事がデシリアライズに必要なアイテムを生成または取得するのに参考になると思います。
Event Hubsに送られてきた推論結果をトリガーにデコード・デシリアライズを実行し、Azure Blob Storageへの保存の実装（ローカル環境）
function_app.py
import azure.functions as func
import json
import logging
import base64
from SmartCamera import ObjectDetectionTop, BoundingBox, BoundingBox2d
app = func.FunctionApp()
@app.function_name(name="DeserializeFunction")
@app.event_hub_message_trigger(arg_name="event", event_hub_name="# Event Hubsのインスタンス名を入力してください", connection="EventHubConnectionString")
@app.blob_output(arg_name="outputblob", path="Blob名を入力してください。", connection="BlobStorageConnection")
def DeserializeFunction(event: func.EventHubEvent, outputblob: func.Out[str]) -> None:
logging.info("DeserializeFunction triggered. Processing message...")
try:
# データのデコードとパース
body = event.get_body().decode('utf-8')
data = json.loads(body)
logging.info(f"Received event body: {body}")
logging.info(f"Parsed data: {json.dumps(data, indent=2)}")
# 'O'フィールドのデコードとデシリアライズ
o_data_base64 = data.get('backdoor-EA_Main/placeholder', {}).get('Inferences', [{}])[0].get('O', '')
try:
o_data = base64.b64decode(o_data_base64)
except Exception as e:
logging.error(f"Base64デコードに失敗しました: {str(e)}")
raise
if not o_data:
raise ValueError("Decoded data is empty")
logging.info(f"Decoded data length: {len(o_data)}")
# デコードされたデータの解析
ppl_out = ObjectDetectionTop.ObjectDetectionTop.GetRootAsObjectDetectionTop(o_data, 0)
obj_data = ppl_out.Perception()
if obj_data is None:
raise ValueError("Cannot deserialize inference data (Only ObjectDetection is supported).")
# 結果の構築
result = {
'id': data.get('id'),
'device_id': data.get('DeviceID'),
'model_id': data.get('backdoor-EA_Main/placeholder', {}).get('ModelID'),
'inference_result': {
'DeviceID': data.get('DeviceID'),
'ModelID': data.get('backdoor-EA_Main/placeholder', {}).get('ModelID'),
'Image': data.get('backdoor-EA_Main/placeholder', {}).get('Image'),
'Inferences': [
{
'T': data.get('backdoor-EA_Main/placeholder', {}).get('Inferences', [{}])[0].get('T'),
'O': {}
}
],
'project_id': data.get('backdoor-EA_Main/placeholder', {}).get('project_id')
}
}
# ObjectDetectionListの処理
inferences = result['inference_result']['Inferences'][0]['O']
res_num = obj_data.ObjectDetectionListLength()
for i in range(res_num):
obj_list = obj_data.ObjectDetectionList(i)
union_type = obj_list.BoundingBoxType()
if union_type != BoundingBox.BoundingBox.BoundingBox2d:
raise ValueError("Cannot deserialize inference data (Only ObjectDetection is supported).")
bbox_2d = BoundingBox2d.BoundingBox2d()
bbox_2d.Init(obj_list.BoundingBox().Bytes, obj_list.BoundingBox().Pos)
inferences[str(i + 1)] = {
"C": obj_list.ClassId(),
"P": obj_list.Score(),
"X": bbox_2d.Left(),
"Y": bbox_2d.Top(),
"x": bbox_2d.Right(),
"y": bbox_2d.Bottom()
}
logging.info(f"Deserialized data: {json.dumps(result, indent=2)}")
# JSONデータの作成
json_data = json.dumps(result, ensure_ascii=False, indent=2)
# Blobに保存
outputblob.set(json_data)
logging.info(f"データがBlobに保存されました: {data.get('id')}")
except ValueError as ve:
logging.error(f"値エラーが発生しました: {str(ve)}")
except AttributeError as ae:
logging.error(f"属性エラーが発生しました: {str(ae)}")
except Exception as e:
logging.error(f"エラーが発生しました: {str(e)}")
logging.error(f"エラーの種類: {type(e).__name__}")
logging.error(f"エラーの詳細情報: {e.args}
-
Event Hubsのインスタンス名：Event Hubs名前空間（Namespace）から"＋Event Hub"でインスタンスを作成した名前。
-
path名：ストレージアカウントのストレージブラウザーからBlobコンテナー名を入力するか、ここに入力した名前のBlobコンテナーを自動で生成する。Blobコンテナー名/{ここに保存されるデータ名が入ります。}.json
例.
data/{DateTime}.jsonはdataというBlobコンテナー名への経路。DateTimeだと日時、rand-guidとすると、ランダムのIDがデータ名に付与される。
local.settings.json
{
"IsEncrypted": false,
"Values": {
"AzureWebJobsStorage": "#ストレージアカウントの接続文字列を入力してください",
"FUNCTIONS_WORKER_RUNTIME": "python",
"EventHubConnectionString": "#Event Hubsの接続文字列を入力してください",
"BlobStorageConnection": "#ストレージアカウントの接続文字列を入力してください"
}
}
-
Event Hubsの接続文字列：今回のために作ったEvent Hubs（名前空間）のリソースを選択し、設定の共有アクセスポリシーからポリシーを選択した時の出てくる接続文字列 - 主キーをコピペしてください。
-
ストレージアカウントの接続文字列：今回のために作ったストレージアカウントのリソースを選択し、セキュリティとネットワークのアクセスキーから接続文字列をコピペしてください。
host.json
{
"version": "2.0",
"logging": {
"applicationInsights": {
"samplingSettings": {
"isEnabled": true,
"excludedTypes": "Request"
}
},
"fileLoggingMode": "always"
},
"extensionBundle": {
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[4.*, 5.0.0)"
},
"functions": ["HttpTrigger1", "HttpExample", "DeserializeFunction"]
}
これでコード実装は終了です。
ターミナルで以下のコマンドを打つと、ローカル環境で実行することができます。以下のコマンドはAzure Functions Core Toolsをインストールしている必要があります。
func start --verbose
これでエラーが出ておらず、ログに「データがBlobに保存されました」と出たら、今回のために作ったストレージアカウントを開いてください。左にあるタブのストレージブラウザーを押し、Blobコンテナーから今回保存するコンテナー名を選択してデータが保存されているか確認して下さい。データを選択してダウンロードを押すと中身を確認できます。JSON形式で、下記のような形で出力されているはずです。
{
"id": "0ebb0e0f-77e8-4e6f-abb6-1ba8f6257ad7",
"device_id": "sid---------------------",
"model_id": "0311030016010100",
"inference_result": {
"DeviceID": "sid---------------------",
"ModelID": "0311030016010100",
"Image": false,
"Inferences": [
{
"T": "20241015082645655",
"O": {
"1": {
"C": 0,
"P": 0.578125,
"X": 137,
"Y": 23,
"x": 240,
"y": 165
}
}
}
],
"project_id": "--------------"
}
}
これが出力されていたらローカル環境での実装は成功です。
3. AzureFunctionへのデプロイについて
まずはVSCodeにAzure ResourcesとAzure Functionsの拡張機能をインストールして有効にしてください。
VSCodeを開き、Azureにサインインしてください。
その後、左下にある（下記の画像のようなところ）Azure Functionsのマークを押し、「Deploy to Azure...」を選択して、今回のために作った関数アプリ（Function App）名（ない場合は、ここで作ることも可能です）を選択してデプロイを開始します。
デプロイ完了後、デプロイした関数アプリ（Function App）を開き、左側サイドバーの設定の環境変数を選択します。アプリ設定で追加を押し、EventHubConnectionStringとBlobStorageConnectionを追加し、local.settings.jsonで記載したものに合わせて値も入力してください。
このあとに、ストレージアカウントでローカル環境で試したときと同じように、データが追加されているか確認してください。データが追加されていたら実装が完全に終了です。
困った時は
もし、記事の途中でうまくいかなかった場合は、気軽にこの記事にコメントいただいたり、以下のサポートのページもご覧ください。
コメントのお返事にはお時間をいただく可能性もありますがご了承ください。
また、記事の内容以外で AITRIOS についてお困りごとなどあれば以下よりお問い合わせください。
終わりに
初めての機能実装でしたが、2週間ほどでここまで実装することができました（皆さんならもっと早く実装できると思います！）。他にも不便な点などがあればどんどん効率化して記事にしていきたいと思うので、そちらについてもコメントお願いします。
他にも諸先輩方が面白い記事を書いているので、是非AITRIOSのQiita Organization をフォローいただければ幸いです。
同時に、#AITRIOSのタグをつけてのAITRIOS で何かやってみた、や遊んでみたのような記事の投稿も大歓迎ですので、こちらもぜひよろしくお願いします！
改めて記事をお読みいただきありがとうございました！