Spaces:
Build error
Build error
File size: 2,831 Bytes
f797e13 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
import sys
from dataclasses import dataclass, field
from queue import Queue
import numpy as np
import speech_recognition as sr
def get_microphone(default_microphone: str | None = "pulse", sample_rate: int = 16000) -> sr.Microphone:
"""Get the specified system microphone if available."""
# Important for linux users.
# Prevents permanent application hang and crash by using the wrong Microphone
if "linux" in sys.platform:
mic_name = default_microphone
if not mic_name or mic_name == "list":
mic_names = "\n".join(f"- {n}" for n in sr.Microphone.list_microphone_names())
err_msg = f"No microphone selected. Available microphone devices are:\n{mic_names}"
raise ValueError(err_msg)
else:
for index, name in enumerate(sr.Microphone.list_microphone_names()):
if mic_name in name:
return sr.Microphone(sample_rate=sample_rate, device_index=index)
return sr.Microphone(sample_rate=sample_rate)
def get_speech_recognizer(energy_threshold: int = 300) -> sr.Recognizer:
"""Set up a speech recognizer with a custom energy threshold."""
# We use SpeechRecognizer to record our audio because it has a nice feature where
# it can detect when speech ends.
speech_recognizer = sr.Recognizer()
speech_recognizer.energy_threshold = energy_threshold
# Definitely do this, dynamic energy compensation lowers the energy threshold dramatically
# to a point where the SpeechRecognizer never stops recording.
speech_recognizer.dynamic_energy_threshold = False
return speech_recognizer
def to_audio_array(audio_data: bytes) -> np.ndarray:
"""
Convert in-ram buffer to something the model can use directly without needing a temp file.
Convert data from 16 bit wide integers to floating point with a width of 32 bits.
Clamp the audio stream frequency to a PCM wavelength compatible default of 32768hz max.
"""
audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
return audio_np
def get_all_audio_queue(data_queue: Queue) -> bytes:
"""Returns all audio in the queue."""
audio_data = b"".join(data_queue.queue)
data_queue.queue.clear()
return audio_data
@dataclass
class AudioChunk:
start_time: float
end_time: float | None = None
audio_array: np.ndarray = field(default_factory=lambda: np.array([]))
@property
def duration(self) -> float | None:
return None if self.end_time is None else self.end_time - self.start_time
@property
def is_complete(self) -> bool:
return (self.end_time is not None) and (self.audio_array.size > 0)
def update_array(self, new_audio: np.ndarray) -> None:
self.audio_array = np.concatenate((self.audio_array, new_audio))
|