diff --git a/sources/speech_to_text.py b/sources/speech_to_text.py index a888326..9d6917b 100644 --- a/sources/speech_to_text.py +++ b/sources/speech_to_text.py @@ -3,11 +3,18 @@ from typing import List, Tuple, Type, Dict import queue import threading import numpy as np -import torch import time -from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline -import librosa -import pyaudio + +IMPORT_FOUND = True + +try: + import torch + import librosa + import pyaudio + from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline +except ImportError: + print(Fore.RED + "Speech To Text disabled." + Fore.RESET) + IMPORT_FOUND = False audio_queue = queue.Queue() done = False @@ -23,13 +30,18 @@ class AudioRecorder: self.chunk = chunk self.record_seconds = record_seconds self.verbose = verbose - self.audio = pyaudio.PyAudio() - self.thread = threading.Thread(target=self._record, daemon=True) + self.thread = None + self.audio = None + if IMPORT_FOUND: + self.audio = pyaudio.PyAudio() + self.thread = threading.Thread(target=self._record, daemon=True) def _record(self) -> None: """ Record audio from the microphone and add it to the audio queue. """ + if not IMPORT_FOUND: + return stream = self.audio.open(format=self.format, channels=self.channels, rate=self.rate, input=True, frames_per_buffer=self.chunk) if self.verbose: @@ -58,10 +70,14 @@ class AudioRecorder: def start(self) -> None: """Start the recording thread.""" + if not IMPORT_FOUND: + return self.thread.start() def join(self) -> None: """Wait for the recording thread to finish.""" + if not IMPORT_FOUND: + return self.thread.join() class Transcript: @@ -69,6 +85,9 @@ class Transcript: Transcript is a class that transcribes audio from the audio queue and adds it to the transcript. """ def __init__(self): + if not IMPORT_FOUND: + print(Fore.RED + "Transcript: Speech to Text is disabled." + Fore.RESET) + return self.last_read = None device = self.get_device() torch_dtype = torch.float16 if device == "cuda" else torch.float32 @@ -91,6 +110,8 @@ class Transcript: ) def get_device(self) -> str: + if not IMPORT_FOUND: + return "cpu" if torch.backends.mps.is_available(): return "mps" if torch.cuda.is_available(): @@ -108,6 +129,8 @@ class Transcript: def transcript_job(self, audio_data: np.ndarray, sample_rate: int = 16000) -> str: """Transcribe the audio data.""" + if not IMPORT_FOUND: + return "" if audio_data.dtype != np.float32: audio_data = audio_data.astype(np.float32) / np.iinfo(audio_data.dtype).max if len(audio_data.shape) > 1: @@ -122,6 +145,9 @@ class AudioTranscriber: AudioTranscriber is a class that transcribes audio from the audio queue and adds it to the transcript. """ def __init__(self, ai_name: str, verbose: bool = False): + if not IMPORT_FOUND: + print(Fore.RED + "AudioTranscriber: Speech to Text is disabled." + Fore.RESET) + return self.verbose = verbose self.ai_name = ai_name self.transcriptor = Transcript() @@ -152,6 +178,8 @@ class AudioTranscriber: """ Transcribe the audio data using AI stt model. """ + if not IMPORT_FOUND: + return global done if self.verbose: print(Fore.BLUE + "AudioTranscriber: Started processing..." + Fore.RESET) @@ -185,9 +213,13 @@ class AudioTranscriber: def start(self): """Start the transcription thread.""" + if not IMPORT_FOUND: + return self.thread.start() def join(self): + if not IMPORT_FOUND: + return """Wait for the transcription thread to finish.""" self.thread.join() diff --git a/sources/text_to_speech.py b/sources/text_to_speech.py index b57d782..310bfee 100644 --- a/sources/text_to_speech.py +++ b/sources/text_to_speech.py @@ -5,9 +5,14 @@ import subprocess from sys import modules from typing import List, Tuple, Type, Dict -from kokoro import KPipeline -from IPython.display import display, Audio -import soundfile as sf +IMPORT_FOUND = True +try: + from kokoro import KPipeline + from IPython.display import display, Audio + import soundfile as sf +except ImportError: + print("Speech synthesis disabled. Please install the kokoro package.") + IMPORT_FOUND = False if __name__ == "__main__": from utility import pretty_print, animate_thinking @@ -33,7 +38,7 @@ class Speech(): } self.pipeline = None self.language = language - if enable: + if enable and IMPORT_FOUND: self.pipeline = KPipeline(lang_code=self.lang_map[language]) self.voice = self.voice_map[language][voice_idx] self.speed = 1.2 @@ -57,7 +62,7 @@ class Speech(): sentence (str): The text to convert to speech. Will be pre-processed. voice_idx (int, optional): Index of the voice to use from the voice map. """ - if not self.pipeline: + if not self.pipeline or not IMPORT_FOUND: return if voice_idx >= len(self.voice_map[self.language]): pretty_print("Invalid voice number, using default voice", color="error") @@ -159,6 +164,7 @@ class Speech(): if __name__ == "__main__": # TODO add info message for cn2an, jieba chinese related import + IMPORT_FOUND = False sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) speech = Speech() tosay_en = """