Vitis AI 综合实践（DPU example: dpu

1. 简介

2. 代码解析

2.1 导入库

2.2 图像预处理

2.3 读取标签

2.4 读取图像

2.5 获取IO形状

2.6 申请内存

2.7 运行推理

2.8 后处理

3. 相关类的介绍

3.1 DpuOverlay 类

3.2 Overlay 类

3.3 Bitsteam 类

3.4 Device 类

3.5 DeviceMeta 元类

3.6 type 类

3.7 VART

3.7.1 vart*.so

3.7.2 vart*.cpp

4. 总结

1. 简介

本文以 DPU example: dpu_resnet50.ipynb 例程展开，深入探讨使用 PYNQ 平台进行深度学习推理的过程。

介绍 DPUOverlay 类及其在 PYNQ 中的作用，包括加载比特流和模型、管理 DPU 运行时等功能。
解析图像预处理流程，包括调整图像大小、均值归一化以及中心裁剪等步骤，确保输入数据符合模型要求。
标签读取、图像读取和推理执行的过程，强调内存管理与数据结构的使用。
介绍与 DPU 相关的类和库，如 VART，以便更好地理解整个推理流程及其背后的实现机制。
介绍与 VART 有关的 pybind11 相关知识。

2. 代码解析

2.1 导入库

import os
import time
import numpy as np
import cv2
import matplotlib.pyplot as plt
from pynq_dpu import DpuOverlay
%matplotlib inlineoverlay = DpuOverlay("dpu.bit")
overlay.load_model("dpu_resnet50.xmodel")

DpuOverlay 继承自 PYNQ 的 Overlay 类，并在此基础上增加了一些特定于 DPU 的功能：

加载 DPU 比特流
下载 Overlay
加载模型
管理 DPU 运行时

2.2 图像预处理

_R_MEAN = 123.68
_G_MEAN = 116.78
_B_MEAN = 103.94MEANS = [_B_MEAN,_G_MEAN,_R_MEAN]def resize_shortest_edge(image, size):H, W = image.shape[:2]if H >= W:nW = sizenH = int(float(H)/W * size)else:nH = sizenW = int(float(W)/H * size)return cv2.resize(image,(nW,nH))def mean_image_subtraction(image, means):B, G, R = cv2.split(image)B = B - means[0]G = G - means[1]R = R - means[2]image = cv2.merge([R, G, B])return imagedef central_crop(image, crop_height, crop_width):image_height = image.shape[0]image_width = image.shape[1]offset_height = (image_height - crop_height) // 2offset_width = (image_width - crop_width) // 2return image[offset_height:offset_height + crop_height, offset_width:offset_width + crop_width, :]def preprocess_fn(image, crop_height = 224, crop_width = 224):image = resize_shortest_edge(image, 256)image = mean_image_subtraction(image, MEANS)image = central_crop(image, crop_height, crop_width)return image

在处理图像数据时，使用 ImageNet 数据集的均值和方差进行标准化是一个常见的做法。确保根据输入数据的范围（0-1或0-255）选择合适的均值和方差进行处理。

1). ImageNet 数据集的标准均值和方差

均值 (Mean): RGB 通道均值 = (0.485, 0.456, 0.406)
方差 (Standard Deviation): RGB 通道方差 = (0.229, 0.224, 0.225)

2). 说明

这些均值和方差是基于百万张图像计算得出的，通常在训练深度学习模型时使用它们进行标准化处理。
以上均值和方差是针对像素值在 [0, 1] 范围内的图像进行计算的。

3). 对于 [0, 255] 范围的输入

如果输入图像的像素值在 [0, 255] 范围内，可以通过将均值乘以 255 来得到推荐的 RGB 均值：
- R: 0.485 * 255 ≈ 123.68
- G: 0.456 * 255 ≈ 116.78
- B: 0.406 * 255 ≈ 103.94

2.3 读取标签

with open('img/words.txt', 'r') as file:class_names = [line.strip() for line in file]print(class_names)
---
['tench, Tinca tinca', 'goldfish, Carassius auratus'...]

例子原始语句是这样的：

with open("img/words.txt", "r") as f:lines = f.readlines()

使用 .readlines() 方法直接读取文件的所有行到一个列表中。每个列表元素都是一个包含行末换行符的字符串，这种方式比较直接。

使用了列表推导式来读取文件中的每一行，并且立即使用 .strip() 方法去除每行字符串末尾的空白字符（包括换行符\n）。这种方法的优点是代码简洁，且可以在读取每行的同时进行处理，这样可以节省后续可能需要的处理步骤。

