在 Azure Custom Vision 上训练自定义对象检测模型（第 3 部分）：部署和运行模型

这个分为三部分的文章系列探讨了如何使用 Azure Custom Vision 训练、测试和部署一个 AI 模型来检测车辆前方的行人。

上一篇文章测试、重新训练并再次测试了一个对象检测模型。然而，如果没有最终部署，它的用处是有限的。

本文将介绍下载训练模型的副本以供离线使用。还将演示如何将模型部署到 Raspberry Pi 并对其进行修改，使其能够充当实时对象检测器。

访问 GitHub 获取此演示的完整代码。

导出模型

本系列的前一篇文章多次测试了模型并进行了迭代以提高模型的性能。现在，是时候导出模型并在离线状态下运行它了。

在 Custom Vision 服务 Web 门户中选择您的项目。转到项目的 **性能** 选项卡，选择要导出的迭代，然后在顶部菜单栏中点击 **导出**。

下一个页面会提示您选择平台。选择 **Dockerfile**，因为您将在边缘设备上运行模型。

下一步是指定您将在其上运行模型的平台类型。从菜单中选择 **ARM (Raspberry Pi 3)**，然后点击 **导出**。然后，点击 **下载**。

上面的选项会下载一个 `.zip` 文件，其中包含在 Raspberry Pi 设备上运行模型的现成解决方案。

既然模型已准备好离线运行，我们将设置 Raspberry Pi 来托管该服务。

设置您的 Raspberry Pi

本文假定您的设备使用 Raspberry Pi 操作系统 (OS)。如果没有，请按照 Raspberry Pi 的 入门指南 在您的设备上安装操作系统。安装完操作系统后，安装运行对象检测模型所需的其他软件包。

**注意：** 我们在本篇文章的设置和代码是在 Raspberry Pi 4 Model B 上使用 Debian Buster (现称为 Raspberry Pi Legacy OS) 和 Raspberry Pi Camera V2 和 Python 3.7 进行测试的。但是，代码应该适用于任何版本的操作系统、摄像头和所需模块。

在安装所需模块之前，请使用以下命令更新所有软件包

$ sudo apt-get update && sudo apt-get upgrade

您应该创建一个虚拟环境来安装所需的模块。如下创建并激活一个名为 `env` 的虚拟环境

$ python3 -m venv env
$ source env/bin/activate

激活虚拟环境后，安装所需的软件包。Custom Vision 模型在边缘设备上需要 TensorFlow。

执行以下命令安装 Python 3.7 的 TensorFlow 2.4.0。

$ pip install https://github.com/bitsy-ai/tensorflow-arm-bin/releases/download/v2.4.0-rc2/tensorflow-2.4.0rc2-cp37-none-linux_armv7l.whl

然后安装 Pillow 库以支持打开、操作和保存图像。

$ pip install pillow

最后，安装 Flask 以运行服务器。

$ pip install flask

安装上述模块后，您的环境就可以托管 Custom Vision 模型了。

接下来，在 Raspberry Pi 上运行您的 Custom Vision 模型。先前导出的训练模型是 `.zip` 格式的，应该很容易将其传输到 Raspberry Pi。

在 Raspberry Pi 上解压导出的模型并进入应用程序目录。该目录包含训练模型、标签和用于处理图像预测的 Flask 应用程序文件。

首先，检查 `predict.py` 文件

import tensorflow as tf
import numpy as np
import PIL.Image
from datetime import datetime
from urllib.request import urlopen

MODEL_FILENAME = 'model.pb'
LABELS_FILENAME = 'labels.txt'

od_model = None
labels = None

class ObjectDetection:
INPUT_TENSOR_NAME = 'image_tensor:0'
OUTPUT_TENSOR_NAMES = ['detected_boxes:0', 'detected_scores:0', 'detected_classes:0']


def __init__(self, model_filename):
graph_def = tf.compat.v1.GraphDef()
with open(model_filename, 'rb') as f:
graph_def.ParseFromString(f.read())


self.graph = tf.Graph()
with self.graph.as_default():
tf.import_graph_def(graph_def, name='')


# Get input shape
with tf.compat.v1.Session(graph=self.graph) as sess:
self.input_shape = sess.graph.get_tensor_by_name(self.INPUT_TENSOR_NAME).shape.as_list()[1:3]


