audio_codebuddy/hardware-integration/pc-simulator/recorder.py

# 录音模块（电脑麦克风）

import pyaudio
import wave
import numpy as np
from config import Config


def record_until_silence(max_seconds=10, silence_threshold=500, silence_duration_ms=800):
    """
    录音直到检测到静音

    返回：bytes (WAV 格式音频数据)
    """
    CHUNK = 1024
    FORMAT = pyaudio.paInt16
    CHANNELS = Config.AUDIO_CHANNELS
    RATE = Config.AUDIO_SAMPLE_RATE

    audio = pyaudio.PyAudio()

    # 列出输入设备
    print('可用输入设备:')
    for i in range(audio.get_device_count()):
        info = audio.get_device_info_by_index(i)
        if info['maxInputChannels'] > 0:
            print(f'  [{i}] {info["name"]}')

    stream = audio.open(
        format=FORMAT,
        channels=CHANNELS,
        rate=RATE,
        input=True,
        frames_per_buffer=CHUNK,
    )

    print('开始录音...（说完后会自动停止）')

    frames = []
    silent_chunks = 0
    chunks_per_silence = int(RATE * silence_duration_ms / 1000 / CHUNK)
    max_chunks = int(RATE * max_seconds / CHUNK)

    for _ in range(max_chunks):
        try:
            data = stream.read(CHUNK, exception_on_overflow=False)
        except Exception as e:
            print(f'录音错误: {e}')
            break

        frames.append(data)

        # 检测音量
        audio_data = np.frombuffer(data, dtype=np.int16)
        volume = np.abs(audio_data).mean()

        if volume < silence_threshold:
            silent_chunks += 1
            if silent_chunks >= chunks_per_silence and len(frames) > chunks_per_silence:
                print('检测到静音，停止录音')
                break
        else:
            silent_chunks = 0

    stream.stop_stream()
    stream.close()
    audio.terminate()

    # 编码为 WAV
    wav_data = encode_wav(frames, RATE, CHANNELS)
    print(f'录音完成：{len(wav_data) / 1024:.1f} KB')
    return wav_data


def encode_wav(frames, rate, channels):
    """编码为 WAV bytes"""
    import io
    buffer = io.BytesIO()
    with wave.open(buffer, 'wb') as wf:
        wf.setnchannels(channels)
        wf.setsampwidth(2)
        wf.setframerate(rate)
        wf.writeframes(b''.join(frames))
    return buffer.getvalue()


if __name__ == '__main__':
    wav_data = record_until_silence()
    with open('test_recording.wav', 'wb') as f:
        f.write(wav_data)
    print('已保存为 test_recording.wav')