Update app.py

app.py

@@ -1,366 +1,17 @@
 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
+from transformers import pipeline
 
-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()
+# Load the DeepSeek model
+pipe = pipeline("text-generation", model="deepseek-ai/DeepSeek-R1", trust_remote_code=True)
 
+# Function to interact with the chatbot
+def chat_with_bot(message, chat_history):
+    messages = [{"role": "user", "content": message}]
+    response = pipe(messages, max_length=512)
+    return response[0]["generated_text"]
 
+# Create Gradio UI
+interface = gr.ChatInterface(fn=chat_with_bot, title="DeepSeek AI Chatbot")
 
+# Launch the chatbot
+interface.launch()