app.py

import gradio as gr
import requests
import os
import time
import re
import logging
import tempfile
import folium
import concurrent.futures
import torch
from PIL import Image
from datetime import datetime
from transformers import pipeline, AutoModelForSpeechSeq2Seq, AutoProcessor
from googlemaps import Client as GoogleMapsClient
from gtts import gTTS
from diffusers import StableDiffusionPipeline
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_pinecone import PineconeVectorStore
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain.chains.conversation.memory import ConversationBufferWindowMemory
from langchain.agents import Tool, initialize_agent
from huggingface_hub import login
from transformers.models.speecht5.number_normalizer import EnglishNumberNormalizer
from parler_tts import ParlerTTSForConditionalGeneration
from transformers import AutoTokenizer, AutoFeatureExtractor, set_seed
from scipy.io.wavfile import write as write_wav
from pydub import AudioSegment
from string import punctuation
import librosa
from pathlib import Path
import torchaudio

# Check if the token is already set in the environment variables
hf_token = os.getenv("HF_TOKEN")
if hf_token is None:
    print("Please set your Hugging Face token in the environment variables.")
else:
    login(token=hf_token)

logging.basicConfig(level=logging.DEBUG)

embeddings = OpenAIEmbeddings(api_key=os.environ['OPENAI_API_KEY'])

from pinecone import Pinecone
pc = Pinecone(api_key=os.environ['PINECONE_API_KEY'])

index_name = "birminghumsummary1"
vectorstore = PineconeVectorStore(index_name=index_name, embedding=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={'k': 5})

chat_model = ChatOpenAI(api_key=os.environ['OPENAI_API_KEY'], temperature=0, model='gpt-4o')

conversational_memory = ConversationBufferWindowMemory(
    memory_key='chat_history',
    k=10,
    return_messages=True
)

def get_current_time_and_date():
    now = datetime.now()
    return now.strftime("%Y-%m-%d %H:%M:%S")

current_time_and_date = get_current_time_and_date()


# def fetch_local_events():
#     api_key = os.environ['SERP_API']
#     url = f'https://serpapi.com/search.json?engine=google_events&q=Events+in+Birmingham&hl=en&gl=us&api_key={api_key}'
#     response = requests.get(url)
#     if response.status_code == 200:
#         events_results = response.json().get("events_results", [])
#         events_html = """
#         <h2 style="font-family: 'Georgia', serif; color: #ff0000; background-color: #f8f8f8; padding: 10px; border-radius: 10px;">Local Events</h2>
#         <style>
#             table {
#                 font-family: 'Verdana', sans-serif;
#                 color: #333;
#                 border-collapse: collapse;
#                 width: 100%;
#             }
#             th, td {
#                 border: 1px solid #ddd;
#                 padding: 8px;
#             }
#             th {
#                 background-color: #f2f2f2;
#                 color: #333;
#                 text-align: left;
#             }
#             tr:hover {
#                 background-color: #f5f5f5;
#             }
#             .event-link {
#                 color: #1E90FF;
#                 text-decoration: none;
#             }
#             .event-link:hover {
#                 text-decoration: underline;
#             }
#         </style>
#         <table>
#             <tr>
#                 <th>Title</th>
#                 <th>Date and Time</th>
#                 <th>Location</th>
#             </tr>
#         """
#         for event in events_results:
#             title = event.get("title", "No title")
#             date_info = event.get("date", {})
#             date = f"{date_info.get('start_date', '')} {date_info.get('when', '')}".replace("{", "").replace("}", "")
#             location = event.get("address", "No location")
#             if isinstance(location, list):
#                 location = " ".join(location)
#             location = location.replace("[", "").replace("]", "")
#             link = event.get("link", "#")
#             events_html += f"""
#             <tr>
#                 <td><a class='event-link' href='{link}' target='_blank'>{title}</a></td>
#                 <td>{date}</td>
#                 <td>{location}</td>
#             </tr>
#             """
#         events_html += "</table>"
#         return events_html
#     else:
#         return "<p>Failed to fetch local events</p>"

