import os
import sys
import time
import wave
import tempfile
import threading
import torch
import pyaudiowpatch as pyaudio
from faster_whisper import WhisperModel
AUDIO_BUFFER = 5
def record_audio(p, device):
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
filename = f.name
wave_file = wave.open(filename, "wb")
wave_file.setnchannels(int(device["maxInputChannels"]))
wave_file.setsampwidth(p.get_sample_size(pyaudio.paInt16))
wave_file.setframerate(int(device["defaultSampleRate"]))
def callback(in_data, frame_count, time_info, status):
"""写入音频帧"""
wave_file.writeframes(in_data)
return (in_data, pyaudio.paContinue)
try:
stream = p.open(format=pyaudio.paInt16, channels=int(device["maxInputChannels"]), rate=int(device["defaultSampleRate"]), frames_per_buffer=1024, input=True, input_device_index=device["index"], stream_callback=callback,)
time.sleep(AUDIO_BUFFER)
except Exception as e:
print(f"录音出错:{e}")
finally:
if 'stream' in locals():
stream.stop_stream()
stream.close()
wave_file.close()
return filename
def whisper_audio(filename, model):
"""调用模型进行转录"""
try:
segments, info = model.transcribe(filename, beam_size=5, language="zh", vad_filter=True, vad_parameters=dict(min_silence_duration_ms=500))
for segment in segments:
print("[%.2fs -> %.2fs] %s" % (segment.start, segment.end, segment.text))
except Exception as e:
print(f"转录出错:{e}")
finally:
if os.path.exists(filename):
os.remove(filename)
def main():
print("正在加载 Whisper 模型...")
if torch.cuda.is_available():
device = "cuda"
compute_type = "float16"
print("使用 GPU (CUDA) 进行推理")
else:
device = "cpu"
compute_type = "int8"
print("使用 CPU 进行推理")
model_path = "large-v3"
try:
model = WhisperModel(model_path, device=device, compute_type=compute_type, local_files_only=True)
print("模型加载成功!")
except Exception as e:
print(f"模型加载失败:{e}")
return
with pyaudio.PyAudio() as p:
try:
default_mic = p.get_default_input_device_info()
print(f"\n当前使用的麦克风:{default_mic['name']} (Index: {default_mic['index']})")
print(f"采样率:{default_mic['defaultSampleRate']}, 通道数:{default_mic['maxInputChannels']}")
print("-" * 50)
print("开始持续录音 (按 Ctrl+C 停止)...")
while True:
filename = record_audio(p, default_mic)
thread = threading.Thread(target=whisper_audio, args=(filename, model))
thread.start()
except OSError:
print("未找到默认麦克风,请检查系统声音设置。")
except KeyboardInterrupt:
print("\n停止录音,程序退出。")
except Exception as e:
print(f"\n发生未知错误:{e}")
if __name__ == '__main__':
main()