2.4 读取图像

查看 img 目录下所有的 JPEG 格式的图片，并打印出来：

image_folder = 'img'
image_paths = [os.path.join(image_folder, i) for i in os.listdir(image_folder) if i.endswith("JPEG")]
image_paths
---
['img/irishterrier-696543.JPEG','img/bellpeppe-994958.JPEG','img/jinrikisha-911722.JPEG','img/greyfox-672194.JPEG']

通过 image_paths[i] 选择一幅图片，并通过 openCV 读取，存入变量 img 中。

img = cv2.imread(image_paths[0])

注意，openCV 读取的图像时 BGR 格式的，需要转换成 RGB 后才能给到模型推理，这个转换过程是在预处理函数 preprocess_fn 的 mean_image_subtraction 子函数中进行的：

def mean_image_subtraction(image, means):B, G, R = cv2.split(image)B = B - means[0]G = G - means[1]R = R - means[2]image = cv2.merge([R, G, B])return image

2.5 获取IO形状

dpu = overlay.runnerinputTensors = dpu.get_input_tensors()
outputTensors = dpu.get_output_tensors()shapeIn = tuple(inputTensors[0].dims)
shapeOut = tuple(outputTensors[0].dims)
outputSize = int(outputTensors[0].get_data_size() / shapeIn[0])softmax = np.empty(outputSize)

在调用 DpuOverlay 的加载 .xmodel 模型后，会自动创建一个 vart.Runner 实例，用于与 vart API通信。而 dpu = overlay.runner 是一个引用的过程。

try:import vart
except:print("Couldn't import vart, check if library installed and is on path.")
...class DpuOverlay(pynq.Overlay):...if not model.endswith(".xmodel"):raise RuntimeError("Currently only xmodel files can be loaded.")else:self.graph = xir.Graph.deserialize(abs_model)subgraphs = get_child_subgraph_dpu(self.graph)assert len(subgraphs) == 1self.runner = vart.Runner.create_runner(subgraphs[0], "run")

2.6 申请内存

output_data = [np.empty(shapeOut, dtype=np.float32, order="C")]
input_data = [np.empty(shapeIn, dtype=np.float32, order="C")]

np.empty 函数用于创建一个未初始化的数组。它的参数包含：

shapeOut、shapeIn：指定数组的形状。
dtype=np.float32：指定数组的数据类型为 32 位浮点数。
order="C"：指定数组的内存布局为 C 风格（行优先）。

问题：以下三种类型的赋值，有什么区别？

-----------------------------------
# 情况一
-----------------------------------
image = input_data[0]
image[0,...] = preprocess_fn(img)-----------------------------------
# 情况二
-----------------------------------
input_data = [[preprocess_fn(img)]]-----------------------------------
# 情况三
-----------------------------------
input_data[0][0]= preprocess_fn(img)

1). 情况一是 PYNQ 例程中原始的赋值方式。首先创建一个具有指定形状和数据类型的未初始化 NumPy 数组input_data ，并将 input_data[0] 放入一个列表中(image)。最终，通过变量引用了数组，preprocessed 函数处理后的数据将赋值给 image 的第一个位置。

情况一比较绕，难以理解。

2). 情况二和三是想简化赋值过程。

情况二是想对 preprocess_fn(img) 结果“升维”，然后将结果赋值给 input_data 变量。

情况三是想对 input_data “降维”，然后 preprocess_fn(img) 将结果赋值给 input_data 变量。

情况二和情况三有重大差别！

情况二中，preprocess_fn(img) 会申请新的内存空间，将其“升维”赋值给 input_data 会导致其原先申请的内存地址变更，即 input_data 会指向 preprocess_fn(img) 所申请的内存空间。这意味着 input_data 现在指向一个新的内存位置，而不是原来的内存位置。

情况三中，preprocess_fn(img) 同样会申请新的内存空间，但是新的内存空间数据会被完全复制到 input_data[0][0]，这个赋值过程也就是深拷贝，而不是对原始图像的引用。

2.7 运行推理

job_id = dpu.execute_async(input_data, output_data)
dpu.wait(job_id)

异步执行和等待任务完成：

dpu.execute_async(input_data, output_data)：用于启动一个异步的 DPU 任务。该函数返回一个 job_id，用于标识这个异步任务。
dpu.wait(job_id)：用于等待指定的异步任务完成。

2.8 后处理

