roboto.experimental.video#

Video APIs in active refinement; see roboto.experimental for the stability contract.

Decodes compressed-video topic data (one Annex B-framed H.264 frame per MCAP message, as written by Roboto ingestion for foxglove.CompressedVideo and compatible schemas) into still frames on demand. Decoding requires the roboto[video] extra; the bitstream-inspection helpers in roboto.experimental.video.h264 are dependency-free.

Submodules#

Package Contents#

roboto.experimental.video.DEFAULT_KEYFRAME_LOOKBACK_NS: int = 10000000000#

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.

class roboto.experimental.video.DecodedVideoFrame(log_time, frame)#

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.

Parameters:

  • log_time (int)

  • frame (av.VideoFrame)

property height: int#

Frame height in pixels.

Return type:

int

property is_keyframe: bool#

Whether this frame is a keyframe (clean decoder entry point).

Return type:

bool

property log_time: int#

Log time (nanoseconds since Unix epoch) of the message this frame came from.

Return type:

int

to_image()#

Convert the frame to a PIL RGB image.

Returns:

The frame’s pixels as a PIL.Image.Image in RGB mode.

Raises:

ImportError – If Pillow is not installed (roboto[video]).

Return type:

PIL.Image.Image

to_ndarray()#

Convert the frame to an (height, width, 3) uint8 RGB array.

Returns:

The frame’s pixels as a numpy array in RGB channel order.

Raises:

ImportError – If numpy is not installed (roboto[video]).

Return type:

numpy.ndarray

property width: int#

Frame width in pixels.

Return type:

int

roboto.experimental.video.MessageRangeLoader#

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.

class roboto.experimental.video.NalUnit#

One NAL unit split out of an Annex B-framed H.264 frame.

data: bytes#

header byte included, start code excluded.

Type:

The unit’s bytes

type: int#

nal_unit_type from the unit’s header byte.

class roboto.experimental.video.NalUnitType#

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.

IDR_SLICE = 5#
NON_IDR_SLICE = 1#
PICTURE_PARAMETER_SET = 8#
SEI = 6#
SEQUENCE_PARAMETER_SET = 7#
roboto.experimental.video.decode_frames_in_range(load_messages, start_time, end_time, keyframe_lookback_ns=DEFAULT_KEYFRAME_LOOKBACK_NS)#

Decode the H.264 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.

Parameters:

  • load_messages (MessageRangeLoader) – Loads the topic’s (log_time, data) messages for a time range; called once for the requested range and, when the range starts mid-GOP, once more for the keyframe lookback before it.

  • start_time (int) – Start of the range in nanoseconds since Unix epoch (inclusive).

  • end_time (int) – End of the range in nanoseconds since Unix epoch; passed through to load_messages.

  • keyframe_lookback_ns (int) – Upper bound on how far before start_time to search for the keyframe that anchors the range’s leading delta frames.

Yields:

One DecodedVideoFrame per decodable frame with log_time >= start_time, in ascending log-time order.

Raises:

ImportError – If PyAV is not installed (roboto[video]).

Return type:

collections.abc.Generator[roboto.experimental.video.decoder.DecodedVideoFrame, None, None]

Examples

>>> 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()
roboto.experimental.video.decode_h264_stream(encoded_frames)#

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

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.

Parameters:

encoded_frames (collections.abc.Iterable[tuple[int, bytes]]) – (log_time, data) pairs in ascending log-time order, where data is one Annex B-framed H.264 frame.

Yields:

One DecodedVideoFrame per decodable input frame, carrying the source message’s log time.

Raises:

ImportError – If PyAV is not installed (roboto[video]).

Return type:

collections.abc.Generator[DecodedVideoFrame, None, None]

Examples

>>> 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")
roboto.experimental.video.find_nal_units(data)#

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

Parameters:

data (bytes) – One frame’s Annex B-framed bytes, using 3-byte or 4-byte start codes.

Returns:

The frame’s NAL units in bitstream order. Each unit’s data is a copy of the unit’s bytes (header included, start code excluded).

Return type:

list[NalUnit]

Examples

>>> 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]
roboto.experimental.video.is_keyframe(data)#

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.

Parameters:

data (bytes) – One frame’s Annex B-framed bytes.

Returns:

True when the frame contains an IDR slice.

Return type:

bool

On this page