def predict_image(self, image):
image = image.convert('RGB') if image.mode != 'RGB' else image
image = image.resize(self.input_shape)


inputs = np.array(image, dtype=np.float32)[np.newaxis, :, :, :]
with tf.compat.v1.Session(graph=self.graph) as sess:
output_tensors = [sess.graph.get_tensor_by_name(n) for n in self.OUTPUT_TENSOR_NAMES]
outputs = sess.run(output_tensors, {self.INPUT_TENSOR_NAME: inputs})
return outputs

def initialize():
global od_model
od_model = ObjectDetection(MODEL_FILENAME)
global labels
with open(LABELS_FILENAME) as f:
labels = [l.strip() for l in f.readlines()]


def predict_url(image_url):
with urlopen(image_url) as binary:
image = PIL.Image.open(binary)
return predict_image(image)


def predict_image(image):
predictions = od_model.predict_image(image)


predictions = [{'probability': round(float(p[1]), 8),
'tagId': int(p[2]),
'tagName': labels[p[2]],
'boundingBox': {
'left': round(float(p[0][0]), 8),
'top': round(float(p[0][1]), 8),
'width': round(float(p[0][2] - p[0][0]), 8),
'height': round(float(p[0][3] - p[0][1]), 8)
}
} for p in zip(*predictions)]


response = {'id': '', 'project': '', 'iteration': '', 'created': datetime.utcnow().isoformat(),
'predictions': predictions}


print("Results: " + str(response))
return response

上面的代码定义了以下方法

• `ObjectDetection` 类加载训练好的模型。`ObjectDetection` 类还定义了一个名为 `predict_image` 的方法，该方法以图像作为输入，使用模型对其进行评分，并返回预测结果。
• `initialize` 方法创建 `ObjectDetection` 类的对象，并从指定的标签文件中加载标签。
• `predict_url` 方法以图像 URL 作为参数，从指定的 URL 加载图像，并返回对 `predict_image` 方法的调用。
• `predict_image` 方法以图像文件作为参数，并调用 `ObjectDetection` 类的 `predict_image` 方法。它将返回的响应保存在 `predictions` 变量中，并分别在屏幕上显示 `probability`、`tagId`、`tagName` 和 `boundingBox`。

现在，检查 `app.py` 文件

import json
import os
import io

# Imports for the REST API
from flask import Flask, request, jsonify

# Imports for image processing
from PIL import Image

# Imports for prediction
from predict import initialize, predict_image, predict_url

app = Flask(__name__)

# 4MB Max image size limit
app.config['MAX_CONTENT_LENGTH'] = 4 * 1024 * 1024 

# Default route just shows simple text
@app.route('/')
def index():
return 'CustomVision.ai model host harness'

# Like the CustomVision.ai Prediction service /image route handles either
# - octet-stream image file 
# - a multipart/form-data with files in the imageData parameter
@app.route('/image', methods=['POST'])
@app.route('/<project>/image', methods=['POST'])
@app.route('/<project>/image/nostore', methods=['POST'])
@app.route('/<project>/classify/iterations/<publishedName>/image', methods=['POST'])
@app.route('/<project>/classify/iterations/<publishedName>/image/nostore', methods=['POST'])
@app.route('/<project>/detect/iterations/<publishedName>/image', methods=['POST'])
@app.route('/<project>/detect/iterations/<publishedName>/image/nostore', methods=['POST'])
def predict_image_handler(project=None, publishedName=None):
try:
imageData = None
if ('imageData' in request.files):
imageData = request.files['imageData']
elif ('imageData' in request.form):
imageData = request.form['imageData']
else:
imageData = io.BytesIO(request.get_data())

img = Image.open(imageData)
results = predict_image(img)
return jsonify(results)
except Exception as e:
print('EXCEPTION:', str(e))
return 'Error processing image', 500