def fetch_local_events():
   
    
    api_key = os.environ['SERP_API']
    url = f'https://serpapi.com/search.json?engine=google_events&q=Events+in+Birmingham&hl=en&gl=us&api_key={api_key}'
    response = requests.get(url)
    if response.status_code == 200:
        events_results = response.json().get("events_results", [])
        events_html = """
        <h2 style="font-family: 'Georgia', serif; color: #ff0000; background-color: #f8f8f8; padding: 10px; border-radius: 10px;">Local Events</h2>
        <style>
            table {
                font-family: 'Verdana', sans-serif;
                color: #333;
                border-collapse: collapse;
                width: 100%;
            }
            th, td {
                border: 1px solid #fff !important;
                padding: 8px;
            }
            th {
                background-color: #f2f2f2;
                color: #333;
                text-align: left;
            }
            tr:hover {
                background-color: #f5f5f5;
            }
            .event-link {
                color: #1E90FF;
                text-decoration: none;
            }
            .event-link:hover {
                text-decoration: underline;
            }
        </style>
        <table>
            <tr>
                <th>Title</th>
                <th>Date and Time</th>
                <th>Location</th>
            </tr>
        """
        for event in events_results:
            title = event.get("title", "No title")
            date_info = event.get("date", {})
            date = f"{date_info.get('start_date', '')} {date_info.get('when', '')}".replace("{", "").replace("}", "")
            location = event.get("address", "No location")
            if isinstance(location, list):
                location = " ".join(location)
            location = location.replace("[", "").replace("]", "")
            link = event.get("link", "#")
            events_html += f"""
            <tr>
                <td><a class='event-link' href='{link}' target='_blank'>{title}</a></td>
                <td>{date}</td>
                <td>{location}</td>
            </tr>
            """
        events_html += "</table>"
        return events_html
    else:
        return "<p>Failed to fetch local events</p>"




def fetch_local_weather():
    try:
        api_key = os.environ['WEATHER_API']
        url = f'https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline/birmingham?unitGroup=metric&include=events%2Calerts%2Chours%2Cdays%2Ccurrent&key={api_key}'
        response = requests.get(url)
        response.raise_for_status()
        jsonData = response.json()
        
        current_conditions = jsonData.get("currentConditions", {})
        temp_celsius = current_conditions.get("temp", "N/A")
        
        if temp_celsius != "N/A":
            temp_fahrenheit = int((temp_celsius * 9/5) + 32)
        else:
            temp_fahrenheit = "N/A"
            
        condition = current_conditions.get("conditions", "N/A")
        humidity = current_conditions.get("humidity", "N/A")

        weather_html = f"""
        <div class="weather-theme">
            <h2 style="font-family: 'Georgia', serif; color: #ff0000; background-color: #f8f8f8; padding: 10px; border-radius: 10px;">Local Weather</h2>
            <div class="weather-content">
                <div class="weather-icon">
                    <img src="https://www.weatherbit.io/static/img/icons/{get_weather_icon(condition)}.png" alt="{condition}" style="width: 100px; height: 100px;">
                </div>
                <div class="weather-details">
                    <p style="font-family: 'Verdana', sans-serif; color: #333; font-size: 1.2em;">Temperature: {temp_fahrenheit}°F</p>
                    <p style="font-family: 'Verdana', sans-serif; color: #333; font-size: 1.2em;">Condition: {condition}</p>
                    <p style="font-family: 'Verdana', sans-serif; color: #333; font-size: 1.2em;">Humidity: {humidity}%</p>
                </div>
            </div>
        </div>
        <style>
            .weather-theme {{
                animation: backgroundAnimation 10s infinite alternate;
                border-radius: 10px;
                padding: 10px;
                margin-bottom: 15px;
                background: linear-gradient(45deg, #ffcc33, #ff6666, #ffcc33, #ff6666);
                background-size: 400% 400%;
                box-shadow: 0 4px 8px rgba(0, 0, 0, 0.1);
                transition: box-shadow 0.3s ease, background-color 0.3s ease;
            }}
            .weather-theme:hover {{
                box-shadow: 0 8px 16px rgba(0, 0, 0, 0.2);
                background-position: 100% 100%;
            }}
            @keyframes backgroundAnimation {{
                0% {{ background-position: 0% 50%; }}
                100% {{ background-position: 100% 50%; }}
            }}
            .weather-content {{
                display: flex;
                align-items: center;
            }}
            .weather-icon {{
                flex: 1;
            }}
            .weather-details {{
                flex: 3;
            }}
        </style>
        """
        return weather_html
    except requests.exceptions.RequestException as e:
        return f"<p>Failed to fetch local weather: {e}</p>"

