背景

手里有块 Rockchip RV1126B 的板子，就是EASY-EAI-PI2这块，上面跑着 Linux 6.1，接了颗 4K MIPI 摄像头。需求很简单：录 10 秒视频，保存为 MP4。

听起来 ffmpeg -f v4l2 -i /dev/video0 -t 10 output.mp4 一条命令的事。但现实是——

[video4linux2,v4l2 @ ...] Not a video capture device.
/dev/video0: No such device

这就开始了。

第一道坎：ffmpeg 不认摄像头

板子上 /dev/video0 到 /dev/video51 满满当当 52 个 video 设备。ffmpeg 挨个试了一遍，全部返回 Not a video capture device。

用 Python 直接 ioctl 查 V4L2 capabilities：

fd = os.open('/dev/video31', os.O_RDWR)
caps = fcntl.ioctl(fd, VIDIOC_QUERYCAP, buf)
# caps = 0x00000000  ← 一个能力位都没设

原来 Rockchip 的 ISP 驱动把这批设备注册成了 V4L2 OUTPUT 类型——它们是给 userspace 往硬件写数据用的，不是从硬件读数据。ffmpeg 的 v4l2 模块硬编码检查 V4L2_CAP_VIDEO_CAPTURE 标志位，拿不到就拒绝。

解决： 用 GStreamer。v4l2src 的 mplane 插件能绕过这个检查，正常从设备取帧。

ffmpeg 直采 → ❌ Not a video capture device
GStreamer   → ✓ 能取到 NV12 3840×2160 @ 120fps

于是方案变成 GStreamer 采集 + ffmpeg 编码 的管道架构。

第二道坎：画面被压扁了

好不容易拿到图像，发现画面是方的。看参数：1920×2160，宽高比 8:9 ≈ 0.89，接近正方形。

用 media-ctl 查 ISP 链路拓扑：

sensor → rkcif-mipi-lvds2          SGBRG10_1X10 / 3840×2160
           ↓
         rkisp-isp-subdev           YUYV8_2X8   / 3840×2160
           ├── rkisp_mainpath       3840×2160    ← 16:9 ✓
           └── rkisp_selfpath       1920×2160    ← 8:9  ✗

ISP 内部处理的是 3840×2160（标准 16:9），但 mainpath 被 rkaiq 3A 引擎独占，只能拿到 selfpath 的 1920×2160——宽度砍半但高度没变，画面横向压缩。

解决： 用 selfpath，采集后裁掉上下黑边：

1920×2160 → videocrop top=540 bottom=540 → 1920×1080 (16:9)

第三道坎：画面极暗 + 色彩偏差

selfpath 能用了，但画面暗得几乎什么都看不清。抓一帧分析：

Y 亮度: [28, 118]  平均 32
暗部(<50): 100%   亮部(>180): 0%

原来之前为了用 mainpath，每次录制都 systemctl stop rkaiq_3A。rkaiq 是 3A 引擎（Auto Exposure / Auto White Balance / Auto Focus），一停 ISP 的曝光参数就冻住了。

强行用 videobalance brightness=0.2 把亮度提上去后，Y 从 32→107 是亮了，但暗部信噪比太低，色度通道提亮后出现偏差——黄色变成了蓝色。

真正的解决： 不停止 rkaiq。selfpath 本身就不需要独占设备，可以和 rkaiq 并存。rkaiq 全程运行，3A 持续工作，画面亮度和色彩自然正常。

第四道坎：NV12 → I420 转换的色彩坑

中间走过一段弯路：GStreamer 管道里加了 videoconvert 把 NV12 转 I420 再交给 ffmpeg。

NV12 的 UV 是交错排列 (UVUVUV...)，I420 是分开排列 (UUU...VVV...)。GStreamer 的 videoconvert 在 pipe 场景下对这个转换的处理有问题，导致色偏。PSNR 对比确认转换引入额外损失。

解决： 不做任何转换。ffmpeg 原生支持 NV12 输入：

ffmpeg -f rawvideo -pixel_format nv12 -video_size 1920x1080 ...

同帧编码→解码对比：PSNR Y:45.1dB UV:49.7dB，几乎无损。

最终架构

