Update app.py

app.py

@@ -14,7 +14,6 @@ from transformers import pipeline, AutoModelForSpeechSeq2Seq, AutoProcessor
 from googlemaps import Client as GoogleMapsClient
 from gtts import gTTS
 from diffusers import StableDiffusion3Pipeline
-
 from langchain_openai import OpenAIEmbeddings, ChatOpenAI
 from langchain_pinecone import PineconeVectorStore
 from langchain.prompts import PromptTemplate
@@ -22,27 +21,21 @@ 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
 
 # Check if the token is already set in the environment variables
 hf_token = os.getenv("HF_TOKEN")
-
 if hf_token is None:
-    # If the token is not set, prompt for it (this should be done securely)
     print("Please set your Hugging Face token in the environment variables.")
 else:
-    # Login using the token
     login(token=hf_token)
 
-# Your application logic goes here
-print("Logged in successfully to Hugging Face Hub!")
-
-# Set up logging
 logging.basicConfig(level=logging.DEBUG)
 
-# Initialize OpenAI embeddings
 embeddings = OpenAIEmbeddings(api_key=os.environ['OPENAI_API_KEY'])
 
-# Initialize Pinecone
 from pinecone import Pinecone
 pc = Pinecone(api_key=os.environ['PINECONE_API_KEY'])
 
@@ -50,9 +43,7 @@ index_name = "birmingham-dataset"
 vectorstore = PineconeVectorStore(index_name=index_name, embedding=embeddings)
 retriever = vectorstore.as_retriever(search_kwargs={'k': 5})
 
-# Initialize ChatOpenAI model
-chat_model = ChatOpenAI(api_key=os.environ['OPENAI_API_KEY'],
-                        temperature=0, model='gpt-4o')
+chat_model = ChatOpenAI(api_key=os.environ['OPENAI_API_KEY'], temperature=0, model='gpt-4o')
 
 conversational_memory = ConversationBufferWindowMemory(
     memory_key='chat_history',
@@ -64,13 +55,11 @@ def get_current_time_and_date():
     now = datetime.now()
     return now.strftime("%Y-%m-%d %H:%M:%S")
 
-# Example usage
 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", [])
@@ -193,11 +182,6 @@ def get_weather_icon(condition):
     }
     return condition_map.get(condition, "c04d")
 
-# Update prompt templates to include fetched details
-
-current_time_and_date = get_current_time_and_date()
-
-# Define prompt templates
 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
