FFmpeg实战：高效转换与读取μ-law编码音频缓冲区数据

import subprocess
import numpy as np
import io

def ffmpeg_read_mulaw(bpayload: bytes, sampling_rate: int = 8000) -> np.ndarray:
    """
    Helper function to read mu-law encoded audio buffer data through ffmpeg.

    Args:
        bpayload (bytes): The mu-law encoded audio buffer data.
        sampling_rate (int): The sampling rate of the mu-law audio. Defaults to 8000 Hz.

    Returns:
        np.ndarray: A NumPy array containing the decoded audio as float32 samples.

    Raises:
        ValueError: If ffmpeg is not found or decoding fails.
    """
    ar = f"{sampling_rate}"
    ac = "1"  # Assuming mono channel for mu-law phone audio
    format_for_conversion = "f32le" # Output format: 32-bit float, little-endian

    # FFmpeg command to decode mu-law from stdin and output f32le PCM to stdout
    ffmpeg_command = [
        "ffmpeg",
        "-f", "mulaw",          # Explicitly specify input format as mu-law
        "-ar", ar,              # Input sampling rate
        "-ac", ac,              # Input audio channels (mono)
        "-i", "pipe:0",         # Read input from stdin
        "-b:a", "256k",         # Output audio bitrate (can be adjusted or omitted for raw PCM output)
        "-f", format_for_conversion, # Output format: 32-bit float PCM
        "-hide_banner",         # Suppress FFmpeg banner
        "-loglevel", "quiet",   # Suppress FFmpeg logging
        "pipe:1",               # Write output to stdout
    ]

    try:
        # Execute FFmpeg as a subprocess, piping input and capturing output
        with subprocess.Popen(ffmpeg_command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as ffmpeg_process:
            output_stream, _ = ffmpeg_process.communicate(bpayload)
    except FileNotFoundError as error:
        raise ValueError(
            "ffmpeg was not found but is required to load audio files from filename. "
            "Please ensure ffmpeg is installed and accessible in your system's PATH."
        ) from error

    out_bytes = output_stream
    # Convert raw bytes output from FFmpeg into a NumPy array of float32 samples
    audio = np.frombuffer(out_bytes, np.float32)

    if audio.shape[0] == 0:
        # If no audio data is produced, it indicates a decoding failure
        raise ValueError("Failed to decode mu-law encoded data with FFMPEG. "
                         "Check input data integrity and ffmpeg parameters.")

    return audio

# 假设这是你接收到的μ-law编码的缓冲区数据
# 这是一个非常简短的示例，实际数据会更长
mu_encoded_data = b"\x7F\xFF\x80\x01\x7F\xFF\x00\x10\x7F\xFF\x80\x01"
sampling_rate = 8000

try:
    decoded_audio = ffmpeg_read_mulaw(mu_encoded_data, sampling_rate)
    print("成功解码μ-law音频数据，形状:", decoded_audio.shape)
    print("前5个解码后的音频样本:", decoded_audio[:5])
    print("数据类型:", decoded_audio.dtype)
except ValueError as e:
    print(f"解码失败: {e}")

# 你可以将decoded_audio用于后续的音频处理任务，例如语音识别模型的输入