RKVision开发日志(六)

一、相机进程增加功能

今天打算完成相机进程的第二个功能，即显示预览画面传递给前端，并注意预留重写接口
瑞芯微上预留的设备有：

# Rockchip特定设备
- /dev/mpp_service:/dev/mpp_service
- /dev/rga:/dev/rga
- /dev/dma_heap:/dev/dma_heap

其中，

/dev/mpp_service: Rockchip媒体处理平台服务设备，提供视频编解码的硬件加速接口 。这是RKMPP（Rockchip Media Process Platform）的核心接口，用于H.264/H.265等视频格式的硬件编解码。

/dev/rga: Rockchip 2D图形加速器设备，提供2D图像处理硬件加速。RGA可以执行图像缩放、格式转换、旋转等操作，在视频处理流水线中用于高效的图像预处理。

/dev/dma_heap: DMA内存堆设备，用于直接内存访问的内存分配。这个设备允许应用程序分配连续的物理内存，供硬件加速器（如NPU、VPU）直接访问，避免内存拷贝开销。

经查询资料，MPP和RGA可通过ffmpeg上调用，不过我在鲁班猫上用调ffmpeg会有BUG，调gstreamer倒是能用。

参阅frigate的代码，MPP是这样调用的，以下是AI的回答，我认为解释的很清楚了：

这段代码并不是直接用 Python “编写”了驱动或底层硬件逻辑，而是充当了一个“命令生成器”（Command Builder）的角色。
它的核心原理是：Python 负责根据硬件类型拼接好 FFmpeg 的命令行参数字符串，然后由 FFmpeg 去调用底层的硬件 API。
为了让你看懂，我们需要把这段代码拆解成三个部分：模板、填空、执行。

---

1. 核心概念：字典里的“模板” (The Templates)

看 PRESETS_HW_ACCEL_ENCODE_TIMELAPSE 这个字典。它的 Value（值）不是乱码，而是 FFmpeg 能够识别的命令行参数模板。
FFmpeg 本身是一个极其强大的工具，它通过特定的命令行参数来调用不同厂商的硬件加速接口。这段 Python 代码只是把这些繁琐的参数针对不同硬件预设好了。
举个例子，针对 Nvidia 显卡的配置：

"preset-nvidia-h264": "{0} -hide_banner -hwaccel cuda -hwaccel_output_format cuda -extra_hw_frames 8 {1} -c:v h264_nvenc {2}",

这里的每一个词都在指挥 FFmpeg 做事：

• -hwaccel cuda: 告诉 FFmpeg 解码时使用 CUDA（也就是 Nvidia GPU）。
• -hwaccel_output_format cuda: 告诉 FFmpeg 解码后的画面不要搬回 CPU 内存，直接留在 GPU 显存里（这是硬件加速高性能的关键，避免 PCIe 总线传输）。
• -c:v h264_nvenc: 告诉 FFmpeg 编码时使用 Nvidia 的 NVENC 硬件编码器。
• {0}, {1}, {2}: 这是 Python 的占位符，等着后续填入具体的输入文件、分辨率等参数。
  对比一下其他硬件：
• Intel QSV: -c:v h264_qsv (调用 Intel Quick Sync Video)
• 树莓派: -c:v h264_v4l2m2m (调用 Video4Linux2 内存到内存的驱动接口)
• AMD: -c:v h264_amf (调用 AMD 的 AMF 接口)
• Rockchip (瑞芯微): -c:v h264_rkmpp (调用 Rockchip 的 Media Process Platform)
  总结： 这段代码并没有直接操作硬件，它只是把告诉 FFmpeg “用哪个硬件”和“怎么用”的参数字符串写好了。

---

2. 填空过程：parse_preset_hardware_acceleration_decode 函数

看这个函数，它负责把上面提到的模板“填空”。

def parse_preset_hardware_acceleration_decode(
    arg: Any, # 比如 "preset-nvidia-h264"
    fps: int, # 比如 30
    width: int, # 比如 1920
    height: int, # 比如 1080
    gpu: int, # GPU 编号
) -> list[str]:
    # 1. 去字典里找对应的模板字符串
    decode = PRESETS_HW_ACCEL_DECODE.get(arg, None)
    if not decode:
        return None
    # 2. 获取 GPU 参数 (比如 /dev/dri/renderD128 或者 cuda:0)
    gpu_arg = _gpu_selector.get_gpu_arg(arg, gpu)
    # 3. 关键步骤：format()
    # 它把 decode 模板里的 {0}, {1}, {2}, {3} 替换成具体的数值
    # 比如变成: "ffmpeg -hwaccel cuda ... -r 30 -s 1920x1080 ..."
    formatted_string = decode.format(fps, width, height, gpu_arg)
    
    # 4. 切割成列表
    # 变成: ["ffmpeg", "-hwaccel", "cuda", ..., "-r", "30", ...]
    return formatted_string.split(" ")

