图像分类应用样例开发介绍（Python）

样例介绍

“图片分类应用”，即标识图片所属的分类。

图1 图片分类应用

本例中使用的是Caffe框架的ResNet-50模型。可以直接使用训练好的开源模型，也可以基于开源模型的源码进行修改、重新训练，还可以基于算法、框架构建适合的模型。

模型的输入数据与输出数据格式：

• 输入数据：RGB格式、224*224分辨率的输入图片。
• 输出数据：图片的类别标签及其对应置信度。

• 置信度是指图片所属某个类别可能性。
• 类别标签和类别的对应关系与训练模型时使用的数据集有关，需要查阅对应数据集的标签及类别的对应关系。

业务模块介绍

业务模块的操作流程如图2 业务流程图所示。

图2 业务流程图

1. 图片输入：收集待分类图片数据集做为输入数据。
2. 输入预处理：将图片读入为numpy array，调整为模型输入大小，并修改像素值范围，与训练数据保持一致。
3. 模型推理：经过模型推理后得到图片分类结果。
4. 后处理：根据预测概率排序，找出概率最高的5个预测类别。
5. 可视化或保存到文件：将预测结果显示在界面或保存到文件。

获取代码

1. 获取代码文件。

   单击获取链接或使用wget命令获取代码（使用wget时需确保开发者套件能够连接外网），下载代码文件压缩包，以root用户登录开发者套件。

   wget https://ascend-repo.obs.cn-east-2.myhuaweicloud.com/Atlas%20200I%20DK%20A2/DevKit/models/sdk_cal_samples/resnet50_acl_python_sample.zip

2. 将“resnet50_acl_python_sample.zip”压缩包上传到开发者套件，解压并进入解压后的目录。

   unzip resnet50_acl_python_sample.zip
   cd resnet50_acl_python_sample

   代码目录结构如下所示，按照正常开发流程，需要将框架模型文件转换成昇腾AI处理器支持推理的om格式模型文件，鉴于当前是入门内容，用户可直接获取已转换好的om模型进行推理。

   resnet50_acl_python_sample  
   ├── data  
   │   ├── test.jpg                    // 测试图片  
   ├── model     
   │   ├── resnet50.om                    // ResNet-50网络的om模型                 
   ├── main.py                             // 运行程序的脚本  
   ├── imagenet-labels.json              // 图片标签文件
   ├── font.ttf                             // 字体文件，用于推理结果可视化

3. 准备用于推理的图片数据。

   如2所示文件结构，内置测试图片为“test.jpg”，用户也可从imagenet数据集中获取其它图片。

代码解析

开发代码过程中，在“resnet50_acl_python_sample/main.py”文件中已包含读入数据、前处理、推理、后处理等功能，串联整个应用代码逻辑，此处仅对代码进行解析。

1. 在“main.py”文件的开头有如下代码，用于导入第三方库与调用AscendCL接口推理所需文件。

   import json 
   import os
   
   import numpy as np  # 用于对多维数组进行计算
   from PIL import Image, ImageDraw, ImageFont  # 图片处理库，用于在图片上画出推理结果
   
   import acl  # acl推理相关接口
   
   
   ACL_MEM_MALLOC_HUGE_FIRST = 0  # 内存分配策略
   ACL_SUCCESS = 0  # 成功状态值
   IMG_EXT = ['.jpg', '.JPG', '.png', '.PNG', '.bmp', '.BMP', '.jpeg', '.JPEG']  # 所支持的图片格式

2. 定义一个main函数，串联整个应用的代码逻辑。

   def main():
       # 参数初始化
       device = 0  # 设备id
       model_path = "./model/resnet50.om"  # 模型路径
       images_path = "./data"  # 数据集路径
       label_path = "./imagenet-labels.json"  # 数据集标签
       output_path = "./output"  # 推理结果保存路径
   
       os.makedirs(output_path, exist_ok=True)  # 构造输出路径
       with open(label_path) as f:
           idx2label_list = json.load(f)  # 加载标签列表
       net = Net(device, model_path, idx2label_list)  # 初始化模型
   
       # 加载文件夹中每张图片路径
       images_list = [os.path.join(images_path, img)
                   for img in os.listdir(images_path)
                   if os.path.splitext(img)[1] in IMG_EXT]
   
       # 对每张图片进行预处理、推理、以及保存预测结果
       for img_path in images_list:
           print("images:{}".format(img_path))  # 输出图片路径
           img = preprocess_img(img_path)  # 预处理
           pred_dict = net.run([img])  # 推理
           save_image(img_path, pred_dict, output_path)  # 保存预测结果
   
       # 释放 acl 资源
       print("*****run finish******")
       net.release_resource()
   
   if __name__ == '__main__':
       main()

