大音频文件的流式音频波形生成方案需求及现有实现

问题背景 #

需要为MP3、WAV等格式的音频文件生成波形数据，现有实现使用scipy.io.wavfile.read加载整个文件到内存，处理大于300MB的文件时会出现内存不足问题，需要流式读取或更优实现思路。

现有实现代码 #

import pandas as pd
import numpy as np
import scipy.signal as sps
from scipy.io.wavfile import read as read_wav
import fsspec

def gen_audio_wave_form(audio_file_path):
    with fsspec.open(audio_file_path) as file_obj:
        sample_rate, audio_buffer = read_wav(file_obj)
        duration = audio_buffer.shape[0] / sample_rate
    # resample to limit number of samples
    data = sps.resample(audio_buffer, 2000)

    # normalize values so that it is in the range 0-1
    normalized_audio_buffer = normalize_nd_array(data)

    target_data = create_target_data(
        audio_data=normalized_audio_buffer, duration=duration
            )
    return target_data

def normalize_nd_array(array):
    abs_array = np.abs(array)
    max_x = np.max(abs_array)
    normalized_array = np.array([x / max_x for x in abs_array])
    return normalized_array

def create_target_data(audio_data, duration: float) -> pd.DataFrame:            
    if np.ndim(audio_data) > 1:
        audio_data = [x[0] for x in audio_data]
    else:
        audio_data = audio_data.tolist()
    df = pd.DataFrame()
    df["waveform_data"] = [audio_data]
    df["duration_in_sec"] = duration
    return df

优化方案 #

核心思路 #

波形显示只需要降采样后的稀疏数据（目标2000个点），无需加载全部采样。通过分块读取音频，计算每块的关键统计值（如绝对值峰值、均方根），再将这些统计值组合成波形数据，最后归一化并调整到目标点数，彻底避免大文件内存过载问题。

方案1：使用soundfile分块读取（支持WAV/FLAC等无损格式） #

soundfile原生支持流式分块读取，适合处理大体积无损音频：

import pandas as pd
import numpy as np
import soundfile as sf
import scipy.signal as sps
import fsspec

def gen_audio_wave_form(audio_file_path):
    target_points = 2000
    block_stats = []
    max_global = 0.0

    with fsspec.open(audio_file_path, 'rb') as file_obj:
        # 获取音频元信息
        info = sf.info(file_obj)
        sample_rate = info.samplerate
        total_frames = info.frames
        duration = total_frames / sample_rate

        # 计算块大小，使块数接近目标点数
        block_size = max(1, total_frames // target_points)

        # 分块读取并计算每块绝对值峰值
        for block in sf.blocks(file_obj, blocksize=block_size, dtype='float32'):
            # 多声道取第一声道
            if block.ndim > 1:
                block = block[:, 0]
            block_peak = np.max(np.abs(block))
            block_stats.append(block_peak)
            # 更新全局最大值用于归一化
            if block_peak > max_global:
                max_global = block_peak

    # 将块统计值调整到目标点数
    if len(block_stats) != target_points:
        block_stats = sps.resample(np.array(block_stats), target_points)
    
    # 归一化到0-1范围
    normalized_data = np.array(block_stats) / max_global if max_global != 0 else block_stats

    return create_target_data(normalized_data, duration)

def create_target_data(audio_data, duration: float) -> pd.DataFrame:            
    audio_data = audio_data.tolist()
    df = pd.DataFrame()
    df["waveform_data"] = [audio_data]
    df["duration_in_sec"] = duration
    return df

方案2：处理MP3等压缩格式（使用pydub） #

如果需要支持MP3等压缩音频，pydub可按需读取音频片段，避免一次性加载全部文件：

import pandas as pd
import numpy as np
from pydub import AudioSegment
import scipy.signal as sps

def gen_audio_wave_form(audio_file_path):
    target_points = 2000
    block_stats = []
    max_global = 0.0

    # 按需加载音频，不一次性读入内存
    audio = AudioSegment.from_file(audio_file_path)
    sample_rate = audio.frame_rate
    duration = len(audio) / 1000  # 转换为秒
    channels = audio.channels

    # 计算块时长（毫秒），使块数接近目标点数
    block_duration_ms = max(1, int(duration * 1000 // target_points))

    # 分块处理音频
    for i in range(0, len(audio), block_duration_ms):
        block = audio[i:i+block_duration_ms]
        samples = np.array(block.get_array_of_samples())
        # 多声道取第一声道
        if channels > 1:
            samples = samples[::channels]
        block_peak = np.max(np.abs(samples))
        block_stats.append(block_peak)
        if block_peak > max_global:
            max_global = block_peak

    # 调整到目标点数
    if len(block_stats) != target_points:
        block_stats = sps.resample(np.array(block_stats), target_points)
    
    # 归一化
    normalized_data = np.array(block_stats) / max_global if max_global != 0 else block_stats

    return create_target_data(normalized_data, duration)

def create_target_data(audio_data, duration: float) -> pd.DataFrame:            
    audio_data = audio_data.tolist()
    df = pd.DataFrame()
    df["waveform_data"] = [audio_data]
    df["duration_in_sec"] = duration
    return df

关键优化点 #

• 流式分块读取：彻底避免大文件一次性加载到内存
• 统计值替代全采样：用每块峰值代表该段波形高度，既减少数据量，又保留波形视觉特征
• 动态归一化：分块过程中记录全局最大值，确保归一化准确性
• 多声道兼容：自动处理单/多声道音频，统一输出单声道波形数据

内容的提问来源于stack exchange，提问作者Puneeth R