rkaiq_3A (全程运行, 无中断)
    │
    ▼
/dev/video32 (rkisp_selfpath)
    │ NV12 1920×2160 @ 120fps
    ▼
videobalance (可选亮度微调)
    │
    ▼
videocrop top=540 bottom=540
    │ NV12 1920×1080 (16:9)
    ▼
fdsink → pipe → ffmpeg libx264 ultrafast
    │ bt709 yuv420p
    ▼
MP4 文件 (10s, 15fps)

使用方式

# ffmpeg 后端 (默认, ~15fps)
python3 camera_record.py -b ffmpeg -t 10

# GStreamer 纯后端 (~8fps, 无 ffmpeg 依赖)
python3 camera_record.py -b gst -t 10

# 微调亮度
python3 camera_record.py -b ffmpeg -B 0.1

代码实现：

#!/usr/bin/env python3
"""
摄像头视频录制脚本
从摄像头采集视频，H.264编码保存为MP4文件
适用平台: Rockchip RV1126B (rkisp_mainpath + rkaiq 3A)

用法:
  python3 camera_record.py [--backend ffmpeg|gst] [--time N] [--brightness N]

  --backend ffmpeg   GStreamer采集 + ffmpeg libx264编码 (默认, ~15fps)
  --backend gst      GStreamer全链路采集+编码 (~8fps, 纯GStreamer)
  --time N           录制时长，默认10秒
  --brightness N     亮度补偿 (-1~1), 默认0.2
"""

import subprocess
import os
import sys
import time
import json
import argparse
from datetime import datetime

# ============ 默认配置 ============
CAMERA_DEVICE = "/dev/video32"           # rkisp_selfpath, 与 rkaiq 共存, 3A正常
OUTPUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "videos")
OUTPUT_FPS = 15
SRC_WIDTH = 1920                         # selfpath 原生
SRC_HEIGHT = 2160
CROP_TOP = 540                           # 裁掉上下黑边 → 1920x1080 (16:9)
CROP_BOTTOM = 540
OUT_WIDTH = 1920
OUT_HEIGHT = 1080
BITRATE = "2000k"
CRF = "26"
BRIGHTNESS = 0.0                         # rkaiq 3A 正常曝光, 一般不需补偿
SUDO_PASSWORD = "******"
# =============================


def sudo(cmd):
    return subprocess.run(
        ["sudo", "-S"] + cmd,
        input=SUDO_PASSWORD + "\n",
        capture_output=True, text=True
    )


def rkaiq_stop():
    print("暂停 rkaiq_3A ...", end=" ")
    r = sudo(["systemctl", "stop", "rkaiq_3A"])
    print("失败" if r.returncode != 0 else "✓")
    time.sleep(1)
    sudo(["fuser", "-k", CAMERA_DEVICE])


def rkaiq_start():
    print("恢复 rkaiq_3A ...", end=" ")
    r = sudo(["systemctl", "start", "rkaiq_3A"])
    print("失败" if r.returncode != 0 else "✓")


def check_camera():
    if not os.path.exists(CAMERA_DEVICE):
        print(f"[错误] 设备 {CAMERA_DEVICE} 不存在")
        return False
    if not os.access(CAMERA_DEVICE, os.R_OK | os.W_OK):
        return "sg_needed"
    return True


def get_output_path(backend):
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    os.makedirs(OUTPUT_DIR, mode=0o755, exist_ok=True)
    return os.path.join(OUTPUT_DIR, f"record_{timestamp}_{backend}.mp4")


def probe_video(path):
    try:
        r = subprocess.run(
            ["ffprobe", "-v", "quiet", "-print_format", "json",
             "-show_format", "-show_streams", path],
            capture_output=True, text=True, timeout=5
        )
        info = json.loads(r.stdout) if r.stdout else {}
        fmt = info.get("format", {})
        for s in info.get("streams", []):
            if s.get("codec_type") == "video":
                return {
                    "duration": float(fmt.get("duration", 0)),
                    "width": s.get("width"),
                    "height": s.get("height"),
                    "fps": s.get("r_frame_rate"),
                    "bitrate": int(fmt.get("bit_rate", 0)) // 1000,
                    "pix_fmt": s.get("pix_fmt"),
                    "colorspace": s.get("color_space"),
                }
    except Exception:
        pass
    return None


