remode sd and pydub

app/stt.py

@@ -1,244 +1,244 @@
-import os
-import time
-import yaml
-import json
-from decouple import config as decouple_config  # if still using .env for other values
-from groq import Groq
-import sounddevice as sd
-import soundfile as sf
-import numpy as np
-import noisereduce as nr
-import keyboard  
-import queue
-import sys
-import datetime
-from pydub import AudioSegment
-import os
-import tempfile
-
-class AudioProcessor:
-    def __init__(self):
-        """
-        Initialize the AudioProcessor by loading configuration and setting up the Groq client.
-        """
-        # Compute the absolute path to the config file located at <project_root>/config/config.yml.
-        base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
-        config_path = os.path.join(base_dir, "config", "config.yml")
-        
-        # Load the configuration from the YAML file.
-        with open(config_path, "r") as file:
-            self.config_data = yaml.safe_load(file)
-        
-        # Access the Groq settings from the configuration.
-        self.groq_api_key = self.config_data["groq"]["api_key"]
-        self.groq_model = self.config_data["groq"]["model"]
-        
-        # Initialize the Groq client.
-        self.groq_client = Groq(api_key=self.groq_api_key)
-        
-        # Store default audio settings from the config.
-        self.sample_rate = self.config_data.get("audio", {}).get("sample_rate", 44100)
-        self.channels = self.config_data.get("audio", {}).get("channels", 1)
-        self.default_duration = self.config_data.get("audio", {}).get("duration", 5)
-        self.default_output_file = self.config_data.get("audio", {}).get("output_file", "denoised_output.wav")
-        
-        # Setup logging folder (assuming logs folder is at project_root/logs)
-        base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
-        self.logs_folder = os.path.join(base_dir, "logs")
-        os.makedirs(self.logs_folder, exist_ok=True)
-        self.log_file = os.path.join(self.logs_folder, "conversation_logs.json")
-    
-    def wait_for_key(self, target_key):
-        """
-        Wait for the target key to be pressed. If 's' is pressed, exit the program.
-        """
-        while True:
-            if keyboard.is_pressed('s'):
-                print("Stop key 's' pressed. Exiting program.")
-                sys.exit(0)
-            if keyboard.is_pressed(target_key):
-                # Debounce: wait until key is released.
-                while keyboard.is_pressed(target_key):
-                    time.sleep(0.1)
-                break
-            time.sleep(0.1)
-
-    def record_and_denoise(self, fs=None, channels=None, filename=None):
-        """
-        Waits for the user to press 'q' to start recording and again 'q' to stop.
-        Also checks if 's' is pressed to exit immediately.
-        Then it performs noise reduction and saves the cleaned audio.
-        
-        Parameters:
-            fs (int): Sampling rate (e.g., 44100 Hz). Defaults to configuration value.
-            channels (int): Number of audio channels (1 for mono, 2 for stereo). Defaults to configuration value.
-            filename (str): Output filename for the denoised audio. Defaults to configuration value.
-        
-        Returns:
-            np.ndarray: The denoised audio signal.
-            int: The sampling rate.
-        """
-        fs = fs if fs is not None else self.sample_rate
-        channels = channels if channels is not None else self.channels
-        filename = filename if filename is not None else self.default_output_file
-
-        print("Press 'q' to start recording (or 's' to exit).")
-        self.wait_for_key('q')
-        print("Recording started. Press 'q' again to stop (or 's' to exit).")
-        
-        # Set up a queue to collect recorded audio data.
-        audio_queue = queue.Queue()
-        
-        def callback(indata, frames, time_info, status):
-            if status:
-                print(status)
-            audio_queue.put(indata.copy())
-        
-        # Start the input stream.
-        stream = sd.InputStream(samplerate=fs, channels=channels, callback=callback)
-        stream.start()
-        
-        # Instead of a single blocking wait, check in a loop.
-        while True:
-            if keyboard.is_pressed('s'):
-                print("Stop key 's' pressed during recording. Exiting program.")
-                stream.stop()
-                sys.exit(0)
-            if keyboard.is_pressed('q'):
-                # Debounce the key press.
-                while keyboard.is_pressed('q'):
-                    time.sleep(0.1)
-                break
-            time.sleep(0.1)
-        stream.stop()
-        
-        # Combine all audio chunks from the queue.
-        audio_chunks = []
-        while not audio_queue.empty():
-            audio_chunks.append(audio_queue.get())
-        if audio_chunks:
-            audio = np.concatenate(audio_chunks, axis=0)
-        else:
-            print("No audio was recorded.")
-            return None, fs
-        
-        # If mono, remove the extra dimension.
-        if channels == 1 and audio.ndim > 1:
-            audio = np.squeeze(audio)
-        
-        print("Recording stopped. Reducing background noise...")
-        noise_duration = 0.5
-        noise_samples = int(noise_duration * fs)
-        noise_sample = audio if audio.shape[0] < noise_samples else audio[:noise_samples]
-        
-        # Perform noise reduction.
-        denoised_audio = nr.reduce_noise(y=audio, sr=fs, y_noise=noise_sample, prop_decrease=1.0)
-        
-        # Save the denoised audio to a file.
-        sf.write(filename, denoised_audio, fs)
-        print(f"Denoised audio saved as '{filename}'")
-        
-        return denoised_audio, fs
-
-    def prepare_audio(self, file_path):
-        """
-        Check if the provided audio file is in WAV format.
-        If not, convert it to WAV and return the path to the converted file.
-        """
-        ext = os.path.splitext(file_path)[1].lower()
-        if ext != ".wav":
-            try:
-                audio = AudioSegment.from_file(file_path)
-                # Create a temporary file to hold the WAV data.
-                temp_wav_fd, temp_wav_path = tempfile.mkstemp(suffix=".wav")
-                os.close(temp_wav_fd)  # Close the file descriptor.
-                audio.export(temp_wav_path, format="wav")
-                print(f"Converted {file_path} to WAV format at {temp_wav_path}")
-                return temp_wav_path
-            except Exception as e:
-                print(f"Error converting audio file: {e}")
-                return file_path
-        return file_path
-
-    def transcribe_audio(self, file_path, language="en", prompt=None, temperature=0.0):
-        """
-        Transcribe an audio file using the Groq API with the specified Whisper model.
-        This function now accepts any voice input format by converting non-WAV files to WAV.
-        
-        Parameters:
-            file_path (str): Path to the audio file.
-            language (str): Language code (default "en").
-            prompt (str): Optional transcription prompt.
-            temperature (float): Temperature parameter for transcription (default 0.0).
-            
-        Returns:
-            str: The transcribed text.
-        """
-        # Ensure the audio is in WAV format.
-        prepared_file = self.prepare_audio(file_path)
-        with open(prepared_file, "rb") as f:
-            audio_bytes = f.read()
-        
-        transcription = self.groq_client.audio.transcriptions.create(
-            file=(os.path.basename(prepared_file), audio_bytes),
-            model=self.groq_model,
-            prompt=prompt,
-            response_format="json",
-            language=language,
-            temperature=temperature
-        )
-        
-        # If a temporary file was created, remove it.
-        if prepared_file != file_path:
-            try:
-                os.remove(prepared_file)
-            except Exception as e:
-                print(f"Error removing temporary file: {e}")
-        
-        return transcription.text
-
-    def log_conversation(self, user_text, bot_text=""):
-        """
-        Append a conversation entry to the JSON log file.
-        Each entry has the following format:
-        {
-            "timestamp": "YYYY-MM-DD HH:MM:SS",
-            "User": { "text": "User's transcribed text" },
-            "bot": { "text": "Bot's reply text" }
-        }
-        
-        Parameters:
-            user_text (str): The text from the user (transcribed).
-            bot_text (str): The bot's reply text (empty until LLM integration).
-        """
-        entry = {
-            "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-            "User": {"text": user_text},
-            "bot": {"text": bot_text}
-        }
-        
-        # Load existing logs if the file exists, else start with an empty list.
-        if os.path.exists(self.log_file):
-            try:
-                with open(self.log_file, "r", encoding="utf-8") as f:
-                    logs = json.load(f)
-            except json.JSONDecodeError:
-                logs = []
-        else:
-            logs = []
-        
-        logs.append(entry)
-        
-        # Save the updated logs.
-        with open(self.log_file, "w", encoding="utf-8") as f:
-            json.dump(logs, f, indent=4)
-        print("Conversation logged.")
-
-# Example usage:
-if __name__ == "__main__":
-    processor = AudioProcessor()
-    denoised_audio, fs = processor.record_and_denoise()
-    if denoised_audio:
-        transcription = processor.transcribe_audio(processor.default_output_file, language="en")
-        print("Transcribed Text:\n", transcription)
-        processor.log_conversation(transcription, bot_text="")
+import os
+import time
+import yaml
+import json
+from decouple import config as decouple_config  # if still using .env for other values
+from groq import Groq
+# import sounddevice as sd
+import soundfile as sf
+import numpy as np
+import noisereduce as nr
+import keyboard  
+import queue
+import sys
+import datetime
+# from pydub import AudioSegment
+import os
+import tempfile
+
+class AudioProcessor:
+    def __init__(self):
+        """
+        Initialize the AudioProcessor by loading configuration and setting up the Groq client.
+        """
+        # Compute the absolute path to the config file located at <project_root>/config/config.yml.
+        base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+        config_path = os.path.join(base_dir, "config", "config.yml")
+        
+        # Load the configuration from the YAML file.
+        with open(config_path, "r") as file:
+            self.config_data = yaml.safe_load(file)
+        
+        # Access the Groq settings from the configuration.
+        self.groq_api_key = self.config_data["groq"]["api_key"]
+        self.groq_model = self.config_data["groq"]["model"]
+        
+        # Initialize the Groq client.
+        self.groq_client = Groq(api_key=self.groq_api_key)
+        
+        # Store default audio settings from the config.
+        self.sample_rate = self.config_data.get("audio", {}).get("sample_rate", 44100)
+        self.channels = self.config_data.get("audio", {}).get("channels", 1)
+        self.default_duration = self.config_data.get("audio", {}).get("duration", 5)
+        self.default_output_file = self.config_data.get("audio", {}).get("output_file", "denoised_output.wav")
+        
+        # Setup logging folder (assuming logs folder is at project_root/logs)
+        base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+        self.logs_folder = os.path.join(base_dir, "logs")
+        os.makedirs(self.logs_folder, exist_ok=True)
+        self.log_file = os.path.join(self.logs_folder, "conversation_logs.json")
+    
+    def wait_for_key(self, target_key):
+        """
+        Wait for the target key to be pressed. If 's' is pressed, exit the program.
+        """
+        while True:
+            if keyboard.is_pressed('s'):
+                print("Stop key 's' pressed. Exiting program.")
+                sys.exit(0)
+            if keyboard.is_pressed(target_key):
+                # Debounce: wait until key is released.
+                while keyboard.is_pressed(target_key):
+                    time.sleep(0.1)
+                break
+            time.sleep(0.1)
+
+    def record_and_denoise(self, fs=None, channels=None, filename=None):
+        """
+        Waits for the user to press 'q' to start recording and again 'q' to stop.
+        Also checks if 's' is pressed to exit immediately.
+        Then it performs noise reduction and saves the cleaned audio.
+        
+        Parameters:
+            fs (int): Sampling rate (e.g., 44100 Hz). Defaults to configuration value.
+            channels (int): Number of audio channels (1 for mono, 2 for stereo). Defaults to configuration value.
+            filename (str): Output filename for the denoised audio. Defaults to configuration value.
+        
+        Returns:
+            np.ndarray: The denoised audio signal.
+            int: The sampling rate.
+        """
+        fs = fs if fs is not None else self.sample_rate
+        channels = channels if channels is not None else self.channels
+        filename = filename if filename is not None else self.default_output_file
+
+        print("Press 'q' to start recording (or 's' to exit).")
+        self.wait_for_key('q')
+        print("Recording started. Press 'q' again to stop (or 's' to exit).")
+        
+        # Set up a queue to collect recorded audio data.
+        audio_queue = queue.Queue()
+        
+        def callback(indata, frames, time_info, status):
+            if status:
+                print(status)
+            audio_queue.put(indata.copy())
+        
+        # Start the input stream.
+        stream = sd.InputStream(samplerate=fs, channels=channels, callback=callback)
+        stream.start()
+        
+        # Instead of a single blocking wait, check in a loop.
+        while True:
+            if keyboard.is_pressed('s'):
+                print("Stop key 's' pressed during recording. Exiting program.")
+                stream.stop()
+                sys.exit(0)
+            if keyboard.is_pressed('q'):
+                # Debounce the key press.
+                while keyboard.is_pressed('q'):
+                    time.sleep(0.1)
+                break
+            time.sleep(0.1)
+        stream.stop()
+        
+        # Combine all audio chunks from the queue.
+        audio_chunks = []
+        while not audio_queue.empty():
+            audio_chunks.append(audio_queue.get())
+        if audio_chunks:
+            audio = np.concatenate(audio_chunks, axis=0)
+        else:
+            print("No audio was recorded.")
+            return None, fs
+        
+        # If mono, remove the extra dimension.
+        if channels == 1 and audio.ndim > 1:
+            audio = np.squeeze(audio)
+        
+        print("Recording stopped. Reducing background noise...")
+        noise_duration = 0.5
+        noise_samples = int(noise_duration * fs)
+        noise_sample = audio if audio.shape[0] < noise_samples else audio[:noise_samples]
+        
+        # Perform noise reduction.
+        denoised_audio = nr.reduce_noise(y=audio, sr=fs, y_noise=noise_sample, prop_decrease=1.0)
+        
+        # Save the denoised audio to a file.
+        sf.write(filename, denoised_audio, fs)
+        print(f"Denoised audio saved as '{filename}'")
+        
+        return denoised_audio, fs
+
+    def prepare_audio(self, file_path):
+        """
+        Check if the provided audio file is in WAV format.
+        If not, convert it to WAV and return the path to the converted file.
+        """
+        ext = os.path.splitext(file_path)[1].lower()
+        if ext != ".wav":
+            try:
+                audio = AudioSegment.from_file(file_path)
+                # Create a temporary file to hold the WAV data.
+                temp_wav_fd, temp_wav_path = tempfile.mkstemp(suffix=".wav")
+                os.close(temp_wav_fd)  # Close the file descriptor.
+                audio.export(temp_wav_path, format="wav")
+                print(f"Converted {file_path} to WAV format at {temp_wav_path}")
+                return temp_wav_path
+            except Exception as e:
+                print(f"Error converting audio file: {e}")
+                return file_path
+        return file_path
+
+    def transcribe_audio(self, file_path, language="en", prompt=None, temperature=0.0):
+        """
+        Transcribe an audio file using the Groq API with the specified Whisper model.
+        This function now accepts any voice input format by converting non-WAV files to WAV.
+        
+        Parameters:
+            file_path (str): Path to the audio file.
+            language (str): Language code (default "en").
+            prompt (str): Optional transcription prompt.
+            temperature (float): Temperature parameter for transcription (default 0.0).
+            
+        Returns:
+            str: The transcribed text.
+        """
+        # Ensure the audio is in WAV format.
+        prepared_file = self.prepare_audio(file_path)
+        with open(prepared_file, "rb") as f:
+            audio_bytes = f.read()
+        
+        transcription = self.groq_client.audio.transcriptions.create(
+            file=(os.path.basename(prepared_file), audio_bytes),
+            model=self.groq_model,
+            prompt=prompt,
+            response_format="json",
+            language=language,
+            temperature=temperature
+        )
+        
+        # If a temporary file was created, remove it.
+        if prepared_file != file_path:
+            try:
+                os.remove(prepared_file)
+            except Exception as e:
+                print(f"Error removing temporary file: {e}")
+        
+        return transcription.text
+
+    def log_conversation(self, user_text, bot_text=""):
+        """
+        Append a conversation entry to the JSON log file.
+        Each entry has the following format:
+        {
+            "timestamp": "YYYY-MM-DD HH:MM:SS",
+            "User": { "text": "User's transcribed text" },
+            "bot": { "text": "Bot's reply text" }
+        }
+        
+        Parameters:
+            user_text (str): The text from the user (transcribed).
+            bot_text (str): The bot's reply text (empty until LLM integration).
+        """
+        entry = {
+            "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            "User": {"text": user_text},
+            "bot": {"text": bot_text}
+        }
+        
+        # Load existing logs if the file exists, else start with an empty list.
+        if os.path.exists(self.log_file):
+            try:
+                with open(self.log_file, "r", encoding="utf-8") as f:
+                    logs = json.load(f)
+            except json.JSONDecodeError:
+                logs = []
+        else:
+            logs = []
+        
+        logs.append(entry)
+        
+        # Save the updated logs.
+        with open(self.log_file, "w", encoding="utf-8") as f:
+            json.dump(logs, f, indent=4)
+        print("Conversation logged.")
+
+# Example usage:
+if __name__ == "__main__":
+    processor = AudioProcessor()
+    denoised_audio, fs = processor.record_and_denoise()
+    if denoised_audio:
+        transcription = processor.transcribe_audio(processor.default_output_file, language="en")
+        print("Transcribed Text:\n", transcription)
+        processor.log_conversation(transcription, bot_text="")