app.py

import gradio as gr
import whisper
from transformers import pipeline
import requests
import cv2
import string
import numpy as np
import tensorflow as tf
import edge_tts
import asyncio
import tempfile

# Load models
whisper_model = whisper.load_model("base")
sentiment_analysis = pipeline(
    "sentiment-analysis", framework="pt", model="SamLowe/roberta-base-go_emotions"
)

def load_sign_language_model():
    return tf.keras.models.load_model("best_model.h5")

sign_language_model = load_sign_language_model()

# Get available voices asynchronously
async def get_voices():
    voices = await edge_tts.list_voices()
    return {
        f"{v['ShortName']} - {v['Locale']} ({v['Gender']})": v["ShortName"]
        for v in voices
    }

# Audio-based functions
def analyze_sentiment(text):
    results = sentiment_analysis(text)
    sentiment_results = {result["label"]: result["score"] for result in results}
    return sentiment_results

def display_sentiment_results(sentiment_results, option):
    sentiment_text = ""
    for sentiment, score in sentiment_results.items():
        if option == "Sentiment Only":
            sentiment_text += f"{sentiment}\n"
        elif option == "Sentiment + Score":
            sentiment_text += f"{sentiment}: {score:.2f}\n"
    return sentiment_text

def search_text(text, api_key):
    api_endpoint = "https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash-latest:generateContent"
    headers = {"Content-Type": "application/json"}
    payload = {"contents": [{"parts": [{"text": text}]}]}

    try:
        response = requests.post(api_endpoint, headers=headers, json=payload, params={"key": api_key})
        response.raise_for_status()
        response_json = response.json()
        if "candidates" in response_json and response_json["candidates"]:
            content_parts = response_json["candidates"][0]["content"]["parts"]
            if content_parts:
                return content_parts[0]["text"].strip()
        return "No relevant content found."
    except requests.exceptions.RequestException as e:
        return {"error": str(e)}

async def text_to_speech(text, voice, rate, pitch):
    if not text.strip():
        return None, gr.Warning("Please enter text to convert.")
    if not voice:
        return None, gr.Warning("Please select a voice.")

    voice_short_name = voice.split(" - ")[0]
    rate_str = f"{rate:+d}%"
    pitch_str = f"{pitch:+d}Hz"
    communicate = edge_tts.Communicate(text, voice_short_name, rate=rate_str, pitch=pitch_str)
    
    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
        tmp_path = tmp_file.name
        await communicate.save(tmp_path)
    
    return tmp_path, None

async def tts_interface(text, voice, rate, pitch):
    return await text_to_speech(text, voice, rate, pitch)

async def inference_audio(audio, sentiment_option, api_key, tts_voice, tts_rate, tts_pitch):
    if audio is None:
        return "No audio file provided.", "", "", "", None

    audio = whisper.load_audio(audio)
    audio = whisper.pad_or_trim(audio)
    mel = whisper.log_mel_spectrogram(audio).to(whisper_model.device)

    _, probs = whisper_model.detect_language(mel)
    lang = max(probs, key=probs.get)

    options = whisper.DecodingOptions(fp16=False)
    result = whisper.decode(whisper_model, mel, options)

    sentiment_results = analyze_sentiment(result.text)
    sentiment_output = display_sentiment_results(sentiment_results, sentiment_option)

    search_results = search_text(result.text, api_key)

    explanation_audio, _ = await tts_interface(search_results, tts_voice, tts_rate, tts_pitch)

    return lang.upper(), result.text, sentiment_output, search_results, explanation_audio

async def classify_sign_language(image, api_key):
    img = np.array(image)
    gray_img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    gray_img = cv2.resize(gray_img, (28, 28))
    normalized_img = gray_img / 255.0
    input_img = np.expand_dims(normalized_img, axis=0)

    output = sign_language_model.predict(input_img)
    output = np.argmax(output, axis=1).item()
    uppercase_alphabet = string.ascii_uppercase
    output = output + 1 if output > 7 else output
    pred = uppercase_alphabet[output]

    explanation = search_text(f"Explain the American Sign Language letter '{pred}'.", api_key)
    explanation_audio, _ = await tts_interface(explanation, None, 0, 0)

    return pred, explanation, explanation_audio

async def process_input(input_type, audio=None, image=None, sentiment_option=None, api_key=None, tts_voice=None, tts_rate=0, tts_pitch=0):
    if input_type == "Audio":
        return await inference_audio(audio, sentiment_option, api_key, tts_voice, tts_rate, tts_pitch)
    elif input_type == "Image":
        return await classify_sign_language(image, api_key)

async def main():
    voices = await get_voices()

    with gr.Blocks() as demo:
        gr.Markdown("# Speak & Sign AI Assistant")

        with gr.Row():
            with gr.Column():
                gr.Markdown("### User Input")
                input_type = gr.Radio(label="Choose Input Type", choices=["Audio", "Image"], value="Audio")
                api_key_input = gr.Textbox(label="API Key", placeholder="Your API key here", type="password")
                audio_input = gr.Audio(label="Upload or Record Audio", type="filepath")
                sentiment_option = gr.Radio(choices=["Sentiment Only", "Sentiment + Score"], label="Sentiment Output", value="Sentiment Only")
                image_input = gr.Image(label="Upload Image", type="pil", visible=False)
                tts_voice = gr.Dropdown(label="Select Voice", choices=[""] + list(voices.keys()), value="")
                tts_rate = gr.Slider(minimum=-50, maximum=50, value=0, label="Speech Rate Adjustment (%)", step=1)
                tts_pitch = gr.Slider(minimum=-20, maximum=20, value=0, label="Pitch Adjustment (Hz)", step=1)

                def update_visibility(input_type):
                    return gr.update(visible=input_type == "Audio"), gr.update(visible=input_type == "Image")

                input_type.change(update_visibility, inputs=[input_type], outputs=[audio_input, image_input])
                submit_btn = gr.Button("Submit")

            with gr.Column():
                gr.Markdown("### Bot Response")
                lang_str = gr.Textbox(label="Detected Language", interactive=False)
                text = gr.Textbox(label="Transcription or Prediction", interactive=False)
                sentiment_output = gr.Textbox(label="Sentiment Analysis Results", interactive=False)
                search_results = gr.Textbox(label="Explanation", interactive=False)
                audio_output = gr.Audio(label="Generated Explanation Audio", type="filepath", interactive=False)

        submit_btn.click(process_input, inputs=[input_type, audio_input, image_input, sentiment_option, api_key_input, tts_voice, tts_rate, tts_pitch], outputs=[lang_str, text, sentiment_output, search_results, audio_output])

    demo.launch(share=True)

asyncio.create_task(main())