RDK X5 MIPI 摄像头 + FCOS 目标检测完整代码逐段深度解析

整体工程定位

这是地平线官方标准流水线： MIPI Sensor(IMX415/IMX477) → hobot_vio(srcampy) ISP采集NV12裸流 → hobot_dnn_rdkx5 BPU推理FCOS → libpostprocess.so C后处理 → HDMI叠加框渲染显示

和之前test_resnet18.py最大区别：

1. 不再使用 OpenCV VideoCapture 读取摄像头（USB摄像头），改用 libsrcampy（地平线 VIO 采集库，专门适配 MIPI CSI）
2. 任务从 ImageNet 分类 → FCOS Anchor-Free 目标检测（COCO 80 类）
3. 引入多进程并发后处理，防止后处理阻塞采集流水线
4. 硬件通路：摄像头数据流直通 HDMI；AI 框是图形层叠加，不会修改原始视频流

一、模块导入部分

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

1. hobot_vio / hobot_vio_rdkx5
   • RDK X3 / X5 的 MIPI ISP 采集 SDK；专门读取板载 CSI 摄像头
   • 直接输出NV12 内存裸数据，无需 BGR2NV12 转换，节省大量 CPU
   • ❗不要和cv2.VideoCapture混用读取 MIPI
2. hobot_dnn系列：和之前分类代码完全一致，BPU 推理 API
3. multiprocessing + BoundedSemaphore：实现多进程并行后处理 Python GIL 限制，线程无法真正并行计算，后处理重负载使用多进程

二、ctypes C 结构体区域（和分类代码对比重点）

python

运行

class FcosPostProcessInfo_t(ctypes.Structure):
    _fields_ = [
        ("height",ctypes.c_int),       # 模型输入尺寸 h=512
        ("width",ctypes.c_int),        # 模型输入尺寸 w=512
        ("ori_height",ctypes.c_int),   # 原图（HDMI画面分辨率）
        ("ori_width",ctypes.c_int),
        ("score_threshold",ctypes.c_float), # 置信度阈值0.5
        ("nms_threshold",ctypes.c_float),   # NMS IOU阈值0.6
        ("nms_top_k",ctypes.c_int),
        ("is_pad_resize",ctypes.c_int)      # 0=拉伸resize
    ]

✅ 对比分类结构体差异： 分类：ClassificationPostProcessInfo_t 检测：FcosPostProcessInfo_t

注意：此结构体缺少 FCOS 必须的 strides、class_num 参数，参数在底层 libpostprocess 写死适配这个 512 FCOS 模型；自定义训练模型需要扩展结构体！

python

运行

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

绑定 C 库 FCOS 后处理入口，对应分类的ClassificationPostProcess。

三、显示分辨率自动适配函数 get_display_res()

python

运行

def get_display_res():
    disp = srcampy.Display()
    resolution_list = disp.get_display_res()
    # 优先匹配1920×1080；不匹配自动选择支持的更小分辨率

作用：查询 HDMI 显示器支持分辨率，避免 disp.display () 参数不兼容黑屏。 全局变量 disp_w, disp_h = HDMI 画面输出分辨率（原始相机画面分辨率）。

四、辅助工具函数

1. scale_bbox(bbox, input_w, input_h, output_w, output_h)

• 模型输入是 512×512
• 相机原始画面 disp_w × disp_h（1920×1080） FCOS 后处理解码出的框坐标是512 图坐标系，该函数做坐标缩放映射回原图画面

关键：is_pad_resize=0 拉伸缩放；如果设置 1（等比例 + padding），坐标换算必须额外补偿 padding 偏移！

2. limit_display_cord()

防止边框坐标超出画面边界，避免 HDMI 渲染接口报错越界。

3. get_classes()

COCO 80 类别名称列表，和 FCOS 预训练模型匹配。

五、并行执行器 ParallelExector【重点设计】

python

运行

