"""MSEB Metric Aggregation, SVQ Macro-Averaging, and DataFrame Matrix Engine. This module aggregates raw EvaluationRecord data across tasks and datasets, computes task averages, normalizes metric directions for overall SVQ macro-averaging, generates wide task DataFrames and the Overall summary leaderboard DataFrame, and provides interactive search and column filtering. """ from __future__ import annotations import logging import math from typing import Any, Dict, List, Optional, Tuple import numpy as np import pandas as pd from src.config import get_source_file_url from src.config import MSEB_TASKS from src.config import normalize_metric_for_svq from src.config import ORDERED_TASK_KEYS from src.config import resolve_results_dir from src.config import TASK_PRIMARY_METRICS from src.data.models import EvaluationRecord from src.data.models import LeaderboardData from src.data.models import ModelEntry from src.data.models import ModelScores from src.data.models import TaskScoreSummary from src.data.parser import discover_models from src.data.parser import load_all_evaluation_data logger = logging.getLogger(__name__) def format_score_for_task_table(val: Optional[float]) -> float: """Formats a raw score for display in a task-specific DataFrame.""" if val is None: return np.nan try: f_val = float(val) except (ValueError, TypeError): return np.nan if math.isnan(f_val): return np.nan return round(f_val * 100.0, 2) def _get_base_model_name(entry: ModelEntry) -> str: """Returns the base model name without sub_config suffix.""" name = entry.display_name or entry.top_model if entry.sub_config and name.endswith(f" ({entry.sub_config})"): return name[: -len(f" ({entry.sub_config})")].strip() return name def _format_dataset_display_name(dataset_name: str, task_key: str) -> str: """Formats dataset display name for task tables and column selectors.""" match task_key.lower().strip(): case "transcription": return dataset_name.replace("SpeechTranscription", "") case "classification": return dataset_name.replace("Classification", "") case "clustering": return dataset_name.replace("Clustering", "") case "reasoning": return dataset_name.replace("Reasoning", "") case "reranking": return dataset_name.replace("Reranking", "") case "retrieval": return dataset_name.replace("Retrieval", "") case _: # Optional: handle default or unmatched keys return dataset_name def build_overall_dataframe( records: Optional[List[EvaluationRecord]] = None, models: Optional[List[ModelEntry]] = None, ) -> pd.DataFrame: """Builds the master summary DataFrame for the Overall Leaderboard tab. Args: records: List of parsed EvaluationRecord objects. models: Optional list of discovered ModelEntry objects. If None, extracted from records. Returns: pd.DataFrame sorted descending by 'Average ⬆️' with task score columns, metadata columns, and NaN for unevaluated tasks. """ task_cols = [t.capitalize() for t in ORDERED_TASK_KEYS] if not records and not models: return pd.DataFrame(columns=["Model", "Tags", "Average ⬆️"] + task_cols) # 1. Index models models_dict: Dict[str, ModelEntry] = {} if models: for m in models: models_dict[m.entry_id] = m for rec in records or []: if rec.model_entry.entry_id not in models_dict: models_dict[rec.model_entry.entry_id] = rec.model_entry if not models_dict: return pd.DataFrame(columns=["Model", "Tags", "Average ⬆️"] + task_cols) # 2. Accumulate raw and SVQ scores per model, task, and dataset raw_dataset_scores: Dict[str, Dict[str, Dict[str, List[float]]]] = { mid: {t: {} for t in ORDERED_TASK_KEYS} for mid in models_dict } svq_dataset_scores: Dict[str, Dict[str, Dict[str, List[float]]]] = { mid: {t: {} for t in ORDERED_TASK_KEYS} for mid in models_dict } task_source_files: Dict[str, Dict[str, str]] = { mid: {} for mid in models_dict } for rec in records or []: mid = rec.model_entry.entry_id tname = rec.task_name if tname not in raw_dataset_scores[mid]: continue if tname in MSEB_TASKS and not MSEB_TASKS[tname].is_active: continue val = rec.main_score_value if val is None or math.isnan(val): continue if rec.source_file: task_source_files[mid][tname] = rec.source_file ds = rec.dataset_name raw_dataset_scores[mid][tname].setdefault(ds, []).append(val) # Compute normalized SVQ score norm_s = normalize_metric_for_svq(val, rec.main_score_name) if not math.isnan(norm_s): svq_dataset_scores[mid][tname].setdefault(ds, []).append(norm_s) # 3. Build row per model rows: List[Dict[str, Any]] = [] for mid, entry in models_dict.items(): row: Dict[str, Any] = { "Model": _get_base_model_name(entry), "Tags": entry.sub_config or "-", "entry_id": entry.entry_id, "top_model": entry.top_model, "sub_config": entry.sub_config or "", "url": entry.url, } for t in ORDERED_TASK_KEYS: col_name = t.capitalize() t_raw_ds = raw_dataset_scores[mid][t] src_f = task_source_files[mid].get(t) if src_f: row[f"{col_name}__url"] = get_source_file_url(src_f) if t_raw_ds: # Average per dataset first (handles multiple noise runs/subtasks) ds_means = [ sum(scores) / len(scores) for scores in t_raw_ds.values() if scores ] if ds_means: task_raw_mean = sum(ds_means) / len(ds_means) row[col_name] = format_score_for_task_table(task_raw_mean) else: row[col_name] = np.nan else: row[col_name] = np.nan # SVQ Macro-average across all tasks, treating missing tasks as 0.0 task_norm_scores: List[float] = [] for t in ORDERED_TASK_KEYS: col_name = t.capitalize() disp_val = row.get(col_name) if pd.notna(disp_val): try: f_val = float(disp_val) if t == "transcription": norm_val = 100.0 - f_val else: norm_val = f_val task_norm_scores.append(norm_val) except (ValueError, TypeError): task_norm_scores.append(0.0) else: task_norm_scores.append(0.0) if task_norm_scores: row["Average ⬆️"] = round(float(np.mean(task_norm_scores)), 2) else: row["Average ⬆️"] = 0.0 rows.append(row) df = pd.DataFrame(rows) if not df.empty: meta_cols = ["entry_id", "top_model", "sub_config", "url"] task_cols = [ t.capitalize() for t in ORDERED_TASK_KEYS if t.capitalize() in df.columns ] url_cols = [c for c in df.columns if c.endswith("__url")] other_cols = [ c for c in df.columns if c not in ["Model", "Tags", "Average ⬆️"] + meta_cols + task_cols + url_cols ] ordered_cols = ( ["Model", "Tags", "Average ⬆️"] + task_cols + [m for m in meta_cols if m in df.columns] + url_cols + other_cols ) df = df[[c for c in ordered_cols if c in df.columns]] if "Average ⬆️" in df.columns: df = df.sort_values(by=["Average ⬆️"], ascending=False).reset_index( drop=True ) return df def build_task_dataframe( records: Optional[List[EvaluationRecord]], task_name: str, models: Optional[List[ModelEntry]] = None, ) -> pd.DataFrame: """Builds a fine-grained DataFrame for an individual MSEB task tab. Args: records: List of parsed EvaluationRecord objects. task_name: Normalized task key (e.g. 'reasoning', 'classification'). models: Optional list of discovered ModelEntry objects. Returns: pd.DataFrame with columns ['Model', 'Tags', , ..., , 'Task Average'] sorted descending by 'Task Average'. """ del models t_clean = task_name.lower().strip() if t_clean in MSEB_TASKS and not MSEB_TASKS[t_clean].is_active: return pd.DataFrame(columns=["Model", "Tags", "Task Average"]) task_records = [r for r in records or [] if r.task_name == t_clean] if not task_records: return pd.DataFrame(columns=["Model", "Tags", "Task Average"]) # Discover all evaluated datasets for this task all_datasets = sorted( list( set( _format_dataset_display_name(r.dataset_name, t_clean) for r in task_records ) ) ) # Index models evaluated in this task model_recs: Dict[str, Tuple[ModelEntry, Dict[str, List[float]]]] = {} dataset_source_files: Dict[str, Dict[str, str]] = {} for r in task_records: mid = r.model_entry.entry_id ds_name = _format_dataset_display_name(r.dataset_name, t_clean) if mid not in model_recs: model_recs[mid] = (r.model_entry, {}) dataset_source_files[mid] = {} if r.source_file: dataset_source_files[mid][ds_name] = r.source_file val = r.main_score_value if val is not None: try: f_val = float(val) if not math.isnan(f_val): model_recs[mid][1].setdefault(ds_name, []).append(f_val) except (ValueError, TypeError): pass rows: List[Dict[str, Any]] = [] for mid, (entry, ds_dict) in model_recs.items(): row: Dict[str, Any] = { "Model": _get_base_model_name(entry), "Tags": entry.sub_config or "-", "entry_id": entry.entry_id, "top_model": entry.top_model, "sub_config": entry.sub_config or "", "url": entry.url, } dataset_vals: List[float] = [] for ds in all_datasets: src_f = dataset_source_files.get(mid, {}).get(ds) if src_f: row[f"{ds}__url"] = get_source_file_url(src_f) scores = ds_dict.get(ds, []) if scores: mean_s = sum(scores) / len(scores) disp_s = format_score_for_task_table(mean_s) row[ds] = disp_s dataset_vals.append(disp_s) else: row[ds] = np.nan dataset_vals.append(0.0) if dataset_vals: avg_val = float(np.mean(dataset_vals)) if t_clean == "transcription": row["Task Average"] = round(100.0 - avg_val, 2) else: row["Task Average"] = round(avg_val, 2) else: row["Task Average"] = 0.0 rows.append(row) df = pd.DataFrame(rows) if not df.empty: meta_cols = ["entry_id", "top_model", "sub_config", "url"] ds_cols = [d for d in all_datasets if d in df.columns] url_cols = [c for c in df.columns if c.endswith("__url")] ordered_cols = ( ["Model", "Tags", "Task Average"] + ds_cols + [m for m in meta_cols if m in df.columns] + url_cols ) df = df[[c for c in ordered_cols if c in df.columns]] if "Task Average" in df.columns: spec = MSEB_TASKS.get(t_clean) if t_clean == "transcription": is_lib = False # Transcription value is 100-val. elif spec: is_lib = spec.metric_direction.name == "LOWER_IS_BETTER" else: is_lib = False df = df.sort_values( by=["Task Average"], ascending=is_lib, na_position="last" ).reset_index(drop=True) return df def _build_typed_model_scores( records: Optional[List[EvaluationRecord]], models: List[ModelEntry], overall_df: pd.DataFrame, ) -> List[ModelScores]: """Constructs typed ModelScores and TaskScoreSummary objects.""" recs = records or [] # Map entry_id to overall_svq_score from overall_df overall_scores_map: Dict[str, float] = {} if ( not overall_df.empty and "entry_id" in overall_df.columns and "Average ⬆️" in overall_df.columns ): for _, r in overall_df.iterrows(): if pd.notna(r["Average ⬆️"]): overall_scores_map[r["entry_id"]] = float(r["Average ⬆️"]) # Group records by (mid, task) rec_group: Dict[str, Dict[str, List[EvaluationRecord]]] = { m.entry_id: {t: [] for t in ORDERED_TASK_KEYS} for m in models } for r in recs: mid = r.model_entry.entry_id t = r.task_name.lower().strip() if t in MSEB_TASKS and not MSEB_TASKS[t].is_active: continue if mid in rec_group and t in rec_group[mid]: rec_group[mid][t].append(r) result: List[ModelScores] = [] for m in models: mid = m.entry_id task_summaries: Dict[str, TaskScoreSummary] = {} total_eval_datasets = 0 total_recs = 0 for t in ORDERED_TASK_KEYS: t_recs = rec_group[mid][t] total_recs += len(t_recs) primary_metric = TASK_PRIMARY_METRICS.get(t, "Score") if not t_recs: task_summaries[t] = TaskScoreSummary( task_name=t, primary_metric=primary_metric, ) continue # Group scores by dataset ds_raw_scores: Dict[str, List[float]] = {} ds_svq_scores: Dict[str, List[float]] = {} for r in t_recs: if r.main_score_value is not None: try: f_val = float(r.main_score_value) if not math.isnan(f_val): ds_name = _format_dataset_display_name(r.dataset_name, t) ds_raw_scores.setdefault(ds_name, []).append(f_val) norm_s = normalize_metric_for_svq(f_val, r.main_score_name) if not math.isnan(norm_s): ds_svq_scores.setdefault(ds_name, []).append(norm_s) except (ValueError, TypeError): pass ds_means: Dict[str, float] = { ds: float(np.mean(vals)) for ds, vals in ds_raw_scores.items() if vals } svq_means: Dict[str, float] = { ds: float(np.mean(vals)) for ds, vals in ds_svq_scores.items() if vals } all_mean = float(np.mean(list(ds_means.values()))) if ds_means else None svq_mean = float(np.mean(list(svq_means.values()))) if svq_means else None total_eval_datasets += len(ds_means) task_summaries[t] = TaskScoreSummary( task_name=t, primary_metric=primary_metric, all_mean=round(all_mean, 4) if all_mean is not None else None, svq_mean=round(svq_mean, 4) if svq_mean is not None else None, normalized_svq_score=round(svq_mean, 4) if svq_mean is not None else None, dataset_scores=ds_means, dataset_count=len(ds_means), svq_dataset_count=len(svq_means), ) eval_task_cnt = sum( 1 for ts in task_summaries.values() if ts.dataset_count > 0 ) result.append( ModelScores( model_entry=m, task_summaries=task_summaries, overall_svq_score=overall_scores_map.get(mid), evaluated_task_count=eval_task_cnt, evaluated_dataset_count=total_eval_datasets, records_count=total_recs, ) ) return result def build_leaderboard_tables( records: Optional[List[EvaluationRecord]] = None, results_dir: Optional[str] = None, models: Optional[List[ModelEntry]] = None, ) -> LeaderboardData: """High-level coordinator that ingests results and constructs all leaderboard tables. Args: records: Optional list of parsed EvaluationRecord objects. results_dir: Optional directory path to search for results if records is None. models: Optional list of ModelEntry objects. Returns: LeaderboardData dataclass containing overall_df, task_dfs, models, and model_scores. """ if records is None: resolved_dir = resolve_results_dir(results_dir) models = discover_models(resolved_dir) records = load_all_evaluation_data(resolved_dir) if models is None: models_map = {r.model_entry.entry_id: r.model_entry for r in records or []} models = list(models_map.values()) overall_df = build_overall_dataframe(records=records, models=models) task_dfs: Dict[str, pd.DataFrame] = {} for t in ORDERED_TASK_KEYS: task_dfs[t] = build_task_dataframe( records=records, task_name=t, models=models ) model_scores = _build_typed_model_scores( records=records, models=models, overall_df=overall_df ) return LeaderboardData( overall_df=overall_df, task_dfs=task_dfs, models=models, model_scores=model_scores, ) def filter_leaderboard_df( df: pd.DataFrame, search_query: str = "", selected_columns: Optional[List[str]] = None, cant_deselect: Optional[List[str]] = None, ) -> pd.DataFrame: """Filters a leaderboard DataFrame by model name search query and visible column selection. Args: df: Input DataFrame. search_query: Search string to filter Model names. selected_columns: Subset of columns to display. cant_deselect: Mandatory columns that cannot be hidden (e.g., 'Model'). Returns: Filtered pd.DataFrame with reset index. """ if df.empty: return df.copy() filtered = df.copy() # 1. Search Query Filtering query = (search_query or "").strip() if query: mask = pd.Series(False, index=filtered.index) if "Model" in filtered.columns: mask |= ( filtered["Model"] .astype(str) .str.contains(query, case=False, regex=False) ) if "Tags" in filtered.columns: mask |= ( filtered["Tags"] .astype(str) .str.contains(query, case=False, regex=False) ) if "top_model" in filtered.columns: mask |= ( filtered["top_model"] .astype(str) .str.contains(query, case=False, regex=False) ) if "sub_config" in filtered.columns: mask |= ( filtered["sub_config"] .astype(str) .str.contains(query, case=False, regex=False) ) if "entry_id" in filtered.columns: mask |= ( filtered["entry_id"] .astype(str) .str.contains(query, case=False, regex=False) ) filtered = filtered[mask] # 2. Column Selection & Retention if selected_columns is not None: raw_keep = list(selected_columns) if cant_deselect: for c in reversed(cant_deselect): if c in df.columns and c not in raw_keep: raw_keep.insert(0, c) # Deduplicate while preserving order keep = [] for c in raw_keep: if c not in keep and c in filtered.columns: keep.append(c) if keep: filtered = filtered[keep] return filtered.reset_index(drop=True) def compute_dynamic_overall_average( df: pd.DataFrame, selected_task_columns: Optional[List[str]] = None, ) -> pd.DataFrame: """Dynamically recalculates the 'Average ⬆️' column. Computes the macro-average of the currently selected task columns for each model row, properly taking metric orientation into account (e.g. inverting lower-is-better WER for Transcription), and re-sorting rows by the dynamic Average in descending order. Args: df: Overall leaderboard DataFrame containing model rows and task columns. selected_task_columns: List of visible task column names (e.g. ['Classification', 'Reasoning']). Returns: pd.DataFrame with recomputed 'Average ⬆️' sorted descending. """ if df.empty: return df.copy() df_copy = df.copy() task_keys = [t.capitalize() for t in ORDERED_TASK_KEYS] if selected_task_columns is None: active_tasks = [c for c in task_keys if c in df_copy.columns] elif len(selected_task_columns) == 0: active_tasks = [] else: active_tasks = [ c for c in selected_task_columns if c in df_copy.columns and c in task_keys ] averages: List[float] = [] for _, row in df_copy.iterrows(): row_scores: List[float] = [] for col in active_tasks: val = row[col] if pd.notna(val): try: f_val = float(val) t_key = col.lower().strip() if t_key == "transcription": norm_val = 100.0 - f_val else: norm_val = f_val row_scores.append(norm_val) except (ValueError, TypeError): row_scores.append(0.0) else: row_scores.append(0.0) if active_tasks: averages.append(round(float(np.mean(row_scores)), 2)) else: averages.append(np.nan) df_copy["Average ⬆️"] = averages # Sort descending by Average ⬆️ with NaN at bottom if "Average ⬆️" in df_copy.columns: df_copy = df_copy.sort_values( by=["Average ⬆️"], ascending=False, na_position="last", ).reset_index(drop=True) return df_copy def compute_dynamic_task_average( df: pd.DataFrame, selected_dataset_columns: Optional[List[str]] = None, task_key: str = "", ) -> pd.DataFrame: """Dynamically recalculates the 'Task Average' column. Computes the average of the currently selected dataset score columns for each model row in a task tab, and re-sorts rows by the dynamic Task Average. Args: df: Task DataFrame containing model rows and dataset columns. selected_dataset_columns: List of visible dataset column names. task_key: Normalized task key (e.g. 'classification', 'transcription'). Returns: pd.DataFrame with recomputed 'Task Average' sorted appropriately. """ if df.empty: return df.copy() df_copy = df.copy() non_ds_cols = { "Model", "Tags", "Task Average", "entry_id", "top_model", "sub_config", "url", "Rank", } all_dataset_cols = [ c for c in df_copy.columns if c not in non_ds_cols and not c.endswith("__url") ] if selected_dataset_columns is None: active_datasets = all_dataset_cols elif len(selected_dataset_columns) == 0: active_datasets = [] else: active_datasets = [ c for c in selected_dataset_columns if c in df_copy.columns and c in all_dataset_cols ] averages: List[float] = [] for _, row in df_copy.iterrows(): row_scores: List[float] = [] for col in active_datasets: val = row[col] if pd.notna(val): try: f_val = float(val) row_scores.append(f_val) except (ValueError, TypeError): row_scores.append(0.0) else: row_scores.append(0.0) if active_datasets: avg_val = float(np.mean(row_scores)) t_clean = task_key.lower().strip() if t_clean == "transcription": averages.append(round(100.0 - avg_val, 2)) else: averages.append(round(avg_val, 2)) else: averages.append(np.nan) df_copy["Task Average"] = averages # Sort appropriately by Task Average if "Task Average" in df_copy.columns: t_clean = task_key.lower().strip() spec = MSEB_TASKS.get(t_clean) if t_clean == "transcription": is_lib = False # Transcription value is 100-val. elif spec: is_lib = spec.metric_direction.name == "LOWER_IS_BETTER" else: is_lib = False df_copy = df_copy.sort_values( by=["Task Average"], ascending=is_lib, na_position="last", ).reset_index(drop=True) return df_copy