order_assistant.py

import base64
import random
import numpy as np
import speech_recognition as sr
import edge_tts
import tempfile
import json

# Load the food menu from menu.json
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,
                    }
                ],
            },
        }

    @staticmethod
    def process_audio_response(data):
        audio_data = base64.b64decode(data)
        return np.frombuffer(audio_data, dtype=np.int16)


class OrderAssistant:
    def __init__(self, food_menu):
        self.food_menu = food_menu
        self.audio_processor = AudioProcessor()

    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."

    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"

    async def text_to_speech(self, text):
        # Ensure consistent indentation for the body of the function
        communicate = edge_tts.Communicate(text, "en-US-AriaNeural", rate="+0%", pitch="+0Hz")  # Fixed indentation
        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
            tmp_path = tmp_file.name
            await communicate.save(tmp_path)
        return tmp_path

    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)

        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

        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)

        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"