Run Custom Models on the Board

Overview

After completing model conversion, the next step is to deploy and run custom models on the MB-E30P development board. This chapter describes the two main deployment methods in detail and demonstrates how to write a complete image classification application through a hands-on example.

5.1 Comparison of the Two Deployment Methods

Python API Deployment

Advantages

• High development efficiency: Python has concise syntax and fast development speed
• Convenient debugging: Rich debugging tools and library support
• Rich ecosystem: A large number of third-party libraries are directly usable
• Prototype validation: Suitable for rapid prototyping and algorithm validation
• Strong flexibility: Easy to modify and extend functionality

Disadvantages

• Performance overhead: Python is interpreted and has performance losses
• Memory usage: Relatively more memory usage than C++
• Startup time: Interpreter startup and module loading take time
• Complex dependencies: Requires a Python runtime environment and related libraries

Applicable Scenarios

# 适合以下场景:
scenarios = [
    "算法原型验证",
    "快速功能演示",
    "教学和学习",
    "复杂数据处理",
    "与其他Python服务集成",
    "对性能要求不严格的应用"
]

C/C++ API Deployment

Advantages

• Excellent performance: Compiled execution with high running efficiency
• Memory efficiency: Low memory usage and high resource utilization
• Fast startup: No interpreter overhead, short startup time
• Simple deployment: Compiled executable files with few dependencies
• System integration: Easy to integrate with system services and other C/C++ programs

Disadvantages

• Development complexity: Need to handle memory management, pointers, etc.
• Difficult debugging: Debugging tools are relatively limited
• Development cycle: The compile-test cycle is longer
• Flexibility: Modifying functionality requires recompilation

Applicable Scenarios

// 适合以下场景:
std::vector<std::string> scenarios = {
    "生产环境部署",
    "实时性要求高的应用",
    "资源受限的嵌入式系统",
    "系统服务和守护进程",
    "与硬件底层交互",
    "大规模批量处理"
};

Performance Comparison Test

# 性能测试脚本
import time
import numpy as np
from rknnlite.api import RKNNLite

def benchmark_python_api(model_path, test_data, iterations=100):
    """Python API性能测试"""
    rknn = RKNNLite()

    # 加载模型
    start_time = time.time()
    ret = rknn.load_rknn(model_path)
    load_time = time.time() - start_time

    if ret != 0:
        print("加载模型失败")
        return None

    # 初始化运行时
    start_time = time.time()
    ret = rknn.init_runtime()
    init_time = time.time() - start_time

    if ret != 0:
        print("初始化运行时失败")
        return None

    # 推理性能测试
    inference_times = []

    for i in range(iterations):
        start_time = time.time()
        outputs = rknn.inference(inputs=[test_data])
        inference_time = time.time() - start_time
        inference_times.append(inference_time)

    # 计算统计信息
    avg_inference_time = np.mean(inference_times)
    min_inference_time = np.min(inference_times)
    max_inference_time = np.max(inference_times)
    std_inference_time = np.std(inference_times)

    # 释放资源
    rknn.release()

    results = {
        'load_time': load_time,
        'init_time': init_time,
        'avg_inference_time': avg_inference_time,
        'min_inference_time': min_inference_time,
        'max_inference_time': max_inference_time,
        'std_inference_time': std_inference_time,
        'fps': 1.0 / avg_inference_time
    }

    return results

def print_benchmark_results(python_results, cpp_results=None):
    """打印性能测试结果"""
    print("=" * 60)
    print("性能测试结果对比")
    print("=" * 60)

    print(f"\nPython API 性能:")
    print(f"  模型加载时间: {python_results['load_time']:.4f}s")
    print(f"  运行时初始化: {python_results['init_time']:.4f}s")
    print(f"  平均推理时间: {python_results['avg_inference_time']:.4f}s")
    print(f"  最小推理时间: {python_results['min_inference_time']:.4f}s")
    print(f"  最大推理时间: {python_results['max_inference_time']:.4f}s")
    print(f"  推理时间标准差: {python_results['std_inference_time']:.4f}s")
    print(f"  平均FPS: {python_results['fps']:.2f}")

    if cpp_results:
        print(f"\nC++ API 性能:")
        print(f"  模型加载时间: {cpp_results['load_time']:.4f}s")
        print(f"  运行时初始化: {cpp_results['init_time']:.4f}s")
        print(f"  平均推理时间: {cpp_results['avg_inference_time']:.4f}s")
        print(f"  平均FPS: {cpp_results['fps']:.2f}")

        print(f"\n性能提升:")
        speedup = python_results['avg_inference_time'] / cpp_results['avg_inference_time']
        print(f"  C++ vs Python 推理速度提升: {speedup:.2f}x")

        fps_improvement = (cpp_results['fps'] - python_results['fps']) / python_results['fps'] * 100
        print(f"  FPS 提升: {fps_improvement:.1f}%")

# 使用示例
if __name__ == "__main__":
    model_path = "models/resnet18_rk3568.rknn"
    test_data = np.random.rand(1, 3, 224, 224).astype(np.float32)

    python_results = benchmark_python_api(model_path, test_data)
    print_benchmark_results(python_results)

5.2 Python API Deployment Explanation

Basic API Usage

Core Classes and Methods

from rknnlite.api import RKNNLite
import numpy as np
import cv2

class RKNNInference:
    """RKNN推理封装类"""

    def __init__(self, model_path, verbose=True):
        self.model_path = model_path
        self.rknn = RKNNLite(verbose=verbose)
        self.is_loaded = False
        self.is_initialized = False

    def load_model(self):
        """加载RKNN模型"""
        print(f"加载模型: {self.model_path}")

        ret = self.rknn.load_rknn(self.model_path)
        if ret != 0:
            raise RuntimeError(f"加载模型失败，错误码: {ret}")

        self.is_loaded = True
        print("模型加载成功")

    def init_runtime(self, target='rk3568', device_id=None):
        """初始化运行时环境"""
        if not self.is_loaded:
            raise RuntimeError("请先加载模型")

        print("初始化运行时环境...")

        ret = self.rknn.init_runtime(target=target, device_id=device_id)
        if ret != 0:
            raise RuntimeError(f"初始化运行时失败，错误码: {ret}")

        self.is_initialized = True
        print("运行时初始化成功")

    def inference(self, input_data):
        """执行推理"""
        if not self.is_initialized:
            raise RuntimeError("请先初始化运行时")

        # 确保输入数据格式正确
        if isinstance(input_data, np.ndarray):
            inputs = [input_data]
        elif isinstance(input_data, list):
            inputs = input_data
        else:
            raise ValueError("输入数据必须是numpy数组或数组列表")

        # 执行推理
        outputs = self.rknn.inference(inputs=inputs)

        if outputs is None:
            raise RuntimeError("推理失败")

        return outputs

    def get_model_info(self):
        """获取模型信息"""
        if not self.is_loaded:
            raise RuntimeError("请先加载模型")

        # 获取输入输出信息
        input_info = self.rknn.get_input_info()
        output_info = self.rknn.get_output_info()

        return {
            'input_info': input_info,
            'output_info': output_info
        }

    def release(self):
        """释放资源"""
        if hasattr(self, 'rknn') and self.rknn:
            self.rknn.release()
            print("资源释放完成")

# 使用示例
def basic_usage_example():
    """基础使用示例"""
    model_path = "models/resnet18_rk3568.rknn"

    # 创建推理对象
    inference = RKNNInference(model_path)

    try:
        # 加载模型
        inference.load_model()

        # 初始化运行时
        inference.init_runtime()

        # 获取模型信息
        model_info = inference.get_model_info()
        print("模型信息:", model_info)

        # 准备测试数据
        test_data = np.random.rand(1, 3, 224, 224).astype(np.float32)

        # 执行推理
        outputs = inference.inference(test_data)
        print(f"推理输出形状: {[output.shape for output in outputs]}")

    finally:
        # 释放资源
        inference.release()

if __name__ == "__main__":
    basic_usage_example()

Image Preprocessing Module

import cv2
import numpy as np
from typing import Tuple, List, Optional

