如何使用Python PedalBoard为PyAudio流添加混响并创建AudioStream

一、为PyAudio流添加混响效果 #

你可以通过PedalBoard创建效果器链，在PyAudio的音频回调函数中对输入音频应用混响效果，具体实现如下：

步骤1：导入依赖并创建效果器链 #

import pyaudio
import numpy as np
from pedalboard import PedalBoard, Reverb

# 初始化混响效果器链，可调整room_size（房间大小）和wet_level（湿信号比例）参数
effect_chain = PedalBoard([Reverb(room_size=0.8, wet_level=0.4)])

步骤2：定义音频回调函数处理音频 #

p = pyaudio.PyAudio()

def process_audio(in_data, frame_count, time_info, status):
    # 将PyAudio传入的字节数据转换为浮点数数组
    audio_data = np.frombuffer(in_data, dtype=np.float32)
    # 调整数组形状为PedalBoard要求的格式：(通道数, 采样数)
    audio_data = audio_data.reshape((1, -1))
    # 应用混响效果
    processed_data = effect_chain(audio_data, sample_rate=44100)
    # 转换回字节格式返回给PyAudio
    return (processed_data.tobytes(), pyaudio.paContinue)

步骤3：启动PyAudio流并运行 #

# 打开音频流，指定格式、通道数、采样率等参数
audio_stream = p.open(
    format=pyaudio.paFloat32,
    channels=1,
    rate=44100,
    input=True,
    output=True,
    stream_callback=process_audio
)

audio_stream.start_stream()
# 保持运行直到手动中断
try:
    while audio_stream.is_active():
        pass
except KeyboardInterrupt:
    print("停止处理")

# 清理资源
audio_stream.stop_stream()
audio_stream.close()
p.terminate()

二、基于外部PyAudio流创建带效果的AudioStream #

PedalBoard的AudioStream类并不直接接受已打开的PyAudio流作为参数，但你可以复用已有PyAudio流的设备和参数，结合效果器链实现类似需求：

import pyaudio
from pedalboard import PedalBoard, Reverb, AudioStream

p = pyaudio.PyAudio()
# 初始化你自己的PyAudio流
py_stream = p.open(
    format=pyaudio.paFloat32,
    channels=1,
    rate=44100,
    input=True,
    output=True
)

# 获取已有流的关键参数
device_info = py_stream.get_input_device_info()
sample_rate = py_stream.get_sample_rate()
channels = py_stream.get_channels()

# 创建效果器链
effect_chain = PedalBoard([Reverb(room_size=0.75)])

# 使用PedalBoard的AudioStream，复用相同设备参数并应用效果
with AudioStream(
    input_device_name=device_info["name"],
    output_device_name=device_info["name"],
    sample_rate=sample_rate,
    channels=channels,
    allow_feedback=True,
    effect=effect_chain
) as stream:
    print("效果处理中，按Ctrl+C终止")
    try:
        while True:
            pass
    except KeyboardInterrupt:
        pass

# 关闭原PyAudio流并清理资源
py_stream.close()
p.terminate()

如果需要直接在已有的PyAudio流上处理音频，参考第一个问题中的回调函数方式即可，无需额外创建AudioStream。

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