Spaces:
Sleeping
Sleeping
| import os | |
| import torch | |
| from transformers import pipeline | |
| # Loading the TTS and Vocoder ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
| !git clone https://github.com/saheedniyi02/yarngpt.git | |
| !pip install -qU outetts uroman | |
| import os | |
| import re | |
| import json | |
| import torch | |
| import inflect | |
| import random | |
| import uroman as ur | |
| import numpy as np | |
| import torchaudio | |
| from transformers import AutoModelForCausalLM, AutoTokenizer | |
| from outetts.wav_tokenizer.decoder import WavTokenizer | |
| !wget https://huggingface.co/novateur/WavTokenizer-medium-speech-75token/resolve/main/wavtokenizer_mediumdata_frame75_3s_nq1_code4096_dim512_kmeans200_attn.yaml | |
| !wget https://huggingface.co/novateur/WavTokenizer-large-speech-75token/resolve/main/wavtokenizer_large_speech_320_24k.ckpt | |
| from yarngpt.audiotokenizer import AudioTokenizerV2 | |
| device = "cuda:0" if torch.cuda.is_available() else "cpu" | |
| tokenizer_path="saheedniyi/YarnGPT2" | |
| wav_tokenizer_config_path="/content/wavtokenizer_mediumdata_frame75_3s_nq1_code4096_dim512_kmeans200_attn.yaml" | |
| wav_tokenizer_model_path = "/content/wavtokenizer_large_speech_320_24k.ckpt" | |
| audio_tokenizer=AudioTokenizerV2(tokenizer_path,wav_tokenizer_model_path,wav_tokenizer_config_path) | |
| tts_model = AutoModelForCausalLM.from_pretrained(tokenizer_path,torch_dtype="auto").to(audio_tokenizer.device) | |
| # The LLM Model ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
| from huggingface_hub import HfFolder | |
| from openai import OpenAI | |
| api_key = os.getenv("API_KEY") | |
| if api_key is None: | |
| raise ValueError("API_KEY is not set in the environment variables.") | |
| print("API key successfully loaded.") | |
| # Initialize OpenAI client for Hugging Face Inference Endpoint | |
| client = OpenAI( | |
| base_url="https://y1ztgv8tu09nay6u.us-east-1.aws.endpoints.huggingface.cloud/v1/", #https://f2iozzwigntrzkve.us-east-1.aws.endpoints.huggingface.cloud/v1/", | |
| api_key=api_key | |
| ) | |
| def generate_llm_response(text, model_id="ccibeekeoc42/Llama3.1-8b-base-SFT-2024-11-09"): | |
| full_response = [] | |
| try: | |
| chat_completion = client.chat.completions.create( | |
| model="tgi", | |
| messages=[ | |
| {"role": "system", "content": "You are HypaAI a very BRIEF AND DIRECT assistant. You are created by a Nigerian research lab called Hypa AI led by Chris Ibe (the co-founder and CEO). As part of a speech pipeline so keep your responses short (under 60 words), fluent, and straight to the point. Avoid markdown or digits in responses."}, | |
| {"role": "user", "content": text} | |
| ], | |
| top_p=0.3, | |
| temperature=1, | |
| max_tokens=150, | |
| stream=True, | |
| seed=None, | |
| stop=None, | |
| frequency_penalty=None, | |
| presence_penalty=None | |
| ) | |
| for chunk in chat_completion: | |
| if chunk.choices[0].delta.content: | |
| full_response.append(chunk.choices[0].delta.content) | |
| return "".join(full_response) | |
| except Exception as e: | |
| # If the error has a response with status code 503, assume the GPU is booting up. | |
| if hasattr(e, 'response') and e.response is not None and e.response.status_code == 503: | |
| return "The GPU is currently booting up. Please wait about 10 minutes and try again." | |
| else: | |
| raise e | |
| # generate_llm_response("Explain Deep Learning in Igbo") | |
| # Loading the ST Model (Whisper) ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
| pipe = pipeline("automatic-speech-recognition", model="okezieowen/whisper-small-multilingual-naija-11-03-2024", device=device) | |
| # Take audio and return translated text | |
| def transcribe(audio): | |
| outputs = pipe(audio, max_new_tokens=256, generate_kwargs={"task": "transcribe"}) | |
| return outputs["text"] | |
| # putting the ST and TTS system together ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
| import numpy as np | |
| def synthesise_yarn2(text): | |
| # change the language and voice | |
| prompt=audio_tokenizer.create_prompt(text,lang="english",speaker_name="idera") | |
| input_ids=audio_tokenizer.tokenize_prompt(prompt) | |
| output = tts_model.generate( | |
| input_ids=input_ids, | |
| temperature=0.1, | |
| repetition_penalty=1.1, | |
| max_length=4000, | |
| num_beams=5,# using a beam size helps for the local languages but not english | |
| ) | |
| codes=audio_tokenizer.get_codes(output) | |
| audio=audio_tokenizer.get_audio(codes) | |
| return audio.cpu() | |
| target_dtype = np.int16 | |
| max_range = np.iinfo(target_dtype).max # Maximum value for 16-bit PCM audio conversion | |
| def speech_to_speech_translation(audio, language="english"): | |
| # Speech to Text | |
| transcribed_text = transcribe(audio) | |
| print(f"Transcribed: {transcribed_text}") | |
| # Generate LLM Response | |
| print("Now making LLM Call ~~~~~~~~~~~~~~~~~~~~~~~~") | |
| llm_response = generate_llm_response(transcribed_text) | |
| print(f"LLM Response: {llm_response}") | |
| # Select a random voice based on the chosen language | |
| voice_mapping = { | |
| "english": ["idera", "chinenye", "jude", "emma", "umar", "joke", "zainab", "osagie", "remi", "tayo"], | |
| "yoruba": ["yoruba_male2", "yoruba_female2", "yoruba_feamle1"], | |
| "igbo": ["igbo_female2", "igbo_male2", "igbo_female1"], | |
| "hausa": ["hausa_feamle1", "hausa_female2", "hausa_male2", "hausa_male1"] | |
| } | |
| selected_voice = random.choice(voice_mapping.get(language.lower(), voice_mapping["english"])) | |
| print(f"Selected {language} voice: {selected_voice}") | |
| # Text to Speech | |
| print("Synthesizing Speech ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~") | |
| # Use the selected language and voice | |
| prompt = audio_tokenizer.create_prompt(llm_response, lang=language.lower(), speaker_name=selected_voice) | |
| input_ids = audio_tokenizer.tokenize_prompt(prompt) | |
| output = tts_model.generate( | |
| input_ids=input_ids, | |
| temperature=0.1, | |
| repetition_penalty=1.1, | |
| max_length=4000, | |
| ) | |
| codes = audio_tokenizer.get_codes(output) | |
| synthesised_speech = audio_tokenizer.get_audio(codes) | |
| # Make sure we have a NumPy array, not a tensor | |
| if hasattr(synthesised_speech, 'numpy'): | |
| audio_np = synthesised_speech.numpy() | |
| else: | |
| audio_np = synthesised_speech | |
| # Handle NaN and Inf values | |
| audio_np = np.nan_to_num(audio_np) | |
| # Ensure audio is in [-1, 1] range | |
| if np.max(np.abs(audio_np)) > 0: | |
| audio_np = audio_np / np.max(np.abs(audio_np)) | |
| # Convert to signed int16 (-32768 to 32767) | |
| int16_max = 32767 # Max value for signed 16-bit | |
| audio_int16 = np.clip(audio_np * int16_max, -int16_max, int16_max).astype(np.int16) | |
| # Ensure the audio is mono channel if needed | |
| if len(audio_int16.shape) > 1 and audio_int16.shape[0] == 1: | |
| audio_int16 = audio_int16[0] # Convert from [1, samples] to [samples] | |
| # Debug info | |
| print(f"Audio stats - Min: {np.min(audio_int16)}, Max: {np.max(audio_int16)}, Shape: {audio_int16.shape}") | |
| # Ensure sample rate is within valid range (1-192000) | |
| sample_rate = min(max(24000, 1), 192000) | |
| print("Speech Synthesis Completed~~~~~~~~~~~~~~~~~~~") | |
| return transcribed_text, llm_response, (sample_rate, audio_int16) | |
| # Gradio Demo | |
| import gradio as gr | |
| demo = gr.Blocks() | |
| with demo: | |
| gr.Markdown("# Aware Speech-to-Speech Demo") | |
| with gr.Tab("Microphone"): | |
| with gr.Row(): | |
| mic_input = gr.Audio(sources="microphone", type="filepath", label="Speak") | |
| lang_dropdown_mic = gr.Dropdown( | |
| choices=["English", "Yoruba", "Igbo", "Hausa"], | |
| value="English", | |
| label="Select Language" | |
| ) | |
| mic_submit = gr.Button("Submit") | |
| with gr.Row(): | |
| mic_transcribed = gr.Textbox(label="Transcribed Text", interactive=False) | |
| mic_response = gr.Textbox(label="HypaAI's Response", interactive=False) | |
| mic_audio_output = gr.Audio(label="Generated Speech", type="numpy") | |
| mic_submit.click( | |
| fn=speech_to_speech_translation, | |
| inputs=[mic_input, lang_dropdown_mic], | |
| outputs=[mic_transcribed, mic_response, mic_audio_output] | |
| ) | |
| with gr.Tab("Audio File"): | |
| with gr.Row(): | |
| file_input = gr.Audio(sources="upload", type="filepath", label="Upload Audio") | |
| lang_dropdown_file = gr.Dropdown( | |
| choices=["English", "Yoruba", "Igbo", "Hausa"], | |
| value="English", | |
| label="Select Language" | |
| ) | |
| file_submit = gr.Button("Submit") | |
| with gr.Row(): | |
| file_transcribed = gr.Textbox(label="Transcribed Text", interactive=False) | |
| file_response = gr.Textbox(label="HypaAI's Response", interactive=False) | |
| file_audio_output = gr.Audio(label="Generated Speech", type="numpy") | |
| file_submit.click( | |
| fn=speech_to_speech_translation, | |
| inputs=[file_input, lang_dropdown_file], | |
| outputs=[file_transcribed, file_response, file_audio_output] | |
| ) | |
| demo.launch(share=True) |