class ImagePreprocessor:
    """图像预处理类"""

    def __init__(self, target_size=(224, 224), mean=(0.485, 0.456, 0.406),
                 std=(0.229, 0.224, 0.225), bgr_to_rgb=True):
        self.target_size = target_size
        self.mean = np.array(mean, dtype=np.float32)
        self.std = np.array(std, dtype=np.float32)
        self.bgr_to_rgb = bgr_to_rgb

    def resize_image(self, image: np.ndarray, keep_ratio: bool = True) -> np.ndarray:
        """调整图像大小"""
        h, w = image.shape[:2]
        target_h, target_w = self.target_size

        if keep_ratio:
            # 保持宽高比的缩放
            scale = min(target_w / w, target_h / h)
            new_w, new_h = int(w * scale), int(h * scale)

            # 缩放图像
            resized = cv2.resize(image, (new_w, new_h), interpolation=cv2.INTER_LINEAR)

            # 创建目标大小的画布
            canvas = np.zeros((target_h, target_w, 3), dtype=image.dtype)

            # 计算粘贴位置（居中）
            start_h = (target_h - new_h) // 2
            start_w = (target_w - new_w) // 2

            canvas[start_h:start_h + new_h, start_w:start_w + new_w] = resized

            return canvas, scale, (start_w, start_h)
        else:
            # 直接缩放到目标大小
            resized = cv2.resize(image, (target_w, target_h), interpolation=cv2.INTER_LINEAR)
            return resized, None, None

    def normalize(self, image: np.ndarray) -> np.ndarray:
        """图像归一化"""
        # 转换为float32并归一化到[0,1]
        image = image.astype(np.float32) / 255.0

        # 标准化
        image = (image - self.mean) / self.std

        return image

    def preprocess(self, image: np.ndarray, keep_ratio: bool = True) -> Tuple[np.ndarray, dict]:
        """完整的预处理流程"""
        original_shape = image.shape[:2]

        # BGR转RGB
        if self.bgr_to_rgb and len(image.shape) == 3:
            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)

        # 调整大小
        if keep_ratio:
            image, scale, offset = self.resize_image(image, keep_ratio=True)
            preprocess_info = {
                'original_shape': original_shape,
                'scale': scale,
                'offset': offset,
                'target_size': self.target_size
            }
        else:
            image, _, _ = self.resize_image(image, keep_ratio=False)
            preprocess_info = {
                'original_shape': original_shape,
                'target_size': self.target_size
            }

        # 归一化
        image = self.normalize(image)

        # 转换为NCHW格式
        image = np.transpose(image, (2, 0, 1))
        image = np.expand_dims(image, axis=0)

        return image, preprocess_info

# 使用示例
def preprocess_example():
    """预处理使用示例"""
    # 创建预处理器
    preprocessor = ImagePreprocessor(
        target_size=(224, 224),
        mean=(0.485, 0.456, 0.406),
        std=(0.229, 0.224, 0.225)
    )

    # 读取图像
    image_path = "test_images/cat.jpg"
    image = cv2.imread(image_path)

    if image is None:
        print(f"无法读取图像: {image_path}")
        return

    print(f"原始图像形状: {image.shape}")

    # 预处理
    processed_image, info = preprocessor.preprocess(image, keep_ratio=True)

    print(f"处理后图像形状: {processed_image.shape}")
    print(f"预处理信息: {info}")

    return processed_image, info

if __name__ == "__main__":
    preprocess_example()

Postprocessing Module

import numpy as np
from typing import List, Tuple, Dict, Optional

class PostProcessor:
    """后处理基类"""

    def __init__(self):
        pass

    def process(self, outputs: List[np.ndarray], **kwargs) -> Dict:
        """处理模型输出"""
        raise NotImplementedError

class ClassificationPostProcessor(PostProcessor):
    """分类任务后处理"""

    def __init__(self, class_names: Optional[List[str]] = None, top_k: int = 5):
        super().__init__()
        self.class_names = class_names
        self.top_k = top_k

    def softmax(self, x: np.ndarray) -> np.ndarray:
        """Softmax激活函数"""
        exp_x = np.exp(x - np.max(x, axis=-1, keepdims=True))
        return exp_x / np.sum(exp_x, axis=-1, keepdims=True)

    def process(self, outputs: List[np.ndarray], **kwargs) -> Dict:
        """处理分类模型输出"""
        if len(outputs) == 0:
            raise ValueError("输出为空")

        # 获取第一个输出（通常是分类logits）
        logits = outputs[0]

        # 如果是4D张量，压缩为2D
        if len(logits.shape) == 4:
            logits = logits.reshape(logits.shape[0], -1)
        elif len(logits.shape) == 3:
            logits = logits.reshape(logits.shape[0], -1)

        # 应用softmax
        probabilities = self.softmax(logits)

        results = []

        for i, probs in enumerate(probabilities):
            # 获取top-k结果
            top_indices = np.argsort(probs)[::-1][:self.top_k]
            top_probs = probs[top_indices]

            # 构建结果
            predictions = []
            for idx, prob in zip(top_indices, top_probs):
                prediction = {
                    'class_id': int(idx),
                    'probability': float(prob),
                    'confidence': float(prob)
                }

                if self.class_names and idx < len(self.class_names):
                    prediction['class_name'] = self.class_names[idx]

                predictions.append(prediction)

            results.append({
                'predictions': predictions,
                'top1_class_id': int(top_indices[0]),
                'top1_probability': float(top_probs[0])
            })

        return {
            'results': results,
            'batch_size': len(results)
        }

class DetectionPostProcessor(PostProcessor):
    """目标检测后处理"""

    def __init__(self, class_names: Optional[List[str]] = None,
                 conf_threshold: float = 0.5, nms_threshold: float = 0.4):
        super().__init__()
        self.class_names = class_names
        self.conf_threshold = conf_threshold
        self.nms_threshold = nms_threshold

    def xywh2xyxy(self, boxes: np.ndarray) -> np.ndarray:
        """转换边界框格式从xywh到xyxy"""
        xyxy = boxes.copy()
        xyxy[:, 0] = boxes[:, 0] - boxes[:, 2] / 2  # x1
        xyxy[:, 1] = boxes[:, 1] - boxes[:, 3] / 2  # y1
        xyxy[:, 2] = boxes[:, 0] + boxes[:, 2] / 2  # x2
        xyxy[:, 3] = boxes[:, 1] + boxes[:, 3] / 2  # y2
        return xyxy

    def nms(self, boxes: np.ndarray, scores: np.ndarray, threshold: float) -> List[int]:
        """非极大值抑制"""
        if len(boxes) == 0:
            return []

        # 计算面积
        areas = (boxes[:, 2] - boxes[:, 0]) * (boxes[:, 3] - boxes[:, 1])

        # 按分数排序
        order = scores.argsort()[::-1]

        keep = []
        while len(order) > 0:
            i = order[0]
            keep.append(i)

            if len(order) == 1:
                break

            # 计算IoU
            xx1 = np.maximum(boxes[i, 0], boxes[order[1:], 0])
            yy1 = np.maximum(boxes[i, 1], boxes[order[1:], 1])
            xx2 = np.minimum(boxes[i, 2], boxes[order[1:], 2])
            yy2 = np.minimum(boxes[i, 3], boxes[order[1:], 3])

            w = np.maximum(0, xx2 - xx1)
            h = np.maximum(0, yy2 - yy1)
            intersection = w * h

            union = areas[i] + areas[order[1:]] - intersection
            iou = intersection / union

            # 保留IoU小于阈值的框
            indices = np.where(iou <= threshold)[0]
            order = order[indices + 1]

        return keep

    def process(self, outputs: List[np.ndarray], input_shape: Tuple[int, int] = (640, 640),
                original_shape: Optional[Tuple[int, int]] = None) -> Dict:
        """处理检测模型输出"""
        if len(outputs) == 0:
            raise ValueError("输出为空")

        # YOLOv5输出格式: [batch, num_anchors, 85] (4 + 1 + 80)
        predictions = outputs[0]

        if len(predictions.shape) == 3:
            predictions = predictions[0]  # 取第一个batch

        # 过滤低置信度检测
        conf_mask = predictions[:, 4] >= self.conf_threshold
        predictions = predictions[conf_mask]

        if len(predictions) == 0:
            return {'detections': [], 'count': 0}

        # 提取边界框、置信度和类别概率
        boxes = predictions[:, :4]
        confidences = predictions[:, 4]
        class_probs = predictions[:, 5:]

        # 计算类别分数
        class_scores = confidences[:, np.newaxis] * class_probs
        class_ids = np.argmax(class_scores, axis=1)
        scores = np.max(class_scores, axis=1)

        # 转换边界框格式
        boxes = self.xywh2xyxy(boxes)

        # 缩放到原始图像尺寸
        if original_shape:
            scale_x = original_shape[1] / input_shape[1]
            scale_y = original_shape[0] / input_shape[0]
            boxes[:, [0, 2]] *= scale_x
            boxes[:, [1, 3]] *= scale_y

        # 应用NMS
        keep_indices = self.nms(boxes, scores, self.nms_threshold)

        # 构建最终结果
        detections = []
        for i in keep_indices:
            detection = {
                'bbox': boxes[i].tolist(),
                'confidence': float(scores[i]),
                'class_id': int(class_ids[i])
            }

            if self.class_names and class_ids[i] < len(self.class_names):
                detection['class_name'] = self.class_names[class_ids[i]]

            detections.append(detection)

        return {
            'detections': detections,
            'count': len(detections)
        }

