Spaces:
Running
Running
File size: 4,227 Bytes
25e6afd | 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 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | import speech_recognition as sr
import logging
import threading
import time
import queue
from typing import Optional, Callable
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("VoiceProcessor")
class VoiceProcessor:
def __init__(self):
"""Initialize voice components for backend use."""
# Voice recognition components
self.recognizer = sr.Recognizer()
self.recognizer.pause_threshold = 1.0 # Longer pauses allowed
self.recognizer.energy_threshold = 3000 # Better for different voice volumes
# Control variables
self.is_running = False
self.last_detected_text = "No text detected yet."
self.voice_lock = threading.Lock()
self.voice_queue = queue.Queue()
self.callback = None
def set_callback(self, callback: Callable[[str], None]):
"""Set a callback function to be called when text is detected."""
self.callback = callback
def voice_listener(self):
"""Voice listener thread function."""
try:
with sr.Microphone() as source:
self.recognizer.adjust_for_ambient_noise(source, duration=1)
logger.info("Voice listener started")
while self.is_running:
try:
logger.info("Listening for speech...")
audio = self.recognizer.listen(source, timeout=5, phrase_time_limit=7)
text = self.recognizer.recognize_google(audio)
logger.info(f"Detected speech: {text}")
with self.voice_lock:
self.last_detected_text = text
self.voice_queue.put(text)
# Call the callback if set
if self.callback:
self.callback(text)
except sr.WaitTimeoutError:
continue
except sr.UnknownValueError:
logger.info("Speech not recognized")
continue
except Exception as e:
logger.error(f"Voice recognition error: {e}")
continue
except Exception as e:
logger.error(f"Error in voice listener: {e}")
def start_listening(self):
"""Start the voice-to-text system."""
if self.is_running:
logger.info("Voice processor is already running")
return False
try:
self.is_running = True
# Start voice listener thread
listener_thread = threading.Thread(target=self.voice_listener, daemon=True)
listener_thread.start()
logger.info("Voice processor started")
return True
except Exception as e:
logger.error(f"Error starting voice processor: {e}")
self.is_running = False
return False
def stop_listening(self):
"""Stop the voice-to-text system."""
logger.info("Stopping voice processor...")
self.is_running = False
return True
def get_last_detected_text(self) -> str:
"""Get the last detected text in a thread-safe manner."""
with self.voice_lock:
return self.last_detected_text
def get_next_text(self, timeout: float = 1.0) -> Optional[str]:
"""Get the next detected text from the queue."""
try:
return self.voice_queue.get(timeout=timeout)
except queue.Empty:
return None
# For testing
if __name__ == "__main__":
processor = VoiceProcessor()
processor.start_listening()
try:
while True:
text = processor.get_next_text()
if text:
print(f"Detected: {text}")
time.sleep(0.1)
except KeyboardInterrupt:
processor.stop_listening()
print("Stopped") |