app.py

import gradio as gr
import torch
import numpy as np
import librosa
from transformers import (
    pipeline, AutoTokenizer, AutoModelForCausalLM, 
    WhisperProcessor, WhisperForConditionalGeneration
)
import soundfile as sf
import json
import time
from datetime import datetime
import os
import warnings

# Import Dia model correctly[2]
try:
    from dia.model import Dia
    DIA_AVAILABLE = True
    print("✅ Dia model imported successfully")
except ImportError as e:
    print(f"⚠️ Dia import failed: {e}")
    DIA_AVAILABLE = False

warnings.filterwarnings("ignore")

class MayaAI:
    def __init__(self):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        print(f"🚀 Initializing Maya AI on {self.device}")
        
        # Load Whisper ASR with FORCED English
        self.asr_processor = WhisperProcessor.from_pretrained("openai/whisper-large-v3")
        self.asr_model = WhisperForConditionalGeneration.from_pretrained(
            "openai/whisper-large-v3",
            torch_dtype=torch.float16 if self.device == "cuda" else torch.float32
        ).to(self.device)
        
        # FORCE English transcription
        self.asr_model.config.forced_decoder_ids = self.asr_processor.get_decoder_prompt_ids(
            language="english", 
            task="transcribe"
        )
        print("✅ Whisper ASR loaded with FORCED English")
        
        # Load FREE LLM with FIXED attention mask
        self.llm_tokenizer = AutoTokenizer.from_pretrained("microsoft/DialoGPT-large")
        # FIX: Set pad_token to eos_token to avoid attention mask warnings
        if self.llm_tokenizer.pad_token is None:
            self.llm_tokenizer.pad_token = self.llm_tokenizer.eos_token
        
        self.llm_model = AutoModelForCausalLM.from_pretrained(
            "microsoft/DialoGPT-large",
            torch_dtype=torch.float16 if self.device == "cuda" else torch.float32,
            device_map="auto",
            pad_token_id=self.llm_tokenizer.eos_token_id
        )
        print("✅ DialoGPT-Large loaded with FIXED attention masks")
        
        # Load Emotion Recognition
        self.emotion_model = pipeline(
            "audio-classification",
            model="superb/wav2vec2-base-superb-er",
            device=self.device
        )
        print("✅ Emotion recognition loaded")
        
        # Load REAL Dia TTS Model[2]
        if DIA_AVAILABLE:
            try:
                # Load Dia model with correct parameters[2]
                self.dia_model = Dia.from_pretrained(
                    "nari-labs/Dia-1.6B", 
                    compute_dtype="float16" if self.device == "cuda" else "float32",
                    device=self.device
                )
                print("✅ Dia TTS loaded (Ultra-realistic dialogue generation)")
                self.use_dia = True
            except Exception as e:
                print(f"⚠️ Dia loading failed: {e}")
                self.use_dia = False
                self._load_fallback_tts()
        else:
            print("⚠️ Dia not available, using fallback TTS")
            self.use_dia = False
            self._load_fallback_tts()
        
        # Conversation storage
        self.conversations = {}
        self.call_active = False
        self.speaker_turn = 1  # Track speaker turns for Dia[2]
        
    def _load_fallback_tts(self):
        """Load fallback TTS if Dia is not available"""
        try:
            from transformers import SpeechT5Processor, SpeechT5ForTextToSpeech, SpeechT5HifiGan
            from datasets import load_dataset
            
            self.tts_processor = SpeechT5Processor.from_pretrained("microsoft/speecht5_tts")
            self.tts_model = SpeechT5ForTextToSpeech.from_pretrained(
                "microsoft/speecht5_tts",
                torch_dtype=torch.float32
            ).to(self.device)
            self.vocoder = SpeechT5HifiGan.from_pretrained(
                "microsoft/speecht5_hifigan",
                torch_dtype=torch.float32
            ).to(self.device)
            
            # Load female speaker embeddings
            embeddings_dataset = load_dataset("Matthijs/cmu-arctic-xvectors", split="validation")
            self.speaker_embeddings = torch.tensor(
                embeddings_dataset[7306]["xvector"], 
                dtype=torch.float32
            ).unsqueeze(0).to(self.device)
            print("✅ SpeechT5 TTS loaded as fallback")
        except Exception as e:
            print(f"❌ Fallback TTS loading failed: {e}")
    
    def transcribe_with_whisper(self, audio_path):
        """Transcribe using Whisper with FORCED English"""
        try:
            if audio_path is None:
                return "No audio provided"
            
            # Load and preprocess audio
            audio, sr = librosa.load(audio_path, sr=16000, mono=True)
            
            # Process with Whisper - FORCE English
            inputs = self.asr_processor(
                audio, 
                sampling_rate=16000, 
                return_tensors="pt",
                language="english"
            ).to(self.device)
            
            with torch.no_grad():
                predicted_ids = self.asr_model.generate(
                    inputs.input_features,
                    max_new_tokens=150,
                    do_sample=False,
                    forced_decoder_ids=self.asr_model.config.forced_decoder_ids
                )
            
            transcription = self.asr_processor.batch_decode(
                predicted_ids, 
                skip_special_tokens=True
            )[0]
            
            return transcription.strip()
            
        except Exception as e:
            return f"Transcription error: {str(e)}"
    
    def recognize_emotion_from_audio(self, audio_path):
        """Recognize emotion using superb model"""
        try:
            if audio_path is None:
                return "neutral"
            
            result = self.emotion_model(audio_path)
            emotion_label = result[0]["label"].lower()
            
            # Map to human emotions
            emotion_map = {
                "ang": "angry", "hap": "happy", "exc": "excited",
                "sad": "sad", "fru": "frustrated", "fea": "fearful",
                "sur": "surprised", "neu": "neutral", "dis": "disgusted"
            }
            
            return emotion_map.get(emotion_label, emotion_label)
        except:
            return "neutral"
    
    def generate_with_free_llm(self, text, emotion, history):
        """Generate response using FREE LLM with FIXED attention masks"""
        try:
            # Emotional context prompting
            emotion_prompts = {
                "angry": "I understand you're frustrated. Let me help calm this situation.",
                "sad": "I can hear the sadness in your voice. I'm here to support you.",
                "happy": "Your joy is infectious! I love your positive energy.",
                "excited": "Your enthusiasm is amazing! Tell me more!",
                "fearful": "I sense your concern. Let's work through this together.",
                "surprised": "That sounds unexpected! What happened?",
                "neutral": "I'm listening carefully. Please continue."
            }
            
            emotion_context = emotion_prompts.get(emotion, "I'm here to help.")
            
            # Build conversation context
            context_text = ""
            if history:
                for entry in history[-2:]:
                    context_text += f"User: {entry.get('user_input', '')}\nMaya: {entry.get('ai_response', '')}\n"
            
            prompt = f"{context_text}User: {text}\nMaya:"
            
            # Tokenize input with PROPER attention mask
            inputs = self.llm_tokenizer(
                prompt,
                return_tensors="pt",
                truncation=True,
                max_length=1024,
                padding=True,
                add_special_tokens=True
            ).to(self.device)
            
            # Generate response with PROPER attention mask
            with torch.no_grad():
                outputs = self.llm_model.generate(
                    input_ids=inputs.input_ids,
                    attention_mask=inputs.attention_mask,
                    max_new_tokens=80,
                    temperature=0.7,
                    do_sample=True,
                    pad_token_id=self.llm_tokenizer.pad_token_id,
                    eos_token_id=self.llm_tokenizer.eos_token_id
                )
            
            # Decode response
            response = self.llm_tokenizer.decode(
                outputs[0][inputs.input_ids.shape[1]:], 
                skip_special_tokens=True
            ).strip()
            
            # Clean up response
            if not response or len(response) < 5:
                return emotion_context
            
            return response
            
        except Exception as e:
            return f"{emotion_prompts.get(emotion, 'I understand.')} Could you tell me more about that?"
    
    def synthesize_with_dia(self, text, emotion):
        """Generate ultra-realistic dialogue using Dia[2]"""
        try:
            if not text or len(text.strip()) == 0:
                return None
            
            if self.use_dia:
                # Format text for Dia with proper speaker tags[2]
                speaker_tag = f"[S{self.speaker_turn}]"
                
                # Add emotional non-verbals based on emotion[2]
                if emotion == "happy":
                    emotional_text = f"{speaker_tag} {text} (laughs)"
                elif emotion == "sad":
                    emotional_text = f"{speaker_tag} {text} (sighs)"
                elif emotion == "excited":
                    emotional_text = f"{speaker_tag} {text}!"
                elif emotion == "angry":
                    emotional_text = f"{speaker_tag} {text} (frustrated tone)"
                elif emotion == "surprised":
                    emotional_text = f"{speaker_tag} {text} (gasps)"
                else:
                    emotional_text = f"{speaker_tag} {text}"
                
                # Generate with Dia[2]
                output = self.dia_model.generate(
                    emotional_text, 
                    use_torch_compile=True if self.device == "cuda" else False,
                    verbose=False
                )
                
                # Toggle speaker for next turn[2]
                self.speaker_turn = 2 if self.speaker_turn == 1 else 1
                
                return output
            else:
                # Fallback to SpeechT5
                return self._synthesize_with_fallback(text, emotion)
            
        except Exception as e:
            print(f"Dia TTS error: {e}")
            return self._synthesize_with_fallback(text, emotion)
    
    def _synthesize_with_fallback(self, text, emotion):
        """Fallback TTS synthesis"""
        try:
            clean_text = text.replace("[", "").replace("]", "").strip()
            if len(clean_text) > 200:
                clean_text = clean_text[:200] + "..."
            
            # Add emotional inflection through punctuation
            if emotion == "happy":
                clean_text = clean_text.replace(".", "!")
            elif emotion == "excited":
                clean_text = clean_text + "!"
            elif emotion == "sad":
                clean_text = clean_text.replace("!", ".")
            
            inputs = self.tts_processor(text=clean_text, return_tensors="pt")
            inputs = {k: v.to(self.device) for k, v in inputs.items()}
            
            with torch.no_grad():
                speech = self.tts_model.generate_speech(
                    inputs["input_ids"], 
                    self.speaker_embeddings, 
                    vocoder=self.vocoder
                )
            
            if isinstance(speech, torch.Tensor):
                speech = speech.cpu().numpy().astype(np.float32)
            
            return speech
        except Exception as e:
            print(f"Fallback TTS error: {e}")
            return None
    
    def start_call(self):
        """Start a new call session"""
        self.call_active = True
        self.speaker_turn = 1  # Reset speaker turn[2]
        greeting = "Hello! I'm Maya, your AI conversation partner. I'm here to chat with you naturally and understand your emotions. How are you feeling today?"
        
        greeting_audio = self.synthesize_with_dia(greeting, "happy")
        
        # Dia outputs at 24kHz, fallback at 22050Hz[2]
        sample_rate = 24000 if self.use_dia else 22050
        return greeting, (sample_rate, greeting_audio) if greeting_audio is not None else None, "📞 Call started! Maya is greeting you with ultra-realistic speech..."
    
    def end_call(self, user_id="default"):
        """End call and clear conversation"""
        self.call_active = False
        if user_id in self.conversations:
            self.conversations[user_id] = []
        
        farewell = "Thank you for chatting with me! It was wonderful talking with you. Have a great day!"
        farewell_audio = self.synthesize_with_dia(farewell, "happy")
        
        sample_rate = 24000 if self.use_dia else 22050
        return farewell, (sample_rate, farewell_audio) if farewell_audio is not None else None, "📞 Call ended. Conversation cleared!"
    
    def process_conversation(self, audio_input, user_id="default"):
        """Main conversation processing pipeline"""
        if not self.call_active:
            return "Please start a call first by clicking the 'Start Call' button", None, "No active call"
        
        if audio_input is None:
            return "Please record some audio", None, "No audio input"
        
        start_time = time.time()
        
        if user_id not in self.conversations:
            self.conversations[user_id] = []
        
        try:
            # Step 1: ASR with FORCED English
            transcription = self.transcribe_with_whisper(audio_input)
            
            # Step 2: Emotion recognition
            emotion = self.recognize_emotion_from_audio(audio_input)
            
            # Step 3: FREE LLM generation with FIXED attention masks
            response_text = self.generate_with_free_llm(
                transcription, emotion, self.conversations[user_id]
            )
            
            # Step 4: Ultra-realistic TTS with Dia[2]
            response_audio = self.synthesize_with_dia(response_text, emotion)
            
            # Step 5: Update conversation history
            processing_time = time.time() - start_time
            conversation_entry = {
                "timestamp": datetime.now().strftime("%H:%M:%S"),
                "user_input": transcription,
                "user_emotion": emotion,
                "ai_response": response_text,
                "processing_time": processing_time
            }
            
            self.conversations[user_id].append(conversation_entry)
            
            # Keep last 1000 exchanges as requested[5]
            if len(self.conversations[user_id]) > 1000:
                self.conversations[user_id] = self.conversations[user_id][-1000:]
            
            history = self.format_conversation_history(user_id)
            
            sample_rate = 24000 if self.use_dia else 22050
            return transcription, (sample_rate, response_audio) if response_audio is not None else None, history
            
        except Exception as e:
            return f"Processing error: {str(e)}", None, "Error in processing"
    
    def format_conversation_history(self, user_id):
        """Format conversation history for display"""
        if user_id not in self.conversations or not self.conversations[user_id]:
            return "No conversation history yet."
        
        history = []
        for i, entry in enumerate(self.conversations[user_id][-10:], 1):
            history.append(f"**Exchange {i}** ({entry['timestamp']})")
            history.append(f"🎤 **You** ({entry['user_emotion']}): {entry['user_input']}")
            history.append(f"🤖 **Maya**: {entry['ai_response']}")
            history.append(f"⏱️ *{entry['processing_time']:.2f}s*")
            history.append("---")
        
        return "\n".join(history)

