app.py

import os, time, requests, tempfile, asyncio, logging
import gradio as gr
from transformers import pipeline
import edge_tts
from collections import Counter

# ─── Configuration ──────────────────────────────────────────────────────────────
ENDPOINT_URL = "https://xzup8268xrmmxcma.us-east-1.aws.endpoints.huggingface.cloud/invocations"
HF_TOKEN     = os.getenv("HF_TOKEN")

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ─── Helpers ───────────────────────────────────────────────────────────────────
# 1) Speech→Text
asr = pipeline("automatic-speech-recognition", "facebook/wav2vec2-base-960h")
def speech_to_text(audio):
    if not audio:
        return ""
    # Gradio supplies a tuple (sr, ndarray)
    if isinstance(audio, tuple):
        sr, arr = audio
        return asr(arr, sampling_rate=sr)["text"]
    # filepath
    return asr(audio)["text"]

# 2) Prompt formatting
def format_prompt(message, history):
    fixed_prompt = """
    You are a smart mood analyzer tasked with determining the user's mood for a music recommendation system. Your goal is to classify the user's mood into one of four categories: Happy, Sad, Instrumental, or Party.
    Instructions:
    1. Engage in a conversation with the user to understand their mood.
    2. Ask relevant questions to guide the conversation towards mood classification.
    3. If the user's mood is clear, respond with a single word: "Happy", "Sad", "Instrumental", or "Party".
    4. If the mood is unclear, continue the conversation with a follow-up question.
    5. Limit the conversation to a maximum of 5 exchanges.
    6. Do not classify the mood prematurely if it's not evident from the user's responses.
    7. Focus on the user's emotional state rather than specific activities or preferences.
    8. If unable to classify after 5 exchanges, respond with "Unclear" to indicate the need for more information.
    Remember: Your primary goal is mood classification. Stay on topic and guide the conversation towards understanding the user's emotional state.
    """
    prompt = f"{fixed_prompt}\n"
    for i, (u, b) in enumerate(history):
        prompt += f"User: {u}\nAssistant: {b}\n"
        if i == 3:
            prompt += "Note: This is the last exchange. Classify the mood if possible or respond with 'Unclear'.\n"
    prompt += f"User: {message}\nAssistant:"
    return prompt

# 3) Call HF Invocation Endpoint
def query_model(prompt, max_new_tokens=64, temperature=0.1):
    headers = {
        "Authorization": f"Bearer {HF_TOKEN}",
        "Content-Type": "application/json",
    }
    payload = {
        "inputs": prompt,
        "parameters": {"max_new_tokens": max_new_tokens, "temperature": temperature},
    }
    resp = requests.post(ENDPOINT_URL, headers=headers, json=payload, timeout=30)
    resp.raise_for_status()
    return resp.json()[0]["generated_text"]

# 4) Aggregate mood from history
def aggregate_mood_from_history(history):
    mood_words = {"happy", "sad", "instrumental", "party"}
    counts = Counter()
    for _, bot_response in history:
        for tok in bot_response.split():
            w = tok.strip('.,?!;"\'').lower()
            if w in mood_words:
                counts[w] += 1
    if not counts:
        return None
    return counts.most_common(1)[0][0]

# 5) Text→Speech
def text_to_speech(text):
    communicate = edge_tts.Communicate(text)
    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
        # save synchronously to simplify callback
        asyncio.get_event_loop().run_until_complete(communicate.save(tmp.name))
        return tmp.name

# ─── Gradio Callbacks ───────────────────────────────────────────────────────────
def user_turn(user_input, history):
    history = history + [(user_input, None)]
    formatted = format_prompt(user_input, history)
    raw = query_model(formatted)
    # temporarily assign raw
    history[-1] = (user_input, raw)
    # aggregate mood
    mood = aggregate_mood_from_history(history)
    if mood:
        reply = f"Playing {mood.capitalize()} playlist for you!"
    else:
        reply = raw
    history[-1] = (user_input, reply)
    return history, history, ""

async def bot_audio(history):
    last = history[-1][1]
    return text_to_speech(last)

def speech_callback(audio):
    return speech_to_text(audio)

