Opus Encoder (Run)

This article explains how you can use Transcoder.run to encode a WAV file to Opus audio stored in an OGG container.

The code snippets in this article are from the enc_opus_file sample.

Source Audio

For source we use the equinox-48KHz.wav file from the AVBlocks Assets repository. After downloading and unzipping you will find equinox-48KHz.wav in the aud subdirectory.

Code

This code takes a WAV file and encodes it to Opus audio in an OGG container.

Initialize AVBlocks

The first step in any AVBlocks application is to initialize the library. This must be done before using any other AVBlocks functionality. The Library.initialize() method sets up the internal state and loads necessary codecs. Always remember to call Library.shutdown() at the end of your program to properly clean up resources and release any allocated memory.

Library.initialize()

# Set license information. Without this AVBlocks runs in Demo mode.
# Library.set_license("<license-string>")

result = encode(input_file, output_file)

Library.shutdown()

sys.exit(0 if result else 2)

Configure Output Socket

The output socket defines where and how the encoded audio data will be written. In this case, we’re configuring it to output an Opus stream in an OGG container. The socket represents the output destination, while the pin represents the specific audio stream within that destination. The AudioStreamInfo object specifies the audio format parameters - you can customize the bitrate, sampling rate, or number of channels by uncommenting the relevant lines.

def create_output_socket(output_file: str) -> MediaSocket:
    """Create output socket for Opus file."""
    # create stream info to describe the output audio stream
    asi = AudioStreamInfo()
    asi.stream_type = StreamType.Opus

    # The default bitrate is 128000. You can set it to 192000, 256000, etc.
    # asi.bitrate = 192000

    # Optionally set the sampling rate and the number of the channels, e.g. 48 Khz, Stereo
    # asi.sample_rate = 48000
    # asi.channels = 2

    # create a pin using the stream info
    pin = MediaPin()
    pin.stream_info = asi

    # finally create a socket for the output container format which is OGG in this case
    socket = MediaSocket()
    socket.stream_type = StreamType.OGG
    socket.pins.add(pin)
    socket.file = output_file

    return socket

Configure Transcoder and Encode

This is the main encoding function that ties everything together. First, we create an input socket that points to the WAV file we want to encode. Then we create the output socket using our helper method. The transcoder is the core component that performs the actual encoding - it takes the uncompressed LPCM data from the input and converts it to compressed Opus data for the output.

The allow_demo_mode = True setting allows the transcoder to work even without a valid license (useful for testing, but not recommended for production). We then add our input and output sockets to the transcoder, open it to prepare for processing, run the actual transcoding operation, and finally close it to clean up.

def encode(input_file: str, output_file: str) -> bool:
    """Encode WAV file to Opus."""
    # Transcoder will fail if output exists (by design)
    delete_file(output_file)

    # Create output directory if it doesn't exist
    output_dir = os.path.dirname(output_file)
    if output_dir and not os.path.exists(output_dir):
        os.makedirs(output_dir)

    # Create input socket
    in_socket = MediaSocket()
    in_socket.file = input_file

    # Create output socket
    out_socket = create_output_socket(output_file)

    # Create transcoder
    with Transcoder() as transcoder:
        transcoder.allow_demo_mode = True
        transcoder.inputs.add(in_socket)
        transcoder.outputs.add(out_socket)

        if not transcoder.open():
            print_error("Transcoder open", transcoder.error)
            return False

        if not transcoder.run():
            print_error("Transcoder run", transcoder.error)
            return False

        transcoder.close()

    return True

Complete Python Code

Here’s the complete working example that demonstrates Opus encoding using AVBlocks. This code combines all the previous snippets into a functional program that can be run. The main function handles command-line argument parsing, initializes AVBlocks, performs the encoding operation, and properly shuts down the library before exiting.

"""
Encode WAV file to Opus.

This sample encodes a WAV audio file to Opus format using AVBlocks.
"""

import os
import sys
import click

from avblocks import (
    Library, Transcoder, MediaSocket, MediaPin,
    AudioStreamInfo, StreamType, ErrorFacility
)


def delete_file(filename: str):
    """Delete a file if it exists."""
    try:
        if os.path.exists(filename):
            os.remove(filename)
    except OSError:
        pass


