mcm-26-b/project/q2/compare.py

"""问题二同案配对对比：最优 / 两个基线 / 网格参考，用同一评估器打分。"""

from __future__ import annotations

import math

from mcm26b.config import ProblemConfig
from mcm26b.planner import _worst_posterior_radius

from project.common import (Method, RESULTS_DIR, markdown_table, mean, ratio,
                            write_json, write_text)
from project.q2 import baseline, cases, optimal, reference
from project.q2.region import first_bearing_region

METHODS: tuple[Method, ...] = (optimal.METHOD, baseline.PERP, baseline.MEC,
                               reference.METHOD)
EVAL_WITNESS_SPACING_M = 60.0


def _evaluate(case: dict, config: ProblemConfig) -> list[dict]:
    region = first_bearing_region(case, config)
    station = (case["station"][0], case["station"][1])
    initial_radius = region.mec()[1]
    half_angle = (config.field.bearing_error_half_deg
                  + config.search.bearing_rounding_margin_deg)
    witnesses = region.witnesses(EVAL_WITNESS_SPACING_M)

    plans = [optimal.candidate(case, config),
             baseline.candidate_perp(case, config),
             baseline.candidate_mec(case, config),
             reference.candidate(case, config)]
    rows = []
    for plan in plans:
        q = (plan["position"][0], plan["position"][1])
        guarantee_distance = max(math.dist(q, vertex) for vertex in region.polygon)
        worst = _worst_posterior_radius(
            region, q, witnesses, half_angle,
            config.field.near_radius_m, config.search.polygon_sides)
        move_cost = math.dist(station, q) / config.cost.speed_mps
        rows.append({
            "case_id": case["id"],
            "method": plan["method"],
            "initial_radius_m": round(initial_radius, 4),
            "q": [round(q[0], 2), round(q[1], 2)],
            "guaranteed_reception": bool(guarantee_distance
                                         <= config.field.r_eff_min_m + 1e-6),
            "guarantee_distance_m": round(guarantee_distance, 3),
            "worst_posterior_radius_m": round(worst, 4),
            "improvement_m": round(initial_radius - worst, 4),
            "final_ratio": round(worst / initial_radius, 4) if initial_radius > 0 else None,
            "move_cost_s": round(move_cost, 4),
            "reason": plan["reason"],
        })
    return rows


def _aggregate(rows: list[dict], labels: list[str]) -> dict:
    summary: dict[str, dict] = {}
    initial = mean(r["initial_radius_m"] for r in rows) if rows else 0.0
    for label in labels:
        part = [r for r in rows if r["method"] == label]
        if not part:
            continue
        summary[label] = {
            "cases": len(part),
            "guaranteed_rate": ratio(sum(r["guaranteed_reception"] for r in part),
                                     len(part)),
            "worst_posterior_mean_m": mean(r["worst_posterior_radius_m"] for r in part),
            "worst_posterior_median_m": sorted(
                r["worst_posterior_radius_m"] for r in part)[len(part) // 2],
            "improvement_mean_m": mean(r["improvement_m"] for r in part),
            "final_ratio_mean": mean(r["final_ratio"] for r in part
                                     if r["final_ratio"] is not None),
            "move_cost_mean_s": mean(r["move_cost_s"] for r in part),
        }
    summary["_initial_radius_mean_m"] = initial
    return summary


def _render(rows: list[dict], aggregate: dict, labels: list[str]) -> str:
    headers = ["算例", "方法", "初始 ρ(m)", "第二点 q", "保证接收",
               "q 到 K₁ 最远(m)", "最坏后验 ρ'(m)", "收缩(m)", "移动(s)"]
    body = []
    order = {label: index for index, label in enumerate(labels)}
    for row in sorted(rows, key=lambda r: (r["case_id"], order.get(r["method"], 99))):
        body.append([
            row["case_id"], row["method"], _f(row["initial_radius_m"]),
            f"({row['q'][0]:.0f},{row['q'][1]:.0f})",
            "是" if row["guaranteed_reception"] else "否",
            _f(row["guarantee_distance_m"]), _f(row["worst_posterior_radius_m"]),
            _f(row["improvement_m"]), _f(row["move_cost_s"]),
        ])
    text = ["# 问题二：第二检测点选择对比", "",
            "初始 ρ 为首次示向度候选集 K₁ 的最小包围半径；最坏后验按"
            "目标边界采样 × 读数区间端点统一评估。", "",
            markdown_table(headers, body), "", "## 逐方法汇总", ""]
    agg_headers = ["方法", "保证接收率", "最坏后验均值(m)", "中位(m)",
                   "平均收缩(m)", "终值比 ρ'/ρ", "移动均值(s)"]
    agg_rows = []
    for label in labels:
        s = aggregate.get(label)
        if not s:
            continue
        agg_rows.append([label, f"{s['guaranteed_rate']*100:.1f}%",
                         _f(s["worst_posterior_mean_m"]),
                         _f(s["worst_posterior_median_m"]),
                         _f(s["improvement_mean_m"]),
                         _f(s["final_ratio_mean"], 3),
                         _f(s["move_cost_mean_s"])])
    text.append(markdown_table(agg_headers, agg_rows))
    text.append("")
    text.append(f"初始 ρ 均值：{aggregate['_initial_radius_mean_m']:.2f} m")
    return "\n".join(text) + "\n"


def _f(value, digits: int = 2) -> str:
    if value is None:
        return "—"
    if isinstance(value, bool):
        return "是" if value else "否"
    if isinstance(value, float):
        return f"{value:.{digits}f}"
    return str(value)


def run(quick: bool = False) -> dict:
    config = ProblemConfig()
    case_list = cases.random_cases(8 if quick else 30, config=config)
    rows = [row for case in case_list for row in _evaluate(case, config)]
    labels = [m.key for m in METHODS]
    aggregate = _aggregate(rows, labels)
    payload = {"question": 2, "methods": [m.__dict__ for m in METHODS],
               "cases": case_list, "rows": rows, "aggregate": aggregate}
    write_json(RESULTS_DIR / "q2.json", payload)
    write_text(RESULTS_DIR / "q2.md", _render(rows, aggregate, labels))
    return payload