Spaces:
Running
Running
| """Pure mapping from an OpenEEGBench results DataFrame to the | |
| `braindecode/contents` leaderboard schema. No network, no oeb import.""" | |
| from __future__ import annotations | |
| from typing import Any, Dict, List | |
| from app.config.submission_config import DATASET_ID_TO_ACCURACY_FIELD | |
| AVERAGE_FIELD = "Average ⬆️" # "Average ⬆️" | |
| # Map oeb finetuning kinds to the arena's adapter vocabulary used by the existing | |
| # (paper) results and the frontend filters — frozen linear probing == "probe". | |
| _ADAPTER_ALIAS = {"frozen": "probe"} | |
| def _submitted_date(request: Dict[str, Any]) -> str: | |
| ts = str(request.get("submitted_time", "")) | |
| return ts.split("T", 1)[0] if "T" in ts else ts | |
| def map_oeb_results_to_contents(df, request: Dict[str, Any]) -> List[Dict[str, Any]]: | |
| """One contents row per finetuning strategy present in `df`.""" | |
| if df is None or len(df) == 0: | |
| return [] | |
| kwargs = request.get("benchmark_kwargs", {}) | |
| hub_repo = kwargs.get("hub_repo") | |
| model_cls = kwargs.get("model_cls", "") | |
| arch_fallback = model_cls.split(".")[-1] if model_cls else "" | |
| date = _submitted_date(request) | |
| # No completed classification results (e.g. every experiment failed, so oeb | |
| # produced no metric column) -> nothing to publish. | |
| if "test_balanced_accuracy" not in df.columns: | |
| return [] | |
| metric = df["test_balanced_accuracy"] | |
| if "status" in df.columns: | |
| completed = df[(df["status"] == "completed") & metric.notna()] | |
| else: | |
| completed = df[metric.notna()] | |
| if len(completed) == 0 or "finetuning" not in completed.columns: | |
| return [] | |
| rows: List[Dict[str, Any]] = [] | |
| for strategy, grp in completed.groupby("finetuning"): | |
| backbone = grp["backbone"].iloc[0] if "backbone" in grp.columns else arch_fallback | |
| # mean balanced_accuracy across seeds, per dataset (hf_id) | |
| per_dataset = grp.groupby("dataset")["test_balanced_accuracy"].mean() | |
| accuracies: Dict[str, float] = {} | |
| for hf_id, score in per_dataset.items(): | |
| field = DATASET_ID_TO_ACCURACY_FIELD.get(hf_id) | |
| if field is not None: | |
| accuracies[field] = float(score) | |
| avg_pct = round(sum(accuracies.values()) / len(accuracies) * 100, 2) if accuracies else 0.0 | |
| n_params_m = None | |
| if "trainable_params" in grp.columns and grp["trainable_params"].notna().any(): | |
| n_params_m = round(float(grp["trainable_params"].dropna().iloc[0]) / 1e6, 2) | |
| row: Dict[str, Any] = { | |
| "fullname": request.get("model_name", ""), | |
| "adapter": _ADAPTER_ALIAS.get(str(strategy), str(strategy)), | |
| "Precision": "", | |
| "Model sha": None, | |
| "Architecture": backbone or arch_fallback, | |
| AVERAGE_FIELD: avg_pct, | |
| "#Params (M)": n_params_m, | |
| "Available on the hub": bool(hub_repo), | |
| "Upload To Hub Date": date, | |
| "Submission Date": date, | |
| "Base Model": backbone or arch_fallback, | |
| "Hub License": request.get("hub_license", ""), | |
| "Hub ❤️": 0, | |
| "model_url": request.get("model_url", ""), | |
| "paper_url": request.get("paper_url", ""), | |
| } | |
| # Initialise every benchmark field to 0, then fill the ones we have. | |
| for field in DATASET_ID_TO_ACCURACY_FIELD.values(): | |
| row.setdefault(field, 0) | |
| row.update(accuracies) | |
| rows.append(row) | |
| return rows | |