MMOU Eval

app.py

@@ -0,0 +1,562 @@
+from __future__ import annotations
+
+import json
+import os
+import time
+from collections import defaultdict
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+if "GRADIO_TEMP_DIR" not in os.environ:
+    for candidate in (
+        Path(__file__).resolve().parent / ".gradio_tmp",
+        Path.cwd() / ".gradio_tmp",
+        Path("/tmp") / "gradio",
+    ):
+        try:
+            candidate.mkdir(parents=True, exist_ok=True)
+            probe = candidate / ".write_probe"
+            probe.write_text("ok", encoding="utf-8")
+            probe.unlink()
+            os.environ["GRADIO_TEMP_DIR"] = str(candidate)
+            break
+        except OSError:
+            continue
+
+import gradio as gr
+import pandas as pd
+from huggingface_hub import hf_hub_download
+
+
+DEFAULT_GT_LOCAL_PATH = ""
+DEFAULT_GT_REPO_ID = "nvidia/mmou-gt"
+DEFAULT_GT_FILENAME = "MMOU.json"
+DEFAULT_GT_REPO_TYPE = "dataset"
+DEFAULT_GT_TOKEN_ENV = "HF_TOKEN"
+
+DOMAINS_ORDER = [
+    "Sports",
+    "Travel",
+    "Video Games",
+    "Daily Life",
+    "Academic Lectures",
+    "Film",
+    "Pranks",
+    "Music",
+    "Animation",
+    "News",
+]
+DURATION_BUCKET_ORDER = ["< 5", "5–10", "10–20", "20–30", "> 30", "Overall"]
+GT_LETTER_KEYS = (
+    "correct_option_letter",
+    "correct_answer_letter",
+    "label",
+    "gold_label",
+    "answer_letter",
+)
+GT_DOMAIN_KEYS = ("domain", "category")
+GT_DURATION_KEYS = ("video_duration", "video_duration_sec", "duration", "duration_sec")
+GT_SKILL_KEYS = ("question_type", "skills", "skill", "question_types")
+OPTION_LETTERS = set("ABCDEFGHIJ")
+
+APP_INTRO = """
+# MMOU Evaluator
+
+Upload a `.json` or `.jsonl` file where each entry contains `question_id` and `answer`.
+"""
+
+FORMAT_GUIDE = """
+### Submission Format
+
+Each entry must contain:
+
+- `question_id`
+- `answer`
+
+`answer` must be a single letter from `A` to `J`. Letter matching is case-insensitive. Extra keys are ignored.
+Rows with empty or `null` answers are ignored.
+
+Example JSON:
+
+```json
+[
+  {"question_id": "54aaef4d-2c22-476e-a7e7-37efabde2520", "answer": "C"},
+  {"question_id": "a7f8790d-7828-4ece-a63a-a5d13edf9026", "answer": "B"}
+]
+```
+
+Example JSONL:
+
+```json
+{"question_id": "54aaef4d-2c22-476e-a7e7-37efabde2520", "answer": "C"}
+{"question_id": "a7f8790d-7828-4ece-a63a-a5d13edf9026", "answer": "B"}
+```
+"""
+
+READY_STATUS_MARKDOWN = "### Ready\nUpload a prediction file and click `Evaluate`."
+EMPTY_SUMMARY_MARKDOWN = """
+### Summary
+
+Run an evaluation to populate the aggregate summary.
+"""
+
+LAYOUT_CSS = """
+.gradio-container {
+    max-width: 1100px !important;
+    margin: 0 auto !important;
+    padding-left: 1rem !important;
+    padding-right: 1rem !important;
+    font-size: 16px !important;
+}
+
+.gradio-container .prose,
+.gradio-container .gr-markdown,
+.gradio-container .gr-dataframe,
+.gradio-container label,
+.gradio-container button,
+.gradio-container input,
+.gradio-container textarea {
+    font-size: 1rem !important;
+}
+"""
+
+
+@dataclass(frozen=True)
+class GroundTruthEntry:
+    correct_letter: str
+    domain: str
+    video_duration_sec: float | None
+    skills: tuple[str, ...]
+
+
+def stringify(value: Any) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value.strip()
+    if isinstance(value, (int, float, bool)):
+        return str(value)
+    return json.dumps(value, ensure_ascii=True)
+
+
+def coerce_float(value: Any) -> float | None:
+    if value is None or value == "":
+        return None
+    if isinstance(value, (int, float)):
+        return float(value)
+    if isinstance(value, str):
+        try:
+            return float(value.strip())
+        except ValueError:
+            return None
+    return None
+
+
+def first_present(record: dict[str, Any], keys: tuple[str, ...]) -> Any:
+    return next((record[key] for key in keys if record.get(key) not in (None, "", [])), None)
+
+
+def parse_skill_list(value: Any) -> tuple[str, ...]:
+    items = value if isinstance(value, list) else ([] if value is None else [value])
+    cleaned: list[str] = []
+    seen: set[str] = set()
+    for item in items:
+        text = stringify(item).strip().strip("\"'")
+        if text and text not in seen:
+            seen.add(text)
+            cleaned.append(text)
+    return tuple(cleaned)
+
+
+def safe_pct(correct: int, total: int) -> float:
+    return (100.0 * correct / total) if total else 0.0
+
+
+def duration_bucket(minutes: float) -> str:
+    if minutes < 5:
+        return "< 5"
+    if minutes < 10:
+        return "5–10"
+    if minutes < 20:
+        return "10–20"
+    if minutes < 30:
+        return "20–30"
+    return "> 30"
+
+
+def normalize_answer(value: Any) -> str:
+    answer = stringify(value).upper()
+    if not answer:
+        return ""
+    if len(answer) != 1 or answer not in OPTION_LETTERS:
+        raise ValueError("Each `answer` must be a single letter from A to J.")
+    return answer
+
+
+def load_records(path: str | Path, *, allow_data_key: bool = False) -> tuple[list[dict[str, Any]], str]:
+    file_path = Path(path)
+    suffix = file_path.suffix.lower()
+
+    if suffix in {".jsonl", ".ndjson"}:
+        records: list[dict[str, Any]] = []
+        with file_path.open("r", encoding="utf-8") as handle:
+            for line_number, line in enumerate(handle, start=1):
+                if not line.strip():
+                    continue
+                record = json.loads(line)
+                if not isinstance(record, dict):
+                    raise ValueError(f"Line {line_number} in JSONL must be an object.")
+                records.append(record)
+        return records, "jsonl"
+
+    with file_path.open("r", encoding="utf-8") as handle:
+        payload = json.load(handle)
+
+    if isinstance(payload, list):
+        records = payload
+    elif allow_data_key and isinstance(payload, dict) and isinstance(payload.get("data"), list):
+        records = payload["data"]
+    else:
+        raise ValueError("JSON file must contain a list of objects.")
+
+    if not all(isinstance(item, dict) for item in records):
+        raise ValueError("JSON file must contain only objects.")
+
+    return records, "json"
+
+
+def materialize_ground_truth_file() -> Path:
+    local_path = os.getenv("MMOU_GT_PATH", DEFAULT_GT_LOCAL_PATH).strip()
+    if local_path:
+        path = Path(local_path)
+        if not path.exists():
+            raise FileNotFoundError(
+                "MMOU_GT_PATH is set, but the file does not exist. "
+                "Update the configured path or mount the private file correctly."
+            )
+        return path
+
+    repo_id = os.getenv("MMOU_GT_REPO_ID", DEFAULT_GT_REPO_ID).strip()
+    filename = os.getenv("MMOU_GT_FILENAME", DEFAULT_GT_FILENAME).strip()
+    if repo_id and filename:
+        repo_type = os.getenv("MMOU_GT_REPO_TYPE", DEFAULT_GT_REPO_TYPE).strip() or "dataset"
+        token_env = os.getenv("MMOU_GT_TOKEN_ENV", DEFAULT_GT_TOKEN_ENV).strip() or "HF_TOKEN"
+        token = os.getenv(token_env) or os.getenv("HF_TOKEN", "")
+        return Path(
+            hf_hub_download(
+                repo_id=repo_id,
+                filename=filename,
+                repo_type=repo_type,
+                token=token or None,
+            )
+        )
+
+    raise RuntimeError(
+        "Ground truth is not configured. Set MMOU_GT_PATH or "
+        "MMOU_GT_REPO_ID/MMOU_GT_FILENAME before launching the app."
+    )
+
+
+@lru_cache(maxsize=1)
+def load_ground_truth() -> dict[str, GroundTruthEntry]:
+    records, _ = load_records(materialize_ground_truth_file(), allow_data_key=True)
+    entries: dict[str, GroundTruthEntry] = {}
+
+    for record in records:
+        question_id = stringify(record.get("question_id"))
+        if not question_id:
+            continue
+
+        correct_letter = next(
+            (
+                letter
+                for key in GT_LETTER_KEYS
+                if (letter := stringify(record.get(key)).upper()) in OPTION_LETTERS
+            ),
+            "",
+        )
+        if not correct_letter:
+            continue
+
+        entries[question_id] = GroundTruthEntry(
+            correct_letter=correct_letter,
+            domain=stringify(first_present(record, GT_DOMAIN_KEYS)) or "Unknown",
+            video_duration_sec=coerce_float(first_present(record, GT_DURATION_KEYS)),
+            skills=parse_skill_list(first_present(record, GT_SKILL_KEYS)),
+        )
+
+    if not entries:
+        raise RuntimeError("No usable ground-truth question IDs were found.")
+
+    return entries
+
+
+def build_prediction_map(records: list[dict[str, Any]]) -> tuple[dict[str, str], int, int]:
+    predictions: dict[str, str] = {}
+    duplicates = 0
+    skipped_empty_answers = 0
+
+    for index, record in enumerate(records, start=1):
+        question_id = stringify(record.get("question_id"))
+        if not question_id:
+            raise ValueError(f"Row {index} is missing `question_id`.")
+        answer = normalize_answer(record.get("answer"))
+        if not answer:
+            skipped_empty_answers += 1
+            continue
+        if question_id in predictions:
+            duplicates += 1
+        predictions[question_id] = answer
+
+    return predictions, duplicates, skipped_empty_answers
+
+
+def bump(stats: dict[str, dict[str, int]], keys: list[str], field: str) -> None:
+    for key in keys:
+        stats[key][field] += 1
+
+
+def make_breakdown_dataframe(
+    stats: dict[str, dict[str, int]],
+    label: str,
+    ordered_labels: list[str] | None = None,
+) -> pd.DataFrame:
+    rows = [
+        {
+            label: name,
+            "Official Accuracy (%)": round(safe_pct(counts["correct"], counts["total"]), 2),
+            "Answered Accuracy (%)": round(safe_pct(counts["correct"], counts["answered"]), 2),
+            "Coverage (%)": round(safe_pct(counts["answered"], counts["total"]), 2),
+            "Correct": counts["correct"],
+            "Answered": counts["answered"],
+            "Total": counts["total"],
+        }
+        for name, counts in stats.items()
+    ]
+
+    if not rows:
+        return pd.DataFrame(
+            columns=[
+                label,
+                "Official Accuracy (%)",
+                "Answered Accuracy (%)",
+                "Coverage (%)",
+                "Correct",
+                "Answered",
+                "Total",
+            ]
+        )
+
+    frame = pd.DataFrame(rows)
+    if ordered_labels:
+        rank = {name: idx for idx, name in enumerate(ordered_labels)}
+        frame["_rank"] = frame[label].map(lambda name: rank.get(name, len(rank)))
+        return frame.sort_values(["_rank", label]).drop(columns="_rank").reset_index(drop=True)
+
+    return frame.sort_values(["Answered Accuracy (%)", "Total"], ascending=[False, False]).reset_index(drop=True)
+
+
+def build_metrics_markdown(summary: dict[str, Any]) -> str:
+    return "\n".join(
+        [
+            "### Metrics",
+            f"- Official accuracy: `{summary['official_accuracy_pct']:.2f}%` "
+            f"(`{summary['correct']} / {summary['total_ground_truth']}`)",
+            f"- Answered accuracy: `{summary['answered_accuracy_pct']:.2f}%` "
+            f"(`{summary['correct']} / {summary['answered_predictions']}`)",
+            f"- Coverage: `{summary['coverage_pct']:.2f}%`",
+            f"- Matched IDs: `{summary['matched_prediction_ids']}`",
+            f"- Missing IDs: `{summary['missing_prediction_ids']}`",
+            f"- Extra IDs: `{summary['extra_prediction_ids']}`",
+            f"- Duplicate IDs: `{summary['duplicate_prediction_ids']}`",
+        ]
+    )
+
+
+def build_summary_markdown(domain_df: pd.DataFrame, duration_df: pd.DataFrame, skill_df: pd.DataFrame) -> str:
+    accuracy_column = "Answered Accuracy (%)"
+    best_domain = "n/a"
+    best_duration = "n/a"
+    lowest_skill = "n/a"
+
+    if not domain_df.empty:
+        row = domain_df.sort_values([accuracy_column, "Total"], ascending=[False, False]).iloc[0]
+        best_domain = f"{row['Domain']} ({row[accuracy_column]:.2f}%)"
+
+    if not duration_df.empty:
+        rows = duration_df[duration_df["Duration Bucket"] != "Overall"]
+        if not rows.empty:
+            row = rows.sort_values([accuracy_column, "Total"], ascending=[False, False]).iloc[0]
+            best_duration = f"{row['Duration Bucket']} ({row[accuracy_column]:.2f}%)"
+
+    if not skill_df.empty:
+        rows = skill_df[skill_df["Total"] >= 10]
+        if rows.empty:
+            rows = skill_df
+        row = rows.sort_values([accuracy_column, "Total"], ascending=[True, False]).iloc[0]
+        lowest_skill = f"{row['Skill']} ({row[accuracy_column]:.2f}%)"
+
+    return "\n".join(
+        [
+            "### Summary",
+            f"- Best domain by answered accuracy: `{best_domain}`",
+            f"- Best duration bucket by answered accuracy: `{best_duration}`",
+            f"- Lowest skill bucket by answered accuracy: `{lowest_skill}`",
+        ]
+    )
+
+
+def empty_result(status: str) -> tuple[str, str, str, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+    return status, "", EMPTY_SUMMARY_MARKDOWN, pd.DataFrame(), pd.DataFrame(), pd.DataFrame()
+
+
+def evaluate_submission(
+    prediction_file: str | None,
+) -> tuple[str, str, str, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+    if not prediction_file:
+        return empty_result(
+            "### Upload required\nPlease upload a `.json` or `.jsonl` prediction file before evaluating."
+        )
+
+    started_at = time.time()
+
+    try:
+        ground_truth = load_ground_truth()
+        records, file_format = load_records(prediction_file)
+        if not records:
+            raise ValueError("No valid prediction records were found in the uploaded file.")
+
+        predictions, duplicate_prediction_ids, skipped_empty_answers = build_prediction_map(records)
+        domain_stats: dict[str, dict[str, int]] = defaultdict(lambda: {"correct": 0, "answered": 0, "total": 0})
+        duration_stats: dict[str, dict[str, int]] = defaultdict(lambda: {"correct": 0, "answered": 0, "total": 0})
+        skill_stats: dict[str, dict[str, int]] = defaultdict(lambda: {"correct": 0, "answered": 0, "total": 0})
+
+        correct = 0
+        answered = 0
+        gt_ids = set(ground_truth)
+        pred_ids = set(predictions)
+
+        for question_id, gt in ground_truth.items():
+            duration_key = duration_bucket(gt.video_duration_sec / 60.0) if gt.video_duration_sec is not None else None
+            scopes = [
+                (domain_stats, [gt.domain]),
+                (duration_stats, [duration_key] if duration_key else []),
+                (skill_stats, list(gt.skills)),
+            ]
+
+            for stats, keys in scopes:
+                bump(stats, keys, "total")
+
+            answer = predictions.get(question_id)
+            if not answer:
+                continue
+
+            answered += 1
+            for stats, keys in scopes:
+                bump(stats, keys, "answered")
+
+            if answer == gt.correct_letter:
+                correct += 1
+                for stats, keys in scopes:
+                    bump(stats, keys, "correct")
+
+        total_ground_truth = len(ground_truth)
+        duration_stats["Overall"] = {"total": total_ground_truth, "answered": answered, "correct": correct}
+
+        summary = {
+            "correct": correct,
+            "answered_predictions": answered,
+            "total_ground_truth": total_ground_truth,
+            "official_accuracy_pct": safe_pct(correct, total_ground_truth),
+            "answered_accuracy_pct": safe_pct(correct, answered),
+            "coverage_pct": safe_pct(answered, total_ground_truth),
+            "matched_prediction_ids": len(pred_ids & gt_ids),
+            "missing_prediction_ids": total_ground_truth - len(pred_ids & gt_ids),
+            "extra_prediction_ids": len(pred_ids - gt_ids),
+            "duplicate_prediction_ids": duplicate_prediction_ids,
+        }
+
+        domain_df = make_breakdown_dataframe(domain_stats, "Domain", ordered_labels=DOMAINS_ORDER)
+        duration_df = make_breakdown_dataframe(
+            duration_stats,
+            "Duration Bucket",
+            ordered_labels=DURATION_BUCKET_ORDER,
+        )
+        skill_df = make_breakdown_dataframe(skill_stats, "Skill")
+
+        status_markdown = (
+            "### Evaluation complete\n"
+            f"- Parsed file format: `{file_format}`\n"
+            f"- Uploaded rows: `{len(records)}`\n"
+            f"- Skipped empty answers: `{skipped_empty_answers}`\n"
+            f"- Evaluation time: `{time.time() - started_at:.2f}s`"
+        )
+        return (
+            status_markdown,
+            build_metrics_markdown(summary),
+            build_summary_markdown(domain_df, duration_df, skill_df),
+            domain_df,
+            duration_df,
+            skill_df,
+        )
+
+    except Exception as exc:
+        return empty_result(f"### Evaluation failed\n`{type(exc).__name__}: {exc}`")
+
+
+def clear_outputs() -> tuple[None, str, str, str, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+    return None, READY_STATUS_MARKDOWN, "", EMPTY_SUMMARY_MARKDOWN, pd.DataFrame(), pd.DataFrame(), pd.DataFrame()
+
+
+with gr.Blocks(title="MMOU Evaluator", fill_width=False) as demo:
+    gr.Markdown(APP_INTRO)
+
+    prediction_file = gr.File(label="Upload prediction file", file_types=[".json", ".jsonl"], type="filepath")
+
+    with gr.Row():
+        evaluate_button = gr.Button("Evaluate", variant="primary")
+        clear_button = gr.Button("Clear")
+
+    status_markdown = gr.Markdown(READY_STATUS_MARKDOWN)
+    metrics_markdown = gr.Markdown("")
+    summary_markdown = gr.Markdown(EMPTY_SUMMARY_MARKDOWN)
+    gr.Markdown(FORMAT_GUIDE)
+
+    with gr.Tabs():
+        with gr.Tab("Domain Breakdown"):
+            domain_dataframe = gr.Dataframe(label="Domain breakdown", interactive=False, wrap=True)
+        with gr.Tab("Duration Breakdown"):
+            duration_dataframe = gr.Dataframe(label="Duration breakdown", interactive=False, wrap=True)
+        with gr.Tab("Skill Breakdown"):
+            skill_dataframe = gr.Dataframe(label="Skill breakdown", interactive=False, wrap=True)
+
+    evaluate_button.click(
+        fn=evaluate_submission,
+        inputs=[prediction_file],
+        outputs=[
+            status_markdown,
+            metrics_markdown,
+            summary_markdown,
+            domain_dataframe,
+            duration_dataframe,
+            skill_dataframe,
+        ],
+    )
+    clear_button.click(
+        fn=clear_outputs,
+        outputs=[
+            prediction_file,
+            status_markdown,
+            metrics_markdown,
+            summary_markdown,
+            domain_dataframe,
+            duration_dataframe,
+            skill_dataframe,
+        ],
+    )
+
+
+if __name__ == "__main__":
+    demo.launch(theme=gr.themes.Default(), css=LAYOUT_CSS)

requirements.txt

@@ -0,0 +1,3 @@
+gradio>=6.0.0
+pandas>=2.2.0
+huggingface_hub>=0.30.0