Description #

Video and audio extraction processor, supports multi-stream separation

Key features #

• Supports extracting audio streams from videos via multiple paths or binary input, including local, TOS, HTTP, and more
• Supports selecting multiple audio streams, extracting only the first stream, or all streams
• Supports outputting audio to TOS, returning binary data, sample rate, and more
• Supports extracting audio within a specified time interval (start_second, end_second)
• All output audio is in mp3 format

Operator parameters #

Input #

Output #

Array of structs includes:

• audio_paths: List of audio TOS paths
• audio_arrays: List of audio array data (optional)
• binaries: List of audio binary data (optional)
• original_audio_sampling_rates: Always returned, contains the original sampling rate of each audio stream, corresponds one-to-one to audio_paths and binaries

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 extraction from video.

from __future__ import annotations

import os

import daft
from daft import col
from daft.las.functions.udf import las_udf
from daft.las.functions.video import VideoExtractAudio

if __name__ == "__main__":
    # After extracting the audio, it will be saved to the specified TOS path. Therefore, you need to set the environment variables to ensure you have permission to access and write to TOS, including: ACCESS_KEY, SECRET_KEY, TOS_ENDPOINT, TOS_REGION, TOS_TEST_DIR_URL
    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 = {
        "video_path": [f"https://{TOS_TEST_DIR_URL}/video_extract_audio/sample.mp4"],
    }
    ds = daft.from_pydict(samples)

    # Using Daft to extract audio from video
    output_tos_dir = f"tos://{TOS_TEST_DIR}/video_extract_audio"
    constructor_kwargs = {
        "output_tos_dir": output_tos_dir,
        "output_audio_binary": True,
        "output_sampling_rate": 16000,
        "output_audio_format": "mp3",
    }

    ds = ds.with_column(
        "extract_results",
        las_udf(
            VideoExtractAudio,
            construct_args=constructor_kwargs,
            num_cpus=1,
            concurrency=1,
            batch_size=1,
        )(col("video_path")),
    )

    ds.show()
    # ╭────────────────────────────────┬──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
    # │ video_path                     ┆ extract_results                                                                                                                          │
    # │ ---                            ┆ ---                                                                                                                                      │
    # │ Utf8                           ┆ Struct[audio_paths: List[Utf8], audio_arrays: List[List[Float32]], binaries: List[Binary], original_audio_sampling_rates: List[Float64]] │
    # ╞════════════════════════════════╪══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╡
    # │ tos://tos_bucket/video_extrac… ┆ {audio_paths: [tos://las-ai-c…                                                                                                           │
    # ╰────────────────────────────────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