def get_weather_icon(condition):
    condition_map = {
        "Clear": "c01d",
        "Partly Cloudy": "c02d",
        "Cloudy": "c03d",
        "Overcast": "c04d",
        "Mist": "a01d",
        "Patchy rain possible": "r01d",
        "Light rain": "r02d",
        "Moderate rain": "r03d",
        "Heavy rain": "r04d",
        "Snow": "s01d",
        "Thunderstorm": "t01d",
        "Fog": "a05d",
    }
    return condition_map.get(condition, "c04d")

template1 = """You are an expert concierge who is helpful and a renowned guide for Birmingham,Alabama. Based on weather being a sunny bright day and the today's date is 1st july 2024, use the following pieces of context, 
memory, and message history, along with your knowledge of perennial events in Birmingham,Alabama, to answer the question at the end. If you don't know the answer, just say "Homie, I need to get more data for this," and don't try to make up an answer. 
Use fifteen sentences maximum. Keep the answer as detailed as possible. Always include the address, time, date, and
event type and description.And also add this Birmingham,AL at the end of each address, Always say "It was my pleasure!" at the end of the answer.
{context}
Question: {question}
Helpful Answer:"""


# template2 = """You are an expert concierge who is helpful and a renowned guide for Birmingham,Alabama. Based on today's weather being a sunny bright day and today's date is 1st july 2024, take the location or address but don't show the location or address on the output prompts. Use the following pieces of context, 
# memory, and message history, along with your knowledge of perennial events in Birmingham,Alabama, to answer the question at the end. If you don't know the answer, just say "Homie, I need to get more data for this," and don't try to make up an answer. 
# Keep the answer short and sweet and crisp. Always say "It was my pleasure!" at the end of the answer.
# {context}
# Question: {question}
# Helpful Answer:"""

template2 = """You are an expert concierge who is helpful and a renowned guide for Birmingham,Alabama. Based on today's weather being a sunny bright day and today's date is 16th july 2024, take the location or address but don't show the location or address on the output prompts. Use the following pieces of context, 
memory, and message history, along with your knowledge of perennial events in Birmingham,Alabama, to answer the question at the end. If you don't know the answer, just say "Homie, I need to get more data for this," and don't try to make up an answer. 
Keep the answer short ,sweet and crisp and in one shot. Always say "It was my pleasure!" at the end of the answer.
{context}
Question: {question}
Helpful Answer:"""

QA_CHAIN_PROMPT_1 = PromptTemplate(input_variables=["context", "question"], template=template1)
QA_CHAIN_PROMPT_2 = PromptTemplate(input_variables=["context", "question"], template=template2)

def build_qa_chain(prompt_template):
    qa_chain = RetrievalQA.from_chain_type(
        llm=chat_model,
        chain_type="stuff",
        retriever=retriever,
        chain_type_kwargs={"prompt": prompt_template}
    )
    tools = [
        Tool(
            name='Knowledge Base',
            func=qa_chain,
            description='Use this tool when answering general knowledge queries to get more information about the topic'
        )
    ]
    return qa_chain, tools

def initialize_agent_with_prompt(prompt_template):
    qa_chain, tools = build_qa_chain(prompt_template)
    agent = initialize_agent(
        agent='chat-conversational-react-description',
        tools=tools,
        llm=chat_model,
        verbose=False,
        max_iteration=5,
        early_stopping_method='generate',
        memory=conversational_memory
    )
    return agent


def generate_answer(message, choice):
    logging.debug(f"generate_answer called with prompt_choice: {choice}")
    
    if choice == "Details":
        agent = initialize_agent_with_prompt(QA_CHAIN_PROMPT_1)
    elif choice == "Conversational":
        agent = initialize_agent_with_prompt(QA_CHAIN_PROMPT_2)
    else:
        logging.error(f"Invalid prompt_choice: {choice}. Defaulting to 'Conversational'")
        agent = initialize_agent_with_prompt(QA_CHAIN_PROMPT_2)
    response = agent(message)

    addresses = extract_addresses(response['output'])
    return response['output'], addresses