temp = np.reshape(output_data, (-1, 1000))
softmax = np.exp(temp)
predict_label = lines[np.argmax(softmax)-1]print("Classification: {}".format(predict_label))
plt.imshow(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
plt.axis('off')
plt.show()

ResNet50 的输出通常不包含 Softmax 层。ResNet50 的最后一层是一个全连接层，输出的是 logits（未归一化的分数）。在实际应用中，通常会在损失函数中隐式地应用 Softmax。

3. 相关类的介绍

3.1 DpuOverlay 类

继承自 pynq.Overlay，包含四个方法：

初始化：__init__();
重载：download(self);
复制 xclbin：copy_xclbin(self);
加载 xmodel：load_model(self, model);

在 Jupyter Lab 中，通过“显示上下文帮助”查看源码，或者在 KV260 目录查看源码：

/usr/local/share/pynq-venv/lib/python3.10/site-packages/pynq_dpu/dpu.py

class DpuOverlay(pynq.Overlay):"""DPU Overlay该类继承自 PYNQ Overlay。初始化方法类似，但有额外的bit文件搜索路径。"""def __init__(self, bitfile_name, dtbo=None,download=True, ignore_version=False, device=None):"""初始化方法。默认情况下，将在以下路径中搜索比特文件：(1) 本模块内；(2) 绝对路径；(3) 当前工作目录的相对路径。默认情况下，此类将设置运行时为 `dnndk`。"""def download(self):"""下载 Overlay此方法重写了在覆盖类中定义的现有 `download()` 方法。它将下载比特流，设置 AXI 数据宽度，复制 xclbin 和 ML 模型文件。"""def copy_xclbin(self):"""将 xclbin 文件复制到特定位置。此方法将 xclbin 文件复制到目标目录以确保 VART 库可以正常工作。如果未明确设置，xclbin 文件应位于与比特流和 hwh 文件相同的文件夹中。默认的目标文件夹是 `/usr/lib`。"""def load_model(self, model):"""加载 DPU 模型以供 VART 使用。如果未明确设置，ML 模型文件应位于与比特流和 hwh 文件相同的文件夹中。这还将创建一个 vart.Runner 实例，用于与 vart API 通信。参数----------model : strML 模型二进制文件的名称。可以是绝对路径或相对路径。"""...self.runner = vart.Runner.create_runner(subgraphs[0], "run")

3.2 Overlay 类

Overlay 类继承自 Bitstream 类，用于记录单个 bitsteam 的状态和内容。

1). Overlay 类存储四个字典：IP、GPIO、中断控制器和中断引脚字典。

ol.ip_dict
ol.gpio_dict
ol.interrupt_controllers
ol.interrupt_pins

2). Overlay 类的属性

bitfile_name：bitstream 的绝对路径
dtbo：dtbo 文件的绝对路径
ip_dict：来自 PS 的所有可寻址 IP
gpio_dict：所有由 PS 控制的 GPIO 引脚
interrupt_controllers：系统中所有连接到 PS 中断线的 AXI 中断控制器
interrupt_pins：设计中所有连接到中断控制器的引脚
pr_dict：从部分可重新配置的层次块的名称映射到已加载部分位流的字典
device：加载 overlay 的设备

3). Overlay 类源码

class Overlay(Bitstream):def __init__(self, bitfile_name, dtbo=None, download=True, ignore_version=False, device=None, gen_cache=False):super().__init__(bitfile_name, dtbo, partial=False, device=device)...def __getattr__(self, key):if self.is_loaded():return getattr(self._ip_map, key)else:raise RuntimeError("Overlay not currently loaded")...def _deepcopy_dict_from(self, source):...def free(self):...def gen_cache(self):super().gen_cache(self.parser)def download(self, dtbo=None):...def pr_download(self, partial_region, partial_bit, dtbo=None, program=True):...def is_loaded(self):...def reset(self):...def load_ip_data(self, ip_name, data):...def __dir__(self):...def _register_drivers(self):...

3.3 Bitsteam 类

Bitsteam 类是基类，为 Overlay 和 DpuOverlay 提供继承。

Bitstream 类与 .pl_server.device.Device 类之间的关系是组合（composition），而非继承。

Bitstream 类在初始化时会创建一个 Device 类的实例，并通过这个实例调用 Device 类的方法来执行各种操作，如下载比特流、生成缓存、插入和移除设备树覆盖等。

通过组合，Bitstream 类可以使用 Device 类的实例来调用其方法，而不需要继承其属性和方法。

