AAC ADTS Decoder (Pull)

This article explains how you can use Transcoder.pull to decode an AAC ADTS elementary stream.

The code snippets in this article are from the dec_aac_adts_pull Python sample.

Source Audio

For source we use the Hydrate-Kenny_Beltrey.aac file from the AVBlocks Assets repository. After downloading and unzipping you will find Hydrate-Kenny_Beltrey.aac in the aud subdirectory.

Code

This code takes an AAC ADTS stream and decodes it to uncompressed LPCM samples. The AAC decoder Transcoder returns the decoded samples with Transcoder.pull, and a second Transcoder writes them to a WAV file using Transcoder.push.

Initialize AVBlocks

The first step in any AVBlocks application is to initialize the library. This must be done before using any other AVBlocks functionality. Call Library.shutdown() at the end of the program to clean up resources.

Library.initialize()

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

result = decode(input_file, output_file)

Library.shutdown()

sys.exit(0 if result else 2)

Configure Transcoder

The decoder Transcoder has the AAC file as input, and an output socket with a single LPCM pin and no file, because the decoded samples are pulled. The WAV writer is a second Transcoder that takes raw LPCM samples as input and writes a WAV file.

def create_decoder_output_socket() -> MediaSocket:
    """Create output socket for decoder (LPCM output)."""
    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_wav_writer(output_file: str) -> Transcoder:
    """Create transcoder for writing WAV file."""
    # Input stream info, pin and socket
    infmt = AudioStreamInfo()
    infmt.stream_type = StreamType.LPCM
    infmt.channels = 2
    infmt.sample_rate = 48000
    infmt.bits_per_sample = 16

    in_pin = MediaPin()
    in_pin.stream_info = infmt

    in_socket = MediaSocket()
    in_socket.stream_type = StreamType.LPCM
    in_socket.pins.add(in_pin)

    # Output stream info, pin and socket
    outfmt = AudioStreamInfo()
    outfmt.stream_type = StreamType.LPCM
    outfmt.channels = 2
    outfmt.sample_rate = 48000
    outfmt.bits_per_sample = 16

    out_pin = MediaPin()
    out_pin.stream_info = outfmt

    out_socket = MediaSocket()
    out_socket.stream_type = StreamType.WAVE
    out_socket.pins.add(out_pin)
    out_socket.file = output_file

    # Create transcoder
    wav_writer = Transcoder()
    wav_writer.allow_demo_mode = True
    wav_writer.inputs.add(in_socket)
    wav_writer.outputs.add(out_socket)

    return wav_writer

Call Transcoder.pull

The decode function opens the decoder and the WAV writer, then pulls PCM samples from the decoder and pushes them to the WAV writer. Each call to Transcoder.pull returns one audio frame. At end of stream (CodecError.EOS) we push None to the WAV writer to signal end of stream.

def decode(input_file: str, output_file: str) -> bool:
    """Decode AAC ADTS file to WAV 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)

    # Create decoder transcoder
    with Transcoder() as decoder:
        decoder.allow_demo_mode = True

        input_socket = MediaSocket()
        input_socket.file = input_file
        decoder.inputs.add(input_socket)
        decoder.outputs.add(create_decoder_output_socket())

        if not decoder.open():
            print_error("Decoder open", decoder.error)
            return False

        # Create WAV writer transcoder
        wav_writer = create_wav_writer(output_file)

        if not wav_writer.open():
            print_error("WAV Writer open", wav_writer.error)
            decoder.close()
            return False

        # Pull-push decoding loop
        pcm_sample = MediaSample()
        decoder_eos = False

        while not decoder_eos:
            # Pull PCM sample from decoder
            res, _ = decoder.pull(pcm_sample)

            if res:
                # Push PCM sample to WAV writer
                if not wav_writer.push(0, pcm_sample):
                    print_error("WAV Writer push", wav_writer.error)
                    decoder.close()
                    wav_writer.close()
                    return False
                continue

            # No more PCM data from decoder
            error = decoder.error
            if error.facility == ErrorFacility.Codec and error.code == CodecError.EOS:
                # Push None to signal EOS to WAV writer
                wav_writer.push(0, None)
                decoder_eos = True
                continue

            # Decoder error
            print_error("Decoder pull", error)
            decoder.close()
            wav_writer.close()
            return False

        wav_writer.close()
        decoder.close()

    print("Decoding completed successfully.")
    return True

Complete Program

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

This sample decodes an AAC ADTS audio file to a WAV file using the pull mode
(manually pulling decoded PCM samples and writing to output).
"""

