icassp2027-experiment/summarize_easycom_event_prelabels.py

#!/usr/bin/env python3
"""Summarize EasyCom AI event pre-labels without treating them as ground truth."""

from __future__ import annotations

import argparse
import csv
import json
import statistics
from collections import Counter
from pathlib import Path
from typing import Any, Iterable


MANUSCRIPT_ROWS = [
    ("Positive: wearer entry", "event", "WEARER_ENTRY"),
    ("HN: same-speaker pause", "subtype", "ENTRY_OTHER_MIDTURN_PAUSE"),
    ("HN: other-to-other transition", "subtype", "ENTRY_OTHER_TO_OTHER"),
    ("HN: wearer backchannel", "subtype", "ENTRY_WEARER_BACKCHANNEL"),
    ("Positive: INT-yield", "event", "INT_YIELD"),
    ("HN: other backchannel", "subtype", "INT_OTHER_BACKCHANNEL"),
    ("HN: non-takeover overlap", "subtype", "INT_NON_TAKEOVER_OVERLAP"),
    ("HN: wearer holds floor", "subtype", "INT_WEARER_HOLDS_FLOOR"),
    ("HN: other", "subtype", "OTHER_HARD_NEGATIVE"),
    ("Excluded: valid but out of scope", "exclude_group", "OUT_OF_SCOPE"),
    ("Excluded: insufficient evidence", "exclude_group", "INSUFFICIENT_EVIDENCE"),
    ("Excluded: data quality", "exclude_group", "DATA_QUALITY"),
    ("Scoring-eligible total", "derived", "SCORING_ELIGIBLE"),
    ("All candidates", "derived", "ALL"),
]


def _read_jsonl(path: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    with path.open(encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, 1):
            if not line.strip():
                continue
            try:
                row = json.loads(line)
            except json.JSONDecodeError as error:
                raise ValueError(f"Invalid JSON at {path}:{line_number}: {error}") from error
            row["_source_file"] = str(path)
            rows.append(row)
    return rows


def _exclude_group(row: dict[str, Any]) -> str | None:
    group = row.get("ai_exclude_reason_group")
    if group:
        return str(group)
    reason = row.get("ai_exclude_reason")
    if not reason:
        return None
    if str(reason).startswith("OUT_OF_SCOPE_"):
        return "OUT_OF_SCOPE"
    if str(reason).startswith("DATA_"):
        return "DATA_QUALITY"
    return str(reason)


def _decision_key(decision: dict[str, Any], *, include_opportunity: bool) -> tuple[Any, ...]:
    keys = ["verdict", "event_type", "subtype", "exclude_reason"]
    if include_opportunity:
        keys.append("selected_opportunity_uid")
    return tuple(
        decision.get(key) for key in keys
    )


def _count(rows: Iterable[dict[str, Any]], dimension: str, value: str) -> int:
    rows = list(rows)
    if dimension == "event":
        return sum(row.get("ai_verdict") == "POSITIVE" and row.get("ai_event_type") == value for row in rows)
    if dimension == "subtype":
        return sum(row.get("ai_verdict") == "HARD_NEGATIVE" and row.get("ai_subtype") == value for row in rows)
    if dimension == "exclude_group":
        return sum(row.get("ai_verdict") == "EXCLUDE" and _exclude_group(row) == value for row in rows)
    if value == "SCORING_ELIGIBLE":
        return sum(row.get("ai_verdict") in {"POSITIVE", "HARD_NEGATIVE"} for row in rows)
    if value == "ALL":
        return len(rows)
    raise ValueError((dimension, value))


def _cohort_summary(name: str, rows: list[dict[str, Any]]) -> dict[str, Any]:
    confidences = [float(row["ai_confidence"]) for row in rows if row.get("ai_confidence") is not None]
    pass_rows = 0
    semantic_agree = 0
    full_agree = 0
    for row in rows:
        decisions = (row.get("llm_adjudication") or {}).get("decisions") or []
        if len(decisions) >= 2:
            pass_rows += 1
            semantic_agree += len({_decision_key(decision, include_opportunity=False) for decision in decisions}) == 1
            full_agree += len({_decision_key(decision, include_opportunity=True) for decision in decisions}) == 1
    verdicts = Counter(str(row.get("ai_verdict")) for row in rows)
    return {
        "cohort": name,
        "sessions": sorted({str(row.get("session")) for row in rows}),
        "num_candidates": len(rows),
        "num_entry_family": sum(row.get("task") == "wearer_entry" for row in rows),
        "num_int_family": sum(row.get("task") == "int_yield" for row in rows),
        "num_positive": verdicts["POSITIVE"],
        "num_hard_negative": verdicts["HARD_NEGATIVE"],
        "num_exclude": verdicts["EXCLUDE"],
        "num_scoring_eligible_prelabels": verdicts["POSITIVE"] + verdicts["HARD_NEGATIVE"],
        "scoring_eligible_prelabel_percent": round(
            100 * (verdicts["POSITIVE"] + verdicts["HARD_NEGATIVE"]) / len(rows), 2
        ) if rows else None,
        "mean_ai_confidence": round(statistics.fmean(confidences), 4) if confidences else None,
        "median_ai_confidence": round(statistics.median(confidences), 4) if confidences else None,
        "num_rows_with_two_passes": pass_rows,
        "num_two_pass_semantic_agreement": semantic_agree,
        "two_pass_semantic_agreement_percent": round(100 * semantic_agree / pass_rows, 2) if pass_rows else None,
        "num_two_pass_full_decision_agreement": full_agree,
        "two_pass_full_decision_agreement_percent": round(100 * full_agree / pass_rows, 2) if pass_rows else None,
        "num_human_reviewed": sum(row.get("human_status") != "UNREVIEWED" for row in rows),
    }


def _write_csv(path: Path, fieldnames: list[str], rows: list[dict[str, Any]]) -> None:
    with path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames, lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)


