openpilot 远程实时摄像头流：用 compressed_vipc.py 在 PC 上解码并显示设备三路相机画面

本篇指南讲解 openpilot 仓库中 openpilot/tools/camerastream 工具链的完整用法：如何先在设备端启动 camerad、encoderd 和 messaging 桥接三件套，再用 compressed_vipc.py 把设备上的 HEVC 压缩码流经网络拉取、在 PC 本地解码，并重新发布到 VisionIPC 供 watch3.py 实时显示。读完并照做后，你可以在自己的电脑上实时看到任意一台运行 openpilot 的设备上三路摄像头（窄角道路、广角道路、车内）的原始画面，并观察到每一帧从采集到显示各阶段的延迟分解——这对调试相机管线、验证模型输入画面、远程观察真车行为都非常有用。

整体数据链路：从设备相机到 PC 屏幕

理解这个工具组最快的方式是先看清帧数据在整条链路上的形态变化：

1. 设备上：camerad 打开三路硬件相机，把裸帧（NV12）写入名为 camerad 的 VisionIPC 服务器；
2. 设备上：encoderd 作为 camerad 的 VisionIPC 客户端订阅这些裸帧，用硬件编码器（Comma 硬件走 V4L2 硬件编码，见 encoderd.cc 中 __COMMA_HARDWARE__ 分支选择 V4LEncoder，否则回退 FfmpegEncoder）将其压成 HEVC 码流，连同帧索引一起发布为 cereal 消息；
3. 设备上：cereal/messaging/bridge 把设备进程间消息总线上的数据通过 ZMQ（TCP）转发出去，这是跨机器的关键一步；
4. PC 上：compressed_vipc.py 作为 ZMQ 订阅端连回设备，按相机逐帧解码 HEVC → NV12；
5. PC 上：解码后的 NV12 数据重新发布到 PC 本地的 VisionIPC 服务器（默认名为 camerad），watch3.py 作为客户端把三路画面画到同一窗口。

对应源码中的消息服务定义可以在 log.capnp 中看到：

narrowRoadEncodeData @86 :EncodeData;
cabinEncodeData @87 :EncodeData;
wideRoadEncodeData @88 :EncodeData;

三个 EncodeData 服务在 services.py 中均被标记为高优先级、队列容量 QueueSize.BIG，因为视频流数据量大、要求实时：

"narrowRoadEncodeData": (False, 20., None, QueueSize.BIG),
"cabinEncodeData": (False, 20., None, QueueSize.BIG),
"wideRoadEncodeData": (False, 20., None, QueueSize.BIG),

前置条件：设备与 PC 必须处于同一 openpilot 版本

文档特别强调：设备和你的 PC 必须运行同一个 openpilot commit。从源码看这是硬约束：

• 设备端 bridge 把每个消息服务名通过 FNV-1a 哈希映射到 TCP 端口（起点 8023），见 bridge_zmq.cc 的 get_port()；PC 端订阅时用同一份 services.h 计算同样的端口。如果两端 commit 不一致，服务定义（端口哈希、消息 schema）就可能对不上，连上也没有有效数据；
• EncodeData 消息里的 idx（encodeId、timestampSof/Eof、flags 等）与 compressed_vipc.py 里的解析逻辑强绑定，schema 不一致会导致解码端持续判定丢包。

设备端操作：启动三个进程

SSH 登录设备后，在三个独立终端分别执行：

cd /data/openpilot && ./openpilot/cereal/messaging/bridge

cd /data/openpilot/system/loggerd && ./encoderd

cd /data/openpilot/system/camerad && ./camerad

说明：

• bridge 无参数运行时走 msgq_to_zmq 方向，即把设备本地消息总线广播出去（bridge.cc 的 main()：argc > 2 才走反向的 zmq_to_msgq）；
• encoderd 负责把 VisionIPC 裸帧编码成 HEVC 并发布 *EncodeData 消息；它还会响应 LivestreamEncoderBitrate、LivestreamRequestKeyframe 两个 Params 参数来动态调整直播码率和请求关键帧（见 encoderd.cc 的 encoder_set_bitrate() / encoder_request_keyframe()）。

