如何优化麦克风音频波形生成函数以提升音频流处理速度？

原函数与使用代码 #

原生成函数 #

def generate_waveform(stream, frames_num, waveform):
    # transform stream into numpy array
    data = stream.read(CHUNK, exception_on_overflow=False)
    waveData = wave.struct.unpack("%dh" % (CHUNK), data)

    npArrayData = np.empty(len(waveData), dtype=np.float32)  # Adjust dtype as needed
    npArrayData = np.asarray(waveData)
    indata = npArrayData

    # update waveform buffer with newer data, flush older data
    waveform_prev = waveform
    waveform[-CHUNK:] = indata
    waveform[:-CHUNK] = waveform_prev[CHUNK:]

    # start and stoppage time for each CHUNK window
    start = chunk_to_seconds(frames_num)
    stop = chunk_to_seconds(frames_num + CHUNK)

    del data, waveData, npArrayData, indata, waveform_prev
    return waveform

使用代码 #

CHUNK = int(sr / updates_per_sec)
p = pyaudio.PyAudio()
stream = p.open(format=pyaudio.paInt16, channels=1, rate=sr, input=True,
                frames_per_buffer=CHUNK)

for i in range(0,sys.maxsize ** 10):
   generate_waveform(stream, CHUNK * i, waveform)  

性能分析结果（单位：毫秒） #

Line #      Hits         Time  Per Hit   % Time  Line Contents
==============================================================
     1                                           def generate_waveform(stream, frames_num, waveform):
     2                                               # transform stream into numpy array
     3         1          1.6      1.6     30.5      data = stream.read(CHUNK, exception_on_overflow=False)
     4         1          0.2      0.2      3.5      waveData = wave.struct.unpack("%dh" % (CHUNK), data)
     5         1          0.2      0.2      3.7      npArrayData = np.empty(len(waveData), dtype=np.float32)  # Adjust dtype as needed
     6                                           
     7         1          1.0      1.0     18.8      npArrayData = np.asarray(waveData) ###
     8         1          0.0      0.0      0.0      indata = npArrayData
     9                                           
    10                                               # update waveform buffer with newer data, flush older data
    11         1          0.0      0.0      0.0      waveform_prev = waveform
    12         1          1.2      1.2     23.4      waveform[-CHUNK:] = indata
    13         1          0.9      0.9     18.2      waveform[:-CHUNK] = waveform_prev[CHUNK:]
    14                                           
    15                                               # start and stoppage time for each CHUNK window
    16         1          0.0      0.0      0.9      start = chunk_to_seconds(frames_num)
    17         1          0.0      0.0      0.0      stop = chunk_to_seconds(frames_num + CHUNK)
    18                                           
    19                                           
    20         1          0.1      0.1      1.1      del data, waveData, npArrayData, indata, waveform_prev
    22         1          0.0      0.0      0.0      return waveform

优化技巧与修改方案 #

1. 删掉完全没用的冗余代码 #

• 第5行创建npArrayData后立刻被第7行覆盖，纯粹浪费资源，直接删除。
• indata = npArrayData属于多余赋值，直接使用npArrayData即可。
• waveform_prev = waveform只是创建了数组引用，后续直接操作waveform本身的切片就行，无需额外变量。
• del语句完全没必要，Python的垃圾回收机制会自动处理无用变量，删除后能节省微小的开销。
• 若start和stop变量未在后续逻辑中使用（当前函数未返回它们），直接删除这两行计算代码。

2. 合并数据转换流程，用numpy直接读取二进制 #

原来先通过wave.struct.unpack转成Python元组再转numpy数组，绕了不必要的弯路。直接用numpy的frombuffer读取原始二进制数据，一步生成目标数组，利用numpy的C实现大幅提速：

# 替换原数据转换部分
data = stream.read(CHUNK, exception_on_overflow=False)
# 跳过unpack步骤，直接从二进制生成numpy数组
indata = np.frombuffer(data, dtype=np.int16).astype(np.float32)

这一步能同时减少wave.struct.unpack和np.asarray的耗时，提升效果明显。

3. 简化波形缓冲区的滚动更新逻辑 #

原代码的waveform_prev变量完全多余，直接操作原数组的切片即可完成滚动更新：

# 替换原缓冲区更新代码
# 先将旧数据向前移动，再写入新数据
waveform[:-CHUNK] = waveform[CHUNK:]
waveform[-CHUNK:] = indata

waveform_prev只是waveform的引用，原waveform_prev[CHUNK:]等价于waveform[CHUNK:]，直接操作原数组即可。

4. 提前预定义格式化字符串（保留unpack时适用） #

如果不想用numpy的frombuffer方式，把"%dh" % CHUNK提前定义成全局常量，避免每次函数调用都重新格式化字符串：

# 在函数外部定义
UNPACK_FORMAT = "%dh" % CHUNK

# 函数内直接使用
waveData = wave.struct.unpack(UNPACK_FORMAT, data)

不过更推荐使用numpy的frombuffer方案，效率更高。

5. 改用PyAudio回调模式解决IO瓶颈 #

stream.read是同步阻塞调用，占总耗时的30.5%，是当前最大的性能瓶颈。改用PyAudio的回调模式实现异步采集，避免主线程被IO阻塞，更适合高频调用的实时场景：

# 回调函数，音频数据到达时自动触发
def callback(in_data, frame_count, time_info, status):
    indata = np.frombuffer(in_data, dtype=np.int16).astype(np.float32)
    waveform[:-CHUNK] = waveform[CHUNK:]
    waveform[-CHUNK:] = indata
    return (None, pyaudio.paContinue)

# 初始化流时指定回调函数
stream = p.open(format=pyaudio.paInt16, channels=1, rate=sr, input=True,
                frames_per_buffer=CHUNK, stream_callback=callback)

# 启动流后无需循环调用generate_waveform
stream.start_stream()
while stream.is_active():
    time.sleep(0.1)  # 主线程可处理其他业务逻辑

这种模式下音频数据会自动触发处理，实时性和整体效率远高于主动读取的方式。

优化后的完整函数 #

def generate_waveform(stream, waveform):
    data = stream.read(CHUNK, exception_on_overflow=False)
    indata = np.frombuffer(data, dtype=np.int16).astype(np.float32)
    
    # 滚动更新波形缓冲区
    waveform[:-CHUNK] = waveform[CHUNK:]
    waveform[-CHUNK:] = indata
    
    return waveform

（注：如果frames_num参数未实际使用，可直接移除，进一步简化函数）

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