Rename src/leaderboard.py to src/evaluation.py

src/evaluation.py

@@ -0,0 +1,403 @@
+# src/evaluation.py
+import pandas as pd
+import numpy as np
+from sacrebleu.metrics import BLEU, CHRF
+from rouge_score import rouge_scorer
+import Levenshtein
+from collections import defaultdict
+from transformers.models.whisper.english_normalizer import BasicTextNormalizer
+from typing import Dict, List, Tuple
+from config import ALL_UG40_LANGUAGES, GOOGLE_SUPPORTED_LANGUAGES, METRICS_CONFIG
+from src.utils import get_all_language_pairs, get_google_comparable_pairs
+
+def calculate_sentence_metrics(reference: str, prediction: str) -> Dict[str, float]:
+    """Calculate all metrics for a single sentence pair."""
+    
+    # Handle empty predictions
+    if not prediction or not isinstance(prediction, str):
+        prediction = ""
+    
+    if not reference or not isinstance(reference, str):
+        reference = ""
+    
+    # Normalize texts
+    normalizer = BasicTextNormalizer()
+    pred_norm = normalizer(prediction)
+    ref_norm = normalizer(reference)
+    
+    metrics = {}
+    
+    # BLEU score
+    try:
+        bleu = BLEU(effective_order=True)
+        metrics['bleu'] = bleu.sentence_score(pred_norm, [ref_norm]).score
+    except:
+        metrics['bleu'] = 0.0
+    
+    # ChrF score
+    try:
+        chrf = CHRF()
+        metrics['chrf'] = chrf.sentence_score(pred_norm, [ref_norm]).score / 100.0
+    except:
+        metrics['chrf'] = 0.0
+    
+    # Character Error Rate (CER)
+    try:
+        if len(ref_norm) > 0:
+            metrics['cer'] = Levenshtein.distance(ref_norm, pred_norm) / len(ref_norm)
+        else:
+            metrics['cer'] = 1.0 if len(pred_norm) > 0 else 0.0
+    except:
+        metrics['cer'] = 1.0
+    
+    # Word Error Rate (WER)
+    try:
+        ref_words = ref_norm.split()
+        pred_words = pred_norm.split()
+        if len(ref_words) > 0:
+            metrics['wer'] = Levenshtein.distance(ref_words, pred_words) / len(ref_words)
+        else:
+            metrics['wer'] = 1.0 if len(pred_words) > 0 else 0.0
+    except:
+        metrics['wer'] = 1.0
+    
+    # Length ratio
+    try:
+        if len(ref_norm) > 0:
+            metrics['len_ratio'] = len(pred_norm) / len(ref_norm)
+        else:
+            metrics['len_ratio'] = 1.0 if len(pred_norm) == 0 else float('inf')
+    except:
+        metrics['len_ratio'] = 1.0
+    
+    # ROUGE scores
+    try:
+        scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
+        rouge_scores = scorer.score(ref_norm, pred_norm)
+        
+        metrics['rouge1'] = rouge_scores['rouge1'].fmeasure
+        metrics['rouge2'] = rouge_scores['rouge2'].fmeasure
+        metrics['rougeL'] = rouge_scores['rougeL'].fmeasure
+    except:
+        metrics['rouge1'] = 0.0
+        metrics['rouge2'] = 0.0
+        metrics['rougeL'] = 0.0
+    
+    # Quality score (composite metric)
+    try:
+        quality_components = [
+            metrics['bleu'] / 100.0,  # Normalize BLEU to 0-1
+            metrics['chrf'],
+            1.0 - min(metrics['cer'], 1.0),  # Invert error rates
+            1.0 - min(metrics['wer'], 1.0),
+            metrics['rouge1'],
+            metrics['rougeL']
+        ]
+        metrics['quality_score'] = np.mean(quality_components)
+    except:
+        metrics['quality_score'] = 0.0
+    
+    return metrics
+
+def evaluate_predictions(predictions: pd.DataFrame, test_set: pd.DataFrame) -> Dict:
+    """Evaluate predictions against test set targets."""
+    
+    print("Starting evaluation...")
+    
+    # Merge predictions with test set (which contains targets)
+    merged = test_set.merge(
+        predictions, 
+        on='sample_id', 
+        how='inner',
+        suffixes=('', '_pred')
+    )
+    
+    if len(merged) == 0:
+        return {
+            'error': 'No matching samples found between predictions and test set',
+            'evaluated_samples': 0
+        }
+    
+    print(f"Evaluating {len(merged)} samples...")
+    
+    # Calculate metrics for each sample
+    sample_metrics = []
+    for idx, row in merged.iterrows():
+        metrics = calculate_sentence_metrics(row['target_text'], row['prediction'])
+        metrics['sample_id'] = row['sample_id']
+        metrics['source_language'] = row['source_language']
+        metrics['target_language'] = row['target_language']
+        metrics['google_comparable'] = row.get('google_comparable', False)
+        sample_metrics.append(metrics)
+    
+    sample_df = pd.DataFrame(sample_metrics)
+    
+    # Aggregate by language pairs
+    pair_metrics = {}
+    overall_metrics = defaultdict(list)
+    google_comparable_metrics = defaultdict(list)
+    
+    # Calculate metrics for each language pair
+    for src_lang in ALL_UG40_LANGUAGES:
+        for tgt_lang in ALL_UG40_LANGUAGES:
+            if src_lang != tgt_lang:
+                pair_data = sample_df[
+                    (sample_df['source_language'] == src_lang) & 
+                    (sample_df['target_language'] == tgt_lang)
+                ]
+                
+                if len(pair_data) > 0:
+                    pair_key = f"{src_lang}_to_{tgt_lang}"
+                    pair_metrics[pair_key] = {}
+                    
+                    # Calculate averages for this pair
+                    for metric in METRICS_CONFIG['primary_metrics'] + METRICS_CONFIG['secondary_metrics']:
+                        if metric in pair_data.columns:
+                            avg_value = float(pair_data[metric].mean())
+                            pair_metrics[pair_key][metric] = avg_value
+                            
+                            # Add to overall averages
+                            overall_metrics[metric].append(avg_value)
+                            
+                            # Add to Google comparable if applicable
+                            if (src_lang in GOOGLE_SUPPORTED_LANGUAGES and 
+                                tgt_lang in GOOGLE_SUPPORTED_LANGUAGES):
+                                google_comparable_metrics[metric].append(avg_value)
+                    
+                    pair_metrics[pair_key]['sample_count'] = len(pair_data)
+    
+    # Calculate overall averages
+    averages = {}
+    for metric in overall_metrics:
+        if overall_metrics[metric]:
+            averages[metric] = float(np.mean(overall_metrics[metric]))
+        else:
+            averages[metric] = 0.0
+    
+    # Calculate Google comparable averages
+    google_averages = {}
+    for metric in google_comparable_metrics:
+        if google_comparable_metrics[metric]:
+            google_averages[metric] = float(np.mean(google_comparable_metrics[metric]))
+        else:
+            google_averages[metric] = 0.0
+    
+    # Generate evaluation summary
+    summary = {
+        'total_samples': len(sample_df),
+        'language_pairs_covered': len([k for k in pair_metrics if pair_metrics[k]['sample_count'] > 0]),
+        'google_comparable_pairs': len([k for k in pair_metrics 
+                                      if '_to_' in k and 
+                                      k.split('_to_')[0] in GOOGLE_SUPPORTED_LANGUAGES and
+                                      k.split('_to_')[1] in GOOGLE_SUPPORTED_LANGUAGES]),
+        'primary_metrics': {metric: averages.get(metric, 0.0) 
+                          for metric in METRICS_CONFIG['primary_metrics']},
+        'secondary_metrics': {metric: averages.get(metric, 0.0) 
+                            for metric in METRICS_CONFIG['secondary_metrics']}
+    }
+    
+    return {
+        'sample_metrics': sample_df,
+        'pair_metrics': pair_metrics,
+        'averages': averages,
+        'google_comparable_averages': google_averages,
+        'summary': summary,
+        'evaluated_samples': len(sample_df),
+        'error': None
+    }
+
+def compare_with_baseline(results: Dict, baseline_results: Dict = None) -> Dict:
+    """Compare results with baseline (e.g., Google Translate)."""
+    
+    if baseline_results is None:
+        return {
+            'comparison_available': False,
+            'message': 'No baseline available for comparison'
+        }
+    
+    comparison = {
+        'comparison_available': True,
+        'overall_comparison': {},
+        'pair_comparisons': {},
+        'better_pairs': [],
+        'worse_pairs': []
+    }
+    
+    # Compare overall metrics
+    for metric in METRICS_CONFIG['primary_metrics']:
+        if metric in results['averages'] and metric in baseline_results['averages']:
+            user_score = results['averages'][metric]
+            baseline_score = baseline_results['averages'][metric]
+            
+            # For error metrics (cer, wer), lower is better
+            if metric in ['cer', 'wer']:
+                improvement = baseline_score - user_score  # Positive = improvement
+            else:
+                improvement = user_score - baseline_score  # Positive = improvement
+            
+            comparison['overall_comparison'][metric] = {
+                'user_score': user_score,
+                'baseline_score': baseline_score,
+                'improvement': improvement,
+                'improvement_percent': (improvement / max(baseline_score, 0.001)) * 100
+            }
+    
+    # Compare by language pairs (only Google comparable ones)
+    google_pairs = [k for k in results['pair_metrics'] 
+                   if '_to_' in k and 
+                   k.split('_to_')[0] in GOOGLE_SUPPORTED_LANGUAGES and
+                   k.split('_to_')[1] in GOOGLE_SUPPORTED_LANGUAGES]
+    
+    for pair in google_pairs:
+        if pair in baseline_results['pair_metrics']:
+            pair_comparison = {}
+            
+            for metric in METRICS_CONFIG['primary_metrics']:
+                if (metric in results['pair_metrics'][pair] and 
+                    metric in baseline_results['pair_metrics'][pair]):
+                    
+                    user_score = results['pair_metrics'][pair][metric]
+                    baseline_score = baseline_results['pair_metrics'][pair][metric]
+                    
+                    if metric in ['cer', 'wer']:
+                        improvement = baseline_score - user_score
+                    else:
+                        improvement = user_score - baseline_score
+                    
+                    pair_comparison[metric] = {
+                        'user_score': user_score,
+                        'baseline_score': baseline_score,
+                        'improvement': improvement
+                    }
+            
+            comparison['pair_comparisons'][pair] = pair_comparison
+            
+            # Determine if this pair is better or worse overall
+            quality_improvement = pair_comparison.get('quality_score', {}).get('improvement', 0)
+            if quality_improvement > 0.01:  # Threshold for significance
+                comparison['better_pairs'].append(pair)
+            elif quality_improvement < -0.01:
+                comparison['worse_pairs'].append(pair)
+    
+    return comparison
+
+def generate_evaluation_report(results: Dict, model_name: str = "", comparison: Dict = None) -> str:
+    """Generate human-readable evaluation report."""
+    
+    if results.get('error'):
+        return f"❌ **Evaluation Error**: {results['error']}"
+    
+    report = []
+    
+    # Header
+    report.append(f"# Evaluation Report: {model_name or 'Submission'}")
+    report.append(f"Generated: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    report.append("")
+    
+    # Summary
+    summary = results['summary']
+    report.append("## 📊 Summary")
+    report.append(f"- **Total Samples Evaluated**: {summary['total_samples']:,}")
+    report.append(f"- **Language Pairs Covered**: {summary['language_pairs_covered']}")
+    report.append(f"- **Google Comparable Pairs**: {summary['google_comparable_pairs']}")
+    report.append("")
+    
+    # Primary metrics
+    report.append("## 🎯 Primary Metrics")
+    for metric, value in summary['primary_metrics'].items():
+        formatted_value = f"{value:.4f}" if metric != 'bleu' else f"{value:.2f}"
+        report.append(f"- **{metric.upper()}**: {formatted_value}")
+    
+    # Quality ranking (if comparison available)
+    if comparison and comparison.get('comparison_available'):
+        quality_comp = comparison['overall_comparison'].get('quality_score', {})
+        if quality_comp:
+            improvement = quality_comp.get('improvement', 0)
+            if improvement > 0.01:
+                report.append(f"  - 🟢 **{improvement:.3f}** better than baseline")
+            elif improvement < -0.01:
+                report.append(f"  - 🔴 **{abs(improvement):.3f}** worse than baseline")
+            else:
+                report.append(f"  - 🟡 Similar to baseline")
+    
+    report.append("")
+    
+    # Secondary metrics
+    report.append("## 📈 Secondary Metrics")
+    for metric, value in summary['secondary_metrics'].items():
+        formatted_value = f"{value:.4f}"
+        report.append(f"- **{metric.upper()}**: {formatted_value}")
+    report.append("")
+    
+    # Language pair performance (top and bottom 5)
+    pair_metrics = results['pair_metrics']
+    if pair_metrics:
+        # Sort pairs by quality score
+        sorted_pairs = sorted(
+            [(k, v.get('quality_score', 0)) for k, v in pair_metrics.items() if v.get('sample_count', 0) > 0],
+            key=lambda x: x[1], 
+            reverse=True
+        )
+        
+        report.append("## 🏆 Best Performing Language Pairs")
+        for pair, score in sorted_pairs[:5]:
+            src, tgt = pair.replace('_to_', ' → ').split(' → ')
+            report.append(f"- **{src} → {tgt}**: {score:.3f}")
+        
+        if len(sorted_pairs) > 5:
+            report.append("")
+            report.append("## 📉 Challenging Language Pairs")
+            for pair, score in sorted_pairs[-3:]:
+                src, tgt = pair.replace('_to_', ' → ').split(' → ')
+                report.append(f"- **{src} → {tgt}**: {score:.3f}")
+    
+    # Comparison with baseline
+    if comparison and comparison.get('comparison_available'):
+        report.append("")
+        report.append("## 🔍 Comparison with Baseline")
+        
+        better_count = len(comparison.get('better_pairs', []))
+        worse_count = len(comparison.get('worse_pairs', []))
+        total_comparable = better_count + worse_count + (comparison.get('google_comparable_pairs', 0) - better_count - worse_count)
+        
+        if total_comparable > 0:
+            report.append(f"- **Better than baseline**: {better_count}/{total_comparable} pairs")
+            report.append(f"- **Worse than baseline**: {worse_count}/{total_comparable} pairs")
+            
+            if comparison['better_pairs']:
+                report.append("  - Strong pairs: " + ", ".join(comparison['better_pairs'][:3]))
+            
+            if comparison['worse_pairs']:
+                report.append("  - Weak pairs: " + ", ".join(comparison['worse_pairs'][:3]))
+    
+    return "\n".join(report)
+
+def create_sample_analysis(results: Dict, n_samples: int = 10) -> pd.DataFrame:
+    """Create sample analysis showing best and worst translations."""
+    
+    if 'sample_metrics' not in results:
+        return pd.DataFrame()
+    
+    sample_df = results['sample_metrics']
+    
+    # Get best and worst samples by quality score
+    best_samples = sample_df.nlargest(n_samples // 2, 'quality_score')
+    worst_samples = sample_df.nsmallest(n_samples // 2, 'quality_score')
+    
+    analysis_samples = pd.concat([best_samples, worst_samples])
+    
+    # Add category
+    analysis_samples['category'] = ['Best'] * len(best_samples) + ['Worst'] * len(worst_samples)
+    
+    return analysis_samples[['sample_id', 'source_language', 'target_language', 
+                           'quality_score', 'bleu', 'chrf', 'category']]
+
+def get_google_translate_baseline() -> Dict:
+    """Get Google Translate baseline results (if available)."""
+    
+    try:
+        # This would load pre-computed Google Translate results
+        # For now, return empty dict - implement when Google Translate baseline is available
+        return {}
+    except:
+        return {}

src/leaderboard.py

@@ -1,183 +0,0 @@
-# src/leaderboard.py
-import pandas as pd
-from datasets import Dataset, load_dataset
-from huggingface_hub import hf_hub_download, upload_file
-import json
-import datetime
-from typing import Dict, List, Optional
-import os
-from config import LEADERBOARD_DATASET, HF_TOKEN
-from src.utils import format_model_name, create_submission_id
-
-def initialize_leaderboard() -> Dataset:
-    """Initialize empty leaderboard dataset."""
-    empty_data = {
-        'submission_id': [],
-        'model_path': [],
-        'model_display_name': [],
-        'author': [],
-        'submission_date': [],
-        'bleu': [],
-        'chrf': [],
-        'quality_score': [],
-        'cer': [],
-        'wer': [],
-        'rouge1': [],
-        'rouge2': [],
-        'rougeL': [],
-        'len_ratio': [],
-        'detailed_metrics': [],
-        'evaluation_samples': [],
-        'model_type': []
-    }
-    return Dataset.from_dict(empty_data)
-
-def load_leaderboard() -> pd.DataFrame:
-    """Load current leaderboard from HuggingFace dataset."""
-    try:
-        dataset = load_dataset(LEADERBOARD_DATASET, split='train')
-        df = dataset.to_pandas()
-        
-        # Ensure all required columns exist
-        required_columns = [
-            'submission_id', 'model_path', 'model_display_name', 'author',
-            'submission_date', 'bleu', 'chrf', 'quality_score', 'cer', 'wer',
-            'rouge1', 'rouge2', 'rougeL', 'len_ratio', 'detailed_metrics',
-            'evaluation_samples', 'model_type'
-        ]
-        
-        for col in required_columns:
-            if col not in df.columns:
-                if col in ['bleu', 'chrf', 'quality_score', 'cer', 'wer', 'rouge1', 'rouge2', 'rougeL', 'len_ratio', 'evaluation_samples']:
-                    df[col] = 0.0
-                else:
-                    df[col] = ''
-        
-        return df
-        
-    except Exception as e:
-        print(f"Error loading leaderboard: {e}")
-        print("Initializing empty leaderboard...")
-        return initialize_leaderboard().to_pandas()
-
-def save_leaderboard(df: pd.DataFrame) -> bool:
-    """Save leaderboard back to HuggingFace dataset."""
-    try:
-        # Convert DataFrame to Dataset
-        dataset = Dataset.from_pandas(df)
-        
-        # Push to HuggingFace Hub
-        dataset.push_to_hub(
-            LEADERBOARD_DATASET,
-            token=HF_TOKEN,
-            commit_message=f"Update leaderboard - {datetime.datetime.now().isoformat()}"
-        )
-        
-        print("Leaderboard saved successfully!")
-        return True
-        
-    except Exception as e:
-        print(f"Error saving leaderboard: {e}")
-        return False
-
-def add_model_results(
-    model_path: str,
-    author: str,
-    metrics: Dict,
-    detailed_metrics: Dict,
-    evaluation_samples: int,
-    model_type: str
-) -> pd.DataFrame:
-    """Add new model results to leaderboard."""
-    
-    # Load current leaderboard
-    df = load_leaderboard()
-    
-    # Check if model already exists
-    existing = df[df['model_path'] == model_path]
-    if not existing.empty:
-        print(f"Model {model_path} already exists. Updating with new results.")
-        # Remove existing entry
-        df = df[df['model_path'] != model_path]
-    
-    # Create new entry
-    new_entry = {
-        'submission_id': create_submission_id(),
-        'model_path': model_path,
-        'model_display_name': format_model_name(model_path),
-        'author': author,
-        'submission_date': datetime.datetime.now().isoformat(),
-        'bleu': metrics.get('bleu', 0.0),
-        'chrf': metrics.get('chrf', 0.0),
-        'quality_score': metrics.get('quality_score', 0.0),
-        'cer': metrics.get('cer', 0.0),
-        'wer': metrics.get('wer', 0.0),
-        'rouge1': metrics.get('rouge1', 0.0),
-        'rouge2': metrics.get('rouge2', 0.0),
-        'rougeL': metrics.get('rougeL', 0.0),
-        'len_ratio': metrics.get('len_ratio', 0.0),
-        'detailed_metrics': json.dumps(detailed_metrics),
-        'evaluation_samples': evaluation_samples,
-        'model_type': model_type
-    }
-    
-    # Add to dataframe
-    new_df = pd.concat([df, pd.DataFrame([new_entry])], ignore_index=True)
-    
-    # Sort by quality score descending
-    new_df = new_df.sort_values('quality_score', ascending=False).reset_index(drop=True)
-    
-    # Save updated leaderboard
-    save_leaderboard(new_df)
-    
-    return new_df
-
-def get_leaderboard_summary(df: pd.DataFrame) -> Dict:
-    """Get summary statistics for the leaderboard."""
-    if df.empty:
-        return {
-            'total_models': 0,
-            'avg_quality_score': 0.0,
-            'best_model': 'None',
-            'latest_submission': 'None'
-        }
-    
-    return {
-        'total_models': len(df),
-        'avg_quality_score': df['quality_score'].mean(),
-        'best_model': df.iloc[0]['model_display_name'] if not df.empty else 'None',
-        'latest_submission': df['submission_date'].max() if not df.empty else 'None'
-    }
-
-def get_top_models(df: pd.DataFrame, n: int = 10) -> pd.DataFrame:
-    """Get top N models by quality score."""
-    return df.nlargest(n, 'quality_score')
-
-def search_models(df: pd.DataFrame, query: str) -> pd.DataFrame:
-    """Search models by name or author."""
-    if not query:
-        return df
-    
-    query = query.lower()
-    mask = (
-        df['model_display_name'].str.lower().str.contains(query, na=False) |
-        df['author'].str.lower().str.contains(query, na=False) |
-        df['model_path'].str.lower().str.contains(query, na=False)
-    )
-    
-    return df[mask]
-
-def export_results(df: pd.DataFrame, format: str = 'csv') -> str:
-    """Export leaderboard results in specified format."""
-    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
-    
-    if format == 'csv':
-        filename = f"salt_leaderboard_{timestamp}.csv"
-        df.to_csv(filename, index=False)
-        return filename
-    elif format == 'json':
-        filename = f"salt_leaderboard_{timestamp}.json"
-        df.to_json(filename, orient='records', indent=2)
-        return filename
-    else:
-        raise ValueError(f"Unsupported format: {format}")