也可以把三进程合成一条命令一次性启动，按一次 Ctrl+C 即可全部停止：

(
  cd /data/openpilot
  ./openpilot/cereal/messaging/bridge &

  cd /data/openpilot/system/camerad/
  ./camerad &

  cd /data/openpilot/system/loggerd/
  ./encoderd &

  wait
) ; trap 'kill $(jobs -p)' SIGINT

PC 端操作：解码并重新发布到 VisionIPC

在 PC 上的 openpilot 检出目录中运行解码脚本：

cd ~/openpilot/tools/camerastream && ./compressed_vipc.py <ip>

<ip> 是设备的网络地址（文档示例中为设备的 SSH 主机名/IP，如 comma-ffffffff）。完整参数来自 compressed_vipc.py 的 argparse 定义：

$ python3 compressed_vipc.py -h
usage: compressed_vipc.py [-h] [--cams CAMS] [--server SERVER] [--silent] addr

Decode video streams and broadcast on VisionIPC

positional arguments:
  addr             Address of comma three

options:
  -h, --help       show this help message and exit
  --cams CAMS      Cameras to decode
  --server SERVER  choose vipc server name
  --silent         Suppress debug output

各参数在源码中的实际语义：

参数	源码位置	默认值	说明
addr	位置参数	必填	设备地址；每个解码子进程用它建立到设备的 ZMQ 订阅
--cams	args.cams.split(",")	"0,1,2"（即全部三路）	逗号分隔的流编号，映射到 VisionStreamType：0=NARROW_ROAD（窄角道路）、1=CABIN（车内）、2=WIDE_ROAD（广角道路），枚举定义见 visionipc.py
--server	VisionIpcServer(server_name)	"camerad"	PC 本地 VisionIPC 服务器名，watch3.py 按这个名字取流
--silent	debug=(not args.silent)	关闭	去掉每帧的丢包/重同步/延迟调试输出

文档给出的最小示例——只解码第一路（窄角）并在另一终端显示：

cd ~/openpilot/tools/camerastream && ./compressed_vipc.py comma-ffffffff --cams 0
cd ~/openpilot/selfdrive/ui/ && ./watch3.py

PC 端显示：watch3.py 三路同屏

watch3.py 是一个只有十几行的 Raylib 演示程序，把三路画面平铺到同一窗口（右上为窄角道路、左下为车内、右下为广角）：

