如何在Python的finally语句中停止PyAudio录音并导出音频

问题描述 #

我想咨询是否可以在Python的finally语句中停止PyAudio录音并导出已录制的音频内容。例如，当脚本处于录音状态时突然崩溃，是否有办法实现停止录音并导出已录制的部分？我曾尝试编写一个独立的停止录音并导出的函数，将其放在finally语句中执行，但并未生效，仅生成了一个新文件。

原代码 #

import pyaudio
import wave
import time
import signal
from datetime import datetime

# Configuration
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 44100
RECORD_SECONDS = 10
# Initialize PyAudio
audio = pyaudio.PyAudio()

def record_audio(filename):
    stream = audio.open(format=FORMAT, channels=CHANNELS,
                        rate=RATE, input=True, frames_per_buffer=CHUNK)
    print("Recording started")

    frames = []
    data = stream.read(CHUNK)
    frames.append(data)

    print("Recording stopped")
    stream.stop_stream()
    stream.close()

    # Save to WAV file
    wf = wave.open(filename, 'wb')
    wf.setnchannels(CHANNELS)
    wf.setsampwidth(audio.get_sample_size(FORMAT))
    wf.setframerate(RATE)
    wf.writeframes(b''.join(frames))
    wf.close()

def main():
    try:
       while True:
                timestamp = datetime.now().strftime("%m-%d-%Y_%H-%M-%S")
                wav_filename = f"/home/henrikkoehler/Desktop/Audio_Files/{timestamp}.wav"

                record_audio(wav_filename)

         
                print(f"Audio saved to {wav_filename}")

                # Wait a bit to avoid multiple recordings from a single press
                time.sleep(1)

    except KeyboardInterrupt:
        print("Exiting...")

    except:
        time.sleep(1)
        t_end = time.time() + 15 * 1
        while time.time() < t_end:
            print("Error! Restart!")

    finally:
        audio.terminate()

if __name__ == "__main__":
    main()

问题分析 #

1. 录音逻辑不完整：原record_audio函数仅读取了一帧音频数据，没有实现持续录音的循环，无法完成完整的录音流程。
2. 状态无法被finally访问：录音的流对象和帧数据都被封装在record_audio函数内部，finally块无法获取这些状态，自然无法在崩溃时处理未完成的录音。
3. 异常处理逻辑冗余：通用except块中的重启逻辑会阻塞程序，导致finally无法及时执行。

解决方案 #

要实现崩溃时保存已录制的音频，需要将录音的关键状态（流、帧、文件名）提升到全局或外层作用域，让finally块能够访问并处理这些资源。以下是修改后的代码：

import pyaudio
import wave
import time
from datetime import datetime

# 全局变量存储录音状态，让finally可以访问
current_stream = None
current_frames = []
current_filename = ""

# 配置参数
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 44100
RECORD_SECONDS = 10

audio = pyaudio.PyAudio()

def save_recording():
    """保存已录制的音频帧到文件"""
    global current_frames, current_filename
    if current_frames and current_filename:
        print(f"正在保存已录制的音频到 {current_filename}")
        wf = wave.open(current_filename, 'wb')
        wf.setnchannels(CHANNELS)
        wf.setsampwidth(audio.get_sample_size(FORMAT))
        wf.setframerate(RATE)
        wf.writeframes(b''.join(current_frames))
        wf.close()
        current_frames = []  # 清空缓存

def record_audio(filename):
    global current_stream, current_frames, current_filename
    current_filename = filename
    current_frames = []
    
    current_stream = audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        input=True,
        frames_per_buffer=CHUNK
    )
    print("Recording started")
    
    try:
        # 持续录音的循环
        for _ in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
            data = current_stream.read(CHUNK)
            current_frames.append(data)
    except Exception as e:
        print(f"录音过程中出现异常: {e}")
        raise  # 重新抛出异常，让外层处理
    
    print("Recording stopped")
    current_stream.stop_stream()
    current_stream.close()
    current_stream = None
    save_recording()

def main():
    try:
        while True:
            timestamp = datetime.now().strftime("%m-%d-%Y_%H-%M-%S")
            wav_filename = f"/home/henrikkoehler/Desktop/Audio_Files/{timestamp}.wav"
            
            record_audio(wav_filename)
            print(f"Audio saved to {wav_filename}")
            
            time.sleep(1)

    except KeyboardInterrupt:
        print("用户中断程序")
    except Exception as e:
        print(f"程序出现错误: {e}")
    finally:
        print("执行清理操作")
        # 停止当前正在运行的录音流
        if current_stream:
            current_stream.stop_stream()
            current_stream.close()
        # 保存已录制的内容
        save_recording()
        # 终止PyAudio实例
        audio.terminate()

if __name__ == "__main__":
    main()

关键改进点 #

• 全局状态存储：用全局变量保存当前的录音流、帧数据和文件名，确保finally块能访问到这些资源。
• 独立保存函数：提取save_recording函数，统一处理音频保存逻辑，既可以在正常录音结束时调用，也能在崩溃时执行。
• 完善异常处理：移除冗余的重启循环，让finally块能及时执行清理和保存操作。
• 完整录音循环：实现了基于时间的持续录音逻辑，符合RECORD_SECONDS的配置。

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