149 lines
6.5 KiB
Python
149 lines
6.5 KiB
Python
"""统一入口：一次跑完 1-4 问的"最优 vs 基线"对比并生成 results/。
|
||
|
||
用法（仓库根目录）::
|
||
|
||
uv run python -m project.run # 四问全部
|
||
uv run python -m project.run --question 3 # 只跑第三问
|
||
uv run python -m project.run --quick # 缩小算例/基准集规模
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import time
|
||
|
||
from project.common import RESULTS_DIR, markdown_table, write_text
|
||
from project.q1 import compare as q1_compare
|
||
from project.q2 import compare as q2_compare
|
||
from project.q3 import compare as q3_compare
|
||
from project.q4 import compare as q4_compare
|
||
|
||
RUNNERS = {
|
||
1: q1_compare.run,
|
||
2: q2_compare.run,
|
||
3: q3_compare.run,
|
||
4: q4_compare.run,
|
||
}
|
||
|
||
STRUCTURE = """\
|
||
project/
|
||
├── common.py 公共层：方法描述/指标聚合/结果落盘/闭环装配
|
||
├── q1/ 问题一：直径与直径圆覆盖
|
||
│ ├── optimal.py 生产实现：半平面交 + 顶点对直径 + MEC
|
||
│ ├── baseline.py 基线：网格采样 + bbox 对角线代理
|
||
│ ├── reference.py 参考：大框半平面裁剪 + 精确直径/MEC
|
||
│ ├── cases.py 固定示例（examples/q1）+ 随机交会算例
|
||
│ └── compare.py 同案配对对比
|
||
├── q2/ 问题二：第二检测点选择
|
||
│ ├── optimal.py 生产实现：保证接收盘内最小最坏后验半径
|
||
│ ├── baseline.py 基线：垂直 1000 m / MEC 圆心
|
||
│ ├── reference.py 参考：保证接收盘内网格枚举
|
||
│ ├── region.py K1 统一构造
|
||
│ ├── cases.py 随机首次示向度算例
|
||
│ └── compare.py
|
||
├── q3/ 问题三：全向闭环
|
||
│ ├── optimal.py bayes（生产默认策略）
|
||
│ ├── baseline.py grid（49 点网格）/ seven（七点批量）
|
||
│ └── compare.py
|
||
├── q4/ 问题四：混合源闭环
|
||
│ ├── optimal.py directional（定向扩展）
|
||
│ ├── baseline.py bayes / grid 直接搬用
|
||
│ └── compare.py
|
||
├── run.py 统一入口（本文件）
|
||
└── results/ 自动生成：q1.md/json … q4.md/json + summary.md
|
||
"""
|
||
|
||
|
||
def _q1_headline(payload: dict) -> list[str]:
|
||
agg = payload["aggregate_random"]
|
||
return ["1", "直径平均误差 / 覆盖准确率",
|
||
f"{agg['diameter_mae_optimal_m']:.2f} m / "
|
||
f"{agg['coverage_accuracy_optimal']*100:.0f}%",
|
||
f"{agg['diameter_mae_baseline_m']:.2f} m / "
|
||
f"{agg['coverage_accuracy_baseline']*100:.0f}%"]
|
||
|
||
|
||
def _q2_headline(payload: dict) -> list[str]:
|
||
agg = payload["aggregate"]
|
||
opt = agg["optimal"]
|
||
base = agg["baseline_mec"]
|
||
return ["2", "保证接收率 / 最坏后验均值",
|
||
f"{opt['guaranteed_rate']*100:.0f}% / "
|
||
f"{opt['worst_posterior_mean_m']:.1f} m",
|
||
f"{base['guaranteed_rate']*100:.0f}% / "
|
||
f"{base['worst_posterior_mean_m']:.1f} m"]
|
||
|
||
|
||
def _q3_headline(payload: dict) -> list[str]:
|
||
agg = payload["aggregate"]["variants"]
|
||
opt, base = agg["optimal"], agg["baseline"]
|
||
paired = payload["aggregate"]["paired"].get("optimal_vs_baseline", {})
|
||
return ["3", "全清率 / T 均值（相对基线）",
|
||
f"{opt['cleared_ratio_mean']*100:.0f}% / "
|
||
f"{opt['virtual_time_mean_s']:.0f} s "
|
||
f"({paired.get('mean_relative_change', 0)*100:+.0f}%)",
|
||
f"{base['cleared_ratio_mean']*100:.0f}% / "
|
||
f"{base['virtual_time_mean_s']:.0f} s"]
|
||
|
||
|
||
def _q4_headline(payload: dict) -> list[str]:
|
||
agg = payload["aggregate"]["variants"]
|
||
opt, base = agg["optimal"], agg["baseline"]
|
||
return ["4", "全清率 / 缺席误判（问题三策略搬用）",
|
||
f"{opt['cleared_ratio_mean']*100:.0f}% / "
|
||
f"{opt['absent_misjudged_total']}",
|
||
f"{base['cleared_ratio_mean']*100:.0f}% / "
|
||
f"{base['absent_misjudged_total']}"]
|
||
|
||
|
||
def _write_summary(payloads: dict[int, dict]) -> None:
|
||
lines = ["# 1-4 问最优 vs 基线：对比总览", "",
|
||
"本节所有数字由 `uv run python -m project.run` 在同机同口径下生成，",
|
||
"逐问明细见 `q1.md` … `q4.md`。", ""]
|
||
rows = []
|
||
if 1 in payloads:
|
||
rows.append(_q1_headline(payloads[1]))
|
||
if 2 in payloads:
|
||
rows.append(_q2_headline(payloads[2]))
|
||
if 3 in payloads:
|
||
rows.append(_q3_headline(payloads[3]))
|
||
if 4 in payloads:
|
||
rows.append(_q4_headline(payloads[4]))
|
||
lines.append(markdown_table(["问", "核心指标", "最优实现", "基线实现"], rows))
|
||
lines += ["", "## 方法映射", "",
|
||
"| 问 | 最优实现 | 基线实现 | 独立参考 |",
|
||
"| --- | --- | --- | --- |",
|
||
"| 1 | 半平面交 + 顶点对直径 + MEC | 网格采样 + bbox 对角线代理 | 大框半平面裁剪 + 精确直径/MEC |",
|
||
"| 2 | 保证接收盘内最小最坏后验半径 | 垂直 1000 m / MEC 圆心 | 保证接收盘内网格枚举 |",
|
||
"| 3 | bayes（粒子滤波 + 信息增益 + 滚动调度） | grid（49 点网格）/ seven（七点批量） | 冻结基准集 + 逐微秒计时复算 |",
|
||
"| 4 | directional（联合信念 + 后备清除） | bayes / grid 直接搬用 | 冻结基准集 + 真值全清校验 |",
|
||
"", "## 目录结构", "", "```text", STRUCTURE.rstrip(), "```", ""]
|
||
write_text(RESULTS_DIR / "summary.md", "\n".join(lines))
|
||
|
||
|
||
def main(argv=None) -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||
parser.add_argument("--question", choices=("1", "2", "3", "4", "all"),
|
||
default="all")
|
||
parser.add_argument("--quick", action="store_true",
|
||
help="缩小随机算例数/冻结基准集规模")
|
||
args = parser.parse_args(argv)
|
||
|
||
selected = [1, 2, 3, 4] if args.question == "all" else [int(args.question)]
|
||
payloads: dict[int, dict] = {}
|
||
for question in selected:
|
||
start = time.perf_counter()
|
||
print(f"[Q{question}] 运行对比 …", flush=True)
|
||
payloads[question] = RUNNERS[question](quick=args.quick)
|
||
print(f"[Q{question}] 完成，用时 {time.perf_counter() - start:.1f}s",
|
||
flush=True)
|
||
if selected == [1, 2, 3, 4]:
|
||
_write_summary(payloads)
|
||
print(f"总览已写入 {RESULTS_DIR / 'summary.md'}")
|
||
print(f"结果目录：{RESULTS_DIR}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|