Python无损耗连续录音求助：解决sd.wait()耗时导致的音频丢失

问题描述 #

我正在开发一个实时语音转文字项目，需要连续录制音频：每5秒录制一段，每次将最后2段（共10秒）写入文件。但Python脚本中sd.wait()耗时约5.06秒，导致音频丢失。通过时间戳排查确认，仅sd.wait()这一行占用额外时间。

原代码 #

import sounddevice as sd
import numpy as np
from scipy.io.wavfile import write
from pydub import AudioSegment
import time as t
import os


seconds = 5
fs = 44100
recordsMaxLength = 2


def record_audio(seconds=seconds, fs=fs, recordsMaxLength=recordsMaxLength):
    audioIndex = 0
    current_time = 0
    records = []
    output_dir = "records"
    concatedRecords = None
    os.makedirs(output_dir, exist_ok=True)

    try:
        for i in range(3):
            print("time between starts:", t.time()-current_time)
            record = sd.rec(int(seconds * fs), samplerate=fs, channels=2, dtype='int16')
            current_time = t.time()
            sd.wait()
            print("time: ", t.time() - current_time)
            current_time = t.time()
            if len(records) < recordsMaxLength:
                records.append(record)
            else:
                for i in range(recordsMaxLength - 1):
                    records[i] = records[i + 1]
                records[recordsMaxLength - 1] = record

            if len(records) > 1:
                concatedRecords = np.concatenate(records)

            audioIndex += 1
    except KeyboardInterrupt:
        pass

    # If there are no records, there's nothing to save
    if concatedRecords is not None:
        # Convert numpy array to pydub AudioSegment
        audio_segment = AudioSegment(
            data=concatedRecords.tobytes(),
            sample_width=concatedRecords.dtype.itemsize,
            frame_rate=fs,
            channels=2
        )

        # Export the AudioSegment as an MP3 file
        output_path = os.path.join(output_dir, "records.mp3")
        audio_segment.export(output_path, format="mp3")
        print(f"Saved recording as {output_path}")
    else:
        print("No audio recorded to save.")

解决方案 #

核心问题分析 #

sd.wait()会阻塞主线程直到当前录制完成，但实际录制时长略超过设定的5秒，导致下一次录制启动延迟，出现音频间隙。要实现无间隙连续录制，需改用非阻塞录制+回调函数的方式，避免主线程被阻塞。

改进后代码 #

import sounddevice as sd
import numpy as np
from pydub import AudioSegment
import os
import sys

seconds = 5
fs = 44100
records_max_length = 2
records = []
output_dir = "records"
os.makedirs(output_dir, exist_ok=True)

def callback(indata, frames, time, status):
    """音频录制回调函数，实时处理录制数据"""
    global records
    if status:
        print(status, file=sys.stderr)
    # 保存当前录制的片段
    records.append(indata.copy())
    # 仅保留最近的2段音频
    if len(records) > records_max_length:
        records.pop(0)
    # 当凑齐2段时，合并并写入文件
    if len(records) == records_max_length:
        concated_records = np.concatenate(records)
        audio_segment = AudioSegment(
            data=concated_records.tobytes(),
            sample_width=concated_records.dtype.itemsize,
            frame_rate=fs,
            channels=2
        )
        output_path = os.path.join(output_dir, "latest_records.mp3")
        audio_segment.export(output_path, format="mp3")
        print(f"已更新录制文件: {output_path}")

def start_continuous_recording():
    # 启动非阻塞连续录制，每5秒触发一次回调
    with sd.InputStream(samplerate=fs, channels=2, dtype='int16',
                       blocksize=int(seconds * fs), callback=callback):
        print("开始连续录制，按Ctrl+C停止...")
        while True:
            pass

if __name__ == "__main__":
    try:
        start_continuous_recording()
    except KeyboardInterrupt:
        print("\n录制已停止")

关键改进点 #

• 采用sd.InputStream的回调机制，非阻塞录制，彻底避免sd.wait()的阻塞延迟
• 回调函数实时处理音频片段，自动维护最近2段数据，无需手动控制录制循环
• 每次凑齐2段音频后立即写入文件，保证实时性
• 主线程仅保持运行状态，不参与录制阻塞，确保音频录制无间隙

内容的提问来源于stack exchange，提问作者Haluk Umut Berkut