app.py

import gradio as gr
import numpy as np
import cv2
import librosa
import speech_recognition as sr
import tempfile
import wave
import optimum
import os
import tensorflow as tf
from tensorflow.keras.preprocessing.text import tokenizer_from_json
from tensorflow.keras.models import load_model, model_from_json
from sklearn.preprocessing import StandardScaler
from tensorflow.keras.preprocessing.sequence import pad_sequences
import nltk
from nltk.corpus import stopwords
from nltk.stem import WordNetLemmatizer
import pickle
import json
from tensorflow.keras.preprocessing.image import img_to_array, load_img
from collections import Counter
from pydub import AudioSegment
import ffmpeg

nltk.download('punkt')       # Tokenizer
nltk.download('wordnet')     # WordNet lemmatizer
nltk.download('stopwords')   # Stopwords

# Load the text model
with open('model_architecture_for_text_emotion_updated_json.json', 'r') as json_file:
    model_json = json_file.read()
text_model = model_from_json(model_json)
text_model.load_weights("model_for_text_emotion_updated(1).keras")

# Load the encoder and scaler for audio
with open('encoder.pkl', 'rb') as file:
    encoder = pickle.load(file)
with open('scaler.pkl', 'rb') as file:
    scaler = pickle.load(file)

# Load the tokenizer for text
with open('tokenizer.json') as json_file:
    tokenizer_json = json.load(json_file)
tokenizer = tokenizer_from_json(tokenizer_json)

# Load the audio model
audio_model = load_model('my_model.h5')

# Load the image model
image_model = load_model('model_emotion.h5')

# Initialize NLTK
lemmatizer = WordNetLemmatizer()
stop_words = set(stopwords.words('english'))

# Preprocess text function
def preprocess_text(text):
    tokens = nltk.word_tokenize(text.lower())
    tokens = [word for word in tokens if word.isalnum() and word not in stop_words]
    lemmatized_tokens = [lemmatizer.lemmatize(word) for word in tokens]
    return ' '.join(lemmatized_tokens)

# Extract features from audio
import numpy as np
import torch
import torchaudio
import torchaudio.transforms as T

def extract_features(data, sample_rate):
    # List to collect all features
    features = []

    # Zero Crossing Rate (ZCR)
    zcr = T.ZeroCrossingRate()(data)
    features.append(torch.mean(zcr).numpy())

    # Chroma Short-Time Fourier Transform (STFT)
    stft = T.MelSpectrogram(sample_rate)(data)
    chroma_stft = torch.mean(stft, dim=-1).numpy()  # Take mean across the time dimension
    features.append(chroma_stft)

    # Mel Frequency Cepstral Coefficients (MFCC)
    mfcc_transform = T.MFCC(sample_rate=sample_rate, melkwargs={"n_fft": 400, "hop_length": 160, "n_mels": 23})
    mfcc = mfcc_transform(data)
    mfcc = torch.mean(mfcc, dim=-1).numpy()  # Take mean across the time dimension
    features.append(mfcc)

    # Root Mean Square Energy (RMS)
    rms = torch.mean(T.MelSpectrogram(sample_rate)(data), dim=-1)  # Same as RMS feature extraction
    features.append(rms.numpy())

    # Mel Spectrogram
    mel = T.MelSpectrogram(sample_rate)(data)
    mel = torch.mean(mel, dim=-1).numpy()  # Take mean across the time dimension
    features.append(mel)

    # Convert list of features to a single numpy array
    result = np.hstack(features)

    return result


# Predict emotion from text
def find_emotion_using_text(sample_rate, audio_data, recognizer):
    mapping = {0: "anger", 1: "disgust", 2: "fear", 3: "joy", 4: "neutral", 5: "sadness", 6: "surprise"}
    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_audio_file:
        temp_audio_path = temp_audio_file.name
        
        with wave.open(temp_audio_path, 'w') as wf:
            wf.setnchannels(1)
            wf.setsampwidth(2)
            wf.setframerate(sample_rate)
            wf.writeframes(audio_data.tobytes())
        
    with sr.AudioFile(temp_audio_path) as source:
        audio_record = recognizer.record(source)
        text = recognizer.recognize_google(audio_record)
        pre_text = preprocess_text(text)
        title_seq = tokenizer.texts_to_sequences([pre_text])
        padded_title_seq = pad_sequences(title_seq, maxlen=35, padding='post', truncating='post')
        inp1 = np.array(padded_title_seq)
        text_prediction = text_model.predict(inp1)
    
    os.remove(temp_audio_path)
    max_index = text_prediction.argmax()
    return mapping[max_index],text