@@ -216,7 +200,6 @@ Helpful Answer:"""
 QA_CHAIN_PROMPT_1 = PromptTemplate(input_variables=["context", "question"], template=template1)
 QA_CHAIN_PROMPT_2 = PromptTemplate(input_variables=["context", "question"], template=template2)
 
-# Define the retrieval QA chain
 def build_qa_chain(prompt_template):
     qa_chain = RetrievalQA.from_chain_type(
         llm=chat_model,
@@ -233,7 +216,6 @@ def build_qa_chain(prompt_template):
     ]
     return qa_chain, tools
 
-# Define the agent initializer
 def initialize_agent_with_prompt(prompt_template):
     qa_chain, tools = build_qa_chain(prompt_template)
     agent = initialize_agent(
@@ -247,7 +229,6 @@ def initialize_agent_with_prompt(prompt_template):
     )
     return agent
 
-# Define the function to generate answers
 def generate_answer(message, choice):
     logging.debug(f"generate_answer called with prompt_choice: {choice}")
     
@@ -260,23 +241,24 @@ def generate_answer(message, choice):
         agent = initialize_agent_with_prompt(QA_CHAIN_PROMPT_2)
     response = agent(message)
 
-    # Extract addresses for mapping regardless of the choice
     addresses = extract_addresses(response['output'])
     return response['output'], addresses
 
-def bot(history, choice):
+def bot(history, choice, tts_choice):
     if not history:
         return history
     response, addresses = generate_answer(history[-1][0], choice)
     history[-1][1] = ""
     
-    # Generate audio for the entire response in a separate thread
     with concurrent.futures.ThreadPoolExecutor() as executor:
-        audio_future = executor.submit(generate_audio_elevenlabs, response)
+        if tts_choice == "Eleven Labs":
+            audio_future = executor.submit(generate_audio_elevenlabs, response)
+        elif tts_choice == "Parler-TTS":
+            audio_future = executor.submit(generate_audio_parler_tts, response)
         
         for character in response:
             history[-1][1] += character
-            time.sleep(0.05)  # Adjust the speed of text appearance
+            time.sleep(0.05)
             yield history, None
         
         audio_path = audio_future.result()
@@ -293,20 +275,13 @@ def extract_addresses(response):
     if not isinstance(response, str):
         response = str(response)
     address_patterns = [
-        # r'([A-Z].*,\sOmaha,\sNE\s\d{5})',
-        # r'(\d{4}\s.*,\sOmaha,\sNE\s\d{5})',
-        # r'([A-Z].*,\sNE\s\d{5})',
-        # r'([A-Z].*,.*\sSt,\sOmaha,\sNE\s\d{5})',
-        # r'([A-Z].*,.*\sStreets,\sOmaha,\sNE\s\d{5})',
-        # r'(\d{2}.*\sStreets)',
-        # r'([A-Z].*\s\d{2},\sOmaha,\sNE\s\d{5})'
         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-Z].*\s\d{2},\sBirmingham,\sAL\s\d{5})',
         r'([a-zA-Z]\s Birmingham)'
     ]
     addresses = []
@@ -411,7 +386,6 @@ def fetch_local_news():
     else:
         return "<p>Failed to fetch local news</p>"
 
-# Voice Control
 import numpy as np
 import torch
 from transformers import pipeline, AutoModelForSpeechSeq2Seq, AutoProcessor
@@ -419,18 +393,13 @@ 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,
-                                                  #low_cpu_mem_usage=True,
-                                                  use_safetensors=True).to(device)
+model = AutoModelForSpeechSeq2Seq.from_pretrained(model_id, torch_dtype=torch_dtype).to(device)
 processor = AutoProcessor.from_pretrained(model_id)
 
-# Optimized ASR pipeline
 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"
 
-import numpy as np
-
 def transcribe_function(stream, new_chunk):
     try:
         sr, y = new_chunk[0], new_chunk[1]
@@ -469,7 +438,7 @@ def show_map_if_details(history,choice):
 
 def generate_audio_elevenlabs(text):
     XI_API_KEY = os.environ['ELEVENLABS_API']
-    VOICE_ID = 'd9MIrwLnvDeH7aZb61E9'  # Replace with your voice ID
+    VOICE_ID = 'd9MIrwLnvDeH7aZb61E9'
     tts_url = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID}/stream"
     headers = {
         "Accept": "application/json",
@@ -481,7 +450,7 @@ def generate_audio_elevenlabs(text):
         "voice_settings": {
             "stability": 1.0,
             "similarity_boost": 0.0,
-            "style": 0.60,  # Adjust style for more romantic tone
+            "style": 0.60,
             "use_speaker_boost": False
         }
     }
@@ -497,7 +466,49 @@ def generate_audio_elevenlabs(text):
         logging.error(f"Error generating audio: {response.text}")
         return None
 
-# Stable Diffusion setup
+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 generate_audio_parler_tts(text):
+    description = "Thomas speaks with emphasis and excitement at a moderate pace with high quality."
+    inputs = parler_tokenizer(description, return_tensors="pt").to(device)
+    prompt = parler_tokenizer(preprocess(text), 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()
+
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as f:
+        f.write(audio_arr.tobytes())
+        temp_audio_path = f.name
+
+    logging.debug(f"Audio saved to {temp_audio_path}")
+    return temp_audio_path
+
 pipe = StableDiffusion3Pipeline.from_pretrained("stabilityai/stable-diffusion-3-medium-diffusers", torch_dtype=torch.float16)
 pipe = pipe.to("cuda")
 
@@ -510,9 +521,8 @@ def generate_image(prompt):
     ).images[0]
     return image
 
-# Hardcoded prompt for image generation
-hardcoded_prompt_1="Give a high quality photograph of a great looking red 2026 Bentley coupe against a skyline setting in th 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_1 = "Give a high quality photograph of a great looking red 2026 Bentley 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():
@@ -521,10 +531,7 @@ def update_images():
     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()
@@ -535,13 +542,13 @@ with gr.Blocks(theme='Pijush2023/scikit-learn-pijush') as demo:
             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])
-            bot_msg = chat_msg.then(bot, [chatbot, choice], [chatbot, gr.Audio(interactive=False, autoplay=True)])
+            tts_choice = gr.Radio(label="Select TTS System", choices=["Eleven Labs", "Parler-TTS"], value="Eleven Labs")
+            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="SAMLOne_real_time")
 
@@ -552,18 +559,15 @@ with gr.Blocks(theme='Pijush2023/scikit-learn-pijush') as demo:
         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())
+            events_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)
-