# 使用示例
def postprocess_example():
    """后处理使用示例"""
    # 分类后处理示例
    class_names = ['cat', 'dog', 'bird', 'fish', 'horse']
    classifier = ClassificationPostProcessor(class_names=class_names, top_k=3)

    # 模拟分类输出
    classification_output = [np.random.rand(1, 5)]
    classification_results = classifier.process(classification_output)

    print("分类结果:")
    for result in classification_results['results']:
        print(f"Top-1: {result['predictions'][0]}")

    # 检测后处理示例
    coco_classes = ['person', 'bicycle', 'car', 'motorcycle', 'airplane']
    detector = DetectionPostProcessor(class_names=coco_classes)

    # 模拟检测输出
    detection_output = [np.random.rand(1, 25200, 85)]  # YOLOv5输出格式
    detection_results = detector.process(detection_output)

    print(f"\n检测结果: 发现 {detection_results['count']} 个目标")

if __name__ == "__main__":
    postprocess_example()

5.3 C/C++ API Deployment Explanation

Basic C++ API Usage

// rknn_inference.h
#ifndef RKNN_INFERENCE_H
#define RKNN_INFERENCE_H

#include <vector>
#include <string>
#include <memory>
#include "rknn_api.h"

class RKNNInference {
public:
    RKNNInference();
    ~RKNNInference();

    // 基础功能
    int loadModel(const std::string& model_path);
    int initRuntime();
    int inference(const std::vector<void*>& inputs, std::vector<void*>& outputs);
    void release();

    // 信息获取
    rknn_input_output_num getIONum() const { return io_num_; }
    rknn_tensor_attr* getInputAttrs() const { return input_attrs_; }
    rknn_tensor_attr* getOutputAttrs() const { return output_attrs_; }

    // 工具函数
    static void printTensorAttr(const rknn_tensor_attr& attr);
    static size_t getTensorSize(const rknn_tensor_attr& attr);

private:
    rknn_context ctx_;
    rknn_input_output_num io_num_;
    rknn_tensor_attr* input_attrs_;
    rknn_tensor_attr* output_attrs_;
    bool is_loaded_;
    bool is_initialized_;

    void cleanup();
};

#endif // RKNN_INFERENCE_H

// rknn_inference.cpp
#include "rknn_inference.h"
#include <iostream>
#include <fstream>
#include <cstring>

RKNNInference::RKNNInference()
    : ctx_(0), input_attrs_(nullptr), output_attrs_(nullptr),
      is_loaded_(false), is_initialized_(false) {
    memset(&io_num_, 0, sizeof(io_num_));
}

RKNNInference::~RKNNInference() {
    cleanup();
}

int RKNNInference::loadModel(const std::string& model_path) {
    std::cout << "Loading model: " << model_path << std::endl;

    // 读取模型文件
    std::ifstream file(model_path, std::ios::binary | std::ios::ate);
    if (!file.is_open()) {
        std::cerr << "Failed to open model file: " << model_path << std::endl;
        return -1;
    }

    size_t model_size = file.tellg();
    file.seekg(0, std::ios::beg);

    std::vector<char> model_data(model_size);
    if (!file.read(model_data.data(), model_size)) {
        std::cerr << "Failed to read model file" << std::endl;
        return -1;
    }
    file.close();

    // 初始化RKNN上下文
    int ret = rknn_init(&ctx_, model_data.data(), model_size, 0, nullptr);
    if (ret < 0) {
        std::cerr << "rknn_init failed: " << ret << std::endl;
        return ret;
    }

    // 获取输入输出数量
    ret = rknn_query(ctx_, RKNN_QUERY_IN_OUT_NUM, &io_num_, sizeof(io_num_));
    if (ret < 0) {
        std::cerr << "rknn_query RKNN_QUERY_IN_OUT_NUM failed: " << ret << std::endl;
        return ret;
    }

    std::cout << "Model input num: " << io_num_.n_input
              << ", output num: " << io_num_.n_output << std::endl;

    // 获取输入属性
    input_attrs_ = new rknn_tensor_attr[io_num_.n_input];
    memset(input_attrs_, 0, sizeof(rknn_tensor_attr) * io_num_.n_input);

    for (uint32_t i = 0; i < io_num_.n_input; i++) {
        input_attrs_[i].index = i;
        ret = rknn_query(ctx_, RKNN_QUERY_INPUT_ATTR, &input_attrs_[i], sizeof(rknn_tensor_attr));
        if (ret < 0) {
            std::cerr << "rknn_query input attr " << i << " failed: " << ret << std::endl;
            return ret;
        }

        std::cout << "Input " << i << " attr:" << std::endl;
        printTensorAttr(input_attrs_[i]);
    }

    // 获取输出属性
    output_attrs_ = new rknn_tensor_attr[io_num_.n_output];
    memset(output_attrs_, 0, sizeof(rknn_tensor_attr) * io_num_.n_output);

    for (uint32_t i = 0; i < io_num_.n_output; i++) {
        output_attrs_[i].index = i;
        ret = rknn_query(ctx_, RKNN_QUERY_OUTPUT_ATTR, &output_attrs_[i], sizeof(rknn_tensor_attr));
        if (ret < 0) {
            std::cerr << "rknn_query output attr " << i << " failed: " << ret << std::endl;
            return ret;
        }

        std::cout << "Output " << i << " attr:" << std::endl;
        printTensorAttr(output_attrs_[i]);
    }

    is_loaded_ = true;
    std::cout << "Model loaded successfully" << std::endl;
    return 0;
}

int RKNNInference::initRuntime() {
    if (!is_loaded_) {
        std::cerr << "Model not loaded" << std::endl;
        return -1;
    }

    std::cout << "Initializing runtime..." << std::endl;

    int ret = rknn_init_runtime(ctx_, nullptr);
    if (ret < 0) {
        std::cerr << "rknn_init_runtime failed: " << ret << std::endl;
        return ret;
    }

    is_initialized_ = true;
    std::cout << "Runtime initialized successfully" << std::endl;
    return 0;
}

int RKNNInference::inference(const std::vector<void*>& inputs, std::vector<void*>& outputs) {
    if (!is_initialized_) {
        std::cerr << "Runtime not initialized" << std::endl;
        return -1;
    }

    if (inputs.size() != io_num_.n_input) {
        std::cerr << "Input size mismatch: expected " << io_num_.n_input
                  << ", got " << inputs.size() << std::endl;
        return -1;
    }

    // 设置输入
    std::vector<rknn_input> rknn_inputs(io_num_.n_input);
    for (uint32_t i = 0; i < io_num_.n_input; i++) {
        rknn_inputs[i].index = i;
        rknn_inputs[i].buf = inputs[i];
        rknn_inputs[i].size = getTensorSize(input_attrs_[i]);
        rknn_inputs[i].pass_through = 0;
        rknn_inputs[i].type = RKNN_TENSOR_UINT8;
        rknn_inputs[i].fmt = RKNN_TENSOR_NHWC;
    }

    int ret = rknn_inputs_set(ctx_, io_num_.n_input, rknn_inputs.data());
    if (ret < 0) {
        std::cerr << "rknn_inputs_set failed: " << ret << std::endl;
        return ret;
    }

    // 执行推理
    ret = rknn_run(ctx_, nullptr);
    if (ret < 0) {
        std::cerr << "rknn_run failed: " << ret << std::endl;
        return ret;
    }

    // 获取输出
    std::vector<rknn_output> rknn_outputs(io_num_.n_output);
    for (uint32_t i = 0; i < io_num_.n_output; i++) {
        rknn_outputs[i].want_float = 1;
        rknn_outputs[i].is_prealloc = 0;
    }

    ret = rknn_outputs_get(ctx_, io_num_.n_output, rknn_outputs.data(), nullptr);
    if (ret < 0) {
        std::cerr << "rknn_outputs_get failed: " << ret << std::endl;
        return ret;
    }

    // 复制输出数据
    outputs.resize(io_num_.n_output);
    for (uint32_t i = 0; i < io_num_.n_output; i++) {
        size_t output_size = rknn_outputs[i].size;
        outputs[i] = malloc(output_size);
        memcpy(outputs[i], rknn_outputs[i].buf, output_size);
    }

    // 释放RKNN输出缓冲区
    rknn_outputs_release(ctx_, io_num_.n_output, rknn_outputs.data());

    return 0;
}