def bot(history, choice, tts_choice):
    if not history:
        return history
    response, addresses = generate_answer(history[-1][0], choice)
    history[-1][1] = ""
    
    with concurrent.futures.ThreadPoolExecutor() as executor:
        if tts_choice == "Alpha":
            audio_future = executor.submit(generate_audio_elevenlabs, response)
        elif tts_choice == "Beta":
            audio_future = executor.submit(generate_audio_parler_tts, response)
        elif tts_choice == "Gamma":
            audio_future = executor.submit(generate_audio_mars5, response)
        elif tts_choice == "Delta":
            audio_future = executor.submit(generate_audio_fishaudio, response)
        
        for character in response:
            history[-1][1] += character
            time.sleep(0.05)
            yield history, None
        
        audio_path = audio_future.result()
        yield history, audio_path

def add_message(history, message):
    history.append((message, None))
    return history, gr.Textbox(value="", interactive=True, placeholder="Enter message or upload file...", show_label=False)
#----------------------------------part 1-------------------------
# def generate_answer(message, choice):
#     logging.debug(f"generate_answer called with prompt_choice: {choice}")
    
#     if choice == "Details":
#         agent = initialize_agent_with_prompt(QA_CHAIN_PROMPT_1)
#     elif choice == "Conversational":
#         agent = initialize_agent_with_prompt(QA_CHAIN_PROMPT_2)
#     else:
#         logging.error(f"Invalid prompt_choice: {choice}. Defaulting to 'Conversational'")
#         agent = initialize_agent_with_prompt(QA_CHAIN_PROMPT_2)
#     response = agent(message)

#     addresses = extract_addresses(response['output'])
#     return response['output'], addresses



# # def bot(history, choice, tts_choice):
# #     if not history:
# #         return history
# #     response_pair, addresses = generate_answer(history[-1][0], choice)
# #     history[-1] = response_pair  # Update bot response correctly

# #     with concurrent.futures.ThreadPoolExecutor() as executor:
# #         if tts_choice == "Alpha":
# #             audio_future = executor.submit(generate_audio_elevenlabs, response_pair[1])
# #         elif tts_choice == "Beta":
# #             audio_future = executor.submit(generate_audio_parler_tts, response_pair[1])
# #         elif tts_choice == "Gamma":
# #             audio_future = executor.submit(generate_audio_mars5, response_pair[1])
        
# #         for character in response_pair[1]:
# #             history[-1][1] += character
# #             time.sleep(0.05)
# #             yield history, None
        
# #         audio_path = audio_future.result()
# #         yield history, audio_path

# # def add_message(history, message):
# #     history = history or []
# #     history.append([message, ""])  # Ensure it is a list with two elements: message and empty response
# #     return history, gr.Textbox(value="", interactive=True, placeholder="Enter message or upload file...", show_label=False)

# def bot(history, choice, tts_choice):
#     if not history:
#         return history
#     response_pair, addresses = generate_answer(history[-1][0], choice)
#     # Ensure history has the correct format
#     if len(history[-1]) == 1:
#         history[-1].append("")
#     elif len(history[-1]) == 0:
#         history[-1] = [history[-1][0], ""]

#     with concurrent.futures.ThreadPoolExecutor() as executor:
#         if tts_choice == "Alpha":
#             audio_future = executor.submit(generate_audio_elevenlabs, response_pair[1])
#         elif tts_choice == "Beta":
#             audio_future = executor.submit(generate_audio_parler_tts, response_pair[1])
#         elif tts_choice == "Gamma":
#             audio_future = executor.submit(generate_audio_mars5, response_pair[1])
        
#         for character in response_pair[1]:
#             history[-1][1] += character
#             time.sleep(0.05)
#             yield history, None
        
#         audio_path = audio_future.result()
#         yield history, audio_path

# def add_message(history, message):
#     history = history or []
#     history.append([message, ""])  # Ensure it is a list with two elements: message and empty response
#     return history, gr.Textbox(value="", interactive=True, placeholder="Enter message or upload file...", show_label=False)
#--------------------------------------------------------
# def generate_voice_response(history, tts_choice):
#     if not history:
#         return None
    
#     response = history[-1][1]
#     with concurrent.futures.ThreadPoolExecutor() as executor:
#         if tts_choice == "Alpha":
#             audio_future = executor.submit(generate_audio_elevenlabs, response)
#         elif tts_choice == "Beta":
#             audio_future = executor.submit(generate_audio_parler_tts, response)
#         elif tts_choice == "Gamma":
#             audio_future = executor.submit(generate_audio_mars5, response)
        
