[具身智能-631]：RDK X5 地平线 ResNet18 分类 Python 示例完整解析

文火冰糖的硅基工坊

60人浏览 · 2026-07-23 22:18:29

文火冰糖的硅基工坊 · 2026-07-23 22:18:29 发布

ry:
    from hobot_dnn import pyeasy_dnn as dnn
except ImportError:
    from hobot_dnn_rdkx5 import pyeasy_dnn as dnn
import numpy as np
import cv2

import time
import ctypes
import json

output_tensors = None

fcos_postprocess_info = None

class hbSysMem_t(ctypes.Structure):
    _fields_ = [
        ("phyAddr",ctypes.c_double),
        ("virAddr",ctypes.c_void_p),
        ("memSize",ctypes.c_int)
    ]

class hbDNNQuantiShift_yt(ctypes.Structure):
    _fields_ = [
        ("shiftLen",ctypes.c_int),
        ("shiftData",ctypes.c_char_p)
    ]

class hbDNNQuantiScale_t(ctypes.Structure):
    _fields_ = [
        ("scaleLen",ctypes.c_int),
        ("scaleData",ctypes.POINTER(ctypes.c_float)),
        ("zeroPointLen",ctypes.c_int),
        ("zeroPointData",ctypes.c_char_p)
    ]

class hbDNNTensorShape_t(ctypes.Structure):
    _fields_ = [
        ("dimensionSize",ctypes.c_int * 8),
        ("numDimensions",ctypes.c_int)
    ]

class hbDNNTensorProperties_t(ctypes.Structure):
    _fields_ = [
        ("validShape",hbDNNTensorShape_t),
        ("alignedShape",hbDNNTensorShape_t),
        ("tensorLayout",ctypes.c_int),
        ("tensorType",ctypes.c_int),
        ("shift",hbDNNQuantiShift_yt),
        ("scale",hbDNNQuantiScale_t),
        ("quantiType",ctypes.c_int),
        ("quantizeAxis", ctypes.c_int),
        ("alignedByteSize",ctypes.c_int),
        ("stride",ctypes.c_int * 8)
    ]

class hbDNNTensor_t(ctypes.Structure):
    _fields_ = [
        ("sysMem",hbSysMem_t * 4),
        ("properties",hbDNNTensorProperties_t)
    ]


class ClassificationPostProcessInfo_t(ctypes.Structure):
    _fields_ = [
        ("height",ctypes.c_int),
        ("width",ctypes.c_int),
        ("ori_height",ctypes.c_int),
        ("ori_width",ctypes.c_int),
        ("score_threshold",ctypes.c_float),
        ("nms_threshold",ctypes.c_float),
        ("nms_top_k",ctypes.c_int),
        ("is_pad_resize",ctypes.c_int)
    ]


libpostprocess = ctypes.CDLL('/usr/lib/libpostprocess.so')

get_Postprocess_result = libpostprocess.ClassificationPostProcess
get_Postprocess_result.argtypes = [ctypes.POINTER(ClassificationPostProcessInfo_t)]
get_Postprocess_result.restype = ctypes.c_char_p

def get_TensorLayout(Layout):
    if Layout == "NCHW":
        return int(2)
    else:
        return int(0)