void RKNNInference::release() {
    cleanup();
}

void RKNNInference::cleanup() {
    if (input_attrs_) {
        delete[] input_attrs_;
        input_attrs_ = nullptr;
    }

    if (output_attrs_) {
        delete[] output_attrs_;
        output_attrs_ = nullptr;
    }

    if (ctx_) {
        rknn_destroy(ctx_);
        ctx_ = 0;
    }

    is_loaded_ = false;
    is_initialized_ = false;
}

void RKNNInference::printTensorAttr(const rknn_tensor_attr& attr) {
    std::cout << "  index=" << attr.index << ", name=" << attr.name
              << ", n_dims=" << attr.n_dims << ", dims=[";
    for (uint32_t i = 0; i < attr.n_dims; i++) {
        std::cout << attr.dims[i];
        if (i < attr.n_dims - 1) std::cout << ", ";
    }
    std::cout << "], n_elems=" << attr.n_elems;
    std::cout << ", size=" << attr.size << ", fmt=" << attr.fmt
              << ", type=" << attr.type << ", qnt_type=" << attr.qnt_type << std::endl;
}

size_t RKNNInference::getTensorSize(const rknn_tensor_attr& attr) {
    size_t size = 1;
    for (uint32_t i = 0; i < attr.n_dims; i++) {
        size *= attr.dims[i];
    }

    switch (attr.type) {
        case RKNN_TENSOR_FLOAT32:
            return size * sizeof(float);
        case RKNN_TENSOR_FLOAT16:
            return size * sizeof(uint16_t);
        case RKNN_TENSOR_INT8:
        case RKNN_TENSOR_UINT8:
            return size * sizeof(uint8_t);
        case RKNN_TENSOR_INT16:
        case RKNN_TENSOR_UINT16:
            return size * sizeof(uint16_t);
        case RKNN_TENSOR_INT32:
        case RKNN_TENSOR_UINT32:
            return size * sizeof(uint32_t);
        case RKNN_TENSOR_INT64:
        case RKNN_TENSOR_UINT64:
            return size * sizeof(uint64_t);
        default:
            return size;
    }
}

Image Processing Utility Class

// image_utils.h
#ifndef IMAGE_UTILS_H
#define IMAGE_UTILS_H

#include <opencv2/opencv.hpp>
#include <vector>

class ImageUtils {
public:
    struct PreprocessInfo {
        cv::Size original_size;
        cv::Size target_size;
        float scale;
        cv::Point2f offset;
    };

    static cv::Mat resizeKeepRatio(const cv::Mat& image, cv::Size target_size,
                                   PreprocessInfo& info);
    static cv::Mat normalize(const cv::Mat& image,
                           const std::vector<float>& mean = {0.485, 0.456, 0.406},
                           const std::vector<float>& std = {0.229, 0.224, 0.225});
    static std::vector<uint8_t> matToUint8(const cv::Mat& image);
    static cv::Mat uint8ToMat(const std::vector<uint8_t>& data, cv::Size size, int type);

    // 绘制结果
    static void drawClassification(cv::Mat& image, const std::string& class_name,
                                 float confidence, cv::Point position = cv::Point(10, 30));
    static void drawDetection(cv::Mat& image, const cv::Rect& bbox,
                            const std::string& label, float confidence);
};

#endif // IMAGE_UTILS_H

// image_utils.cpp
#include "image_utils.h"
#include <iostream>

cv::Mat ImageUtils::resizeKeepRatio(const cv::Mat& image, cv::Size target_size,
                                   PreprocessInfo& info) {
    info.original_size = image.size();
    info.target_size = target_size;

    float scale_x = static_cast<float>(target_size.width) / image.cols;
    float scale_y = static_cast<float>(target_size.height) / image.rows;
    info.scale = std::min(scale_x, scale_y);

    int new_width = static_cast<int>(image.cols * info.scale);
    int new_height = static_cast<int>(image.rows * info.scale);

    cv::Mat resized;
    cv::resize(image, resized, cv::Size(new_width, new_height), 0, 0, cv::INTER_LINEAR);

    // 创建目标大小的画布
    cv::Mat canvas = cv::Mat::zeros(target_size, image.type());

    // 计算居中位置
    int offset_x = (target_size.width - new_width) / 2;
    int offset_y = (target_size.height - new_height) / 2;
    info.offset = cv::Point2f(offset_x, offset_y);

    // 将缩放后的图像放置到画布中心
    cv::Rect roi(offset_x, offset_y, new_width, new_height);
    resized.copyTo(canvas(roi));

    return canvas;
}

cv::Mat ImageUtils::normalize(const cv::Mat& image,
                            const std::vector<float>& mean,
                            const std::vector<float>& std) {
    cv::Mat normalized;
    image.convertTo(normalized, CV_32F, 1.0 / 255.0);

    std::vector<cv::Mat> channels;
    cv::split(normalized, channels);

    for (size_t i = 0; i < channels.size() && i < mean.size(); i++) {
        channels[i] = (channels[i] - mean[i]) / std[i];
    }

    cv::merge(channels, normalized);
    return normalized;
}

std::vector<uint8_t> ImageUtils::matToUint8(const cv::Mat& image) {
    cv::Mat uint8_image;

    if (image.type() == CV_32F) {
        // 反归一化
        image.convertTo(uint8_image, CV_8U, 255.0);
    } else {
        image.convertTo(uint8_image, CV_8U);
    }

    // 转换为RGB格式（如果是BGR）
    if (uint8_image.channels() == 3) {
        cv::cvtColor(uint8_image, uint8_image, cv::COLOR_BGR2RGB);
    }

    std::vector<uint8_t> data;
    data.assign(uint8_image.datastart, uint8_image.dataend);

    return data;
}

cv::Mat ImageUtils::uint8ToMat(const std::vector<uint8_t>& data, cv::Size size, int type) {
    cv::Mat image(size, type);
    std::memcpy(image.data, data.data(), data.size());
    return image;
}

void ImageUtils::drawClassification(cv::Mat& image, const std::string& class_name,
                                  float confidence, cv::Point position) {
    std::string text = class_name + ": " + std::to_string(confidence);

    int font_face = cv::FONT_HERSHEY_SIMPLEX;
    double font_scale = 0.8;
    int thickness = 2;
    cv::Scalar color(0, 255, 0);  // 绿色

    // 获取文本大小
    int baseline;
    cv::Size text_size = cv::getTextSize(text, font_face, font_scale, thickness, &baseline);

    // 绘制背景矩形
    cv::Rect bg_rect(position.x, position.y - text_size.height - 10,
                     text_size.width + 10, text_size.height + 15);
    cv::rectangle(image, bg_rect, cv::Scalar(0, 0, 0), -1);

    // 绘制文本
    cv::putText(image, text, cv::Point(position.x + 5, position.y - 5),
                font_face, font_scale, color, thickness);
}

void ImageUtils::drawDetection(cv::Mat& image, const cv::Rect& bbox,
                             const std::string& label, float confidence) {
    // 绘制边界框
    cv::Scalar box_color(0, 255, 0);  // 绿色
    cv::rectangle(image, bbox, box_color, 2);

    // 准备标签文本
    std::string text = label + ": " + std::to_string(confidence);

    int font_face = cv::FONT_HERSHEY_SIMPLEX;
    double font_scale = 0.6;
    int thickness = 2;

    // 获取文本大小
    int baseline;
    cv::Size text_size = cv::getTextSize(text, font_face, font_scale, thickness, &baseline);

    // 绘制标签背景
    cv::Point label_pos(bbox.x, bbox.y - 10);
    cv::Rect label_bg(label_pos.x, label_pos.y - text_size.height - 5,
                      text_size.width + 10, text_size.height + 10);
    cv::rectangle(image, label_bg, cv::Scalar(0, 255, 0), -1);

    // 绘制标签文本
    cv::putText(image, text, cv::Point(label_pos.x + 5, label_pos.y - 5),
                font_face, font_scale, cv::Scalar(0, 0, 0), thickness);
}

