---

玩转音频：使用 Python sounddevice 轻松处理声音

在现代计算机应用中，音频处理无处不在，从简单的音乐播放器到复杂的实时语音识别和合成系统。对于 Python 开发者而言，与系统音频设备进行交互似乎是一个潜在的挑战，涉及到低级别API、跨平台兼容性等问题。然而，有了 sounddevice 库，这一切变得前所未有的简单和优雅。

sounddevice 是一个基于 PortAudio 音频库的 Python 模块，它提供了一个简单直观的接口，用于直接访问计算机的音频输入和输出设备。无论是播放音频数据、录制声音、或者进行实时的音频流处理，sounddevice 都能轻松胜任。

本文将带你深入了解 sounddevice 的方方面面，从安装、设备管理，到最核心的音频流处理，并结合丰富的代码示例，帮助你轻松掌握如何在 Python 中玩转音频。

1. 为什么选择 sounddevice？

在 Python 中处理音频，其实有不止一种选择。一些库可能专注于文件格式（如 wave, scipy.io.wavfile, soundfile），一些可能用于音频分析（如 librosa），还有一些可能提供更高级的框架（如 PyAudio）。那么 sounddevice 的独特优势在哪里呢？

• 基于 PortAudio： PortAudio 是一个免费的、跨平台的音频 I/O 库，支持多种操作系统和音频后端（如 Windows MME/DirectSound/WDM-KS, macOS Core Audio, Linux ALSA/PulseAudio等）。sounddevice 利用了 PortAudio 的强大功能，因此具有出色的跨平台兼容性。
• 简洁的 API： sounddevice 的设计哲学是简洁易用。它提供了一套直观的函数和类，使得音频设备的播放、录制和流处理变得非常直接。
• 直接访问设备： 与一些只处理音频文件的库不同，sounddevice 能够直接与音频硬件设备进行交互，支持实时的音频输入和输出。
• 支持 NumPy： sounddevice 与 NumPy 紧密集成，音频数据通常以 NumPy 数组的形式进行处理。这使得它可以方便地与科学计算、信号处理等其他 Python 库结合使用。
• 灵活性： sounddevice 支持阻塞模式和非阻塞（回调）模式的音频流，可以满足不同应用场景的需求，从简单的脚本到需要实时响应的复杂应用。

总而言之，如果你需要在 Python 中直接与音频设备进行交互，进行实时的音频输入、输出或处理，并且追求跨平台兼容性和简洁的API，那么 sounddevice 是一个非常理想的选择。

2. 安装 sounddevice

安装 sounddevice 非常简单，只需要使用 pip 包管理器即可：

bash
pip install sounddevice

sounddevice 依赖于 PortAudio。在大多数操作系统上，pip 安装会自动尝试找到或编译 PortAudio。如果安装过程中遇到 PortAudio相关的错误，你可能需要在系统上先安装 PortAudio 开发库。

• Debian/Ubuntu: sudo apt-get install libportaudio2 libportaudio-ocaml libportaudio-ocaml-dev portaudio19-dev
• Fedora: sudo dnf install portaudio portaudio-devel
• macOS (using Homebrew): brew install portaudio
• Windows: 通常 pip 会自带预编译的 PortAudio 二进制文件，如果不行，可能需要从 PortAudio 官网下载或者寻找其他解决方案。

安装成功后，你就可以在 Python 脚本中导入 sounddevice 了：

python
import sounddevice as sd

习惯上，我们通常将其别名为 sd。

3. 了解你的音频设备

在使用 sounddevice 进行音频处理之前，了解你的系统上有哪些可用的音频设备非常重要。sounddevice 提供了查询设备信息的功能。

3.1 列出所有音频设备

使用 sd.query_devices() 函数可以获取系统上所有可用的音频输入和输出设备的列表。

“`python
import sounddevice as sd

print(sd.query_devices())
“`

运行这段代码，你将看到一个字典列表，每个字典代表一个音频设备，包含了设备的名称、主机API、输入/输出通道数、默认采样率等信息。输出可能类似这样（具体取决于你的系统和连接的音频设备）：

