Add files via upload

groq_ai.py

@@ -0,0 +1,68 @@
+from groq import Groq
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+GROQ_API_KEY = os.getenv("GROQ_API_KEY")
+if not GROQ_API_KEY:
+    raise ValueError("❌ GROQ_API_KEY not found in .env")
+
+client = Groq(api_key=GROQ_API_KEY)
+
+SYSTEM_PROMPT = """
+Give useful and shortly answers,
+Don't forget to add, "sir".
+"""
+
+chat_history = []
+
+def get_best_model():
+    model_list = client.models.list()
+
+    available = [m.id for m in model_list.data]
+    print("Available Groq Models:\n", available)
+
+    priority = [
+        "llama-3.3-70b",
+        "llama-3.2-90b",
+        "llama-3.2-70b",
+        "llama-3.1-8b",
+        "mixtral-8x7b",
+        "gemma2-9b-it"
+    ]
+
+    for name in priority:
+        for model_id in available:
+            if name in model_id.lower():
+                print(f"\n✔ Selected model: {model_id}\n")
+                return model_id
+
+
+    return available[0]
+
+BEST_MODEL = get_best_model()
+
+def get_response(user_text):
+    global chat_history
+
+    chat_history.append({"role": "user", "content": user_text})
+
+    messages = (
+        [{"role": "system", "content": SYSTEM_PROMPT}]
+        + chat_history[-5:]
+    )
+
+    response = client.chat.completions.create(
+        model=BEST_MODEL,
+        messages=messages,
+        max_tokens=200
+    )
+
+    text = response.choices[0].message.content.strip()
+    chat_history.append({"role": "assistant", "content": text})
+    return text
+
+
+def reset_history():
+    global chat_history
+    chat_history = []

main.py

@@ -0,0 +1,47 @@
+import asyncio
+import threading
+from voice_player import VoiceImagePlayer, edge_speak
+from speech_to_text import record_voice
+#from recognition import record_voice
+from groq_ai import get_response
+
+async def get_voice_input():
+    return await asyncio.to_thread(record_voice)
+
+async def ai_loop(player):
+
+    while True:
+        user_text = await get_voice_input()
+        if not user_text:
+            continue
+
+        print("User:", user_text)
+        lowered = user_text.lower().strip()
+
+        if lowered in ["exit", "quit", "goodbye", "bye", "stop","mute"]:
+            edge_speak("Goodbye sir.", player)
+            player.root.after(300, player.root.destroy)
+            break
+
+        response = get_response(user_text)
+        print("AI:", response)
+
+        edge_speak(response, player)
+
+        player.write_log(f"You: {user_text}")
+        player.write_log(f"AI: {response}")
+
+
+def main_loop():
+    print("")
+
+    player = VoiceImagePlayer("face.png", size=(900, 900))
+
+    def thread_target():
+        asyncio.run(ai_loop(player))
+
+    threading.Thread(target=thread_target, daemon=True).start()
+    player.root.mainloop()
+
+if __name__ == "__main__":
+    main_loop()

recognition.py

@@ -0,0 +1,25 @@
+import speech_recognition as sr
+
+recognizer = sr.Recognizer()
+mic = sr.Microphone()
+
+def record_voice(prompt="🎙 I'm listening, sir...", timeout=None, phrase_time_limit=None):
+    with mic as source:
+        print(prompt)
+        recognizer.adjust_for_ambient_noise(source, duration=0.5) 
+        try:
+            audio = recognizer.listen(source, timeout=timeout, phrase_time_limit=phrase_time_limit)
+        except sr.WaitTimeoutError:
+            print("Listening timed out.")
+            return None
+
+    try:
+        text = recognizer.recognize_google(audio, language="en-US")
+        print("👤 You:", text)
+        return text
+    except sr.UnknownValueError:
+        print("🤖 Couldn't recognize the voice")
+        return None
+    except sr.RequestError as e:
+        print("🤖 Google Speech API Error:", e)
+        return None

speech_to_text.py

@@ -0,0 +1,32 @@
+# vosk_voice.py
+import sounddevice as sd
+import vosk
+import queue
+import sys
+import json
+
+MODEL_PATH = "C:/Users/90553/Downloads/vosk/vosk-model-small-en-us-0.15" #put your path in here
+model = vosk.Model(MODEL_PATH)
+
+q = queue.Queue()
+
+def callback(indata, frames, time, status):
+    if status:
+        print(status, file=sys.stderr)
+    q.put(bytes(indata))
+
+def record_voice(prompt="🎙 I'm listening, sir...", timeout=None, phrase_time_limit=None):
+
+    print(prompt)
+    rec = vosk.KaldiRecognizer(model, 16000)
+
+    with sd.RawInputStream(samplerate=16000, blocksize=8000, dtype='int16',
+                           channels=1, callback=callback):
+        while True:
+            data = q.get()
+            if rec.AcceptWaveform(data):
+                result = json.loads(rec.Result())
+                text = result.get("text", "")
+                if text.strip():
+                    print("👤 You:", text)
+                    return text