# ============================================================
#  ffmpeg 后端: GStreamer采集 → 管道 → ffmpeg libx264编码
# ============================================================
def record_ffmpeg(record_seconds, output_path, brightness):
    capture_frames = record_seconds * OUTPUT_FPS

    gst_pipeline = (
        f"v4l2src device={CAMERA_DEVICE} num-buffers={capture_frames} ! "
        f"video/x-raw,format=NV12,width={SRC_WIDTH},height={SRC_HEIGHT} ! "
        f"videobalance brightness={brightness} ! "
        f"videocrop top={CROP_TOP} bottom={CROP_BOTTOM} ! "
        f"fdsink"
    )

    ffmpeg_cmd = [
        "ffmpeg",
        "-f", "rawvideo",
        "-pixel_format", "nv12",
        "-video_size", f"{SRC_WIDTH}x{SRC_HEIGHT - CROP_TOP - CROP_BOTTOM}",
        "-framerate", str(OUTPUT_FPS),
        "-i", "pipe:0",
        "-c:v", "libx264",
        "-preset", "ultrafast",
        "-crf", CRF,
        "-maxrate", BITRATE,
        "-bufsize", str(int(BITRATE.replace("k", "000")) * 2),
        "-pix_fmt", "yuv420p",
        "-colorspace", "bt709",
        "-color_primaries", "bt709",
        "-color_trc", "bt709",
        "-movflags", "+faststart",
        "-y", output_path
    ]

    print("开始录制...\n")

    camera_status = check_camera()
    if camera_status == "sg_needed":
        gst_proc = subprocess.Popen(
            ["sg", "video", "-c", f"gst-launch-1.0 -e {gst_pipeline}"],
            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL
        )
    else:
        gst_proc = subprocess.Popen(
            ["gst-launch-1.0", "-e"] + gst_pipeline.split(),
            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL
        )

    ffmpeg_proc = subprocess.Popen(
        ffmpeg_cmd, stdin=gst_proc.stdout,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
    )
    gst_proc.stdout.close()

    start = time.time()
    _, ffmpeg_stderr = ffmpeg_proc.communicate(
        timeout=record_seconds * 2 + 30
    )
    gst_proc.wait(timeout=5)
    elapsed = time.time() - start

    if os.path.exists(output_path) and os.path.getsize(output_path) > 1000:
        print(f"✓ 录制完成! (耗时 {elapsed:.1f}s)")
        return True, ffmpeg_stderr
    return False, ffmpeg_stderr


# ============================================================
#  GStreamer 后端: 纯GStreamer采集+编码 → filesink
# ============================================================
def record_gst(record_seconds, output_path, brightness):
    capture_frames = record_seconds * OUTPUT_FPS

    gst_pipeline = (
        f"v4l2src device={CAMERA_DEVICE} num-buffers={capture_frames} ! "
        f"video/x-raw,format=NV12,width={SRC_WIDTH},height={SRC_HEIGHT} ! "
        f"videobalance brightness={brightness} ! "
        f"videocrop top={CROP_TOP} bottom={CROP_BOTTOM} ! "
        f"videoconvert ! video/x-raw,format=I420 ! "
        f"openh264enc bitrate={int(BITRATE.replace('k','000'))} complexity=high ! "
        f"h264parse ! "
        f"mp4mux ! "
        f"filesink location={output_path}"
    )

    print("开始录制...\n")

    camera_status = check_camera()
    cmd = f"gst-launch-1.0 -e {gst_pipeline}"
    if camera_status == "sg_needed":
        proc = subprocess.Popen(
            ["sg", "video", "-c", cmd],
            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
        )
    else:
        proc = subprocess.Popen(
            ["gst-launch-1.0", "-e"] + gst_pipeline.split(),
            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
        )

    start = time.time()
    stdout, stderr = proc.communicate(timeout=record_seconds * 2 + 30)
    elapsed = time.time() - start

    if os.path.exists(output_path) and os.path.getsize(output_path) > 1000:
        print(f"✓ 录制完成! (耗时 {elapsed:.1f}s)")
        return True, stderr
    return False, stderr