# ─── Build the Interface ────────────────────────────────────────────────────────
with gr.Blocks() as demo:
    gr.Markdown("## 🎵 Mood-Based Music Buddy")
    chat = gr.Chatbot()
    txt = gr.Textbox(placeholder="Type your mood...", label="Text")
    send = gr.Button("Send")
    mic  = gr.Audio()
    out_audio = gr.Audio(label="Response (Audio)", autoplay=True)
    state = gr.State([])

    def init():
        greeting = "Hi! I'm your music buddy—tell me how you’re feeling today."
        return [("", greeting)], [("", greeting)], None
    demo.load(init, outputs=[state, chat, out_audio])

    txt.submit(user_turn, [txt, state], [state, chat, txt])\
       .then(bot_audio, [state], [out_audio])
    send.click(user_turn, [txt, state], [state, chat, txt])\
        .then(bot_audio, [state], [out_audio])

    mic.change(speech_callback, [mic], [txt])\
       .then(user_turn, [txt, state], [state, chat, txt])\
       .then(bot_audio, [state], [out_audio])

if __name__ == "__main__":
    demo.launch(debug=True)

# import gradio as gr
# import requests
# from transformers import pipeline
# import edge_tts
# import tempfile
# import asyncio
# import os
# import json
# import time
# import logging

# # Set up logging
# logging.basicConfig(level=logging.INFO)
# logger = logging.getLogger(__name__)

# ENDPOINT_URL = "https://xzup8268xrmmxcma.us-east-1.aws.endpoints.huggingface.cloud/invocations"
# hf_token = os.getenv("HF_TOKEN")

# print(f"DEBUG: Starting application at {time.strftime('%Y-%m-%d %H:%M:%S')}")
# print(f"DEBUG: HF_TOKEN available: {bool(hf_token)}")
# print(f"DEBUG: Endpoint URL: {ENDPOINT_URL}")

# try:
#     print("DEBUG: Loading ASR pipeline...")
#     start_time = time.time()
#     asr = pipeline("automatic-speech-recognition", "facebook/wav2vec2-base-960h")
#     print(f"DEBUG: ASR pipeline loaded in {time.time() - start_time:.2f} seconds")
# except Exception as e:
#     print(f"DEBUG: Error loading ASR pipeline: {e}")
#     asr = None

# INITIAL_MESSAGE = "Hi! I'm your music buddy—tell me about your mood and the type of tunes you're in the mood for today!"

# def speech_to_text(speech):
#     print(f"DEBUG: speech_to_text called with input: {speech is not None}")
#     if speech is None:
#         print("DEBUG: No speech input provided")
#         return ""
    
#     try:
#         start_time = time.time()
#         print("DEBUG: Starting speech recognition...")
#         result = asr(speech)["text"]
#         print(f"DEBUG: Speech recognition completed in {time.time() - start_time:.2f} seconds")
#         print(f"DEBUG: Recognized text: '{result}'")
#         return result
#     except Exception as e:
#         print(f"DEBUG: Error in speech_to_text: {e}")
#         return ""

# def classify_mood(input_string):
#     print(f"DEBUG: classify_mood called with: '{input_string}'")
#     input_string = input_string.lower()
#     mood_words = {"happy", "sad", "instrumental", "party"}
#     for word in mood_words:
#         if word in input_string:
#             print(f"DEBUG: Mood classified as: {word}")
#             return word, True
#     print("DEBUG: No mood classified")
#     return None, False

# def generate(prompt, history, temperature=0.1, max_new_tokens=2048):
#     print(f"DEBUG: generate() called at {time.strftime('%H:%M:%S')}")
#     print(f"DEBUG: Prompt length: {len(prompt)}")
#     print(f"DEBUG: History length: {len(history)}")
    
#     if not hf_token:
#         error_msg = "Error: Hugging Face authentication required. Please set your HF_TOKEN."
#         print(f"DEBUG: {error_msg}")
#         return error_msg
    
#     try:
#         print("DEBUG: Formatting prompt...")
#         start_time = time.time()
#         formatted_prompt = format_prompt(prompt, history)
#         print(f"DEBUG: Prompt formatted in {time.time() - start_time:.2f} seconds")
#         print(f"DEBUG: Formatted prompt length: {len(formatted_prompt)}")
        
#         headers = {"Authorization": f"Bearer {hf_token}", "Content-Type": "application/json"}
#         payload = {
#             "inputs": formatted_prompt,
#             "parameters": {
#                 "temperature": temperature,
#                 "max_new_tokens": max_new_tokens
#             }
#         }
        