“`

0 Microsoft Sound Mapper – Input, MME (2 in, 0 out)
< 1 Microphone Array (Realtek(R) Au, WDM-KS (4 in, 0 out)
2 Microsoft Sound Mapper – Output, MME (0 in, 2 out)
< 3 Speakers (Realtek(R) Audio), WDM-KS (0 in, 2 out)
4 Primary Sound Capture Driver, DirectSound (2 in, 0 out)
5 Microphone Array (Realtek(R) Audio), DirectSound (4 in, 0 out)
6 Primary Sound Driver, DirectSound (0 in, 2 out)
7 Speakers (Realtek(R) Audio), DirectSound (0 in, 2 out)
8 Microphone (NVIDIA Broadcast), WDM-KS (2 in, 0 out)
9 Speakers (NVIDIA Broadcast), WDM-KS (0 in, 2 out)
10 Digital Audio (S/PDIF) (Realtek(R, WDM-KS (0 in, 2 out)
11 Digital Audio (HDMI) (NVIDIA High Definition Audio), WDM-KS (0 in, 8 out)
12 CABLE Output (VB-Audio Virtual Cable), WDM-KS (2 in, 0 out)
13 CABLE Input (VB-Audio Virtual Cable), WDM-KS (0 in, 2 out)
“`

列表中的每个设备前面可能有一个 > 或 < 符号。> 表示这是默认的输入设备，< 表示这是默认的输出设备。注意，一个设备可能同时具备输入和输出能力，也可能只有输入或只有输出。括号中的数字表示输入和输出通道的数量。

3.2 获取默认设备

sd.query_devices() 返回的列表中，默认设备通常带有 > 或 < 标记。你也可以使用 sd.default.device 来直接获取当前设置的默认输入和输出设备。它返回一个包含两个元素的元组 (input_device_index, output_device_index) 或者设备名称。

“`python
import sounddevice as sd

default_devices = sd.default.device
print(f”默认输入设备索引和名称: {default_devices[0]}”)
print(f”默认输出设备索引和名称: {default_devices[1]}”)

你也可以通过索引查询详细信息

default_input_info = sd.query_devices(default_devices[0])
default_output_info = sd.query_devices(default_devices[1])
print(“\n默认输入设备详细信息:”)
print(default_input_info)
print(“\n默认输出设备详细信息:”)
print(default_output_info)
“`

sd.default.device 也可以用来设置默认设备：

“`python

设置默认输入设备为索引 1 的设备，默认输出设备为索引 3 的设备

sd.default.device = (1, 3)

设置默认输入设备为某个名称的设备

sd.default.device = (‘Microphone Array (Realtek(R) Au’, ‘Speakers (Realtek(R) Audio)’)

print(“\n新的默认设备设置:”)
print(sd.default.device)
“`

你可以通过设备索引（列表中的序号）或者设备名称（字符串）来指定设备。使用索引通常更稳定，因为名称可能会随着系统变化而改变，但索引也可能在设备插拔后改变。使用名称更具可读性。选择哪种方式取决于你的应用场景。

3.3 获取默认采样率

每个音频设备都有一个或多个支持的采样率。sd.default.samplerate 可以获取当前设置的默认采样率。通常，这是一个合理的默认值，但对于特定的应用，你可能需要使用设备支持的其他采样率。

“`python
import sounddevice as sd

print(f”默认采样率: {sd.default.samplerate} Hz”)

你也可以为输入或输出单独设置默认采样率

sd.default.samplerate = 48000

“`

请注意，修改 sd.default.device 和 sd.default.samplerate 会影响后续使用这些默认值的所有 sounddevice 函数和类。

4. 简单的播放和录制 (阻塞模式)

sounddevice 提供了简单的阻塞模式函数 sd.play() 和 sd.rec()，用于播放和录制固定长度的音频数据。这些函数会暂停程序的执行，直到播放或录制完成。它们非常适合处理短音频片段或简单的演示。

4.1 播放音频数据

sd.play(data, samplerate, device=None, blocking=False) 函数用于播放音频数据。

• data: 一个 NumPy 数组，包含要播放的音频数据。对于单声道，形状是 (样本数,)；对于立体声或其他多通道音频，形状是 (样本数, 通道数)。数据类型可以是 float32 (范围 -1.0 到 1.0), int16, int32 等，但通常推荐使用 float32。
• samplerate: 音频数据的采样率 (Hz)。
• device: 可选参数，指定播放设备。可以是设备索引或名称。如果为 None，则使用默认输出设备。
• blocking: 如果为 True (默认)，函数会阻塞直到播放完成。如果为 False，函数会立即返回，你需要使用 sd.wait() 来等待播放完成。

我们来生成一个简单的正弦波声音并播放它：

“`python
import sounddevice as sd
import numpy as np
import math

音频参数

samplerate = 44100 # 采样率
duration = 2.0 # 持续时间 (秒)
frequency = 440.0 # 音调频率 (Hz), A4 音

生成正弦波数据

计算总样本数

num_samples = int(samplerate * duration)

生成时间轴，从 0 到 duration (不包含duration，因为第一个样本在时间0)

time = np.linspace(0., duration, num_samples, endpoint=False)

生成正弦波信号，幅度设为 0.5 防止溢出

amplitude = 0.5
data = amplitude * np.sin(2. * math.pi * frequency * time)

播放音频

print(f”正在播放 {duration} 秒的 {frequency} Hz正弦波…”)
sd.play(data, samplerate)

等待播放完成

sd.wait()
print(“播放完成.”)
“`

