跨平台虚拟扬声器音频捕获服务 — 捕获系统音频，转发到真实扬声器，录制为 WAV 文件，并提供音频流用于 AI 推理。
支持 macOS、Windows、Linux 三大平台。
系统音频 → 虚拟音频设备 (Driver) → FakeSpeaker 引擎
├→ 真实扬声器 (透传，可听到声音)
├→ WAV 文件录制 (按需启停)
├→ AudioStream (供 AI 推理消费，同步)
├→ AsyncAudioStream (供 AI 推理消费，异步)
└→ 自定义回调 (同步或 async 协程)
sounddevice C 回调 → _post_queue (Queue) → _post_worker 线程
→ SharedAudioBuffer.publish()
├→ 写入环形缓冲区 slot (无锁)
├→ _cond.notify_all() (同步消费者)
└→ call_soon_threadsafe() (异步消费者)
同步消费者 (AudioStream): _cond.wait() → _slots[seq % max]
异步消费者 (AsyncAudioStream): try_read_at() → _slots[seq % max] (完全无锁)
SharedAudioBuffer 使用固定大小环形缓冲区 + seqlock 验证，try_read_at() 读取路径
在 CPython GIL 下完全无锁，不与 publish() 热路径产生竞争。
Driver (抽象基类)
├── MacOSDriver — BlackHole / Soundflower
├── WindowsDriver — VB-CABLE / WASAPI Loopback (免安装)
└── LinuxDriver — PulseAudio / PipeWire null sink (自动创建)
安装 BlackHole 虚拟音频驱动：
brew install blackhole-2ch然后在 系统设置 > 声音 > 输出 中选择 "BlackHole 2ch"。
提示：如希望同时听到声音，可在「音频 MIDI 设置」中创建「多输出设备」， 同时勾选真实扬声器和 BlackHole 2ch。
方式 1 — VB-CABLE（推荐）：
从 vb-audio.com/Cable 下载安装，免费。 然后将系统音频输出设为 "CABLE Input"。
方式 2 — WASAPI Loopback（免安装）：
无需安装任何驱动。FakeSpeaker 会自动通过 WASAPI Loopback 捕获系统默认输出设备的音频。 缺点是不能在系统设置中看到单独的"虚拟扬声器"。
无需安装任何驱动。 FakeSpeaker 会自动通过 pactl 创建 PulseAudio / PipeWire 虚拟 null sink，
并在退出时自动清理。
确保系统有 PulseAudio 或 PipeWire（大多数现代发行版已预装）：
# Debian / Ubuntu
sudo apt install pulseaudio # 或 pipewire pipewire-pulse
# Fedora
sudo dnf install pulseaudio # 或 pipewire pipewire-pulseaudio
# Arch
sudo pacman -S pulseaudio # 或 pipewire pipewire-pulseuv syncuv run python main.py devices# 自动检测平台和虚拟设备
uv run python main.py start
# 通过设备索引指定输入/输出 (索引号见 devices 命令输出)
uv run python main.py start -i 0 -o 7
# 也支持设备名称（部分匹配）
uv run python main.py start -i "BlackHole" -o "MacBook"
# 启动时自动录制
uv run python main.py start --record output.wav
# 不转发到扬声器（静音模式）
uv run python main.py start --no-forward
# 使用 asyncio 引擎启动
uv run python main.py start --async启动后可使用以下命令：
| 命令 | 说明 |
|---|---|
r |
开始/停止录制（自动生成文件名） |
r <文件名> |
录制到指定文件 |
f |
切换音频转发开/关 |
s |
显示当前状态 |
q |
退出 |
from fakespeaker import FakeSpeaker
# 自动检测平台驱动，创建虚拟设备
with FakeSpeaker() as fs:
fs.start_recording("output.wav")
import time; time.sleep(10)
fs.stop_recording()from fakespeaker import FakeSpeaker
fs = FakeSpeaker(virtual_device="BlackHole 2ch") # macOS
fs = FakeSpeaker(virtual_device="CABLE Output") # Windows VB-CABLEfrom fakespeaker.drivers.linux import LinuxDriver
from fakespeaker import FakeSpeaker
driver = LinuxDriver()
fs = FakeSpeaker(driver=driver)
fs.start()
# ...
fs.stop() # 自动调用 driver.teardown() 清理虚拟设备from fakespeaker import FakeSpeaker
with FakeSpeaker() as fs:
stream = fs.create_stream()
# 方式 1: 逐块读取
for chunk in stream:
# chunk 是 numpy float32 数组, shape=(frames, channels)
result = your_ai_model.predict(chunk)
# 方式 2: 读取指定时长
audio_data = stream.read_seconds(5.0) # 读取 5 秒音频
# 方式 3: 读取所有可用数据
chunks = stream.read_all()提示：多个 stream 默认共享底层缓冲以减少拷贝，chunk 不要原地修改。
如需修改，请在创建时使用 copy_chunks=True。
import numpy as np
from fakespeaker import FakeSpeaker
def on_audio(data: np.ndarray):
"""每个音频块都会调用此函数（后台线程）。"""
rms = np.sqrt(np.mean(data ** 2))
if rms > 0.01:
print(f"检测到声音! RMS={rms:.4f}")
with FakeSpeaker() as fs:
fs.set_on_audio(on_audio)
input("Press Enter to stop...")from fakespeaker import FakeSpeaker
import threading
with FakeSpeaker() as fs:
# 每个消费者使用独立的流
stream_asr = fs.create_stream() # 语音识别
stream_vad = fs.create_stream() # 语音活动检测
def asr_worker():
for chunk in stream_asr:
text = speech_to_text(chunk)
def vad_worker():
for chunk in stream_vad:
is_speech = detect_voice(chunk)
t1 = threading.Thread(target=asr_worker, daemon=True)
t2 = threading.Thread(target=vad_worker, daemon=True)
t1.start()
t2.start()
t1.join()import asyncio
from fakespeaker.aio import AsyncFakeSpeaker
async def main():
async with AsyncFakeSpeaker() as fs:
stream = fs.create_stream()
# 方式 1: async for 逐块读取
async for chunk in stream:
result = await your_ai_model.predict(chunk)
# 方式 2: await 读取单个块
chunk = await stream.read(timeout=1.0)
# 方式 3: 读取指定时长
audio_data = await stream.read_seconds(5.0)
# 方式 4: 读取所有可用数据（非阻塞）
chunks = stream.read_all()
asyncio.run(main())import asyncio
from fakespeaker.aio import AsyncFakeSpeaker
async def main():
async with AsyncFakeSpeaker() as fs:
stream_asr = fs.create_stream()
stream_vad = fs.create_stream()
async def asr_worker():
async for chunk in stream_asr:
text = await speech_to_text(chunk)
async def vad_worker():
async for chunk in stream_vad:
is_speech = await detect_voice(chunk)
await asyncio.gather(asr_worker(), vad_worker())
asyncio.run(main())import asyncio
import numpy as np
from fakespeaker.aio import AsyncFakeSpeaker
async def main():
async with AsyncFakeSpeaker() as fs:
# 支持 async 回调函数，自动调度为 asyncio Task
async def on_audio(data: np.ndarray):
rms = np.sqrt(np.mean(data ** 2))
if rms > 0.01:
print(f"检测到声音! RMS={rms:.4f}")
fs.set_on_audio(on_audio)
await asyncio.sleep(30)
asyncio.run(main())from fakespeaker.drivers import get_driver
# 自动检测当前平台
driver = get_driver()
print(driver.platform_name) # e.g. "Linux (pipewire)"
print(driver.is_available()) # True / False
# 查看安装说明
if not driver.is_available():
print(driver.get_install_instructions())
# 手动管理虚拟设备生命周期
info = driver.setup()
print(info) # VirtualDeviceInfo(...)
# ... 使用 info.device_index 等 ...
driver.teardown() # 清理（Linux 会移除 null sink）| 参数 | 默认值 | 说明 |
|---|---|---|
virtual_device |
None (自动检测) | 虚拟设备名称/索引/AudioDevice/VirtualDeviceInfo |
output_device |
None (系统默认) | 真实输出设备 |
driver |
None (自动检测) | 平台驱动实例 |
sample_rate |
44100 | 采样率 (Hz) |
channels |
2 | 声道数 |
block_size |
1024 | 每块帧数 |
dtype |
float32 | 音频数据类型 |
forwarding |
True | 是否转发到真实扬声器 |
fakespeaker/
├── __init__.py # 包入口，导出公共 API
├── engine.py # 核心引擎：捕获、转发、录制、流
├── devices.py # 音频设备发现与管理
├── recorder.py # WAV 文件录制
├── stream.py # 无锁环形缓冲区 + 同步音频流
├── aio/
│ ├── __init__.py # 异步包入口，导出 AsyncFakeSpeaker, AsyncAudioStream
│ ├── engine.py # AsyncFakeSpeaker (异步引擎包装)
│ └── stream.py # AsyncAudioStream (async for / await 音频流)
└── drivers/
├── __init__.py # get_driver() 工厂函数
├── base.py # Driver 抽象基类 + VirtualDeviceInfo
├── macos.py # MacOSDriver (BlackHole)
├── windows.py # WindowsDriver (VB-CABLE / WASAPI)
└── linux.py # LinuxDriver (PulseAudio / PipeWire)