Description #

Audio format conversion processor

Key features #

• Supports audio format conversion (WAV, MP3, FLAC)
• Automatically selects the appropriate encoder
• Supports custom ffmpeg parameters via extra_params
• Supports local files, HTTP/HTTPS URLs, and TOS/S3 storage

Format support #

• Input: WAV, FLAC, MP3, AAC, M4A, OGG, WMA, APE, and other mainstream audio formats
• Output: WAV (pcm_s16le), MP3 (libmp3lame), FLAC (flac)

Operator parameters #

Input #

Output #

An array containing the paths of conversion results. Returns the output path on success, and None on failure.

Parameters #

If a parameter does not have a default value, it is required.

Examples #

The following code demonstrates how to use Daft (for distributed scenarios) to run the operator for audio format conversion. Supports conversion to multiple formats, including MP3, WAV, FLAC, AAC, OGG, and more.

from __future__ import annotations

import os

import daft
from daft import col
from daft.las.functions.audio import AudioConvert
from daft.las.functions.udf import las_udf

if __name__ == "__main__":
    # The converted audio will be saved to the specified TOS path. Therefore, you need to set environment variables to ensure write permissions to TOS, including: ACCESS_KEY, SECRET_KEY, TOS_ENDPOINT, TOS_REGION, TOS_TEST_DIR
    TOS_TEST_DIR_URL = os.getenv("TOS_TEST_DIR_URL", "las-cn-beijing-public-online.tos-cn-beijing.volces.com")
    TOS_TEST_DIR = os.getenv("TOS_TEST_DIR", "tos_bucket")

    if os.getenv("DAFT_RUNNER", "native") == "ray":
        import logging

        import ray

        def configure_logging():
            logging.basicConfig(
                level=logging.INFO,
                format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S",
            )
            logging.getLogger("tracing.span").setLevel(logging.WARNING)
            logging.getLogger("daft_io.stats").setLevel(logging.WARNING)
            logging.getLogger("DaftStatisticsManager").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaScheduler").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaDispatcher").setLevel(logging.WARNING)

        ray.init(dashboard_host="0.0.0.0", runtime_env={"worker_process_setup_hook": configure_logging})
        daft.set_runner_ray()

    daft.set_execution_config(actor_udf_ready_timeout=600)
    daft.set_execution_config(min_cpu_per_task=0)

    samples = {
        "input_path": [f"https://{TOS_TEST_DIR_URL}/public/archive/audio_convert/sample.wav"],
        "output_path": [f"tos://{TOS_TEST_DIR}/audio_convert/sample_converted.mp3"],
    }
    ds = daft.from_pydict(samples)

    # Using Daft to convert audio to MP3 format (as an example)
    # AudioConvert supports multiple formats: mp3, wav, flac, aac, ogg, etc.
    constructor_kwargs = {
        "output_format": "mp3",
        "sample_rate": 44100,
        "audio_map": "auto",
        "extra_params": ["-c:a", "libmp3lame", "-b:a", "192k", "-q:a", "2"],
    }

    ds = ds.with_column(
        "convert_result",
        las_udf(
            AudioConvert,
            construct_args=constructor_kwargs,
            num_cpus=1,
            concurrency=1,
            batch_size=1,
        )(col("input_path"), col("output_path")),
    )

    ds.show()
    # ╭────────────────────────────────┬────────────────────────────────┬────────────────────────────────╮
    # │ input_path                     ┆ output_path                    ┆ convert_result                 │
    # │ ---                            ┆ ---                            ┆ ---                            │
    # │ String                         ┆ String                         ┆ String                         │
    # ╞════════════════════════════════╪════════════════════════════════╪════════════════════════════════╡
    # │ https://las-public-data-qa.to… ┆ tos://tos_bucket/audio_conver… ┆ tos://tos_bucket/audio_conver… │
    # ╰────────────────────────────────┴────────────────────────────────┴────────────────────────────────╯