RDK X5 mipi_camera.py(FCOS 目标检测 Demo)完整解析

运行环境:RDK OS3.x(X5 平台);

核心组件:hobot_vio(相机 / 显示) + hobot_dnn(BPU 推理) + libpostprocess.so(C 后处理) 业务流程:MIPI 相机采集 NV12 图 → BPU FCOS 推理 → C 库后处理解析检测框 → HDMI 叠加绘制目标框

一、整体数据流框图

plaintext

MIPI Sensor(F37) → hobot_vio.Camera采集 → 裁剪缩放512×512 NV12图像
                                 ↓
                         models[0].forward() BPU硬件推理
                                 ↓
                 推理输出tensor → 多进程调度调用libpostprocess.so(FCOS解码)
                                 ↓
                 解析JSON检测结果 → HDMI图层叠加绘制方框+类别文字

二、模块逐段解析

1. 头文件与兼容导入

python

运行

try:
    from hobot_vio import libsrcampy as srcampy
except ImportError:
    from hobot_vio_rdkx5 import libsrcampy as srcampy
try:
    from hobot_dnn import pyeasy_dnn as dnn
except ImportError:
    from hobot_dnn_rdkx5 import pyeasy_dnn as dnn
  • 作用:兼容新旧 SDK 包名;srcampy = 地平线 VIO 库(Camera/Display/Bind);dnn = BPU 推理 Python 接口
  • ⚠️ 你之前报错根源:此模块预编译依赖 numpy<2.0

2. C 类型结构体定义(ctypes 桥接 C 库)

python

运行

class hbSysMem_t / hbDNNTensor_t / FcosPostProcessInfo_t ...
libpostprocess = ctypes.CDLL('/usr/lib/libpostprocess.so')

重点:

  1. Python 通过ctypes直接调用地平线提供的 C 语言后处理动态库 libpostprocess.so
  2. 手动对齐 C 结构体,将 BPU 推理输出 tensor 内存地址传递给 C 函数FcosdoProcess
  3. get_Postprocess_result 返回 JSON 字符串,包含所有目标框坐标、置信度、类别 ID

优势:后处理在 C 中运行,性能远高于纯 Python 解析,避免占用 CPU

3. 显示分辨率自动适配函数 get_display_res()

python

运行

disp = srcampy.Display()
resolution_list = disp.get_display_res()

逻辑:

  1. 优先尝试 1920×1080
  2. 如果 HDMI 不支持,则自动选择小于等于 1080P 的最大分辨率
  3. 返回disp_w/disp_h,后续绘制框时用于坐标缩放

4. 多进程并行后处理 ParallelExector

python

运行

self._pool = multiprocessing.Pool(processes=self.parallel_num, maxtasksperchild=5)
self.workers = BoundedSemaphore(self.parallel_num)

核心设计:

  1. 推理(BPU 硬件)与后处理(CPU)解耦
    • 主线程:持续采集图像 + 调用 BPU 推理(硬件异步)
    • 子进程池:并发运行 FCOS 解析、绘制逻辑
  2. BoundedSemaphore 信号量限制最大并发数,防止进程无限创建、内存暴涨
  3. maxtasksperchild=5:执行若干任务后重建子进程,规避长期运行内存泄漏

⚠️注意:multiprocess 多进程在 RDK 上有坑:全局变量跨进程不共享,代码中image_counter使用multiprocessing.Value实现跨进程计数。

5. run(outputs):单帧后处理核心函数

python

运行

libpostprocess.FcosdoProcess(...)
result_str = get_Postprocess_result(...)
data = json.loads(result_str[14:])

执行步骤:

  1. 将 BPU 输出 tensor 内存地址绑定到 C 结构体
  2. 调用 C 库 FCOS 解码(5 个 stride 层 8/16/32/64/128 特征图)
  3. 返回 json,截取字符串[14:]跳过库内置前缀字符
  4. 坐标缩放:模型输入是512×512,输出框坐标映射回 HDMI 显示分辨率disp_w×disp_h

    python

    运行

    bbox = scale_bbox(bbox, 512, 512, disp_w, disp_h)
    
  5. 边界截断 limit_display_cord,防止框超出画面
  6. HDMI 叠加绘制:

    python

    运行

    disp.set_graph_rect()   # 绘制矩形框
    disp.set_graph_word()   # 绘制类别+置信度文字
    
    • box_color_ARGB:转换成 HDMI 接口要求的 ARGB32 颜色格式

6. 主程序初始化 if __name__ == '__main__':

6.1 加载模型

python

运行

models = dnn.load('../models/fcos_512x512_nv12.bin')
  • *.bin:地平线经过 HBIR 编译完成、适配 BPU 的模型文件;输入格式 NV12(和相机输出格式一致,无需 RGB 转换,节省 CPU)
