MP3 Encoder (Push)

This article explains how you can use Transcoder.push to encode an MP3 audio stream from raw audio frames.

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

Source Audio

As audio input we use the equinox-48KHz.wav file from the AVBlocks Assets archive. After downloading and unzipping you will find equinox-48KHz.wav in the aud subdirectory.

Code

This code shows how you can encode raw uncompressed audio frames into an MP3 stream. Two Transcoder objects are used, one to read the raw LPCM frames from a file, and another to encode the raw frames to MP3 stream. The encoding is done via the Transcoder.push method.

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 Input Socket

The input socket defines the format of the raw LPCM audio frames that will be fed to the MP3 encoder. This socket describes uncompressed 16-bit stereo audio at 48 kHz sample rate. The socket represents the input source, while the pin represents the specific audio stream within that source. The AudioStreamInfo object specifies the audio format parameters that the encoder expects to receive.

def create_input_socket() -> MediaSocket:
    """Create input socket for raw PCM data."""
    socket = MediaSocket()
    socket.stream_type = StreamType.LPCM

    pin = MediaPin()
    asi = AudioStreamInfo()
    asi.stream_type = StreamType.LPCM
    asi.channels = 2
    asi.sample_rate = 48000
    asi.bits_per_sample = 16

    pin.stream_info = asi
    socket.pins.add(pin)

    return socket

Configure Output Socket

The output socket defines where and how the encoded MP3 data will be written. This method sets up the MP3 encoding parameters including the output file location, audio format, and encoding options. You can customize the bitrate, sample rate, and number of channels by uncommenting the relevant lines. The socket represents the output destination, while the pin represents the specific MP3 stream within that destination.

def create_output_socket(output_file: str) -> MediaSocket:
    """Create output socket for MP3 file."""
    socket = MediaSocket()
    socket.stream_type = StreamType.MPEG_Audio
    socket.stream_sub_type = StreamSubType.MPEG_Audio_Layer3
    socket.file = output_file

    pin = MediaPin()
    asi = AudioStreamInfo()
    asi.stream_type = StreamType.MPEG_Audio
    asi.stream_sub_type = StreamSubType.MPEG_Audio_Layer3

    # 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. 44.1 Khz, Mono
    # asi.sample_rate = 44100
    # asi.channels = 1

    pin.stream_info = asi
    socket.pins.add(pin)

    return socket

Configure WAV Reader

This method creates a transcoder that reads WAV files and outputs raw LPCM audio frames. In a real-world scenario, you might have a different audio source like an audio capture device or streaming input. The WAV reader transcoder converts the compressed or container-wrapped audio in the WAV file into raw PCM samples that can be fed to the MP3 encoder. The allow_demo_mode = True setting allows the transcoder to work without a valid license for testing purposes.

def create_wav_reader(input_file: str) -> Transcoder:
    """Create transcoder for reading WAV file and outputting raw PCM."""
    wav_reader = Transcoder()
    wav_reader.allow_demo_mode = True

    # Input socket - it will automatically detect stream info from the file
    wav_input_socket = MediaSocket()
    wav_input_socket.file = input_file
    wav_reader.inputs.add(wav_input_socket)

    # Output socket - we need LPCM stream type
    pcm_output_socket = MediaSocket()
    pcm_output_socket.stream_type = StreamType.LPCM

    pcm_pin = MediaPin()
    pcm_asi = AudioStreamInfo()
    pcm_asi.stream_type = StreamType.LPCM
    pcm_asi.channels = 2
    pcm_asi.sample_rate = 48000
    pcm_asi.bits_per_sample = 16
    pcm_pin.stream_info = pcm_asi
    pcm_output_socket.pins.add(pcm_pin)
    wav_reader.outputs.add(pcm_output_socket)

    return wav_reader

Configure Transcoder and Encode

This is the main encoding function that coordinates the entire MP3 encoding process. It creates both the WAV reader (to provide raw audio frames) and the MP3 encoder (to compress those frames). The function demonstrates the push-based encoding workflow where raw PCM samples are pulled from the WAV reader and pushed to the MP3 encoder. The encoding loop continues until all audio data has been processed, with proper error handling and end-of-stream detection.

def encode(input_file: str, output_file: str) -> bool:
    """Encode WAV file to MP3 using push mode."""
    # 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 WAV reader transcoder
    wav_reader = create_wav_reader(input_file)

    if not wav_reader.open():
        print_error("WAV Reader open", wav_reader.error)
        return False

    # Create encoder transcoder
    encoder = Transcoder()
    encoder.allow_demo_mode = True
    encoder.inputs.add(create_input_socket())
    encoder.outputs.add(create_output_socket(output_file))

    if not encoder.open():
        print_error("Encoder open", encoder.error)
        wav_reader.close()
        return False

    # Pull-push encoding loop
    pcm_sample = MediaSample()
    reader_eos = False

    while not reader_eos:
        # Pull PCM sample from WAV reader
        res, _ = wav_reader.pull(pcm_sample)

        if res:
            # Push PCM sample to encoder
            if not encoder.push(0, pcm_sample):
                print_error("Encoder push", encoder.error)
                wav_reader.close()
                encoder.close()
                return False
            continue

        # No more PCM data from reader
        error = wav_reader.error
        if error.facility == ErrorFacility.Codec and error.code == CodecError.EOS:
            # Flush encoder
            encoder.flush()
            reader_eos = True
            continue

        # Reader error
        print_error("WAV Reader pull", error)
        wav_reader.close()
        encoder.close()
        return False

    wav_reader.close()
    encoder.close()

    print("Encoding completed successfully.")
    return True