#         print("DEBUG: Making API request...")
#         api_start_time = time.time()
#         response = requests.post(ENDPOINT_URL, headers=headers, json=payload, timeout=60)
#         api_duration = time.time() - api_start_time
#         print(f"DEBUG: API request completed in {api_duration:.2f} seconds")
#         print(f"DEBUG: Response status code: {response.status_code}")
        
#         if response.status_code == 200:
#             print("DEBUG: Parsing API response...")
#             result = response.json()                 
#             output = result[0]["generated_text"]
            
#             print(f"DEBUG: Generated output: '{output[:100]}...'")
            
#             mood, is_classified = classify_mood(output)
#             if is_classified:
#                 playlist_message = f"Playing {mood.capitalize()} playlist for you!"
#                 print(f"DEBUG: Returning playlist message: {playlist_message}")
#                 return playlist_message
            
#             print(f"DEBUG: Returning generated output")
#             return output
#         else:
#             error_msg = f"Error: {response.status_code} - {response.text}"
#             print(f"DEBUG: API error: {error_msg}")
#             return error_msg
            
#     except requests.exceptions.Timeout:
#         error_msg = "Error: API request timed out after 60 seconds"
#         print(f"DEBUG: {error_msg}")
#         return error_msg
#     except Exception as e:
#         error_msg = f"Error generating response: {str(e)}"
#         print(f"DEBUG: Exception in generate(): {error_msg}")
#         return error_msg

# def format_prompt(message, history):
#     print("DEBUG: format_prompt called")
#     fixed_prompt = """
#     You are a smart mood analyzer tasked with determining the user's mood for a music recommendation system. Your goal is to classify the user's mood into one of four categories: Happy, Sad, Instrumental, or Party.
#     Instructions:
#     1. Engage in a conversation with the user to understand their mood.
#     2. Ask relevant questions to guide the conversation towards mood classification.
#     3. If the user's mood is clear, respond with a single word: "Happy", "Sad", "Instrumental", or "Party".
#     4. If the mood is unclear, continue the conversation with a follow-up question.
#     5. Limit the conversation to a maximum of 5 exchanges.
#     6. Do not classify the mood prematurely if it's not evident from the user's responses.
#     7. Focus on the user's emotional state rather than specific activities or preferences.
#     8. If unable to classify after 5 exchanges, respond with "Unclear" to indicate the need for more information.
#     Remember: Your primary goal is mood classification. Stay on topic and guide the conversation towards understanding the user's emotional state.
#     """
#     prompt = f"{fixed_prompt}\n"
    
#     for i, (user_prompt, bot_response) in enumerate(history):
#         prompt += f"User: {user_prompt}\nAssistant: {bot_response}\n"
#         if i == 3:
#             prompt += "Note: This is the last exchange. Classify the mood if possible or respond with 'Unclear'.\n"
    
#     prompt += f"User: {message}\nAssistant:"
#     print(f"DEBUG: Final prompt length: {len(prompt)}")
#     return prompt

# async def text_to_speech(text):
#     print(f"DEBUG: text_to_speech called with text length: {len(text)}")
#     try:
#         start_time = time.time()
#         print("DEBUG: Creating TTS communicate object...")
#         communicate = edge_tts.Communicate(text)
        
#         print("DEBUG: Creating temporary file...")
#         with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp_file:
#             tmp_path = tmp_file.name
#             print(f"DEBUG: Saving TTS to: {tmp_path}")
#             await communicate.save(tmp_path)
        
#         duration = time.time() - start_time
#         print(f"DEBUG: TTS completed in {duration:.2f} seconds")
#         print(f"DEBUG: TTS file size: {os.path.getsize(tmp_path) if os.path.exists(tmp_path) else 'File not found'}")
#         return tmp_path
#     except Exception as e:
#         print(f"DEBUG: TTS Error: {e}")
#         return None

# def process_input(input_text, history):
#     print(f"DEBUG: process_input called with text: '{input_text[:50]}...'")
#     if not input_text:
#         print("DEBUG: No input text provided")
#         return history, history, ""
    
#     print("DEBUG: Calling generate function...")
#     start_time = time.time()
#     response = generate(input_text, history)
#     duration = time.time() - start_time
#     print(f"DEBUG: generate() completed in {duration:.2f} seconds")
#     print(f"DEBUG: Response: '{response[:100]}...'")
    