运行这段代码，你应该能听到一个持续 2 秒的 440 Hz 的正弦波声音。

4.2 录制音频数据

sd.rec(frames, samplerate, channels=None, dtype=None, device=None, blocking=False) 函数用于录制固定长度的音频数据。

• frames: 要录制的样本总数。
• samplerate: 录制的采样率 (Hz)。
• channels: 可选参数，指定录制通道数。如果为 None，则使用默认输入设备的通道数。
• dtype: 可选参数，指定录制数据的 NumPy 数组数据类型。如果为 None，则使用默认输入设备的推荐类型。
• device: 可选参数，指定录制设备。可以是设备索引或名称。如果为 None，则使用默认输入设备。
• blocking: 如果为 True (默认)，函数会阻塞直到录制完成。如果为 False，函数会立即返回，你需要使用 sd.wait() 来等待录制完成。

函数返回一个 NumPy 数组，包含录制到的音频数据。

我们来录制 5 秒钟的声音：

“`python
import sounddevice as sd
import numpy as np

音频参数

samplerate = 44100 # 采样率
duration = 5.0 # 持续时间 (秒)
channels = 1 # 录制单声道

计算总样本数

num_frames = int(samplerate * duration)

print(f”正在录制 {duration} 秒音频…”)

录制音频

返回的 recorded_data 形状为 (num_frames, channels)

recorded_data = sd.rec(num_frames, samplerate=samplerate, channels=channels, dtype=’float32′)

等待录制完成

sd.wait()
print(“录制完成.”)

你可以进一步处理 recorded_data，例如保存到文件或播放

例如，简单播放一下录制的内容

print(“正在播放录制内容…”)
sd.play(recorded_data, samplerate)
sd.wait()
print(“播放录制内容完成.”)

注意：如果想保存到文件，需要结合 soundfile 等库

例如:

import soundfile as sf

sf.write(‘my_recording.wav’, recorded_data, samplerate)

print(“已保存到 my_recording.wav”)

“`

这段代码会录制 5 秒的声音，并将其存储在 recorded_data NumPy 数组中，然后立即播放出来。

sd.play() 和 sd.rec() 的阻塞模式对于简单的任务非常方便。然而，对于需要实时处理音频（例如，对输入音频进行修改后立即输出，或者需要长时间不间断地播放/录制），阻塞模式就不适用了。这时就需要使用 sounddevice 的核心功能：音频流 (Streams)。

5. 音频流 (Streams)：实时处理的基石

音频流是 sounddevice 处理实时音频的核心机制。它在单独的线程中运行，与主程序并行，负责不断地从音频输入设备读取数据并/或向音频输出设备写入数据。

sounddevice 提供 sd.Stream 类来管理音频流。创建 sd.Stream 对象后，你可以启动它，让它在后台运行，直到你停止或关闭它。

音频流有两种主要模式：

1. 阻塞流： 你可以在主线程中创建流，并在需要读写数据时调用流对象的方法 (如 stream.read(), stream.write())。这些方法会阻塞直到足够的数据可用或被处理。适合在需要同步处理的场景下使用，但不如回调模式灵活。
2. 回调流： 这是更常用的方式，特别是在需要实时、非阻塞处理时。你提供一个回调函数，sounddevice 会在音频设备需要更多输出数据或有新的输入数据可用时自动调用这个函数。音频数据的读写都在这个回调函数中进行。

我们将重点介绍更强大的回调流模式。

5.1 创建和配置音频流

创建 sd.Stream 对象时，你可以指定多种参数：

python
sd.Stream(
samplerate=None,
blocksize=None,
device=None,
channels=None,
dtype=None,
callback=None,
finished_callback=None,
extra_settings=None,
clip_output=False,
dither_output=False,
never_drop_input=False,
prime_output_buffers_on_start=False
)

关键参数解释：