class ParallelExector(object):
    def __init__(self, counter, parallel_num=4):
        self._pool = multiprocessing.Pool(processes=4)
        self.workers = BoundedSemaphore(4)

    def infer(self, output):
        self.workers.acquire()
        self._pool.apply_async(func=run,args=(output, ),callback=self.task_done)

设计目的

采集、BPU 推理是高速流水线；后处理 + 绘制渲染耗时波动大 如果主线程直接执行 run ()，摄像头取帧会阻塞、丢帧。 方案： 主线程只做：取图 → forward 推理； 把输出 tensor 扔给进程池异步执行后处理。 BoundedSemaphore：限制最大并发进程，防止瞬间创建大量进程内存溢出。

⚠️ 多进程坑：全局变量在子进程是副本；不要在 run 里修改主线程资源。HDMI 显示接口属于共享硬件，并发绘制要注意冲突。

六、核心后处理函数 run (outputs)

python

运行

strides = [8, 16, 32, 64, 128] # FCOS FPN五层步长
for i in range(len(strides)):
    libpostprocess.FcosdoProcess(output_tensors[i], 
                                 output_tensors[i + 5], 
                                 output_tensors[i + 10], 
                                 ctypes.pointer(fcos_postprocess_info), i)

FCOS 模型输出排布规则（至关重要）

fcos_512x512_nv12.bin 一共 15 路输出 Tensor 5 层 FPN (P3~P7)，每层 3 个分支： cls(分类)、reg(偏移ltrb)、centerness(中心度) 索引排布：

plaintext

0~4: cls
5~9: reg
10~14: centerness

循环 i=0~4，逐层送入 C 后处理。

数据流

1. 将推理 outputs 的 numpy 内存指针赋值给hbDNNTensor_t结构体
2. C 库执行：反量化 → 解码 ltrb 框 → score=cls*centerness → 阈值过滤 → NMS
3. 返回 JSON 字符串：result_str[14:] 截断头部前缀（和分类[25:]类似，库版本不同前缀长度不一样）

HDMI 图形叠加（不是 OpenCV 绘图！）

python

运行

disp.set_graph_rect()   # 绘制矩形框
disp.set_graph_word()   # 绘制类别+置信度文字

特点：

1. 图层叠加（OSD）：不修改原始 NV12 视频帧；原始视频流继续直通 HDMI
2. 参数末尾 1/0：
   • 1：刷新整个 OSD 图层
   • 0：增量绘制
3. 颜色使用 ARGB 格式，直接调用底层 VIO 图形驱动，CPU 开销极低

七、主流程 main 函数（流水线骨架）

1. 加载 FCOS 模型

python

运行

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

模型输入尺寸固定：512×512，输入格式 NV12。

2. 初始化 FCOS 后处理结构体

python

运行

fcos_postprocess_info.height = 512
fcos_postprocess_info.width = 512
fcos_postprocess_info.ori_height = disp_h
fcos_postprocess_info.ori_width = disp_w
fcos_postprocess_info.score_threshold = 0.5
fcos_postprocess_info.nms_threshold = 0.6
fcos_postprocess_info.is_pad_resize = 0

3. MIPI 摄像头、HDMI 显示初始化（srcampy 核心）

python：MIPI 摄像头

运行

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

参数解读：

• 0：camera 通道号（RDK X5 CSI0 / CSI1 对应 0、1，对应 /dev/video0/video1）
• [w, disp_w]：一路输出 512（送入 BPU 推理），一路输出 1920×1080（直通 HDMI） 👉 ISP 硬件双通路输出！ 硬件同时输出两路图像： 通路 A：缩放 512×512 NV12 → 给 AI 推理 通路 B：原始 1080P NV12 → 直接送到 HDMI 显示 ✅ 巨大优势：硬件 ISP 完成缩放，CPU 零消耗

python：HDMI 显示

运行

disp = srcampy.Display()
disp.display(0, disp_w, disp_h)
srcampy.bind(cam, disp)    # 将摄像头原始视频流绑定HDMI直出
disp.display(3, disp_w, disp_h)  # 切换图层3用于绘制检测框OSD