def print_error(action: str, error) -> None:
    """Print error information."""
    if action:
        print(f"{action}: ", end="")

    if error.facility == ErrorFacility.Success:
        print("Success")
        return

    print(f"{error.message or ''}, facility:{error.facility} code:{error.code} hint:{error.hint or ''}")


def create_output_socket(output_file: str) -> MediaSocket:
    """Create output socket for Opus file."""
    # create stream info to describe the output audio stream
    asi = AudioStreamInfo()
    asi.stream_type = StreamType.Opus

    # The default bitrate is 128000. You can set it to 192000, 256000, etc.
    # asi.bitrate = 192000

    # Optionally set the sampling rate and the number of the channels, e.g. 48 Khz, Stereo
    # asi.sample_rate = 48000
    # asi.channels = 2

    # create a pin using the stream info
    pin = MediaPin()
    pin.stream_info = asi

    # finally create a socket for the output container format which is OGG in this case
    socket = MediaSocket()
    socket.stream_type = StreamType.OGG
    socket.pins.add(pin)
    socket.file = output_file

    return socket


def encode(input_file: str, output_file: str) -> bool:
    """Encode WAV file to Opus."""
    # Transcoder will fail if output exists (by design)
    delete_file(output_file)

    # Create output directory if it doesn't exist
    output_dir = os.path.dirname(output_file)
    if output_dir and not os.path.exists(output_dir):
        os.makedirs(output_dir)

    # Create input socket
    in_socket = MediaSocket()
    in_socket.file = input_file

    # Create output socket
    out_socket = create_output_socket(output_file)

    # Create transcoder
    with Transcoder() as transcoder:
        transcoder.allow_demo_mode = True
        transcoder.inputs.add(in_socket)
        transcoder.outputs.add(out_socket)

        if not transcoder.open():
            print_error("Transcoder open", transcoder.error)
            return False

        if not transcoder.run():
            print_error("Transcoder run", transcoder.error)
            return False

        transcoder.close()

    return True


@click.command()
@click.option('-i', '--input', 'input_file',
              help='Input WAV file',
              type=click.Path(exists=True))
@click.option('-o', '--output', 'output_file',
              help='Output Opus file',
              type=click.Path())
def main(input_file: str, output_file: str):
    """Encode WAV file to Opus."""

    # Set default options if not provided
    if not input_file or not output_file:
        script_dir = os.path.dirname(os.path.abspath(__file__))

        if not input_file:
            input_file = os.path.join(script_dir, "../../assets/aud/equinox-48KHz.wav")
        if not output_file:
            output_file = os.path.join(script_dir, "../../output/enc_opus_file/equinox-48KHz.opus")

        print("Using default options:")
        print(f"  --input {input_file}")
        print(f"  --output {output_file}")
        print()

    # Validate options
    print(f"Input file: {input_file}")
    print(f"Output file: {output_file}")

    Library.initialize()

    # Set license information. Without this AVBlocks runs in Demo mode.
    # Library.set_license("<license-string>")

    result = encode(input_file, output_file)

    Library.shutdown()

    sys.exit(0 if result else 2)


if __name__ == '__main__':
    main()

How to Run

See the Python sample for details. The samples are installed as command-line tools when you install the package from the repository root with uv pip install --editable . (see Installation).

Command Line

enc-opus-file --input <wav file> --output <opus file>

Examples

List options:

enc-opus-file --help
Usage: enc-opus-file [OPTIONS]

  Encode WAV file to Opus.

Options:
  -i, --input PATH   Input WAV file
  -o, --output PATH  Output Opus file
  --help             Show this message and exit.

Encode the input file ./assets/aud/equinox-48KHz.wav into output file ./output/enc_opus_file/equinox-48KHz.opus:

# Linux and macOS
mkdir -p ./output/enc_opus_file

enc-opus-file \
    --input ./assets/aud/equinox-48KHz.wav \
    --output ./output/enc_opus_file/equinox-48KHz.opus

# Windows
New-Item -ItemType Directory -Force -Path ./output/enc_opus_file

enc-opus-file `
    --input ./assets/aud/equinox-48KHz.wav `
    --output ./output/enc_opus_file/equinox-48KHz.opus