fix : conditional stt & tts activation

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_FOUND = True
-import pyaudio
+
+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 = None
-        self.thread = threading.Thread(target=self._record, daemon=True)
+        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()
 

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
+IMPORT_FOUND = True
-from IPython.display import display, Audio
+try:
-import soundfile as sf
+    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 = """