app.py

import gradio as gr
import pandas as pd
from datasets import load_dataset
from jiwer import wer, cer
import os
from datetime import datetime
import re

from huggingface_hub import login


token = os.environ.get("HG_TOKEN")
login(token)

print("Loading dataset...")
try:
    dataset = load_dataset("sudoping01/bambara-speech-recognition-benchmark", name="default")["eval"]
    references = {row["id"]: row["text"] for row in dataset}
    print(f"Successfully loaded dataset with {len(references)} samples")
except Exception as e:
    print(f"Error loading dataset: {str(e)}")
    references = {}
    print("WARNING: Using empty references dictionary due to dataset loading error")

# Initialize leaderboard file with consistent column names
leaderboard_file = "leaderboard.csv"
if not os.path.exists(leaderboard_file):
    # Create with Model_Name consistently
    pd.DataFrame(columns=["Model_Name", "WER", "CER", "Combined_Score", "timestamp"]).to_csv(leaderboard_file, index=False)
    print("Created new leaderboard file")
    
    # Add example entries for first-time visitors
    example_data = [
        ["Example Model 1", 0.35, 0.20, 0.305, "2023-01-01 00:00:00"],
        ["Example Model 2", 0.40, 0.18, 0.334, "2023-01-02 00:00:00"],
        ["Example Model 3", 0.32, 0.25, 0.299, "2023-01-03 00:00:00"]
    ]
    example_df = pd.DataFrame(
        example_data, 
        columns=["Model_Name", "WER", "CER", "Combined_Score", "timestamp"]
    )
    example_df.to_csv(leaderboard_file, index=False)
    print("Added example data to empty leaderboard for demonstration")
else:
    # Load existing leaderboard
    leaderboard_df = pd.read_csv(leaderboard_file)
    
    # Rename column if needed for consistency
    if "submitter" in leaderboard_df.columns and "Model_Name" not in leaderboard_df.columns:
        leaderboard_df = leaderboard_df.rename(columns={"submitter": "Model_Name"})
        leaderboard_df.to_csv(leaderboard_file, index=False)
        print("Renamed 'submitter' column to 'Model_Name' for consistency")
    
    # Add Combined_Score column if it doesn't exist
    if "Combined_Score" not in leaderboard_df.columns:
        leaderboard_df["Combined_Score"] = leaderboard_df["WER"] * 0.7 + leaderboard_df["CER"] * 0.3 # WER 70% and CER 30%
        leaderboard_df.to_csv(leaderboard_file, index=False)
        print("Added Combined_Score column to existing leaderboard")
    
    print(f"Loaded existing leaderboard with {len(leaderboard_df)} entries")

def normalize_text(text):
    """
    Normalize text for WER/CER calculation:
    - Convert to lowercase
    - Remove punctuation
    - Replace multiple spaces with single space
    - Strip leading/trailing spaces
    """
    if not isinstance(text, str):
        text = str(text)
    
    text = text.lower()
    text = re.sub(r'[^\w\s]', '', text)
    text = re.sub(r'\s+', ' ', text).strip()
    return text

def calculate_metrics(predictions_df):
    """Calculate WER and CER for predictions."""
    results = []
    total_ref_words = 0
    total_ref_chars = 0

    for _, row in predictions_df.iterrows():
        id_val = row["id"]
        if id_val not in references:
            continue
            
        reference = normalize_text(references[id_val])
        hypothesis = normalize_text(row["text"])
        
        
        if not reference or not hypothesis:
            print(f"Warning: Empty reference or hypothesis for ID {id_val}")
            continue
            
        reference_words = reference.split()
        hypothesis_words = hypothesis.split()
        reference_chars = list(reference)
        
        try:

            sample_wer = wer(reference, hypothesis)
            sample_cer = cer(reference, hypothesis)
            
            sample_wer = min(sample_wer, 2.0)  
            sample_cer = min(sample_cer, 2.0)  
            
         
            total_ref_words += len(reference_words)
            total_ref_chars += len(reference_chars)
            
                
            results.append({
                "id": id_val,
                "reference": reference,
                "hypothesis": hypothesis,
                "ref_word_count": len(reference_words),
                "ref_char_count": len(reference_chars),
                "wer": sample_wer,
                "cer": sample_cer
            })
        except Exception as e:
            print(f"Error calculating metrics for ID {id_val}: {str(e)}")
    
    if not results:
        raise ValueError("No valid samples for WER/CER calculation")
        
    # Calculate standard average metrics
    avg_wer = sum(item["wer"] for item in results) / len(results)
    avg_cer = sum(item["cer"] for item in results) / len(results)
    
    # Calculate weighted average metrics based on reference length
    weighted_wer = sum(item["wer"] * item["ref_word_count"] for item in results) / total_ref_words
    weighted_cer = sum(item["cer"] * item["ref_char_count"] for item in results) / total_ref_chars
    
    return avg_wer, avg_cer, weighted_wer, weighted_cer, results

