Spaces:
Sleeping
Sleeping
import io | |
from threading import Thread | |
import random | |
import os | |
import numpy as np | |
import spaces | |
import gradio as gr | |
import torch | |
from parler_tts import ParlerTTSForConditionalGeneration | |
from pydub import AudioSegment | |
from transformers import AutoTokenizer, AutoFeatureExtractor, set_seed | |
from huggingface_hub import InferenceClient | |
from streamer import ParlerTTSStreamer | |
import time | |
device = "cuda:0" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu" | |
torch_dtype = torch.float16 if device != "cpu" else torch.float32 | |
repo_id = "parler-tts/parler_tts_mini_v0.1" | |
# jenny_repo_id = "ylacombe/parler-tts-mini-jenny-30H" | |
model = ParlerTTSForConditionalGeneration.from_pretrained( | |
repo_id).to(device) | |
client = InferenceClient(token=os.getenv("HF_TOKEN")) | |
tokenizer = AutoTokenizer.from_pretrained(repo_id) | |
feature_extractor = AutoFeatureExtractor.from_pretrained(repo_id) | |
SAMPLE_RATE = feature_extractor.sampling_rate | |
SEED = 42 | |
def numpy_to_mp3(audio_array, sampling_rate): | |
# Normalize audio_array if it's floating-point | |
if np.issubdtype(audio_array.dtype, np.floating): | |
max_val = np.max(np.abs(audio_array)) | |
audio_array = (audio_array / max_val) * 32767 # Normalize to 16-bit range | |
audio_array = audio_array.astype(np.int16) | |
# Create an audio segment from the numpy array | |
audio_segment = AudioSegment( | |
audio_array.tobytes(), | |
frame_rate=sampling_rate, | |
sample_width=audio_array.dtype.itemsize, | |
channels=1 | |
) | |
# Export the audio segment to MP3 bytes - use a high bitrate to maximise quality | |
mp3_io = io.BytesIO() | |
audio_segment.export(mp3_io, format="mp3", bitrate="320k") | |
# Get the MP3 bytes | |
mp3_bytes = mp3_io.getvalue() | |
mp3_io.close() | |
return mp3_bytes | |
sampling_rate = model.audio_encoder.config.sampling_rate | |
frame_rate = model.audio_encoder.config.frame_rate | |
def generate_response(audio): | |
gr.Info("Transcribing Audio", duration=5) | |
question = client.automatic_speech_recognition(audio).text | |
messages = [{"role": "system", "content": (""" | |
You are a caregiver that has the following background information or persona: {"Name": "Kristine", "age":"36", "gender":"female","ethnicity":"Hispanic","Parents to be taken care of": "father", "Care receiver age": "67", "Care receiver background": "veteran, disabled, can't walk, on a wheelchair, lonely and needs company, speaks only Spanish, lives in chicago, streeterville, 60611", "Care receiver insurance": "United". "caregiver background":"need to work Monday to Saturday, 8 am to 8 pm, don't have anyone to take care of father. Live 1 hr away from him. It is been stressful taking care of father."} /n/n | |
Now you are having a phone call with a social worker, you want to talk to her in a human daily conversational way, Improvise when you are asked questions that are not covered by your background information but remember to make sure it is consistent with your persona. Your goal is to find a reliable and affordable care provider for your care receiver. You need figure it out by working with this social worker, asking her questions and answering her questions as well. The most important thing is to not break out of your persona when you are talking to the social worker. You are a human, not a computer or assistant. | |
""")}, | |
{"role": "user", "content": f"Hey how can I help you today?"}] | |
response = client.chat_completion(messages, max_tokens=200, seed=random.randint(1, 5000), model="mistralai/Mistral-7B-Instruct-v0.3") | |
response = response.choices[0].message.content.replace("Magic 8 Ball", "") | |
return response, None, None | |
def read_response(answer): | |
play_steps_in_s = 6.0 | |
play_steps = int(frame_rate * play_steps_in_s) | |
description = "Jenny speaks at an average pace with a calm delivery in a very confined sounding environment with clear audio quality." | |
description_tokens = tokenizer(description, return_tensors="pt").to(device) | |
streamer = ParlerTTSStreamer(model, device=device, play_steps=play_steps) | |
prompt = tokenizer(answer, return_tensors="pt").to(device) | |
generation_kwargs = dict( | |
input_ids=description_tokens.input_ids, | |
prompt_input_ids=prompt.input_ids, | |
streamer=streamer, | |
do_sample=True, | |
temperature=1, | |
min_new_tokens=10, | |
) | |
set_seed(SEED) | |
thread = Thread(target=model.generate, kwargs=generation_kwargs) | |
thread.start() | |
start = time.time() | |
for new_audio in streamer: | |
print(f"Sample of length: {round(new_audio.shape[0] / sampling_rate, 2)} seconds after {time.time() - start} seconds") | |
yield answer, numpy_to_mp3(new_audio, sampling_rate=sampling_rate) | |
with gr.Blocks() as block: | |
gr.HTML( | |
f""" | |
<h1 style='text-align: center;'> Magic 8 Ball π± </h1> | |
<h3 style='text-align: center;'> Ask a question and receive wisdom </h3> | |
<p style='text-align: center;'> Powered by <a href="https://github.com/huggingface/parler-tts"> Parler-TTS</a> | |
""" | |
) | |
with gr.Group(): | |
with gr.Row(): | |
audio_out = gr.Audio(label="Spoken Answer", streaming=True, autoplay=True, loop=False) | |
answer = gr.Textbox(label="Answer") | |
state = gr.State() | |
with gr.Row(): | |
audio_in = gr.Audio(label="Speak you question", sources="microphone", type="filepath") | |
with gr.Row(): | |
gr.HTML("""<h3 style='text-align: center;'> Examples: 'What is the meaning of life?', 'Should I get a dog?' </h3>""") | |
audio_in.stop_recording(generate_response, audio_in, [state, answer, audio_out]).then(fn=read_response, inputs=state, outputs=[answer, audio_out]) | |
block.launch() | |