# moq_media.opus

> Python API reference for moq_media.opus: The Opus encoder, behind the opus extra.

URL: https://www.caprid.com/docs/reference/python/moq_media-opus/

From the `caprid-moq-media[opus]` package: `import moq_media.opus`

Opus encoding and decoding on PyAV's libopus, for a server-side talker such as a voice bot.

Install with the `opus` extra (`caprid-moq-media[opus]`). This is the one module in the package that
needs PyAV; `import moq_media` does not load it.

## Functions

### loc_properties

```python
def loc_properties(
    timestamp_us: int,
    *,
    timescale: int = MICROSECONDS_PER_SECOND,
    video_config: bytes | None = None,
    audio_config: bytes | None = None,
) -> list[KeyValuePair]
```

Package a microsecond timestamp; pass decoder configuration only on keyframes.

Other timescales round to the nearest tick, with ties rounded upward.

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-media/src/moq_media/loc.py#L40)</small>

## Classes

### KeyValuePair

```python
class KeyValuePair
```

A generic MOQT Key-Value-Pair.

Even types carry a varint value, odd types a byte field. Used for Setup Options and Track/Object
Properties (message parameters use per-type encodings instead; see [`MessageParameters`](../caprid_relay/#messageparameters)).

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-core/src/moq_core/control_messages.py#L374)</small>

**Attributes**

- **`type`**: `int`
- **`value`**: `int | bytes`

### OpusPacket

```python
class OpusPacket
```

One encoded Opus packet and where it starts on the broadcast timeline.

It has the attributes `caprid_relay.media.FrameWriter` takes, so it can be written as a frame as it is.

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-media/src/moq_media/opus.py#L34)</small>

**Attributes**

- **`timestamp_us`**: `int`
- **`payload`**: `bytes`
- **`duration_us`**: `int`
- **`key`**: `bool` — Every Opus packet decodes on its own: one group per frame (LOC).
- **`discontinuity`**: `bool`
- **`properties`**: `list[KeyValuePair]` *(property)* — The LOC object properties for the packet: its timestamp in microseconds.

### OpusEncoder

```python
class OpusEncoder
```

Encodes PCM into 20 ms Opus packets stamped on the broadcast timeline, with voice defaults.

Input is a run of audio: the first `encode` call after construction or `flush` gives the
timeline timestamp of its first sample, and every later call continues from where the samples so far
end. `flush` ends the run, the end of an utterance, padding its last partial frame with silence.

Packet *k* of a run is stamped at the run's timestamp plus 20 ms per packet before it, as WebCodecs stamps its
output. Decoded audio trails those stamps by the encoder's look-ahead (6.5 ms, or 2.5 ms for `lowdelay`),
and a flushed run ends with one packet more than the frames it was given, which carries that look-ahead's tail.

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-media/src/moq_media/opus.py#L54)</small>

**Attributes**

- **`input_rate`**
- **`input_channels`**
- **`channels`**
- **`bitrate`**
- **`application`**: `OpusApplication`
- **`next_timestamp_us`**: `int | None` *(property)* — Where the next sample handed in would land on the timeline, or `None` between runs.

#### OpusEncoder.__init__

```python
def __init__(
    self,
    *,
    input_rate: int = OPUS_SAMPLE_RATE,
    input_channels: int = 1,
    channels: int = 1,
    bitrate: int | None = 32000,
    application: OpusApplication = 'voip',
) -> None
```

Make an encoder for s16 PCM at `input_rate` with `input_channels`, encoding `channels`.

`bitrate` is in bits per second, or `None` for libopus's own choice.

#### OpusEncoder.encode

```python
def encode(
    self,
    pcm: Pcm | av.AudioFrame,
    *,
    timestamp_us: int | None = None,
) -> list[OpusPacket]
```

Encode audio and return the packets it completed, which may be none.

`pcm` is interleaved signed 16-bit little-endian samples at the encoder's input rate and channel
count, or a PyAV `AudioFrame` of any format, rate and layout, the same throughout a run.
`timestamp_us` is required to start a run and refused within one.

#### OpusEncoder.flush

```python
def flush(self) -> list[OpusPacket]
```

End the run: pad its last partial frame with silence and return every packet left.

The next `encode` starts a new run and needs a timestamp again. Flushing between runs
returns nothing.

### OpusDecoder

```python
class OpusDecoder
```

Decodes Opus packets into PCM at the rate and channel count the caller wants, such as 16 kHz mono for STT.

The decoder keeps state from packet to packet. Call `reset` wherever the packets stop following
one another: a gap, a resync, or a restart (a frame's `discontinuity`).

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-media/src/moq_media/opus.py#L213)</small>

**Attributes**

- **`rate`**
- **`channels`**

#### OpusDecoder.__init__

```python
def __init__(self, *, rate: int = OPUS_SAMPLE_RATE, channels: int = 1) -> None
```

Make a decoder whose output is interleaved s16 PCM at `rate` with `channels`.

#### OpusDecoder.decode

```python
def decode(self, payload: bytes | bytearray | memoryview) -> bytes
```

Decode one packet into interleaved signed 16-bit little-endian PCM.

The output's length can vary from packet to packet while resampling, and the first packets after a
reset can return less, or nothing, while the resampler fills.

#### OpusDecoder.reset

```python
def reset(self) -> None
```

Forget the previous packets, for a discontinuity. Audio still in the resampler is dropped.

## Constants

### OPUS_FRAME_US

```python
OPUS_FRAME_US = 20000
```

The frame length [`moq_media.opus.OpusEncoder`](#opusencoder) emits, and the LOC mapping's one frame per group.

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-media/src/moq_media/audio.py#L6)</small>

### OPUS_SAMPLE_RATE

```python
OPUS_SAMPLE_RATE = 48000
```

The rate every Opus stream decodes at (RFC 7845 §5.1), whatever rate the encoder was fed.

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-media/src/moq_media/audio.py#L3)</small>

### MICROSECONDS_PER_SECOND

```python
MICROSECONDS_PER_SECOND = 1000000
```

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-media/src/moq_media/loc.py#L11)</small>

### OpusApplication

```python
OpusApplication = Literal['voip', 'audio', 'lowdelay']
```

libopus's tuning: `voip` favours speech intelligibility, `audio` fidelity, `lowdelay` the shortest delay.

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-media/src/moq_media/opus.py#L23)</small>

### Pcm

```python
Pcm = bytes | bytearray | memoryview
```

Interleaved signed 16-bit little-endian samples.

<small>[Source](https://github.com/caprids/relay-sdks/blob/main/python/packages/moq-media/src/moq_media/opus.py#L30)</small>