在 Azure IoT Edge 裝置上啟用機器學習推斷

Azure IoT Edge

Azure IoT 中樞

邊緣上的 AI 是最受歡迎的邊緣案例之一。 此案例的實作包括影像分類、物件偵測、本文、臉部和手勢分析，以及影像操作。 此架構指南說明如何使用 Azure IoT Edge 來支持這些案例。

您可以藉由更新 AI 模型來改善 AI 精確度，但在某些情況下，邊緣裝置網路環境不好。 例如，在風能和石油工業中，設備可能位於沙漠或海洋中。

IoT Edge 模組對應項可用來實作動態載入的 AI 模型。 IoT Edge 模組是以 Docker 為基礎。 AI 環境中 IoT Edge 模組的映像大小通常至少為 1 GB，因此以累加方式更新 AI 模型在窄頻寬網路中很重要。 該考慮是本文的主要焦點。 其概念是建立IoT Edge AI模組，以載入LiteRT（先前稱為 TensorFlow Lite）或開放式類神經網路交換 （ONNX） 物件偵測模型。 您也可以將模組啟用為 Web API，以便使用它來讓其他應用程式或模組受益。

本文所述的解決方案可透過下列方式協助您：

• 在邊緣裝置上啟用 AI 推斷。
• 將邊緣上部署和更新 AI 模型的網路成本降到最低。 解決方案可以為您或您的客戶節省成本，特別是在窄頻寬網路環境中。
• 在 IoT Edge 裝置的本機記憶體中建立和管理 AI 模型存放庫。
• 當邊緣裝置切換 AI 模型時，幾乎零停機。

TensorFlow 和 LiteRT 是 Google Inc 的商標。使用此標記時未隱含任何背書。

架構

下載此架構的 Visio 檔案。

資料流程

1. AI 模型會上傳至 Azure Blob 儲存體 或 Web 服務。 此模型可以是預先定型的 LiteRT 或 ONNX 模型，或是在 Azure 機器學習 中建立的模型。 IoT Edge 模組可以存取此模型，並在稍後將它下載到邊緣裝置。 如果您需要更好的安全性，請考慮使用 Blob 記憶體與邊緣裝置之間的私人端點連線。
2. Azure IoT 中樞 自動同步處理裝置模組對應項與 AI 模型資訊。 即使 IoT Edge 已離線，也會發生同步處理。 （在某些情況下，IoT 裝置會以每小時、每日或每周的排程時間連線到網路，以節省電源或減少網路流量。
3. 載入器模組會透過 API 監視模組對應項的更新。 當它偵測到更新時，它會取得機器學習模型 SAS 令牌，然後下載 AI 模型。
   • 如需詳細資訊，請參閱 建立容器或 Blob 的 SAS 令牌。
   • 您可以使用 ExpiresOn 屬性來設定資源的到期日。 如果你的裝置預計會長時間離線，你可以延長有效期限。
4. 載入器模組會將 AI 模型儲存在 IoT Edge 模組的共用本機記憶體中。 您必須在 IoT Edge 部署 JSON 檔案中設定共用本機記憶體。
5. 載入器模組會透過 LiteRT 或 ONNX API 從本機記憶體載入 AI 模型。
6. 載入器模組會啟動 Web API，以透過 POST 要求接收二進位相片，並在 JSON 檔案中傳回結果。

若要更新 AI 模型，您可以將新版本上傳至 Blob 記憶體，並再次同步裝置模組對應項以進行累加式更新。 不需要更新整個IoT Edge模組映像。

案例詳細資料

在此解決方案中，IoT Edge 模組可用來下載 AI 模型，然後啟用機器學習推斷。 您可以在此解決方案中使用預先定型的 LiteRT 或 ONNX 模型。

LiteRT 的

• 檔案 .tflite 是預先定型的 AI 模型。 您可以從 TensorFlow.org 下載一個。這是一般 AI 模型，您可以在 iOS 和 Android 等跨平台應用程式中使用。 LiteRT 支援來自 TensorFlow、PyTorch、JAX 和 Keras 的模型。 如需元數據和相關聯字段的詳細資訊（例如， labels.txt），請參閱 從模型讀取元數據。

