Local Weather
-
-
-
- }.png)
-
-
-
-
- Temperature: {temp_fahrenheit}°F
-Condition: {condition}
-Humidity: {humidity}%
-| Title | -Date and Time | -Location | -
|---|---|---|
| {title} | -{date} | -{location} | -
Failed to fetch local events
" +# 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 = """ +#| Title | +#Date and Time | +#Location | +#
|---|---|---|
| {title} | +#{date} | +#{location} | +#
Failed to fetch local events
" -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() +# 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") +# 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" +# 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") +# condition = current_conditions.get("conditions", "N/A") +# humidity = current_conditions.get("humidity", "N/A") - weather_html = f""" -
- Temperature: {temp_fahrenheit}°F
-Condition: {condition}
-Humidity: {humidity}%
-Failed to fetch local weather: {e}
" +# weather_html = f""" +#
+# Temperature: {temp_fahrenheit}°F
+#Condition: {condition}
+#Humidity: {humidity}%
+#Failed to fetch local weather: {e}
" -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") +# 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. Always say "It was my pleasure!" at the end of the answer. -{context} -Question: {question} -Helpful Answer:""" +# 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. 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 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:""" -QA_CHAIN_PROMPT_1 = PromptTemplate(input_variables=["context", "question"], template=template1) -QA_CHAIN_PROMPT_2 = PromptTemplate(input_variables=["context", "question"], template=template2) +# 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 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 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}") +# 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) +# 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 +# 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] = "" +# 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 == "Eleven Labs": - audio_future = executor.submit(generate_audio_elevenlabs, response) - elif tts_choice == "Parler-TTS": - audio_future = executor.submit(generate_audio_parler_tts, response) - elif tts_choice == "MARS5": - audio_future = executor.submit(generate_audio_mars5, response) - elif tts_choice == "FishAudio": - audio_future = executor.submit(generate_audio_fishaudio, response) +# with concurrent.futures.ThreadPoolExecutor() as executor: +# 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) +# elif tts_choice == "MARS5": +# audio_future = executor.submit(generate_audio_mars5, response) +# elif tts_choice == "FishAudio": +# audio_future = executor.submit(generate_audio_fishaudio, response) - for character in response: - history[-1][1] += character - time.sleep(0.05) - yield history, None +# for character in response: +# history[-1][1] += character +# time.sleep(0.05) +# yield history, None - audio_path = audio_future.result() - yield history, audio_path +# 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) +# 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) -def print_like_dislike(x: gr.LikeData): - print(x.index, x.value, x.liked) +# 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)' - ] - addresses = [] - for pattern in address_patterns: - addresses.extend(re.findall(pattern, response)) - return addresses +# 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)' +# ] +# addresses = [] +# for pattern in address_patterns: +# addresses.extend(re.findall(pattern, response)) +# return addresses -all_addresses = [] +# all_addresses = [] -def generate_map(location_names): - global all_addresses - all_addresses.extend(location_names) +# 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) +# api_key = os.environ['GOOGLEMAPS_API_KEY'] +# gmaps = GoogleMapsClient(key=api_key) - m = folium.Map(location=[33.5175,-86.809444], zoom_start=16) +# 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) +# 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 +# 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 = """ -{snippet}
-Failed to fetch local news
" +# 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 = """ +#{snippet}
+#Failed to fetch local news
" -import numpy as np -import torch -from transformers import pipeline, AutoModelForSpeechSeq2Seq, AutoProcessor +# 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) +# 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) +# 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" +# 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 +# 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)) +# y = y.astype(np.float32) / np.max(np.abs(y)) - if stream is not None: - stream = np.concatenate([stream, y]) - else: - stream = 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) +# result = pipe_asr({"array": stream, "sampling_rate": sr}, return_timestamps=False) - full_text = result.get("text", "") +# 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 "" +# return stream, full_text, result -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 update_map_with_response(history): +# if not history: +# return "" +# response = history[-1][1] +# addresses = extract_addresses(response) +# return generate_map(addresses) -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 +# def clear_textbox(): +# return "" -repo_id = "parler-tts/parler-tts-mini-expresso" +# 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), "" -parler_model = ParlerTTSForConditionalGeneration.from_pretrained(repo_id).to(device) -parler_tokenizer = AutoTokenizer.from_pretrained(repo_id) -parler_feature_extractor = AutoFeatureExtractor.from_pretrained(repo_id) +# 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 -SAMPLE_RATE = parler_feature_extractor.sampling_rate -SEED = 42 +# repo_id = "parler-tts/parler-tts-mini-expresso" -def preprocess(text): - number_normalizer = EnglishNumberNormalizer() - text = number_normalizer(text).strip() - if text[-1] not in punctuation: - text = f"{text}." +# parler_model = ParlerTTSForConditionalGeneration.from_pretrained(repo_id).to(device) +# parler_tokenizer = AutoTokenizer.from_pretrained(repo_id) +# parler_feature_extractor = AutoFeatureExtractor.from_pretrained(repo_id) - abbreviations_pattern = r'\b[A-Z][A-Z\.]+\b' +# SAMPLE_RATE = parler_feature_extractor.sampling_rate +# SEED = 42 - def separate_abb(chunk): - chunk = chunk.replace(".", "") - return " ".join(chunk) +# def preprocess(text): +# number_normalizer = EnglishNumberNormalizer() +# text = number_normalizer(text).strip() +# if text[-1] not in punctuation: +# text = f"{text}." - abbreviations = re.findall(abbreviations_pattern, text) - for abv in abbreviations: - if abv in text: - text = text.replace(abv, separate_abb(abv)) - return text +# abbreviations_pattern = r'\b[A-Z][A-Z\.]+\b' -def chunk_text(text, max_length=250): - words = text.split() - chunks = [] - current_chunk = [] - current_length = 0 +# def separate_abb(chunk): +# chunk = chunk.replace(".", "") +# return " ".join(chunk) - 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 +# abbreviations = re.findall(abbreviations_pattern, text) +# for abv in abbreviations: +# if abv in text: +# text = text.replace(abv, separate_abb(abv)) +# return text - if current_chunk: - chunks.append(' '.join(current_chunk)) +# def chunk_text(text, max_length=250): +# words = text.split() +# chunks = [] +# current_chunk = [] +# current_length = 0 - return chunks +# 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 -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 = [] +# if current_chunk: +# chunks.append(' '.join(current_chunk)) - for chunk in chunks: - inputs = parler_tokenizer(description, return_tensors="pt").to(device) - prompt = parler_tokenizer(chunk, return_tensors="pt").to(device) +# return chunks - set_seed(SEED) - generation = parler_model.generate(input_ids=inputs.input_ids, prompt_input_ids=prompt.input_ids) - audio_arr = generation.cpu().numpy().squeeze() +# 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 = [] - 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)) +# for chunk in chunks: +# inputs = parler_tokenizer(description, return_tensors="pt").to(device) +# prompt = parler_tokenizer(chunk, return_tensors="pt").to(device) - 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") +# set_seed(SEED) +# generation = parler_model.generate(input_ids=inputs.input_ids, prompt_input_ids=prompt.input_ids) +# audio_arr = generation.cpu().numpy().squeeze() - logging.debug(f"Audio saved to {combined_audio_path}") - return combined_audio_path +# 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)) -# Load the MARS5 model -mars5, config_class = torch.hub.load('Camb-ai/mars5-tts', 'mars5_english', trust_repo=True) +# 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") -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 - } +# logging.debug(f"Audio saved to {combined_audio_path}") +# return combined_audio_path - chunks = chunk_text(preprocess(text)) - audio_segments = [] +# # Load the MARS5 model +# mars5, config_class = torch.hub.load('Camb-ai/mars5-tts', 'mars5_english', trust_repo=True) - 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) +# 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)) +# 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") +# 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 +# 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) +# 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 +# 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 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_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." +# hardcoded_prompt_1="Give a high quality photograph of a great looking red 2026 toyota 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_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 +# 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 -# New TTS function for FishAudio -def generate_audio_fishaudio(text): - pipe = pipeline("text-to-speech", model="fishaudio/fish-speech-1.2") - audio = pipe(text) +# # New TTS function for FishAudio +# def generate_audio_fishaudio(text): +# pipe = pipeline("text-to-speech", model="fishaudio/fish-speech-1.2") +# audio = pipe(text) - with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as f: - audio.save(f.name) - temp_audio_path = f.name +# with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as f: +# audio.save(f.name) +# temp_audio_path = f.name - logging.debug(f"Audio saved to {temp_audio_path}") - return temp_audio_path +# logging.debug(f"Audio saved to {temp_audio_path}") +# return temp_audio_path -with gr.Blocks(theme='Pijush2023/scikit-learn-pijush') as demo: - with gr.Row(): - with gr.Column(): - state = gr.State() +# 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") +# 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("Date: {date}
Location: {location}
Failed to fetch local events
" +current_time_and_date = get_current_time_and_date() # def fetch_local_events(): # api_key = os.environ['SERP_API'] @@ -817,599 +777,639 @@ demo.launch(share=True) # events_html = """ #| Title | -#Date and Time | -#Location | -#
|---|---|---|
| {title} | -#{date} | -#{location} | -#
Failed to fetch local events
" +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 = """ +| Title | +Date and Time | +Location | +
|---|---|---|
| {title} | +{date} | +{location} | +
Failed to fetch local events
" -# 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() + +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") + 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" + 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") + condition = current_conditions.get("conditions", "N/A") + humidity = current_conditions.get("humidity", "N/A") -# weather_html = f""" -#
-# Temperature: {temp_fahrenheit}°F
-#Condition: {condition}
-#Humidity: {humidity}%
-#Failed to fetch local weather: {e}
" + weather_html = f""" +
+ Temperature: {temp_fahrenheit}°F
+Condition: {condition}
+Humidity: {humidity}%
+Failed to fetch local weather: {e}
" -# 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") +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. Always say "It was my pleasure!" at the end of the answer. -# {context} -# Question: {question} -# Helpful Answer:""" +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. 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 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:""" -# QA_CHAIN_PROMPT_1 = PromptTemplate(input_variables=["context", "question"], template=template1) -# QA_CHAIN_PROMPT_2 = PromptTemplate(input_variables=["context", "question"], template=template2) +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 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 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}") +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) + 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 + 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] = "" +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 == "Eleven Labs": -# audio_future = executor.submit(generate_audio_elevenlabs, response) -# elif tts_choice == "Parler-TTS": -# audio_future = executor.submit(generate_audio_parler_tts, response) -# elif tts_choice == "MARS5": -# audio_future = executor.submit(generate_audio_mars5, response) - + with concurrent.futures.ThreadPoolExecutor() as executor: + 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) + elif tts_choice == "MARS5": + audio_future = executor.submit(generate_audio_mars5, 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) - -# 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)' -# ] -# 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 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 = """ -#{snippet}
-#Failed to fetch local news
" + for character in response: + history[-1][1] += character + time.sleep(0.05) + yield history, None + + audio_path = audio_future.result() + yield history, audio_path -# import numpy as np -# import torch -# from transformers import pipeline, AutoModelForSpeechSeq2Seq, AutoProcessor +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) -# 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) +def print_like_dislike(x: gr.LikeData): + print(x.index, x.value, x.liked) -# 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) +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)' + ] + addresses = [] + for pattern in address_patterns: + addresses.extend(re.findall(pattern, response)) + return addresses -# base_audio_drive = "/data/audio" +all_addresses = [] -# 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 +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 -# y = y.astype(np.float32) / np.max(np.abs(y)) +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 = """ +{snippet}
+Failed to fetch local news
" -# if stream is not None: -# stream = np.concatenate([stream, y]) -# else: -# stream = y +import numpy as np +import torch +from transformers import pipeline, AutoModelForSpeechSeq2Seq, AutoProcessor -# result = pipe_asr({"array": stream, "sampling_rate": sr}, return_timestamps=False) +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) -# full_text = result.get("text", "") +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 + 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 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 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 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 +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" +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) +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 +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}." +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' + abbreviations_pattern = r'\b[A-Z][A-Z\.]+\b' -# def separate_abb(chunk): -# chunk = chunk.replace(".", "") -# return " ".join(chunk) + 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 + 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 +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 + 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)) + if current_chunk: + chunks.append(' '.join(current_chunk)) -# return chunks + 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 = [] +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) + 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() + 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)) + 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") + 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 + 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) +# 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 -# } +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 = [] + 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) + 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)) + 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") + 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 + 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) +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 +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." +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 +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() +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") + 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("