voice_player.py

@@ -0,0 +1,141 @@
+import os
+import io
+import random
+import threading
+import asyncio
+import edge_tts
+import sounddevice as sd
+import soundfile as sf
+from PIL import Image, ImageTk
+import tkinter as tk
+from tkinter.scrolledtext import ScrolledText
+import re
+
+def normalize_punctuation(text: str) -> str:
+    text = re.sub(r'\.\s+', ', ', text) #period
+    text = re.sub(r'!\s+', ', ', text) #exclamation point
+    text = re.sub(r'\?\s+', ', ', text) #question mark
+
+    if text.endswith("."):
+        text = text[:-1]
+
+    return text.strip()
+
+class VoiceImagePlayer:
+    def __init__(self, face_path, size=(760, 760)):
+        self.root = tk.Tk()
+        self.root.title("F.A.I.T.H")
+        self.root.resizable(False, False)
+        self.root.geometry("760x900")
+
+        self.fixed_size = size
+
+        self.face_original = Image.open(face_path).resize(self.fixed_size, Image.LANCZOS)
+        self.face_label = tk.Label(self.root, bg="#001f4d", borderwidth=0, highlightthickness=0)
+        self.face_label.place(relx=0.5, rely=0.4, anchor="center")
+
+        self.scale = 1.0
+        self.playing = False
+
+        self.text_box = ScrolledText(
+            self.root,
+            fg="#00ffff",
+            bg="#001f4d",
+            insertbackground="#00ffff",
+            height=12,
+            borderwidth=0,
+            highlightthickness=0,
+            relief="flat",
+            wrap="word",
+            font=("Consolas", 10),
+            padx=10,
+            pady=10
+        )
+        
+        self.text_box.place(relx=0.5, rely=0.85, anchor="center")
+
+        self.animate()
+        self.root.protocol("WM_DELETE_WINDOW", self.on_close)
+        
+    def on_close(self):
+        os._exit(0)
+
+    def write_log(self, text):
+        self.text_box.configure(state="normal")
+        self.text_box.insert(tk.END, text + "\n")
+        self.text_box.see(tk.END)
+        self.text_box.configure(state="disabled")
+
+    # Animation
+    def start_animation(self):
+        self.playing = True
+
+    def stop_animation(self):
+        self.playing = False
+
+    def animate(self):
+        if self.playing:
+            self.scale += random.uniform(-0.015, 0.02)
+            self.scale = min(max(self.scale, 1.02), 1.10)
+        else:
+            if abs(self.scale - 1.0) < 0.003:
+                self.scale = 1.0
+            elif self.scale > 1.0:
+                self.scale -= 0.009
+            else:
+                self.scale += 0.005
+
+        base_w, base_h = self.fixed_size
+        new_size = (int(base_w * self.scale), int(base_h * self.scale))
+        resized = self.face_original.resize(new_size, Image.LANCZOS)
+        tk_img = ImageTk.PhotoImage(resized)
+        self.face_label.configure(image=tk_img)
+        self.face_label.image = tk_img
+
+        self.root.after(50, self.animate)
+
+VOICE = "en-GB-OllieMultilingualNeural"
+
+def edge_speak(text: str, img_player: VoiceImagePlayer, voice: str = VOICE):
+    if not text.strip():
+        return
+
+    optimized_text = normalize_punctuation(text)
+
+    def _thread_target():
+        async def _speak():
+            communicate = edge_tts.Communicate(
+                optimized_text,
+                voice,
+                pitch="+2Hz",
+                rate="+8%"
+            )
+
+            audio_buffer = bytearray()
+            first_audio_received = False
+
+            async for chunk in communicate.stream():
+                if chunk["type"] == "audio":
+                    audio_buffer.extend(chunk["data"])
+                    if not first_audio_received:
+                        first_audio_received = True
+                        img_player.start_animation()
+
+            wav_bytes = io.BytesIO(audio_buffer)
+
+            try:
+                data, samplerate = sf.read(wav_bytes, dtype="float32")
+            except Exception as e:
+                print("Decode error:", e)
+                return
+
+            sd.play(data, samplerate)
+            sd.wait()
+            img_player.stop_animation()
+
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(_speak())
+        loop.close()
+
+    threading.Thread(target=_thread_target, daemon=True).start()