def summarize(paths: list[Path], output_dir: Path) -> dict[str, Any]:
    all_rows: list[dict[str, Any]] = []
    for path in paths:
        all_rows.extend(_read_jsonl(path))
    if not all_rows:
        raise ValueError("No rows found")

    duplicate_uids = [uid for uid, count in Counter(row.get("uid") for row in all_rows).items() if count > 1]
    if duplicate_uids:
        raise ValueError(f"Duplicate uid across inputs: {duplicate_uids[:5]}")

    sessions = sorted({str(row.get("session")) for row in all_rows})
    by_session = {session: [row for row in all_rows if row.get("session") == session] for session in sessions}
    test_rows = [row for row in all_rows if row.get("session") in {"Session_1", "Session_2", "Session_3"}]
    session4_rows = by_session.get("Session_4", [])
    cohorts: dict[str, list[dict[str, Any]]] = {**by_session}
    if test_rows:
        cohorts["Test pooled (Sessions 1-3)"] = test_rows
    if session4_rows:
        cohorts["Session 4 audit (train split)"] = session4_rows
    cohorts["All supplied (descriptive only)"] = all_rows

    metadata = {
        "protocol_versions": sorted({str(row.get("protocol_version")) for row in all_rows}),
        "schema_versions": sorted({str(row.get("schema_version")) for row in all_rows}),
        "prompt_versions": sorted({str((row.get("llm_adjudication") or {}).get("prompt_version")) for row in all_rows}),
        "models": sorted({str((row.get("llm_adjudication") or {}).get("model")) for row in all_rows}),
        "passes": sorted({str((row.get("llm_adjudication") or {}).get("passes")) for row in all_rows}),
        "human_statuses": dict(sorted(Counter(str(row.get("human_status")) for row in all_rows).items())),
    }
    for key in ("protocol_versions", "schema_versions", "prompt_versions", "models", "passes"):
        if len(metadata[key]) != 1:
            raise ValueError(f"Cannot aggregate mixed {key}: {metadata[key]}")
    summary = {
        "artifact_type": "AI_PRELABEL_DESCRIPTIVE_STATISTICS_NOT_GROUND_TRUTH",
        "warning": "All supplied rows require human adjudication. Do not place these counts in the manuscript table labeled after human adjudication.",
        "inputs": [str(path) for path in paths],
        "metadata": metadata,
        "cohorts": [_cohort_summary(name, rows) for name, rows in cohorts.items()],
    }

    output_dir.mkdir(parents=True, exist_ok=True)
    (output_dir / "summary.json").write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
    cohort_rows = summary["cohorts"]
    _write_csv(output_dir / "cohort_summary.csv", list(cohort_rows[0]), cohort_rows)

    taxonomy_rows: list[dict[str, Any]] = []
    source_rows: list[dict[str, Any]] = []
    for name, rows in cohorts.items():
        total = len(rows)
        dimensions = {
            "verdict": Counter(str(row.get("ai_verdict")) for row in rows),
            "positive_event": Counter(str(row.get("ai_event_type")) for row in rows if row.get("ai_verdict") == "POSITIVE"),
            "hard_negative_subtype": Counter(str(row.get("ai_subtype")) for row in rows if row.get("ai_verdict") == "HARD_NEGATIVE"),
            "exclude_reason": Counter(str(row.get("ai_exclude_reason")) for row in rows if row.get("ai_verdict") == "EXCLUDE"),
            "exclude_group": Counter(str(_exclude_group(row)) for row in rows if row.get("ai_verdict") == "EXCLUDE"),
        }
        for dimension, counts in dimensions.items():
            for label, count in sorted(counts.items()):
                taxonomy_rows.append({
                    "cohort": name, "dimension": dimension, "label": label,
                    "count": count, "percent_of_all_candidates": round(100 * count / total, 2),
                })
        for source, count in sorted(Counter(str(row.get("candidate_source")) for row in rows).items()):
            source_rows.append({
                "cohort": name, "candidate_source": source, "count": count,
                "percent_of_all_candidates": round(100 * count / total, 2),
            })
    _write_csv(
        output_dir / "taxonomy_counts.csv",
        ["cohort", "dimension", "label", "count", "percent_of_all_candidates"],
        taxonomy_rows,
    )
    _write_csv(
        output_dir / "candidate_source_counts.csv",
        ["cohort", "candidate_source", "count", "percent_of_all_candidates"],
        source_rows,
    )

    train_label = "Session 4 audit (train split)"
    test_label = "Test pooled (Sessions 1-3)"
    table_lines = [
        "% AI pre-label audit only; do not substitute for the human-adjudicated manuscript table.",
        "\\begin{tabular}{lrr}",
        "\\hline",
        "AI pre-label outcome & Session 4 audit & Test Sessions 1--3 \\\\",
        "\\hline",
    ]
    for label, dimension, value in MANUSCRIPT_ROWS:
        train_count = _count(cohorts.get(train_label, []), dimension, value)
        test_count = _count(cohorts.get(test_label, []), dimension, value)
        table_lines.append(f"{label} & {train_count} & {test_count} \\\\")
    table_lines.extend(["\\hline", "\\end{tabular}", ""])
    (output_dir / "manuscript_ai_prelabel_table.tex").write_text("\n".join(table_lines), encoding="utf-8")

    report_lines = [
        "# EasyCom event AI pre-label audit",
        "",
        "> These are DeepSeek pre-labels awaiting mandatory human adjudication, not ground truth and not final manuscript annotation counts.",
        "",
        f"Inputs contain {len(all_rows):,} unique candidates from {', '.join(sessions)}. "
        f"Protocol: `{', '.join(metadata['protocol_versions'])}`; prompt: `{', '.join(metadata['prompt_versions'])}`.",
        "",
        "## Cohort overview",
        "",
        "| Cohort | Candidates | Positive | Hard negative | Exclude | Prelabel scoring-eligible | Two-pass semantic/full agreement | Human reviewed |",
        "|---|---:|---:|---:|---:|---:|---:|---:|",
    ]
    for item in cohort_rows:
        report_lines.append(
            f"| {item['cohort']} | {item['num_candidates']} | {item['num_positive']} | "
            f"{item['num_hard_negative']} | {item['num_exclude']} | "
            f"{item['num_scoring_eligible_prelabels']} ({item['scoring_eligible_prelabel_percent']:.2f}%) | "
            f"{item['num_two_pass_semantic_agreement']}/{item['num_rows_with_two_passes']} "
            f"({item['two_pass_semantic_agreement_percent']:.2f}%) / "
            f"{item['num_two_pass_full_decision_agreement']}/{item['num_rows_with_two_passes']} "
            f"({item['two_pass_full_decision_agreement_percent']:.2f}%) | {item['num_human_reviewed']} |"
        )
    report_lines.extend([
        "",
        "The official EasyCom test sessions (1--3) are pooled only for descriptive reporting; these AI pre-labels must not be used to tune thresholds, taxonomy, prompts, or model choices. Session 4 is shown separately as a train-split audit and is not the complete Sessions 4--12 training corpus.",
        "",
        "Detailed counts are in `taxonomy_counts.csv` and `candidate_source_counts.csv`. "
        "`manuscript_ai_prelabel_table.tex` mirrors the manuscript row order but is deliberately labeled as an AI audit; retain the manuscript TODOs until human adjudication is frozen.",
        "",
        "## Reproduce",
        "",
        "```bash",
        "python3 summarize_easycom_event_prelabels.py \\",
    ])
    for index, path in enumerate(paths):
        suffix = " \\" if index < len(paths) - 1 else " \\"
        report_lines.append(f"  --input {path}{suffix}")
    report_lines.extend([f"  --output-dir {output_dir}", "```", ""])
    (output_dir / "README.md").write_text("\n".join(report_lines), encoding="utf-8")
    return summary


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", action="append", type=Path, required=True)
    parser.add_argument("--output-dir", type=Path, required=True)
    args = parser.parse_args()
    summarize(args.input, args.output_dir)


if __name__ == "__main__":
    main()