3. 定义推理相关实现类，包含初始化acl，加载模型，创建输入输出数据类型、推理、解析模型输出、释放acl资源等功能。
   使用AscendCL接口开发应用时，必须先初始化AscendCL ，否则可能会导致后续系统内部资源初始化出错，进而导致其它业务异常。

   class Net(object):
       def __init__(self, device_id, model_path, idx2label_list):
           self.device_id = device_id  # 设备id
           self.model_path = model_path  # 模型路径
           self.model_id = None  # 模型id
           self.context = None  # 用于管理资源，可详见https://www.hiascend.com/document/detail/zh/CANNCommunityEdition/600alpha006/infacldevg/aclpythondevg/aclpythondevg_01_0065.html
   
           self.model_desc = None  # 模型描述信息，包括模型输入个数、输入维度、输出个数、输出维度等信息
           self.load_input_dataset = None  # 输入数据集，aclmdlDataset类型
           self.load_output_dataset = None  # 输出数据集，aclmdlDataset类型
   
           self.init_resource()  # 初始化 acl 资源
           self.idx2label_list = idx2label_list  # 加载的标签列表
   
       def init_resource(self):
           """初始化 acl 相关资源"""
           print("init resource stage:")
   
           ret = acl.init()  # 初始化 acl
           check_ret("acl.init", ret)
   
           ret = acl.rt.set_device(self.device_id)  # 指定 device
           check_ret("acl.rt.set_device", ret)
   
           self.context, ret = acl.rt.create_context(self.device_id)  # 创建 context
           check_ret("acl.rt.create_context", ret)
   
           self.model_id, ret = acl.mdl.load_from_file(self.model_path)  # 加载模型
           check_ret("acl.mdl.load_from_file", ret)
   
           self.model_desc = acl.mdl.create_desc()  # 创建描述模型基本信息的数据类型
           print("init resource success")
   
           ret = acl.mdl.get_desc(self.model_desc, self.model_id)  # 根据模型ID获取模型基本信息
           check_ret("acl.mdl.get_desc", ret)
   
   
       def _gen_input_dataset(self, input_list):
           ''' 组织输入数据的dataset结构 '''
           input_num = acl.mdl.get_num_inputs(self.model_desc)  # 根据模型信息得到模型输入个数
           self.load_input_dataset = acl.mdl.create_dataset()  # 创建输入dataset结构
           for i in range(input_num):
               item = input_list[i]  # 获取第 i 个输入数据
               data = acl.util.bytes_to_ptr(item.tobytes())  # 获取输入数据字节流
               size = item.size * item.itemsize  # 获取输入数据字节数
               dataset_buffer = acl.create_data_buffer(data, size)  # 创建输入dataset buffer结构, 填入输入数据
               _, ret = acl.mdl.add_dataset_buffer(self.load_input_dataset, dataset_buffer)  # 将dataset buffer加入dataset
           print("create model input dataset success")
   
   
       def _gen_output_dataset(self):
           ''' 组织输出数据的dataset结构 '''
           output_num = acl.mdl.get_num_outputs(self.model_desc)  # 根据模型信息得到模型输出个数
           self.load_output_dataset = acl.mdl.create_dataset()  # 创建输出dataset结构
           for i in range(output_num):
               temp_buffer_size = acl.mdl.get_output_size_by_index(self.model_desc, i)  # 获取模型输出个数
               temp_buffer, ret = acl.rt.malloc(temp_buffer_size, ACL_MEM_MALLOC_HUGE_FIRST)  # 为每个输出申请device内存
               dataset_buffer = acl.create_data_buffer(temp_buffer, temp_buffer_size)  # 创建输出的data buffer结构,将申请的内存填入data buffer
               _, ret = acl.mdl.add_dataset_buffer(self.load_output_dataset, dataset_buffer)  # 将 data buffer 加入输出dataset
           print("create model output dataset success")
   
   
       def run(self, images):
           """ 数据集构造、模型推理、解析输出"""
           self._gen_input_dataset(images)  # 构造输入数据集
           self._gen_output_dataset()  # 构造输出数据集
   
           # 模型推理，推理完成后，输出会放入 self.load_output_dataset
           print('execute stage:')
           ret = acl.mdl.execute(self.model_id, self.load_input_dataset, self.load_output_dataset)
           check_ret("acl.mdl.execute", ret)
   
           # 解析输出
           result = []
           output_num = acl.mdl.get_num_outputs(self.model_desc)  # 根据模型信息得到模型输出个数
           for i in range(output_num):
               buffer = acl.mdl.get_dataset_buffer(self.load_output_dataset, i)  # 从输出dataset中获取buffer
               data = acl.get_data_buffer_addr(buffer)  # 获取输出数据内存地址
               size = acl.get_data_buffer_size(buffer)  # 获取输出数据字节数
               narray = acl.util.ptr_to_bytes(data, size)  # 将指针转为字节流数据
   
               # 根据模型输出的维度和数据类型,将字节流数据解码为numpy数组
               dims, ret = acl.mdl.get_cur_output_dims(self.model_desc, i)  # 得到当前输出的维度
               out_dim = dims['dims']  # 提取维度信息
               output_nparray = np.frombuffer(narray, dtype=np.float32).reshape(tuple(out_dim))  # 解码为numpy数组
               result.append(output_nparray)
   
           pred_dict = self._print_result(result)  # 按一定格式打印输出
   
           # 释放模型输入输出数据集
           self._destroy_dataset()
           print('execute stage success')
   
           return pred_dict
   
       def _print_result(self, result):
           """打印预测结果"""
           vals = np.array(result).flatten()  # 将结果展开为一维
           top_k = vals.argsort()[-1:-6:-1]  # 将置信度从大到小排列，并得到top5的下标
   
           print("======== top5 inference results: =============")
           pred_dict = {}
           for j in top_k:
               print(f'{self.idx2label_list[j]}: {vals[j]}')  # 打印出对应类别及概率
               pred_dict[self.idx2label_list[j]] = vals[j]  # 将类别信息和概率存入 pred_dict
           return pred_dict
   
       def _destroy_dataset(self):
           """ 释放模型输入输出数据 """
           for dataset in [self.load_input_dataset, self.load_output_dataset]:
               if not dataset:
                   continue
               number = acl.mdl.get_dataset_num_buffers(dataset)  # 获取输入buffer个数
               for i in range(number):
                   data_buf = acl.mdl.get_dataset_buffer(dataset, i)  # 获取每个输入buffer
                   if data_buf:
                       ret = acl.destroy_data_buffer(data_buf)  # 销毁每个输入buffer (销毁 aclDataBuffer 类型)
                       check_ret("acl.destroy_data_buffer", ret)
               ret = acl.mdl.destroy_dataset(dataset)  # 销毁输入数据 (销毁 aclmdlDataset类型的数据)
               check_ret("acl.mdl.destroy_dataset", ret)
   
   
       def release_resource(self):
           """释放 acl 相关资源"""
           print("Releasing resources stage:")
           ret = acl.mdl.unload(self.model_id)  # 卸载模型
           check_ret("acl.mdl.unload", ret)
           if self.model_desc:
               acl.mdl.destroy_desc(self.model_desc)  # 释放模型描述信息
               self.model_desc = None
   
           if self.context:
               ret = acl.rt.destroy_context(self.context)  # 释放 Context
               check_ret("acl.rt.destroy_context", ret)
               self.context = None
   
           ret = acl.rt.reset_device(self.device_id)  # 释放 device 资源
           check_ret("acl.rt.reset_device", ret)
   
           ret = acl.finalize()  # ACL去初始化
           check_ret("acl.finalize", ret)
           print('Resources released successfully.')
   
   def check_ret(message, ret):
       ''' 用于检查各个返回值是否正常，若否，则抛出对应异常信息 '''
       if ret != ACL_SUCCESS:
           raise Exception("{} failed ret={}".format(message, ret))