#     history.append((input_text, response))
#     print(f"DEBUG: Updated history length: {len(history)}")
#     return history, history, ""

# async def generate_audio(history):
#     print(f"DEBUG: generate_audio called with history length: {len(history)}")
#     if history and len(history) > 0:
#         last_response = history[-1][1]
#         print(f"DEBUG: Generating audio for: '{last_response[:50]}...'")
#         start_time = time.time()
#         audio_path = await text_to_speech(last_response)
#         duration = time.time() - start_time
#         print(f"DEBUG: Audio generation completed in {duration:.2f} seconds")
#         return audio_path
#     print("DEBUG: No history available for audio generation")
#     return None

# async def init_chat():
#     print("DEBUG: init_chat called")
#     try:
#         history = [("", INITIAL_MESSAGE)]
#         print("DEBUG: Generating initial audio...")
#         start_time = time.time()
#         audio_path = await text_to_speech(INITIAL_MESSAGE)
#         duration = time.time() - start_time
#         print(f"DEBUG: Initial audio generated in {duration:.2f} seconds")
#         print("DEBUG: init_chat completed successfully")
#         return history, history, audio_path
#     except Exception as e:
#         print(f"DEBUG: Error in init_chat: {e}")
#         return [("", INITIAL_MESSAGE)], [("", INITIAL_MESSAGE)], None

# def handle_voice_upload(audio_file):
#     print(f"DEBUG: handle_voice_upload called with file: {audio_file}")
#     if audio_file is None:
#         print("DEBUG: No audio file provided")
#         return ""
    
#     try:
#         start_time = time.time()
#         result = speech_to_text(audio_file)
#         duration = time.time() - start_time
#         print(f"DEBUG: Voice upload processing completed in {duration:.2f} seconds")
#         return result
#     except Exception as e:
#         print(f"DEBUG: Error in handle_voice_upload: {e}")
#         return ""

# print("DEBUG: Creating Gradio interface...")

# with gr.Blocks() as demo:
#     gr.Markdown("# Mood-Based Music Recommender with Continuous Voice Chat")
    
#     chatbot = gr.Chatbot()
    
#     with gr.Row():
#         msg = gr.Textbox(
#             placeholder="Type your message here...", 
#             label="Text Input",
#             scale=4
#         )
#         submit = gr.Button("Send", scale=1)
    
#     with gr.Row():
#         voice_input = gr.Audio(
#             label="🎤 Record your voice or upload audio file",
#             sources=["microphone", "upload"],
#             type="filepath"
#         )
        
#     audio_output = gr.Audio(label="AI Response", autoplay=True)

#     state = gr.State([])

#     print("DEBUG: Setting up Gradio event handlers...")

#     demo.load(init_chat, outputs=[state, chatbot, audio_output])

#     def submit_and_generate_audio(input_text, history):
#         print(f"DEBUG: submit_and_generate_audio called at {time.strftime('%H:%M:%S')}")
#         start_time = time.time()
#         new_state, new_chatbot, empty_msg = process_input(input_text, history)
#         duration = time.time() - start_time
#         print(f"DEBUG: submit_and_generate_audio completed in {duration:.2f} seconds")
#         return new_state, new_chatbot, empty_msg

#     msg.submit(
#         submit_and_generate_audio,
#         inputs=[msg, state],
#         outputs=[state, chatbot, msg]
#     ).then(
#         generate_audio,
#         inputs=[state],
#         outputs=[audio_output]
#     )
    
#     submit.click(
#         submit_and_generate_audio,
#         inputs=[msg, state],
#         outputs=[state, chatbot, msg]
#     ).then(
#         generate_audio,
#         inputs=[state],
#         outputs=[audio_output]
#     )

#     voice_input.upload(
#         handle_voice_upload,
#         inputs=[voice_input],
#         outputs=[msg]
#     ).then(
#         submit_and_generate_audio,
#         inputs=[msg, state],
#         outputs=[state, chatbot, msg]
#     ).then(
#         generate_audio,
#         inputs=[state],
#         outputs=[audio_output]
#     )

# print("DEBUG: Gradio interface created successfully")

# if __name__ == "__main__":
#     print("DEBUG: Launching Gradio app...")
#     demo.launch(share=True, debug=True)