def update_ranking(method):
    """Update leaderboard ranking based on selected method"""
    try:
        current_lb = pd.read_csv(leaderboard_file)
        
        if "Combined_Score" not in current_lb.columns:
            current_lb["Combined_Score"] = current_lb["WER"] * 0.7 + current_lb["CER"] * 0.3 # 70% for WER
        
        if method == "WER Only":
            return current_lb.sort_values("WER")
        elif method == "CER Only":
            return current_lb.sort_values("CER")
        else:  # Combined Score
            return current_lb.sort_values("Combined_Score")
    except Exception as e:
        print(f"Error updating ranking: {str(e)}")
        # Return empty dataframe if something goes wrong
        return pd.DataFrame(columns=["Model_Name", "WER", "CER", "Combined_Score", "timestamp"])

def process_submission(model_name, csv_file):
    try:
        df = pd.read_csv(csv_file)
        print(f"Processing submission from {model_name} with {len(df)} rows")
        
        if len(df) == 0:
            return "Error: Uploaded CSV is empty.", None
            
        if set(df.columns) != {"id", "text"}:
            return f"Error: CSV must contain exactly 'id' and 'text' columns. Found: {', '.join(df.columns)}", None
            
        if df["id"].duplicated().any():
            dup_ids = df[df["id"].duplicated()]["id"].unique()
            return f"Error: Duplicate IDs found: {', '.join(map(str, dup_ids[:5]))}", None

        # Check if IDs match the reference dataset
        missing_ids = set(references.keys()) - set(df["id"])
        extra_ids = set(df["id"]) - set(references.keys())
        
        if missing_ids:
            return f"Error: Missing {len(missing_ids)} IDs in submission. First few missing: {', '.join(map(str, list(missing_ids)[:5]))}", None
            
        if extra_ids:
            return f"Error: Found {len(extra_ids)} extra IDs not in reference dataset. First few extra: {', '.join(map(str, list(extra_ids)[:5]))}", None
        
        # Calculate WER and CER
        try:
            avg_wer, avg_cer, weighted_wer, weighted_cer, detailed_results = calculate_metrics(df)
            
            # suspiciously low values
            if avg_wer < 0.001:
                return "Error: WER calculation yielded suspicious results (near-zero). Please check your submission CSV.", None
                
        except Exception as e:
            return f"Error calculating metrics: {str(e)}", None
        
        leaderboard = pd.read_csv(leaderboard_file)
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        #  (70% WER, 30% CER)
        combined_score = avg_wer * 0.7 + avg_cer * 0.3
        
        new_entry = pd.DataFrame(
            [[model_name, avg_wer, avg_cer, combined_score, timestamp]],
            columns=["Model_Name", "WER", "CER", "Combined_Score", "timestamp"]
        )
        
        updated_leaderboard = pd.concat([leaderboard, new_entry]).sort_values("Combined_Score")
        updated_leaderboard.to_csv(leaderboard_file, index=False)
        
        return f"Submission processed successfully! WER: {avg_wer:.4f}, CER: {avg_cer:.4f}, Combined Score: {combined_score:.4f}", updated_leaderboard
        
    except Exception as e:
        return f"Error processing submission: {str(e)}", None

