This commit is contained in:
Adeeb Shihadeh
2026-07-18 09:04:41 -07:00
parent 3f49e2d33c
commit f0841b827a
5 changed files with 315 additions and 79 deletions
+6 -6
View File
@@ -44,18 +44,18 @@ To actually display the stream, run `watch3` in separate terminal:
## compressed_vipc.py usage
```
$ python3 compressed_vipc.py -h
usage: compressed_vipc.py [-h] [--nvidia] [--cams CAMS] [--silent] addr
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
addr Address of comma three
options:
-h, --help show this help message and exit
--nvidia Use nvidia instead of ffmpeg
--cams CAMS Cameras to decode
--silent Suppress debug output
-h, --help show this help message and exit
--cams CAMS Cameras to decode
--server SERVER choose vipc server name
--silent Suppress debug output
```
+48 -53
View File
@@ -1,17 +1,15 @@
#!/usr/bin/env python3
import av
import av.video.format
import os
import sys
import argparse
import numpy as np
import multiprocessing
import time
import signal
from collections import deque
import openpilot.cereal.messaging as messaging
from msgq.visionipc import VisionIpcServer, VisionStreamType
from openpilot.tools.camerastream.ffmpeg_decoder import Decoder, FFmpegError
V4L2_BUF_FLAG_KEYFRAME = 8
@@ -25,23 +23,12 @@ ENCODE_SOCKETS = {
VisionStreamType.VISION_STREAM_WIDE_ROAD: "wideRoadEncodeData",
}
def decoder(addr, vipc_server, vst, nvidia, W, H, debug=False):
def decoder(addr, vipc_server, vst, W, H, debug=False):
sock_name = ENCODE_SOCKETS[vst]
if debug:
print(f"start decoder for {sock_name}, {W}x{H}")
if nvidia:
os.environ["NV_LOW_LATENCY"] = "3" # both bLowLatency and CUVID_PKT_ENDOFPICTURE
sys.path += os.environ["LD_LIBRARY_PATH"].split(":")
import PyNvCodec as nvc
nvDec = nvc.PyNvDecoder(W, H, nvc.PixelFormat.NV12, nvc.CudaVideoCodec.HEVC, 0)
cc1 = nvc.ColorspaceConversionContext(nvc.ColorSpace.BT_709, nvc.ColorRange.JPEG)
conv_yuv = nvc.PySurfaceConverter(W, H, nvc.PixelFormat.NV12, nvc.PixelFormat.YUV420, 0)
nvDwn_yuv = nvc.PySurfaceDownloader(W, H, nvc.PixelFormat.YUV420, 0)
img_yuv = np.ndarray((H*W//2*3), dtype=np.uint8)
else:
codec = av.CodecContext.create("hevc", "r")
codec = Decoder("hevc")
os.environ["ZMQ"] = "1"
messaging.reset_context()
@@ -50,13 +37,22 @@ def decoder(addr, vipc_server, vst, nvidia, W, H, debug=False):
last_idx = -1
seen_iframe = False
time_q = []
time_q = deque()
def resync():
nonlocal seen_iframe
codec.reset()
seen_iframe = False
time_q.clear()
while 1:
msgs = messaging.drain_sock(sock, wait_for_one=True)
for evt in msgs:
evta = getattr(evt, evt.which())
if debug and evta.idx.encodeId != 0 and evta.idx.encodeId != (last_idx+1):
print("DROP PACKET!")
if last_idx != -1 and evta.idx.encodeId != (last_idx + 1):
if debug:
print("DROP PACKET!")
resync()
last_idx = evta.idx.encodeId
if not seen_iframe and not (evta.idx.flags & V4L2_BUF_FLAG_KEYFRAME):
if debug:
@@ -67,48 +63,48 @@ def decoder(addr, vipc_server, vst, nvidia, W, H, debug=False):
frame_latency = ((evta.idx.timestampEof/1e9) - (evta.idx.timestampSof/1e9))*1000
process_latency = ((evt.logMonoTime/1e9) - (evta.idx.timestampEof/1e9))*1000
# put in header (first)
# put in header (first) — VPS/SPS/PPS only, no frame expected
if not seen_iframe:
if nvidia:
nvDec.DecodeSurfaceFromPacket(np.frombuffer(evta.header, dtype=np.uint8))
else:
codec.decode(av.packet.Packet(evta.header))
try:
codec.decode(evta.header)
except FFmpegError as e:
if debug:
print(f"HEADER ERROR: {e}")
resync()
continue
seen_iframe = True
if nvidia:
rawSurface = nvDec.DecodeSurfaceFromPacket(np.frombuffer(evta.data, dtype=np.uint8))
if rawSurface.Empty():
if debug:
print("DROP SURFACE")
continue
convSurface = conv_yuv.Execute(rawSurface, cc1)
nvDwn_yuv.DownloadSingleSurface(convSurface, img_yuv)
else:
frames = codec.decode(av.packet.Packet(evta.data))
if len(frames) == 0:
if debug:
print("DROP SURFACE")
continue
assert len(frames) == 1
img_yuv = frames[0].to_ndarray(format=av.video.format.VideoFormat('yuv420p')).flatten()
uv_offset = H*W
y = img_yuv[:uv_offset]
uv = img_yuv[uv_offset:].reshape(2, -1).ravel('F')
img_yuv = np.hstack((y, uv))
try:
img_yuv = codec.decode(evta.data)
except FFmpegError as e:
if debug:
print(f"DECODE ERROR: {e}")
resync()
continue
vipc_server.send(vst, img_yuv.data, cnt, int(time_q[0]*1e9), int(time.monotonic()*1e9))
if img_yuv is None:
if debug:
print("DROP SURFACE")
continue
if codec.width != W or codec.height != H:
if debug:
print(f"DECODE ERROR: decoded frame is {codec.width}x{codec.height}, expected {W}x{H}")
resync()
continue
frame_start_time = time_q.popleft()
vipc_server.send(vst, img_yuv.data, cnt, int(frame_start_time*1e9), int(time.monotonic()*1e9))
cnt += 1
pc_latency = (time.monotonic()-time_q[0])*1000
time_q = time_q[1:]
pc_latency = (time.monotonic()-frame_start_time)*1000
if debug:
print(f"{len(msgs):2d} {evta.idx.encodeId:4d} {evt.logMonoTime/1e9:.3f} {evta.idx.timestampEof/1e6:.3f} \
roll {frame_latency:6.2f} ms latency {process_latency:6.2f} ms + {network_latency:6.2f} ms + {pc_latency:6.2f} ms \
= {process_latency+network_latency+pc_latency:6.2f} ms", len(evta.data), sock_name)
class CompressedVipc:
def __init__(self, addr, vision_streams, server_name, nvidia=False, debug=False):
def __init__(self, addr, vision_streams, server_name, debug=False):
print("getting frame sizes")
os.environ["ZMQ"] = "1"
messaging.reset_context()
@@ -127,7 +123,7 @@ class CompressedVipc:
self.procs = []
for vst in vision_streams:
ed = sm[ENCODE_SOCKETS[vst]]
p = multiprocessing.Process(target=decoder, args=(addr, self.vipc_server, vst, nvidia, ed.width, ed.height, debug))
p = multiprocessing.Process(target=decoder, args=(addr, self.vipc_server, vst, ed.width, ed.height, debug))
p.start()
self.procs.append(p)
@@ -143,7 +139,6 @@ class CompressedVipc:
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Decode video streams and broadcast on VisionIPC")
parser.add_argument("addr", help="Address of comma three")
parser.add_argument("--nvidia", action="store_true", help="Use nvidia instead of ffmpeg")
parser.add_argument("--cams", default="0,1,2", help="Cameras to decode")
parser.add_argument("--server", default="camerad", help="choose vipc server name")
parser.add_argument("--silent", action="store_true", help="Suppress debug output")
@@ -156,7 +151,7 @@ if __name__ == "__main__":
]
vsts = [vision_streams[int(x)] for x in args.cams.split(",")]
cvipc = CompressedVipc(args.addr, vsts, args.server, args.nvidia, debug=(not args.silent))
cvipc = CompressedVipc(args.addr, vsts, args.server, debug=(not args.silent))
# register exit handler
signal.signal(signal.SIGINT, lambda sig, frame: cvipc.kill())
@@ -0,0 +1,261 @@
import ctypes
import errno
import os
import ffmpeg
import numpy as np
AV_INPUT_BUFFER_PADDING_SIZE = 64
AV_LOG_QUIET = -8
SWS_FAST_BILINEAR = 1
class FFmpegError(RuntimeError):
pass
class AVPacket(ctypes.Structure):
# Public prefix of AVPacket. Only data and size are modified here; the packet
# remains non-refcounted and points at Decoder._packet_buffer.
_fields_ = [
("buf", ctypes.c_void_p),
("pts", ctypes.c_int64),
("dts", ctypes.c_int64),
("data", ctypes.POINTER(ctypes.c_uint8)),
("size", ctypes.c_int),
]
class AVFrame(ctypes.Structure):
# Public prefix of AVFrame through format. Stable within a libavutil major
_fields_ = [
("data", ctypes.POINTER(ctypes.c_uint8) * 8),
("linesize", ctypes.c_int * 8),
("extended_data", ctypes.POINTER(ctypes.POINTER(ctypes.c_uint8))),
("width", ctypes.c_int),
("height", ctypes.c_int),
("nb_samples", ctypes.c_int),
("format", ctypes.c_int),
]
def _bind(fn, restype, *argtypes):
fn.restype = restype
fn.argtypes = list(argtypes)
return fn
def _load_libraries():
avutil = ctypes.CDLL(os.path.join(ffmpeg.LIB_DIR, "libavutil.so.59"), mode=ctypes.RTLD_GLOBAL)
avcodec = ctypes.CDLL(os.path.join(ffmpeg.LIB_DIR, "libavcodec.so.61"), mode=ctypes.RTLD_GLOBAL)
swscale = ctypes.CDLL(os.path.join(ffmpeg.LIB_DIR, "libswscale.so.8"), mode=ctypes.RTLD_GLOBAL)
c_int, c_char_p, c_void_p, c_size_t = ctypes.c_int, ctypes.c_char_p, ctypes.c_void_p, ctypes.c_size_t
c_uint8_p = ctypes.POINTER(ctypes.c_uint8)
c_void_p_p = ctypes.POINTER(c_void_p)
_bind(avutil.av_log_set_level, None, c_int)
_bind(avutil.av_opt_set, c_int, c_void_p, c_char_p, c_char_p, c_int)
_bind(avutil.av_strerror, c_int, c_int, c_char_p, c_size_t)
_bind(avutil.av_get_pix_fmt, c_int, c_char_p)
_bind(avcodec.avcodec_find_decoder_by_name, c_void_p, c_char_p)
_bind(avcodec.avcodec_alloc_context3, c_void_p, c_void_p)
_bind(avcodec.avcodec_open2, c_int, c_void_p, c_void_p, c_void_p)
_bind(avcodec.avcodec_free_context, None, c_void_p_p)
_bind(avcodec.avcodec_flush_buffers, None, c_void_p)
_bind(avcodec.avcodec_send_packet, c_int, c_void_p, ctypes.POINTER(AVPacket))
_bind(avcodec.avcodec_receive_frame, c_int, c_void_p, ctypes.POINTER(AVFrame))
_bind(avcodec.av_packet_alloc, ctypes.POINTER(AVPacket))
_bind(avcodec.av_packet_free, None, ctypes.POINTER(ctypes.POINTER(AVPacket)))
_bind(avcodec.av_frame_alloc, ctypes.POINTER(AVFrame))
_bind(avcodec.av_frame_free, None, ctypes.POINTER(ctypes.POINTER(AVFrame)))
_bind(avcodec.av_frame_unref, None, ctypes.POINTER(AVFrame))
_bind(swscale.sws_getCachedContext, c_void_p,
c_void_p, c_int, c_int, c_int, c_int, c_int, c_int, c_int, c_void_p, c_void_p, c_void_p)
_bind(swscale.sws_scale, c_int,
c_void_p, ctypes.POINTER(c_uint8_p), ctypes.POINTER(c_int),
c_int, c_int, ctypes.POINTER(c_uint8_p), ctypes.POINTER(c_int))
_bind(swscale.sws_freeContext, None, c_void_p)
avutil.av_log_set_level(AV_LOG_QUIET)
return avutil, avcodec, swscale
_avutil, _avcodec, _swscale = _load_libraries()
_DataArray = ctypes.POINTER(ctypes.c_uint8) * 4
_LinesizeArray = ctypes.c_int * 4
AV_PIX_FMT_NV12 = _avutil.av_get_pix_fmt(b"nv12")
assert AV_PIX_FMT_NV12 >= 0
def _error_string(code: int) -> str:
buf = ctypes.create_string_buffer(256)
if _avutil.av_strerror(code, buf, len(buf)) == 0:
return buf.value.decode(errors="replace")
return f"FFmpeg error {code}"
def _check(code: int, operation: str) -> None:
if code < 0:
raise FFmpegError(f"{operation}: {_error_string(code)}")
class Decoder:
def __init__(self, codec_name: str = "hevc"):
self.closed = True
self._sws_context = ctypes.c_void_p()
self._packet_buffer = bytearray()
self._packet_address = 0
self._packet_data = None
self._output = np.empty(0, dtype=np.uint8)
self._dst_data = _DataArray()
self._dst_linesize = _LinesizeArray()
self.width = 0
self.height = 0
self._src_format = -1
codec = _avcodec.avcodec_find_decoder_by_name(codec_name.encode())
if not codec:
raise FFmpegError(f"decoder not found: {codec_name}")
self._context = ctypes.c_void_p(_avcodec.avcodec_alloc_context3(codec))
if not self._context:
raise MemoryError("avcodec_alloc_context3 failed")
self._packet = _avcodec.av_packet_alloc()
if not self._packet:
_avcodec.avcodec_free_context(ctypes.byref(self._context))
raise MemoryError("av_packet_alloc failed")
self._frame = _avcodec.av_frame_alloc()
if not self._frame:
_avcodec.av_packet_free(ctypes.byref(self._packet))
_avcodec.avcodec_free_context(ctypes.byref(self._context))
raise MemoryError("av_frame_alloc failed")
try:
# Frame threading holds decoded frames to populate worker pipelines.
# Slice threads can reduce decode time without adding that frame queue;
# four was the latency minimum on the replay camera workload.
_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")
_check(_avcodec.avcodec_open2(self._context, codec, None), "open decoder")
except Exception:
_avcodec.av_frame_free(ctypes.byref(self._frame))
_avcodec.av_packet_free(ctypes.byref(self._packet))
_avcodec.avcodec_free_context(ctypes.byref(self._context))
raise
self.closed = False
def __enter__(self):
self._ensure_open()
return self
def __exit__(self, exc_type, exc_value, traceback):
self.close()
def _ensure_open(self) -> None:
if self.closed:
raise RuntimeError("decoder is closed")
def _prepare_packet(self, data) -> None:
size = len(data)
required = size + AV_INPUT_BUFFER_PADDING_SIZE
if len(self._packet_buffer) < required:
# Grow-only; the address stays valid until the next reallocation.
self._packet_buffer = bytearray(required)
self._packet_address = ctypes.addressof(ctypes.c_uint8.from_buffer(self._packet_buffer))
self._packet_data = ctypes.cast(self._packet_address, ctypes.POINTER(ctypes.c_uint8))
self._packet_buffer[:size] = data
ctypes.memset(self._packet_address + size, 0, AV_INPUT_BUFFER_PADDING_SIZE)
self._packet.contents.data = self._packet_data
self._packet.contents.size = size
def _prepare_output(self, frame: AVFrame) -> None:
width, height, src_format = frame.width, frame.height, frame.format
if width <= 0 or height <= 0 or width % 2 or height % 2:
raise FFmpegError(f"unsupported frame dimensions: {width}x{height}")
if (width, height, src_format) == (self.width, self.height, self._src_format):
return
sws_context = _swscale.sws_getCachedContext(
self._sws_context, width, height, src_format,
width, height, AV_PIX_FMT_NV12, SWS_FAST_BILINEAR,
None, None, None,
)
if not sws_context:
raise FFmpegError("sws_getCachedContext failed")
self._sws_context = ctypes.c_void_p(sws_context)
self.width, self.height = width, height
self._src_format = src_format
self._output = np.empty(width * height * 3 // 2, dtype=np.uint8)
output_address = self._output.ctypes.data
self._dst_data = _DataArray(
ctypes.cast(output_address, ctypes.POINTER(ctypes.c_uint8)),
ctypes.cast(output_address + width * height, ctypes.POINTER(ctypes.c_uint8)),
None,
None,
)
self._dst_linesize = _LinesizeArray(width, width, 0, 0)
def _receive(self) -> np.ndarray | None:
"""Return one NV12 frame, or None if the decoder needs more input.
The returned buffer is reused on the next successful decode; callers must
use or copy it before calling decode again.
"""
result = _avcodec.avcodec_receive_frame(self._context, self._frame)
if result == -errno.EAGAIN:
return None
_check(result, "receive decoded frame")
try:
frame = self._frame.contents
self._prepare_output(frame)
rows = _swscale.sws_scale(
self._sws_context, frame.data, frame.linesize, 0, frame.height,
self._dst_data, self._dst_linesize,
)
if rows != frame.height:
raise FFmpegError(f"convert decoded frame: produced {rows} of {frame.height} rows")
return self._output
finally:
_avcodec.av_frame_unref(self._frame)
def decode(self, data) -> np.ndarray | None:
self._ensure_open()
if len(data) == 0:
return None
self._prepare_packet(data)
result = _avcodec.avcodec_send_packet(self._context, self._packet)
# The packet buffer is ours, not FFmpeg's. Clear the borrowed pointer so
# packet teardown can never attempt to release it.
self._packet.contents.data = None
self._packet.contents.size = 0
_check(result, "send packet to decoder")
return self._receive()
def reset(self) -> None:
"""Discard decoder state after a stream discontinuity."""
self._ensure_open()
_avcodec.avcodec_flush_buffers(self._context)
def close(self) -> None:
if self.closed:
return
self.closed = True
_swscale.sws_freeContext(self._sws_context)
_avcodec.av_frame_free(ctypes.byref(self._frame))
_avcodec.av_packet_free(ctypes.byref(self._packet))
_avcodec.avcodec_free_context(ctypes.byref(self._context))
def __del__(self):
self.close()
-3
View File
@@ -37,9 +37,6 @@ dependencies = [
"comma-deps-git-lfs",
"comma-deps-gcc-arm-none-eabi",
# body / webrtcd
"av",
# logging
"pyzmq",
"sentry-sdk",
Generated
-17
View File
@@ -17,21 +17,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" },
]
[[package]]
name = "av"
version = "16.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/78/cd/3a83ffbc3cc25b39721d174487fb0d51a76582f4a1703f98e46170ce83d4/av-16.1.0.tar.gz", hash = "sha256:a094b4fd87a3721dacf02794d3d2c82b8d712c85b9534437e82a8a978c175ffd", size = 4285203, upload-time = "2026-01-11T07:31:33.772Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/9c/84/2535f55edcd426cebec02eb37b811b1b0c163f26b8d3f53b059e2ec32665/av-16.1.0-cp312-cp312-macosx_11_0_x86_64.whl", hash = "sha256:640f57b93f927fba8689f6966c956737ee95388a91bd0b8c8b5e0481f73513d6", size = 26945785, upload-time = "2026-01-09T20:18:34.486Z" },
{ url = "https://files.pythonhosted.org/packages/b6/17/ffb940c9e490bf42e86db4db1ff426ee1559cd355a69609ec1efe4d3a9eb/av-16.1.0-cp312-cp312-macosx_14_0_arm64.whl", hash = "sha256:ae3fb658eec00852ebd7412fdc141f17f3ddce8afee2d2e1cf366263ad2a3b35", size = 21481147, upload-time = "2026-01-09T20:18:36.716Z" },
{ url = "https://files.pythonhosted.org/packages/15/c1/e0d58003d2d83c3921887d5c8c9b8f5f7de9b58dc2194356a2656a45cfdc/av-16.1.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:27ee558d9c02a142eebcbe55578a6d817fedfde42ff5676275504e16d07a7f86", size = 39517197, upload-time = "2026-01-11T09:57:31.937Z" },
{ url = "https://files.pythonhosted.org/packages/32/77/787797b43475d1b90626af76f80bfb0c12cfec5e11eafcfc4151b8c80218/av-16.1.0-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:7ae547f6d5fa31763f73900d43901e8c5fa6367bb9a9840978d57b5a7ae14ed2", size = 41174337, upload-time = "2026-01-11T09:57:35.792Z" },
{ url = "https://files.pythonhosted.org/packages/8e/ac/d90df7f1e3b97fc5554cf45076df5045f1e0a6adf13899e10121229b826c/av-16.1.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:8cf065f9d438e1921dc31fc7aa045790b58aee71736897866420d80b5450f62a", size = 40817720, upload-time = "2026-01-11T09:57:39.039Z" },
{ url = "https://files.pythonhosted.org/packages/80/6f/13c3a35f9dbcebafd03fe0c4cbd075d71ac8968ec849a3cfce406c35a9d2/av-16.1.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:a345877a9d3cc0f08e2bc4ec163ee83176864b92587afb9d08dff50f37a9a829", size = 42267396, upload-time = "2026-01-11T09:57:42.115Z" },
{ url = "https://files.pythonhosted.org/packages/c8/b9/275df9607f7fb44317ccb1d4be74827185c0d410f52b6e2cd770fe209118/av-16.1.0-cp312-cp312-win_amd64.whl", hash = "sha256:f49243b1d27c91cd8c66fdba90a674e344eb8eb917264f36117bf2b6879118fd", size = 31752045, upload-time = "2026-01-11T09:57:45.106Z" },
]
[[package]]
name = "certifi"
version = "2026.6.17"
@@ -717,7 +702,6 @@ name = "openpilot"
version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "av" },
{ name = "cffi" },
{ name = "comma-deps-acados" },
{ name = "comma-deps-bootstrap-icons" },
@@ -794,7 +778,6 @@ standalone = [
[package.metadata]
requires-dist = [
{ name = "av" },
{ name = "cffi" },
{ name = "codespell", marker = "extra == 'testing'" },
{ name = "comma-deps-acados" },