• samplerate: 流的采样率。如果为 None，使用默认采样率。
• blocksize: 每个数据块（buffer）的大小，以帧 (frames) 为单位。PortAudio 会以这个大小的数据块调用回调函数。较小的 blocksize 会降低延迟，但会增加 CPU 开销。如果为 None，PortAudio 会选择一个合适的值。通常推荐指定一个 2 的幂次方值，如 1024 或 512。
• device: 使用的音频设备。可以是单个设备的索引或名称（用于同时输入/输出），或者一个包含输入和输出设备索引/名称的元组 (input_device, output_device)。如果为 None，使用默认输入/输出设备。
• channels: 指定输入和输出通道数。可以是一个整数（同时用于输入和输出），或者一个包含输入和输出通道数的元组 (input_channels, output_channels)。如果为 None，使用默认设备的通道数。
• dtype: 指定输入和输出数据的 NumPy 数组数据类型。可以是一个 NumPy dtype（同时用于输入和输出），或者一个包含输入和输出 dtype 的元组 (input_dtype, output_dtype)。如果为 None，使用默认设备的推荐类型，通常是 float32。
• callback: 这是 回调模式 的核心。一个可调用对象（函数），sounddevice 会在需要处理音频数据时调用它。回调函数的签名必须是 callback(indata, outdata, frames, time, status)。详细解释见下文。
• finished_callback: 可选参数。当音频流正常结束时（例如，播放完所有数据或停止流），会调用这个函数。
• indata 和 outdata 参数：创建 sd.Stream 时，你需要通过 channels, dtype 等参数告诉 sounddevice 你是需要输入（录制）、输出（播放）还是同时需要。如果只需要输入，只指定 channels 为整数即可；如果只需要输出，指定 channels 为整数即可；如果同时需要，指定 channels=(input_channels, output_channels)。相应的，在回调函数中，如果你需要输入，indata 将是一个非 None 的 NumPy 数组；如果你需要输出，outdata 将是一个非 None 的 NumPy 数组。

5.2 回调函数的签名和参数

音频流的回调函数是实时处理音频数据的地方。它的标准签名如下：

“`python
def callback(indata, outdata, frames, time, status):
“””
音频流回调函数

参数:
    indata (np.ndarray): 输入音频数据数组 (frames, input_channels)。如果流没有输入，则为 None。
    outdata (np.ndarray): 输出音频数据数组 (frames, output_channels)。你需要将要播放的数据写入这里。如果流没有输出，则为 None。
    frames (int): 当前数据块中的帧数 (与 blocksize 相同)。
    time (CData): 一个结构体，包含当前数据块的时间戳信息 (inputBufferAdcTime, currentTime, outputBufferDacTime)。
    status (Flags): 一个标志对象，包含流的状态信息，如是否发生溢出/下溢。
"""
# 在这里处理音频数据
pass

“`

• indata: 如果流配置了输入通道，这是一个 NumPy 数组，形状为 (frames, input_channels)，包含从输入设备读取的最新音频数据。
• outdata: 如果流配置了输出通道，这是一个 NumPy 数组，形状为 (frames, output_channels)。你需要将要播放的音频数据写入这个数组。
• frames: 当前数据块的大小，等于你创建流时指定的 blocksize。
• time: 提供了当前数据块的时间信息，对于需要精确同步的应用可能有用。
• status: 非常重要！这是一个标志对象，用来指示流在处理当前数据块时是否发生了错误，例如输入缓冲区溢出 (status.input_overflow) 或输出缓冲区下溢 (status.output_underflow)。在回调函数中 务必 检查 status 对象，并在发生错误时进行处理（例如打印警告）。

5.3 回调模式示例：简单的声音输入到输出 (Pass-through)

这个例子演示如何创建一个同时具有输入和输出的音频流，将麦克风捕获的声音实时通过扬声器播放出来（监听功能）。