class Bitstream:def __init__(self, bitfile_name, dtbo=None, partial=False, device=None):if not isinstance(bitfile_name, str):raise TypeError("Bitstream name has to be a string.")if device is None:from .pl_server.device import Devicedevice = Device.active_deviceself.device = device...def download(self, parser=None):self.device.download(self, parser)def gen_cache(self, parser=None):self.device.gen_cache(self, parser)def remove_dtbo(self):self.device.remove_device_tree(self.dtbo)def insert_dtbo(self, dtbo=None):if dtbo:resolved_dtbo = _find_dtbo_file(dtbo, self.bitfile_name)if resolved_dtbo:self.dtbo = resolved_dtboelse:raise IOError("DTBO file {} does not exist.".format(dtbo))if not self.dtbo:raise ValueError("DTBO path has to be specified.")self.device.insert_device_tree(self.dtbo)

3.4 Device 类

Device 类继承自元类 DeviceMeta。

Device 类构建一个新的设备实例，并提供一个全局唯一的设备标识符。

class Device(metaclass=DeviceMeta):def __init__(self, tag, warn=False):# Args validationif type(tag) is not str:raise ValueError("Argument 'tag' must be a string")self.tag = tagself.parser = Nonedef set_bitfile_name(self, bitfile_name: str) -> None:self.bitfile_name = bitfile_nameself.parser = self.get_bitfile_metadata(self.bitfile_name)self.mem_dict = self.parser.mem_dictself.ip_dict = self.parser.ip_dictself.gpio_dict = self.parser.gpio_dictself.interrupt_pins = self.parser.interrupt_pinsself.interrupt_controllers = self.parser.interrupt_controllersself.hierarchy_dict = self.parser.hierarchy_dictself.systemgraph = self.parser.systemgraph...

在 Python 中，类可以继承自另一个类，也可以指定一个元类。

1). 普通继承：类直接继承另一个类的属性和方法。

class Parent:def __init__(self):self.value = "I'm the parent"def show(self):print(self.value)class Child(Parent):def __init__(self):super().__init__()self.value = "I'm the child"# 使用示例
child_instance = Child()
child_instance.show()  # 输出: I'm the child

在这个例子中，Child类继承了Parent类的属性和方法。Child类实例化后，可以调用Parent类中的方法，并且可以重写父类的方法。

2). 元类继承：类通过元类来控制其创建过程。

# 定义一个元类
class MyMeta(type):def __init__(cls, name, bases, attrs):print(f"Creating class {name}")super().__init__(name, bases, attrs)# 使用元类创建一个类
class MyClass(metaclass=MyMeta):def __init__(self, value):self.value = valuedef display(self):print(f"Value: {self.value}")# 创建 MyClass 的实例
obj = MyClass(10)
obj.display()---
Creating class MyClass
Value: 10

在这个例子中：

定义元类：
- MyMeta 继承自 type，并重写了 __init__ 方法。在类创建时，它会打印出类的名称。
使用元类：
- MyClass 使用 metaclass=MyMeta 来指定它的元类为 MyMeta。那么在创建 MyClass 时，会调用 MyMeta 的 __init__ 方法。
创建实例：
- 创建 MyClass 的实例 obj，并调用 display 方法。

3.5 DeviceMeta 元类

DeviceMeta 类是所有类型设备的元类，它负责枚举系统中的设备，并选择一个default_device，供不考虑多设备场景的应用程序使用。

DeviceMeta 类主要的实现是 Device 类，每种支持的硬件类型都应该继承该类。每个子类应该有一个_probe_函数，该函数返回一个Device对象数组，以及一个用于确定默认设备的_probe_priority_常量。

class DeviceMeta(type):_subclasses = {}def __init__(cls, name, bases, attrs):if "_probe_" in attrs:priority = attrs["_probe_priority_"]if (priority in DeviceMeta._subclassesand DeviceMeta._subclasses[priority].__name__ != name):raise RuntimeError("Multiple Device subclasses with same priority")DeviceMeta._subclasses[priority] = clssuper().__init__(name, bases, attrs)...

3.6 type 类

在 Python 中，type 是所有类的元类（metaclass）。当你定义一个类时，实际上是通过 type 来创建这个类的。元类允许你在类创建时自定义类的行为和属性。

type 类的核心功能：

动态创建和初始化类。
提供类的元数据（如基类、大小、模块、名称等）。
支持类的调用和检查操作。
管理类的继承关系和方法解析顺序。
支持类型联合操作和类型参数。

type 类属性和方法

1)基本属性：