• 對象偵測模型會定型，以偵測多個物件類別的存在和位置。 例如，模型可能會使用包含各種水果片段的影像來定型，以及指定它們所代表水果類別的標籤，以及指定影像中每個對象出現位置的數據。

  當影像提供給模型時，它會輸出所偵測的物件清單、每個物件的周框方塊位置，以及指出偵測信賴度的分數。

• 如果您想要建置或自定義調整 AI 模型，請參閱 LiteRT 模型製作者。

• 您可以在偵測 Zoo 取得更多免費的預先定型偵測模型，並具有各種延遲和精確度特性。 每個模型都會使用下列程式代碼範例中顯示的輸入和輸出簽章。

ONNX

ONNX 是表示機器學習模型的開放標準格式。 它受到合作夥伴社群的支持，這些合作夥伴已在許多架構和工具中實作。

• ONNX 支援建置和部署模型的工具，以及完成其他工作。 如需詳細資訊，請參閱 支援的 ONNX 工具。
• 您可以使用 ONNX 執行時間來執行 ONNX 預先定型的模型。 如需預先定型模型的資訊，請參閱 ONNX 模型動物園。
• 在此案例中，您可以使用對象偵測和影像分割模型： Tiny YOLOv3。

ONNX 社群 提供工具來 協助您建立和部署深度學習模型。

下載定型的 AI 模型

若要下載已定型的 AI 模型，建議您在新的模型就緒時，使用裝置對應項來接收通知。 即使裝置脫機，訊息仍可在 IoT 中樞 中快取，直到邊緣裝置重新上線為止。 訊息會自動同步。

以下 Python 程式碼範例會為裝置雙胞胎登錄通知，然後下載 AI 模型的 ZIP 檔。 它也會在下載的檔案上執行進一步的作業。

程式代碼會執行下列工作：

1. 接收裝置對應項通知。 通知包含檔名、檔案下載位址和 MD5 驗證令牌。 （在檔案名中，您可以包含版本資訊，例如 1.0。
2. 將 AI 模型下載為 ZIP 檔案至本機記憶體。
3. 選擇性地執行 MD5 總和檢查碼。 MD5 驗證有助於防止網路傳輸期間遭到竄改的 ZIP 檔案。
4. 將 ZIP 檔案解壓縮並在本機儲存。
5. 將通知傳送至 IoT 中樞 或路由訊息，以報告新的 AI 模型已就緒。

# 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 模型來偵測邊緣裝置上的物件。

程式代碼會執行下列工作：

1. 動態載入 LiteRT AI 模型。
2. 執行影像標準化。
3. 偵測物件。
4. 計算偵測分數。

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 Edge裝置包含上述程式代碼和功能，則邊緣裝置具有 AI 影像物件偵測，並支援 AI 模型的動態更新。 如果您想要邊緣模組透過 Web API 提供 AI 功能給其他應用程式或模組，您可以在模組中建立 Web API。

Flask 架構是一個可用來快速建立 API 的工具範例。 您可以接收影像做為二進位數據、使用 AI 模型進行偵測，然後以 JSON 格式傳回結果。 如需詳細資訊，請參閱 Visual Studio Code中的 Flask：Flask 教學課程。

參與者

本文由 Microsoft 維護。 原始投稿人如下。

主要作者：

• 王波 |資深軟體工程師

其他投稿人：

• Freddy Ayala |雲端解決方案架構師

若要查看非公開的 LinkedIn 設定檔，請登入 LinkedIn。

下一步

• 了解和使用 IoT 中樞的模組對應項
• 了解如何在 IoT Edge 中部署模組及建立路由
• 為模組提供裝置本機記憶體的存取權
• 瞭解單一裝置或大規模IoT Edge自動部署
• 開啟類神經網路交換
• ONNX 教學課程
• 在IoT和邊緣裝置上部署機器學習模型

相關資源