“`python
import sounddevice as sd
import numpy as np
import sys

音频参数

samplerate = 44100 # 采样率

blocksize = 1024 # 数据块大小 (可以尝试不同的值)

channels = (1, 1) # 输入1通道，输出1通道

dtype = ‘float32’ # 数据类型

设置默认参数，也可以在 Stream 中单独指定

sd.default.samplerate = samplerate

sd.default.blocksize = blocksize

sd.default.channels = channels

sd.default.dtype = dtype

print(f”正在监听（从默认输入到默认输出）。按 Ctrl+C 停止…”)

def audio_callback(indata, outdata, frames, time, status):
“””This is called (protested) for each audio block.”””
if status:
print(f”Stream status flags: {status}”, file=sys.stderr)

# 将输入数据直接复制到输出数据
# 如果 indata 或 outdata 的通道数不同，或者 dtype 不同，需要进行转换
# 这里假设输入输出通道数和 dtype 相同
if indata is not None and outdata is not None:
     # Ensure data type matches
    if indata.dtype != outdata.dtype or indata.shape[1] != outdata.shape[1]:
         # Simple pass-through assumes matching shapes and dtypes.
         # For different configurations, you might need resample/remap.
         print("Warning: Input and output shapes/dtypes do not match for simple pass-through.", file=sys.stderr)
         # For demonstration, let's try to make them compatible if possible
         # Example: mono input to stereo output (simplistic)
         if indata.shape[1] == 1 and outdata.shape[1] == 2:
             outdata[:, 0] = indata[:, 0]
             outdata[:, 1] = indata[:, 0] # Copy mono to both stereo channels
         elif indata.shape[1] == outdata.shape[1] and indata.dtype != outdata.dtype:
              # Try converting dtype
              outdata[:] = indata.astype(outdata.dtype)
         else:
             # Fallback or more complex conversion needed
             print("Complex conversion needed, skipping frame.", file=sys.stderr)
             outdata.fill(0) # Output silence
    else:
        outdata[:] = indata # Direct copy

elif indata is not None: # Only input, not configured for output
    # Handle input data (e.g., analyze it, save it)
    pass # For this example, we ignore input if no output is configured
elif outdata is not None: # Only output, not configured for input
    # Generate output data (e.g., read from a file, synthesize)
    outdata.fill(0) # Output silence if no input
else:
    # This case should not happen if channels are specified correctly
    pass

try:
# 创建并启动流
# channels=(input_channels, output_channels)
# dtype=(input_dtype, output_dtype)
with sd.Stream(callback=audio_callback, channels=(sd.default.channels[0], sd.default.channels[1]), dtype=(sd.default.dtype[0], sd.default.dtype[1])):
# 流在后台运行，主线程在这里等待中断信号
import threading
event = threading.Event()
event.wait() # 等待事件被设置 (例如，通过信号处理函数)

except KeyboardInterrupt:
print(“\n监听停止.”)
except Exception as e:
print(f”发生错误: {e}”, file=sys.stderr)

“`

重要提示： 上面的 audio_callback 函数中的数据复制 outdata[:] = indata 假设输入和输出的通道数 (shape[1]) 和数据类型 (dtype) 是匹配的。在实际应用中，你需要根据你的设备和需求检查并可能进行通道映射、数据类型转换等操作。为了演示简单，我加入了基本的形状/类型检查和一些简单的处理逻辑，但在更复杂的场景下，你需要更 robust 的处理。

这个例子使用了 threading.Event() 来让主线程等待，而不是直接退出，从而允许后台的音频流线程持续运行。当用户按下 Ctrl+C 时，KeyboardInterrupt 异常会被捕获，然后主程序退出，with sd.Stream(...) 块结束，流会自动停止和关闭。

5.4 回调模式示例：录制音频到队列

在回调函数中进行复杂的处理（如文件写入、网络传输等）可能会导致回调函数执行时间过长，从而错过 PortAudio 下一次调用回调的时间，导致音频下溢或溢出。更好的做法是，在回调函数中只进行少量快速的操作（如将数据放入一个队列），然后在主线程或另一个工作线程中从队列中取出数据进行处理。

这个例子演示如何使用队列在回调函数和主线程之间传递录制的音频数据。

“`python
import sounddevice as sd
import numpy as np
import queue
import sys
import threading

如果想保存到文件，需要 soundfile

import soundfile as sf

音频参数

samplerate = 44100 # 采样率
duration = 10.0 # 录制持续时间 (秒)
channels = 1 # 录制通道数
dtype = ‘float32’ # 数据类型

blocksize = 1024 # 数据块大小

队列用于在回调线程和主线程之间传递数据块

q = queue.Queue()

def audio_callback(indata, outdata, frames, time, status):
“””This is called (from a separate thread) for each audio block.”””
if status:
print(f”Stream status flags: {status}”, file=sys.stderr)

# 将输入数据块放入队列
# 使用 .copy() 确保数据被复制，而不是只传递引用，因为 indata 是临时的
q.put_nowait(indata.copy())

用于控制录制持续时间

event = threading.Event()

print(f”正在录制 {duration} 秒音频到队列。按 Ctrl+C 提前停止…”)

try:
# 创建只用于输入的流 (没有 outdata 参数)
with sd.InputStream(
samplerate=samplerate,
# blocksize=blocksize, # 可以省略让 PortAudio 自动选择
channels=channels,
dtype=dtype,
callback=audio_callback
):
# 计算需要接收的总帧数
total_frames_to_record = int(samplerate * duration)
recorded_frames = 0
all_recorded_data = [] # 用于存储所有录制到的数据块

    # 在主线程中从队列取出数据并处理
    while recorded_frames < total_frames_to_record and not event.is_set():
        try:
            # 从队列获取数据块，设置超时防止无限等待
            data_chunk = q.get(timeout=0.1)
            all_recorded_data.append(data_chunk)
            recorded_frames += len(data_chunk)
            # print(f"已接收数据块，总帧数: {recorded_frames}") # 可选：打印进度
        except queue.Empty:
            # 队列为空，继续等待或检查事件
            pass

    print("\n录制处理完成.")

except KeyboardInterrupt:
print(“\n录制被中断.”)
event.set() # 设置事件，通知主线程停止等待
except Exception as e:
print(f”发生错误: {e}”, file=sys.stderr)

finally:
# 流在 with 块结束时自动停止和关闭

# 将所有数据块拼接成一个 NumPy 数组
if all_recorded_data:
    final_recorded_data = np.concatenate(all_recorded_data, axis=0)
    print(f"总共录制到 {len(final_recorded_data)} 帧数据.")
    # 在这里你可以进一步处理 final_recorded_data
    # 例如播放或保存到文件
    # print("正在播放录制内容...")
    # sd.play(final_recorded_data, samplerate)
    # sd.wait()
    # print("播放录制内容完成.")

    # 保存到 WAV 文件 (需要 soundfile 库)
    # try:
    #     sf.write('recorded_from_queue.wav', final_recorded_data, samplerate)
    #     print("已保存到 recorded_from_queue.wav")
    # except Exception as e:
    #      print(f"保存文件出错: {e}", file=sys.stderr)
else:
    print("没有录制到任何数据.")

“`

