roboto.experimental.video.decoder#

GOP-aware H.264 stream decoding backed by PyAV.

Compressed-video topics store one Annex B-framed H.264 frame per message. Unlike still images, those frames are not independently decodable: a delta frame is only meaningful relative to the frames since the preceding keyframe (its group of pictures, “GOP”). The decoder here consumes an in-order stream of encoded frames, maintains that state, and yields decoded frames mapped back to their source messages’ log times.

Requires the roboto[video] extra (PyAV; Pillow/numpy for the pixel accessors on DecodedVideoFrame).

Module Contents#

class roboto.experimental.video.decoder.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.decoder.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.decoder.logger#
On this page