mj-new
Added support for datasets without secret test split
3533dd6
raw
history blame
26.4 kB
import json
import seaborn as sns
import matplotlib.pyplot as plt
import pandas as pd
import os
import numpy as np
# move to consts
buckets_age=['teens','twenties', 'thirties', 'fourties', 'fifties', 'sixties', 'seventies', 'eighties', 'nineties']
buckets_sex=["male", "female"]
def load_bigos_analyzer_report(fp:str)->dict:
with open(fp, 'r') as f:
data = json.load(f)
return data
def num_of_samples_per_split(dataset_hf):
# input - huggingface dataset object
# output - dictionary with statistics about number of samples per split
out_dict = {}
# number of samples per subset and split
metric = "samples"
print("Calculating {}".format(metric))
for split in dataset_hf.keys():
samples = dataset_hf[split].num_rows
##print(split, samples)
out_dict[split] = samples
# add number of samples for all splits
out_dict["all_splits"] = sum(out_dict.values())
return out_dict
def total_audio_duration_per_split(dataset_hf):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
metric = "audio[h]"
print("Calculating {}".format(metric))
for split in dataset_hf.keys():
#sampling_rate = dataset_hf[split]["sampling_rate"][0]
#audio_total_length_samples = 0
#audio_total_length_samples = sum(len(audio_file["array"]) for audio_file in dataset_hf["test"]["audio"])
audio_total_length_seconds = sum(dataset_hf[split]["audio_duration_seconds"])
audio_total_length_hours = round(audio_total_length_seconds / 3600,2)
out_dict[split] = audio_total_length_hours
#print(split, audio_total_length_hours)
# add number of samples for all splits
out_dict["all_splits"] = sum(out_dict.values())
return out_dict
def average_audio_duration_per_split(dataset_hf):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
metric = "average_audio_duration[s]"
print("Calculating {}".format(metric))
samples_all=0
audio_length_total_seconds=0
for split in dataset_hf.keys():
#sampling_rate = dataset_hf[split]["sampling_rate"][0]
#audio_total_length_samples = 0
#audio_total_length_samples = sum(len(audio_file["array"]) for audio_file in dataset_hf["test"]["audio"])
audio_length_split_seconds = sum(dataset_hf[split]["audio_duration_seconds"])
audio_length_total_seconds += audio_length_split_seconds
samples_split = len(dataset_hf[split]["audio_duration_seconds"])
samples_all += samples_split
audio_average_length_seconds = round(audio_length_split_seconds / samples_split,2)
out_dict[split] = audio_average_length_seconds
#print(split, audio_total_length_hours)
# add number of samples for all splits
out_dict["all_splits"] = round(audio_length_total_seconds / samples_all,2)
return out_dict
def average_utterance_length_chars_per_split(dataset_hf, dataset_hf_secret):
# input - huggingface dataset object
# output - dictionary with statistics about average utterance length per split
out_dict = {}
metric = "average_utterance_length[chars]"
print("Calculating {}".format(metric))
chars_all=0
samples_all=0
for split in dataset_hf.keys():
# extract speakers from file_id
if (split=="test" and dataset_hf_secret is not None):
utts_split = dataset_hf_secret[split]["ref_orig"]
else:
utts_split = dataset_hf[split]["ref_orig"]
words_split = " ".join(utts_split).split(" ")
chars_split = " ".join(words_split)
chars_split_count = len(chars_split)
chars_all += chars_split_count
samples_split = len(utts_split)
samples_all += samples_split
#print(split, chars_all_count)
out_dict[split] = round(chars_split_count/samples_split, 2)
# add number of samples for all splits
out_dict["all_splits"] = round(chars_all/samples_all, 2)
return out_dict
def average_utterance_length_words_per_split(dataset_hf, dataset_hf_secret):
# input - huggingface dataset object
# output - dictionary with statistics about average utterance length per split
out_dict = {}
metric = "average_utterance_length[words]"
print("Calculating {}".format(metric))
words_all=0
samples_all=0
for split in dataset_hf.keys():
# extract speakers from file_id
if (split=="test" and dataset_hf_secret is not None):
utts_split = dataset_hf_secret[split]["ref_orig"]
else:
utts_split = dataset_hf[split]["ref_orig"]
words_split_count = len(" ".join(utts_split).split(" "))
words_all += words_split_count
samples_split = len(utts_split)
samples_all += samples_split
#print(split, chars_all_count)
out_dict[split] = round(words_split_count/samples_split , 2)
# add number of samples for all splits
out_dict["all_splits"] = round(words_all/samples_all, 2)
return out_dict
def speakers_per_split(dataset_hf):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
metric = "speakers"
print("Calculating {}".format(metric))
for split in dataset_hf.keys():
# extract speakers from file_id
speakers_ids_all = [str(fileid).split("-")[4] for fileid in dataset_hf[split]["audioname"]]
speakers_ids_uniq = list(set(speakers_ids_all))
speakers_count = len(speakers_ids_uniq)
#print(split, speakers_count)
out_dict[split] = speakers_count
# add number of samples for all splits
out_dict["all_splits"] = sum(out_dict.values())
return out_dict
def uniq_utts_per_split(dataset_hf, dataset_hf_secret):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
metric = "utts_unique"
print("Calculating {}".format(metric))
utts_all = []
for split in dataset_hf.keys():
# extract speakers from file_id
if (split == "test" and dataset_hf_secret is not None):
utts_split = dataset_hf_secret[split]["ref_orig"]
else:
utts_split = dataset_hf[split]["ref_orig"]
utts_all = utts_all + utts_split
utts_uniq = list(set(utts_split))
utts_uniq_count = len(utts_uniq)
#print(split, utts_uniq_count)
out_dict[split] = utts_uniq_count
# add number of samples for all splits
out_dict["all_splits"] = len(list(set(utts_all)))
return out_dict,utts_all
def words_per_split(dataset_hf, dataset_hf_secret):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
metric = "words"
print("Calculating {}".format(metric))
for split in dataset_hf.keys():
# extract speakers from file_id
if (split == "test" and dataset_hf_secret is not None):
utts_all = dataset_hf_secret[split]["ref_orig"]
else:
utts_all = dataset_hf[split]["ref_orig"]
utts_lenghts = [len(utt.split(" ")) for utt in utts_all]
words_all_count = sum(utts_lenghts)
#print(split, words_all_count)
out_dict[split] = words_all_count
# add number of samples for all splits
out_dict["all_splits"] = sum(out_dict.values())
return out_dict
def uniq_words_per_split(dataset_hf, dataset_hf_secret):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
out_words_list = []
metric = "words_unique"
print("Calculating {}".format(metric))
for split in dataset_hf.keys():
# extract speakers from file_id
if (split == "test" and dataset_hf_secret is not None):
utts_all = dataset_hf_secret[split]["ref_orig"]
else:
utts_all = dataset_hf[split]["ref_orig"]
words_all = " ".join(utts_all).split(" ")
words_uniq = list(set(words_all))
out_words_list = out_words_list + words_uniq
words_uniq_count = len(words_uniq)
#print(split, words_uniq_count)
out_dict[split] = words_uniq_count
# add number of samples for all splits
out_words_uniq = list(set((out_words_list)))
out_words_uniq_count = len(out_words_uniq)
out_dict["all_splits"] = out_words_uniq_count
#print("all", out_words_uniq_count)
return out_dict, out_words_uniq
def chars_per_split(dataset_hf, dataset_hf_secret):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
metric = "chars"
print("Calculating {}".format(metric))
for split in dataset_hf.keys():
# extract speakers from file_id
if (split=="test" and dataset_hf_secret is not None):
utts_all = dataset_hf_secret[split]["ref_orig"]
else:
utts_all = dataset_hf[split]["ref_orig"]
words_all = " ".join(utts_all).split(" ")
chars_all = " ".join(words_all)
chars_all_count = len(chars_all)
#print(split, chars_all_count)
out_dict[split] = chars_all_count
# add number of samples for all splits
out_dict["all_splits"] = sum(out_dict.values())
return out_dict
def uniq_chars_per_split(dataset_hf, dataset_hf_secret):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
out_chars_list = []
metric = "chars_unique"
print("Calculating {}".format(metric))
for split in dataset_hf.keys():
# extract speakers from file_id
if(split == "test" and dataset_hf_secret is not None):
utts_all = dataset_hf_secret[split]["ref_orig"]
else:
utts_all = dataset_hf[split]["ref_orig"]
words_all = " ".join(utts_all).split(" ")
words_uniq = list(set(words_all))
chars_uniq = list(set("".join(words_uniq)))
chars_uniq_count = len(chars_uniq) + 1
#print(split, chars_uniq_count)
out_dict[split] = chars_uniq_count
out_chars_list = out_chars_list + chars_uniq
# add number of samples for all splits
out_chars_uniq = list(set((out_chars_list)))
out_chars_uniq_count = len(out_chars_uniq)
out_dict["all_splits"] = out_chars_uniq_count
#print("all", out_chars_uniq_count)
return out_dict, out_chars_uniq
def meta_cov_per_split(dataset_hf, meta_field):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
no_meta=False
# TODO move to config
if meta_field == 'speaker_age':
buckets = buckets_age
if meta_field == 'speaker_sex':
buckets = buckets_sex
out_dict = {}
metric = "meta_cov_" + meta_field
print("Calculating {}".format(metric))
meta_info_all = 0
meta_info_not_null_all = 0
for split in dataset_hf.keys():
# extract speakers from file_id
meta_info = dataset_hf[split][meta_field]
meta_info_count = len(meta_info)
meta_info_all += meta_info_count
# calculate coverage
meta_info_not_null_count = len([x for x in meta_info if x != "N/A"])
if meta_info_not_null_count == 0:
out_dict[split] = "N/A"
continue
meta_info_not_null_all += meta_info_not_null_count
meta_info_coverage = round(meta_info_not_null_count / meta_info_count * 100, 2)
out_dict[split] = meta_info_coverage
if (meta_info_not_null_all == 0):
out_dict["all_splits"] = "N/A"
else:
out_dict["all_splits"] = round(meta_info_not_null_all/meta_info_all * 100,2 )
return out_dict
def speech_rate_words_per_split(dataset_hf, dataset_hf_secret):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
metric = "words_per_second"
print("Calculating {}".format(metric))
words_all_count = 0
audio_total_length_seconds = 0
for split in dataset_hf.keys():
# extract speakers from file_id
if (split == "test" and dataset_hf_secret is not None):
utts_split = dataset_hf_secret[split]["ref_orig"]
else:
utts_split = dataset_hf[split]["ref_orig"]
words_split = " ".join(utts_split).split(" ")
words_split_count = len(words_split)
words_all_count += words_split_count
audio_split_length_seconds = sum(dataset_hf[split]["audio_duration_seconds"])
audio_total_length_seconds += audio_split_length_seconds
speech_rate = round(words_split_count / audio_split_length_seconds, 2)
out_dict[split] = speech_rate
out_dict["all_splits"] = round(words_all_count / audio_total_length_seconds, 2)
return out_dict
def speech_rate_chars_per_split(dataset_hf, dataset_hf_secret):
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
metric = "chars_per_second"
print("Calculating {}".format(metric))
chars_all_count = 0
audio_total_length_seconds = 0
for split in dataset_hf.keys():
# extract speakers from file_id
if (split == "test" and dataset_hf_secret is not None):
utts_split = dataset_hf_secret[split]["ref_orig"]
else:
utts_split = dataset_hf[split]["ref_orig"]
words_split = " ".join(utts_split).split(" ")
chars_split_count = len("".join(words_split))
chars_all_count += chars_split_count
audio_split_length_seconds = sum(dataset_hf[split]["audio_duration_seconds"])
audio_total_length_seconds += audio_split_length_seconds
speech_rate = round(chars_split_count / audio_split_length_seconds, 2)
out_dict[split] = speech_rate
out_dict["all_splits"] = round(chars_all_count / audio_total_length_seconds, 2)
return out_dict
# distribution of speaker age
def meta_distribution_text(dataset_hf, meta_field):
no_meta=False
if meta_field == 'speaker_age':
buckets = buckets_age
if meta_field == 'speaker_sex':
buckets = buckets_sex
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict = {}
metric = "distribution_" + meta_field
print("Calculating {}".format(metric))
values_count_total = {}
for bucket in buckets:
values_count_total[bucket]=0
for split in dataset_hf.keys():
out_dict[split] = {}
# extract speakers from file_id
meta_info = dataset_hf[split][meta_field]
meta_info_not_null = [x for x in meta_info if x != "N/A"]
if len(meta_info_not_null) == 0:
out_dict[split]="N/A"
no_meta=True
continue
for bucket in buckets:
values_count = meta_info_not_null.count(bucket)
values_count_total[bucket] += values_count
out_dict[split][bucket] = round(values_count/len(meta_info_not_null),2)
#print(split, out_dict[split])
if (no_meta):
out_dict["all_splits"] = "N/A"
return out_dict
out_dict["all_splits"] = {}
# calculate total number of samples in values_count_total
for bucket in buckets:
total_samples = sum(values_count_total.values())
out_dict["all_splits"][bucket] = round(values_count_total[bucket]/total_samples,2)
return out_dict
def recordings_per_speaker(dataset_hf):
recordings_per_speaker_stats_dict = {}
# input - huggingface dataset object
# output - dictionary with statistics about audio duration per split
out_dict_stats = {}
out_dict_contents = {}
metric = "recordings_per_speaker"
print("Calculating {}".format(metric))
recordings_per_speaker_stats_dict_all = {}
recordings_total=0
speakers_total = 0
for split in dataset_hf.keys():
# extract speakers from file_id
audiopaths = dataset_hf[split]["audioname"]
speaker_prefixes = [str(fileid).split("-")[0:5] for fileid in audiopaths]
speakers_dict_split = {}
# create dictionary with list of audio paths matching speaker prefix
# Create initial dictionary keys from speaker prefixes
for speaker_prefix in speaker_prefixes:
speaker_prefix_str = "-".join(speaker_prefix)
speakers_dict_split[speaker_prefix_str] = []
# Populate the dictionary with matching audio paths
for audio_path in audiopaths:
for speaker_prefix_str in speakers_dict_split.keys():
if speaker_prefix_str in audio_path:
speakers_dict_split[speaker_prefix_str].append(audio_path)
# iterate of speaker_dict prefixes and calculate number of recordings per speaker.
recordings_per_speaker_stats_dict_split = {}
for speaker_prefix_str in speakers_dict_split.keys():
recordings_per_speaker_stats_dict_split[speaker_prefix_str] = len(speakers_dict_split[speaker_prefix_str])
out_dict_contents[split] = {}
out_dict_contents[split] = recordings_per_speaker_stats_dict_split
# use recordings_per_speaker_stats to calculate statistics like min, max, avg, median, std
out_dict_stats[split] = {}
speakers_split = len(list(recordings_per_speaker_stats_dict_split.keys()))
speakers_total += speakers_split
recordings_split = len(audiopaths)
recordings_total += recordings_split
average_recordings_per_speaker = round( recordings_split / speakers_split,2)
out_dict_stats[split]["average"] = average_recordings_per_speaker
out_dict_stats[split]["std"] = round(np.std(list(recordings_per_speaker_stats_dict_split.values())),2)
out_dict_stats[split]["median"] = np.median(list(recordings_per_speaker_stats_dict_split.values()))
out_dict_stats[split]["min"] = min(recordings_per_speaker_stats_dict_split.values())
out_dict_stats[split]["max"] = max(recordings_per_speaker_stats_dict_split.values())
recordings_per_speaker_stats_dict_all = recordings_per_speaker_stats_dict_all | recordings_per_speaker_stats_dict_split
# add number of samples for all splits
average_recordings_per_speaker_all = round( recordings_total / speakers_total , 2)
out_dict_stats["all_splits"] = {}
out_dict_stats["all_splits"]["average"] = average_recordings_per_speaker_all
out_dict_stats["all_splits"]["std"] = round(np.std(list(recordings_per_speaker_stats_dict_all.values())),2)
out_dict_stats["all_splits"]["median"] = np.median(list(recordings_per_speaker_stats_dict_all.values()))
out_dict_stats["all_splits"]["min"] = min(recordings_per_speaker_stats_dict_all.values())
out_dict_stats["all_splits"]["max"] = max(recordings_per_speaker_stats_dict_all.values())
out_dict_contents["all_splits"] = recordings_per_speaker_stats_dict_all
return out_dict_stats, out_dict_contents
def meta_distribution_bar_plot(dataset_hf, output_dir, dimension = "speaker_sex"):
pass
def meta_distribution_violin_plot(dataset_hf, output_dir, metric = "audio_duration_seconds", dimension = "speaker_sex"):
# input - huggingface dataset object
# output - figure with distribution of audio duration per sex
out_dict = {}
print("Generating violin plat for metric {} for dimension {}".format(metric, dimension))
# drop samples for which dimension column values are equal to "N/A"
for split in dataset_hf.keys():
df_dataset = pd.DataFrame(dataset_hf[split])
# remove values equal to "N/A" for column dimension
df_filtered = df_dataset[df_dataset[dimension] != "N/A"]
df_filtered = df_filtered[df_filtered[dimension] != "other"]
df_filtered = df_filtered[df_filtered[dimension] != "unknown"]
if df_filtered.empty:
print("No data for split {} and dimension {}".format(split, dimension))
continue
if (len(df_filtered)>=5000):
sample_size = 5000
print("Selecting sample of size {}".format(sample_size))
else:
sample_size = len(df_filtered)
print("Selecting full split of size {}".format(sample_size))
df = df_filtered.sample(sample_size)
# if df_filtered is empty, skip violin plot generation for this split and dimension
print("Generating plot")
plt.figure(figsize=(20, 15))
plot = sns.violinplot(data = df, hue=dimension, x='dataset', y=metric, split=True, fill = False,inner = 'quart', legend='auto', common_norm=True)
plot.set_xticklabels(plot.get_xticklabels(), rotation = 30, horizontalalignment = 'right')
plt.title('Violin plot of {} by {} for split {}'.format(metric, dimension, split))
plt.xlabel(dimension)
plt.ylabel(metric)
#plt.show(
# save figure to file
os.makedirs(output_dir, exist_ok=True)
output_fn = os.path.join(output_dir, metric + "-" + dimension + "-" + split + ".png")
plt.savefig(output_fn)
print("Plot generation completed")
def read_reports(dataset_name):
json_contents = "./reports/{}/dataset_contents.json".format(dataset_name)
json_stats = "reports/{}/dataset_statistics.json".format(dataset_name)
with open(json_contents, 'r') as file:
contents_dict = json.load(file)
with open(json_stats, 'r') as file:
stats_dict = json.load(file)
return(stats_dict, contents_dict)
def add_test_split_stats_from_secret_dataset(stats_dict_public, stats_dict_secret):
# merge contents if dictionaries for fields utts, words, words_unique, chars, chars_unique and speech_rate
for dataset in stats_dict_public.keys():
print(dataset)
for metric in stats_dict_secret[dataset].keys():
for split in stats_dict_secret[dataset][metric].keys():
if split == "test":
stats_dict_public[dataset][metric][split] = stats_dict_secret[dataset][metric][split]
return(stats_dict_public)
def dict_to_multindex_df(dict_in, all_splits=False):
# Creating a MultiIndex DataFrame
rows = []
for dataset, metrics in dict_in.items():
if (dataset == "all"):
continue
for metric, splits in metrics.items():
for split, value in splits.items():
if (all_splits):
if (split == "all_splits"):
rows.append((dataset, metric, split, value))
else:
if (split == "all_splits"):
continue
rows.append((dataset, metric, split, value))
# Convert to DataFrame
df = pd.DataFrame(rows, columns=['dataset', 'metric', 'split', 'value'])
df.set_index(['dataset', 'metric', 'split'], inplace=True)
return(df)
def dict_to_multindex_df_all_splits(dict_in):
# Creating a MultiIndex DataFrame
rows = []
for dataset, metrics in dict_in.items():
if (dataset == "all"):
continue
for metric, splits in metrics.items():
for split, value in splits.items():
if (split == "all_splits"):
rows.append((dataset, metric, split, value))
# Convert to DataFrame
df = pd.DataFrame(rows, columns=['dataset', 'metric', 'split', 'value'])
df.set_index(['dataset', 'metric', 'split'], inplace=True)
return(df)
def extract_stats_to_agg(df_multindex_per_split, metrics, add_total=True):
# input - multiindex dataframe has three indexes - dataset, metric, split
# select only relevant metrics
df_agg_splits = df_multindex_per_split.loc[(slice(None), metrics), :]
# unstack - move rows per split to columns
df_agg_splits = df_agg_splits.unstack(level ='split')
# aggregate values for all splits
df_agg_splits['value', 'total'] = df_agg_splits['value'].sum(axis=1)
# drop columns with splits
df_agg_splits.columns = df_agg_splits.columns.droplevel(0)
columns_to_drop = ['test', 'train', 'validation']
df_agg_splits.drop(columns = columns_to_drop, inplace = True)
# move rows corresponding to specific metrics into specific columns
df_agg_splits = df_agg_splits.unstack(level ='metric')
df_agg_splits.columns = df_agg_splits.columns.droplevel(0)
if(add_total):
# add row with the sum of all rows
df_agg_splits.loc['total'] = df_agg_splits.sum()
return(df_agg_splits)
def extract_stats_all_splits(df_multiindex_all_splits, metrics):
df_all_splits = df_multiindex_all_splits.loc[(slice(None), metrics), :]
df_all_splits = df_all_splits.unstack(level ='metric')
df_all_splits.columns = df_all_splits.columns.droplevel(0)
#print(df_all_splits)
df_all_splits = df_all_splits.droplevel('split', axis=0)
return(df_all_splits)
def extract_stats_for_dataset_card(df_multindex_per_split, subset, metrics, add_total=False):
#print(df_multindex_per_split)
df_metrics_subset = df_multindex_per_split
df_metrics_subset = df_metrics_subset.unstack(level ='split')
df_metrics_subset.columns = df_metrics_subset.columns.droplevel(0)
df_metrics_subset = df_metrics_subset.loc[(slice(None), metrics), :]
df_metrics_subset = df_metrics_subset.query("dataset == '{}'".format(subset))
# change order of columns to train validation test
df_metrics_subset.reset_index(inplace=True)
if (add_total):
new_columns = ['metric', 'train', 'validation', 'test', 'total']
total = df_metrics_subset[['train', 'validation','test']].sum(axis=1)
df_metrics_subset['total'] = total
else:
new_columns = ['metric', 'train', 'validation', 'test']
df_metrics_subset = df_metrics_subset.reindex(columns=new_columns)
df_metrics_subset.set_index('metric', inplace=True)
return(df_metrics_subset)