# Initialize Maya AI
print("🚀 Starting Maya AI with REAL Dia TTS...")
maya = MayaAI()
print("✅ Maya AI ready with ultra-realistic dialogue generation!")

# Gradio Interface Functions
def start_call_handler():
    return maya.start_call()

def end_call_handler():
    return maya.end_call()

def process_audio_handler(audio):
    return maya.process_conversation(audio)

# Create Gradio Interface[7]
with gr.Blocks(
    title="Maya AI - Dia-Powered Sesame Killer",
    theme=gr.themes.Soft()
) as demo:
    
    gr.Markdown("""
    # 🎤 Maya AI - Dia-Powered Sesame Killer
    *Ultra-realistic dialogue generation with Dia TTS - Natural breathing, laughter, and human-like responses*
    
    **Features:** ✅ Real Dia TTS ✅ English-only ASR ✅ Emotion Recognition ✅ FREE LLM ✅ Ultra-realistic Speech
    """)
    
    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("### 📞 Call Controls")
            
            start_call_btn = gr.Button("📞 Start Call", variant="primary", size="lg")
            end_call_btn = gr.Button("📞 End Call", variant="stop", size="lg")
            
            gr.Markdown("### 🎙️ Voice Input")
            audio_input = gr.Audio(
                sources=["microphone"],
                type="filepath",
                label="Record your message in English"
            )
            
            process_btn = gr.Button("🎯 Process Audio", variant="primary")
            
        with gr.Column(scale=2):
            gr.Markdown("### 💬 Ultra-Realistic Conversation")
            
            transcription_output = gr.Textbox(
                label="📝 What you said (English)",
                lines=2,
                interactive=False
            )
            
            audio_output = gr.Audio(
                label="🔊 Maya's Ultra-Realistic Response (Dia TTS)",
                interactive=False,
                autoplay=True
            )
            
            conversation_display = gr.Textbox(
                label="💭 Live Conversation (FREE & Ultra-Realistic)",
                lines=15,
                interactive=False,
                show_copy_button=True
            )
    
    # Event Handlers
    start_call_btn.click(
        fn=start_call_handler,
        outputs=[transcription_output, audio_output, conversation_display]
    )
    
    end_call_btn.click(
        fn=end_call_handler,
        outputs=[transcription_output, audio_output, conversation_display]
    )
    
    process_btn.click(
        fn=process_audio_handler,
        inputs=[audio_input],
        outputs=[transcription_output, audio_output, conversation_display]
    )
    
    audio_input.stop_recording(
        fn=process_audio_handler,
        inputs=[audio_input],
        outputs=[transcription_output, audio_output, conversation_display]
    )

if __name__ == "__main__":
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True
    )