连接蓝牙设备时实时生成并播放可变纯音的技术咨询

用PyAudio完全能实现你的需求，它支持实时音频流输出，刚好适配你动态生成正弦波的场景。下面是一套整合bleak、asyncio和PyAudio的完整方案：

1. 实时音频流输出：通过PyAudio的回调模式，每次音频设备需要样本时，实时生成对应频率和音量的正弦波，无需预存音频文件。
2. 线程安全的参数传递：由于bleak的蓝牙数据回调运行在asyncio事件循环线程，而PyAudio的音频回调在后台线程，用线程锁保护共享的频率、音量变量，避免竞态问题。
3. 异步整合：在asyncio事件循环中同时维护蓝牙连接和音频流，确保两者并行运行。

import pyaudio
import numpy as np
from bleak import BleakClient, BleakScanner
import asyncio
import threading

# 共享变量：当前频率、音量，加锁保护
current_freq = 440.0  # 默认A4音
current_volume = 0.5  # 0-1范围
param_lock = threading.Lock()

# 音频参数
SAMPLE_RATE = 44100
CHANNELS = 1
FORMAT = pyaudio.paInt16
CHUNK_SIZE = 1024  # 每次生成的样本数，越小延迟越低

def audio_callback(in_data, frame_count, time_info, status):
    """PyAudio回调函数，实时生成正弦波样本"""
    with param_lock:
        freq = current_freq
        volume = current_volume
    
    # 生成时间轴
    t = np.arange(frame_count) / SAMPLE_RATE
    # 生成正弦波：2π*freq*t，振幅映射到16位范围（-32768到32767）
    samples = (volume * 32767 * np.sin(2 * np.pi * freq * t)).astype(np.int16)
    # 转换为字节流返回
    return (samples.tobytes(), pyaudio.paContinue)

def bluetooth_data_handler(sender, data):
    """蓝牙传感器数据回调，解析并更新频率、音量"""
    # 这里根据你的传感器协议解析data，示例假设data是两个float：频率（4字节）+音量（4字节）
    # 实际要替换成你的传感器数据格式
    freq = np.frombuffer(data[:4], dtype=np.float32)[0]
    volume = np.frombuffer(data[4:], dtype=np.float32)[0]
    # 限制范围避免异常
    freq = max(20.0, min(20000.0, freq))
    volume = max(0.0, min(1.0, volume))
    
    with param_lock:
        global current_freq, current_volume
        current_freq = freq
        current_volume = volume
    print(f"更新参数：频率={freq:.1f}Hz，音量={volume:.2f}")

async def main():
    # 扫描并连接蓝牙设备（替换成你的传感器UUID或名称）
    device = await BleakScanner.find_device_by_name("你的传感器名称")
    if not device:
        print("未找到蓝牙设备")
        return
    
    # 初始化PyAudio
    p = pyaudio.PyAudio()
    stream = p.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=SAMPLE_RATE,
        output=True,
        frames_per_buffer=CHUNK_SIZE,
        stream_callback=audio_callback
    )
    
    try:
        async with BleakClient(device) as client:
            print(f"已连接到设备：{device.name}")
            # 订阅传感器的特征值（替换成你的传感器特征UUID）
            await client.start_notify("0000ffe1-0000-1000-8000-00805f9b34fb", bluetooth_data_handler)
            # 保持连接直到用户中断
            await asyncio.Future()  # 无限等待
    except asyncio.CancelledError:
        print("连接中断")
    finally:
        # 关闭音频流和PyAudio
        stream.stop_stream()
        stream.close()
        p.terminate()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("用户终止程序")

• 音频参数调整：CHUNK_SIZE越小，音频延迟越低，但CPU占用会略高，可根据你的设备性能调整（比如512或2048）。
• 传感器数据解析：bluetooth_data_handler里的解析逻辑必须和你的传感器协议匹配，比如如果传感器发的是整数，就用np.int16或np.int32解析。
• 线程安全：必须用param_lock保护current_freq和current_volume，否则可能出现参数更新不完整导致的音频异常。
• 替代工具对比：soundplay更适合播放预存音频文件，实时生成场景下PyAudio的灵活性和控制力更强，是更好的选择。

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