5.4 Deployment In Practice: Writing an Image Classification Application

Python Version Image Classification Application

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import sys
import time
import argparse
import cv2
import numpy as np
from pathlib import Path

# 添加项目路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from rknnlite.api import RKNNLite
from image_preprocessor import ImagePreprocessor
from postprocessor import ClassificationPostProcessor

class ImageClassificationApp:
    """图像分类应用"""

    def __init__(self, model_path, class_names_file=None, target_size=(224, 224)):
        self.model_path = model_path
        self.target_size = target_size

        # 加载类别名称
        self.class_names = self.load_class_names(class_names_file)

        # 初始化组件
        self.rknn = RKNNLite()
        self.preprocessor = ImagePreprocessor(target_size=target_size)
        self.postprocessor = ClassificationPostProcessor(
            class_names=self.class_names,
            top_k=5
        )

        # 加载模型
        self.load_model()

    def load_class_names(self, class_names_file):
        """加载类别名称"""
        if class_names_file and os.path.exists(class_names_file):
            with open(class_names_file, 'r', encoding='utf-8') as f:
                return [line.strip() for line in f.readlines()]
        else:
            # 使用ImageNet类别（简化版）
            return [f"class_{i}" for i in range(1000)]

    def load_model(self):
        """加载RKNN模型"""
        print(f"加载模型: {self.model_path}")

        ret = self.rknn.load_rknn(self.model_path)
        if ret != 0:
            raise RuntimeError(f"加载模型失败，错误码: {ret}")

        ret = self.rknn.init_runtime()
        if ret != 0:
            raise RuntimeError(f"初始化运行时失败，错误码: {ret}")

        print("模型加载成功")

    def predict_image(self, image_path):
        """预测单张图像"""
        # 读取图像
        image = cv2.imread(image_path)
        if image is None:
            raise ValueError(f"无法读取图像: {image_path}")

        return self.predict(image)

    def predict(self, image):
        """预测图像"""
        start_time = time.time()

        # 预处理
        processed_image, preprocess_info = self.preprocessor.preprocess(image)
        preprocess_time = time.time() - start_time

        # 推理
        inference_start = time.time()
        outputs = self.rknn.inference(inputs=[processed_image])
        inference_time = time.time() - inference_start

        # 后处理
        postprocess_start = time.time()
        results = self.postprocessor.process(outputs)
        postprocess_time = time.time() - postprocess_start

        total_time = time.time() - start_time

        # 添加时间信息
        results['timing'] = {
            'preprocess_time': preprocess_time,
            'inference_time': inference_time,
            'postprocess_time': postprocess_time,
            'total_time': total_time,
            'fps': 1.0 / total_time
        }

        return results

    def predict_batch(self, image_paths, batch_size=8):
        """批量预测"""
        results = []

        for i in range(0, len(image_paths), batch_size):
            batch_paths = image_paths[i:i + batch_size]
            batch_results = []

            for image_path in batch_paths:
                try:
                    result = self.predict_image(image_path)
                    result['image_path'] = image_path
                    batch_results.append(result)
                except Exception as e:
                    print(f"处理图像 {image_path} 失败: {e}")
                    continue

            results.extend(batch_results)
            print(f"已处理 {len(results)}/{len(image_paths)} 张图像")

        return results

    def predict_video(self, video_path, output_path=None, display=True):
        """预测视频"""
        cap = cv2.VideoCapture(video_path)

        if not cap.isOpened():
            raise ValueError(f"无法打开视频: {video_path}")

        # 获取视频信息
        fps = int(cap.get(cv2.CAP_PROP_FPS))
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

        print(f"视频信息: {width}x{height}, {fps}fps, {total_frames}帧")

        # 初始化视频写入器
        writer = None
        if output_path:
            fourcc = cv2.VideoWriter_fourcc(*'mp4v')
            writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height))

        frame_count = 0
        total_inference_time = 0

        try:
            while True:
                ret, frame = cap.read()
                if not ret:
                    break

                frame_count += 1

                # 预测
                start_time = time.time()
                results = self.predict(frame)
                inference_time = time.time() - start_time
                total_inference_time += inference_time

                # 绘制结果
                if results['results']:
                    top_prediction = results['results'][0]['predictions'][0]
                    class_name = top_prediction.get('class_name', f"Class {top_prediction['class_id']}")
                    confidence = top_prediction['confidence']

                    # 绘制分类结果
                    text = f"{class_name}: {confidence:.3f}"
                    cv2.putText(frame, text, (10, 30), cv2.FONT_HERSHEY_SIMPLEX,
                               1, (0, 255, 0), 2)

                # 绘制FPS信息
                fps_text = f"FPS: {1.0/inference_time:.1f}"
                cv2.putText(frame, fps_text, (10, 70), cv2.FONT_HERSHEY_SIMPLEX,
                           0.7, (255, 255, 255), 2)

                # 保存帧
                if writer:
                    writer.write(frame)

                # 显示
                if display:
                    cv2.imshow('Classification', frame)
                    if cv2.waitKey(1) & 0xFF == ord('q'):
                        break

                # 进度显示
                if frame_count % 30 == 0:
                    progress = frame_count / total_frames * 100
                    avg_fps = frame_count / total_inference_time
                    print(f"进度: {progress:.1f}%, 平均FPS: {avg_fps:.1f}")

        finally:
            cap.release()
            if writer:
                writer.release()
            if display:
                cv2.destroyAllWindows()

        avg_fps = frame_count / total_inference_time if total_inference_time > 0 else 0
        print(f"视频处理完成: {frame_count}帧, 平均FPS: {avg_fps:.1f}")

    def predict_camera(self, camera_id=0, display=True):
        """实时摄像头预测"""
        cap = cv2.VideoCapture(camera_id)

        if not cap.isOpened():
            raise ValueError(f"无法打开摄像头: {camera_id}")

        # 设置摄像头参数
        cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
        cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
        cap.set(cv2.CAP_PROP_FPS, 30)

        print("开始实时分类，按 'q' 退出")

        frame_count = 0
        total_time = 0

        try:
            while True:
                ret, frame = cap.read()
                if not ret:
                    print("无法读取摄像头帧")
                    break

                frame_count += 1

                # 预测
                start_time = time.time()
                results = self.predict(frame)
                inference_time = time.time() - start_time
                total_time += inference_time

                # 绘制结果
                if results['results']:
                    predictions = results['results'][0]['predictions']

                    # 显示Top-3结果
                    for i, pred in enumerate(predictions[:3]):
                        class_name = pred.get('class_name', f"Class {pred['class_id']}")
                        confidence = pred['confidence']

                        text = f"{i+1}. {class_name}: {confidence:.3f}"
                        y_pos = 30 + i * 30
                        cv2.putText(frame, text, (10, y_pos), cv2.FONT_HERSHEY_SIMPLEX,
                                   0.7, (0, 255, 0), 2)

                # 显示性能信息
                timing = results['timing']
                fps_text = f"FPS: {timing['fps']:.1f}"
                cv2.putText(frame, fps_text, (10, frame.shape[0] - 60),
                           cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)

                inference_text = f"Inference: {timing['inference_time']*1000:.1f}ms"
                cv2.putText(frame, inference_text, (10, frame.shape[0] - 30),
                           cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)

                # 显示
                if display:
                    cv2.imshow('Real-time Classification', frame)
                    if cv2.waitKey(1) & 0xFF == ord('q'):
                        break

        finally:
            cap.release()
            if display:
                cv2.destroyAllWindows()

        avg_fps = frame_count / total_time if total_time > 0 else 0
        print(f"实时分类结束: {frame_count}帧, 平均FPS: {avg_fps:.1f}")

    def release(self):
        """释放资源"""
        if hasattr(self, 'rknn') and self.rknn:
            self.rknn.release()