实际运行时的效果可能是这样的：

• 输入: arg="preset-nvidia-h264", fps=30, width=1920, height=1080
• 模板: "{0} ... -hwaccel cuda ... {1} ..." (假设 {0}是输入路径占位，{1}是参数占位)
• 结果: 一个包含几十个字符串的列表，就像你在终端里敲入的命令被空格劈开一样。

---

3. 最终执行：Python 只是个传话的

虽然这段代码里没写，但通常返回这个 list[str] 后，代码会把它传给 Python 的 subprocess 模块：

import subprocess
cmd_list = parse_preset_hardware_acceleration_decode("preset-nvidia-h264", 30, 1920, 1080, 0)
# 此时 cmd_list 大概是: ['ffmpeg', '-hide_banner', '-hwaccel', 'cuda', ..., '-c:v', 'h264_nvenc', ...]
# 真正的调用发生在这里：
# Python 启动一个子进程运行 ffmpeg，并把刚才拼好的参数传给它
process = subprocess.Popen(cmd_list)

总结：它是怎么实现底层调用的？

1. 识别硬件：代码通过字典 Key（如 preset-rpi-64-h264）区分你是树莓派、N卡还是 Intel CPU。
2. 映射驱动：代码根据硬件类型，填入对应的 FFmpeg 编码器名称（h264_nvenc, h264_v4l2m2m 等）。
3. 零拷贝优化：代码加入了 -hwaccel_output_format 等参数，确保数据在内存和显存之间高效流转，不走冤枉路。
4. 生成命令：Python 拼接出完整的命令字符串。
5. 底层干活：FFmpeg 被启动后，它会去加载你系统中安装的 Nvidia 驱动、Intel Media Driver 或 V4L2 内核驱动。真正的硬件编码是由这些驱动和底层的 C/C++ 库完成的，Python 只是个负责把命令念给 FFmpeg 听的“秘书”。

如果你没看懂，可能是因为你不熟悉 FFmpeg 的命令行参数。理解这段代码的关键在于理解：-c:v h264_nvenc 这种参数才是真正的钥匙，而 Python 代码只是帮你管理这些钥匙的钥匙串。

二、 构造硬件管道流

由于我现在手头没有设备，只能是虚空构思一下然后预留接口。AI给我的调用相机加速为：

# 构造 GStreamer 管道字符串
gst_str = (
    "filesrc location=input.mp4 ! "  # 读取文件
    "qtdemux ! "                     # 解复用 MP4
    "h264parse ! "                   # 解析 H.264
    "mppvideodec ! "                 # 使用 MPP (Media Process Platform) 硬件解码
    "videoconvert ! "                # 转换颜色空间 (虽然这可能涉及 CPU，但有些 RK 驱动支持硬件转换)
    "video/x-raw,format=BGR ! "      # 确保输出 BGR 格式给 OpenCV
    "appsink"                        # 将数据送入应用程序
)

# 以 CAP_GSTREAMER 模式打开
cap = cv2.VideoCapture(gst_str, cv2.CAP_GSTREAMER)

图像处理方面最好不调用opencv，而是librga。官方文档为librga/docs/Rockchip_Developer_Guide_RGA_CN.md at main · airockchip/librga
可惜这是C版本的库，python怎么调用有待商榷…

存图也可以通过gstreamer实现，这个已经实践过了，在录制存图视频的时候使用的指令为：

gst-launch-1.0 -e \
  v4l2src device=/dev/video0 io-mode=mmap num-buffers=4500 \
    ! video/x-raw,format=NV12,width=1088,height=816,framerate=15/1 \
    ! mpph264enc \
        bps=16000000         \
        bps-max=20000000     \
        bps-min=8000000      \
        rc-mode=vbr          \
        gop=3               \
        qp-init=24           \
    ! h264parse config-interval=-1 \
    ! mp4mux \
    ! filesink location=Q2.mp4