# Like the CustomVision.ai Prediction service /url route handles url's
# in the body of hte request of the form:
# { 'Url': '<http url>'} 
@app.route('/url', methods=['POST'])
@app.route('/<project>/url', methods=['POST'])
@app.route('/<project>/url/nostore', methods=['POST'])
@app.route('/<project>/classify/iterations/<publishedName>/url', methods=['POST'])
@app.route('/<project>/classify/iterations/<publishedName>/url/nostore', methods=['POST'])
@app.route('/<project>/detect/iterations/<publishedName>/url', methods=['POST'])
@app.route('/<project>/detect/iterations/<publishedName>/url/nostore', methods=['POST'])
def predict_url_handler(project=None, publishedName=None):
try:
image_url = json.loads(request.get_data().decode('utf-8'))['url']
results = predict_url(image_url)
return jsonify(results)
except Exception as e:
print('EXCEPTION:', str(e))
return 'Error processing image'

if __name__ == '__main__':
# Load and intialize the model
initialize()

# Run the server
app.run(host='0.0.0.0', port=80)

上面的代码为对象检测模型创建了一个 Flask REST API。它定义了两个方法：`predict_image_handler` 和 `predict_url_handler`，应用程序会根据预测是针对图像文件还是图像文件 URL 来调用它们。

在探索完模型如何在离线状态下运行时，运行应用程序。

首先，在 `app.py` 文件中进行如下修改主机和端口

# Run the server
app.run(host='127.0.0.1', port=5000)

保存文件并运行应用程序

$ python app.py

保持服务器运行。打开另一个终端，并使用以下 `curl` 命令向服务器发送请求

$ curl -X POST http://127.0.0.1:5000/url -d '{ "url": "<test url here>" }'

上面的命令会调用 Prediction API，预测结果将显示在您的屏幕上。

在 Raspberry Pi 上启用实时对象检测

模型现在可以在离线状态下运行，并可以检测到图像中何时以及何地有人。但是，该模型目前仅适用于静态图像。它必须能够立即确定是否有人进入了视野。

要捕获视频流，提取其图像，通过模型运行它们，并检测到人的存在，您需要在 Raspberry Pi 上安装 OpenCV。此库应该能够实现实时对象检测。

使用以下命令安装 OpenCV

$ pip install opencv-python

OpenCV 成功安装后，实时对象检测器就离您只有几步之遥了。

**注意：** 在继续之前，请测试您的摄像头模块。您可以使用以下命令尝试 Raspberry Pi Camera V2

$ raspistill –o output.jpg

如果您使用的是其他摄像头，请确保它与 Raspberry Pi 兼容并且能够捕获图像和视频。否则，程序将触发错误。

现在，在应用程序目录中创建另一个名为 `pedestrian-detection.py` 的文件，并添加以下代码

import cv2

# Imports for image processing
from PIL import Image

# Imports for prediction
from predict import initialize, predict_image

def main():
# Load and initialize the model
initialize()
# create a video capture object
capture = cv2.VideoCapture(0)
while(True):
# capture the video frame by frame
ret, frame = capture.read()
# pass the frame for detection
predictions = predict_image(Image.fromarray(frame))
# display the frame
cv2.imshow('Pedestrian detector', frame)
# define quitting button
keyCode = cv2.waitKey(30) & 0xFF
if keyCode == 27:
break
# release the object once the loop is over
capture.release()
# destroy all windows
cv2.destroyAllWindows()

if __name__ == '__main__':
main()

上面的程序首先加载并初始化模型。接下来，它创建一个用于捕获视频的对象捕获函数。然后，它读取每一帧，并将该帧传递给 `predict_image` 方法，用于预测和显示该帧。它还定义了要中断循环的按键。最后，循环结束后，它会释放视频捕获并销毁所有窗口。

现在，按如下方式运行程序

$ python pedestrian-detection.py

一旦程序成功运行，一个显示当前摄像头视野的独立窗口就会打开，您将在终端中看到预测结果。

如果当前视野为空，您的模型将返回一个空的预测列表。

摘要

本文演示了如何将模型部署到边缘以进行实时预测。Custom Vision 服务使用户无需对底层 AI 和算法有全面的了解，即可训练和部署复杂的机器学习模型。

本系列文章是 Azure Custom Vision 服务的入门指南。Custom Vision 服务使广泛的开发人员能够训练和部署针对其特定用例量身定制的机器学习模型。

现在您知道了它的简便性，请查看 Azure 的 Custom Vision 服务，了解有关训练和部署机器学习模型的更多信息。

要了解如何通过更快捷的 AI 方法来驱动应用程序创新并获得持久的业务效益，请参阅 Forrester 研究：利用专门的云 AI 服务推动应用程序创新。