使用FFmpeg高效解码mu-law编码音频缓冲区数据

import subprocess
import numpy as np
import io

def ffmpeg_read_mulaw(bpayload: bytes, sampling_rate: int) -> np.array:
    """
    使用FFmpeg解码mu-law编码的音频缓冲区数据。

    参数:
        bpayload (bytes): mu-law编码的原始字节数据。
        sampling_rate (int): 音频的采样率（例如，8000 Hz）。

    返回:
        np.array: 解码后的浮点数数组，表示音频波形。

    抛出:
        ValueError: 如果ffmpeg未找到或解码失败。
    """
    ar = f"{sampling_rate}"
    ac = "1"  # mu-law通常是单声道
    format_for_conversion = "f32le" # 输出为32位小端浮点数

    ffmpeg_command = [
        "ffmpeg",
        "-f", "mulaw",  # 明确指定输入格式为mu-law
        "-ar", ar,      # 指定输入采样率
        "-ac", ac,      # 指定输入声道数
        "-i", "pipe:0", # 从标准输入读取数据
        "-b:a", "256k",  # 设置输出音频比特率，确保转换质量
        "-f", format_for_conversion, # 指定输出格式为32位浮点数
        "-hide_banner", # 隐藏FFmpeg启动时的版权信息
        "-loglevel", "quiet", # 抑制FFmpeg的日志输出
        "pipe:1",       # 将处理结果输出到标准输出
    ]

    try:
        # 使用subprocess.Popen通过管道与FFmpeg交互
        with subprocess.Popen(
            ffmpeg_command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE # 捕获标准错误，以便更好地调试
        ) as ffmpeg_process:
            # 将mu-law数据写入FFmpeg的stdin，并读取stdout
            output_stream, error_stream = ffmpeg_process.communicate(bpayload)
            if ffmpeg_process.returncode != 0:
                raise ValueError(
                    f"FFmpeg process exited with error code {ffmpeg_process.returncode}. "
                    f"Stderr: {error_stream.decode('utf-8')}"
                )
    except FileNotFoundError as error:
        raise ValueError("ffmpeg was not found but is required to load audio files.") from error
    except Exception as e:
        raise ValueError(f"An unexpected error occurred during FFmpeg execution: {e}") from e

    out_bytes = output_stream
    audio = np.frombuffer(out_bytes, np.float32)

    if audio.shape[0] == 0:
        raise ValueError("Failed to decode mu-law encoded data with FFMPEG. Output audio is empty.")

    return audio

# 示例 mu-law 编码数据 (实际数据会更长)
# 这些字节代表了mu-law编码的音频样本
mu_encoded_data = b"\x7F\xFF\x80\x01\x7F\xFF\x00\x02\x7E\xFE\x03\x7D\xFD\x04\x7C\xFC" 
sampling_rate = 8000 # 电话通信中常见的采样率

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

# 你可以将解码后的数据保存为WAV文件进行验证
# import soundfile as sf
# sf.write("decoded_mulaw_audio.wav", decoded_audio, sampling_rate)