#         audio_path = audio_future.result()
#     return audio_path

def print_like_dislike(x: gr.LikeData):
    print(x.index, x.value, x.liked)

def extract_addresses(response):
    if not isinstance(response, str):
        response = str(response)
    address_patterns = [
        r'([A-Z].*,\sBirmingham,\sAL\s\d{5})',
        r'(\d{4}\s.*,\sBirmingham,\sAL\s\d{5})',
        r'([A-Z].*,\sAL\s\d{5})',
        r'([A-Z].*,.*\sSt,\sBirmingham,\sAL\s\d{5})',
        r'([A-Z].*,.*\sStreets,\sBirmingham,\sAL\s\d{5})',
        r'(\d{2}.*\sStreets)',
        r'([A-Z].*\s\d{2},\sBirmingham,\sAL\s\d{5})',
        r'([a-zA-Z]\s Birmingham)',
        r'([a-zA-Z].*,\sBirmingham,\sAL)'
    ]
    addresses = []
    for pattern in address_patterns:
        addresses.extend(re.findall(pattern, response))
    return addresses

all_addresses = []

# def generate_map(location_names):
#     global all_addresses
#     all_addresses.extend(location_names)
    
#     api_key = os.environ['GOOGLEMAPS_API_KEY']
#     gmaps = GoogleMapsClient(key=api_key)
    
#     m = folium.Map(location=[33.5175,-86.809444], zoom_start=16)
    
#     for location_name in all_addresses:
#         geocode_result = gmaps.geocode(location_name)
#         if geocode_result:
#             location = geocode_result[0]['geometry']['location']
#             folium.Marker(
#                 [location['lat'], 'lng'],
#                 tooltip=f"{geocode_result[0]['formatted_address']}"
#             ).add_to(m)
    
#     map_html = m._repr_html_()
#     return map_html

def generate_map(location_names):
    global all_addresses
    all_addresses.extend(location_names)
    
    api_key = os.environ['GOOGLEMAPS_API_KEY']
    gmaps = GoogleMapsClient(key=api_key)
    
    m = folium.Map(location=[33.5175, -86.809444], zoom_start=20)
    
    for location_name in all_addresses:
        geocode_result = gmaps.geocode(location_name)
        if geocode_result:
            location = geocode_result[0]['geometry']['location']
            folium.Marker(
                [location['lat'], location['lng']],
                tooltip=f"{geocode_result[0]['formatted_address']}"
            ).add_to(m)
    
    map_html = m._repr_html_()
    return map_html


def fetch_local_news():
    api_key = os.environ['SERP_API']
    url = f'https://serpapi.com/search.json?engine=google_news&q=birmingham headline&api_key={api_key}'
    response = requests.get(url)
    if response.status_code == 200:
        results = response.json().get("news_results", [])
        news_html = """
        <h2 style="font-family: 'Georgia', serif; color: #ff0000; background-color: #f8f8f8; padding: 10px; border-radius: 10px;">Birmingham Today</h2>
        <style>
            .news-item {
                font-family: 'Verdana', sans-serif;
                color: #333;
                background-color: #f0f8ff;
                margin-bottom: 15px;
                padding: 10px;
                border-radius: 5px;
                transition: box-shadow 0.3s ease, background-color 0.3s ease;
                font-weight: bold;
            }
            .news-item:hover {
                box-shadow: 0 4px 8px rgba(0, 0, 0, 0.1);
                background-color: #e6f7ff;
            }
            .news-item a {
                color: #1E90FF;
                text-decoration: none;
                font-weight: bold;
            }
            .news-item a:hover {
                text-decoration: underline;
            }
            .news-preview {
                position: absolute;
                display: none;
                border: 1px solid #ccc;
                border-radius: 5px;
                box-shadow: 0 2px 4px rgba(0, 0, 0, 0.2);
                background-color: white;
                z-index: 1000;
                max-width: 300px;
                padding: 10px;
                font-family: 'Verdana', sans-serif;
                color: #333;
            }
        </style>
        <script>
            function showPreview(event, previewContent) {
                var previewBox = document.getElementById('news-preview');
                previewBox.innerHTML = previewContent;
                previewBox.style.left = event.pageX + 'px';
                previewBox.style.top = event.pageY + 'px';
                previewBox.style.display = 'block';
            }
            function hidePreview() {
                var previewBox = document.getElementById('news-preview');
                previewBox.style.display = 'none';
            }
        </script>
        <div id="news-preview" class="news-preview"></div>
        """
        for index, result in enumerate(results[:7]):
            title = result.get("title", "No title")
            link = result.get("link", "#")
            snippet = result.get("snippet", "")
            news_html += f"""
            <div class="news-item" onmouseover="showPreview(event, '{snippet}')" onmouseout="hidePreview()">
                <a href='{link}' target='_blank'>{index + 1}. {title}</a>
                <p>{snippet}</p>
            </div>
            """
        return news_html
    else:
        return "<p>Failed to fetch local news</p>"