# Predict emotion from audio
def predict_emotion(audio_data):
    sample_rate, data = audio_data
    data = data.flatten()
    
    if data.dtype != np.float32:
        data = data.astype(np.float32)
    data = data / np.max(np.abs(data))
    
    features = extract_features(data, sample_rate)
    features = np.expand_dims(features, axis=0)
    
    if features.ndim == 3:
        features = np.squeeze(features, axis=2)
    elif features.ndim != 2:
        raise ValueError("Features array has unexpected dimensions.")
    
    scaled_features = scaler.transform(features)
    scaled_features = np.expand_dims(scaled_features, axis=2)
    
    prediction = audio_model.predict(scaled_features)
    emotion_index = np.argmax(prediction)
    
    num_classes = len(encoder.categories_[0])
    emotion_array = np.zeros((1, num_classes))
    emotion_array[0, emotion_index] = 1
    
    emotion_label = encoder.inverse_transform(emotion_array)[0]
    return emotion_label

def preprocess_image(image):
    image = load_img(image, target_size=(48, 48), color_mode="grayscale")
    image = img_to_array(image)
    image = np.expand_dims(image, axis=0)
    image = image / 255.0
    return image

# Predict emotion from image
def predict_emotion_from_image(image):
    preprocessed_image = preprocess_image(image)
    prediction = image_model.predict(preprocessed_image)
    emotion_index = np.argmax(prediction)
    
    mapping = {0: "anger", 1: "disgust", 2: "fear", 3: "joy", 4: "neutral", 5: "sadness", 6: "surprise"}
    return mapping[emotion_index]

def process_video(video_path):
    cap = cv2.VideoCapture(video_path)
    frame_rate = cap.get(cv2.CAP_PROP_FPS)
    
    frame_count = 0
    predictions = []
    
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        
        # Process every nth frame (to speed up processing)
        if frame_count % int(frame_rate) == 0:
            # Convert frame to grayscale as required by your model
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            frame = cv2.resize(frame, (48, 48))  # Resize to match model input size
            frame = img_to_array(frame)
            frame = np.expand_dims(frame, axis=0) / 255.0
            
            # Predict emotion
            prediction = image_model.predict(frame)
            predictions.append(np.argmax(prediction))
        
        frame_count += 1
    
    cap.release()
    cv2.destroyAllWindows()
    
    # Find the most common prediction
    most_common_emotion = Counter(predictions).most_common(1)[0][0]
    mapping = {0: "anger", 1: "disgust", 2: "fear", 3: "joy", 4: "neutral", 5: "sadness", 6: "surprise"}
    return mapping[most_common_emotion]



def process_audio_from_video(video_path):
    text_emotion = "Error in text processing"  # Initialize text_emotion
    text=""
    try:
        # Load the video using an alternative library (e.g., ffmpeg or cv2)
        import ffmpeg

        audio_output = tempfile.NamedTemporaryFile(delete=False, suffix=".wav").name
        ffmpeg.input(video_path).output(audio_output, format="wav").run(quiet=True)

        recognizer = sr.Recognizer()

        with sr.AudioFile(audio_output) as source:
            audio_record = recognizer.record(source)
            text = recognizer.recognize_google(audio_record)
            pre_text = preprocess_text(text)
            title_seq = tokenizer.texts_to_sequences([pre_text])
            padded_title_seq = pad_sequences(title_seq, maxlen=35, padding='post', truncating='post')
            inp1 = np.array(padded_title_seq)
            text_prediction = text_model.predict(inp1)

        os.remove(audio_output)

        max_index = text_prediction.argmax()
        text_emotion = {0: "anger", 1: "disgust", 2: "fear", 3: "joy", 4: "neutral", 5: "sadness", 6: "surprise"}[max_index]

    except Exception as e:
        print(f"Error processing text from audio: {e}")
        text_emotion = "Error in text processing"

    try:
        # Extract audio features for emotion recognition
        sample_rate, data = librosa.load(video_path, sr=None, mono=True)
        data = data.flatten()

        if data.dtype != np.float32:
            data = data.astype(np.float32)
        data = data / np.max(np.abs(data))

        features = extract_features(data, sample_rate)
        features = np.expand_dims(features, axis=0)
        scaled_features = scaler.transform(features)
        scaled_features = np.expand_dims(scaled_features, axis=2)

        prediction = audio_model.predict(scaled_features)
        emotion_index = np.argmax(prediction)

        num_classes = len(encoder.categories_[0])
        emotion_array = np.zeros((1, num_classes))
        emotion_array[0, emotion_index] = 1

        audio_emotion = encoder.inverse_transform(emotion_array)[0]

    except Exception as e:
        print(f"Error processing audio features: {e}")
        audio_emotion = "Error in audio processing"

    return text_emotion, audio_emotion,text