__base__：返回类的直接基类，如果没有基类则返回 None。
__bases__：返回类的所有基类组成的元组。
__basicsize__：返回类的基本大小（以字节为单位）。
__dict__：返回类的属性字典。
__dictoffset__：返回类的字典偏移量。
__flags__：返回类的标志位。
__itemsize__：返回类的项大小。
__module__：返回类所在的模块名。
__mro__：返回类的继承顺序（方法解析顺序）。
__name__：返回类的名称。
__qualname__：返回类的限定名称。
__text_signature__：返回类的文本签名。
__weakrefoffset__：返回类的弱引用偏移量。

2). 构造方法：

__init__：用于初始化类的实例。
__new__：用于创建类的实例。

3). 调用和检查方法：

__call__：使类的实例可以像函数一样被调用。
__subclasses__：返回类的所有子类。
mro：返回类的继承顺序列表。
__instancecheck__：检查实例是否属于类。
__subclasscheck__：检查子类是否属于类。

4). 类方法：

__prepare__：用于准备类的命名空间。

5). 运算符重载（Python 3.10 及以上版本）：

__or__ 和 __ror__：用于类型联合操作。

6). 类型参数（Python 3.12 及以上版本）：

__type_params__：返回类型参数的元组。

3.7 VART

3.7.1 vart*.so

在 KV260 中，查看 VART 的位置：

import vart
import inspect
import osmodule_location = os.path.dirname(inspect.getfile(vart))
print(module_location)
---
'/usr/local/lib/python3.10/dist-packages'ls -l /usr/local/lib/python3.10/dist-packages
---
vaitrace_py
vart.cpython-310-aarch64-linux-gnu.so
xir.cpython-310-aarch64-linux-gnu.so

可以看到 vart 全名为：vart.cpython-310-aarch64-linux-gnu.so

dist-packages：用于系统自带的 Python 版本。系统自带的软件管理器（如apt、yum等）安装的Python包会放在这个目录中。
site-packages：用于用户手动安装的 Python 版本。通过pip或其他包管理工具安装的第三方库通常会放在这个目录中。

3.7.2 vart*.cpp

<Vitis-AI-2.5>/src/Vitis-AI-Runtime/VART/vart/runner/python/runner_py_module.cpp

主要功能：

1). TensorBuffer 类

vart::TensorBuffer 是存储神经网络输入和输出张量的类。张量（tensor）是多维数组，代表模型的输入或输出数据。
CpuFlatTensorBuffer 是 TensorBuffer 的一个具体实现，负责将张量数据从Python的numpy数组转换为可用于Vitis AI推理的格式。
CpuFlatTensorBuffer 类中的 data 方法负责计算数据在内存中的地址和大小，用于访问张量的具体内容。

2). 输入与输出的处理

代码定义了函数如 array_to_tensor_buffer 和 dynamic_array_to_tensor_buffer，用于将Python 中的 numpy 数组（或其他缓冲区）转换为 TensorBuffer，方便推理时使用。
这些函数利用了pybind11将Python的numpy数组格式与Vitis AI要求的C++张量格式桥接。

3). 异步推理
Runner 类的 execute_async 方法可以异步执行推理任务，并且支持动态输入形状的处理。
任务执行完后，可以通过 wait 方法等待任务完成并清理资源。

4). 内存管理

代码中使用了 std::shared_ptr 和 WeakSingleton 来管理 TensorBuffer 的生命周期，确保在推理任务执行时合理分配和回收内存。
save_to_map 方法用于将 TensorBuffer 保存到一个全局的共享映射中，方便在推理结束时释放相关资源。

5). Python 绑定

PYBIND11_MODULE 宏定义了一个 Python 模块接口，名字为 vart，其中导出了多个 C++ 类和方法，使得 Python 端可以直接调用 C++ 中的这些功能。
pybind11 用于处理 C++ 和 Python 之间的数据转换，并使得 Python 代码可以方便地访问TensorBuffer、Runner 类的接口。

改 cpp 是通过 CMake 构建的，如下：

<Vitis-AI-2.5>/src/Vitis-AI-Runtime/VART/vart/runner/CMakeLists.txtif(BUILD_PYTHON)vai_add_pybind11_module(py_runner MODULE_NAME vartpython/runner_py_module.cpp)target_link_libraries(py_runner PRIVATE ${PROJECT_NAME}::util${PROJECT_NAME}::${COMPONENT_NAME})
endif(BUILD_PYTHON)

根据 BUILD_PYTHON 变量来决定是否构建模块。
MODULE_NAME vart: 指定生成的 Python 模块的名称为 vart。
python/runner_py_module.cpp: 生成模块所需的源文件。