这个例子展示了如何使用 queue.Queue 在 sounddevice 回调函数（在单独的线程中）和主线程之间安全地传递数据。回调函数只负责快速地将数据放入队列，主线程则负责从队列中取出数据并进行更耗时的操作（例如拼接数组、保存文件等）。这种模式是处理实时音频流的推荐方式。

5.5 回调模式示例：从文件播放音频

同样，播放一个长音频文件时，不能一次性将所有数据加载到内存并传给 sd.play() (内存可能不足)，也不能在回调函数中同步读取文件 (文件 I/O 可能阻塞)。正确的做法是，提前从文件中读取一部分数据放入缓冲区（例如一个队列），回调函数从缓冲区取出数据填充 outdata。

这里需要 soundfile 库来读取音频文件。

“`python
import sounddevice as sd
import numpy as np
import soundfile as sf # 需要安装 pip install soundfile
import queue
import sys
import threading

音频参数

可以根据文件信息设置或覆盖

samplerate = 44100

channels = 2

dtype = ‘float32’

blocksize = 1024

替换为你的音频文件路径

filename = ‘your_audio_file.wav’ # 请确保这个文件存在

队列用于在主线程（或读取线程）和回调线程之间传递数据块

q = queue.Queue()

用于通知流结束的事件

event = threading.Event()

加载音频文件并获取信息

try:
# sf.read 可以返回数据和采样率，并且可以分块读取
# 为了简化示例，这里先加载全部数据，但在生产环境中应使用 sf.SoundFile 进行分块读取
# 对于非常大的文件，需要另一个线程专门负责从文件读取并放入队列
print(f”正在加载音频文件: {filename}…”)
# sf.read 返回 (data, samplerate)
data, samplerate = sf.read(filename, dtype=’float32′)
channels = data.shape[1] if data.ndim > 1 else 1 # 获取文件通道数
file_info = sf.info(filename) # 获取更多文件信息
# 可以使用 file_info.channels, file_info.samplerate 等

print(f"文件加载完成。采样率: {samplerate} Hz, 通道数: {channels}, 帧数: {len(data)}")

# 在这个简单示例中，我们将所有数据一次性推入队列
# 对于大文件，应该分块推入
total_frames = len(data)
frame_index = 0 # 追踪当前播放到文件中的位置

# 将数据块分割并放入队列
# 假设 stream blocksize 是 1024，我们可以按这个大小分割
# 更灵活的方式是动态从文件中读取并放入队列
# 这里只是一个概念演示，实际应用需要更精细的控制

# 演示：简单的文件读取线程 (更适合大文件)
def file_reader_thread(filename, q, event, blocksize, dtype):
    try:
        with sf.SoundFile(filename, 'r') as f:
            while not event.is_set():
                # 每次读取一个数据块
                data_chunk = f.read(blocksize, dtype=dtype)
                if not data_chunk.shape[0]: # 文件读取完毕
                     # print("文件读取线程：文件读取完毕")
                     q.put_nowait(None) # 放入一个 sentinel 值表示结束
                     break
                q.put_nowait(data_chunk)
            # print("文件读取线程：退出")
    except Exception as e:
         print(f"文件读取线程出错: {e}", file=sys.stderr)
         q.put_nowait(None) # 确保流能结束
         event.set() # 通知主线程结束

# 在回调函数中，从队列取出数据并填充 outdata
def audio_callback(indata, outdata, frames, time, status):
    """Output audio data from the queue."""
    if status:
        print(f"Stream status flags: {status}", file=sys.stderr)

    try:
        # 从队列获取一个数据块
        data_chunk = q.get_nowait()

        if data_chunk is None: # 收到结束 sentinel 值
             # 文件播放完毕
             # print("音频回调：收到结束信号，流停止。")
             raise sd.CallbackStop # 抛出异常停止流

        # 确保数据块大小匹配，如果从文件读取的不足 frames，用静音填充
        # 理论上，如果文件读取线程和回调同步良好，每次读取的块大小应该就是 blocksize
        if len(data_chunk) < frames:
            # print(f"音频回调：数据不足，读取到 {len(data_chunk)} 帧，填充静音。")
            outdata[:len(data_chunk)] = data_chunk
            outdata[len(data_chunk):].fill(0) # 填充静音
        elif len(data_chunk) > frames:
             # 这种情况不应该发生，除非文件读取线程逻辑有问题
             print(f"音频回调：数据块过大 ({len(data_chunk)} 帧)，只使用前 {frames} 帧。", file=sys.stderr)
             outdata[:] = data_chunk[:frames]
        else:
             # 数据块大小正好匹配
             outdata[:] = data_chunk

    except queue.Empty:
        # 队列为空！发生输出下溢！填充静音，并打印警告
        print("音频回调：队列为空，发生下溢！填充静音。", file=sys.stderr)
        outdata.fill(0)
    except Exception as e:
         print(f"音频回调出错: {e}", file=sys.stderr)
         outdata.fill(0) # 出错时输出静音
         raise sd.CallbackAbort # 抛出异常终止流


print(f"正在播放文件 {filename}...")

# 创建并启动只用于输出的流
# channels 可以是一个整数或元组，用于输出时只关心第二个元素
# dtype 也是类似
stream_channels = channels # 使用文件通道数
stream_dtype = data.dtype # 使用文件数据类型
stream_blocksize = sd.default.blocksize if sd.default.blocksize else 1024 # 使用默认或指定一个 blocksize

# 启动文件读取线程
reader_thread = threading.Thread(target=file_reader_thread,
                                 args=(filename, q, event, stream_blocksize, stream_dtype))
reader_thread.start()

with sd.OutputStream(
    samplerate=samplerate,
    blocksize=stream_blocksize,
    channels=stream_channels,
    dtype=stream_dtype,
    callback=audio_callback,
    finished_callback=lambda: print("\n文件播放完成.", file=sys.stderr) # 流正常结束时调用
):
    # 主线程在这里等待事件被设置 (例如，文件读取线程完成并发送None，或用户按下Ctrl+C)
    event.wait()

# 等待文件读取线程结束
reader_thread.join()

except FileNotFoundError:
print(f”错误: 未找到文件 {filename}”, file=sys.stderr)
except sf.SoundFileError as e:
print(f”读取音频文件出错: {e}”, file=sys.stderr)
except KeyboardInterrupt:
print(“\n播放被中断.”)
event.set() # 通知文件读取线程停止
except Exception as e:
print(f”发生错误: {e}”, file=sys.stderr)

“`

