Hugging Face's logo

Spaces:
michaljunczyk
/
pl-asr-bigos-workspace
Sleeping

App Files Community
pl-asr-bigos-workspace / app.py
mj-new
Updated prompts handling
10babf5
raw history blame contribute delete
No virus
18 kB
import gradio as gr
import whisper
import numpy as np
import openai
import os
from gtts import gTTS
import json
import hashlib
import random
import string
import uuid
from datetime import date,datetime
from huggingface_hub import Repository, upload_file
import shutil
from helpers import dict_origin, dict_promptset
HF_TOKEN_WRITE = os.environ.get("HF_TOKEN_WRITE")
print("HF_TOKEN_WRITE", HF_TOKEN_WRITE)
today = date.today()
today_ymd = today.strftime("%Y%m%d")
def greet(name):
return "Hello " + name + "!!"
with open('app.css','r') as f:
css_file = f.read()
markdown="""
# Polish ASR BIGOS workspace
"""
# TODO move to config
WORKING_DATASET_REPO_URL = "https://huggingface.co/datasets/goodmike31/working-db"
REPO_NAME = "goodmike31/working-db"
REPOSITORY_DIR = "data"
LOCAL_DIR = "data_local"
os.makedirs(LOCAL_DIR,exist_ok=True)
def dump_json(thing,file):
with open(file,'w+',encoding="utf8") as f:
json.dump(thing,f)
def get_unique_name():
return ''.join([random.choice(string.ascii_letters
+ string.digits) for n in range(32)])
def get_prompts(project_name, size, language_code,prompts_left_info):
print(f"Retrieving prompts for project {project_name} with method: {type} for language_code {language_code} of size {size}")
size = int(size)
promptset = dict_promptset[project_name][0:size]
prompts_left_info = size
return(promptset, promptset[0],prompts_left_info)
def save_recording_and_meta(project_name, recording, prompt_text, language_code, spk_name, spk_age, spk_accent, spk_city, spk_gender, spk_nativity, promptset, prompt_number, prompts_left_info):
#, name, age, gender):
# TODO save user data in the next version
current_prompt = prompt_text.strip()
print("current_prompt: ", current_prompt)
# check if prompt number is set
if prompt_number == None:
prompt_number = 1
prompt_index = prompt_number - 1
print("prompt_number: ", prompt_number)
print("promptset: ", promptset)
if prompt_number == len(promptset):
next_prompt = "All prompts recorded. Thank you! You can close the app now:)"
else:
next_prompt = promptset[prompt_number]
print("next_prompt: ", next_prompt)
# remove leading and trailing spaces
next_prompt =next_prompt.strip()
# increment prompt number
prompt_number = prompt_number + 1
speaker_metadata={}
speaker_metadata['name'] = spk_name if spk_name != None else 'unknown'
speaker_metadata['gender'] = spk_gender if spk_gender != None else 'unknown'
speaker_metadata['age'] = spk_age if spk_age != None else 'unknown'
speaker_metadata['accent'] = spk_accent if spk_accent != None else 'unknown'
speaker_metadata['city'] = spk_city if spk_city != None else 'unknown'
speaker_metadata['nativity'] = spk_nativity if spk_nativity != None else 'unknown'
# TODO get ISO-693-1 codes
SAVE_ROOT_DIR = os.path.join(LOCAL_DIR, project_name, today_ymd, spk_name)
SAVE_DIR_AUDIO = os.path.join(SAVE_ROOT_DIR, "audio")
SAVE_DIR_META = os.path.join(SAVE_ROOT_DIR, "meta")
os.makedirs(SAVE_DIR_AUDIO, exist_ok=True)
os.makedirs(SAVE_DIR_META, exist_ok=True)
# Write audio to file
#audio_name = get_unique_name()
uuid_name = str(uuid.uuid4())
audio_fn = uuid_name + ".wav"
audio_output_fp = os.path.join(SAVE_DIR_AUDIO, audio_fn)
print (f"Saving {recording} as {audio_output_fp}")
shutil.copy2(recording, audio_output_fp)
# Write metadata.json to file
meta_fn = uuid_name + '.metadata.jsonl'
json_file_path = os.path.join(SAVE_DIR_META, meta_fn)
now = datetime.now()
timestamp_str = now.strftime("%d/%m/%Y %H:%M:%S")
metadata= {'id':uuid_name,
'audio_file': audio_fn,
'language_code':language_code,
'prompt_number':prompt_number,
'prompt':current_prompt,
'name': speaker_metadata['name'],
'age': speaker_metadata['age'],
'gender': speaker_metadata['gender'],
'accent': speaker_metadata['accent'],
'nativity': speaker_metadata['nativity'],
'city': speaker_metadata['city'],
"date":today_ymd,
"timestamp": timestamp_str }
dump_json(metadata, json_file_path)
# Simply upload the audio file and metadata using the hub's upload_file
# Upload the audio
repo_audio_path = os.path.join(REPOSITORY_DIR, project_name, today_ymd, spk_name, "audio", audio_fn)
_ = upload_file(path_or_fileobj = audio_output_fp,
path_in_repo = repo_audio_path,
repo_id = REPO_NAME,
repo_type = 'dataset',
token = HF_TOKEN_WRITE
)
# Upload the metadata
repo_json_path = os.path.join(REPOSITORY_DIR, project_name, today_ymd, spk_name, "meta", meta_fn)
_ = upload_file(path_or_fileobj = json_file_path,
path_in_repo = repo_json_path,
repo_id = REPO_NAME,
repo_type = 'dataset',
token = HF_TOKEN_WRITE
)
output = print(f"Recording {audio_fn} and meta file {meta_fn} successfully saved to repo!")
prompts_left_info = prompts_left_info - 1
# check if this is the last prompt
return [next_prompt, prompt_number, None, prompts_left_info]
def whisper_model_change(radio_whisper_model):
whisper_model = whisper.load_model(radio_whisper_model)
return(whisper_model)
def prompt_gpt_assistant(input_text, api_key, temperature):
#, role, template_prompt, template_answer):
#TODO add option to specify instruction
openai.api_key = api_key
#TODO add specific message for specific role
system_role_message="You are a helpful assistant"
messages = [
{"role": "system", "content": system_role_message}]
if input_text:
messages.append(
{"role": "user", "content": input_text},
)
chat_completion = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
temperature=temperature
)
reply = chat_completion.choices[0].message.content
#TODO save chat completion for future reuse
return reply
def voicebot_pipeline(audio):
asr_out = transcribe(audio)
gpt_out = prompt_gpt_assistant(asr_out)
tts_out = synthesize_speech(gpt_out)
return(tts_out)
def transcribe(audio, language_code, whisper_model, whisper_model_type):
if not whisper_model:
whisper_model=init_whisper_model(whisper_model_type)
print(f"Transcribing {audio} for language_code {language_code} and model {whisper_model_type}")
audio = whisper.load_audio(audio)
audio = whisper.pad_or_trim(audio)
mel = whisper.log_mel_spectrogram(audio)
options = whisper.DecodingOptions(language=language_code, without_timestamps=True, fp16=False)
result = whisper.decode(whisper_model, mel, options)
result_text = result.text
return result_text
def init_whisper_model(whisper_model_type):
print("Initializing whisper model")
print(whisper_model_type)
whisper_model = whisper.load_model(whisper_model_type)
return whisper_model
def synthesize_speech(text, language_code):
audioobj = gTTS(text = text,
lang = language_code,
slow = False)
audioobj.save("Temp.mp3")
return("Temp.mp3")
block = gr.Blocks(css=css_file)
with block:
#state variables
project_name = gr.State("voicebot") # voicebot is default for playground. For recording app, it is selected e.g. bridge
language_code = gr.State("pl")
prompts_type = gr.State()
promptset = gr.State("test.prompts.txt")
prompt_history = gr.State()
current_prompt = gr.State()
prompt_number = gr.State()
finished_recording = gr.State()
temperature = gr.State(0)
whisper_model_type = gr.State("base")
whisper_model = gr.State()
openai_api_key = gr.State()
google_api_key = gr.State()
azure_api_key = gr.State()
spk_age = gr.State("unknown")
spk_accent = gr.State("unknown")
spk_city = gr.State("unknown")
spk_gender = gr.State("unknown")
spk_nativity = gr.State("unknown")
spk_name = gr.State("unknown")
cities = sorted(dict_origin["Poland"]["cities"])
# state handling functions
def change_project(choice):
print("Changing project to")
print(choice)
project=choice
return(project)
def change_prompts_type(choice):
print("Changing promptset type to")
print(choice)
prompts_type=choice
return(prompts_type)
def change_nativity(choice):
print("Changing speaker nativity to")
print(choice)
spk_nativity=choice
return(spk_nativity)
def change_accent(choice):
print("Changing speaker accent to")
print(choice)
spk_accent=choice
return(spk_accent)
def change_age(choice):
print("Changing speaker age to")
print(choice)
spk_age=choice
return(spk_age)
def change_city(choice):
print("Changing speaker city to")
print(choice)
spk_city=choice
return(spk_city)
def change_gender(choice):
print("Changing speaker gender to")
print(choice)
spk_gender=choice
return(spk_gender)
def change_language(choice):
if choice == "Polish":
language_code="pl"
print("Switching to Polish")
print("language_code")
print(language_code)
elif choice == "English":
language_code="en"
print("Switching to English")
print("language_code")
print(language_code)
return(language_code)
def change_whisper_model(choice):
whisper_model_type = choice
print("Switching Whisper model")
print(whisper_model_type)
whisper_model = init_whisper_model(whisper_model_type)
return [whisper_model_type, whisper_model]
def change_prompts_left(prompts_left, current_prompt, promptset_size):
prompts_left = promptset_size - current_prompt
return [prompts_left]
gr.Markdown(markdown)
with gr.Tabs():
"""with gr.TabItem('General settings'):
radio_lang = gr.Radio(["Polish", "English"], label="Language", info="If none is selected, Polish is used")
radio_asr_type = gr.Radio(["Local", "Cloud"], label="Select ASR type", info="Cloud models are faster and more accurate, but costs money")
with gr.Accordion(label="Local ASR settings", open=False):
#radio_asr_type = gr.Radio(["Local", "Cloud"], label="Select ASR type", info="Cloud models are faster and more accurate, but costs money")
#radio_cloud_asr = gr.Radio(["Whisper", "Google", "Azure"], label="Select Cloud ASR provider", info="You need to provide API keys for specific service")
radio_whisper_model = gr.Radio(["tiny", "base", "small", "medium", "large"], label="Whisper ASR model (local)", info="Larger models are more accurate, but slower. Default - base")
with gr.Accordion(label="Cloud ASR settings", open=False):
radio_cloud_asr = gr.Radio(["Whisper", "Google", "Azure"], label="Select Cloud ASR provider", info="You need to provide API keys for specific service")
with gr.Accordion(label="Cloud API Keys",open=False):
gr.HTML("<p class=\"apikey\">Open AI API Key:</p>")
# API key textbox (password-style)
openai_api_key = gr.Textbox(label="", elem_id="pw")
gr.HTML("<p class=\"apikey\">Google Cloud API Key:</p>")
# API key textbox (password-style)
google_api_key = gr.Textbox(label="", elem_id="pw")
gr.HTML("<p class=\"apikey\">Azure Cloud API Key:</p>")
# API key textbox (password-style)
azure_api_key = gr.Textbox(label="", elem_id="pw")
with gr.Accordion(label="Chat GPT settings",open=False):
slider_temp = gr.Slider(minimum=0, maximum= 2, step=0.2, label="ChatGPT temperature")
"""
with gr.TabItem('Speaker information'):
with gr.Row():
spk_name = gr.Textbox(placeholder="Your name", label="Name")
dropdown_spk_nativity = gr.Dropdown(["Polish", "Other"], label="Native language", info="")
dropdown_spk_gender = gr.Dropdown(["Male", "Female", "Other", "Prefer not to say"], label="Gender", info="")
dropdown_spk_age = gr.Dropdown(["under 20", "20-29", "30-39", "40-49", "50-59", "over 60"], label="Age", info="")
dropdown_spk_origin_city = gr.Dropdown(cities, label="Hometown", visible=True, info="Closest city to speaker's place of birth and upbringing")
#radio_gdpr_consent = gr.Radio(["Yes", "No"], label="Personal data processing consent", info="Do you agree for your personal data processing according to the policy (link)")
dropdown_spk_nativity.change(fn=change_nativity, inputs=dropdown_spk_nativity, outputs=spk_age)
dropdown_spk_gender.change(fn=change_gender, inputs=dropdown_spk_gender, outputs=spk_gender)
dropdown_spk_age.change(fn=change_age, inputs=dropdown_spk_age, outputs=spk_age)
dropdown_spk_origin_city.change(fn=change_city, inputs=dropdown_spk_origin_city, outputs=spk_city)
"""with gr.TabItem('Voicebot playground'):
mic_recording = gr.Audio(source="microphone", type="filepath", label='Record your voice')
with gr.Row():
button_transcribe = gr.Button("Transcribe speech")
button_save_audio_and_trans = gr.Button("Save audio recording and transcription")
out_asr = gr.Textbox(placeholder="ASR output",
lines=2,
max_lines=5,
show_label=False)
with gr.Row():
button_prompt_gpt = gr.Button("Prompt ChatGPT")
button_save_gpt_response = gr.Button("Save ChatGPT response")
out_gpt = gr.Textbox(placeholder="ChatGPT output",
lines=4,
max_lines=10,
show_label=False)
with gr.Row():
button_synth_speech = gr.Button("Synthesize speech")
button_save_synth_audio = gr.Button("Save synthetic audio")
synth_recording = gr.Audio()
# Events actions
button_save_audio_and_trans.click(save_recording_and_meta, inputs=[project_name, mic_recording, out_asr, language_code, spk_age, spk_accent, spk_city, spk_gender, spk_nativity], outputs=[])
button_transcribe.click(transcribe, inputs=[mic_recording, language_code, whisper_model,whisper_model_type], outputs=out_asr)
button_prompt_gpt.click(prompt "dates":["20230922"],
"speakers":["Test"]_gpt_assistant, inputs=[out_asr, openai_api_key, slider_temp], outputs=out_gpt)
button_synth_speech.click(synthesize_speech, inputs=[out_gpt, language_code], outputs=synth_recording)
radio_lang.change(fn=change_language, inputs=radio_lang, outputs=language_code)
radio_whisper_model.change(fn=change_whisper_model, inputs=radio_whisper_model, outputs=[whisper_model_type, whisper_model])
"""
with gr.TabItem('Speech recordings app'):
with gr.Accordion(label="Project settings"):
radio_project = gr.Dropdown(["bridge"], label="Select project", info="")
#radio_promptset_type = gr.Radio(["New promptset generation", "Existing promptset use"], label="Language", value ="Existing promptset use", info="New promptset is generated using. Requires providing open AI key in general settings tab")
var_promptset_size = gr.Textbox(label="How many recordings do you intend to make? (max 200)")
button_get_prompts = gr.Button("Save settings and get first prompt!")
prompts_left_info = gr.Number(placeholder='',label="Recordings left",lines=1, max_lines=1, show_label=True, interactive=False)
prompt_text = gr.Textbox(placeholder='Prompt to read during recording',label="Prompt to read")
speech_recording = gr.Audio(source="microphone",label="Select 'record from microphone' and read the prompt displayed above", type="filepath")
radio_project.change(fn=change_project, inputs=radio_project, outputs=project_name)
#radio_promptset_type.change(fn=change_prompts_type, inputs=radio_promptset_type, outputs=prompts_type)
#prompts_left.change(change_prompts_left, inputs = [prompts_left, current_prompt, var_promptset_size], outputs = [prompts_left])
button_save_and_next = gr.Button("Save recording and get the next prompt")
# TODO - add option to generate new promptset on the fly for new projects
button_get_prompts.click(get_prompts, inputs=[radio_project, var_promptset_size, language_code, prompts_left_info], outputs = [promptset, prompt_text, prompts_left_info])
button_save_and_next.click(save_recording_and_meta, inputs=[project_name, speech_recording, prompt_text, language_code, spk_name, spk_age, spk_accent, spk_city, spk_gender, spk_nativity, promptset, prompt_number, prompts_left_info], outputs=[prompt_text, prompt_number, speech_recording,prompts_left_info])
block.launch()