4. 加载及预处理图像。

   def preprocess_img(input_path):
       """图片预处理"""
       # 循环加载图片
       input_path = os.path.abspath(input_path)  # 得到当前图片的绝对路径
       with Image.open(input_path) as image_file:
           image_file = image_file.resize((256, 256))  # 缩放图片
           img = np.array(image_file)  # 转为numpy数组
   
       # 获取图片的高和宽
       height = img.shape[0]
       width = img.shape[1]
   
       # 对图片进行切分，取中间区域
       h_off = (height - 224) // 2
       w_off = (width - 224) // 2
       crop_img = img[h_off:height - h_off, w_off:width - w_off, :]
   
       # 转换bgr格式、数据类型、颜色空间、数据维度等信息
       img = crop_img[:, :, ::-1]  # 改变通道顺序，rgb to bgr
       shape = img.shape  # 暂时存储维度信息
       img = img.astype("float32")  # 转为 float32 数据类型
       img[:, :, 0] -= 104  # 常数104,117,123用于将图像转换到Caffe模型需要的颜色空间
       img[:, :, 1] -= 117
       img[:, :, 2] -= 123
       img = img.reshape([1] + list(shape))  # 扩展第一维度，适应模型输入
       img = img.transpose([0, 3, 1, 2])  # 将 (batch,height,width,channels) 转为 (batch,channels,height,width)
       return img