• IoT 架構
• 在 Azure 中選擇IoT解決方案
• AI 架構設計

邊緣上的 AI 是最受歡迎的邊緣案例之一。 此案例的實作包括影像分類、物件偵測、本文、臉部和手勢分析，以及影像操作。 此架構指南說明如何使用 Azure IoT Edge 來支持這些案例。

您可以藉由更新 AI 模型來改善 AI 精確度，但在某些情況下，邊緣裝置網路環境不好。 例如，在風能和石油工業中，設備可能位於沙漠或海洋中。

IoT Edge 模組對應項可用來實作動態載入的 AI 模型。 IoT Edge 模組是以 Docker 為基礎。 AI 環境中 IoT Edge 模組的映像大小通常至少為 1 GB，因此以累加方式更新 AI 模型在窄頻寬網路中很重要。 該考慮是本文的主要焦點。 其概念是建立IoT Edge AI模組，以載入LiteRT（先前稱為 TensorFlow Lite）或開放式類神經網路交換 （ONNX） 物件偵測模型。 您也可以將模組啟用為 Web API，以便使用它來讓其他應用程式或模組受益。

本文所述的解決方案可透過下列方式協助您：

• 在邊緣裝置上啟用 AI 推斷。
• 將邊緣上部署和更新 AI 模型的網路成本降到最低。 解決方案可以為您或您的客戶節省成本，特別是在窄頻寬網路環境中。
• 在 IoT Edge 裝置的本機記憶體中建立和管理 AI 模型存放庫。
• 當邊緣裝置切換 AI 模型時，幾乎零停機。

TensorFlow 和 LiteRT 是 Google Inc 的商標。使用此標記時未隱含任何背書。

架構

下載此架構的 Visio 檔案。

資料流程

1. AI 模型會上傳至 Azure Blob 儲存體 或 Web 服務。 此模型可以是預先定型的 LiteRT 或 ONNX 模型，或是在 Azure 機器學習 中建立的模型。 IoT Edge 模組可以存取此模型，並在稍後將它下載到邊緣裝置。 如果您需要更好的安全性，請考慮使用 Blob 記憶體與邊緣裝置之間的私人端點連線。
2. Azure IoT 中樞 自動同步處理裝置模組對應項與 AI 模型資訊。 即使 IoT Edge 已離線，也會發生同步處理。 （在某些情況下，IoT 裝置會以每小時、每日或每周的排程時間連線到網路，以節省電源或減少網路流量。
3. 載入器模組會透過 API 監視模組對應項的更新。 當它偵測到更新時，它會取得機器學習模型 SAS 令牌，然後下載 AI 模型。
   • 如需詳細資訊，請參閱 建立容器或 Blob 的 SAS 令牌。
   • 您可以使用 ExpiresOn 屬性來設定資源的到期日。 如果你的裝置預計會長時間離線，你可以延長有效期限。
4. 載入器模組會將 AI 模型儲存在 IoT Edge 模組的共用本機記憶體中。 您必須在 IoT Edge 部署 JSON 檔案中設定共用本機記憶體。
5. 載入器模組會透過 LiteRT 或 ONNX API 從本機記憶體載入 AI 模型。
6. 載入器模組會啟動 Web API，以透過 POST 要求接收二進位相片，並在 JSON 檔案中傳回結果。

若要更新 AI 模型，您可以將新版本上傳至 Blob 記憶體，並再次同步裝置模組對應項以進行累加式更新。 不需要更新整個IoT Edge模組映像。

案例詳細資料

在此解決方案中，IoT Edge 模組可用來下載 AI 模型，然後啟用機器學習推斷。 您可以在此解決方案中使用預先定型的 LiteRT 或 ONNX 模型。

LiteRT 的

• 檔案 .tflite 是預先定型的 AI 模型。 您可以從 TensorFlow.org 下載一個。這是一般 AI 模型，您可以在 iOS 和 Android 等跨平台應用程式中使用。 LiteRT 支援來自 TensorFlow、PyTorch、JAX 和 Keras 的模型。 如需元數據和相關聯字段的詳細資訊（例如， labels.txt），請參閱 從模型讀取元數據。