def main():
    parser = argparse.ArgumentParser(description='RKNN图像分类应用')
    parser.add_argument('--model', required=True, help='RKNN模型路径')
    parser.add_argument('--classes', help='类别名称文件路径')
    parser.add_argument('--image', help='输入图像路径')
    parser.add_argument('--video', help='输入视频路径')
    parser.add_argument('--camera', type=int, help='摄像头ID')
    parser.add_argument('--output', help='输出视频路径')
    parser.add_argument('--batch', nargs='+', help='批量处理图像路径')
    parser.add_argument('--size', type=int, nargs=2, default=[224, 224],
                       help='输入图像尺寸 (width height)')

    args = parser.parse_args()

    # 创建分类应用
    app = ImageClassificationApp(
        model_path=args.model,
        class_names_file=args.classes,
        target_size=tuple(args.size)
    )

    try:
        if args.image:
            # 单张图像预测
            results = app.predict_image(args.image)

            print(f"\n图像: {args.image}")
            print("分类结果:")
            for i, pred in enumerate(results['results'][0]['predictions']):
                class_name = pred.get('class_name', f"Class {pred['class_id']}")
                print(f"  {i+1}. {class_name}: {pred['confidence']:.4f}")

            timing = results['timing']
            print(f"\n性能统计:")
            print(f"  预处理: {timing['preprocess_time']*1000:.1f}ms")
            print(f"  推理: {timing['inference_time']*1000:.1f}ms")
            print(f"  后处理: {timing['postprocess_time']*1000:.1f}ms")
            print(f"  总时间: {timing['total_time']*1000:.1f}ms")
            print(f"  FPS: {timing['fps']:.1f}")

        elif args.batch:
            # 批量预测
            results = app.predict_batch(args.batch)

            print(f"\n批量处理结果 ({len(results)}张图像):")
            for result in results:
                image_path = result['image_path']
                top_pred = result['results'][0]['predictions'][0]
                class_name = top_pred.get('class_name', f"Class {top_pred['class_id']}")

                print(f"{Path(image_path).name}: {class_name} ({top_pred['confidence']:.4f})")

        elif args.video:
            # 视频预测
            app.predict_video(args.video, args.output)

        elif args.camera is not None:
            # 实时摄像头预测
            app.predict_camera(args.camera)

        else:
            print("请指定输入源: --image, --video, --camera 或 --batch")

    finally:
        app.release()

if __name__ == "__main__":
    main()

C++ Version Image Classification Application

// classification_app.h
#ifndef CLASSIFICATION_APP_H
#define CLASSIFICATION_APP_H

#include <string>
#include <vector>
#include <memory>
#include <opencv2/opencv.hpp>
#include "rknn_inference.h"
#include "image_utils.h"

struct ClassificationResult {
    int class_id;
    std::string class_name;
    float confidence;
};

struct TimingInfo {
    double preprocess_time;
    double inference_time;
    double postprocess_time;
    double total_time;
    double fps;
};

class ClassificationApp {
public:
    ClassificationApp(const std::string& model_path,
                     const std::string& class_names_file = "",
                     cv::Size target_size = cv::Size(224, 224));
    ~ClassificationApp();

    // 预测接口
    std::vector<ClassificationResult> predictImage(const std::string& image_path,
                                                   TimingInfo* timing = nullptr);
    std::vector<ClassificationResult> predict(const cv::Mat& image,
                                             TimingInfo* timing = nullptr);

    // 批量和视频处理
    void predictBatch(const std::vector<std::string>& image_paths);
    void predictVideo(const std::string& video_path,
                     const std::string& output_path = "");
    void predictCamera(int camera_id = 0);

private:
    std::unique_ptr<RKNNInference> rknn_;
    std::vector<std::string> class_names_;
    cv::Size target_size_;
    int top_k_;

    // 辅助函数
    bool loadClassNames(const std::string& file_path);
    std::vector<ClassificationResult> postprocess(const std::vector<void*>& outputs);
    void drawResults(cv::Mat& image, const std::vector<ClassificationResult>& results,
                    const TimingInfo& timing);

    // 数据转换
    std::vector<uint8_t> preprocessImage(const cv::Mat& image);
    void softmax(float* data, int size);
};

#endif // CLASSIFICATION_APP_H

// classification_app.cpp
#include "classification_app.h"
#include <iostream>
#include <fstream>
#include <algorithm>
#include <chrono>

ClassificationApp::ClassificationApp(const std::string& model_path,
                                   const std::string& class_names_file,
                                   cv::Size target_size)
    : target_size_(target_size), top_k_(5) {

    // 初始化RKNN推理
    rknn_ = std::make_unique<RKNNInference>();

    if (rknn_->loadModel(model_path) != 0) {
        throw std::runtime_error("Failed to load model: " + model_path);
    }

    if (rknn_->initRuntime() != 0) {
        throw std::runtime_error("Failed to initialize runtime");
    }

    // 加载类别名称
    if (!class_names_file.empty()) {
        loadClassNames(class_names_file);
    } else {
        // 默认类别名称
        for (int i = 0; i < 1000; i++) {
            class_names_.push_back("class_" + std::to_string(i));
        }
    }

    std::cout << "Classification app initialized successfully" << std::endl;
}

ClassificationApp::~ClassificationApp() {
    if (rknn_) {
        rknn_->release();
    }
}

bool ClassificationApp::loadClassNames(const std::string& file_path) {
    std::ifstream file(file_path);
    if (!file.is_open()) {
        std::cerr << "Failed to open class names file: " << file_path << std::endl;
        return false;
    }

    std::string line;
    while (std::getline(file, line)) {
        if (!line.empty()) {
            class_names_.push_back(line);
        }
    }

    std::cout << "Loaded " << class_names_.size() << " class names" << std::endl;
    return true;
}

std::vector<ClassificationResult> ClassificationApp::predictImage(
    const std::string& image_path, TimingInfo* timing) {

    cv::Mat image = cv::imread(image_path);
    if (image.empty()) {
        throw std::runtime_error("Failed to load image: " + image_path);
    }

    return predict(image, timing);
}

std::vector<ClassificationResult> ClassificationApp::predict(
    const cv::Mat& image, TimingInfo* timing) {

    auto start_time = std::chrono::high_resolution_clock::now();

    // 预处理
    auto preprocess_start = std::chrono::high_resolution_clock::now();
    std::vector<uint8_t> input_data = preprocessImage(image);
    auto preprocess_end = std::chrono::high_resolution_clock::now();

    // 推理
    auto inference_start = std::chrono::high_resolution_clock::now();
    std::vector<void*> inputs = {input_data.data()};
    std::vector<void*> outputs;

    if (rknn_->inference(inputs, outputs) != 0) {
        throw std::runtime_error("Inference failed");
    }
    auto inference_end = std::chrono::high_resolution_clock::now();

    // 后处理
    auto postprocess_start = std::chrono::high_resolution_clock::now();
    std::vector<ClassificationResult> results = postprocess(outputs);
    auto postprocess_end = std::chrono::high_resolution_clock::now();

    auto total_end = std::chrono::high_resolution_clock::now();

    // 计算时间
    if (timing) {
        timing->preprocess_time = std::chrono::duration<double>(preprocess_end - preprocess_start).count();
        timing->inference_time = std::chrono::duration<double>(inference_end - inference_start).count();
        timing->postprocess_time = std::chrono::duration<double>(postprocess_end - postprocess_start).count();
        timing->total_time = std::chrono::duration<double>(total_end - start_time).count();
        timing->fps = 1.0 / timing->total_time;
    }

    // 释放输出内存
    for (void* output : outputs) {
        free(output);
    }

    return results;
}

std::vector<uint8_t> ClassificationApp::preprocessImage(const cv::Mat& image) {
    ImageUtils::PreprocessInfo info;

    // 调整大小并保持宽高比
    cv::Mat resized = ImageUtils::resizeKeepRatio(image, target_size_, info);

    // 归一化
    cv::Mat normalized = ImageUtils::normalize(resized);

    // 转换为uint8格式
    return ImageUtils::matToUint8(normalized);
}

std::vector<ClassificationResult> ClassificationApp::postprocess(
    const std::vector<void*>& outputs) {

    if (outputs.empty()) {
        return {};
    }

    // 获取输出属性
    rknn_tensor_attr* output_attrs = rknn_->getOutputAttrs();
    float* output_data = static_cast<float*>(outputs[0]);

    // 获取输出大小
    int output_size = 1;
    for (uint32_t i = 0; i < output_attrs[0].n_dims; i++) {
        output_size *= output_attrs[0].dims[i];
    }

    // 应用softmax
    softmax(output_data, output_size);

    // 获取top-k结果
    std::vector<std::pair<float, int>> score_index_pairs;
    for (int i = 0; i < output_size; i++) {
        score_index_pairs.push_back({output_data[i], i});
    }

    std::sort(score_index_pairs.begin(), score_index_pairs.end(),
              std::greater<std::pair<float, int>>());

    // 构建结果
    std::vector<ClassificationResult> results;
    for (int i = 0; i < std::min(top_k_, static_cast<int>(score_index_pairs.size())); i++) {
        ClassificationResult result;
        result.confidence = score_index_pairs[i].first;
        result.class_id = score_index_pairs[i].second;

        if (result.class_id < static_cast<int>(class_names_.size())) {
            result.class_name = class_names_[result.class_id];
        } else {
            result.class_name = "class_" + std::to_string(result.class_id);
        }

        results.push_back(result);
    }

    return results;
}

