使用 FFmpeg 直接解码 Mu-law 编码音频缓冲区数据

import subprocess
import numpy as np

def ffmpeg_read(bpayload: bytes, sampling_rate: int) -> np.array:
    """
    通过 ffmpeg 读取音频文件的辅助函数。
    此函数设计用于处理带有标准文件头的音频文件。
    """
    ar = f"{sampling_rate}"
    ac = "1"
    format_for_conversion = "f32le"
    ffmpeg_command = [
        "ffmpeg",
        "-i",
        "pipe:0", # 默认尝试从管道推断格式，对原始Mu-law数据会失败
        "-ac", ac,
        "-ar", ar,
        "-f", format_for_conversion,
        "-hide_banner",
        "-loglevel", "quiet",
        "pipe:1",
    ]

    try:
        with subprocess.Popen(ffmpeg_command, stdin=subprocess.PIPE, stdout=subprocess.PIPE) as ffmpeg_process:
            output_stream = ffmpeg_process.communicate(bpayload)
    except FileNotFoundError as error:
        raise ValueError("ffmpeg 未安装，但它是加载音频文件所必需的。") from error

    out_bytes = output_stream[0]
    audio = np.frombuffer(out_bytes, np.float32)

    if audio.shape[0] == 0:
        raise ValueError(
            "音频文件格式不正确或已损坏。请确保音频文件具有有效的扩展名 (例如 wav, flac 或 mp3) 且未损坏。"
        )
    return audio

import subprocess
import numpy as np
import io

def ffmpeg_read_mulaw(bpayload: bytes, sampling_rate: int) -> np.array:
    """
    通过 FFmpeg 直接解码 Mu-law 编码的音频缓冲区数据。

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

    Returns:
        np.array: 解码后的浮点 PCM 格式的 NumPy 数组。

    Raises:
        ValueError: 如果 FFmpeg 未安装或解码失败。
    """
    ar = f"{sampling_rate}"
    ac = "1"  # Mu-law 编码通常是单声道
    format_for_conversion = "f32le" # 输出为 32 位浮点小端序 PCM

    ffmpeg_command = [
        "ffmpeg",
        "-f", "mulaw",  # 关键：明确指定输入格式为 mulaw
        "-ar", ar,      # 指定输入采样率 (对于原始数据至关重要)
        "-ac", ac,      # 指定输入声道数 (对于原始数据至关重要)
        "-i", "pipe:0", # 从标准输入读取数据
        "-b:a", "256k", # 设置输出音频比特率，有助于控制输出质量（对于解码到原始PCM，影响相对较小）
        "-f", format_for_conversion, # 指定输出格式为 32 位浮点 PCM
        "-hide_banner", # 隐藏 FFmpeg 启动信息
        "-loglevel", "quiet", # 设置日志级别为静默，减少控制台输出
        "pipe:1",       # 将处理后的输出写入标准输出
    ]

    try:
        with subprocess.Popen(ffmpeg_command, stdin=subprocess.PIPE, stdout=subprocess.PIPE) as ffmpeg_process:
            # 将 Mu-law 字节数据通过 stdin 传递给 FFmpeg 进程
            output_stream = ffmpeg_process.communicate(bpayload)
    except FileNotFoundError as error:
        raise ValueError("ffmpeg 未安装，但它是加载音频文件所必需的。请确保 FFmpeg 已安装并配置到系统 PATH 中。") from error

    out_bytes = output_stream[0]
    # 将 FFmpeg 输出的原始 PCM 字节转换为 NumPy 浮点数组
    audio = np.frombuffer(out_bytes, np.float32)

    if audio.shape[0] == 0:
        raise ValueError("FFmpeg 解码 Mu-law 编码数据失败，输出为空。请检查输入数据和 FFmpeg 参数是否正确。")

    return audio

# 示例 Mu-law 编码数据（实际数据将根据音频内容更长）
# 这是一个简短的字节序列示例，代表Mu