这个更完整的例子演示了播放音频文件时如何使用一个单独的线程来读取文件并将数据放入队列，而音频回调函数则从队列中取出数据进行播放。当文件读取完毕时，文件读取线程会向队列放入一个特殊值 (None) 来通知回调函数没有更多数据了，回调函数收到这个值后抛出 sd.CallbackStop 异常来停止流。sounddevice 的 finished_callback 在流正常停止（如播放完毕）时会被调用。

对于超大文件，你可能不会一次性加载到内存，而是使用 soundfile.SoundFile 对象分块读取。上面的例子已经包含了使用 sf.SoundFile 在单独线程中读取的逻辑草图。

6. 流状态和错误处理

在音频流的回调函数中，status 参数包含了流当前的状态信息。检查这个对象对于诊断问题非常重要。

status 对象是一个 sounddevice.Flags 类的实例，它有一些布尔属性：

• status.input_overflow: 输入缓冲区溢出。意味着输入设备生成的数据比你的回调函数处理得快，数据丢失了。通常是因为你的回调函数执行时间过长。
• status.output_underflow: 输出缓冲区下溢。意味着输出设备需要数据时，你的回调函数没有及时提供，导致播放出现中断或静音。通常也是因为回调函数执行时间过长，或者数据源（如队列）没有及时填充。
• status.input_late: 输入数据到达得比预期晚。
• status.output_late: 输出数据被提供得比预期晚。
• status.dropped_frames: PortAudio 因为某种原因跳过了一些输入帧。
• status.cpu_load_warning: CPU 负载过高，可能导致音频问题。
• status.prime_output_buffers_on_start: 流启动时输出缓冲区被填充了。
• status.exclusive_mode: 流正在独占模式下运行。

