File size: 3,969 Bytes
42f5a8f | 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 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | import os
from dotenv import load_dotenv
import wave
import pyaudio
from scipy.io import wavfile
import numpy as np
import whisper
from langchain.chains.llm import LLMChain
from langchain_core.prompts import PromptTemplate
from langchain_groq import ChatGroq
from gtts import gTTS
import pygame
from groq import Groq
load_dotenv()
groq_api_key = os.getenv("GROQ_API_KEY")
def is_silence(data, max_amplitude_threshold=3000):
"""Check if audio data contains silence."""
# Find the maximum absolute amplitude in the audio data
max_amplitude = np.max(np.abs(data))
return max_amplitude <= max_amplitude_threshold
def record_audio_chunk(audio, stream, chunk_length=5):
print("Recording...")
frames = []
# Calculate the number of chunks needed for the specified length of recording
# 16000 Hertz -> sufficient for capturing the human voice
# 1024 frames -> the higher, the higher the latency
num_chunks = int(16000 / 1024 * chunk_length)
# Record the audio data in chunks
for _ in range(num_chunks):
data = stream.read(1024)
frames.append(data)
temp_file_path = os.path.dirname(__file__) + "/temp_audio_chunk.wav"
# print(temp_file_path)
print("Writing...")
with wave.open(temp_file_path, 'wb') as wf:
wf.setnchannels(1) # Mono channel
wf.setsampwidth(audio.get_sample_size(pyaudio.paInt16)) # Sample width
wf.setframerate(16000) # Sample rate
wf.writeframes(b''.join(frames)) # Write audio frames
# Check if the recorded chunk contains silence
try:
samplerate, data = wavfile.read(temp_file_path)
if is_silence(data):
# os.remove(temp_file_path)
return True
else:
return False
except Exception as e:
print(f"Error while reading audio file: {e}")
def load_whisper():
model = whisper.load_model("base")
return model
def transcribe_audio(api_key, filename):
print("Transcribing...")
# Print all files in the current directory
print("Current directory files:", os.listdir())
if os.path.dirname(__file__) + "/"+ file_path:
results = model.transcribe(file_path) # , fp16=False
return results['text']
else:
return None
def load_prompt():
input_prompt = """
You are a friendly conversational chatbot
Previous conversation:
{chat_history}
New human question: {question}
Response:
"""
return input_prompt
def load_llm():
chat_groq = ChatGroq(temperature=0, model_name="llama3-8b-8192",
groq_api_key=groq_api_key)
return chat_groq
def get_response_llm(user_question, memory):
input_prompt = load_prompt()
chat_groq = load_llm()
# Look how "chat_history" is an input variable to the prompt template
prompt = PromptTemplate.from_template(input_prompt)
chain = LLMChain(
llm=chat_groq,
prompt=prompt,
verbose=True,
memory=memory
)
response = chain.invoke({"question": user_question})
return response['text']
def play_text_to_speech(text, language='en', slow=False):
# Generate text-to-speech audio from the provided text
tts = gTTS(text=text, lang=language, slow=slow)
# Save the generated audio to a temporary file
temp_audio_file = "temp_audio.mp3"
tts.save(temp_audio_file)
# Initialize the pygame mixer for audio playback
pygame.mixer.init()
# Load the temporary audio file into the mixer
pygame.mixer.music.load(temp_audio_file)
# Start playing the audio
pygame.mixer.music.play()
# Wait until the audio playback finishes
while pygame.mixer.music.get_busy():
pygame.time.Clock().tick(10) # Control the playback speed
# Stop the audio playback
pygame.mixer.music.stop()
# Clean up: Quit the pygame mixer and remove the temporary audio file
pygame.mixer.quit()
os.remove(temp_audio_file) |