125 lines
5.4 KiB
Python
125 lines
5.4 KiB
Python
"""问题四批量仿真基准：固定误差场下统计后备清除成本与总请求量。
|
|
|
|
逐局在进程内驱动 DirectionalExtension（本地模拟器；示向度误差场逐点
|
|
固定，与官方"同一地点误差不变"口径一致），记录 completed、measures、
|
|
clears、total_requests、fallback_clear_attempts、virtual_time_s 等指标，
|
|
落盘 JSON 并打印汇总表。用于：
|
|
- 验证后备清除成本（对跖探点修正后梯子命中率的现实校核）；
|
|
- 为论文问题四结果表提供可复现数字（种子、配置、逐局指标全部留档）。
|
|
|
|
用法示例：
|
|
uv run python scripts/benchmark_q4.py --seeds 0-15 --sources 10 --directional 3
|
|
uv run python scripts/benchmark_q4.py --seeds 20260910 --sources 16 --directional 6
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
from mcm26b.config import ProblemConfig
|
|
from mcm26b.evaluation import recompute_virtual_time
|
|
from mcm26b.planner import DirectionalExtension
|
|
from mcm26b.protocol.client import SimulatorClient
|
|
from mcm26b.sim import LocalSimTransport, LocalSimulator
|
|
from mcm26b.sim.scenes import mixed_scene
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def parse_seeds(spec: str) -> list[int]:
|
|
"""种子列表："0-15" 区间、"0,5,9" 枚举或单个值。"""
|
|
seeds: list[int] = []
|
|
for part in spec.split(","):
|
|
part = part.strip()
|
|
if "-" in part:
|
|
lo, hi = part.split("-", 1)
|
|
seeds.extend(range(int(lo), int(hi) + 1))
|
|
else:
|
|
seeds.append(int(part))
|
|
return seeds
|
|
|
|
|
|
def run_one(seed: int, count: int, directional: int, margin: float) -> dict:
|
|
config = replace(ProblemConfig(),
|
|
search=replace(ProblemConfig().search,
|
|
bearing_rounding_margin_deg=margin))
|
|
sources = mixed_scene(seed=seed, count=count, directional=directional)
|
|
simulator = LocalSimulator(sources, robot_id="BENCH", config=config)
|
|
client = SimulatorClient("BENCH", config=config,
|
|
transport=LocalSimTransport(simulator))
|
|
loop = DirectionalExtension(client, config)
|
|
summary = loop.run()
|
|
report = recompute_virtual_time(client.log.action_events())
|
|
truth = sorted(s.channel for s in sources)
|
|
return {
|
|
"seed": seed,
|
|
"sources": count,
|
|
"directional": directional,
|
|
"completed": summary["completed"],
|
|
"stop_reason": summary["stop_reason"],
|
|
"all_cleared": sorted(summary["cleared_channels"]) == truth,
|
|
"measures": summary["measures"],
|
|
"clears": summary["clears"],
|
|
"total_requests": summary["total_requests"],
|
|
"fallback_cells": [entry["cells"] for entry in summary["fallback_clears"]],
|
|
"fallback_attempts": summary["fallback_clear_attempts"],
|
|
"virtual_time_s": summary["virtual_time_s"],
|
|
"virtual_recompute_mismatches": len(report["mismatches"]),
|
|
"anomaly_count": len(summary["anomalies"]),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
parser.add_argument("--seeds", default="0-15",
|
|
help="种子，如 0-15 / 0,5,9 / 20260910")
|
|
parser.add_argument("--sources", type=int, default=10)
|
|
parser.add_argument("--directional", type=int, default=3)
|
|
parser.add_argument("--bearing-margin-deg", type=float, default=0.005)
|
|
parser.add_argument("--out", default="output/q4/benchmark.json")
|
|
args = parser.parse_args()
|
|
|
|
rows = [run_one(seed, args.sources, args.directional, args.bearing_margin_deg)
|
|
for seed in parse_seeds(args.seeds)]
|
|
completed = sum(1 for row in rows if row["completed"])
|
|
cleared = sum(1 for row in rows if row["all_cleared"])
|
|
summary = {
|
|
"config": {"sources": args.sources, "directional": args.directional,
|
|
"bearing_margin_deg": args.bearing_margin_deg},
|
|
"games": len(rows),
|
|
"completed": completed,
|
|
"all_cleared": cleared,
|
|
"measures_total": sum(row["measures"] for row in rows),
|
|
"clears_total": sum(row["clears"] for row in rows),
|
|
"total_requests_total": sum(row["total_requests"] for row in rows),
|
|
"fallback_attempts_total": sum(row["fallback_attempts"] for row in rows),
|
|
"fallback_attempts_max": max((row["fallback_attempts"] for row in rows),
|
|
default=0),
|
|
"virtual_time_s_max": max((row["virtual_time_s"] for row in rows),
|
|
default=0.0),
|
|
"mismatch_total": sum(row["virtual_recompute_mismatches"] for row in rows),
|
|
"anomaly_total": sum(row["anomaly_count"] for row in rows),
|
|
}
|
|
payload = {"summary": summary, "games": rows}
|
|
out_path = ROOT / args.out
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
out_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8")
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
header = ("seed", "done", "clear", "measures", "clears", "requests",
|
|
"fb_attempts", "virtual_s")
|
|
print("\t".join(header))
|
|
for row in rows:
|
|
print("\t".join(str(row[key]) for key in
|
|
("seed", "completed", "all_cleared", "measures", "clears",
|
|
"total_requests", "fallback_attempts", "virtual_time_s")))
|
|
print(f"\n结果已写入 {out_path}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|