import numpy as np
import torch
from transformers import pipeline, AutoModelForSpeechSeq2Seq, AutoProcessor

model_id = 'openai/whisper-large-v3'
device = "cuda:0" if torch.cuda.is_available() else "cpu"
torch_dtype = torch.float16 if torch.cuda.is_available() else torch.float32
model = AutoModelForSpeechSeq2Seq.from_pretrained(model_id, torch_dtype=torch_dtype).to(device)
processor = AutoProcessor.from_pretrained(model_id)

pipe_asr = pipeline("automatic-speech-recognition", model=model, tokenizer=processor.tokenizer, feature_extractor=processor.feature_extractor, max_new_tokens=128, chunk_length_s=15, batch_size=16, torch_dtype=torch_dtype, device=device, return_timestamps=True)

base_audio_drive = "/data/audio"

def transcribe_function(stream, new_chunk):
    try:
        sr, y = new_chunk[0], new_chunk[1]
    except TypeError:
        print(f"Error chunk structure: {type(new_chunk)}, content: {new_chunk}")
        return stream, "", None

    y = y.astype(np.float32) / np.max(np.abs(y))

    if stream is not None:
        stream = np.concatenate([stream, y])
    else:
        stream = y

    result = pipe_asr({"array": stream, "sampling_rate": sr}, return_timestamps=False)

    full_text = result.get("text", "")
    
    return stream, full_text, result

def update_map_with_response(history):
    if not history:
        return ""
    response = history[-1][1]
    addresses = extract_addresses(response)
    return generate_map(addresses)

def clear_textbox():
    return "" 

def show_map_if_details(history,choice):
    if choice in ["Details", "Conversational"]:
        return gr.update(visible=True), update_map_with_response(history)
    else:
        return gr.update(visible=False), ""

def generate_audio_elevenlabs(text):
    XI_API_KEY = os.environ['ELEVENLABS_API']
    VOICE_ID = 'd9MIrwLnvDeH7aZb61E9'
    tts_url = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID}/stream"
    headers = {
        "Accept": "application/json",
        "xi-api-key": XI_API_KEY
    }
    data = {
        "text": str(text),
        "model_id": "eleven_multilingual_v2",
        "voice_settings": {
            "stability": 1.0,
            "similarity_boost": 0.0,
            "style": 0.60,
            "use_speaker_boost": False
        }
    }
    response = requests.post(tts_url, headers=headers, json=data, stream=True)
    if response.ok:
        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as f:
            for chunk in response.iter_content(chunk_size=1024):
                f.write(chunk)
            temp_audio_path = f.name
        logging.debug(f"Audio saved to {temp_audio_path}")
        return temp_audio_path
    else:
        logging.error(f"Error generating audio: {response.text}")
        return None

repo_id = "parler-tts/parler-tts-mini-expresso"

parler_model = ParlerTTSForConditionalGeneration.from_pretrained(repo_id).to(device)
parler_tokenizer = AutoTokenizer.from_pretrained(repo_id)
parler_feature_extractor = AutoFeatureExtractor.from_pretrained(repo_id)

SAMPLE_RATE = parler_feature_extractor.sampling_rate
SEED = 42

def preprocess(text):
    number_normalizer = EnglishNumberNormalizer()
    text = number_normalizer(text).strip()
    if text[-1] not in punctuation:
        text = f"{text}."

    abbreviations_pattern = r'\b[A-Z][A-Z\.]+\b'

    def separate_abb(chunk):
        chunk = chunk.replace(".", "")
        return " ".join(chunk)

    abbreviations = re.findall(abbreviations_pattern, text)
    for abv in abbreviations:
        if abv in text:
            text = text.replace(abv, separate_abb(abv))
    return text