# ============================================================
def main():
    parser = argparse.ArgumentParser(
        description="摄像头视频录制 - Rockchip RV1126B"
    )
    parser.add_argument(
        "--backend", "-b", choices=["ffmpeg", "gst"], default="ffmpeg",
        help="编码后端: ffmpeg (libx264,较快) / gst (openh264,纯GStreamer)"
    )
    parser.add_argument(
        "--time", "-t", type=int, default=10,
        help="录制时长 (秒), 默认10"
    )
    parser.add_argument(
        "--brightness", "-B", type=float, default=BRIGHTNESS,
        help=f"亮度补偿 (-1~1), 默认{BRIGHTNESS}"
    )
    args = parser.parse_args()

    backend = args.backend
    record_seconds = args.time
    brightness = args.brightness
    output_path = get_output_path(backend)

    encoder_name = {
        "ffmpeg": "libx264 ultrafast (GStreamer→管道→ffmpeg)",
        "gst":    "openh264enc (纯GStreamer)"
    }[backend]

    print(f"============================================")
    print(f"  摄像头视频录制")
    print(f"============================================")
    print(f"  后端:     {backend} ({encoder_name})")
    print(f"  设备:     {CAMERA_DEVICE} (rkisp_selfpath, rkaiq 3A 运行中)")
    print(f"  采集:     {SRC_WIDTH}x{SRC_HEIGHT} NV12 @ 120fps")
    print(f"  裁切:     去掉上下{CROP_TOP}px → {OUT_WIDTH}x{OUT_HEIGHT} (16:9)")
    print(f"  输出:     {OUT_WIDTH}x{OUT_HEIGHT} (16:9) @ {OUTPUT_FPS}fps")
    print(f"  时长:     {record_seconds} 秒")
    print(f"  亮度:     {brightness:+0.1f}")
    print(f"  文件:     {output_path}")
    print(f"============================================")
    print()

    try:
        if backend == "ffmpeg":
            ok, log = record_ffmpeg(record_seconds, output_path, brightness)
        else:
            ok, log = record_gst(record_seconds, output_path, brightness)

        if ok:
            info = probe_video(output_path)
            if info:
                kb = os.path.getsize(output_path) / 1024
                print(f"  文件:     {output_path}")
                print(f"  大小:     {kb:.0f} KB")
                print(f"  时长:     {info['duration']:.2f}s")
                print(f"  分辨率:   {info['width']}x{info['height']} (16:9)")
                print(f"  帧率:     {info['fps']}")
                print(f"  码率:     {info['bitrate']} kb/s")
                print(f"  色彩:     {info['pix_fmt']} / {info['colorspace']}")
            return

        print(f"\n[错误] 录制失败")
        if log:
            lines = [l for l in log.split('\n')
                     if 'mpp' not in l.lower() and 'rga' not in l.lower()]
            relevant = '\n'.join(lines[-8:])
            if relevant.strip():
                print(relevant)
        sys.exit(1)

    except subprocess.TimeoutExpired:
        print("[错误] 录制超时")
        sys.exit(1)
    except FileNotFoundError as e:
        print(f"[错误] 缺少依赖: {e}")
        sys.exit(1)
    except Exception as e:
        print(f"[错误] 异常: {e}")
        sys.exit(1)
    finally:
        pass


if __name__ == "__main__":
    main()

效果：

踩坑清单

1. Rockchip V4L2 设备不设 CAPTURE 标志位，ffmpeg 直接拒，必须用 GStreamer v4l2src

2. mainpath vs selfpath 分辨率不同，mainpath 16:9 但被 rkaiq 独占，selfpath 8:9 需裁切

3. 停止 rkaiq = 曝光冻结，画面暗且提亮后偏色，正确做法是用 selfpath 不停止 rkaiq

4. NV12 直通优于转换，GStreamer videoconvert 在 pipe 场景下 NV12→I420 有损

5. GStreamer openh264enc 比 ffmpeg libx264 ultrafast 慢一倍（8fps vs 15fps）

6. videobalance 的 brightness 只调 Y 不动 UV，不会引起色偏（之前冤枉它了）

7. timeout 要用 -s INT 才能触发 GStreamer -e 的 EOS 机制正确封包