"""최종 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()