def chunk_text(text, max_length=250):
    words = text.split()
    chunks = []
    current_chunk = []
    current_length = 0

    for word in words:
        if current_length + len(word) + 1 <= max_length:
            current_chunk.append(word)
            current_length += len(word) + 1
        else:
            chunks.append(' '.join(current_chunk))
            current_chunk = [word]
            current_length = len(word) + 1

    if current_chunk:
        chunks.append(' '.join(current_chunk))

    return chunks

def generate_audio_parler_tts(text):
    description = "Thomas speaks with emphasis and excitement at a moderate pace with high quality."
    chunks = chunk_text(preprocess(text))
    audio_segments = []

    for chunk in chunks:
        inputs = parler_tokenizer(description, return_tensors="pt").to(device)
        prompt = parler_tokenizer(chunk, return_tensors="pt").to(device)

        set_seed(SEED)
        generation = parler_model.generate(input_ids=inputs.input_ids, prompt_input_ids=prompt.input_ids)
        audio_arr = generation.cpu().numpy().squeeze()

        temp_audio_path = os.path.join(tempfile.gettempdir(), f"parler_tts_audio_{len(audio_segments)}.wav")
        write_wav(temp_audio_path, SAMPLE_RATE, audio_arr)
        audio_segments.append(AudioSegment.from_wav(temp_audio_path))

    combined_audio = sum(audio_segments)
    combined_audio_path = os.path.join(tempfile.gettempdir(), "parler_tts_combined_audio.wav")
    combined_audio.export(combined_audio_path, format="wav")

    logging.debug(f"Audio saved to {combined_audio_path}")
    return combined_audio_path

# Load the MARS5 model
mars5, config_class = torch.hub.load('Camb-ai/mars5-tts', 'mars5_english', trust_repo=True)

def generate_audio_mars5(text):
    description = "Thomas speaks with emphasis and excitement at a moderate pace with high quality."
    kwargs_dict = {
        'temperature': 0.2,
        'top_k': -1,
        'top_p': 0.2,
        'typical_p': 1.0,
        'freq_penalty': 2.6,
        'presence_penalty': 0.4,
        'rep_penalty_window': 100,
        'max_prompt_phones': 360,
        'deep_clone': True,
        'nar_guidance_w': 3
    }

    chunks = chunk_text(preprocess(text))
    audio_segments = []

    for chunk in chunks:
        wav = torch.zeros(1, mars5.sr)  # Use a placeholder silent audio for the reference
        cfg = config_class(**{k: kwargs_dict[k] for k in kwargs_dict if k in config_class.__dataclass_fields__})
        ar_codes, wav_out = mars5.tts(chunk, wav, "", cfg=cfg)
        
        
        temp_audio_path = os.path.join(tempfile.gettempdir(), f"mars5_audio_{len(audio_segments)}.wav")
        torchaudio.save(temp_audio_path, wav_out.unsqueeze(0), mars5.sr)
        audio_segments.append(AudioSegment.from_wav(temp_audio_path))

    combined_audio = sum(audio_segments)
    combined_audio_path = os.path.join(tempfile.gettempdir(), "mars5_combined_audio.wav")
    combined_audio.export(combined_audio_path, format="wav")

    logging.debug(f"Audio saved to {combined_audio_path}")
    return combined_audio_path

pipe = StableDiffusionPipeline.from_pretrained("stabilityai/stable-diffusion-2", torch_dtype=torch.float16)
pipe.to(device)

def generate_image(prompt):
    with torch.cuda.amp.autocast():
        image = pipe(
            prompt,
            num_inference_steps=28,
            guidance_scale=3.0,
        ).images[0]
    return image