import os
import sys
import click

from avblocks import (
    Library, Transcoder, MediaSocket, MediaPin,
    AudioStreamInfo, MediaSample, StreamType,
    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_decoder_output_socket() -> MediaSocket:
    """Create output socket for decoder (LPCM output)."""
    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_wav_writer(output_file: str) -> Transcoder:
    """Create transcoder for writing WAV file."""
    # Input stream info, pin and socket
    infmt = AudioStreamInfo()
    infmt.stream_type = StreamType.LPCM
    infmt.channels = 2
    infmt.sample_rate = 48000
    infmt.bits_per_sample = 16

    in_pin = MediaPin()
    in_pin.stream_info = infmt

    in_socket = MediaSocket()
    in_socket.stream_type = StreamType.LPCM
    in_socket.pins.add(in_pin)

    # Output stream info, pin and socket
    outfmt = AudioStreamInfo()
    outfmt.stream_type = StreamType.LPCM
    outfmt.channels = 2
    outfmt.sample_rate = 48000
    outfmt.bits_per_sample = 16

    out_pin = MediaPin()
    out_pin.stream_info = outfmt

    out_socket = MediaSocket()
    out_socket.stream_type = StreamType.WAVE
    out_socket.pins.add(out_pin)
    out_socket.file = output_file

    # Create transcoder
    wav_writer = Transcoder()
    wav_writer.allow_demo_mode = True
    wav_writer.inputs.add(in_socket)
    wav_writer.outputs.add(out_socket)

    return wav_writer


def decode(input_file: str, output_file: str) -> bool:
    """Decode AAC ADTS file to WAV 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)

    # Create decoder transcoder
    with Transcoder() as decoder:
        decoder.allow_demo_mode = True

        input_socket = MediaSocket()
        input_socket.file = input_file
        decoder.inputs.add(input_socket)
        decoder.outputs.add(create_decoder_output_socket())

        if not decoder.open():
            print_error("Decoder open", decoder.error)
            return False

        # Create WAV writer transcoder
        wav_writer = create_wav_writer(output_file)

        if not wav_writer.open():
            print_error("WAV Writer open", wav_writer.error)
            decoder.close()
            return False

        # Pull-push decoding loop
        pcm_sample = MediaSample()
        decoder_eos = False

        while not decoder_eos:
            # Pull PCM sample from decoder
            res, _ = decoder.pull(pcm_sample)

            if res:
                # Push PCM sample to WAV writer
                if not wav_writer.push(0, pcm_sample):
                    print_error("WAV Writer push", wav_writer.error)
                    decoder.close()
                    wav_writer.close()
                    return False
                continue

            # No more PCM data from decoder
            error = decoder.error
            if error.facility == ErrorFacility.Codec and error.code == CodecError.EOS:
                # Push None to signal EOS to WAV writer
                wav_writer.push(0, None)
                decoder_eos = True
                continue

            # Decoder error
            print_error("Decoder pull", error)
            decoder.close()
            wav_writer.close()
            return False

        wav_writer.close()
        decoder.close()

    print("Decoding completed successfully.")
    return True


@click.command()
@click.option('-i', '--input', 'input_file',
              help='Input AAC file',
              type=click.Path(exists=True))
@click.option('-o', '--output', 'output_file',
              help='Output WAV file',
              type=click.Path())
def main(input_file: str, output_file: str):
    """Decode AAC ADTS file to WAV 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.adts.aac")
        if not output_file:
            output_file = os.path.join(script_dir, "../../output/dec_aac_adts_pull/Hydrate-Kenny_Beltrey.wav")

        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 = decode(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

dec-aac-adts-pull --input <aac file> --output <wav file>

Examples

List options:

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

  Decode AAC ADTS file to WAV using pull mode.

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

Decode the input file ./assets/aud/Hydrate-Kenny_Beltrey.adts.aac into output file ./output/dec_aac_adts_pull/Hydrate-Kenny_Beltrey.wav:

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

dec-aac-adts-pull \
    --input ./assets/aud/Hydrate-Kenny_Beltrey.adts.aac \
    --output ./output/dec_aac_adts_pull/Hydrate-Kenny_Beltrey.wav

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

dec-aac-adts-pull `
    --input ./assets/aud/Hydrate-Kenny_Beltrey.adts.aac `
    --output ./output/dec_aac_adts_pull/Hydrate-Kenny_Beltrey.wav