| import os |
| import time |
| import yaml |
| import json |
| from decouple import config as decouple_config |
| from groq import Groq |
| |
| import soundfile as sf |
| import numpy as np |
| import noisereduce as nr |
| |
| import queue |
| import sys |
| import datetime |
| |
| import os |
| import tempfile |
|
|
| class AudioProcessor: |
| def __init__(self): |
| """ |
| Initialize the AudioProcessor by loading configuration and setting up the Groq client. |
| """ |
| |
| |
| base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) |
| config_path = os.path.join(base_dir, "config", "config.yml") |
| |
| |
| with open(config_path, "r") as file: |
| self.config_data = yaml.safe_load(file) |
| |
| |
| self.groq_api_key = os.getenv("GROQ_API_KEY") |
| self.groq_model = self.config_data["groq"]["model"] |
| |
| |
| if not self.groq_api_key: |
| raise ValueError("Groq API key not found in environment variables. " |
| "Set GROQ_API_KEY in Hugging Face secrets.") |
| |
| if not self.groq_model: |
| raise ValueError("Groq model not configured in config.yml") |
| |
| |
| self.groq_client = Groq(api_key=self.groq_api_key) |
| |
| |
| 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") |
| |
| |
| 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): |
| |
| 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).") |
| |
| |
| audio_queue = queue.Queue() |
| |
| def callback(indata, frames, time_info, status): |
| if status: |
| print(status) |
| audio_queue.put(indata.copy()) |
| |
| |
| stream = sd.InputStream(samplerate=fs, channels=channels, callback=callback) |
| stream.start() |
| |
| |
| 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'): |
| |
| while keyboard.is_pressed('q'): |
| time.sleep(0.1) |
| break |
| time.sleep(0.1) |
| stream.stop() |
| |
| |
| 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 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] |
| |
| |
| denoised_audio = nr.reduce_noise(y=audio, sr=fs, y_noise=noise_sample, prop_decrease=1.0) |
| |
| |
| 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) |
| |
| temp_wav_fd, temp_wav_path = tempfile.mkstemp(suffix=".wav") |
| os.close(temp_wav_fd) |
| 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. |
| """ |
| |
| 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 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} |
| } |
| |
| |
| 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) |
| |
| |
| with open(self.log_file, "w", encoding="utf-8") as f: |
| json.dump(logs, f, indent=4) |
| print("Conversation logged.") |
|
|
| |
| 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="") |
|
|