hardcoded_prompt_1 = "Give a high quality photograph of a great looking red 2026 Toyota coupe against a skyline setting in the night, michael mann style in omaha enticing the consumer to buy this product"
hardcoded_prompt_2 = "A vibrant and dynamic football game scene in the style of Peter Paul Rubens, showcasing the intense match between Alabama and Nebraska. The players are depicted with the dramatic, muscular physiques and expressive faces typical of Rubens' style. The Alabama team is wearing their iconic crimson and white uniforms, while the Nebraska team is in their classic red and white attire. The scene is filled with action, with players in mid-motion, tackling, running, and catching the ball. The background features a grand stadium filled with cheering fans, banners, and the natural landscape in the distance. The colors are rich and vibrant, with a strong use of light and shadow to create depth and drama. The overall atmosphere captures the intensity and excitement of the game, infused with the grandeur and dynamism characteristic of Rubens' work."
hardcoded_prompt_3 = "Create a high-energy scene of a DJ performing on a large stage with vibrant lights, colorful lasers, a lively dancing crowd, and various electronic equipment in the background."

def update_images():
    image_1 = generate_image(hardcoded_prompt_1)
    image_2 = generate_image(hardcoded_prompt_2)
    image_3 = generate_image(hardcoded_prompt_3)
    return image_1, image_2, image_3

with gr.Blocks(theme='Pijush2023/scikit-learn-pijush') as demo:
    with gr.Row():
        with gr.Column():
            state = gr.State()
            
            chatbot = gr.Chatbot([], elem_id="RADAR:Channel 94.1", bubble_full_width=False)
            choice = gr.Radio(label="Select Style", choices=["Details", "Conversational"], value="Conversational")
            
            gr.Markdown("<h1 style='color: red;'>Talk to RADAR</h1>", elem_id="voice-markdown")
            
            chat_input = gr.Textbox(show_copy_button=True, interactive=True, show_label=False, label="ASK Radar !!!")
            chat_msg = chat_input.submit(add_message, [chatbot, chat_input], [chatbot, chat_input],api_name="voice_query")
            tts_choice = gr.Radio(label="Select TTS System", choices=["Alpha", "Beta", "Gamma"], value="Alpha")
            retriver_button = gr.Button("Retriver")
            retriver_button.click(fn=add_message, inputs=[chatbot, chat_input], outputs=[chatbot, chat_input]).then(
            fn=bot, inputs=[chatbot, choice, tts_choice], outputs=[chatbot],api_name="Ask_Retriever")
            #gr.Audio(interactive=False, autoplay=True)]

            # voice_response_button = gr.Button("Voice Response")
            # voice_response_button.click(fn=generate_voice_response, inputs=[chatbot, tts_choice], outputs=[gr.Audio(interactive=False, autoplay=True)],api_name="generate_voice_response")

            bot_msg = chat_msg.then(bot, [chatbot, choice, tts_choice], [chatbot, gr.Audio(interactive=False, autoplay=True)],api_name="generate_voice_response")
            # bot_msg = chat_msg.then(bot, [chatbot, choice, tts_choice], [chatbot, gr.Audio(interactive=False, autoplay=True)])
            bot_msg.then(lambda: gr.Textbox(value="", interactive=True, placeholder="Ask Radar!!!...", show_label=False), None, [chat_input])
            chatbot.like(print_like_dislike, None, None)
            clear_button = gr.Button("Clear")
            clear_button.click(fn=clear_textbox, inputs=None, outputs=chat_input)
           
            audio_input = gr.Audio(sources=["microphone"], streaming=True, type='numpy')
            audio_input.stream(transcribe_function, inputs=[state, audio_input], outputs=[state, chat_input], api_name="voice_query_to_text")


            gr.Markdown("<h1 style='color: red;'>Map</h1>", elem_id="location-markdown")
            location_output = gr.HTML()
            bot_msg.then(show_map_if_details, [chatbot, choice], [location_output, location_output],api_name="map_finder")
        
        
        # with gr.Column():
        #     weather_output = gr.HTML(value=fetch_local_weather())
        #     news_output = gr.HTML(value=fetch_local_news())
        #     news_output = gr.HTML(value=fetch_local_events())
        
        with gr.Column():
            image_output_1 = gr.Image(value=generate_image(hardcoded_prompt_1), width=400, height=400)
            image_output_2 = gr.Image(value=generate_image(hardcoded_prompt_2), width=400, height=400)
            image_output_3 = gr.Image(value=generate_image(hardcoded_prompt_3), width=400, height=400)

            refresh_button = gr.Button("Refresh Images")
            refresh_button.click(fn=update_images, inputs=None, outputs=[image_output_1, image_output_2, image_output_3])

demo.queue()
demo.launch(share=True)