def bgr2nv12_opencv(image):
    height, width = image.shape[0], image.shape[1]
    area = height * width
    yuv420p = cv2.cvtColor(image, cv2.COLOR_BGR2YUV_I420).reshape((area * 3 // 2,))
    y = yuv420p[:area]
    uv_planar = yuv420p[area:].reshape((2, area // 4))
    uv_packed = uv_planar.transpose((1, 0)).reshape((area // 2,))

    nv12 = np.zeros_like(yuv420p)
    nv12[:height * width] = y
    nv12[height * width:] = uv_packed
    return nv12

def print_properties(pro):
    print("tensor type:", pro.tensor_type)
    print("data type:", pro.dtype)
    print("layout:", pro.layout)
    print("shape:", pro.shape)


def get_hw(pro):
    if pro.layout == "NCHW":
        return pro.shape[2], pro.shape[3]
    else:
        return pro.shape[1], pro.shape[2]


if __name__ == '__main__':
    # test classification result
    models = dnn.load('../models/resnet18_224x224_nv12.bin')
    # test input and output properties
    print("=" * 10, "inputs[0] properties", "=" * 10)
    print_properties(models[0].inputs[0].properties)
    print("inputs[0] name is:", models[0].inputs[0].name)

    print("=" * 10, "outputs[0] properties", "=" * 10)
    print_properties(models[0].outputs[0].properties)
    print("outputs[0] name is:", models[0].outputs[0].name)


    img_file = cv2.imread('./zebra_cls.jpg')
    h, w = get_hw(models[0].inputs[0].properties)
    des_dim = (w, h)
    resized_data = cv2.resize(img_file, des_dim, interpolation=cv2.INTER_AREA)
    nv12_data = bgr2nv12_opencv(resized_data)

    outputs = models[0].forward(nv12_data)

    t0 = time.time()
    # 获取结构体信息
    classification_postprocess_info = ClassificationPostProcessInfo_t()
    classification_postprocess_info.height = h
    classification_postprocess_info.width = w
    org_height, org_width = img_file.shape[0:2]
    classification_postprocess_info.ori_height = org_height
    classification_postprocess_info.ori_width = org_width
    classification_postprocess_info.score_threshold = 0.3
    classification_postprocess_info.nms_threshold = 0
    classification_postprocess_info.nms_top_k = 500
    classification_postprocess_info.is_pad_resize = 0

    output_tensors = (hbDNNTensor_t * len(models[0].outputs))()
    for i in range(len(models[0].outputs)):
        output_tensors[i].properties.tensorLayout = get_TensorLayout(outputs[i].properties.layout)
        # print(output_tensors[i].properties.tensorLayout)
        if (len(outputs[i].properties.scale_data) == 0):
            output_tensors[i].properties.quantiType = 0
            output_tensors[i].sysMem[0].virAddr = ctypes.cast(outputs[i].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_float)), ctypes.c_void_p)
        else:
            output_tensors[i].properties.quantiType = 2
            output_tensors[i].properties.scale.scaleData = outputs[i].properties.scale_data.ctypes.data_as(ctypes.POINTER(ctypes.c_float))
            output_tensors[i].sysMem[0].virAddr = ctypes.cast(outputs[i].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_int32)), ctypes.c_void_p)

        for j in range(len(outputs[i].properties.shape)):
            output_tensors[i].properties.validShape.numDimensions = len(outputs[i].properties.shape)
            output_tensors[i].properties.validShape.dimensionSize[j] = outputs[i].properties.shape[j]

        libpostprocess.ClassificationDoProcess(output_tensors[i], ctypes.pointer(classification_postprocess_info), i)

    result_str = get_Postprocess_result(ctypes.pointer(classification_postprocess_info))
    result_str = result_str.decode('utf-8')
    t1 = time.time()
    print("postprocess time is :", (t1 - t0))

    # draw result
    # 解析JSON字符串
    data = json.loads(result_str[25:])

    # 遍历每一个结果
    for result in data:
        prob = result['prob']  # 得分
        label = result['label']  # id
        name = result['class_name']  # 类别名称

        # 打印信息
        print(f"cls id: {label}, Confidence: {prob}, class_name: {name}")

RDK X5 地平线 ResNet18 分类 Python 示例完整解析

这份代码是地平线 RDK 系列（RDK X3/X5） 基于 hobot_dnn 部署 NV12 输入 ResNet18 图像分类模型的工程示例，核心特点：

使用地平线 BPU 原生模型 resnet18_224x224_nv12.bin（模型输入格式为 NV12，不需要在 Python 做归一化、减均值；预处理放在模型内部）
Python 调用 C++ 动态库 libpostprocess.so 完成后处理，不使用 Python 实现 softmax，性能更高
通过ctypes构造 C 语言结构体，桥接 Python 与底层后处理库

运行环境：RDK X5 Ubuntu 系统；依赖地平线 hobot-dnn、OpenCV、numpy

一、整体代码架构分层

plaintext

【头文件&库导入】
    → 自动兼容hobot_dnn / hobot_dnn_rdkx5（X5专用包）
【ctypes C结构体定义】
    → 对齐libpostprocess.so内部C结构体，用于传参
【工具函数】
    NV12转换、Layout转换、模型信息打印
【主流程】
    1. 加载BPU模型
    2. 读取图像 → resize → BGR转NV12
    3. dnn.forward 推理（BPU硬件运行）
    4. 组装输出tensor结构体，调用C后处理
    5. 解析JSON输出分类结果

二、逐段详细解析

1. 版权与模块导入

python

运行

try:
    from hobot_dnn import pyeasy_dnn as dnn
except ImportError:
    from hobot_dnn_rdkx5 import pyeasy_dnn as dnn

RDK X3：包名 hobot_dnn
RDK X5：包名 hobot_dnn_rdkx5 自动兼容两个平台，不用修改代码

python

运行

import numpy as np
import cv2
import time
import ctypes
import json

ctypes：重中之重！Python 调用 C 动态库，必须构造和 C 头文件完全一致的结构体

2. ctypes 结构体定义（核心难点）

所有class XXX(ctypes.Structure) 严格对齐 libpostprocess.so C 头文件内存布局

python

运行

class hbSysMem_t(ctypes.Structure):
    _fields_ = [
        ("phyAddr",ctypes.c_double),
        ("virAddr",ctypes.c_void_p),
        ("memSize",ctypes.c_int)
    ]

地平线系统内存结构体：

virAddr：虚拟地址（Python numpy buffer 内存指针，传给 C 库访问推理输出）
phyAddr：物理地址（BPU 硬件地址，示例里未使用）

python

运行

class hbDNNTensor_t(ctypes.Structure):
    _fields_ = [
        ("sysMem",hbSysMem_t * 4),
        ("properties",hbDNNTensorProperties_t)
    ]

地平线 DNN 输出 Tensor 结构体：承载模型推理结果、量化参数、维度、排布格式。

python

运行

class ClassificationPostProcessInfo_t(ctypes.Structure):
    _fields_ = [
        ("height",ctypes.c_int),       # 模型输入尺寸 h
        ("width",ctypes.c_int),        # 模型输入尺寸 w
        ("ori_height",ctypes.c_int),   # 原图高
        ("ori_width",ctypes.c_int),    # 原图宽
        ("score_threshold",ctypes.c_float), # 置信度阈值（分类基本不用）
        ("nms_threshold",ctypes.c_float),  # NMS阈值，分类任务置0
        ("nms_top_k",ctypes.c_int),
        ("is_pad_resize",ctypes.c_int) # 是否等比例padding缩放（0=直接拉伸resize）
    ]

分类后处理配置结构体，传递给 C 后处理库。

3. 加载后处理动态库

python

运行

libpostprocess = ctypes.CDLL('/usr/lib/libpostprocess.so')
get_Postprocess_result = libpostprocess.ClassificationPostProcess
get_Postprocess_result.argtypes = [ctypes.POINTER(ClassificationPostProcessInfo_t)]
get_Postprocess_result.restype = ctypes.c_char_p

打开 /usr/lib/libpostprocess.so 地平线官方后处理库
指定入参类型、返回值类型（C 字符串 char*）

重点：ctypes 默认不会做类型检查，不写argtypes极易出现内存越界、段错误

4. 工具函数详解

4.1 `get_TensorLayout`

python

运行

def get_TensorLayout(Layout):
    if Layout == "NCHW":
        return int(2)
    else:
        return int(0)

地平线 BPU tensor 布局编码：

NHWC → 0
NCHW → 2 用于填充hbDNNTensor_t.properties.tensorLayout传给 C 库

4.2 `bgr2nv12_opencv(image)`

python

运行

def bgr2nv12_opencv(image):
    height, width = image.shape[0], image.shape[1]
    area = height * width
    yuv420p = cv2.cvtColor(image, cv2.COLOR_BGR2YUV_I420).reshape((area * 3 // 2,))
    y = yuv420p[:area]
    uv_planar = yuv420p[area:].reshape((2, area // 4))
    uv_packed = uv_planar.transpose((1, 0)).reshape((area // 2,))

    nv12 = np.zeros_like(yuv420p)
    nv12[:height * width] = y
    nv12[height * width:] = uv_packed
    return nv12

关键：模型输入是 NV12 格式！

OpenCV 读取图片默认 BGR
先转 I420 (YUV420 planar)：YYYY UUUU VVVV
转换成 NV12：YYYY UVUVUVUV

很多新手踩坑：模型是 nv12 输入，直接喂 BGR/RGB 图像，识别完全错乱！

4.3 `get_hw(pro)`

自动根据张量布局 NHWC/NCHW 获取高宽

python

运行

def get_hw(pro):
    if pro.layout == "NCHW":
        return pro.shape[2], pro.shape[3] # NCHW [N,C,H,W]
    else:
        return pro.shape[1], pro.shape[2] # NHWC [N,H,W,C]

5. Main 主推理流程（最重要）

5.1 加载 BPU 模型

python

运行

models = dnn.load('../models/resnet18_224x224_nv12.bin')

dnn.load 加载地平线 *.bin 离线模型，模型已经完成量化编译，适配 BPU
models [0] 代表第一个模型（支持加载多模型）
打印输入输出 tensor 属性：layout、shape、量化信息

5.2 图像预处理流水线

python

运行

img_file = cv2.imread('./zebra_cls.jpg')
h, w = get_hw(models[0].inputs[0].properties)
des_dim = (w, h)
resized_data = cv2.resize(img_file, des_dim, interpolation=cv2.INTER_AREA)
nv12_data = bgr2nv12_opencv(resized_data)

流程：原图 → resize 到模型输入尺寸 (224,224) → BGR→NV12

5.3 BPU 推理前向

python

运行

outputs = models[0].forward(nv12_data)

forward() 数据直接送入 BPU 硬件运算，输出 dnn.Tensor 对象，包含推理结果 buffer、量化参数、shape、layout。

5.4 构造 C 库需要的输出 Tensor（ctypes 核心操作）

python

运行

output_tensors = (hbDNNTensor_t * len(models[0].outputs))()
for i in range(len(models[0].outputs)):
    output_tensors[i].properties.tensorLayout = get_TensorLayout(outputs[i].properties.layout)
    
    # 判断模型输出是否量化
    if (len(outputs[i].properties.scale_data) == 0):
        # FP32输出，quantiType=0
        output_tensors[i].properties.quantiType = 0
        output_tensors[i].sysMem[0].virAddr = ctypes.cast(
            outputs[i].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_float)), 
            ctypes.c_void_p
        )
    else:
        # INT8量化输出 quantiType=2
        output_tensors[i].properties.quantiType = 2
        output_tensors[i].properties.scale.scaleData = outputs[i].properties.scale_data.ctypes.data_as(ctypes.POINTER(ctypes.c_float))
        output_tensors[i].sysMem[0].virAddr = ctypes.cast(
            outputs[i].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_int32)), 
            ctypes.c_void_p
        )
    # 填充维度信息
    for j in range(len(outputs[i].properties.shape)):
        output_tensors[i].properties.validShape.numDimensions = len(outputs[i].properties.shape)
        output_tensors[i].properties.validShape.dimensionSize[j] = outputs[i].properties.shape[j]

    # 调用C后处理函数 ClassificationDoProcess
    libpostprocess.ClassificationDoProcess(output_tensors[i], ctypes.pointer(classification_postprocess_info), i)

逻辑拆解：

创建hbDNNTensor_t数组，对齐 C 库需要的 tensor 数组
区分 FP32 / INT8 量化输出：
- 有 scale_data：INT8 量化，需要把 scale/zero_point 传给 C 库做反量化
- 无 scale_data：FP32 浮点输出
outputs[i].buffer.ctypes.data_as()：获取 numpy 数组内存指针，直接共享内存，无数据拷贝
调用ClassificationDoProcess，在 C 库内部完成 softmax、类别解析

5.5 获取结果并解析 JSON

python

运行

result_str = get_Postprocess_result(ctypes.pointer(classification_postprocess_info))
result_str = result_str.decode('utf-8')
data = json.loads(result_str[25:])

C 函数返回char*，ctypes 拿到 bytes，decode 转字符串
result_str[25:]：C 返回字符串头部带固定前缀，需要截断才能正常 json.loads（地平线官方库固定格式）

5.6 打印分类输出

python

运行

for result in data:
    prob = result['prob']      # softmax置信度
    label = result['label']     # 类别ID
    name = result['class_name'] # 类别名称
    print(f"cls id: {label}, Confidence: {prob}, class_name: {name}")