File size: 7,587 Bytes
8e73e42 bf4b781 8e73e42 bf4b781 8e73e42 bf4b781 8e73e42 bf4b781 8e73e42 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 |
import torch
import gradio as gr
import yt_dlp as youtube_dl
import numpy as np
from datasets import Dataset, Audio
from scipy.io import wavfile
from transformers import pipeline
from transformers.pipelines.audio_utils import ffmpeg_read
import tempfile
import os
import time
MODEL_NAME = "openai/whisper-large-v3"
BATCH_SIZE = 8
FILE_LIMIT_MB = 1000
YT_LENGTH_LIMIT_S = 3600 # limit to 1 hour YouTube files
device = 0 if torch.cuda.is_available() else "cpu"
pipe = pipeline(
task="automatic-speech-recognition",
model=MODEL_NAME,
chunk_length_s=30,
device=device,
)
def transcribe(inputs_path, task, dataset_name, oauth_token: gr.OAuthToken):
if inputs_path is None:
raise gr.Error("No audio file submitted! Please upload or record an audio file before submitting your request.")
sampling_rate, inputs = wavfile.read(inputs_path)
out = pipe(inputs_path, batch_size=BATCH_SIZE, generate_kwargs={"task": task}, return_timestamps=True)
text = out["text"]
chunks = naive_postprocess_whisper_chunks(out["chunks"])
transcripts = []
audios = []
with tempfile.TemporaryDirectory() as tmpdirname:
for i,chunk in enumerate(chunks):
begin, end = chunk["timestamp"]
begin, end = int(begin*sampling_rate), int(end*sampling_rate)
# TODO: make sure 1D or 2D?
arr = inputs[begin:end]
path = os.path.join(tmpdirname, f"{i}.wav")
wavfile.write(path, sampling_rate, arr)
audios.append(path)
transcripts.append(chunk["text"])
dataset = Dataset.from_dict({"audio": audios, "transcript": transcripts}).cast_column("audio", Audio())
dataset.push_to_hub(dataset_name, token=oauth_token)
return text
def _return_yt_html_embed(yt_url):
video_id = yt_url.split("?v=")[-1]
HTML_str = (
f'<center> <iframe width="500" height="320" src="https://www.youtube.com/embed/{video_id}"> </iframe>'
" </center>"
)
return HTML_str
def download_yt_audio(yt_url, filename):
info_loader = youtube_dl.YoutubeDL()
try:
info = info_loader.extract_info(yt_url, download=False)
except youtube_dl.utils.DownloadError as err:
raise gr.Error(str(err))
file_length = info["duration_string"]
file_h_m_s = file_length.split(":")
file_h_m_s = [int(sub_length) for sub_length in file_h_m_s]
if len(file_h_m_s) == 1:
file_h_m_s.insert(0, 0)
if len(file_h_m_s) == 2:
file_h_m_s.insert(0, 0)
file_length_s = file_h_m_s[0] * 3600 + file_h_m_s[1] * 60 + file_h_m_s[2]
if file_length_s > YT_LENGTH_LIMIT_S:
yt_length_limit_hms = time.strftime("%HH:%MM:%SS", time.gmtime(YT_LENGTH_LIMIT_S))
file_length_hms = time.strftime("%HH:%MM:%SS", time.gmtime(file_length_s))
raise gr.Error(f"Maximum YouTube length is {yt_length_limit_hms}, got {file_length_hms} YouTube video.")
ydl_opts = {"outtmpl": filename, "format": "worstvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best"}
with youtube_dl.YoutubeDL(ydl_opts) as ydl:
try:
ydl.download([yt_url])
except youtube_dl.utils.ExtractorError as err:
raise gr.Error(str(err))
def yt_transcribe(yt_url, task, dataset_name, oauth_token: gr.OAuthToken, max_filesize=75.0, dataset_sampling_rate = 24000):
html_embed_str = _return_yt_html_embed(yt_url)
with tempfile.TemporaryDirectory() as tmpdirname:
filepath = os.path.join(tmpdirname, "video.mp4")
download_yt_audio(yt_url, filepath)
with open(filepath, "rb") as f:
inputs_path = f.read()
inputs = ffmpeg_read(inputs_path, pipe.feature_extractor.sampling_rate)
inputs = {"array": inputs, "sampling_rate": pipe.feature_extractor.sampling_rate}
out = pipe(inputs, batch_size=BATCH_SIZE, generate_kwargs={"task": task}, return_timestamps=True)
text = out["text"]
chunks = naive_postprocess_whisper_chunks(out["chunks"])
inputs = ffmpeg_read(inputs_path, dataset_sampling_rate)
transcripts = []
audios = []
with tempfile.TemporaryDirectory() as tmpdirname:
for i,chunk in enumerate(chunks):
begin, end = chunk["timestamp"]
begin, end = int(begin*dataset_sampling_rate), int(end*dataset_sampling_rate)
# TODO: make sure 1D or 2D?
arr = inputs[begin:end]
path = os.path.join(tmpdirname, f"{i}.wav")
wavfile.write(path, dataset_sampling_rate, arr)
audios.append(path)
transcripts.append(chunk["text"])
dataset = Dataset.from_dict({"audio": audios, "transcript": transcripts}).cast_column("audio", Audio())
dataset.push_to_hub(dataset_name, token=oauth_token)
return html_embed_str, text
def naive_postprocess_whisper_chunks(chunks, stop_chars = ".!:;?", min_duration = 5):
new_chunks = []
while chunks:
current_chunk = chunks.pop(0)
begin, end = current_chunk["timestamp"]
text = current_chunk["text"]
while chunks and (text[-1] not in stop_chars or (end-begin<min_duration)):
ch = chunks.pop(0)
end = ch["timestamp"][1]
text = "".join([text, ch["text"]])
new_chunks.append({
"text": text.strip(),
"timestamp": (begin, end),
})
print(f"LENGTH CHUNK #{len(new_chunks)}: {end-begin}s")
return new_chunks
demo = gr.Blocks()
mf_transcribe = gr.Interface(
fn=transcribe,
inputs=[
gr.Audio(type="filepath"),
gr.Radio(["transcribe", "translate"], label="Task", value="transcribe"),
gr.Textbox(lines=1, placeholder="Place your new dataset name here", label="Dataset name"),
],
outputs="text",
theme="huggingface",
title="Create your own TTS dataset using your own recordings",
description=(
"This demo allows use to create a text-to-speech dataset from an input audio snippet and push it to hub to keep track of it."
f"Demo uses the checkpoint [{MODEL_NAME}](https://huggingface.co/{MODEL_NAME}) and 🤗 Transformers to automatically transcribe audio files"
" of arbitrary length. It then merge chunks of audio and push it to the hub."
),
allow_flagging="never",
)
yt_transcribe = gr.Interface(
fn=yt_transcribe,
inputs=[
gr.Textbox(lines=1, placeholder="Paste the URL to a YouTube video here", label="YouTube URL"),
gr.Radio(["transcribe", "translate"], label="Task", value="transcribe"),
gr.Textbox(lines=1, placeholder="Place your new dataset name here", label="Dataset name"),
],
outputs=["html", "text"],
theme="huggingface",
title="Create your own TTS dataset using Youtube",
description=(
"This demo allows use to create a text-to-speech dataset from an input audio snippet and push it to hub to keep track of it."
f"Demo uses the checkpoint [{MODEL_NAME}](https://huggingface.co/{MODEL_NAME}) and 🤗 Transformers to automatically transcribe audio files"
" of arbitrary length. It then merge chunks of audio and push it to the hub."
),
allow_flagging="never",
)
with demo:
with gr.Row():
gr.LoginButton()
gr.LogoutButton()
gr.TabbedInterface([mf_transcribe, yt_transcribe], ["Microphone or Audio file", "YouTube"])
demo.launch(debug=True)
|