# Make sure we have at least some data for first-time visitors
if os.path.exists(leaderboard_file):
    leaderboard_df = pd.read_csv(leaderboard_file)
    if len(leaderboard_df) == 0:
        # Add example entries if leaderboard is empty
        example_data = [
            ["Example Model 1", 0.35, 0.20, 0.305, "2023-01-01 00:00:00"],
            ["Example Model 2", 0.40, 0.18, 0.334, "2023-01-02 00:00:00"],
            ["Example Model 3", 0.32, 0.25, 0.299, "2023-01-03 00:00:00"]
        ]
        example_df = pd.DataFrame(
            example_data, 
            columns=["Model_Name", "WER", "CER", "Combined_Score", "timestamp"]
        )
        example_df.to_csv(leaderboard_file, index=False)
        print("Added example data to empty leaderboard for demonstration")

with gr.Blocks(title="Bambara ASR Leaderboard") as demo:
    gr.Markdown(
        """
        # Bambara ASR Leaderboard
        
        This leaderboard ranks and evaluates speech recognition models for the Bambara language.
        Models are ranked based on a combined score of WER and CER metrics.
        """
    )
    
    with gr.Tabs() as tabs:
        with gr.TabItem("🏅 Current Rankings"):
            try:
                # Load and make sure we have current leaderboard data
                current_leaderboard = pd.read_csv(leaderboard_file)
                
                if "Combined_Score" not in current_leaderboard.columns:
                    current_leaderboard["Combined_Score"] = current_leaderboard["WER"] * 0.7 + current_leaderboard["CER"] * 0.3
                
                # Sort by combined score
                current_leaderboard = current_leaderboard.sort_values("Combined_Score")
            except Exception as e:
                print(f"Error loading leaderboard: {str(e)}")
                # Create empty dataframe if we can't load the file
                current_leaderboard = pd.DataFrame(columns=["Model_Name", "WER", "CER", "Combined_Score", "timestamp"])
            
            gr.Markdown("### Current ASR Model Rankings")
            
            ranking_method = gr.Radio(
                ["Combined Score (WER 70%, CER 30%)", "WER Only", "CER Only"], 
                label="Ranking Method",
                value="Combined Score (WER 70%, CER 30%)"
            )
            
            leaderboard_view = gr.DataFrame(
                value=current_leaderboard,
                interactive=False,
                label="Models are ranked by selected metric - lower is better"
            )
            
            ranking_method.change(
                fn=update_ranking,
                inputs=[ranking_method],
                outputs=[leaderboard_view]
            )
            
            gr.Markdown(
                """
                ## Metrics Explanation
                - **WER**: Word Error Rate (lower is better) - measures word-level accuracy
                - **CER**: Character Error Rate (lower is better) - measures character-level accuracy
                - **Combined Score**: Weighted average of WER (70%) and CER (30%) - provides a balanced evaluation
                """
            )
        
        with gr.TabItem("📊 Submit New Results"):
            gr.Markdown(
                """
                ### Submit a new model for evaluation
                
                Upload a CSV file with 'id' and 'text' columns to evaluate your ASR predictions.
                The 'id's must match those in the reference dataset.
                """
            )
            
            with gr.Row():
                model_name_input = gr.Textbox(label="Model Name", placeholder="e.g., MALIBA-AI/asr")
                csv_upload = gr.File(label="Upload CSV File", file_types=[".csv"])
                
            submit_btn = gr.Button("Submit")
            output_msg = gr.Textbox(label="Status", interactive=False)
            leaderboard_display = gr.DataFrame(
                label="Updated Leaderboard",
                value=current_leaderboard,
                interactive=False
            )
            
            submit_btn.click(
                fn=process_submission,
                inputs=[model_name_input, csv_upload],
                outputs=[output_msg, leaderboard_display]
            )

print("Starting Bambara ASR Leaderboard app...")

if __name__ == "__main__":
    demo.launch(share=True)