309 lines
13 KiB
Python
309 lines
13 KiB
Python
"""对比框架公共层：方法描述、指标聚合、结果落盘、离线闭环装配。
|
|
|
|
设计口径
|
|
--------
|
|
- **最优实现** = 复用 ``src/mcm26b`` 中已由测试固化的生产实现；
|
|
- **基线实现** = 同题下更朴素但可独立运行的参照实现（不得共享最优实现的
|
|
关键步骤），用于量化"生产实现好在哪里、差多少"；
|
|
- **参考解** = 与分析算法相互独立的数值/枚举标准，用来判定谁对谁错。
|
|
|
|
第三、四问是闭环问题，统一走 ``LocalSimulator`` 内存回环（与官方 HTTP
|
|
协议同一边界），真值只在评测机侧，策略端只看响应。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import statistics
|
|
from dataclasses import dataclass, replace
|
|
from pathlib import Path
|
|
from time import perf_counter
|
|
|
|
from mcm26b.config import ProblemConfig
|
|
from mcm26b.evaluation import recompute_virtual_time
|
|
from mcm26b.harness import LocalSimTransport, LocalSimulator
|
|
from mcm26b.protocol.client import SimulatorClient
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
RESULTS_DIR = Path(__file__).resolve().parent / "results"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Method:
|
|
"""一个可对比方法：key 用作图表列名，kind 用于区分最优/基线/参考。"""
|
|
|
|
key: str
|
|
name: str
|
|
kind: str # optimal | baseline | reference
|
|
note: str = ""
|
|
|
|
|
|
def problem_config(*, persistent_search: bool = False,
|
|
margin_deg: float | None = None) -> ProblemConfig:
|
|
"""对比口径配置：默认关闭"扫到限时"的持续复扫（对齐首次找全证明）。"""
|
|
base = ProblemConfig()
|
|
kwargs: dict = {"persistent_search": persistent_search}
|
|
if margin_deg is not None:
|
|
kwargs["bearing_rounding_margin_deg"] = margin_deg
|
|
return replace(base, search=replace(base.search, **kwargs))
|
|
|
|
|
|
# ---------------------------------------------------------------- 结果落盘
|
|
def write_json(path: Path, payload) -> Path:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2,
|
|
allow_nan=False) + "\n", encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def write_text(path: Path, text: str) -> Path:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(text, encoding="utf-8")
|
|
return path
|
|
|
|
|
|
# ---------------------------------------------------------------- 指标工具
|
|
def mean(values) -> float:
|
|
values = list(values)
|
|
return round(statistics.fmean(values), 4) if values else 0.0
|
|
|
|
|
|
def median(values) -> float:
|
|
values = list(values)
|
|
return round(statistics.median(values), 4) if values else 0.0
|
|
|
|
|
|
def ratio(hit: int, total: int) -> float:
|
|
return round(hit / total, 4) if total else 0.0
|
|
|
|
|
|
def relative_change(value: float, reference: float) -> float | None:
|
|
if reference == 0:
|
|
return None
|
|
return round(value / reference - 1.0, 4)
|
|
|
|
|
|
def markdown_table(headers: list[str], rows: list[list[str]]) -> str:
|
|
lines = ["| " + " | ".join(headers) + " |",
|
|
"| " + " | ".join("---" for _ in headers) + " |"]
|
|
for row in rows:
|
|
lines.append("| " + " | ".join(str(cell) for cell in row) + " |")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _num(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_loop_episode(loop_cls, scene_spec, config: ProblemConfig) -> dict:
|
|
"""在冻结场景上跑一局闭环，返回逐局指标行（与 ``scripts/run_suite.py`` 同口径）。"""
|
|
sources = scene_spec.sources(config) # 真值：仅测试端可见
|
|
simulator = LocalSimulator(sources, robot_id="COMPARE", config=config,
|
|
error_seed=scene_spec.error_seed)
|
|
client = SimulatorClient("COMPARE", config=config,
|
|
transport=LocalSimTransport(simulator))
|
|
start = perf_counter()
|
|
loop = loop_cls(client, config)
|
|
summary = loop.run()
|
|
compute_wall_s = perf_counter() - start
|
|
report = recompute_virtual_time(client.log.action_events())
|
|
|
|
truth = sorted(source.channel for source in sources)
|
|
cleared = sorted(summary["cleared_channels"])
|
|
absent = sorted(summary["absent_channels"])
|
|
misjudged = sorted(set(absent) & set(truth))
|
|
valid = (cleared == truth and not misjudged
|
|
and summary["exit_reason"] == "user_exit"
|
|
and not summary["anomalies"]
|
|
and not report["mismatches"])
|
|
return {
|
|
"scene_id": scene_spec.scene_id,
|
|
"problem": scene_spec.problem,
|
|
"kind": scene_spec.kind,
|
|
"sources": len(truth),
|
|
"directional": scene_spec.directional,
|
|
"cleared": len(cleared),
|
|
"cleared_ratio": ratio(len(cleared), len(truth)),
|
|
"cleared_matches_truth": cleared == truth,
|
|
"absent_misjudged": len(misjudged),
|
|
"virtual_time_s": round(summary["virtual_time_s"], 3),
|
|
"accounted_virtual_time_s": (
|
|
round(summary["accounted_virtual_time_s"], 3)
|
|
if summary.get("accounted_virtual_time_s") is not None else None),
|
|
"measures": summary["measures"],
|
|
"clears": summary["clears"],
|
|
"clear_success": summary["clear_success"],
|
|
"total_requests": summary["total_requests"],
|
|
"move_distance_m": round(summary["move_distance_m"], 2),
|
|
"fallback_clear_attempts": summary.get("fallback_clear_attempts", 0),
|
|
"coverage_points": summary.get("coverage_points", 0),
|
|
"stop_reason": summary["stop_reason"],
|
|
"exit_reason": summary["exit_reason"],
|
|
"anomaly_count": len(summary["anomalies"]),
|
|
"virtual_recompute_mismatches": len(report["mismatches"]),
|
|
"compute_wall_s": round(compute_wall_s, 3),
|
|
"valid": valid,
|
|
}
|
|
|
|
|
|
# 第三/四问统一汇报列（顺序即 Markdown 列序）
|
|
EPISODE_COLUMNS = (
|
|
"cleared_ratio", "cleared_matches_truth", "absent_misjudged",
|
|
"virtual_time_s", "measures", "total_requests", "move_distance_m",
|
|
"fallback_clear_attempts", "compute_wall_s", "valid",
|
|
)
|
|
|
|
|
|
def aggregate_episodes(rows: list[dict], labels: list[str],
|
|
base_label: str | None = None) -> dict:
|
|
"""逐方法汇总 + 相对基线的配对变化（按 scene_id 对齐）。"""
|
|
summary: dict[str, dict] = {}
|
|
for label in labels:
|
|
part = [row for row in rows if row["variant"] == label]
|
|
if not part:
|
|
continue
|
|
summary[label] = {
|
|
"episodes": len(part),
|
|
"valid_episodes": sum(1 for row in part if row["valid"]),
|
|
"cleared_ratio_mean": mean(row["cleared_ratio"] for row in part),
|
|
"cleared_matches_truth": sum(1 for row in part
|
|
if row["cleared_matches_truth"]),
|
|
"absent_misjudged_total": sum(row["absent_misjudged"] for row in part),
|
|
"virtual_time_mean_s": mean(row["virtual_time_s"] for row in part),
|
|
"virtual_time_median_s": median(row["virtual_time_s"] for row in part),
|
|
"virtual_time_max_s": round(max(row["virtual_time_s"] for row in part), 3),
|
|
"measures_mean": mean(row["measures"] for row in part),
|
|
"total_requests_mean": mean(row["total_requests"] for row in part),
|
|
"move_distance_mean_m": mean(row["move_distance_m"] for row in part),
|
|
"fallback_clear_attempts_mean": mean(
|
|
row["fallback_clear_attempts"] for row in part),
|
|
"compute_wall_mean_s": mean(row["compute_wall_s"] for row in part),
|
|
}
|
|
paired: dict[str, dict] = {}
|
|
if base_label and len(labels) > 1:
|
|
base_rows = {row["scene_id"]: row for row in rows
|
|
if row["variant"] == base_label}
|
|
for label in labels:
|
|
if label == base_label:
|
|
continue
|
|
deltas, faster, valid_pairs = [], 0, 0
|
|
for row in rows:
|
|
if row["variant"] != label or row["scene_id"] not in base_rows:
|
|
continue
|
|
base_row = base_rows[row["scene_id"]]
|
|
reference = base_row["virtual_time_s"]
|
|
deltas.append(relative_change(row["virtual_time_s"], reference))
|
|
if not (row["valid"] and base_row["valid"]):
|
|
continue
|
|
valid_pairs += 1
|
|
if row["virtual_time_s"] < reference:
|
|
faster += 1
|
|
paired[f"{label}_vs_{base_label}"] = {
|
|
"mean_relative_change": mean(d for d in deltas if d is not None),
|
|
"median_relative_change": median(d for d in deltas if d is not None),
|
|
"faster_scenes": faster,
|
|
"compared": len(deltas),
|
|
"both_valid_scenes": valid_pairs,
|
|
}
|
|
return {"variants": summary, "paired": paired}
|
|
|
|
|
|
def episodes_markdown(title: str, rows: list[dict], labels: list[str],
|
|
base_label: str | None = None) -> str:
|
|
"""逐场景明细 + 汇总表。"""
|
|
headers = ["scene", "sources", "variant", "cleared", "T(s)", "检测",
|
|
"请求", "位移(m)", "后备清除", "墙钟(s)", "有效"]
|
|
body = []
|
|
for row in sorted(rows, key=lambda r: (r["scene_id"], labels.index(r["variant"]))):
|
|
body.append([
|
|
row["scene_id"], row["sources"], row["variant"],
|
|
f"{row['cleared']}/{row['sources']}",
|
|
_num(row["virtual_time_s"], 1), row["measures"],
|
|
row["total_requests"], _num(row["move_distance_m"], 1),
|
|
row["fallback_clear_attempts"], _num(row["compute_wall_s"], 2),
|
|
_num(row["valid"]),
|
|
])
|
|
text = [f"# {title}", "", "## 逐场景明细", "",
|
|
markdown_table(headers, body), "", "## 逐方法汇总", ""]
|
|
agg = aggregate_episodes(rows, labels, base_label)
|
|
agg_headers = ["方法", "有效局", "全清率", "全清对数", "缺席误判",
|
|
"T 均值(s)", "T 中位(s)", "T 最大(s)", "检测均值",
|
|
"请求均值", "位移均值(m)", "后备清除均值", "墙钟均值(s)"]
|
|
agg_rows = []
|
|
for label in labels:
|
|
s = agg["variants"].get(label)
|
|
if not s:
|
|
continue
|
|
agg_rows.append([
|
|
label, f"{s['valid_episodes']}/{s['episodes']}",
|
|
_num(s["cleared_ratio_mean"]), s["cleared_matches_truth"],
|
|
s["absent_misjudged_total"], _num(s["virtual_time_mean_s"], 1),
|
|
_num(s["virtual_time_median_s"], 1), _num(s["virtual_time_max_s"], 1),
|
|
_num(s["measures_mean"], 1), _num(s["total_requests_mean"], 1),
|
|
_num(s["move_distance_mean_m"], 1),
|
|
_num(s["fallback_clear_attempts_mean"], 2),
|
|
_num(s["compute_wall_mean_s"], 2),
|
|
])
|
|
text.append(markdown_table(agg_headers, agg_rows))
|
|
if agg["paired"]:
|
|
text += ["", "## 相对基线的配对变化（虚拟时间）", ""]
|
|
pair_rows = [[name, f"{v['mean_relative_change']*100:+.1f}%",
|
|
f"{v['median_relative_change']*100:+.1f}%",
|
|
f"{v['faster_scenes']}/{v['compared']}",
|
|
v["both_valid_scenes"]]
|
|
for name, v in agg["paired"].items()]
|
|
text.append(markdown_table(
|
|
["配对", "均值变化", "中位变化", "更快场景", "双方均有效"], pair_rows))
|
|
invalid = [name for name, value in agg["paired"].items()
|
|
if value["both_valid_scenes"] < value["compared"]]
|
|
if invalid:
|
|
text += ["",
|
|
"> 注：" + "、".join(invalid)
|
|
+ " 存在未通过有效性判定的配对；其虚拟时间仅作参考，"
|
|
"基线未完成全清任务时不构成「更快」证据"
|
|
"（以「双方均有效」列为准）。"]
|
|
return "\n".join(text) + "\n"
|
|
|
|
|
|
def json_safe(payload):
|
|
"""把 numpy 标量等转成 json 可写类型（防御式，非必需）。"""
|
|
return json.loads(json.dumps(payload, ensure_ascii=False, default=float))
|
|
|
|
|
|
# ---------------------------------------------------------------- 闭环批量执行
|
|
def execute_variants(scenes, variants, config: ProblemConfig) -> list[dict]:
|
|
"""对每个场景跑遍所有变体（串行；对比规模小，进程池收益有限且便于复现）。"""
|
|
rows = []
|
|
for scene in scenes:
|
|
for label, loop_cls in variants:
|
|
row = run_loop_episode(loop_cls, scene, config)
|
|
row["variant"] = label
|
|
rows.append(row)
|
|
return rows
|
|
|
|
|
|
def closed_loop_payload(question: int, suite_name: str, scenes, variants,
|
|
rows: list[dict], labels: list[str],
|
|
base_label: str, config: ProblemConfig) -> dict:
|
|
return {
|
|
"question": question,
|
|
"suite": suite_name,
|
|
"scenes": [scene.as_dict() for scene in scenes],
|
|
"variants": [{"label": label, "class": loop_cls.__name__}
|
|
for label, loop_cls in variants],
|
|
"config": {
|
|
"persistent_search": config.search.persistent_search,
|
|
"bearing_rounding_margin_deg": config.search.bearing_rounding_margin_deg,
|
|
"q4_coverage_plan": config.search.q4_coverage_plan,
|
|
},
|
|
"rows": rows,
|
|
"aggregate": aggregate_episodes(rows, labels, base_label),
|
|
}
|