在和AI亲切友好的沟通学习后，强大的gstreamer可以构造多条管道，并通过阀门”value”实现开启关闭，还能调用RGA进行缩放…

AI给的管道长这个样子：

[Camera] -> [Tee 分流器]
   ├─> [Queue] -> [Valve (录制开关)] -> [mpph264enc (硬编码)] -> [mp4mux] -> [filesink]
   ├─> [Queue] -> [rkrga (硬件缩放)] -> [appsink (预览图)] -> 写入共享内存A
   └─> [Queue] -> [appsink (原图)] -> 写入共享内存B

给的python调用代码长这个样子:

import cv2
import multiprocessing as mp
import numpy as np
import sys

# ---------------- 全局配置 ----------------
WIDTH, HEIGHT = 1088, 816
FPS = 15
PREVIEW_W, PREVIEW_H = 480, 360

# ---------------- 共享内存定义 ----------------
# 使用 multiprocessing.shared_memory 避免 IPC 时的数据拷贝
# 预览图共享内存
shm_preview = mp.shared_memory.SharedMemory(create=True, size=PREVIEW_W * PREVIEW_H * 3)
# 原图共享内存 (为了省事，这里假定处理进程接受 BGR，如果处理进程能接受 NV12 就更好了)
shm_raw = mp.shared_memory.SharedMemory(create=True, size=WIDTH * HEIGHT * 3)

def get_gst_pipeline():
    # 核心逻辑：使用 tee 分流，valve 控制录制，rkrga 负责缩放
    pipeline = f"""
    v4l2src device=/dev/video0 io-mode=mmap !
    video/x-raw,format=NV12,width={WIDTH},height={HEIGHT},framerate={FPS}/1 !
    tee name=t !

    %s 录制分支 %
    t. ! queue leaky=downstream max-size-buffers=2 !
    valve name=v_record drop=true !  <--- 关键：默认 drop=true (关闭录制)
    mpph264enc bps=16000000 !
    h264parse !
    mp4mux !
    filesink location=output.mp4 !

    %s 预览分支 (RGA 硬件缩放) %
    t. ! queue leaky=downstream max-size-buffers=2 !
    rkrga scale-w={PREVIEW_W} scale-h={PREVIEW_H} !  <--- RGA 硬件缩放
    video/x-raw,format=BGR !
    appsink name=appsink_preview emit-signals=false max-buffers=1 sync=false !

    %s 处理分支 (原始数据) %
    t. ! queue leaky=downstream max-size-buffers=2 !
    videoconvert ! 
    video/x-raw,format=BGR !
    appsink name=appsink_raw emit-signals=false max-buffers=1 sync=false
    """
    return pipeline

def update_preview(frame):
    # 将预览帧写入共享内存 (零拷贝或极低开销)
    # 这里为了演示简单使用了 numpy copy，直接操作 shm.buf 更快
    np_view = np.ndarray((PREVIEW_H, PREVIEW_W, 3), dtype=np.uint8, buffer=shm_preview.buf)
    np.copyto(np_view, frame)

def update_raw(frame):
    # 将原图写入共享内存
    np_view = np.ndarray((HEIGHT, WIDTH, 3), dtype=np.uint8, buffer=shm_raw.buf)
    np.copyto(np_view, frame)

def main():
    # 启动 GStreamer 管道
    gst_str = get_gst_pipeline()
    cap = cv2.VideoCapture(gst_str, cv2.CAP_GSTREAMER)
    
    # 获取管道中的元素，用于控制
    # 注意：OpenCV 没有直接暴露 GStreamer Element 的控制接口，
    # 实际工程中，这里通常建议直接使用 gi.repository.Gst (Python原生GStreamer绑定)
    # 但为了保持与 cv2.VideoCapture 的一致性，这里演示逻辑。
    # 如果必须用 Valve 控制，建议切换到 PyGst，或者用一个简单的 flag 逻辑。
    
    # 由于 cv2.VideoCapture 难以控制中间的 valve，这里给出控制逻辑的建议：
    # 我们可以维护一个全局状态 is_recording
    # 但因为 cv2 封装限制，下面我会说明如何绕过。
    
    print("Capture Process Running...")
    
    # 为了演示，我们假设只能通过启动子进程来控制录制，
    # 或者建议你改用 PyGst 来获取 valve 元素的句柄。
    # 如果你坚持用 cv2，只能把录制分支拆出去。
    
    # (此处为了代码可运行性，采用折中方案：Valve 控制在 GStreamer 内部，
    # Python 这里只负责读图。如果真的要按键控制 Valve，需要引入 PyGST)
    
    while True:
        # 1. 读取预览分支 (RGA 处理后的小图)
        ret_prev, frame_prev = cap.read(cv2.CAP_PROP_FRAME_WIDTH, cv2.CAP_PROP_FRAME_HEIGHT) 
        # 注意：cv2.VideoCapture 只能读取一个输出。
        # 在复杂的多路输出场景下，建议使用 GStreamer 的 Python 绑定，或者开两个 VideoCapture (如果是文件)。
        # 但对于 v4l2 实时流，开两个 cap 会争抢设备。
        
        # --- 结论：在极其复杂的分流需求下，cv2.VideoCapture 已经不够用了 ---
        # --- 建议使用 Python 的 gi.repository.GST ---
        
        break # 下面直接给出 PyGst 版本的伪代码核心逻辑

