speech-explorer / app.py
ylacombe's picture
ylacombe HF staff
Update app.py
5511788 verified
from collections import Counter
from itertools import count, groupby, islice
from operator import itemgetter
from typing import Any, Iterable, TypeVar, List, Dict, Tuple, Optional
import gradio as gr
import requests
import pandas as pd
from datasets import Features
from gradio_huggingfacehub_search import HuggingfaceHubSearch
from requests.adapters import HTTPAdapter, Retry
from analyze import run_dataspeech
# import spaces
MAX_ROWS = 100
T = TypeVar("T")
session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504])
session.mount('http://', HTTPAdapter(max_retries=retries))
def stream_rows(dataset: str, config: str, split: str) -> Iterable[Dict[str, Any]]:
batch_size = 100
for i in count():
rows_resp = session.get(f"https://datasets-server.huggingface.co/rows?dataset={dataset}&config={config}&split={split}&offset={i * batch_size}&length={batch_size}", timeout=10).json()
if "error" in rows_resp:
raise RuntimeError(rows_resp["error"])
if not rows_resp["rows"]:
break
for row_item in rows_resp["rows"]:
yield row_item["row"]
class track_iter:
def __init__(self, it: Iterable[T]):
self.it = it
self.next_idx = 0
def __iter__(self) -> T:
for item in self.it:
self.next_idx += 1
yield item
def report(next_row_idx: int, num_rows: int) -> Dict[str, float]:
if num_rows == next_row_idx:
return f"Scan finished: {num_rows} samples analyzed"
else:
return f"Tagging in progress - {next_row_idx/num_rows*100}% of rows analyzed..."
# @spaces.GPU(duration=80)
def analyze_dataset(dataset: str, audio_column_name: str, text_column_name: str, configuration_name: Optional[str] = None, split_name: Optional[str] = None) -> Tuple[str, List[List[Any]]]:
info_resp = session.get(f"https://datasets-server.huggingface.co/info?dataset={dataset}", timeout=3).json()
if "error" in info_resp:
yield "❌ " + info_resp["error"], pd.DataFrame()
return
if configuration_name in info_resp["dataset_info"]:
config = configuration_name
elif configuration_name != "" and configuration_name is not None:
yield "❌ " + f"The configuration you've passed `{configuration_name}` was not found in the dataset configs: {', '.join(info_resp['dataset_info'].keys())}. Try again with the right config name.", gr.DataFrame()
return
else:
config = "default" if "default" in info_resp["dataset_info"] else next(iter(info_resp["dataset_info"]))
features = Features.from_dict(info_resp["dataset_info"][config]["features"])
if split_name in info_resp["dataset_info"][config]["splits"]:
split = split_name
elif split_name != "" and split_name is not None:
yield "❌ " + f"The splt you've passed `{split_name}` was not found in the dataset splits: {', '.join(info_resp['dataset_info'][config]['splits'])}. Try again with the right config name.", gr.DataFrame()
return
else:
split = "train" if "train" in info_resp["dataset_info"][config]["splits"] else next(iter(info_resp["dataset_info"][config]["splits"]))
num_rows = min(info_resp["dataset_info"][config]["splits"][split]["num_examples"], MAX_ROWS)
rows = track_iter(islice(stream_rows(dataset, config, split), MAX_ROWS))
if audio_column_name not in features:
yield "❌ " + f"The audio column name you've passed `{audio_column_name}` was not found in the dataset columns: {', '.join(features.keys())}. Try again with the right column name.", gr.DataFrame()
return
if text_column_name not in features:
yield "❌ " + f"The text column name you've passed `{text_column_name}` was not found in the dataset columns: {', '.join(features.keys())}. Try again with the right column name.", gr.DataFrame()
return
if "gender" in features:
yield "Gender has been detected. We'll compute pitch.", pd.DataFrame()
dataframe = []
for batch in run_dataspeech(
rows, audio_column_name, text_column_name
):
headers = list(batch[0].keys())
batch = [list(sample.values()) for sample in batch]
dataframe.extend(batch)
datatype = ["str" if col != audio_column_name else "markdown" for col in headers]
yield (report(next_row_idx=rows.next_idx, num_rows=num_rows), gr.DataFrame(dataframe, headers=headers, datatype=datatype, wrap=True))
yield (report(next_row_idx=rows.next_idx, num_rows=num_rows), gr.DataFrame(dataframe, headers=headers, datatype=datatype, wrap=True))
with gr.Blocks() as demo:
gr.Markdown("# Analyze speech dataset using Data-Speech")
gr.Markdown("The space takes an HF dataset name as an input, as well as the audio column name to analyze, and returns the speaking rate, noise level, reverberation level, monotony level and pitch. Note that pitch is only computed if a `speaker_id` column and a `gender` column are found.")
hub_search = HuggingfaceHubSearch(
label="Hub Dataset ID",
placeholder="Search for dataset id on Huggingface",
search_type="dataset",
)
audio_column_name = gr.Textbox(
value="audio",
label="Audio column name.",
)
text_column_name = gr.Textbox(
value="text",
label="Transcription column name.",
)
with gr.Accordion("(Optional) specify configuration and split of the dataset to be analysed", open=False):
configuration_name = gr.Textbox(
value=None,
label="Configuration name.",
)
split_name = gr.Textbox(
value=None,
label="Split name.",
)
button = gr.Button("Run Data-Speech Scan")
outputs = [
gr.Label(show_label=False),
gr.DataFrame(),
]
button.click(analyze_dataset, [hub_search, audio_column_name, text_column_name, configuration_name, split_name], outputs)
gr.Examples(
[
["blabble-io/libritts_r", "audio", "text_normalized", "clean"],
["blabble-io/libritts_r", "audio", "text_normalized", "other"],
["espnet/yodas", "audio", "text", "en000",],
["ylacombe/english_dialects", "audio", "text"]
],
[hub_search, audio_column_name, text_column_name, configuration_name],
outputs,
fn=analyze_dataset,
run_on_click=True,
cache_examples=False,
)
demo.launch(debug=False)