Update order_assistant.py

order_assistant.py

@@ -1,68 +1,86 @@
-import json
+import base64
 import random
+import numpy as np
 import speech_recognition as sr
 import edge_tts
-import asyncio
 import tempfile
+import json
 
 # Load the food menu from menu.json
-with open("menu.json") as f:
-    food_menu = json.load(f)
+def load_food_menu():
+    with open("menu.json") as f:
+        return json.load(f)
+
+food_menu = load_food_menu()
+
+class AudioProcessor:
+    @staticmethod
+    def encode_audio(data, sample_rate):
+        encoded = base64.b64encode(data.tobytes()).decode("UTF-8")
+        return {
+            "realtimeInput": {
+                "mediaChunks": [
+                    {
+                        "mimeType": f"audio/pcm;rate={sample_rate}",
+                        "data": encoded,
+                    }
+                ],
+            },
+        }
 
-# Function to get food suggestions based on user input
-def get_food_suggestion(food_type, filter_type):
-    if food_type in food_menu:
-        return random.choice(food_menu[food_type].get(filter_type, []))
-    return "No suggestion available"
+    @staticmethod
+    def process_audio_response(data):
+        audio_data = base64.b64decode(data)
+        return np.frombuffer(audio_data, dtype=np.int16)
 
-# Speech-to-Text function
-def recognize_speech_from_mic():
-    recognizer = sr.Recognizer()
-    with sr.Microphone() as source:
-        print("Say something!")
-        audio = recognizer.listen(source)
-        try:
-            return recognizer.recognize_google(audio)  # Use Google Speech Recognition
-        except sr.UnknownValueError:
-            return "Sorry I didn't catch that"
-        except sr.RequestError:
-            return "Sorry, I'm having trouble reaching the service"
 
-# Text-to-Speech function
-async def text_to_speech(text):
-    communicate = edge_tts.Communicate(text, "en-US-AriaNeural", rate="0%", pitch="0Hz")
-    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
-        tmp_path = tmp_file.name
-        await communicate.save(tmp_path)
-    return tmp_path
+class OrderAssistant:
+    def __init__(self, food_menu):
+        self.food_menu = food_menu
+        self.audio_processor = AudioProcessor()
 
-# AI Food Ordering Assistant Function
-async def food_order_assistant(audio_input, food_type):
-    # Greeting and introduction
-    greeting_text = "Welcome to the restaurant! What would you like to order today? Are you looking for Vegan, Halal, or Guilt-Free options?"
-    audio_path = await text_to_speech(greeting_text)
+    def get_food_suggestion(self, food_type, dietary_preference):
+        if food_type in self.food_menu:
+            return random.choice(self.food_menu[food_type].get(dietary_preference, []))
+        return "Sorry, we don't have that option."
 
-    # Recognize customer's preference (Vegan, Halal, Guilt-Free)
-    dietary_preference = recognize_speech_from_mic().lower()
+    def recognize_speech_from_mic(self):
+        recognizer = sr.Recognizer()
+        with sr.Microphone() as source:
+            print("Say something!")
+            audio = recognizer.listen(source)
+            try:
+                return recognizer.recognize_google(audio)  # Google Speech Recognition
+            except sr.UnknownValueError:
+                return "Sorry I didn't catch that"
+            except sr.RequestError:
+                return "Sorry, I'm having trouble reaching the service"
 
-    if dietary_preference not in ["vegan", "halal", "guilt-free"]:
-        dietary_preference = "vegan"  # Default to Vegan if unrecognized
+    async def text_to_speech(self, text):
+        communicate = edge_tts.Communicate(text, "en-US-AriaNeural", rate="0%", pitch="0Hz")
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
+            tmp_path = tmp_file.name
+            await communicate.save(tmp_path)
+        return tmp_path
 
-    # Get food suggestion based on preference
-    suggestion = get_food_suggestion(food_type, dietary_preference)
+    async def food_order_assistant(self, audio_input, food_type):
+        greeting_text = "Welcome to the restaurant! What would you like to order today? Are you looking for Vegan, Halal, or Guilt-Free options?"
+        audio_path = await self.text_to_speech(greeting_text)
 
-    suggestion_text = f"I suggest you try {suggestion}. Does that sound good?"
-    suggestion_audio = await text_to_speech(suggestion_text)
+        dietary_preference = self.recognize_speech_from_mic().lower()
+        if dietary_preference not in ["vegan", "halal", "guilt-free"]:
+            dietary_preference = "vegan"  # Default to Vegan if unrecognized
 
-    # Wait for customer confirmation
-    confirmation = recognize_speech_from_mic().lower()
+        suggestion = self.get_food_suggestion(food_type, dietary_preference)
+        suggestion_text = f"I suggest you try {suggestion}. Does that sound good?"
+        suggestion_audio = await self.text_to_speech(suggestion_text)
 
-    if "yes" in confirmation:
-        confirmation_text = f"Your order for {suggestion} has been confirmed. Sending the order to the kitchen."
-        confirmation_audio = await text_to_speech(confirmation_text)
-        # Simulate sending order to kitchen
-        return confirmation_audio, f"Confirmed order: {suggestion}"
-    else:
-        cancellation_text = "Okay, let's try again."
-        cancellation_audio = await text_to_speech(cancellation_text)
-        return cancellation_audio, "Order not confirmed"
+        confirmation = self.recognize_speech_from_mic().lower()
+        if "yes" in confirmation:
+            confirmation_text = f"Your order for {suggestion} has been confirmed. Sending the order to the kitchen."
+            confirmation_audio = await self.text_to_speech(confirmation_text)
+            return confirmation_audio, f"Confirmed order: {suggestion}"
+        else:
+            cancellation_text = "Okay, let's try again."
+            cancellation_audio = await self.text_to_speech(cancellation_text)
+            return cancellation_audio, "Order not confirmed"