---
sidebar:
  hidden: true
title: roboto.experimental.video.decoder
---
GOP-aware compressed-video stream decoding backed by PyAV.

Compressed-video topics store one encoded 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.

The decoder is codec-agnostic: a [`VideoCodec`](/docs/reference/python-sdk/roboto/experimental/video/codec#roboto.experimental.video.codec.VideoCodec) supplies the PyAV decoder name and keyframe predicate, so H.265/VP9/AV1 reuse this GOP machinery unchanged (see [`codec`](/docs/reference/python-sdk/roboto/experimental/video/codec)).

Requires the `roboto[video]` extra (PyAV; Pillow/numpy for the pixel accessors on [`DecodedVideoFrame`](/docs/reference/python-sdk/roboto/experimental/video/decoder#roboto.experimental.video.decoder.DecodedVideoFrame)).

## Module Contents

### DecodedVideoFrame

```python
class roboto.experimental.video.decoder.DecodedVideoFrame(
    log_time: int,
    frame: av.VideoFrame,
)
```

`from roboto.experimental.video import DecodedVideoFrame`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/experimental/video/decoder.py#L41-L95)

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`)

**Properties**

- **DecodedVideoFrame.height** (`int`): Frame height in pixels.
- **DecodedVideoFrame.is_keyframe** (`bool`): Whether this frame is a keyframe (clean decoder entry point).
- **DecodedVideoFrame.log_time** (`int`): Log time (nanoseconds since Unix epoch) of the message this frame came from.
- **DecodedVideoFrame.width** (`int`): Frame width in pixels.

#### DecodedVideoFrame.to_image()

```python
def to_image() -> PIL.Image.Image
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/experimental/video/decoder.py#L73-L83)

Convert the frame to a PIL RGB image.

**Returns**

- `PIL.Image.Image`: The frame's pixels as a `PIL.Image.Image` in RGB mode.

**Raises**

- `ImportError`: If Pillow is not installed (`roboto[video]`).

#### DecodedVideoFrame.to_ndarray()

```python
def to_ndarray() -> numpy.ndarray
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/experimental/video/decoder.py#L85-L95)

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

**Returns**

- `numpy.ndarray`: The frame's pixels as a numpy array in RGB channel order.

**Raises**

- `ImportError`: If numpy is not installed (`roboto[video]`).

### decode_h264_stream()

```python
def roboto.experimental.video.decoder.decode_h264_stream(
    encoded_frames: collections.abc.Iterable[tuple[int, bytes]],
) -> collections.abc.Generator[DecodedVideoFrame, None, None]
```

`from roboto.experimental.video import decode_h264_stream`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/experimental/video/decoder.py#L171-L195)

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

Thin wrapper over [`decode_stream()`](/reference/python-sdk/roboto/experimental/video/decoder#roboto.experimental.video.decoder.decode_stream) bound to the H.264 codec; see it for the GOP, corruption-recovery, and log-time-mapping semantics.

**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`](/reference/python-sdk/roboto/experimental/video/decoder#roboto.experimental.video.decoder.DecodedVideoFrame) per decodable input frame, carrying the source message's log time.

**Raises**

- `ImportError`: If PyAV is not installed (`roboto[video]`).

**Returns**

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

**Usage**

```python
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_stream()

```python
def roboto.experimental.video.decoder.decode_stream(
    encoded_frames: collections.abc.Iterable[tuple[int, bytes]],
    codec: roboto.experimental.video.codec.VideoCodec,
) -> collections.abc.Generator[DecodedVideoFrame, None, None]
```

`from roboto.experimental.video import decode_stream`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/experimental/video/decoder.py#L98-L168)

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.

**Parameters**

- **encoded_frames** (`collections.abc.Iterable[tuple[int, bytes]]`): `(log_time, data)` pairs in ascending log-time order, where `data` is one encoded frame in `codec`'s bitstream format.
- **codec** (`roboto.experimental.video.codec.VideoCodec`): The codec of every frame in `encoded_frames`; supplies the PyAV decoder name and the keyframe predicate.

**Yields**

- One [`DecodedVideoFrame`](/reference/python-sdk/roboto/experimental/video/decoder#roboto.experimental.video.decoder.DecodedVideoFrame) per decodable input frame, carrying the source message's log time.

**Raises**

- `ImportError`: If PyAV is not installed (`roboto[video]`).

**Returns**

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

**Usage**

```python
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")
```

### logger

```python
roboto.experimental.video.decoder.logger
```

`from roboto.experimental.video.decoder import logger`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/experimental/video/decoder.py#L38-L38)