你应该总是在回调函数的开头检查 if status:，并在发生错误时打印警告或采取其他措施。

此外，在创建和启动流的 with 块或 stream.start() 调用外部，可以使用标准的 try...except 块来捕获 sounddevice 相关的异常，例如设备不可用、参数错误等。

7. 与 soundfile 协同工作

如前所述，sounddevice 负责与音频设备交互，而 soundfile 库则负责读写多种音频文件格式（如 WAV, FLAC, OGG等）。它们是处理音频文件 I/O 和设备 I/O 的黄金搭档。

简单的文件播放：

“`python
import sounddevice as sd
import soundfile as sf
import sys

filename = ‘your_audio_file.wav’ # 替换为你的文件

try:
# 使用 soundfile 读取整个文件 (小文件适用)
data, samplerate = sf.read(filename, dtype=’float32′)

print(f"正在播放文件: {filename}")
# 使用 sounddevice 播放读取到的数据
sd.play(data, samplerate)

# 等待播放完成
sd.wait()
print("播放完成.")

except FileNotFoundError:
print(f”错误: 未找到文件 {filename}”, file=sys.stderr)
except sf.SoundFileError as e:
print(f”读取音频文件出错: {e}”, file=sys.stderr)
except Exception as e:
print(f”发生错误: {e}”, file=sys.stderr)
“`

简单的文件录制：

“`python
import sounddevice as sd
import soundfile as sf
import numpy as np
import sys

filename = ‘my_recording.wav’
duration = 5.0 # 录制持续时间 (秒)
samplerate = 44100 # 采样率
channels = 1 # 通道数
dtype = ‘float32’ # 数据类型

try:
print(f”正在录制 {duration} 秒音频到 {filename}…”)
# 使用 sounddevice 录制音频
# sd.rec 是阻塞的，会等待录制完成
recorded_data = sd.rec(int(samplerate * duration),
samplerate=samplerate,
channels=channels,
dtype=dtype)

sd.wait() # 等待录制完成
print("录制完成.")

# 使用 soundfile 将录制到的数据保存到文件
sf.write(filename, recorded_data, samplerate)
print(f"已保存到 {filename}")

except Exception as e:
print(f”发生错误: {e}”, file=sys.stderr)

“`

对于大型文件和实时流处理，如前面播放文件到队列和录制到队列的例子所示，soundfile.SoundFile 对象的 read() 和 write() 方法可以在循环或单独线程中与 sounddevice.Stream 结合使用。

8. 更多高级话题和应用

• 多设备同时使用： 可以创建多个 sd.Stream 对象，分别对应不同的设备，实现更复杂的路由或同时进行输入/输出。
• 音频效果处理： 在回调函数中，你可以对 indata 进行各种数字信号处理 (DSP) 操作，然后将结果写入 outdata，实现实时音频效果，如均衡器、混响、失真等。
• 音频分析： 在回调函数中，将 indata 数据传递给分析库（如 NumPy, SciPy, librosa）进行频谱分析、特征提取等。
• 语音应用： 结合语音识别 (ASR) 或语音合成 (TTS) 库，构建实时语音交互应用。
• 音乐合成： 在回调函数中动态生成音频波形数据并写入 outdata，实现软件合成器。

9. 总结

sounddevice 库为 Python 开发者提供了一个强大、灵活且易于使用的接口，用于直接与音频输入和输出设备进行交互。通过简洁的函数和强大的流（Stream）机制，你可以轻松实现音频的播放、录制以及复杂的实时音频处理任务。

本文详细介绍了 sounddevice 的安装、设备管理、简单的阻塞式播放和录制，以及最核心的音频流（Stream）机制，特别是基于回调函数的实时处理模式。我们还探讨了如何结合 soundfile 库进行文件读写，并通过队列机制解决了回调函数与主线程之间的数据同步问题。

掌握 sounddevice 将为你打开音频处理的大门，无论是开发一个简单的音频工具，还是构建一个复杂的实时音频应用，它都能成为你的得力助手。现在，是时候动手尝试，用 Python 和 sounddevice 创造属于你的声音世界了！

---