void ClassificationApp::softmax(float* data, int size) {
    // 找到最大值
    float max_val = *std::max_element(data, data + size);

    // 计算exp并求和
    float sum = 0.0f;
    for (int i = 0; i < size; i++) {
        data[i] = std::exp(data[i] - max_val);
        sum += data[i];
    }

    // 归一化
    for (int i = 0; i < size; i++) {
        data[i] /= sum;
    }
}

void ClassificationApp::drawResults(cv::Mat& image,
                                   const std::vector<ClassificationResult>& results,
                                   const TimingInfo& timing) {
    // 绘制分类结果
    for (size_t i = 0; i < std::min(results.size(), static_cast<size_t>(3)); i++) {
        std::string text = std::to_string(i + 1) + ". " + results[i].class_name +
                          ": " + std::to_string(results[i].confidence);

        cv::Point position(10, 30 + i * 30);
        ImageUtils::drawClassification(image, text, results[i].confidence, position);
    }

    // 绘制性能信息
    std::string fps_text = "FPS: " + std::to_string(timing.fps);
    cv::putText(image, fps_text, cv::Point(10, image.rows - 60),
                cv::FONT_HERSHEY_SIMPLEX, 0.7, cv::Scalar(255, 255, 255), 2);

    std::string inference_text = "Inference: " +
                                std::to_string(timing.inference_time * 1000) + "ms";
    cv::putText(image, inference_text, cv::Point(10, image.rows - 30),
                cv::FONT_HERSHEY_SIMPLEX, 0.7, cv::Scalar(255, 255, 255), 2);
}

void ClassificationApp::predictBatch(const std::vector<std::string>& image_paths) {
    std::cout << "Processing " << image_paths.size() << " images..." << std::endl;

    for (size_t i = 0; i < image_paths.size(); i++) {
        try {
            TimingInfo timing;
            std::vector<ClassificationResult> results = predictImage(image_paths[i], &timing);

            std::cout << "Image " << (i + 1) << "/" << image_paths.size()
                      << ": " << image_paths[i] << std::endl;

            if (!results.empty()) {
                std::cout << "  Top prediction: " << results[0].class_name
                          << " (" << results[0].confidence << ")" << std::endl;
            }

            std::cout << "  Inference time: " << timing.inference_time * 1000 << "ms" << std::endl;

        } catch (const std::exception& e) {
            std::cerr << "Error processing " << image_paths[i] << ": " << e.what() << std::endl;
        }
    }
}

void ClassificationApp::predictVideo(const std::string& video_path,
                                    const std::string& output_path) {
    cv::VideoCapture cap(video_path);
    if (!cap.isOpened()) {
        throw std::runtime_error("Failed to open video: " + video_path);
    }

    // 获取视频信息
    int fps = static_cast<int>(cap.get(cv::CAP_PROP_FPS));
    int width = static_cast<int>(cap.get(cv::CAP_PROP_FRAME_WIDTH));
    int height = static_cast<int>(cap.get(cv::CAP_PROP_FRAME_HEIGHT));
    int total_frames = static_cast<int>(cap.get(cv::CAP_PROP_FRAME_COUNT));

    std::cout << "Video info: " << width << "x" << height
              << ", " << fps << "fps, " << total_frames << " frames" << std::endl;

    // 初始化视频写入器
    cv::VideoWriter writer;
    if (!output_path.empty()) {
        writer.open(output_path, cv::VideoWriter::fourcc('m', 'p', '4', 'v'),
                   fps, cv::Size(width, height));
    }

    cv::Mat frame;
    int frame_count = 0;
    double total_inference_time = 0;

    while (cap.read(frame)) {
        frame_count++;

        try {
            TimingInfo timing;
            std::vector<ClassificationResult> results = predict(frame, &timing);
            total_inference_time += timing.total_time;

            // 绘制结果
            drawResults(frame, results, timing);

            // 保存帧
            if (writer.isOpened()) {
                writer.write(frame);
            }

            // 显示
            cv::imshow("Classification", frame);
            if (cv::waitKey(1) == 'q') {
                break;
            }

            // 进度显示
            if (frame_count % 30 == 0) {
                double progress = static_cast<double>(frame_count) / total_frames * 100;
                double avg_fps = frame_count / total_inference_time;
                std::cout << "Progress: " << progress << "%, Avg FPS: " << avg_fps << std::endl;
            }

        } catch (const std::exception& e) {
            std::cerr << "Error processing frame " << frame_count << ": " << e.what() << std::endl;
        }
    }

    cap.release();
    if (writer.isOpened()) {
        writer.release();
    }
    cv::destroyAllWindows();

    double avg_fps = frame_count / total_inference_time;
    std::cout << "Video processing completed: " << frame_count
              << " frames, Avg FPS: " << avg_fps << std::endl;
}

void ClassificationApp::predictCamera(int camera_id) {
    cv::VideoCapture cap(camera_id);
    if (!cap.isOpened()) {
        throw std::runtime_error("Failed to open camera: " + std::to_string(camera_id));
    }

    // 设置摄像头参数
    cap.set(cv::CAP_PROP_FRAME_WIDTH, 640);
    cap.set(cv::CAP_PROP_FRAME_HEIGHT, 480);
    cap.set(cv::CAP_PROP_FPS, 30);

    std::cout << "Starting real-time classification, press 'q' to quit" << std::endl;

    cv::Mat frame;
    int frame_count = 0;
    double total_time = 0;

    while (cap.read(frame)) {
        frame_count++;

        try {
            TimingInfo timing;
            std::vector<ClassificationResult> results = predict(frame, &timing);
            total_time += timing.total_time;

            // 绘制结果
            drawResults(frame, results, timing);

            // 显示
            cv::imshow("Real-time Classification", frame);
            if (cv::waitKey(1) == 'q') {
                break;
            }

        } catch (const std::exception& e) {
            std::cerr << "Error processing frame: " << e.what() << std::endl;
        }
    }

    cap.release();
    cv::destroyAllWindows();

    double avg_fps = frame_count / total_time;
    std::cout << "Real-time classification ended: " << frame_count
              << " frames, Avg FPS: " << avg_fps << std::endl;
}

Main Program

// main.cpp
#include <iostream>
#include <vector>
#include <string>
#include "classification_app.h"

void printUsage(const char* program_name) {
    std::cout << "Usage: " << program_name << " [options]" << std::endl;
    std::cout << "Options:" << std::endl;
    std::cout << "  --model <path>        RKNN model path (required)" << std::endl;
    std::cout << "  --classes <path>      Class names file path" << std::endl;
    std::cout << "  --image <path>        Input image path" << std::endl;
    std::cout << "  --video <path>        Input video path" << std::endl;
    std::cout << "  --camera <id>         Camera ID (default: 0)" << std::endl;
    std::cout << "  --output <path>       Output video path" << std::endl;
    std::cout << "  --batch <paths...>    Batch process images" << std::endl;
    std::cout << "  --size <w> <h>        Input size (default: 224 224)" << std::endl;
    std::cout << "  --help                Show this help message" << std::endl;
}