srcampy.bind(cam, disp)：摄像头原始数据流硬件通路直连 HDMI，不需要 CPU 转发画面，延迟极低。

4. 主循环（核心流水线）

python

运行

while not is_stop:
    img = cam.get_img(2, 512, 512)       # 获取512×512 NV12 buffer
    img = np.frombuffer(img, dtype=np.uint8)
    outputs = models[0].forward(img)     # BPU推理
    parallel_exe.infer(output_array)     # 异步投递后处理

cam.get_img(2,512,512) 参数2：选择 ISP 输出的 512 分辨率通路；直接拿到 NV12 字节流。

对比之前图片测试代码： 旧代码：BGR → cv2.resize → bgr2nv12_opencv ()（CPU 大量运算） 本代码：ISP 硬件输出 NV12，完全消除图像格式转换 CPU 开销。

八、整套流水线时序图

plaintext

MIPI Sensor → ISP
                ├─通路1：1080P NV12 → 硬件bind直连HDMI（原始画面）
                └─通路2：硬件缩放512×512 NV12 → cam.get_img读取
                        → np.frombuffer
                        → dnn.forward(BPU推理)
                        → 扔进多进程池异步run()
                             ↳ ctypes调用libpostprocess FCOS后处理
                             ↳ 解析框坐标、缩放映射原图
                             ↳ disp OSD叠加矩形与文字

九、高频踩坑清单（工程实操重点）

1. 摄像头通道

cam.open_cam(0,...) → CSI0；改为 1 读取 CSI1，实现双摄。 ⚠️ 不要同时使用 srcampy + cv2.VideoCapture 读取同一个 MIPI 通道，会抢占驱动！

2. 模型与分辨率绑定

fcos_512x512_nv12.bin 固定输入 512；不要强行送入其他尺寸 NV12。

3. is_pad_resize 坐标大坑

• 0：拉伸 resize（当前代码）
• 1：等比例缩放 + 上下 / 左右填充黑边 如果修改为 1，scale_bbox 必须增加 padding 补偿逻辑，否则框位置偏移！

4. JSON 字符串切片

python

运行

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

libpostprocess 版本不同，前缀长度变化，切片数字报错就会 json 解析失败。

5. 多进程隐患

进程池频繁创建销毁会有内存泄漏；长时间 7×24 运行产品需要增加进程重启逻辑。

6. 输出 tensor 顺序不能乱

FCOS 输出必须严格：cls [0-4], reg [5-9], centerness [10-14]，onnx 导出顺序一旦颠倒，后处理直接失效。

7. HDMI OSD 图层冲突

多个进程同时调用set_graph_rect会出现框闪烁；尽量把所有绘制收敛到同一个进程。

十、和之前代码横向对比总结

表格

项目	test_resnet18.py（图片分类）	本 MIPI-FCOS 检测代码
图像来源	本地图片 cv2.imread	MIPI CSI hobot_vio ISP 硬件采集
格式转换	CPU 执行 BGR→NV12	ISP 直接输出 NV12，无 CPU 转换
任务	ImageNet1k 分类	FCOS COCO 80 类目标检测
模型输出	单一路 logits	15 路 tensor（cls/reg/centerness 五层 FPN）
后处理执行	主线程同步阻塞	多进程异步并发，防丢帧
画面显示	无显示	硬件 HDMI 直出 + OSD 叠加检测框
采集方式	离线图片	实时视频流循环采集

十一、改造拓展方向

1. 支持双摄 CSI0+CSI1 两路 FCOS 并行检测；
2. 把 FCOS 模型更换骨干：VarGConvNet/MobileNetV1 轻量化版本；
3. 去掉 libpostprocess.so，Python 实现 FCOS 后处理（方便自定义数据集）；
4. 增加保存检测截图、RTSP 推流；
5. 增加 ROI 感兴趣区域过滤，只识别画面指定区域目标。