板端运行自定义模型
概述
在完成模型转换后，下一步就是在MB-E30P开发板上部署和运行自定义模型。本章将详细介绍两种主要的部署方式，并通过实际案例演示如何编写完整的图像分类应用。
5.1 两种部署方式比较
Python API 部署
优势
- 开发效率高: Python语法简洁，开发速度快
- 调试方便: 丰富的调试工具和库支持
- 生态丰富: 大量的第三方库可直接使用
- 原型验证: 适合快速原型开发和算法验证
- 灵活性强: 易于修改和扩展功能
劣势
- 性能开销: Python解释执行，存在性能损失
- 内存占用: 相对C++占用更多内存
- 启动时间: 解释器启动和模块加载需要时间
- 依赖复杂: 需要Python运行环境和相关库
适用场景
# 适合以下场景:
scenarios = [
"算法原型验证",
"快速功能演示",
"教学和学习",
"复杂数据处理",
"与其他Python服务集成",
"对性能要求不严格的应用"
]C/C++ API 部署
优势
- 性能优异: 编译执行，运行效率高
- 内存效率: 内存占用小，资源利用率高
- 启动快速: 无解释器开销，启动时间短
- 部署简单: 编译后的可执行文件，依赖少
- 系统集成: 易于与系统服务和其他C/C++程序集成
劣势
- 开发复杂: 需要处理内存管理、指针等
- 调试困难: 调试工具相对有限
- 开发周期: 编译-测试周期较长
- 灵活性: 修改功能需要重新编译
适用场景
// 适合以下场景:
std::vector<std::string> scenarios = {
"生产环境部署",
"实时性要求高的应用",
"资源受限的嵌入式系统",
"系统服务和守护进程",
"与硬件底层交互",
"大规模批量处理"
};性能对比测试
# 性能测试脚本
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 部署讲解
基础API使用
核心类和方法
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()图像预处理模块
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()后处理模块
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 部署讲解
基础C++ API使用
// 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_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 部署实战：编写图像分类应用
Python版本图像分类应用
#!/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++版本图像分类应用
// 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.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;
}编译脚本
#!/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
fiCMakeLists.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)性能优化和最佳实践
内存管理优化
# 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()多线程处理
// 多线程处理示例
#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;
}
}
}
};总结
通过本章的学习，您已经掌握了：
- 部署方式对比: 了解了Python和C++ API的优劣势和适用场景
- Python API部署: 掌握了完整的Python部署流程和代码实现
- C++ API部署: 学会了高性能的C++部署方法
- 实战应用: 完成了完整的图像分类应用开发
- 性能优化: 学会了内存管理和多线程优化技巧
至此，您已经具备了在MB-E30P开发板上部署和运行自定义NPU模型的完整能力。