"""최종 selector에서 구성획보다 약한 merge 후보의 must-not-link weight를 validation 선택한다.""" from __future__ import annotations import argparse 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 scripts.crohme_lattice_common import load_cache_for_samples, load_cached_split, writer_fit_validation from scripts.evaluate_crohme_lattice_ocr_fusion import _fit_geometry from scripts.evaluate_crohme_tray_joint_selector import _prepared_signals, _weighted from scripts.sweep_math_ink_06_multistroke_family_guard import _family_metrics06 from scripts.sweep_math_ink_06_x_grouping_guard import _target_metrics06 from scripts.train_crohme_segmentation_lattice_joint_selector import _metrics def _parse_args() -> argparse.Namespace: """필요 변수: CROHME split·cache·full selector head. 작동 원리: test 비개입 competition sweep CLI를 만든다.""" parser = argparse.ArgumentParser(description="Sweep Math Ink 0.6 component competition guard") 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("--maximum-x-regression-pp", type=float, default=1.0) parser.add_argument("--maximum-family-regression-pp", type=float, default=2.0) parser.add_argument("--maximum-pair-f1-regression-pp", type=float, default=0.25) parser.add_argument("--output", type=Path, required=True) return parser.parse_args() def _evaluate_weight06( samples: list[dict[str, Any]], prepared: list[dict[str, Any]], weight: float, ) -> dict[str, Any]: """필요 변수: 한 split의 cached signal·competition weight. 작동 원리: 전역/x/family 지표를 같은 partition에서 계산한다.""" weighted = _weighted( prepared, tray_weight=4.0, symbol_weight=4.0, fraction_weight=8.0, infix_weight=8.0, competition_weight=weight, ) return { "component_competition_weight": weight, "global": _metrics(weighted, -2.0), "behavior_targets": _target_metrics06( samples, weighted, group_bias=-2.0, ), "families": _family_metrics06(samples, weighted), } def _delta_pp06(candidate: float, reference: float) -> float: """필요 변수: 후보·기준 비율. 작동 원리: 채택 판단용 percentage-point 차이를 반환한다.""" return (candidate - reference) * 100.0 def main() -> None: """필요 변수: writer-validation·official test. 작동 원리: 보호 gate 안 exact winner와 기준만 test에서 비교한다.""" 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) validation_cache = load_cache_for_samples( validation, args.cache_dir, split="validation", profile=args.profile, bundle=args.bundle, version=2, ) validation_prepared = _prepared_signals( validation, validation_cache, geometry_model, equality_model=equality_model, cross_model=cross_model, cross_gap_ratio=0.40, multistroke_family_boost=6.0, ) trials = [ _evaluate_weight06(validation, validation_prepared, weight) for weight in (0.0, 1.0, 2.0, 3.0, 4.0, 6.0, 8.0, 12.0) ] reference = trials[0] minimum_x = ( reference["behavior_targets"]["x"]["grouping_recall"] - args.maximum_x_regression_pp / 100.0 ) minimum_family = ( reference["families"]["grouping_recall"] - args.maximum_family_regression_pp / 100.0 ) minimum_pair_f1 = ( reference["global"]["pair_f1"] - args.maximum_pair_f1_regression_pp / 100.0 ) eligible = [ row for row in trials if ( row["behavior_targets"]["x"]["grouping_recall"] >= minimum_x and row["families"]["grouping_recall"] >= minimum_family and row["global"]["pair_f1"] >= minimum_pair_f1 ) ] winner = max(eligible, key=lambda row: ( row["global"]["exact_partition"], row["global"]["pair_f1"], row["global"]["exact_group_recall"], -row["component_competition_weight"], )) test, test_cache = load_cached_split( args.test_root, args.cache_dir, split="official_test", profile=args.profile, bundle=args.bundle, version=2, ) test_prepared = _prepared_signals( test, test_cache, geometry_model, equality_model=equality_model, cross_model=cross_model, cross_gap_ratio=0.40, multistroke_family_boost=6.0, ) official_reference = _evaluate_weight06(test, test_prepared, 0.0) official_winner = _evaluate_weight06( test, test_prepared, float(winner["component_competition_weight"]), ) deltas = { "exact_partition_pp": _delta_pp06( official_winner["global"]["exact_partition"], official_reference["global"]["exact_partition"], ), "pair_f1_pp": _delta_pp06( official_winner["global"]["pair_f1"], official_reference["global"]["pair_f1"], ), "x_grouping_pp": _delta_pp06( official_winner["behavior_targets"]["x"]["grouping_recall"], official_reference["behavior_targets"]["x"]["grouping_recall"], ), "family_grouping_pp": _delta_pp06( official_winner["families"]["grouping_recall"], official_reference["families"]["grouping_recall"], ), } adopted = bool( float(winner["component_competition_weight"]) > 0.0 and deltas["exact_partition_pp"] > 0.0 and deltas["pair_f1_pp"] >= -args.maximum_pair_f1_regression_pp and deltas["x_grouping_pp"] >= -args.maximum_x_regression_pp and deltas["family_grouping_pp"] >= -args.maximum_family_regression_pp ) report = { "experiment": "R-MATH-INK-06-COMPONENT-COMPETITION-GUARD-001", "generated_at": datetime.now(timezone.utc).isoformat(), "selection_contract": { "split": "CROHME trainData writer-validation only", "cross_gap_ratio": 0.40, "multistroke_family_boost": 6.0, "maximum_x_regression_pp": args.maximum_x_regression_pp, "maximum_family_regression_pp": args.maximum_family_regression_pp, "maximum_pair_f1_regression_pp": args.maximum_pair_f1_regression_pp, }, "reference_validation": reference, "winner_validation": winner, "trials": trials, "official_test_reference": official_reference, "official_test_winner": official_winner, "official_test_deltas": deltas, "decision": { "adopted": adopted, "selected_component_competition_weight": ( float(winner["component_competition_weight"]) if adopted else 0.0 ), "reason": ( "exact partition improves within x/family/pair-F1 guards" if adopted else "official adoption gate failed" ), }, "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 != "trials"}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()