int main(int argc, char* argv[]) {
    if (argc < 2) {
        printUsage(argv[0]);
        return -1;
    }

    std::string model_path;
    std::string classes_path;
    std::string image_path;
    std::string video_path;
    std::string output_path;
    std::vector<std::string> batch_paths;
    int camera_id = 0;
    cv::Size input_size(224, 224);
    bool use_camera = false;

    // 解析命令行参数
    for (int i = 1; i < argc; i++) {
        std::string arg = argv[i];

        if (arg == "--help") {
            printUsage(argv[0]);
            return 0;
        } else if (arg == "--model" && i + 1 < argc) {
            model_path = argv[++i];
        } else if (arg == "--classes" && i + 1 < argc) {
            classes_path = argv[++i];
        } else if (arg == "--image" && i + 1 < argc) {
            image_path = argv[++i];
        } else if (arg == "--video" && i + 1 < argc) {
            video_path = argv[++i];
        } else if (arg == "--camera" && i + 1 < argc) {
            camera_id = std::stoi(argv[++i]);
            use_camera = true;
        } else if (arg == "--output" && i + 1 < argc) {
            output_path = argv[++i];
        } else if (arg == "--batch") {
            // 收集所有后续的图像路径
            while (i + 1 < argc && argv[i + 1][0] != '-') {
                batch_paths.push_back(argv[++i]);
            }
        } else if (arg == "--size" && i + 2 < argc) {
            input_size.width = std::stoi(argv[++i]);
            input_size.height = std::stoi(argv[++i]);
        }
    }

    if (model_path.empty()) {
        std::cerr << "Error: Model path is required" << std::endl;
        printUsage(argv[0]);
        return -1;
    }

    try {
        // 创建分类应用
        ClassificationApp app(model_path, classes_path, input_size);

        if (!image_path.empty()) {
            // 单张图像预测
            TimingInfo timing;
            std::vector<ClassificationResult> results = app.predictImage(image_path, &timing);

            std::cout << "\nImage: " << image_path << std::endl;
            std::cout << "Classification results:" << std::endl;
            for (size_t i = 0; i < results.size(); i++) {
                std::cout << "  " << (i + 1) << ". " << results[i].class_name
                          << ": " << results[i].confidence << std::endl;
            }

            std::cout << "\nTiming:" << std::endl;
            std::cout << "  Preprocess: " << timing.preprocess_time * 1000 << "ms" << std::endl;
            std::cout << "  Inference: " << timing.inference_time * 1000 << "ms" << std::endl;
            std::cout << "  Postprocess: " << timing.postprocess_time * 1000 << "ms" << std::endl;
            std::cout << "  Total: " << timing.total_time * 1000 << "ms" << std::endl;
            std::cout << "  FPS: " << timing.fps << std::endl;

        } else if (!batch_paths.empty()) {
            // 批量预测
            app.predictBatch(batch_paths);

        } else if (!video_path.empty()) {
            // 视频预测
            app.predictVideo(video_path, output_path);

        } else if (use_camera) {
            // 实时摄像头预测
            app.predictCamera(camera_id);

        } else {
            std::cout << "Please specify input source: --image, --video, --camera, or --batch" << std::endl;
        }

    } catch (const std::exception& e) {
        std::cerr << "Error: " << e.what() << std::endl;
        return -1;
    }

    return 0;
}

Build Script

#!/bin/bash
# build.sh

# 设置编译参数
CXX=g++
CXXFLAGS="-std=c++11 -O3 -Wall"
INCLUDES="-I/usr/include/opencv4 -I/usr/include/rknn"
LIBS="-lopencv_core -lopencv_imgproc -lopencv_imgcodecs -lopencv_videoio -lopencv_highgui -lrknn_api"

# 编译
echo "Compiling RKNN Classification App..."

$CXX $CXXFLAGS $INCLUDES \
    rknn_inference.cpp \
    image_utils.cpp \
    classification_app.cpp \
    main.cpp \
    -o classification_app \
    $LIBS

if [ $? -eq 0 ]; then
    echo "Compilation successful!"
    echo "Usage: ./classification_app --model <model_path> [options]"
else
    echo "Compilation failed!"
    exit 1
fi

CMakeLists.txt

cmake_minimum_required(VERSION 3.10)
project(RKNNClassificationApp)

set(CMAKE_CXX_STANDARD 11)
set(CMAKE_CXX_STANDARD_REQUIRED ON)

# 查找依赖包
find_package(OpenCV REQUIRED)
find_package(PkgConfig REQUIRED)

# 设置包含目录
include_directories(${OpenCV_INCLUDE_DIRS})
include_directories(/usr/include/rknn)

# 添加可执行文件
add_executable(classification_app
    rknn_inference.cpp
    image_utils.cpp
    classification_app.cpp
    main.cpp
)

# 链接库
target_link_libraries(classification_app
    ${OpenCV_LIBS}
    rknn_api
)

# 编译选项
target_compile_options(classification_app PRIVATE -O3 -Wall)

# 安装
install(TARGETS classification_app DESTINATION bin)

Performance Optimization and Best Practices

Memory Management Optimization

# Python内存优化
import gc
import psutil
import os

class MemoryOptimizer:
    """内存优化工具"""

    def __init__(self):
        self.process = psutil.Process(os.getpid())
        self.initial_memory = self.get_memory_usage()

    def get_memory_usage(self):
        """获取当前内存使用量(MB)"""
        return self.process.memory_info().rss / 1024 / 1024

    def optimize_memory(self):
        """优化内存使用"""
        # 强制垃圾回收
        gc.collect()

        # 清理OpenCV缓存
        cv2.setUseOptimized(True)

        current_memory = self.get_memory_usage()
        saved_memory = self.initial_memory - current_memory

        print(f"内存优化: 节省 {saved_memory:.1f}MB")
        return saved_memory

    def monitor_memory(self, func, *args, **kwargs):
        """监控函数内存使用"""
        start_memory = self.get_memory_usage()

        result = func(*args, **kwargs)

        end_memory = self.get_memory_usage()
        memory_used = end_memory - start_memory

        print(f"函数 {func.__name__} 使用内存: {memory_used:.1f}MB")

        return result

# 使用示例
optimizer = MemoryOptimizer()

# 在推理循环中定期优化内存
for i, image_path in enumerate(image_paths):
    result = app.predict_image(image_path)

    # 每处理100张图像优化一次内存
    if i % 100 == 0:
        optimizer.optimize_memory()

Multi-Threaded Processing

// 多线程处理示例
#include <thread>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <atomic>

class ThreadSafeQueue {
private:
    std::queue<cv::Mat> queue_;
    std::mutex mutex_;
    std::condition_variable condition_;
    std::atomic<bool> finished_{false};

public:
    void push(const cv::Mat& item) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(item);
        condition_.notify_one();
    }

    bool pop(cv::Mat& item) {
        std::unique_lock<std::mutex> lock(mutex_);
        condition_.wait(lock, [this] { return !queue_.empty() || finished_; });

        if (queue_.empty()) {
            return false;
        }

        item = queue_.front();
        queue_.pop();
        return true;
    }

    void finish() {
        finished_ = true;
        condition_.notify_all();
    }
};

class MultiThreadClassifier {
private:
    std::unique_ptr<ClassificationApp> app_;
    ThreadSafeQueue input_queue_;
    ThreadSafeQueue output_queue_;
    std::vector<std::thread> workers_;
    int num_threads_;

public:
    MultiThreadClassifier(const std::string& model_path, int num_threads = 4)
        : num_threads_(num_threads) {
        app_ = std::make_unique<ClassificationApp>(model_path);

        // 启动工作线程
        for (int i = 0; i < num_threads_; i++) {
            workers_.emplace_back(&MultiThreadClassifier::workerThread, this);
        }
    }

    ~MultiThreadClassifier() {
        input_queue_.finish();
        for (auto& worker : workers_) {
            if (worker.joinable()) {
                worker.join();
            }
        }
    }

    void processImages(const std::vector<std::string>& image_paths) {
        // 添加图像到队列
        for (const auto& path : image_paths) {
            cv::Mat image = cv::imread(path);
            if (!image.empty()) {
                input_queue_.push(image);
            }
        }

        input_queue_.finish();

        // 等待所有工作线程完成
        for (auto& worker : workers_) {
            if (worker.joinable()) {
                worker.join();
            }
        }
    }

private:
    void workerThread() {
        cv::Mat image;
        while (input_queue_.pop(image)) {
            try {
                auto results = app_->predict(image);
                // 处理结果...
            } catch (const std::exception& e) {
                std::cerr << "Worker thread error: " << e.what() << std::endl;
            }
        }
    }
};

Summary

Through this chapter, you have mastered:

1. Deployment method comparison: understood the advantages, disadvantages, and applicable scenarios of the Python and C++ APIs
2. Python API deployment: mastered the complete Python deployment flow and code implementation
3. C++ API deployment: learned the high-performance C++ deployment method
4. Hands-on application: completed the development of a complete image classification application
5. Performance optimization: learned memory-management and multi-threading optimization techniques

At this point, you have the complete capability to deploy and run custom NPU models on the MB-E30P development board.