6.2 FCOS 后处理参数配置

python

运行

fcos_postprocess_info.height = 512    # 模型输入尺寸
fcos_postprocess_info.score_threshold = 0.5   # 置信度阈值
fcos_postprocess_info.nms_threshold = 0.6      # NMS阈值
fcos_postprocess_info.nms_top_k = 5            # 单帧最多输出5个目标
6.3 MIPI 相机打开

python

运行

cam.open_cam(0, -1, -1, [w, disp_w], [h, disp_h],sensor_height,sensor_width)

参数含义(RDK 标准 API):

  1. 0:camera 通道号
  2. [w, disp_w]:两路输出分辨率
    • 一路 512×512 → 送入 BPU 推理
    • 一路 disp_w×disp_h → HDMI 原始预览画面

VIO 硬件完成图像缩放,不需要 CPU 做 resize,零拷贝优势

6.4 HDMI 显示分层(关键!)

python

运行

disp.display(0, disp_w, disp_h)
srcampy.bind(cam, disp)        # 图层0:原始相机画面
disp.display(3, disp_w, disp_h) # 图层3:图形叠加层(画框/文字)

地平线 HDMI 支持多图层叠加:

  • Layer0:原始摄像头视频流
  • Layer3:OSD 图层(绘制检测框,透明叠加在视频上) srcampy.bind(cam, disp) 实现相机数据流直接硬件输送到 HDMI主线程不转发原始图像,大幅降低 CPU 占用。

7. 主循环采集链路

python

运行

img = cam.get_img(2, 512, 512)
img = np.frombuffer(img, dtype=np.uint8)
outputs = models[0].forward(img)
parallel_exe.infer(output_array)
  1. cam.get_img(2,512,512):获取缩放后 512×512 NV12 图像;参数2代表获取推理支路图像
  2. np.frombuffer:把 C 内存块封装成 numpy 数组(零拷贝,不复制像素数据
  3. models[0].forward(img):提交任务到 BPU 硬件异步推理
  4. 将推理输出送入进程池执行后处理,主线程立刻继续取下一帧,流水线运行

8. FPS 统计逻辑

python

运行

if image_counter.value == 100:
    print(f"fps: {100/(finish_time - start_time)}")

每累计 100 帧输出一次平均帧率;multiprocessing.Value保证跨进程计数器同步。

9. 信号退出处理

python

运行

signal.signal(signal.SIGINT, signal_handler)

捕获Ctrl+C,置位is_stop标志,退出 while 循环,执行:

python

运行

cam.close_cam()
disp.close()

释放相机、HDMI 硬件资源,防止硬件句柄泄漏。

三、关键技术亮点(地平线 RDK 平台特有)

  1. NV12 原生输入 MIPI 相机输出 NV12,模型直接接收 NV12,省去NV12→BGRCPU 颜色转换,降低时延。
  2. VIO 硬件多路缩放 + 硬件 HDMI 图层叠加 图像缩放、预览输出、OSD 绘制全部由 VPU 硬件完成,CPU 仅负责调度和 AI 业务。
  3. 零拷贝内存传输 frombuffer、ctypes 直接传递物理内存地址,像素数据不在用户态来回拷贝。
  4. BPU 推理与 CPU 后处理流水线并行 多进程分离推理、解码绘制,充分利用 X5 的 BPU + 多核 CPU。

四、结合你遇到问题的重点坑点清单

  1. numpy 版本限制 hobot_dnn二进制包基于 numpy1.x 编译,numpy >=2.0 直接触发numpy.core.multiarray failed to import ✅修复:pip install "numpy<2.0" --force-reinstall

  2. libpostprocess.so 路径依赖 代码硬编码路径/usr/lib/libpostprocess.so,升级镜像、容器环境如果缺少该库会直接崩溃。

  3. 多进程 + ctypes 注意事项 硬件句柄(cam/disp)不能放到子进程操作;本代码所有采集、硬件 API 全部在主线程,子进程仅做解析绘制。

  4. 分辨率匹配问题 模型固定输入 512×512;如果你更换其他尺寸模型,必须同步修改FcosPostProcessInfo_t宽高参数。

  5. MIPI Sensor 适配 当前默认 F37 传感器;更换其他 MIPI 镜头需要修改设备树 / 相机 yaml 配置,否则open_cam失败。

五、常见可改造方向

  1. 保存检测框图片:在run()函数内增加 cv2 保存逻辑
  2. 关闭 HDMI 输出,改为 RTSP 网络推流
  3. 替换 YOLO 系列模型:需要更换对应的libpostprocess.so和后处理结构体
  4. 去掉多进程,简化为单线程(适合低负载场景,调试更简单)
Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