Rerun 视频流VideoStream查询与解码实战从 Catalog 服务器读取、随机访问帧到 MP4 导出【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerun本篇技术指南以 Rerun 的VideoStream视频流为核心讲解如何从 Catalog 服务器查询编码后的视频样本数据、解码指定帧、利用关键帧信息实现高效随机访问以及通过 remuxing 方式将整个流导出为 MP4 文件。读完本文你将掌握一套可直接运行的 Python 视频数据查询管线从启动本地 Catalog 服务器、按列过滤VideoStream:codec与VideoStream:sample到用 PyAV 精确解码任意目标帧并理解其中时间戳、关键帧与 B 帧约束等关键细节。为什么视频流查询需要特殊处理Rerun 支持多种图像/视频的日志方式它们在「简单、无损、体积大」到「复杂、有损、体积小」之间取舍详见 视频概念文档无压缩以大量Image记录压缩为 JPEG 等格式以多个EncodedImage记录压缩为单个视频文件以AssetVideo记录如 MP4压缩为一系列编码样本以VideoStream记录如 H.264 编码帧。VideoStream提供最好的压缩比适合相机画面这类高频数据但由于帧之间采用帧间压缩inter-frame compression大多数帧只保存与前一帧的差异因此从 Catalog 服务器查询回来之后不能像普通图像那样直接取出某一帧渲染而必须携带解码信息、从流的起点或最近的关键帧开始顺序解码。这正是本文要解决的「特殊处理」。需要说明的是本主题的示例代码目前仅在 Python SDK 中实现见 snippets.toml 中对howto/query_videos与howto/query_video_keyframes的标注C 与 Rust 均标记为 Not implemented。环境准备示例依赖两个核心包rerun-sdk[all]提供 Catalog 客户端、服务器与列式写入 APIavPyAV负责视频解码、关键帧检测与 MP4 封装。此外还需要numpy、pyarrow数据表操作与datafusion列式查询引擎Rerun 的reader基于其构建。完整的可运行示例位于 docs/snippets/all/howto/query_videos.py 与 docs/snippets/all/howto/query_video_keyframes.py下文所有代码片段均取自这两个文件。启动 Catalog 服务器并建立查询通道官方示例使用本地服务器演示完整流程实际生产场景中将CATALOG_URL替换为你的云实例地址即可。from fractions import Fraction from io import BytesIO from pathlib import Path import av import numpy as np import pyarrow as pa from datafusion import col import rerun as rr sample_video_path ( Path(__file__).parents[4] / tests / assets / rrd / video_sample ) server rr.server.Server(datasets{video_dataset: sample_video_path}) CATALOG_URL server.url() client rr.catalog.CatalogClient(CATALOG_URL) dataset client.get_dataset(namevideo_dataset) df dataset.filter_contents([/video_stream/**]).reader(indexlog_time) times pa.table(df.select(log_time))[log_time].to_numpy()这段代码的要点rr.server.Server(datasets{video_dataset: sample_video_path})把本地 RRD 文件注册为一个名为video_dataset的数据集并启动服务rr.catalog.CatalogClient(CATALOG_URL)创建 Catalog 客户端之后所有查询都通过它完成dataset.filter_contents([/video_stream/**])按实体路径前缀过滤只保留视频流相关的数据.reader(indexlog_time)返回一个以log_time为索引的列式查询器DataFrame后续的select、filter、aggregate都基于它执行times是全部样本时间戳的 numpy 数组后续随机访问帧时用它定位目标时刻。理解 VideoStream 数据结构VideoStream原型见 video_stream.md包含以下字段字段类型说明codec必填VideoCodec视频编码格式如 H.264sample推荐VideoSample编码后的视频样本帧数据is_keyframe可选IsKeyframe该样本是否为关键帧opacity/draw_order可选—渲染相关属性除sample外其余组件通常在实体上静态记录一次sample则随每条时间线逐帧反复记录。查询时你会打交道的关键列是/video_stream:VideoStream:codec—— 视频编码格式/video_stream:VideoStream:sample—— 编码后的视频帧数据H.264 采用 Annex B 格式。列名遵循/{entity_path}:{archetype}:{component}的约定后续查询与关键帧图层中的is_keyframe列也遵循同样的命名模式。VideoCodec组件定义了以下取值枚举值为对应 WebCodec 字符串的 fourcc 大端表示详见 video_codec.md枚举值样本格式要求AV10x61763031Low overhead bitstream 格式OBU 序列关键帧样本需在KEY_FRAMEOBU 前包含 sequence header OBUH2640x61766331Annex B 规范注意与 MP4 中的 AVCC 格式不同关键帧IDR需包含 SPSH2650x68657631Annex B 规范关键帧IRAP需包含 SPSVP80x76703038—VP90x76703039—当前仓库的 viewer 对编解码器的支持情况浏览器 vs 原生详见 视频概念文档其中对播放器而言推荐 AV1而编码耗时敏感时推荐 H.264/avc。校验视频编码格式在处理视频数据之前先确认 codec 与你的预期一致避免后续用错误的解码器解析codec_column /video_stream:VideoStream:codec num_codec_matches df.select( col(codec_column)[0] rr.VideoCodec.H264.value ).count() if num_codec_matches ! df.select(codec_column).count(): raise ValueError( fExpected H.264 codec {rr.VideoCodec.H264.value}, fgot {df.select(codec_column).limit(1)} )这里利用datafusion的col()构造列表达式统计「codec 列首元素等于rr.VideoCodec.H264.value」的匹配行数并与 codec 列的总行数比较若不等则说明存在非 H.264 样本并抛出异常。这样可以在进入解码环节前快速兜底。解码指定帧从流起点顺序解码由于帧间压缩要解码某一目标帧必须从流的起点或最近的关键帧开始逐包解码并向前迭代av会在解码过程中内部处理关键帧检测video_column /video_stream:VideoStream:sample selected_frame_index 3 # Pick an arbitrary frame to decode # Query all samples up to and including the target frame. # We need to decode from the start (or a keyframe) to reach our target. selected_time times[selected_frame_index] video_df df.filter(col(log_time) selected_time).select( log_time, video_column ) pa_table pa.table(video_df) # Concatenate samples into a byte buffer samples pa_table[video_column].to_numpy() sample_times pa_table[log_time].to_numpy() sample_bytes b for sample in samples: sample_bytes sample[0].tobytes() data_buffer BytesIO(sample_bytes) # Decode using PyAV container av.open(data_buffer, formath264, moder) video_stream: av.video.stream.VideoStream container.streams.video[0] start_time sample_times[0] # Decode all frames up to our target, keeping only the last one frame None for packet, time in zip( container.demux(video_stream), sample_times, strictFalse ): packet.time_base Fraction(1, 1_000_000_000) # Timestamps in nanoseconds packet.pts int(time - start_time) packet.dts packet.pts # No B-frames, so dts pts for decoded_frame in packet.decode(): frame decoded_frame if not isinstance(frame, av.VideoFrame): raise RuntimeError(Failed to decode frame.) image np.asarray(frame.to_image()) print(fDecoded frame shape: {image.shape})关键步骤拆解查询范围df.filter(col(log_time) selected_time)取到目标帧为止的全部样本保证解码依赖链完整拼接字节流把所有样本sample[0].tobytes()拼接成单个 Annex B 字节流交给av.open(data_buffer, formath264, moder)对齐时间戳packet.time_base Fraction(1, 1_000_000_000)纳秒packet.pts int(time - start_time)把 Rerun 的纳秒时间戳转换为相对流起点的 PTS逐包解码container.demux(video_stream)与sample_times一一对应strictFalse容忍长度不完全匹配每包解码出的一或多帧中只保留最后一帧循环结束后frame即目标帧转图像frame.to_image()得到 PIL 图像np.asarray(...)转 numpy 数组供后续处理。注意此处设置了packet.dts packet.pts其前提是流中不存在 B 帧详见下文「B 帧限制」一节。基于关键帧信息的高效随机访问上面的例子从流的起点查询全部样本对长视频而言既不必要也低效。更优的做法是把关键帧信息作为数据集的附加图层layer注册查询时只取「最近关键帧 → 目标帧」之间的样本大幅减少需要拉取和解码的数据量。本节的完整代码见 docs/snippets/all/howto/query_video_keyframes.py。将关键帧信息注册为图层这是一个预处理步骤把视频整体解码一次用packet.is_keyframe找出关键帧时间戳写成稀疏数据并注册为独立图层# Preprocessing step: Add keyframe information to existing video data as a layer # This is typically done once to make subsequent queries faster # Query all video samples from the existing recording video_samples_df df.select(log_time, video_column) video_table pa.table(video_samples_df) sample_times video_table[log_time].to_numpy() samples video_table[video_column].to_numpy() # Concatenate all samples to analyze keyframes sample_bytes b for sample in samples: sample_bytes sample[0].tobytes() # Decode the video to detect keyframes data_buffer BytesIO(sample_bytes) container av.open(data_buffer, formath264, moder) video_stream container.streams.video[0] # Identify which samples are keyframes keyframe_times [] for packet, ts in zip(container.demux(video_stream), sample_times): if packet.is_keyframe: keyframe_times.append(ts) container.close() keyframe_values [True] * len(keyframe_times) print(fFound {len(keyframe_times)} keyframes) # Save keyframe data as a separate layer # Get the segment ID to align with the original recording segment_ids dataset.segment_ids() first_segment_id segment_ids[0] # Create time column and content using the columnar API # Make sure the timeline matches the original video stream timeline log_time time_column rr.TimeColumn(timelinetimeline, timestampkeyframe_times) content rr.DynamicArchetype.columns( archetypeKeyframeData, components{is_keyframe: keyframe_values} ) # Write to a new file as a layer layer_path TMP_DIR / keyframe_layer.rrd with rr.RecordingStream( application_idkeyframes, recording_idfirst_segment_id, # Match original recording_id ) as rec: rec.save(layer_path) rec.send_columns(/video_stream, indexes[time_column], columns[*content]) # Register the layer with the dataset dataset.register([layer_path.as_uri()], layer_namekeyframes) print(fRegistered keyframe layer at {layer_path})该预处理流程的特点解码一次用packet.is_keyframe判定每个 packet 是否为关键帧只记录关键帧的时间戳稀疏数据keyframe_values [True] * len(keyframe_times)仅在关键帧时间点有值普通帧位置不存在该列数据查询时表现为is_keyframe列为空写入独立 RRD通过rr.RecordingStream打开一个recording_id与原始录制相同的录制流rec.save(layer_path)落盘再用rec.send_columns(/video_stream, indexes[time_column], columns[*content])以列式 API 写入KeyframeData原型动态原型 is_keyframe组件注册为图层dataset.register([layer_path.as_uri()], layer_namekeyframes)将新文件注册为数据集上的keyframes图层。这里涉及 Rerun Catalog 的对象模型数据集由若干段segments组成每段按图层layers组织默认图层名为base把recording_id即 segment ID相同的.rrd用不同的 layer_name注册即可实现追加详见 catalog-object-model.md。注册成功后图层数据会作为额外列出现在查询结果中——本例即/video_stream:is_keyframe。利用关键帧图层进行高效查询有了关键帧图层就可以先找到「不晚于目标时刻的最近关键帧」再只查询该关键帧到目标帧之间的样本# Query using keyframe information for efficient random access # Assume weve already added keyframe information via the preprocessing step # above target_frame_index 42 target_time times[target_frame_index] # Create a reader that includes the keyframe layer data # The column name follows the pattern: /{entity_path}:{component_name} keyframe_column /video_stream:is_keyframe full_df dataset.filter_contents([/video_stream/**]).reader(indexlog_time) # Query to find the most recent keyframe at or before the target time. # Since we only log when is_keyframeTrue, any row with this column present # is a keyframe keyframe_slice full_df.filter( (col(log_time) target_time) col(keyframe_column).is_not_null() ) closest_keyframe_df keyframe_slice.aggregate( [], [ F.last_value(col(log_time), order_by[col(log_time)]).alias( latest_keyframe ) ], ) keyframe_result pa.table(closest_keyframe_df) # Start decoding from the most recent keyframe start_time keyframe_result[latest_keyframe].to_numpy()[0] start_frame_idx np.searchsorted(times, start_time) frames_saved target_frame_index - start_frame_idx print( fFound keyframe at frame {start_frame_idx}, fsaved decoding {frames_saved} frames ) # Query only the video samples from keyframe to target (much more efficient!) efficient_video_df df.filter( col(log_time).between(start_time, target_time) ).select(log_time, video_column) efficient_table pa.table(efficient_video_df) frames_to_decode len(efficient_table) print( fDecoding {frames_to_decode} frames f(vs {target_frame_index 1} without keyframe info) ) # Now decode just this smaller range samples efficient_table[video_column].to_numpy() sample_times efficient_table[log_time].to_numpy() sample_bytes b for sample in samples: sample_bytes sample[0].tobytes() data_buffer BytesIO(sample_bytes) container av.open(data_buffer, formath264, moder) video_stream container.streams.video[0] # Decode to the target frame frame None for packet, time in zip( container.demux(video_stream), sample_times, strictFalse ): packet.time_base Fraction(1, 1_000_000_000) packet.pts int(time - sample_times[0]) packet.dts packet.pts for decoded_frame in packet.decode(): frame decoded_frame if isinstance(frame, av.VideoFrame): image np.asarray(frame.to_image()) print( fEfficiently decoded frame {target_frame_index} fwith shape: {image.shape} )查询逻辑要点因为只在关键帧时间点写了is_keyframe所以col(keyframe_column).is_not_null()过滤后留下的每一行都是关键帧F.last_value(col(log_time), order_by[col(log_time)])取「不晚于目标时刻」的最近关键帧时间需要从datafusion.functions导入Fnp.searchsorted(times, start_time)把关键帧时间定位到帧序号从而计算省去了多少帧的解码col(log_time).between(start_time, target_time)只拉取关键帧到目标帧之间的样本解码字节流与目标帧数量大幅缩小后续解码流程与「从起点解码」完全一致只是起点从流的头部变成了最近关键帧。该方案在以下场景收益尤其明显长视频序列——从起点顺序解码代价高昂随机访问模式——需要跳到任意帧高分辨率视频——带宽与解码时间是显著瓶颈交互式应用——需要 seek 到特定时间戳。导出为 MP4remuxing视频流数据可以不重新编码地导出为 MP4 文件即所谓的 remuxing把编码样本原样重新封装进容器格式# Query all video samples video_df df.select(log_time, /video_stream:VideoStream:sample) pa_table pa.table(video_df) all_times pa_table[log_time] all_samples pa_table[/video_stream:VideoStream:sample] # Concatenate samples into a single byte buffer sample_bytes np.concatenate([ sample[0] for sample in all_samples.to_numpy() ]).tobytes() sample_bytes_io BytesIO(sample_bytes) # Setup input container (H.264 Annex B stream) input_container av.open(sample_bytes_io, moder, formath264) input_stream input_container.streams.video[0] # Setup output container (MP4) output_path TMP_DIR / output.mp4 output_container av.open(output_path, modew) output_stream output_container.add_stream_from_template(input_stream) # Remux packets with correct timestamps start_time all_times.chunk(0)[0] for packet, time in zip( input_container.demux(input_stream), all_times, strictFalse ): packet.time_base Fraction(1, 1_000_000_000) packet.pts int(time.value - start_time.value) packet.dts packet.pts packet.stream output_stream output_container.mux(packet) input_container.close() output_container.close() print(fExported video to {output_path})与解码示例相比这里不再调用packet.decode()而是用np.concatenate(...).tobytes()一次性拼接全部样本all_times、all_samples是 Arrow chunked 数组所以取首元素时用.chunk(0)[0]/.value以formath264打开输入容器读取其视频流模板output_container.add_stream_from_template(input_stream)复用输入流的编码参数codec、分辨率、时间基等创建输出流从而避免重新编码逐包修正time_base、pts、dts后将packet.stream指向输出流并mux写入 MP4 容器。由于输出文件写入临时目录TMP_DIR脚本退出时通过atexit自动清理。重要注意事项关键帧处理视频流普遍使用帧间压缩大多数帧只存储与前一帧的差异因此解码任意帧必须从流起点或最近的关键帧开始av内部会自动处理关键帧检测但若要高效随机访问建议在录制时而不是查询时就单独记录关键帧指示信息若选择在查询侧做预处理则像上文那样把关键帧信息写成独立图层只需解码一次即可被后续所有查询复用。时间戳处理Rerun 中的视频时间戳通常以纳秒为单位存储使用 PyAV 解码或封装时务必设置正确的time_base通常为Fraction(1, 1_000_000_000)packet.pts应转换为相对流起点的偏移int(time - start_time)以保证解码器与容器封装的时序正确。B 帧限制当前 Rerun 的VideoStream不支持 B 帧因此dts解码时间戳恒等于pts显示时间戳示例代码中直接令packet.dts packet.pts是安全且正确的该限制同样记录在 视频概念文档 的当前限制列表中B 帧支持问题跟踪中。进一步阅读视频概念与编解码器支持Image/EncodedImage/AssetVideo/VideoStream的取舍、浏览器与原生 viewer 的 codec 支持矩阵VideoStream 原型参考字段定义与 API 链接VideoCodec 组件参考各编解码器枚举值与样本格式要求Catalog 对象模型数据集、段segment与图层layer的注册与追加机制查询图像与视频流查询配套的图像查询指南完整可运行示例query_videos.py 与 query_video_keyframes.py。【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerun创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考