#!/usr/bin/env python3

################################################################################
# Copyright (c) 2024,D-Robotics.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################

from hobot_dnn 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/mobilenetv1_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:])

    print("=" * 10, "Classification result", "=" * 10)
    # 遍历每一个结果
    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}")

完整代码通俗解析

基础信息

1. 运行平台：地平线 RDK（X3/X5）
2. 模型文件：mobilenetv1_224x224_nv12.bin
   • MobileNetV1 图像分类网络
   • 输入尺寸：224×224，输入格式 NV12
   • .bin：地平线 OX 编译完成、面向 BPU 的 INT8 离线模型
3. 核心架构
   • hobot_dnn：调用 BPU 硬件推理
   • libpostprocess.so：地平线闭源 C 动态库，提供分类后处理
   • ctypes：Python ↔ C 库互通（传递张量指针、结构体）
4. 任务：读取本地图片 → 预处理转为 NV12 → BPU 推理 → C 库后处理 → 输出分类 Top-k 结果

和上一份 FCOS 目标检测 Demo 是同源框架，区别仅在于： FCOS = 目标检测；这份 = 图像分类，结构体、后处理函数名全部换成 Classification 分类系列。

一、头部导入 & C 结构体定义

python

运行

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)
    ]

✅ 作用： Python 模拟 C 语言结构体内存布局，用来给libpostprocess.so（C代码）传递参数。

注意：分类任务不需要 NMS，但结构体复用了和检测一样的模板，所以依然保留nms_threshold等字段，代码里直接填 0 即可。

python

运行

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

加载地平线闭源后处理库，调用两个核心 C 函数：

1. ClassificationDoProcess：解析网络输出张量
2. ClassificationPostProcess：返回 JSON 格式分类结果字符串

其余结构体 hbSysMem_t / hbDNNTensor_t 和 FCOS 代码完全一致，用途：封装 BPU 推理输出张量的内存地址、量化参数、shape，传给 C 后处理。

二、公共工具函数

1. bgr2nv12_opencv(image)

和之前 FCOS 代码一模一样。 输入 OpenCV BGR 图片 → 转换成一维 NV12 uint8 数组，满足模型xxx_nv12.bin输入要求。

2. print_properties(pro)

打印 tensor 信息：类型、数据格式、排布（NCHW/NHWC）、shape，调试用。

3. get_hw(pro)

自动根据 tensor 布局提取高宽：

• NCHW → shape[2]=H shape[3]=W
• NHWC → shape[1]=H shape[2]=W

三、主程序逐段解析

python

运行

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

加载地平线编译好的 BPU 离线模型，只能在 RDK 开发板运行。

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)

流水线：

1. 读取本地图片（BGR HWC）
2. 获取模型输入尺寸：224×224
3. resize 原图到 224×224
4. BGR → NV12，准备送入 BPU

python

运行

outputs = models[0].forward(nv12_data)

BPU 硬件推理 输入一维 NV12 buffer，返回推理输出张量（MobileNetV1 分类输出：各类别概率 logits）。

填充后处理参数结构体

python

运行

classification_postprocess_info = ClassificationPostProcessInfo_t()
classification_postprocess_info.height = h        # 模型输入尺寸224
classification_postprocess_info.width = w
classification_postprocess_info.ori_height = org_height # 原始图片分辨率
classification_postprocess_info.ori_width = org_width
classification_postprocess_info.score_threshold = 0.3   # 只返回置信度>0.3类别
classification_postprocess_info.nms_threshold = 0       # 分类任务不用NMS，置0
classification_postprocess_info.nms_top_k = 500        # 最多返回前500个类别
classification_postprocess_info.is_pad_resize = 0       # resize直接拉伸，不做等比例padding

构建 output_tensors，向 C 库传递张量指针

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模型
        output_tensors[i].properties.quantiType = 0
        output_tensors[i].sysMem[0].virAddr = ctypes.cast(...) # float内存指针
    else:
        # INT8量化模型（当前mobilenetv1_224x224_nv12.bin属于此类）
        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(...) # int32量化数据指针

    # 填充tensor维度信息
    for j in range(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)

核心逻辑：

1. outputs是 BPU 推理结果（INT8 量化数据）
2. 通过 ctypes 拿到 numpy 数组底层虚拟内存地址
3. 把地址、量化 scale、shape 塞进 C 结构体
4. 调用ClassificationDoProcess，C 库内部完成：INT8 反量化、softmax 计算

关键点：模型是 PTQ 静态量化 INT8，网络输出不是概率，是量化后整数；反量化、softmax 全部在闭源libpostprocess.so内部完成，Python 看不到这一步实现。

获取解析结果

python

运行

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

1. C 库返回 char * 字符串
2. result_str[25:]：跳过头部一段无关前缀字符，截取合法 JSON
3. JSON 示例格式：

json

[
 {"label":340, "prob":0.96, "class_name":"zebra"},
 {"label":123, "prob":0.02, "class_name":"horse"}
]

打印输出

python

运行

for result in data:
    prob = result['prob']
    label = result['label']
    name = result['class_name']
    print(f"cls id: {label}, Confidence: {prob}, class_name: {name}")

四、完整数据流总图

plaintext

zebra_cls.jpg → cv2.imread → BGR(HWC)
        ↓ cv2.resize → 224×224 BGR
        ↓ bgr2nv12_opencv → NV12一维数组
        ↓ models[0].forward() → BPU推理
        ↓ outputs：INT8量化张量
        ↓ ctypes传递内存指针 → libpostprocess.so
        ↓ C库：反量化 + softmax 过滤低置信度类别
        ↓ 返回JSON字符串
        ↓ Python解析打印分类ID、置信度、类别名

五、和 FCOS 目标检测 Demo 关键差异汇总

表格

项目	FCOS 检测 Demo	MobileNet 分类 Demo
结构体	FcosPostProcessInfo_t	ClassificationPostProcessInfo_t
C 库函数	FcosdoProcess / FcosPostProcess	ClassificationDoProcess / ClassificationPostProcess
网络输出	15 个 tensor（5 层偏移 / 置信度 / 中心度）	1 个输出 tensor（类别 logits）
后处理逻辑	解码框 + NMS	反量化 + softmax，无 NMS
画面绘制	draw_bboxs 画框、HDMI 输出	仅本地图片推理，无摄像头、无 HDMI 显示代码

六、重点坑点

1. 依赖 libpostprocess.so 精简固件如果缺少该库，直接运行崩溃；脱离该库，需要自行实现 INT8 反量化 + softmax。
2. result_str[25:] 切片不能写错 切片数字由后处理 so 版本决定，版本改动前缀长度变化会导致 json.loads 报错。
3. NV12 格式硬性要求 图片 resize 后宽高必须偶数（224 满足），格式错误推理结果错乱。
4. 量化模式匹配 quantiType=2对应 INT8 模型；如果是 FP32 模型分支走另一条代码。

七、拓展改造方向

1. 从本地图片 → 修改成 USB 摄像头实时流
2. 移除 libpostprocess.so 依赖，Python 手动实现 MobileNet 后处理
3. 添加结果可视化，在图片上打印类别名称和置信度
4. 批量推理文件夹内所有图片