pl-asr-leaderboard / utils.py
mj-new
Updated leaderboard code and requirements
37d493c
import pandas as pd
import streamlit as st
import seaborn as sns
import matplotlib.pyplot as plt
import os
import requests
import numpy as np
from datasets import Dataset
from huggingface_hub import hf_hub_download
import matplotlib.patches as mpatches
import matplotlib as mpl
from constants import asr_systems_colors_mapping
from matplotlib.lines import Line2D
def download_tsv_from_google_sheet(sheet_url):
# Modify the Google Sheet URL to export it as TSV
tsv_url = sheet_url.replace('/edit#gid=', '/export?format=tsv&gid=')
# Send a GET request to download the TSV file
response = requests.get(tsv_url)
response.encoding = 'utf-8'
# Check if the request was successful
if response.status_code == 200:
# Read the TSV content into a pandas DataFrame
from io import StringIO
tsv_content = StringIO(response.text)
df = pd.read_csv(tsv_content, sep='\t', encoding='utf-8')
return df
else:
print("Failed to download the TSV file.")
return None
def generate_path_to_latest_tsv(dataset_name, split, type_of_result):
fn = os.path.join("./data", dataset_name, split, "eval_results-{}-latest.tsv".format(type_of_result))
#print(fn)
return(fn)
@st.cache_data
def read_latest_results(dataset_name, split, codename_to_shortname_mapping):
# Set your Hugging Face API token as an environment variable
# Define the path to your dataset directory
repo_id = os.getenv('HF_SECRET_REPO_ID')
#"michaljunczyk/bigos-eval-results-secret"
dataset = dataset_name
dataset_path = os.path.join("leaderboard_input", dataset, split)
print(dataset_path)
fn_results_per_dataset = 'eval_results-per_dataset-latest.tsv'
fn_results_per_sample = 'eval_results-per_sample-latest.tsv'
fp_results_per_dataset_repo = os.path.join(dataset_path, fn_results_per_dataset)
print(fp_results_per_dataset_repo)
fp_results_per_sample_repo = os.path.join(dataset_path, fn_results_per_sample)
# Download the file from the Hugging Face Hub
local_fp_per_dataset = hf_hub_download(repo_id=repo_id, repo_type='dataset', filename=fp_results_per_dataset_repo, use_auth_token=os.getenv('HF_TOKEN'))
local_fp_per_sample = hf_hub_download(repo_id=repo_id, repo_type='dataset', filename=fp_results_per_sample_repo, use_auth_token=os.getenv('HF_TOKEN'))
# Read the TSV file into a pandas DataFrame
df_per_dataset = pd.read_csv(local_fp_per_dataset, delimiter='\t')
df_per_sample = pd.read_csv(local_fp_per_sample, delimiter='\t')
# Print the DataFrame
print(df_per_dataset)
print(df_per_sample)
#replace column system with Shortname
if (codename_to_shortname_mapping):
df_per_sample['system'] = df_per_sample['system'].replace(codename_to_shortname_mapping)
df_per_dataset['system'] = df_per_dataset['system'].replace(codename_to_shortname_mapping)
return df_per_sample, df_per_dataset
@st.cache_data
def retrieve_asr_systems_meta_from_the_catalog(asr_systems_list):
#print("Retrieving ASR systems metadata for systems: ", asr_systems_list)
#print("Number of systems: ", len(asr_systems_list))
#print("Reading ASR systems catalog")
asr_systems_cat_url = "https://docs.google.com/spreadsheets/d/1fVsE98Ulmt-EIEe4wx8sUdo7RLigDdAVjQxNpAJIrH8/edit#gid=681521237"
#print("Reading the catalog from: ", asr_systems_cat_url)
catalog = download_tsv_from_google_sheet(asr_systems_cat_url)
#print("ASR systems catalog read")
#print("Catalog contains information about {} ASR systems".format(len(catalog)))
##print("Catalog columns: ", catalog.columns)
##print("ASR systems available in the catalog: ", catalog["Codename"] )
#print("Filter only the systems we are interested in")
catalog = catalog[(catalog["Codename"].isin(asr_systems_list)) | (catalog["Shortname"].isin(asr_systems_list))]
return catalog
def basic_stats_per_dimension(df_input, metric, dimension):
#Median value
df_median = df_input.groupby(dimension)[metric].median().sort_values().round(2)
#Average value
df_avg = df_input.groupby(dimension)[metric].mean().sort_values().round(2)
#Standard deviation
df_std = df_input.groupby(dimension)[metric].std().sort_values().round(2)
# Min
df_min = df_input.groupby(dimension)[metric].min().sort_values().round(2)
# Max
df_max = df_input.groupby(dimension)[metric].max().sort_values().round(2)
# concatanate all WER statistics
df_stats = pd.concat([df_median, df_avg, df_std, df_min, df_max], axis=1)
df_stats.columns = ["med_{}".format(metric), "avg_{}".format(metric), "std_{}".format(metric), "min_{}".format(metric), "max_{}".format(metric)]
# sort by median values
df_stats = df_stats.sort_values(by="med_{}".format(metric))
return df_stats
def ser_from_per_sample_results(df_per_sample, dimension):
# group by dimension e.g dataset or sample and calculate fraction of samples with WER equal to 0
df_ser = df_per_sample.groupby(dimension)["WER"].apply(lambda x: (x != 0).mean()*100).sort_values().round(2)
# change column names
df_ser.name = "SER"
return df_ser
def get_total_audio_duration(df_per_sample):
# filter the df_per_sample dataframe to leave only unique audio recordings
df_per_sample_unique_audio = df_per_sample.drop_duplicates(subset='id')
# calculate the total size of the dataset in hours based on the list of unique audio recordings
total_duration_hours = df_per_sample_unique_audio['audio_duration'].sum() / 3600
#print(f"Total duration of the dataset: {total_duration_hours:.2f} hours")
return total_duration_hours
def extend_meta_per_sample_words_chars(df_per_sample):
# extend the results with the number of words in the reference and hypothesis
df_per_sample['ref_words'] = df_per_sample['ref'].apply(lambda x: len(x.split()))
df_per_sample['hyp_words'] = df_per_sample['hyp'].apply(lambda x: len(x.split()))
# extend the df_per_sample with the number of words per seconds (based on duration column) for reference and hypothesis
df_per_sample['ref_wps'] = df_per_sample['ref_words'] / df_per_sample['audio_duration'].round(2)
df_per_sample['hyp_wps'] = df_per_sample['hyp_words'] / df_per_sample['audio_duration'].round(2)
# extend the df_per_sample with the number of characters per seconds (based on duration column) for reference and hypothesis
df_per_sample['ref_cps'] = df_per_sample['ref'].apply(lambda x: len(x)) / df_per_sample['audio_duration'].round(2)
df_per_sample['hyp_cps'] = df_per_sample['hyp'].apply(lambda x: len(x)) / df_per_sample['audio_duration'].round(2)
# extend the df_per_sample with the number of characters per words for reference and hypothesis
df_per_sample['ref_cpw'] = df_per_sample['ref'].apply(lambda x: len(x)) / df_per_sample['ref_words'].round(2)
df_per_sample['hyp_cpw'] = df_per_sample['hyp'].apply(lambda x: len(x)) / df_per_sample['hyp_words'].round(2)
# extend metadata with number of words and characters
return df_per_sample
def filter_top_outliers(df_input, metric, max_threshold):
# filter out outliers exceeding max_threshold
df_filtered = df_input[df_input[metric] < max_threshold]
return df_filtered
def filter_bottom_outliers(df_input, metric, min_threshold):
# filter out outliers below min_threshold
df_filtered = df_input[df_input[metric] > min_threshold]
return df_filtered
def box_plot_per_dimension(df_input, metric, dimension, title, xlabel, ylabel):
# Box plot for WER per dataset
fig, ax = plt.subplots(figsize=(12, 8))
# generate box plot without outliers
sns.boxplot(x=dimension, y=metric, data=df_input, order=df_input.groupby(dimension)[metric].median().sort_values().index, showfliers=False)
plt.title(title)
plt.xlabel(xlabel)
plt.ylabel(ylabel)
plt.xticks(rotation=90)
#return figure
return plt
def box_plot_per_dimension_subsets(df_input, metric, dimension, title, xlabel, ylabel, category_column, y_limit=100):
"""
Plots a box plot with individual data points colored and marked by a specified category.
Parameters:
- df_input (pd.DataFrame): Input DataFrame containing data to plot.
- metric (str): Column name for the metric to plot on the y-axis.
- dimension (str): Column name for the dimension (x-axis categories).
- title (str): Title of the plot.
- xlabel (str): Label for the x-axis.
- ylabel (str): Label for the y-axis.
- category_column (str): Column name to use for differentiating data points by color and marker.
- y_limit (float, optional): Maximum value for the y-axis to limit extreme outliers.
Returns:
- fig: The matplotlib figure object.
"""
# Set up the figure and axis with a larger size for readability
fig, ax = plt.subplots(figsize=(14, 8))
# Create a sorted order for the dimension based on the median values of the metric
order = df_input.groupby(dimension)[metric].median().sort_values().index
# Generate box plot without showing extreme outliers
boxplot = sns.boxplot(
x=dimension, y=metric, data=df_input,
order=order, showfliers=False, width=0.6, ax=ax,
color="white"
)
# Make the box plots transparent by adjusting the facecolor of each box
for patch in boxplot.artists:
patch.set_facecolor("white")
patch.set_alpha(0.2) # Set transparency
# Define category-specific colors and marker styles
categories = df_input[category_column].unique()
markers = ['o', 's', '^', 'D', 'X', 'P', '*'] # Different marker styles
colors = sns.color_palette("Set2", len(categories)) # Use a color palette with distinct colors
category_style_map = {category: {'color': colors[i % len(colors)], 'marker': markers[i % len(markers)]}
for i, category in enumerate(categories)}
# Overlay individual data points with category-specific colors and markers
for category, style in category_style_map.items():
# Filter data for each category
category_data = df_input[(df_input[category_column] == category) & (df_input[metric] <= y_limit)]
sns.stripplot(
x=dimension, y=metric, data=category_data,
order=order, color=style['color'], marker=style['marker'],
size=5, jitter=True, alpha=1, ax=ax
)
# Set title and axis labels
ax.set_title(title)
ax.set_xlabel(xlabel)
ax.set_ylabel(ylabel)
ax.set_xticklabels(ax.get_xticklabels(), rotation=45, ha='right')
# Add gridlines for easier comparison
plt.grid(axis='y', linestyle='--', alpha=0.5)
# Set y-axis limit to improve readability
# Calculate the y-axis maximum as the next multiple of 5 above the data’s max value
# Make sure the max value does not contain any extreme outliers. Threhold at 98th percentile
max_value = df_input[metric].quantile(0.99)
y_max = (int(max_value / 5) + 1) * 5
# Set y-axis ticks with evenly spaced intervals of 5
ax.set_yticks(range(0, y_max + 1, 5))
ax.set_ylim(0, y_max)
# Create a custom legend with unique entries for each category
legend_handles = [
Line2D([0], [0], marker=style['marker'], color='w', markerfacecolor=style['color'], markersize=8, label=category)
for category, style in category_style_map.items()
]
ax.legend(handles=legend_handles, title=category_column, bbox_to_anchor=(1.05, 1), loc='upper left')
# Return the updated figure
return fig
def box_plot_per_dimension_with_colors(df_input, metric, dimension, title, xlabel, ylabel, system_col, type_col):
# Create a figure and axis object
fig, ax = plt.subplots(figsize=(12, 8))
# Define the order of categories based on the median of the metric
order = df_input.groupby(dimension)[metric].median().sort_values().index.tolist()
# Create custom color mapping for systems
unique_systems = df_input[system_col].unique()
# Define your custom colors here
system_color_mapping = asr_systems_colors_mapping
# For systems not specified, assign colors from a palette
remaining_systems = [s for s in unique_systems if s not in system_color_mapping]
palette = sns.color_palette("tab10", len(remaining_systems))
system_color_mapping.update(dict(zip(remaining_systems, palette)))
# Create hatching patterns for types
unique_types = df_input[type_col].unique()
type_hatch_mapping = {
'free': '', # No hatching
'commercial': '///', # Diagonal hatching
# Add more patterns if needed
}
# For types not specified, assign default hatches
default_hatches = ['', '///', '\\\\', 'xx', '++', '--', '...']
for idx, t in enumerate(unique_types):
if t not in type_hatch_mapping:
type_hatch_mapping[t] = default_hatches[idx % len(default_hatches)]
# Map colors and hatches to each dimension based on system and type
dimension_system_mapping = df_input.drop_duplicates(subset=dimension).set_index(dimension)[system_col].reindex(order)
colors = dimension_system_mapping.map(system_color_mapping).tolist()
dimension_type_mapping = df_input.drop_duplicates(subset=dimension).set_index(dimension)[type_col].reindex(order)
hatches = dimension_type_mapping.map(type_hatch_mapping).tolist()
# Generate box plot without specifying hue
sns.boxplot(
x=dimension,
y=metric,
data=df_input,
order=order,
ax=ax,
showfliers=False,
linewidth=1.5,
boxprops=dict(facecolor='white') # Set initial facecolor to white
)
# Access the box artists
box_patches = [patch for patch in ax.artists if isinstance(patch, mpatches.PathPatch)]
# Alternatively, you can use ax.patches if ax.artists doesn't work
if not box_patches:
box_patches = [patch for patch in ax.patches if isinstance(patch, mpatches.PathPatch)]
# Color the boxes and apply hatching patterns
for patch, color, hatch in zip(box_patches, colors, hatches):
patch.set_facecolor(color)
patch.set_edgecolor('black')
patch.set_linewidth(1.5)
patch.set_hatch(hatch)
# Create custom legend for systems (colors)
system_handles = []
for system in unique_systems:
color = system_color_mapping[system]
handle = mpatches.Patch(facecolor=color, edgecolor='black', label=system)
system_handles.append(handle)
# Create custom legend for types (hatching patterns)
type_handles = []
for typ in unique_types:
hatch = type_hatch_mapping[typ]
handle = mpatches.Patch(facecolor='white', edgecolor='black', hatch=hatch, label=typ)
type_handles.append(handle)
# Add legends to the plot
legend1 = ax.legend(handles=system_handles, title='System', bbox_to_anchor=(0.01, 1), loc='upper left')
legend2 = ax.legend(handles=type_handles, title='Type', bbox_to_anchor=(0.01, 0.6), loc='upper left')
ax.add_artist(legend1) # Add the first legend back to the plot
ax.set_title(title)
ax.set_xlabel(xlabel)
ax.set_ylabel(ylabel)
# improve readibility of the x-axis labels
# decrease the font size of x-axis labels
ax.tick_params(axis='x', labelsize=8)
# shift left to align the x-axis labels with the boxes
ax.set_xticklabels(ax.get_xticklabels(), ha='right')
# rotate them by 90 degrees
ax.set_xticklabels(ax.get_xticklabels(), rotation=55)
# add more granularity to the y-axis. Make sure the y-axis contains 20 ticks
ax.yaxis.set_major_locator(plt.MaxNLocator(20))
plt.tight_layout()
# Return the figure object
return fig
def check_impact_of_normalization(data_in, ref_type='orig'):
# Filter the data to include only the specific reference type
data_ref_type = data_in[data_in['ref_type'] == ref_type]
data = data_ref_type.drop(columns=['system','subset', 'ref_type'])
# Calculate the average impact of each normalization type on the metrics
average_impact = data.groupby('norm_type').mean()
baseline_metrics = average_impact.loc['none']
# Calculate the difference in metrics compared to the baseline
difference_metrics = average_impact.subtract(baseline_metrics)
# Removing the baseline row for clarity
difference_metrics = difference_metrics.drop(index='none')
# Rounding the results to 2 decimal places
difference_metrics_rounded = difference_metrics.round(2)
# add column with average impact on error reduction for all metric types
difference_metrics_rounded['Average'] = difference_metrics_rounded.mean(axis=1).round(2)
# Sorting the results based on the average impact on error reduction. The lower the absolute value, the higher the impact
difference_metrics_sorted_abs = difference_metrics_rounded.sort_values(by='Average', key=abs)
# Display the resulting differences
return(difference_metrics_sorted_abs)
def calculate_wer_per_meta_category(df_per_sample, selected_systems, metric, analysis_dimension = 'speaker_gender'):
# filter out from df_per_sample rows where analysis_dimension is null
df_per_sample_dimension = df_per_sample[df_per_sample[analysis_dimension].notnull()]
#print(df_per_sample_dimension)
meta_values = df_per_sample_dimension[analysis_dimension].unique()
if (analysis_dimension == 'speaker_age'):
# sort values in the meta_values list, so the order of the values is consistent, starting from teens, twenties, thirties, fourties, fifties, sixties, seventies, eighties, nineties
# Example usage:
sorted_values = sort_age_categories(meta_values)
#print(sorted_values)
print("meta values sorted:", sorted_values)
meta_values = sorted_values
# calculate number of available systems for specific category
#print(df_per_sample_dimension)
# create table with number of samples in df_per_sample_single_system for each meta category from meta_values
df_per_sample_single_system = df_per_sample_dimension[df_per_sample['system'] == selected_systems[0]]
# select the value with the smallest number of available samples for all systems
min_samples = 0
df_available_samples_per_category_per_system = {}
for system in selected_systems:
df_per_sample_single_system = df_per_sample_dimension[df_per_sample['system'] == system]
df_available_samples_per_category_per_system[system] = df_per_sample_single_system.groupby(analysis_dimension)[metric].count().reset_index()
df_available_samples_per_category_per_system[system] = df_available_samples_per_category_per_system[system].rename(columns={metric: 'available_samples'})
# replace index with values from analysis_dimension
df_available_samples_per_category_per_system[system] = df_available_samples_per_category_per_system[system].set_index(analysis_dimension)
#print(df_available_samples_per_category_per_system[system])
min_samples_system = df_available_samples_per_category_per_system[system]['available_samples'].min()
if (min_samples_system < min_samples) or (min_samples == 0):
min_samples = min_samples_system
#print(min_samples)
# get the subset of the df_per_sample_dimension with results for all systems to analyze
df_per_sample_selected_systems = df_per_sample_dimension[df_per_sample['system'].isin(selected_systems)]
#print(df_per_sample_selected_systems)
# select equal number of samples for each system and analysis_dimension equal to the number of samples for the dimension with the smallest number of samples (min_samples)
df_per_sample_selected_systems = df_per_sample_selected_systems.groupby(['system',analysis_dimension]).apply(lambda x: x.sample(min_samples)).reset_index(drop=True)
#print(df_per_sample_selected_systems)
df_per_sample_metric_dimension = df_per_sample_selected_systems.groupby(['system', analysis_dimension])[metric].mean().round(2).reset_index()
df_per_sample_metric_dimension_pivot = df_per_sample_metric_dimension.pivot(index=analysis_dimension, columns='system', values=metric)
df_per_sample_metric_dimension_pivot = df_per_sample_metric_dimension_pivot.round(2)
# add row with the difference between the male and female metric values for values. Add "Difference" row at the end of the dataframe to the index
# calculate the difference between the smallest and largest metric values
# if there are only two values in the analysis_dimension, calculate the difference between them
if len(meta_values) == 2:
gap_metrics = ['Difference']
df_per_sample_metric_dimension_pivot.loc[gap_metrics[0]] = df_per_sample_metric_dimension_pivot.loc[meta_values[0]] - df_per_sample_metric_dimension_pivot.loc[meta_values[1]]
# if there are more than two values in the analysis_dimension, calculate the difference between the smallest and the largest value
elif len(meta_values) > 2:
gap_metrics = ['Std Dev', 'MAD', 'Range']
metrics = pd.DataFrame([])
df = df_per_sample_metric_dimension_pivot
print(df)
# calculate the standard deviation of the metric values
metrics[gap_metrics[0]] = df.std()
# calculate the mean absolute deviation of the metric values
metrics[gap_metrics[1]] = df.apply(lambda x: np.mean(np.abs(x - np.mean(x))), axis=0)
# calculate the difference between the smallest and largest metric values
metrics[gap_metrics[2]] = df.max() - df.min()
metrics_t = metrics.round(2).transpose()
print(metrics_t)
#concatante the metrics dataframe to the df_per_sample_metric_dimension_pivot
df_per_sample_metric_dimension_pivot = pd.concat([df_per_sample_metric_dimension_pivot, metrics_t], axis=0)
print(df_per_sample_metric_dimension_pivot)
# transpose the dataframe to have systems as rows
# sort by the average difference from the smallest to the largest value
df_per_sample_metric_dimension_pivot = df_per_sample_metric_dimension_pivot.transpose()
df_per_sample_metric_dimension_pivot = df_per_sample_metric_dimension_pivot.sort_values(by=gap_metrics[0], axis=0)
# add average, median and standard deviation as the last 3 rows to the dataframe
# calculate average, median, and standard deviation of the difference between the smallest and largest metric values
avg_difference = df_per_sample_metric_dimension_pivot.mean().round(2)
median_difference = df_per_sample_metric_dimension_pivot.median().round(2)
std_difference = df_per_sample_metric_dimension_pivot.std().round(2)
# add average, median, and standard deviation as the last 3 rows to the dataframe
df_per_sample_metric_dimension_pivot.loc['median'] = median_difference
df_per_sample_metric_dimension_pivot.loc['average'] = avg_difference
df_per_sample_metric_dimension_pivot.loc['std'] = std_difference
analyzed_samples_per_category = min_samples
# round all values to 2 decimal places
df_per_sample_metric_dimension_pivot = df_per_sample_metric_dimension_pivot.round(2)
# keep the order of columns as in the meta_values list
columns = list(meta_values) + gap_metrics
print(columns)
df_per_sample_metric_dimension_pivot = df_per_sample_metric_dimension_pivot[columns]
return df_per_sample_metric_dimension_pivot, df_available_samples_per_category_per_system, analyzed_samples_per_category
def sort_age_categories(meta_values):
order = ["teens", "twenties", "thirties", "fourties", "fifties", "sixties", "seventies", "eighties", "nineties"]
order_dict = {age: index for index, age in enumerate(order)}
sorted_values = sorted(meta_values, key=lambda x: order_dict.get(x, float('inf')))
return sorted_values
def calculate_wer_per_audio_feature(df_per_sample, selected_systems, audio_feature_to_analyze, metric, no_of_buckets):
# filter out results for selected systems
print(df_per_sample)
feature_values_uniq = df_per_sample[audio_feature_to_analyze].unique()
df_per_sample_selected_systems = df_per_sample[df_per_sample['system'].isin(selected_systems)]
# create buckets based on speech rate words unique values (min, max,step)
min_feature_value = round(min(feature_values_uniq), 1)
max_feature_value = round(max(feature_values_uniq), 1)
step = max_feature_value / no_of_buckets
audio_feature_buckets = [min_feature_value + i * step for i in range(no_of_buckets)]
# add column with speech_rate_words rounded to nearest bucket value.
# map audio duration to the closest bucket
df_per_sample[audio_feature_to_analyze + '_bucket'] = df_per_sample[audio_feature_to_analyze].apply(
lambda x: min(audio_feature_buckets, key=lambda y: abs(x - y)))
# calculate average WER per audio duration bucket
df_per_sample_wer_feature = df_per_sample_selected_systems.groupby(['system', audio_feature_to_analyze])[metric].mean().reset_index()
# add column with number of samples for specific audio bucket size
df_per_sample_wer_feature['number_of_samples'] = df_per_sample_selected_systems.groupby(['system', audio_feature_to_analyze])[metric].count().values
df_per_sample_wer_feature = df_per_sample_wer_feature.sort_values(by=audio_feature_to_analyze)
# round values in WER column in df_per_sample_wer to 2 decimal places
df_per_sample_wer_feature[metric].round(2)
# transform df_per_sample_wer. Use system values as columns, while audio_duration_buckets as main index
df_per_sample_wer_feature_pivot = df_per_sample_wer_feature.pivot(index=audio_feature_to_analyze, columns='system', values=metric)
df_per_sample_wer_feature_pivot = df_per_sample_wer_feature_pivot.round(2)
df_per_sample_wer_feature_pivot['number_of_samples'] = df_per_sample_wer_feature[
df_per_sample_wer_feature['system'] == selected_systems[0]].groupby(audio_feature_to_analyze)[
'number_of_samples'].sum().values
# put number_of_samples as the first column after index
df_per_sample_wer_feature_pivot = df_per_sample_wer_feature_pivot[
['number_of_samples'] + [col for col in df_per_sample_wer_feature_pivot.columns if col != 'number_of_samples']]
return df_per_sample_wer_feature_pivot, df_per_sample_wer_feature