Complete Python Code

Here’s the complete working example that demonstrates MP3 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 MP3 using push mode.

This sample encodes a WAV audio file to MP3 format using the push mode.
It first decodes the WAV file to PCM samples using a "reader" transcoder,
then pushes those PCM samples to an "encoder" transcoder.
"""

import os
import sys
import click

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


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_input_socket() -> MediaSocket:
    """Create input socket for raw PCM data."""
    socket = MediaSocket()
    socket.stream_type = StreamType.LPCM

    pin = MediaPin()
    asi = AudioStreamInfo()
    asi.stream_type = StreamType.LPCM
    asi.channels = 2
    asi.sample_rate = 48000
    asi.bits_per_sample = 16

    pin.stream_info = asi
    socket.pins.add(pin)

    return socket


def create_output_socket(output_file: str) -> MediaSocket:
    """Create output socket for MP3 file."""
    socket = MediaSocket()
    socket.stream_type = StreamType.MPEG_Audio
    socket.stream_sub_type = StreamSubType.MPEG_Audio_Layer3
    socket.file = output_file

    pin = MediaPin()
    asi = AudioStreamInfo()
    asi.stream_type = StreamType.MPEG_Audio
    asi.stream_sub_type = StreamSubType.MPEG_Audio_Layer3

    # 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. 44.1 Khz, Mono
    # asi.sample_rate = 44100
    # asi.channels = 1

    pin.stream_info = asi
    socket.pins.add(pin)

    return socket


def create_wav_reader(input_file: str) -> Transcoder:
    """Create transcoder for reading WAV file and outputting raw PCM."""
    wav_reader = Transcoder()
    wav_reader.allow_demo_mode = True

    # Input socket - it will automatically detect stream info from the file
    wav_input_socket = MediaSocket()
    wav_input_socket.file = input_file
    wav_reader.inputs.add(wav_input_socket)

    # Output socket - we need LPCM stream type
    pcm_output_socket = MediaSocket()
    pcm_output_socket.stream_type = StreamType.LPCM

    pcm_pin = MediaPin()
    pcm_asi = AudioStreamInfo()
    pcm_asi.stream_type = StreamType.LPCM
    pcm_asi.channels = 2
    pcm_asi.sample_rate = 48000
    pcm_asi.bits_per_sample = 16
    pcm_pin.stream_info = pcm_asi
    pcm_output_socket.pins.add(pcm_pin)
    wav_reader.outputs.add(pcm_output_socket)

    return wav_reader


def encode(input_file: str, output_file: str) -> bool:
    """Encode WAV file to MP3 using push mode."""
    # 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 WAV reader transcoder
    wav_reader = create_wav_reader(input_file)

    if not wav_reader.open():
        print_error("WAV Reader open", wav_reader.error)
        return False

    # Create encoder transcoder
    encoder = Transcoder()
    encoder.allow_demo_mode = True
    encoder.inputs.add(create_input_socket())
    encoder.outputs.add(create_output_socket(output_file))

    if not encoder.open():
        print_error("Encoder open", encoder.error)
        wav_reader.close()
        return False

    # Pull-push encoding loop
    pcm_sample = MediaSample()
    reader_eos = False

    while not reader_eos:
        # Pull PCM sample from WAV reader
        res, _ = wav_reader.pull(pcm_sample)

        if res:
            # Push PCM sample to encoder
            if not encoder.push(0, pcm_sample):
                print_error("Encoder push", encoder.error)
                wav_reader.close()
                encoder.close()
                return False
            continue

        # No more PCM data from reader
        error = wav_reader.error
        if error.facility == ErrorFacility.Codec and error.code == CodecError.EOS:
            # Flush encoder
            encoder.flush()
            reader_eos = True
            continue

        # Reader error
        print_error("WAV Reader pull", error)
        wav_reader.close()
        encoder.close()
        return False

    wav_reader.close()
    encoder.close()

    print("Encoding completed successfully.")
    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 MP3 file',
              type=click.Path())
def main(input_file: str, output_file: str):
    """Encode WAV file to MP3 using push mode."""

    # 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/Hydrate-Kenny_Beltrey.wav")
        if not output_file:
            output_file = os.path.join(script_dir, "../../output/enc_mp3_push/Hydrate-Kenny_Beltrey.mp3")

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

    # Validate options
    print(f"--input: {input_file}")
    print(f"--output: {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-mp3-push --input <wav file> --output <mp3 file>

Examples

List options:

enc-mp3-push --help
Usage: enc-mp3-push [OPTIONS]

  Encode WAV file to MP3 using push mode.

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

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

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

enc-mp3-push \
    --input ./assets/aud/equinox-48KHz.wav \
    --output ./output/enc_mp3_push/equinox-48KHz.mp3

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

enc-mp3-push `
    --input ./assets/aud/equinox-48KHz.wav `
    --output ./output/enc_mp3_push/equinox-48KHz.mp3