import torch
import gradio as gr
from huggingface_hub import InferenceClient
from transformers import AutoTokenizer, AutoModelForCausalLM

# Hugging Face Inference Client (equivalent to the reference code's client)
client = InferenceClient("TheBloke/Mistral-7B-Instruct-v0.1-GPTQ")

# Tokenizer and model loading (still necessary if you want to process locally)
tokenizer = AutoTokenizer.from_pretrained("TheBloke/Mistral-7B-Instruct-v0.1-GPTQ")
model = AutoModelForCausalLM.from_pretrained("TheBloke/Mistral-7B-Instruct-v0.1-GPTQ")


def respond(message, history: list[tuple[str, str]], system_message, max_tokens, temperature, top_p):
    messages = [{"role": "system", "content": system_message}]
    
    # Format history with user and bot messages
    for val in history:
        if val[0]:
            messages.append({"role": "user", "content": val[0]})
        if val[1]:
            messages.append({"role": "assistant", "content": val[1]})

    messages.append({"role": "user", "content": message})

    response = ""
    
    # Stream response from the model
    for message in client.chat_completion(
        messages,
        max_tokens=max_tokens,
        stream=True,
        temperature=temperature,
        top_p=top_p,
    ):
        token = message.choices[0].delta.content
        response += token
        yield response


# Function to handle video processing and interaction
def transcribe_and_predict_video(video, user_input, chat_history=[]):
    # Process the video for emotions (use your own emotion detection functions)

    if chat_history is None:
        chat_history = []
    image_emotion = process_video(video)
    text_emotion, audio_emotion,text = process_audio_from_video(video)
    em = [image_emotion, text_emotion, audio_emotion]

    # Format the conversation history
    history_text = "".join([f"User ({msg[2]}): {msg[0]}\nBot: {msg[1]}\n" for msg in chat_history])

    # Construct the prompt with emotion context and history
    prompt = f"""
    You are a helpful AI assistant. Respond like a human while considering the user's emotion.

    User's Emotion: {em}
    video text context: {text}
    Conversation History:
    {history_text}

    User ({em}): {user_input}
    Bot:"""

    # Tokenize input
    inputs = tokenizer(prompt, return_tensors="pt").to("cpu")

    # Generate response
    output = model.generate(**inputs, max_length=512, temperature=0.7, top_p=0.9, do_sample=True)
    response = tokenizer.decode(output[0], skip_special_tokens=True).split("Bot:")[-1].strip()

    # Store the current emotion for the user input (modify emotion detection as needed)
    emotion = detect_emotion(user_input)  # Assuming `detect_emotion` is a function that returns the user's emotion

    # Update the chat history with the current conversation and emotion
    chat_history.append((user_input, response, emotion))

    return response, chat_history


# Gradio interface setup
iface = gr.Interface(
    fn=transcribe_and_predict_video,
    inputs=[gr.Video(), gr.Textbox(), gr.State()],  # Accepting video input, user text, and chat history
    outputs=[gr.Textbox(), gr.State()],  # Output is the response and updated chat history
    title="Multimodal Emotion Recognition from Video",
    description="Upload a video to get text, audio, and image emotion predictions and interact with the chatbot."
)

# Launch the Gradio interface
if __name__ == "__main__":
    iface.launch()