road = CameraView("camerad", VisionStreamType.VISION_STREAM_NARROW_ROAD)
driver = CameraView("camerad", VisionStreamType.VISION_STREAM_CABIN)
wide = CameraView("camerad", VisionStreamType.VISION_STREAM_WIDE_ROAD)
for _ in gui_app.render():
  road.render(rl.Rectangle(gui_app.width // 4, 0, gui_app.width // 2, gui_app.height // 2))
  driver.render(rl.Rectangle(0, gui_app.height // 2, gui_app.width // 2, gui_app.height // 2))
  wide.render(rl.Rectangle(gui_app.width // 2, gui_app.height // 2, gui_app.width // 2, gui_app.height // 2))

（见 watch3.py。注意它硬编码连接名为 camerad 的 VisionIPC 服务器，所以 compressed_vipc.py --server 如果改名，watch3 也需要同步调整。）

解码流程深潜：compressed_vipc.py 的关键机制

compressed_vipc.py 的结构可以分三层来看（compressed_vipc.py）：

1. 先探测帧尺寸，再建 VisionIPC 缓冲。 CompressedVipc.__init__ 先用 SubMaster 订阅各 *EncodeData 服务并 update() 等待第一帧，只为读取 ed.width/ed.height；随后在 PC 本地 VisionIpcServer 上 create_buffers(vst, 4, W, H)（4 个环形缓冲）并启动监听，最后为每一路相机 fork 一个独立的解码子进程：

p = multiprocessing.Process(target=decoder, args=(addr, self.vipc_server, vst, ed.width, ed.height, debug))

2. 解码前必须先等到关键帧。 HEVC 解码需要 VPS/SPS/PPS 参数集加一个 I 帧才能开始。解码循环里：

if not seen_iframe and not (evta.idx.flags & V4L2_BUF_FLAG_KEYFRAME):
  continue  # waiting for iframe

其中 V4L2_BUF_FLAG_KEYFRAME = 8 来自 V4L2 编码器驱动的标志位定义。收到关键帧后先 codec.decode(evta.header) 把参数集喂给解码器，再解码后续帧。

3. 丢包自动重同步。 每帧消息带有 idx.encodeId，如果 encodeId 不连续（evta.idx.encodeId != last_idx + 1），调用 resync()：清空 FFmpeg 解码器缓冲（avcodec_flush_buffers）、丢弃时间戳队列、等待下一个关键帧重新起解。这一机制让网络偶发抖动不会导致画面永久花屏。

4. 每帧延迟分解输出。 默认（非 --silent）模式下每帧打印一行延迟统计：

roll {frame_latency:6.2f} ms latency {process_latency:6.2f} ms + {network_latency:6.2f} ms + {pc_latency:6.2f} ms = {total:6.2f} ms

四项分别是：设备上帧采集时间（timestampEof - timestampSof）、设备编码到发布耗时（logMonoTime - timestampEof）、网络传输耗时（PC 收到时刻减 unixTimestampNanos）、PC 端解码耗时。这个输出是评估"从路面上发生事件到你在 PC 上看到画面"端到端延迟的现成工具。

解码器实现：ctypes 直接封装 FFmpeg 库

解码器在 ffmpeg_decoder.py 中实现，不依赖 FFmpeg 的 Python 绑定，而是用 ctypes.CDLL 直接加载 libavutil.so.59、libavcodec.so.61、libswscale.so.8（因此依赖 PC 上安装的 FFmpeg 动态库版本与这里写死的 so 主版本号一致）。几个值得注意的解码参数，在 Decoder.__init__ 中设置：

_check(_avutil.av_opt_set(self._context, b"threads", b"4", 0), "set decoder threads")
_check(_avutil.av_opt_set(self._context, b"thread_type", b"slice", 0), "set decoder thread type")
_check(_avutil.av_opt_set(self._context, b"flags", b"+low_delay", 0), "set low-delay mode")

源码注释解释了取舍：4 个 slice 线程是该工作负载下延迟的最低点——帧级线程会在解码器内部多囤帧（增加延迟），而 slice 线程能在不加帧队列的前提下缩短单帧解码时间。解码输出经 sws_scale 用 SWS_FAST_BILINEAR 转成 NV12（紧凑 numpy 数组，尺寸 W*H*3/2），供 VisionIPC 直接发送。

端到端验证与排查清单

按下面的顺序可以快速定位"看不到画面"类问题：

1. 设备与 PC commit 不一致 → 先同步版本，这是文档明示的第一前提；
2. 三个设备端进程是否都在：bridge、encoderd、camerad 缺一不可，encoderd 依赖 camerad 的 VisionIPC 流存在（VisionIpcClient::getAvailableStreams("camerad") 拿不到流就不建编码线程）；
3. 看 compressed_vipc.py 的调试输出：持续打印 waiting for iframe 说明没收到关键帧（可检查设备端编码是否正常）；频繁 DROP PACKET! 说明网络丢包，解码器会自动重同步；HEADER ERROR / DECODE ERROR 则触发同样重同步；
4. 尺寸不匹配：若打印 decoded frame is WxH, expected WxH，通常是两端 commit 不一致导致编码器分辨率与订阅端探测值不符，同样会触发 resync；
5. 只想要低带宽调试：用 --cams 0 只拉窄角一路，能显著降低带宽占用。

小结

tools/camerastream 这一组工具串起了 openpilot 相机管线的"远端可观测"能力：设备端 camerad → encoderd → bridge 三件套把三路 HEVC 直播推上网络，PC 端 compressed_vipc.py 以每进程一路相机的方式解码并回发布到 VisionIPC，watch3.py 完成三屏同显。其源码中"关键帧门控 + encodeId 连续性检测 + 自动重同步 + 逐帧延迟分解"的解码策略，本身就是一份高质量的低延迟网络视频解码参考实现，相关文件见 compressed_vipc.py 与 ffmpeg_decoder.py。