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

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.

Source

Classes

KeyValuePair

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

Source

Attributes

• type: int
• value: int | bytes

OpusPacket

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.

Source

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

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.

Source

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

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

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

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

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

Source

Attributes

• rate
• channels

OpusDecoder.init

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

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

def reset(self) -> None

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

Constants

OPUS_FRAME_US

OPUS_FRAME_US = 20000

The frame length moq_media.opus.OpusEncoder emits, and the LOC mapping’s one frame per group.

Source

OPUS_SAMPLE_RATE

OPUS_SAMPLE_RATE = 48000

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

Source

MICROSECONDS_PER_SECOND

MICROSECONDS_PER_SECOND = 1000000

Source

OpusApplication

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

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

Source

Pcm

Pcm = bytes | bytearray | memoryview

Interleaved signed 16-bit little-endian samples.

Source