if __name__ == "__main__":
    # 实际上，为了满足你“热切换”、“多分支”、“硬件加速”的所有需求，
    # 最稳健的方式是采集进程不再使用 cv2.VideoCapture，而是直接使用 Gst。
    
    import gi
    gi.require_version('Gst', '1.0')
    gi.require_version('GstRkmpi', '1.0')
    from gi.repository import Gst, GLib
    
    Gst.init(None)
    
    pipeline_str = get_gst_pipeline()
    pipeline = Gst.parse_launch(pipeline_str)
    
    # 获取 valve 元件句柄，实现热切换
    valve_record = pipeline.get_by_name("v_record")
    
    # 获取 appsink
    appsink_preview = pipeline.get_by_name("appsink_preview")
    appsink_raw = pipeline.get_by_name("appsink_raw")
    
    # 设置回调
    def on_preview_frame(sink, data):
        sample = sink.emit("pull-sample")
        buf = sample.get_buffer()
        # ... 提取数据写入 shm_preview ...
        return Gst.FlowReturn.OK

    def on_raw_frame(sink, data):
        sample = sink.emit("pull-sample")
        # ... 提取数据写入 shm_raw ...
        return Gst.FlowReturn.OK
        
    appsink_preview.connect("new-sample", on_preview_frame, None)
    appsink_raw.connect("new-sample", on_raw_frame, None)
    
    pipeline.set_state(Gst.State.PLAYING)
    
    # 模拟主循环控制
    loop = GLib.MainLoop()
    
    # 模拟接收外部指令 (比如通过 socket 或 multiprocessing Queue)
    # def toggle_recording():
    #     current_drop = valve_record.get_property("drop")
    #     valve_record.set_property("drop", not current_drop)
    #     print(f"Recording switched to: {not current_drop}")
    
    try:
        loop.run()
    except KeyboardInterrupt:
        pass
        
    pipeline.set_state(Gst.State.NULL)

三、GStreamer

看力竭了…先学习一下gi这个库吧
gi是PyGObject，官网地址是概述 — PyGObject — Overview — PyGObject
不过看起来官网描述，像是类QT的UI类库…

[!NOTE] How does it work?
PyGObject uses GLib, GObject, GIRepository, libffi and other libraries to access the C library (libgtk-4.so) in combination with the additional metadata from the accompanying typelib file (Gtk-4.0.typelib) and dynamically provides a Python interface based on that information.

PyGObject 使用 GLib 、 GObject 、 GIRepository 、 libffi 和其他库来访问 C 库 (libgtk-4.so)，并结合随附的类型库文件 (Gtk-4.0.typelib) 中的附加元数据，动态地提供基于该信息的 Python 接口。

是PyGObject包括了 GObject包括了 GStreamer，所以下一个PyGObject，调用GStreamer。

框架梳理结束，接下来需要了解：

1. GStreamer 能实现什么功能，能否分流、resize，能否用阀门控制各个管道开关
2. python中如何拉流，如何控制阀门开关，如何嵌入参数

想进一步测试，但是这两天被喊去调录视频，而且尝试安装PyGObject库的时候发现没有发行版，只能自行编译，这对于跨端调试很不友好

经过和AI的友好沟通之后，ffmpeg也能实现上述功能，而且用的人也多，资料也多，跨端友好，还有一个库可以实现ffmpeg的api调用，再考察一下

参考资料：

• Intuduction | GStreamer中文教程
• 构建应用程序 — Building an Application