5. 将预测结果保存为图片。

   def save_image(path, pred_dict, output_path):
       """保存预测图片"""
       font = ImageFont.truetype('font.ttf', 20)  # 指定字体和字号
       color = "#fff"  # 指定颜色
       im = Image.open(path)  # 打开图片
       im = im.resize((800, 500))  # 对图片进行缩放
   
       start_y = 20  # 在图片上画分类结果时的纵坐标初始值
       draw = ImageDraw.Draw(im)  # 准备画图
       for label, pred in pred_dict.items():
           draw.text(xy = (20, start_y), text=f'{label}: {pred:.2f}', font=font, fill=color)  # 将预测的类别与置信度添加到图片
           start_y += 30  # 每一行文字往下移30个像素
   
       im.save(os.path.join(output_path, os.path.basename(path)))  # 保存图片到输出路径

运行推理

python main.py

======== top5 inference results: =============
Standard Poodle: 0.935546875
Miniature Poodle: 0.041107177734375
Toy Poodle: 0.0191192626953125
Cocker Spaniels: 0.0028858184814453125
Afghan Hound: 0.00031375885009765625

在“resnet50_acl_python_sample/output”目录下，保存相应的预测结果图片，如图3所示。

图3 预测结果图

样例总结与扩展

以上代码包括以下几个步骤：

1. 初始化acl资源：在调用AscendCL相关资源时，必须先初始化AscendCL，否则可能会导致后续系统内部资源初始化出错。

2. 对图片进行前处理：在此样例中，首先利用PIL读入图片，再对图片进行裁剪，并转换数据类型、颜色空间、数据维度等操作。

3. 推理：利用AscendCL接口对图片进行推理，其中包括构建输入输出数据集结构、推理、从输出数据结构中解析出numpy数据等步骤。

4. 保存结果：取出置信度最高的前五个预测类别，保存带有分类结果的图片。

5. 资源销毁：最后记得释放相关资源，包括卸载模型、销毁输入输出数据集、释放 Context、释放指定的计算设备、以及AscendCL去初始化等操作。

MindX SDK接口分类总结：

理解各个接口含义后，用户可进行灵活运用。除此外，此样例中只示范了图片推理，若需要对视频流数据进行推理，可用三种方式输入视频流数据：USB摄像头和手机摄像头。具体使用方式可参考《摄像头拉流》，用户只需将前处理、推理及后处理代码放入摄像头推理代码的循环中即可，注意有些细节地方需进行相应修改，下面以USB摄像头为例，其他摄像头使用方式可按相应逻辑修改。

1. 修改引入三方库部分如下，加入import cv2，用于USB视频流的读取和保存。

   import json 
   import os
   
   import numpy as np  # 用于对多维数组进行计算
   from PIL import Image, ImageDraw, ImageFont  # 图片处理库, 用于在图片上画出推理结果
   import cv2
   
   import acl  # acl推理相关接口

