roboto.experimental.video

How far before a requested range to search for the keyframe that makes the range’s leading delta frames decodable. Bounds the cost of the backward walk on streams with pathological keyframe intervals.

A decoded video frame paired with the log time of its source message.

Pixel data stays in the decoder’s native representation until one of the accessors is called, so frames that a caller samples past are never converted.

DecodedVideoFrame.to_image()

DecodedVideoFrame.to_ndarray()

Loads a topic’s (log_time, data) messages for a (start_time, end_time) range.

Both times are nanoseconds since Unix epoch; messages must come back in ascending log-time order. Whether end_time is inclusive follows the underlying reader — frame emission is bounded by log-time filtering, not by the loader’s boundary semantics.

One NAL unit split out of Annex B-framed data.

Bases: enum.IntEnum

H.264 NAL unit types (ITU-T H.264 table 7-1) relevant to this module.

NAL headers can carry any 5-bit type value, so NalUnit.type is a plain int; compare against these members for the types that matter here.

The per-codec hooks the generic stream decoder depends on.

A codec is data plus one function: the PyAV decoder name and a predicate that recognizes an independently decodable keyframe from an encoded frame’s bytes. Everything else about decoding (GOP skipping, log-time remapping, error recovery) is codec-agnostic and lives in the decoder.

Decode the compressed-video frames of a time range.

Frames in the range that are undecodable — e.g. delta frames whose keyframe lies further back than keyframe_lookback_ns — are silently omitted; no player could render them either.

Usage

from roboto.experimental.video import decode_frames_in_range
frames = decode_frames_in_range(
    load_messages=my_loader, start_time=start_ns, end_time=end_ns
)
next(frames).to_image()

Decode an in-order stream of Annex B-framed H.264 frames.

Thin wrapper over decode_stream() bound to the H.264 codec; see it for the GOP, corruption-recovery, and log-time-mapping semantics.

Usage

from roboto.experimental.video import decode_h264_stream
for frame in decode_h264_stream(encoded_frames):
    frame.to_image().save(f"frame-{frame.log_time}.jpeg")

Decode an in-order stream of encoded frames of a single codec.

Frames before the first keyframe are skipped — without the preceding GOP state no decoder can render them. Individually corrupt frames are likewise skipped (with a warning) and decoding resumes at the next decodable frame. A fresh decoder session is created per call; concatenating unrelated streams into one call is only valid if each starts with a keyframe.

Usage

from roboto.experimental.video.codec import H264
from roboto.experimental.video import decode_stream
for frame in decode_stream(encoded_frames, H264):
    frame.to_image().save(f"frame-{frame.log_time}.jpeg")

Split Annex B-framed H.264 data into its NAL units.

Usage

from roboto.experimental.video import NalUnitType, find_nal_units
units = find_nal_units(annex_b_frame_bytes)
[unit.type for unit in units]
# [7, 8, 5]

Whether the Annex B-framed H.264 frame contains an IDR slice.

An IDR slice is a clean decoder entry point: decoding may start at this frame without any earlier GOP state. Frames without one are delta frames, decodable only after the preceding keyframe has been fed to the decoder.

Return the codec registered for a per-message format token.

The per-message format tokens that have a decode path, lowercase.