エッジでの AI は、最も一般的なエッジ シナリオの 1 つです。 このシナリオの実装には、画像分類と、物体検出と、身体、顔、ジェスチャの分析と、画像操作が含まれます。 このアーキテクチャ ガイドでは、Azure IoT Edge を使用してこれらのシナリオをサポートする方法について説明します。
AI モデルを更新することで AI の精度を向上させることができます。しかし、シナリオによってはエッジ デバイスのネットワーク環境が望ましくない場合があります。 たとえば、風力および石油業界では、砂漠や海上に機器が設置されている可能性があります。
IoT Edge モジュール ツインは、動的に読み込まれた AI モデルを実装するために使用します。 IoT Edge モジュールは、Docker を基盤にしています。 AI 環境における IoT Edge モジュールのイメージのサイズは、通常、少なくとも 1 GB あります。したがって、帯域幅の狭いネットワークでは AI モデルを段階的に更新することが重要です。 この記事では、この考慮事項に焦点を当てています。 LiteRT (旧称 TensorFlow Lite) または Open Neural Network Exchange (ONNX) 物体検出モデルを読み込むことができる、IoT Edge AI モジュールの作成に取り組みます。 また、このモジュールを Web API として使用できるようにすることで、それを介して他のアプリケーションまたはモジュールにメリットを提供することもできます。
この記事で説明するソリューションは、次を行う点で役立ちます。
- エッジ デバイスで AI 推論を有効にします。
- エッジにおける AI モデルのデプロイと更新のネットワーク コストを最小限に抑えます。 このソリューションを利用すれば、特に、帯域幅の狭いネットワーク環境で貴社または顧客のコストを節約できます。
- IoT エッジ デバイスのローカル ストレージで AI モデル リポジトリを作成して管理します。
- エッジ デバイスで AI モデルが切り替えられるときのダウンタイムをほぼゼロにします。
TensorFlow および LiteRT は、Google Inc. の商標です。この商標の使用は、保証を意味するものではありません。
Architecture
このアーキテクチャの Visio ファイル をダウンロードします。
データフロー
- AI モデルが、Azure Blob Storage または Web サービスにアップロードされます。 モデルとして、事前トレーニング済みの LiteRT または ONNX モデル、あるいは Azure Machine Learning で作成されたモデルを使用できます。 IoT Edge モジュールは後で、このモデルにアクセスし、エッジ デバイスにダウンロードできます。 セキュリティを強化する必要がある場合は、Blob Storage とエッジ デバイスの間でプライベート エンドポイント接続を使用することを検討してください。
- Azure IoT Hub が、デバイス モジュール ツインを AI モデル情報と自動的に同期します。 この同期は、IoT Edge がオフラインになっている場合でも行われます。 (節電またはネットワーク トラフィックの削減のために、IoT デバイスが、時間単位、日単位、または週単位でスケジュールされた時間にネットワークに接続される場合があります。)
- ローダー モジュールが、API を介してモジュール ツインの更新を監視します。 更新プログラムを検出すると、機械学習モデルの SAS トークンを取得し、AI モデルをダウンロードします。
- 詳細については、「コンテナーまたは BLOB のサービス SAS を作成する」を参照してください。
- ExpiresOn プロパティを使用して、リソースの有効期限を設定できます。 デバイスが長時間オフラインになる場合は、有効期限を延長することができます。
- ローダー モジュールが、IoT Edge モジュールの共有ローカル ストレージに AI モデルを保存します。 共有ローカル ストレージは、IoT Edge のデプロイ JSON ファイルで構成する必要があります。
- ローダー モジュールが、LiteRT または ONNX API を介してローカル ストレージから AI モデルを読み込みます。
- ローダー モジュールが、POST 要求を介してバイナリ画像を受け取り、JSON ファイルで結果を返す、Web API を開始します。
AI モデルを更新するには、新しいバージョンを Blob Storage にアップロードし、デバイス モジュール ツインを再同期して増分更新を行います。 IoT Edge モジュール イメージ全体を更新する必要はありません。
シナリオの詳細
このソリューションでは、IoT Edge モジュールを使用して AI モデルをダウンロードし、機械学習推論を有効にします。 事前トレーニング済みの LiteRT または ONNX モデルをこのソリューションで使用できます。
LiteRT
.tfliteファイルは、事前トレーニング済みの AI モデルです。 TensorFlow.org からダウンロードできます。これは、iOS や Android などのクロスプラットフォーム アプリケーションで使用できる汎用 AI モデルです。 LiteRT では、TensorFlow、PyTorch、JAX、Keras のモデルがサポートされています。 メタデータおよび関連フィールド (たとえば、labels.txt) の詳細については、「モデルからメタデータを読み取る」を参照してください。物体検出モデルは、複数のクラスの物体の存在と位置を検出するようにトレーニングされます。 たとえば、あるモデルは、さまざまな果物を含む画像と、それらが表す果物のクラス (たとえば、リンゴ) を指定するラベルと、画像内での各物体の位置を指定するデータを使用してトレーニングされるかもしれません。
画像をモデルに提供すると、検出された物体の一覧と、各物体の境界ボックスの位置と、検出の信頼度を示すスコアが出力されます。
AI モデルを構築またはカスタムチューニングする場合は、「LiteRT モデル メーカー」を参照してください。
Detection Zoo では、さまざまな遅延および精度特性を備えた、無料の事前トレーニング済みモデルをさらに入手できます。 各モデルでは、下記のコード サンプルに示す入出力シグネチャを使用します。
ONNX
ONNX は、機械学習モデルを表すためのオープン スタンダードのフォーマットです。 これを多数のフレームワークとツールに実装したことのあるパートナーのコミュニティによってサポートされています。
- ONNX は、モデルの構築とデプロイ、その他のタスクを行うためのツールをサポートしています。 詳細については、 サポートされている ONNX ツールに関するページを参照してください。
- ONNX Runtime を使用して、ONNX の事前トレーニング済みモデルを実行することができます。 事前トレーニング済みモデルの詳細については、 ONNX Model Zooを参照してください。
- このシナリオでは、物体検出および画像セグメント化のモデルである Tiny YOLOv3 を使用することができます。
ONNX コミュニティは、ディープ ラーニング モデルの作成とデプロイに役立つ ツールを提供 しています。
トレーニング済みの AI モデルをダウンロードする
トレーニング済みの AI モデルをダウンロードするために、デバイス ツインを使用して、新しいモデルが準備されたときに通知を受信することをお勧めします。 デバイスがオフラインの場合でも、エッジ デバイスがオンラインに戻るまで、IoT Hub にメッセージをキャッシュできます。 メッセージは自動的に同期されます。
下記の Python コードの例では、デバイス ツインの通知を登録し、ZIP ファイルで AI モデルをダウンロードします。 また、ダウンロードしたファイルに対してさらに操作を実行します。
このコードは、次のタスクを実行します。
- デバイス ツインの通知を受信します。 通知には、ファイル名、ファイルのダウンロード アドレス、MD5 認証トークンが含まれます。 (ファイル名には、バージョン情報 (1.0 など) を含めることができます。)
- AI モデルを ZIP ファイルとしてローカル ストレージにダウンロードします。
- 必要に応じて、MD5 チェックサムを実行します。 MD5 検証は、ネットワーク転送中に改ざんされた ZIP ファイルを阻止するのに役立ちます。
- ZIP ファイルを解凍し、ローカルに保存します。
- 新しい AI モデルが準備されたことを報告するために、通知を IoT Hub に、またはルーティング メッセージを送信します。
# define behavior for receiving a twin patch
async def twin_patch_handler(patch):
try:
print( "######## The data in the desired properties patch was: %s" % patch)
if "FileName" in patch:
FileName = patch["FileName"]
if "DownloadUrl" in patch:
DownloadUrl = patch["DownloadUrl"]
if "ContentMD5" in patch:
ContentMD5 = patch["ContentMD5"]
FilePath = "/iotedge/storage/" + FileName
# download AI model
r = requests.get(DownloadUrl)
print ("######## download AI Model Succeeded.")
ffw = open(FilePath, 'wb')
ffw.write(r.content)
ffw.close()
print ("######## AI Model File: " + FilePath)
# MD5 checksum
md5str = content_encoding(FilePath)
if md5str == ContentMD5:
print ( "######## New AI Model MD5 checksum succeeded")
# decompressing the ZIP file
unZipSrc = FilePath
targeDir = "/iotedge/storage/"
filenamenoext = get_filename_and_ext(unZipSrc)[0]
targeDir = targeDir + filenamenoext
unzip_file(unZipSrc,targeDir)
# ONNX
local_model_path = targeDir + "/tiny-yolov3-11.onnx"
local_labelmap_path = targeDir + "/coco_classes.txt"
# LiteRT
# local_model_path = targeDir + "/ssd_mobilenet_v1_1_metadata_1.tflite"
# local_labelmap_path = targeDir + "/labelmap.txt"
# message to module
if client is not None:
print ( "######## Send AI Model Info AS Routing Message")
data = "{\"local_model_path\": \"%s\",\"local_labelmap_path\": \"%s\"}" % (filenamenoext+"/tiny-yolov3-11.onnx", filenamenoext+"/coco_classes.txt")
await client.send_message_to_output(data, "DLModelOutput")
# update the reported properties
reported_properties = {"LatestAIModelFileName": FileName }
print("######## Setting reported LatestAIModelName to {}".format(reported_properties["LatestAIModelFileName"]))
await client.patch_twin_reported_properties(reported_properties)
else:
print ( "######## New AI Model MD5 checksum failed")
except Exception as ex:
print ( "Unexpected error in twin_patch_handler: %s" % ex )
推論
AI モデルがダウンロードされたら、次の手順としてエッジ デバイスでモデルを使用します。 モデルを動的に読み込み、エッジ デバイスで物体検出を実行できます。 以下のコード例は、LiteRT AI モデルを使用してエッジ デバイスで物体を検出する方法を示しています。
このコードは、次のタスクを実行します。
- LiteRT AI モデルを動的に読み込みます。
- 画像の標準化を実行します。
- 物体を検出します。
- 検出スコアを計算します。
class InferenceProcedure():
def detect_object(self, imgBytes):
results = []
try:
model_full_path = AI_Model_Path.Get_Model_Path()
if(model_full_path == ""):
raise Exception ("PLEASE SET AI MODEL FIRST")
if '.tflite' in model_full_path:
interpreter = tf.lite.Interpreter(model_path=model_full_path)
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()
input_shape = input_details[0]['shape']
# bytes to numpy.ndarray
im_arr = np.frombuffer(imgBytes, dtype=np.uint8)
img = cv2.imdecode(im_arr, flags=cv2.IMREAD_COLOR)
im_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
im_rgb = cv2.resize(im_rgb, (input_shape[1], input_shape[2]))
input_data = np.expand_dims(im_rgb, axis=0)
interpreter.set_tensor(input_details[0]['index'], input_data)
interpreter.invoke()
output_data = interpreter.get_tensor(output_details[0]['index'])
detection_boxes = interpreter.get_tensor(output_details[0]['index'])
detection_classes = interpreter.get_tensor(output_details[1]['index'])
detection_scores = interpreter.get_tensor(output_details[2]['index'])
num_boxes = interpreter.get_tensor(output_details[3]['index'])
label_names = [line.rstrip('\n') for line in open(AI_Model_Path.Get_Labelmap_Path())]
label_names = np.array(label_names)
new_label_names = list(filter(lambda x : x != '???', label_names))
for i in range(int(num_boxes[0])):
if detection_scores[0, i] > .5:
class_id = int(detection_classes[0, i])
class_name = new_label_names[class_id]
# top, left, bottom, right
results_json = "{'Class': '%s','Score': '%s','Location': '%s'}" % (class_name, detection_scores[0, i],detection_boxes[0, i])
results.append(results_json)
print(results_json)
except Exception as e:
print ( "detect_object unexpected error %s " % e )
raise
# return results
return json.dumps(results)
上記のコードの ONNX バージョンを以下に示します。 手順はほぼ同じです。 唯一の違いは、 Labelmap およびモデル出力パラメーターが異なるために、検出スコアの処理方法が異なることです。
class InferenceProcedure():
def letterbox_image(self, image, size):
'''resize image with unchanged aspect ratio using padding'''
iw, ih = image.size
w, h = size
scale = min(w/iw, h/ih)
nw = int(iw*scale)
nh = int(ih*scale)
image = image.resize((nw,nh), Image.BICUBIC)
new_image = Image.new('RGB', size, (128,128,128))
new_image.paste(image, ((w-nw)//2, (h-nh)//2))
return new_image
def preprocess(self, img):
model_image_size = (416, 416)
boxed_image = self.letterbox_image(img, tuple(reversed(model_image_size)))
image_data = np.array(boxed_image, dtype='float32')
image_data /= 255.
image_data = np.transpose(image_data, [2, 0, 1])
image_data = np.expand_dims(image_data, 0)
return image_data
def detect_object(self, imgBytes):
results = []
try:
model_full_path = AI_Model_Path.Get_Model_Path()
if(model_full_path == ""):
raise Exception ("PLEASE SET AI MODEL FIRST")
if '.onnx' in model_full_path:
# input
image_data = self.preprocess(imgBytes)
image_size = np.array([imgBytes.size[1], imgBytes.size[0]], dtype=np.float32).reshape(1, 2)
labels_file = open(AI_Model_Path.Get_Labelmap_Path())
labels = labels_file.read().split("\n")
# Loading ONNX model
print("loading Tiny YOLO...")
start_time = time.time()
sess = rt.InferenceSession(model_full_path)
print("loaded after", time.time() - start_time, "s")
input_name00 = sess.get_inputs()[0].name
input_name01 = sess.get_inputs()[1].name
pred = sess.run(None, {input_name00: image_data,input_name01:image_size})
boxes = pred[0]
scores = pred[1]
indices = pred[2]
results = []
out_boxes, out_scores, out_classes = [], [], []
for idx_ in indices[0]:
out_classes.append(idx_[1])
out_scores.append(scores[tuple(idx_)])
idx_1 = (idx_[0], idx_[2])
out_boxes.append(boxes[idx_1])
results_json = "{'Class': '%s','Score': '%s','Location': '%s'}" % (labels[idx_[1]], scores[tuple(idx_)],boxes[idx_1])
results.append(results_json)
print(results_json)
except Exception as e:
print ( "detect_object unexpected error %s " % e )
raise
# return results
return json.dumps(results)
IoT エッジ デバイスに上記のコードと機能を組み込めば、そのエッジ デバイスで AI 画像物体検出を行い、AI モデルの動的な更新をサポートすることができます。 エッジ モジュールで Web API を介して他のアプリケーションまたはモジュールに AI 機能を提供する場合は、モジュールで Web API を作成することができます。
Flask フレームワークは、API をすばやく作成するために使用できるツールの一例です。 画像をバイナリ データとして受け取り、AI モデルを使用して検出を行い、結果を JSON 形式で返すことができます。 詳細については、 Flask:「Flask Tutorial in Visual Studio Code (Visual Studio Code での Flask チュートリアル)」を参照してください。
共同作成者
この記事は、Microsoft によって保守されています。 当初の寄稿者は以下のとおりです。
プリンシパル作成者:
- Bo Wang | シニア ソフトウェア エンジニア
その他の共同作成者:
- Freddy Ayala | クラウド ソリューション アーキテクト
パブリックでない LinkedIn プロファイルを表示するには、LinkedIn にサインインします。
次の手順
- IoT Hub のモジュール ツインの理解と使用
- Learn how to deploy modules and establish routes in IoT Edge (IoT Edge にモジュールをデプロイしてルートを確立する方法)
- モジュールにデバイスのローカル ストレージへのアクセスを許可する
- 1 台のデバイスまたは多数のデバイスを対象とした IoT Edge 自動デプロイについて
- Open Neural Network Exchange
- ONNX チュートリアル
- Deploy a machine learning model on IoT and edge devices (IoT およびエッジ デバイスに機械学習モデルをデプロイする)