2. 修改main函数如下，删除了图片读取相关代码，并加入了USB摄像头读取以及视频保存等相关代码。

   def main():
       # 参数初始化、加载需要的路径
       device = 0  # 设备id
       model_path = "./model/resnet50.om"  # 模型路径
       label_path = "./imagenet-labels.json"  # 数据集标签
       output_path = "./output"  # 推理结果保存路径
       os.makedirs(output_path, exist_ok=True)  # 构造输出路径
       with open(label_path) as f:
           idx2label_list = json.load(f)  # 加载标签列表
   
       # 加载模型
       net = Net(device, model_path, idx2label_list)  # 初始化模型
   
       # 打开摄像头
       cap = cv2.VideoCapture(0)  # 打开摄像头
   
       # 获取保存视频相关变量
       fps = cap.get(cv2.CAP_PROP_FPS) 
       fourcc = cv2.VideoWriter_fourcc(*'mp4v')
       outfile = os.path.join(output_path, 'video_result.mp4')
       video_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
       video_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
       writer = cv2.VideoWriter(outfile, fourcc, fps, (video_width, video_height))
   
       try:
           while(cap.isOpened()):  # 在摄像头打开的情况下循环执行
               ret, frame = cap.read()  # 此处 frame 为 bgr 格式图片
   
               # 这里放入模型前处理、推理、后处理相关代码
               img = preprocess_img(frame)  # 前处理
               pred_dict = net.run([img])  # 推理
               img_res = save_image(frame, pred_dict)  # 保存预测结果
               writer.write(img_res)  # 将推理结果写入视频
   
       except KeyboardInterrupt:
           cap.release()
           writer.release()
       finally:
           cap.release()
           writer.release()
   
       # 释放 acl 资源
       print("*****run finish******")
       net.release_resource()

3. 修改前处理函数preprocess_img如下，其中原代码是利用PIL（pillow）直接读入图片，为rgb格式，而此处传入为摄像头读入的帧，为bgr格式，所以删除了PIL相关代码，并利用了img_rgb = img_bgr[:,:,::-1]这一句将摄像头帧转换为了rgb，再使用cv2.resize对图片进行了缩放，其他行的代码保持不变。

   def preprocess_img(img_bgr):
       """图片预处理"""
       img_rgb = img_bgr[:,:,::-1]
       img = cv2.resize(img_rgb, (256, 256))
   
       # 获取图片的高和宽
       height = img.shape[0]
       width = img.shape[1]
   
       # 对图片进行切分, 取中间区域
       h_off = (height - 224) // 2
       w_off = (width - 224) // 2
       crop_img = img[h_off:height - h_off, w_off:width - w_off, :]
   
       # 转换bgr格式、数据类型、颜色控件、数据维度等信息
       img = crop_img[:, :, ::-1]  # 改变通道顺序, rgb to bgr
       shape = img.shape  # 暂时存储维度信息
       img = img.astype("float32")  # 转为 float32 数据类型
       img[:, :, 0] -= 104  # 常数104,117,123用于将图像转换到Caffe模型需要的颜色空间
       img[:, :, 1] -= 117
       img[:, :, 2] -= 123
       img = img.reshape([1] + list(shape))  # 扩展第一维度, 适应模型输入
       img = img.transpose([0, 3, 1, 2])  # 将 (batch,height,width,channels) 转为 (batch,channels,height,width)
       return img

4. 修改save_image函数如下，由于图片是利用PIL（pillow）读入原图并画出结果，但此处传入直接为摄像头帧，所以直接cv2将结果画在图片上比较方便，即使用cv2.putText函数。

   def save_image(img_bgr, pred_dict):
       """保存预测图片"""
       start_y = 20  # 在图片上画分类结果时的纵坐标初始值
       for label, pred in pred_dict.items():
           img_bgr = cv2.putText(img_bgr, f'{label}: {pred:.2f}', (20, start_y), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 1)  # 参数含义依次为：原图、结果文字、左上角坐标、字体、字体大小、颜色、字体粗细
           start_y += 30  # 每一行文字往下移30个像素
   
       return img_bgr