| |
| """Aggregate focused KSL+CASL+NSL experiment results.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import csv |
| import json |
| import statistics |
| from pathlib import Path |
| from typing import Any, Iterable |
|
|
|
|
| POOLED_NAME = "E9.1 pooled pose-only" |
| PROPOSED_NAMES = { |
| "exp9_kcn_proposed_pose": "E9.2 proposed pose-only", |
| "exp9_kcn_proposed_rgb": "E9.3 proposed RGB/image-only", |
| "exp9_kcn_proposed_multimodal": "E9.4 proposed RGB+pose/image", |
| "exp9_kcn_v2_pose": "E9.5 research-v2 pose-only", |
| "exp9_kcn_v2_rgb": "E9.6 research-v2 RGB/image-only", |
| "exp9_kcn_v2_multimodal": "E9.7 research-v2 RGB+pose/image", |
| } |
|
|
|
|
| def as_float(value: Any) -> float | None: |
| try: |
| return float(value) |
| except (TypeError, ValueError): |
| return None |
|
|
|
|
| def mean_std(values: Iterable[Any]) -> tuple[float | None, float | None]: |
| nums = [float(v) for v in values if as_float(v) is not None] |
| if not nums: |
| return None, None |
| return sum(nums) / len(nums), statistics.stdev(nums) if len(nums) > 1 else 0.0 |
|
|
|
|
| def pct(value: Any) -> str: |
| num = as_float(value) |
| return "" if num is None else f"{num * 100:.2f}%" |
|
|
|
|
| def add_pooled_rows(rows: list[dict[str, Any]], root: Path) -> None: |
| for path in sorted((root / "exp9_kcn_pooled_pose").glob("**/pooled_arch_results.json")): |
| item = json.loads(path.read_text(encoding="utf-8")) |
| args = item.get("args", {}) |
| seed = args.get("seed", "") |
| metrics = item.get("final_metrics", {}) |
| base = { |
| "model": POOLED_NAME, |
| "run_name": "kcn_pooled_pose_only", |
| "seed": seed, |
| "split": "test", |
| "source": str(path), |
| "params": item.get("params", ""), |
| } |
| for key, metric in metrics.items(): |
| if not isinstance(metric, dict): |
| continue |
| if key.startswith("macro"): |
| rows.append( |
| { |
| **base, |
| "row_type": "macro", |
| "task_key": key, |
| "language_code": "", |
| "modality": "pose", |
| "level": "", |
| "top1": metric.get("accuracy"), |
| "top5": metric.get("top5"), |
| "macro_f1": metric.get("macro_f1"), |
| "n": "", |
| "num_classes": "", |
| } |
| ) |
| else: |
| rows.append( |
| { |
| **base, |
| "row_type": "task", |
| "task_key": f"pose_{key}", |
| "language_code": "casl" if key == "casl_si" else key, |
| "modality": "pose", |
| "level": "word" if key in {"casl_si", "casl", "ksl"} else "image", |
| "top1": metric.get("accuracy"), |
| "top5": metric.get("top5"), |
| "macro_f1": metric.get("macro_f1"), |
| "n": metric.get("n", ""), |
| "num_classes": (item.get("num_classes") or {}).get(key, ""), |
| } |
| ) |
|
|
|
|
| def add_proposed_rows(rows: list[dict[str, Any]], root: Path) -> None: |
| for dirname, model_name in PROPOSED_NAMES.items(): |
| for path in sorted((root / dirname).glob("*_results.json")): |
| item = json.loads(path.read_text(encoding="utf-8")) |
| base = { |
| "model": model_name, |
| "run_name": item.get("run_name", dirname), |
| "seed": item.get("seed", ""), |
| "source": str(path), |
| "params": item.get("params", ""), |
| } |
| for split in ("val", "test"): |
| for key, metric in (item.get(split) or {}).items(): |
| if not isinstance(metric, dict): |
| continue |
| rows.append( |
| { |
| **base, |
| "split": split, |
| "row_type": "macro" if key.startswith("macro_") else "task", |
| "task_key": key, |
| "language_code": metric.get("language_code", ""), |
| "modality": metric.get("modality", ""), |
| "level": metric.get("level", ""), |
| "top1": metric.get("top1"), |
| "top5": metric.get("top5"), |
| "macro_f1": metric.get("macro_f1"), |
| "n": metric.get("n", ""), |
| "num_classes": metric.get("num_classes", metric.get("n_tasks", "")), |
| } |
| ) |
|
|
|
|
| def summarize(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| groups: dict[tuple[str, str, str, str], list[dict[str, Any]]] = {} |
| for row in rows: |
| key = (row["model"], row["split"], row["row_type"], row["task_key"]) |
| groups.setdefault(key, []).append(row) |
| summary: list[dict[str, Any]] = [] |
| for (_model, _split, _row_type, _task), group in sorted(groups.items()): |
| first = group[0] |
| top1_mean, top1_std = mean_std(row.get("top1") for row in group) |
| f1_mean, f1_std = mean_std(row.get("macro_f1") for row in group) |
| top5_mean, top5_std = mean_std(row.get("top5") for row in group) |
| summary.append( |
| { |
| "model": first["model"], |
| "split": first["split"], |
| "row_type": first["row_type"], |
| "task_key": first["task_key"], |
| "language_code": first["language_code"], |
| "modality": first["modality"], |
| "level": first["level"], |
| "runs": len(group), |
| "top1_mean": top1_mean, |
| "top1_std": top1_std, |
| "top5_mean": top5_mean, |
| "top5_std": top5_std, |
| "macro_f1_mean": f1_mean, |
| "macro_f1_std": f1_std, |
| "n": first["n"], |
| "num_classes": first["num_classes"], |
| "params": first["params"], |
| } |
| ) |
| return summary |
|
|
|
|
| def write_csv(path: Path, rows: list[dict[str, Any]]) -> None: |
| if not rows: |
| return |
| with path.open("w", newline="", encoding="utf-8") as handle: |
| writer = csv.DictWriter(handle, fieldnames=list(rows[0].keys())) |
| writer.writeheader() |
| writer.writerows(rows) |
|
|
|
|
| def md_table(rows: list[dict[str, Any]]) -> str: |
| if not rows: |
| return "No rows yet." |
| lines = [ |
| "| Model | Split | Row | Task | Modality | Runs | Top-1 | Macro-F1 | Classes/Tasks |", |
| "|---|---|---|---|---:|---:|---:|---:|---:|", |
| ] |
| for row in rows: |
| lines.append( |
| f"| {row['model']} | {row['split']} | {row['row_type']} | {row['task_key']} | {row['modality']} | " |
| f"{row['runs']} | {pct(row['top1_mean'])} | {pct(row['macro_f1_mean'])} | {row.get('num_classes', '')} |" |
| ) |
| return "\n".join(lines) |
|
|
|
|
| def write_markdown(path: Path, summary: list[dict[str, Any]]) -> None: |
| test_macros = [r for r in summary if r["split"] == "test" and r["row_type"] == "macro"] |
| test_tasks = [r for r in summary if r["split"] == "test" and r["row_type"] == "task"] |
| val_macros = [r for r in summary if r["split"] == "val" and r["row_type"] == "macro"] |
| test_macros.sort(key=lambda r: (r["model"], r["task_key"])) |
| test_tasks.sort(key=lambda r: (r["model"], r["modality"], r["task_key"])) |
| val_macros.sort(key=lambda r: (r["model"], r["task_key"])) |
| text = "# KSL + CASL + NSL Focused Experiments\n\n" |
| text += "## Test Macro Rows\n\n" + md_table(test_macros) + "\n\n" |
| text += "## Test Task Rows\n\n" + md_table(test_tasks) + "\n\n" |
| text += "## Validation Macro Rows\n\n" + md_table(val_macros) + "\n" |
| path.write_text(text, encoding="utf-8") |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--results-root", type=Path, default=Path("results")) |
| parser.add_argument("--out-dir", type=Path, default=Path("results/exp9_kcn_focus_summary")) |
| return parser.parse_args() |
|
|
|
|
| def main() -> None: |
| args = parse_args() |
| args.out_dir.mkdir(parents=True, exist_ok=True) |
| rows: list[dict[str, Any]] = [] |
| add_pooled_rows(rows, args.results_root) |
| add_proposed_rows(rows, args.results_root) |
| if not rows: |
| raise SystemExit(f"No KCN focused results found under {args.results_root}") |
| summary = summarize(rows) |
| raw = args.out_dir / "kcn_focus_raw_rows.csv" |
| csv_path = args.out_dir / "kcn_focus_summary.csv" |
| json_path = args.out_dir / "kcn_focus_summary.json" |
| md_path = args.out_dir / "kcn_focus_summary.md" |
| write_csv(raw, rows) |
| write_csv(csv_path, summary) |
| json_path.write_text(json.dumps({"raw_rows": rows, "summary": summary}, indent=2), encoding="utf-8") |
| write_markdown(md_path, summary) |
| print(md_path.read_text(encoding="utf-8")) |
| print("Saved:") |
| print(" ", raw) |
| print(" ", csv_path) |
| print(" ", json_path) |
| print(" ", md_path) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