• 對象偵測模型會定型，以偵測多個物件類別的存在和位置。 例如，模型可能會使用包含各種水果片段的影像來定型，以及指定它們所代表水果類別的標籤，以及指定影像中每個對象出現位置的數據。

  當影像提供給模型時，它會輸出所偵測的物件清單、每個物件的周框方塊位置，以及指出偵測信賴度的分數。

• 如果您想要建置或自定義調整 AI 模型，請參閱 LiteRT 模型製作者。

• 您可以在偵測 Zoo 取得更多免費的預先定型偵測模型，並具有各種延遲和精確度特性。 每個模型都會使用下列程式代碼範例中顯示的輸入和輸出簽章。

ONNX

ONNX 是表示機器學習模型的開放標準格式。 它受到合作夥伴社群的支持，這些合作夥伴已在許多架構和工具中實作。

• ONNX 支援建置和部署模型的工具，以及完成其他工作。 如需詳細資訊，請參閱 支援的 ONNX 工具。
• 您可以使用 ONNX 執行時間來執行 ONNX 預先定型的模型。 如需預先定型模型的資訊，請參閱 ONNX 模型動物園。
• 在此案例中，您可以使用對象偵測和影像分割模型： Tiny YOLOv3。

ONNX 社群 提供工具來 協助您建立和部署深度學習模型。

下載定型的 AI 模型

若要下載已定型的 AI 模型，建議您在新的模型就緒時，使用裝置對應項來接收通知。 即使裝置脫機，訊息仍可在 IoT 中樞 中快取，直到邊緣裝置重新上線為止。 訊息會自動同步。

以下 Python 程式碼範例會為裝置雙胞胎登錄通知，然後下載 AI 模型的 ZIP 檔。 它也會在下載的檔案上執行進一步的作業。

程式代碼會執行下列工作：

1. 接收裝置對應項通知。 通知包含檔名、檔案下載位址和 MD5 驗證令牌。 （在檔案名中，您可以包含版本資訊，例如 1.0。
2. 將 AI 模型下載為 ZIP 檔案至本機記憶體。
3. 選擇性地執行 MD5 總和檢查碼。 MD5 驗證有助於防止網路傳輸期間遭到竄改的 ZIP 檔案。
4. 將 ZIP 檔案解壓縮並在本機儲存。
5. 將通知傳送至 IoT 中樞 或路由訊息，以報告新的 AI 模型已就緒。

# 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 模型來偵測邊緣裝置上的物件。

程式代碼會執行下列工作：

1. 動態載入 LiteRT AI 模型。
2. 執行影像標準化。
3. 偵測物件。
4. 計算偵測分數。

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 Edge裝置包含上述程式代碼和功能，則邊緣裝置具有 AI 影像物件偵測，並支援 AI 模型的動態更新。 如果您想要邊緣模組透過 Web API 提供 AI 功能給其他應用程式或模組，您可以在模組中建立 Web API。

Flask 架構是一個可用來快速建立 API 的工具範例。 您可以接收影像做為二進位數據、使用 AI 模型進行偵測，然後以 JSON 格式傳回結果。 如需詳細資訊，請參閱 Visual Studio Code中的 Flask：Flask 教學課程。

參與者

本文由 Microsoft 維護。 原始投稿人如下。

主要作者：

• 王波 |資深軟體工程師

其他投稿人：

• Freddy Ayala |雲端解決方案架構師

若要查看非公開的 LinkedIn 設定檔，請登入 LinkedIn。

下一步

• 了解和使用 IoT 中樞的模組對應項
• 了解如何在 IoT Edge 中部署模組及建立路由
• 為模組提供裝置本機記憶體的存取權
• 瞭解單一裝置或大規模IoT Edge自動部署
• 開啟類神經網路交換
• ONNX 教學課程
• 在IoT和邊緣裝置上部署機器學習模型

相關資源

• IoT 架構
• 在 Azure 中選擇IoT解決方案
• AI 架構設計