AAC ADTS Encoder (Pull)

This article explains how you can use Transcoder.pull to encode a WAV file to AAC (Advanced Audio Coding) ADTS (Audio Data Transport Stream) elementary stream.

The code snippets in this article are from the enc_aac_adts_pull 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 compressed AAC ADTS format using the pull method, which allows you to retrieve encoded samples one at a time.

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 the format of the encoded audio data. Unlike the run method, when using pull we don’t specify an output file in the socket. Instead, we configure only the stream format and parameters. The socket represents the output format configuration, while the pin represents the specific audio stream. The AudioStreamInfo object specifies the audio format parameters - you can customize the number of channels or sampling rate by uncommenting the relevant lines.

def create_output_socket() -> MediaSocket:
    """Create output socket for AAC ADTS encoding (no file)."""
    socket = MediaSocket()
    socket.stream_type = StreamType.AAC
    socket.stream_sub_type = StreamSubType.AAC_ADTS

    pin = MediaPin()
    socket.pins.add(pin)

    asi = AudioStreamInfo()
    pin.stream_info = asi

    asi.stream_type = StreamType.AAC
    asi.stream_sub_type = StreamSubType.AAC_ADTS

    # You can change the number of channels
    # asi.channels = 1
    # or the sampling rate
    # asi.sample_rate = 44100

    return socket

Configure Transcoder and Encode with Pull

This is the main encoding function that uses the pull method to retrieve encoded samples. The pull method allows you to control when and how encoded data is retrieved, giving you more flexibility in processing. The transcoder reads from the input WAV file and encodes it to AAC ADTS format, but instead of writing directly to a file, it provides the encoded samples through the pull() method.

The allow_demo_mode = True setting allows the transcoder to work even without a valid license (useful for testing, but not recommended for production). After opening the transcoder, we continuously pull encoded samples and write them to the output file. The pull loop continues until pull() returns False, which happens when all input has been processed.

The function checks the error code to determine if the encoding completed successfully (EOS - End of Stream) or if an actual error occurred.

def encode(input_file: str, output_file: str) -> bool:
    """Encode WAV file to AAC ADTS using pull 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)

    with open(output_file, 'wb') as outfile:
        # Create input socket
        in_socket = MediaSocket()
        in_socket.file = input_file

        # Create output socket (no file - we'll manually write)
        out_socket = create_output_socket()

        # 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

            # Encode by pulling encoded samples
            sample = MediaSample()

            while True:
                res, _ = transcoder.pull(sample)
                if res:
                    outfile.write(bytes(sample.buffer.data))
                else:
                    break

            error = transcoder.error
            print_error("Transcoder pull", error)

            success = False
            if error.facility == ErrorFacility.Codec and error.code == CodecError.EOS:
                # ok - end of stream
                success = True

            transcoder.close()

            return success

Complete Python Code

Here’s the complete working example that demonstrates AAC ADTS encoding using the pull method. 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 using pull, and properly shuts down the library before exiting.

"""
Encode WAV file to AAC ADTS using pull mode.

This sample encodes a WAV audio file to AAC ADTS format using the pull mode
(manually pulling encoded samples and writing to output file).
"""

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_output_socket() -> MediaSocket:
    """Create output socket for AAC ADTS encoding (no file)."""
    socket = MediaSocket()
    socket.stream_type = StreamType.AAC
    socket.stream_sub_type = StreamSubType.AAC_ADTS

    pin = MediaPin()
    socket.pins.add(pin)

    asi = AudioStreamInfo()
    pin.stream_info = asi

    asi.stream_type = StreamType.AAC
    asi.stream_sub_type = StreamSubType.AAC_ADTS

    # You can change the number of channels
    # asi.channels = 1
    # or the sampling rate
    # asi.sample_rate = 44100

    return socket


def encode(input_file: str, output_file: str) -> bool:
    """Encode WAV file to AAC ADTS using pull 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)

    with open(output_file, 'wb') as outfile:
        # Create input socket
        in_socket = MediaSocket()
        in_socket.file = input_file

        # Create output socket (no file - we'll manually write)
        out_socket = create_output_socket()

        # 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

            # Encode by pulling encoded samples
            sample = MediaSample()

            while True:
                res, _ = transcoder.pull(sample)
                if res:
                    outfile.write(bytes(sample.buffer.data))
                else:
                    break

            error = transcoder.error
            print_error("Transcoder pull", error)

            success = False
            if error.facility == ErrorFacility.Codec and error.code == CodecError.EOS:
                # ok - end of stream
                success = True

            transcoder.close()

            return success


@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 AAC ADTS file',
              type=click.Path())
def main(input_file: str, output_file: str):
    """Encode WAV file to AAC ADTS using pull 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_aac_adts_pull/Hydrate-Kenny_Beltrey.aac")

        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-aac-adts-pull --input <wav file> --output <aac file>

Examples

List options:

enc-aac-adts-pull --help
Usage: enc-aac-adts-pull [OPTIONS]

  Encode WAV file to AAC ADTS using pull mode.

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

Encode the input file ./assets/aud/equinox-48KHz.wav into output file ./output/enc_aac_adts_pull/equinox-48KHz.adts.aac:

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

enc-aac-adts-pull \
    --input ./assets/aud/equinox-48KHz.wav \
    --output ./output/enc_aac_adts_pull/equinox-48KHz.adts.aac

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

enc-aac-adts-pull `
    --input ./assets/aud/equinox-48KHz.wav `
    --output ./output/enc_aac_adts_pull/equinox-48KHz.adts.aac