| """최종 selector의 overmerge를 이웃 truth·OCR family·Tray penalty 기준으로 분해한다.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| from collections import Counter |
| from datetime import datetime, timezone |
| import json |
| from pathlib import Path |
| import sys |
| from typing import Any |
|
|
| PROJECT_ROOT = Path(__file__).parents[1] |
| SOURCE_ROOT = PROJECT_ROOT / "src" |
| for path in (PROJECT_ROOT, SOURCE_ROOT): |
| if str(path) not in sys.path: |
| sys.path.insert(0, str(path)) |
|
|
| from math_grid_drawer.research.cross_visual import CrossVisualModel |
| from math_grid_drawer.research.equality_visual import EqualityVisualModel |
| from math_grid_drawer.research.segmentation_lattice import ( |
| LATTICE_FEATURE_NAMES, |
| select_lattice_partition, |
| ) |
| from scripts.crohme_lattice_common import load_cached_split, writer_fit_validation |
| from scripts.evaluate_crohme_gt_free_grouping import _truth_partition |
| from scripts.evaluate_crohme_lattice_ocr_fusion import _fit_geometry |
| from scripts.evaluate_crohme_structure_presence import _truth_structures |
| from scripts.evaluate_crohme_tray_joint_selector import _prepared_signals, _weighted |
|
|
|
|
| def _parse_args() -> argparse.Namespace: |
| """필요 변수: 공식 test·cache·full selector head. 작동 원리: 최종 overmerge 감사 CLI를 만든다.""" |
|
|
| parser = argparse.ArgumentParser(description="Audit Math Ink 0.6 local-baseline overmerge") |
| parser.add_argument( |
| "--train-root", type=Path, |
| default=PROJECT_ROOT / "research/data/R_noncommercial/ICFHR_package/CROHME2012_data/trainData", |
| ) |
| parser.add_argument( |
| "--test-root", type=Path, |
| default=PROJECT_ROOT / "research/data/R_noncommercial/ICFHR_package/CROHME2012_data/testDataGT", |
| ) |
| parser.add_argument( |
| "--cache-dir", type=Path, |
| default=PROJECT_ROOT / "research/runs/crohme_lattice_ocr_cache_v2_20260722", |
| ) |
| parser.add_argument( |
| "--bundle", type=Path, |
| default=Path(r"research\runs\aiflow_ocr_05_dual_trajectory_3seed_20260720\bundle.manifest.json"), |
| ) |
| parser.add_argument( |
| "--cross-model", type=Path, |
| default=PROJECT_ROOT / "research/runs/crohme_cross_visual_loop3_polyline_20260722/cross_visual.json", |
| ) |
| parser.add_argument( |
| "--equality-model", type=Path, |
| default=PROJECT_ROOT / "research/runs/crohme_equality_visual_loop1_20260722/equality_visual.json", |
| ) |
| parser.add_argument("--profile", default="median_height_32") |
| parser.add_argument("--output", type=Path, required=True) |
| return parser.parse_args() |
|
|
|
|
| def main() -> None: |
| """필요 변수: gap40·family6 보호 selector. 작동 원리: overmerge candidate와 침범 truth를 1:1 연결한다.""" |
|
|
| args = _parse_args() |
| fit, _validation = writer_fit_validation(args.train_root, args.profile) |
| geometry_model = _fit_geometry(fit) |
| equality_model = EqualityVisualModel.load(args.equality_model) |
| cross_model = CrossVisualModel.load(args.cross_model) |
| samples, cached = load_cached_split( |
| args.test_root, |
| args.cache_dir, |
| split="official_test", |
| profile=args.profile, |
| bundle=args.bundle, |
| version=2, |
| ) |
| prepared = _prepared_signals( |
| samples, |
| cached, |
| geometry_model, |
| equality_model=equality_model, |
| cross_model=cross_model, |
| cross_gap_ratio=0.40, |
| multistroke_family_boost=6.0, |
| ) |
| weighted = _weighted( |
| prepared, |
| tray_weight=4.0, |
| symbol_weight=4.0, |
| fraction_weight=8.0, |
| infix_weight=8.0, |
| ) |
| path_by_id = {path.stem: path for path in sorted(args.test_root.rglob("*.inkml"))} |
| truth_labels: Counter[str] = Counter() |
| candidate_labels: Counter[str] = Counter() |
| candidate_families: Counter[str] = Counter() |
| invaded_pairs: Counter[str] = Counter() |
| structure_counts: Counter[str] = Counter() |
| fraction_penalty_counts: Counter[str] = Counter() |
| local_penalty_counts: Counter[str] = Counter() |
| rows: list[dict[str, Any]] = [] |
| unique_bad_candidates: dict[tuple[str, tuple[int, ...]], dict[str, Any]] = {} |
| correct_multistroke_rows: list[dict[str, Any]] = [] |
| for sample, row in zip(samples, weighted, strict=True): |
| truth_groups, labels = _truth_partition(sample, "aiflow_geometry") |
| label_by_group = dict(zip(truth_groups, (str(value) for value in labels), strict=True)) |
| predicted = set(select_lattice_partition( |
| row["candidates"], row["logits"], row["stroke_count"], group_bias=-2.0, |
| )) |
| candidate_index = { |
| frozenset(int(value) for value in candidate["source_indices"]): index |
| for index, candidate in enumerate(row["candidates"]) |
| } |
| families = row.get("ocr_families") or [""] * len(row["candidates"]) |
| structures = _truth_structures(path_by_id[sample["sample_id"]]) |
| for truth, truth_label in label_by_group.items(): |
| if truth in predicted: |
| if len(truth) > 1: |
| index = candidate_index[truth] |
| correct_multistroke_rows.append({ |
| "sample_id": sample["sample_id"], |
| "truth_label": truth_label, |
| "truth_group": sorted(truth), |
| "candidate_label": str(row["ocr_labels"][index]), |
| "candidate_family": str(families[index]), |
| "features": { |
| "ocr_top1": float( |
| row["features"][index][len(LATTICE_FEATURE_NAMES)] |
| ), |
| "merge_top1_gain": float( |
| row["features"][index][len(LATTICE_FEATURE_NAMES) + 6] |
| ), |
| "merge_entropy_gain": float( |
| row["features"][index][len(LATTICE_FEATURE_NAMES) + 7] |
| ), |
| "pair_gap_max": float(row["features"][index][12]), |
| }, |
| }) |
| continue |
| overmerged = [ |
| group for group in predicted |
| if group & truth and bool(group - truth) |
| ] |
| for group in overmerged: |
| index = candidate_index[group] |
| candidate_label = str(row["ocr_labels"][index]) |
| candidate_family = str(families[index]) |
| invaded = [ |
| other_label |
| for other_group, other_label in label_by_group.items() |
| if other_group != truth and other_group & group |
| ] |
| truth_labels[truth_label] += 1 |
| candidate_labels[candidate_label] += 1 |
| candidate_families[candidate_family] += 1 |
| for other_label in invaded: |
| invaded_pairs[f"{truth_label} -> {other_label}"] += 1 |
| for structure in structures or {"plain"}: |
| structure_counts[structure] += 1 |
| fraction_penalty = float(row["fraction_penalty"][index]) |
| fraction_penalty_counts[ |
| "nonzero" if fraction_penalty > 0.0 else "zero" |
| ] += 1 |
| raw_local_penalty = float(row["raw_local_baseline_penalty"][index]) |
| local_penalty = float(row["local_baseline_penalty"][index]) |
| if local_penalty > 0.0: |
| local_penalty_counts["effective_nonzero"] += 1 |
| elif raw_local_penalty > 0.0: |
| local_penalty_counts["protected_by_positive_signal"] += 1 |
| else: |
| local_penalty_counts["not_detected"] += 1 |
| covered_truth = [ |
| other_group for other_group in truth_groups if other_group & group |
| ] |
| replacement_scores = [ |
| float(row["logits"][candidate_index[other_group]]) |
| for other_group in covered_truth if other_group in candidate_index |
| ] |
| oracle_margin = ( |
| float(row["logits"][index]) - sum(replacement_scores) |
| + 2.0 * (len(replacement_scores) - 1) |
| if len(replacement_scores) == len(covered_truth) else None |
| ) |
| detail = { |
| "sample_id": sample["sample_id"], |
| "structures": sorted(structures), |
| "truth_label": truth_label, |
| "truth_group": sorted(truth), |
| "candidate_group": sorted(group), |
| "candidate_label": candidate_label, |
| "candidate_family": candidate_family, |
| "invaded_truth_labels": invaded, |
| "fraction_penalty": fraction_penalty, |
| "raw_local_baseline_penalty": raw_local_penalty, |
| "local_baseline_penalty": local_penalty, |
| "oracle_truth_partition_margin": oracle_margin, |
| "tray_signal": float(row["tray_signal"][index]), |
| "symbol_signal": float(row["symbol_signal"][index]), |
| "infix_signal": float(row["infix_signal"][index]), |
| "score": float(row["logits"][index]), |
| "geometry": { |
| "width_ref": float(row["features"][index][2]), |
| "height_ref": float(row["features"][index][3]), |
| "aspect_log": float(row["features"][index][4]), |
| "temporal_span": float(row["features"][index][5]), |
| "pair_gap_max": float(row["features"][index][12]), |
| }, |
| "ocr_features": { |
| "ocr_top1": float( |
| row["features"][index][len(LATTICE_FEATURE_NAMES)] |
| ), |
| "merge_top1_gain": float( |
| row["features"][index][len(LATTICE_FEATURE_NAMES) + 6] |
| ), |
| "merge_entropy_gain": float( |
| row["features"][index][len(LATTICE_FEATURE_NAMES) + 7] |
| ), |
| }, |
| } |
| rows.append(detail) |
| unique_bad_candidates[(sample["sample_id"], tuple(sorted(group)))] = detail |
| report = { |
| "experiment": "R-MATH-INK-06-LOCAL-BASELINE-OVERMERGE-AUDIT-001", |
| "generated_at": datetime.now(timezone.utc).isoformat(), |
| "configuration": { |
| "cross_gap_ratio": 0.40, |
| "multistroke_family_boost": 6.0, |
| }, |
| "overmerge_events": len(rows), |
| "truth_labels": truth_labels.most_common(), |
| "candidate_labels": candidate_labels.most_common(), |
| "candidate_families": candidate_families.most_common(), |
| "invaded_pairs": invaded_pairs.most_common(), |
| "structures": structure_counts.most_common(), |
| "fraction_penalty": dict(fraction_penalty_counts), |
| "local_baseline_penalty": dict(local_penalty_counts), |
| "unique_bad_candidate_count": len(unique_bad_candidates), |
| "unique_bad_candidates": list(unique_bad_candidates.values()), |
| "correct_multistroke_rows": correct_multistroke_rows, |
| "rows": rows, |
| "track": "R_noncommercial_only", |
| "product_validation": False, |
| } |
| args.output.parent.mkdir(parents=True, exist_ok=True) |
| args.output.write_text( |
| json.dumps(report, ensure_ascii=False, indent=2) + "\n", |
| encoding="utf-8", |
| ) |
| print(json.dumps({ |
| key: value for key, value in report.items() |
| if key not in {"rows", "unique_bad_candidates", "correct_multistroke_rows"} |
| }, ensure_ascii=False, indent=2)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
