1369 lines
67 KiB
Python
1369 lines
67 KiB
Python
"""把 1-4 问的生产代码抽取成单文件 ipynb（无辅助/测评工具、无未采用方案）。
|
||
|
||
用法::
|
||
|
||
python scripts/build_core_notebook.py [-o mcm26b-核心代码.ipynb]
|
||
|
||
设计：
|
||
- 代码正文**逐字复制**自 `src/mcm26b` 的生产实现，只做三类机械化改写：
|
||
1. 去掉包内相对导入，改用单命名空间的前置导入；
|
||
2. 去掉 `if __name__ == "__main__"` 与文件 IO 入口（覆盖计划改为内嵌字面量）；
|
||
3. 去掉文件头部的模块 docstring（其内容改写为独立的说明 markdown 单元）。
|
||
- 明确排除：`referee/`（独立评测机）、`harness/`（测试适配/内存回环/回放）、
|
||
`cli.py`/`geometry/__main__.py`（命令行外壳）、`scripts/`（采集/基准/可视化）、
|
||
`project/`（最优-基线对比框架）、`tests/`，以及未被任何生产策略调用的
|
||
`routing.py` 离线度量族与 `planner/seven.py` 的未采用变体。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
SRC = ROOT / "src" / "mcm26b"
|
||
|
||
|
||
def read_module(relative: str) -> str:
|
||
"""读取生产源文件并剥离模块级 docstring（正文原样保留）。"""
|
||
text = (SRC / relative).read_text(encoding="utf-8")
|
||
return strip_module_docstring(text)
|
||
|
||
|
||
def strip_module_docstring(text: str) -> str:
|
||
"""删除文件开头的模块 docstring（含其后的空行）。"""
|
||
match = re.match(r'\s*("""|\'\'\')(?:.|\n)*?\1\s*\n', text)
|
||
if match is None:
|
||
return text
|
||
return text[match.end():].lstrip("\n")
|
||
|
||
|
||
def cut(text: str, start: str, end: str | None = None) -> str:
|
||
"""截取 [start, end) 片段（含 start，不含 end）；找不到则报错。"""
|
||
begin = text.index(start)
|
||
if end is None:
|
||
return text[begin:]
|
||
return text[begin:text.index(end, begin)]
|
||
|
||
|
||
def drop(text: str, start: str, end: str | None = None) -> str:
|
||
"""删除 [start, end) 片段（end 为 None 时删到文末），保留其余部分。"""
|
||
begin = text.index(start)
|
||
if end is None:
|
||
return text[:begin]
|
||
return text[:begin] + text[text.index(end, begin):]
|
||
|
||
|
||
# --------------------------------------------------------------------- 源文本
|
||
CONFIG = read_module("config.py")
|
||
RULES = read_module("protocol/rules.py")
|
||
TIMING = read_module("protocol/timing.py")
|
||
STATE = read_module("protocol/state.py")
|
||
TRANSPORT = read_module("protocol/transport.py")
|
||
# 单文件版把 requests 改成惰性导入：notebook 只依赖标准库（+可选 numpy）即可运行，
|
||
# 真正调用 HTTP 时才需要 requests。逻辑与仓库实现完全等价。
|
||
TRANSPORT = TRANSPORT.replace(
|
||
"from dataclasses import dataclass\n\nimport requests\n",
|
||
"from dataclasses import dataclass\n")
|
||
TRANSPORT = TRANSPORT.replace(
|
||
""" def post(self, url, body, headers, timeout):
|
||
try:""",
|
||
""" def post(self, url, body, headers, timeout):
|
||
import requests # 惰性导入：仅真正发起 HTTP 时才需要该依赖
|
||
|
||
try:""")
|
||
LOG = read_module("protocol/log.py")
|
||
CLIENT = read_module("protocol/client.py")
|
||
LOCALIZATION = read_module("geometry/localization.py")
|
||
PLANNING = read_module("geometry/planning.py")
|
||
BELIEF = read_module("belief/__init__.py")
|
||
COVERAGE = read_module("coverage/__init__.py")
|
||
COVER = read_module("guarantee/cover.py")
|
||
PARTICLES = read_module("particles.py")
|
||
JOINT = read_module("belief/joint.py")
|
||
ROUTING = read_module("routing.py")
|
||
PLANNER = read_module("planner/__init__.py")
|
||
BAYES = read_module("planner/bayes_loop.py")
|
||
SEVEN = read_module("planner/seven.py")
|
||
|
||
# ---- routing.py：只保留生产路径用到的开放路径族 --------------------------------
|
||
ROUTING = drop(ROUTING, "\ndef or_opt_open(", "\ndef improve_open_path(")
|
||
ROUTING = ROUTING.replace("import math\n\nfrom .config import Vec2\n",
|
||
"import math\n\nVec2 = tuple[float, float]\n\n")
|
||
ROUTING = ROUTING.replace(
|
||
'EXACT_LIMIT = 13 # Held-Karp 状态数 2^n·n：n=13 → 106 496，纯 Python 可接受\n'
|
||
'OR_OPT_LIMIT = 60 # 作业点数超过此值时跳过高阶 Or-opt（见 improve_open_path）\n\n\n', "")
|
||
ROUTING = drop(ROUTING, "\ndef held_karp_open(", "\ndef _dedupe(")
|
||
ROUTING = ROUTING.replace(
|
||
''' """最近邻 + 2-opt + Or-opt 的开放路径改进（大规模兜底）。
|
||
|
||
Or-opt 的搬移评估是 O(n³)，作业点数超过 :data:`OR_OPT_LIMIT` 时只做
|
||
O(n²) 的 2-opt（度量口径下 Or-opt 的相对收益远小于其墙钟代价）。
|
||
"""
|
||
order = nearest_neighbor_order(points, start)
|
||
order = two_opt_open(order, start, max_passes)
|
||
if len(order) > OR_OPT_LIMIT:
|
||
return order
|
||
return or_opt_open(order, start, max_passes)''',
|
||
''' """最近邻打底 + 2-opt 改进的开放路径（后备清除走格顺序用）。"""
|
||
order = nearest_neighbor_order(points, start)
|
||
return two_opt_open(order, start, max_passes)''')
|
||
# _dedupe 只服务已被排除的离线度量族，一并删除。
|
||
ROUTING = drop(ROUTING, "\ndef _dedupe(", "\ndef route_metrics(")
|
||
ROUTING = drop(ROUTING, "\ndef route_metrics(")
|
||
|
||
# ---- guarantee/cover.py：内嵌计划替代文件 IO ----------------------------------
|
||
COVER = COVER.replace("import hashlib\nimport json\nimport math\n",
|
||
"import json\nimport math\n")
|
||
COVER = COVER.replace("from dataclasses import dataclass, field, fields\n"
|
||
"from functools import lru_cache\nfrom pathlib import Path\n",
|
||
"from dataclasses import dataclass, field, fields\n")
|
||
COVER = COVER.replace('PLAN_DIR = Path(__file__).resolve().parent / "plans"\n', "")
|
||
COVER = drop(COVER, "\ndef _load_plan_file(", "\ndef plan_instance(")
|
||
COVER = COVER.replace(
|
||
'''def plan_instance(name: str, verify: bool = True) -> CoveragePlan:
|
||
"""统一入口：支持内置计划名与磁盘上的证书计划。"""
|
||
if name == "square49":
|
||
return square_49_plan()
|
||
if name == "seven":
|
||
return seven_plan(1200.0)
|
||
return load_plan(name, verify)
|
||
|
||
|
||
def plan_names() -> list[str]:
|
||
"""全部可加载的计划名（含内置）。"""
|
||
builtin = ["square49", "seven"]
|
||
return builtin + sorted(p.stem for p in PLAN_DIR.glob("*.json"))
|
||
|
||
|
||
def write_plan(plan: CoveragePlan, path: Path) -> None:
|
||
"""落盘计划 + 证书（生成器用）。"""
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
payload = plan.as_dict()
|
||
payload["generator"] = dict(payload.get("generator") or {})
|
||
payload["sha256"] = hashlib.sha256(
|
||
json.dumps(payload["points"]).encode("utf-8")).hexdigest()
|
||
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\\n",
|
||
encoding="utf-8")''',
|
||
'''def plan_instance(name: str, verify: bool = True) -> CoveragePlan:
|
||
"""统一入口：按名称取覆盖计划（内置解析计划 / 内嵌证书计划）。"""
|
||
if name == "square49":
|
||
return square_49_plan()
|
||
if name == "seven":
|
||
return seven_plan(1200.0)
|
||
if name not in EMBEDDED_PLANS:
|
||
raise FileNotFoundError(f"未找到覆盖计划 {name!r}；现有：{plan_names()}")
|
||
plan = embedded_plan(name)
|
||
if verify and plan.kind == "certified":
|
||
certificate = verify_plan(plan)
|
||
if not certificate.valid:
|
||
raise ValueError(
|
||
f"覆盖计划 {name!r} 运行期校验失败："
|
||
f"{certificate.uncovered}/{certificate.targets} 个 (p,h) 未被覆盖")
|
||
plan = CoveragePlan(
|
||
name=plan.name, problem=plan.problem, kind=plan.kind,
|
||
coverage=plan.coverage, note=plan.note, generator=plan.generator,
|
||
certificate=CoverageCertificate(
|
||
kind=certificate.kind, target_step_m=certificate.target_step_m,
|
||
headings=certificate.headings, r_eff_min_m=certificate.r_eff_min_m,
|
||
targets=certificate.targets, uncovered=certificate.uncovered,
|
||
worst_hit_distance_m=certificate.worst_hit_distance_m,
|
||
nn_tour_m=certificate.nn_tour_m,
|
||
generator_sha256=(plan.certificate.generator_sha256
|
||
if plan.certificate else "")),
|
||
points=plan.points)
|
||
return plan
|
||
|
||
|
||
def plan_names() -> list[str]:
|
||
"""全部可加载的计划名（内置解析计划 + 内嵌证书计划）。"""
|
||
return ["square49", "seven"] + sorted(EMBEDDED_PLANS)''')
|
||
|
||
# ---- particles.py：单命名空间下不再需要可选 numpy 的跨模块探测 ------------------
|
||
PARTICLES = PARTICLES.replace(
|
||
"from .config import ProblemConfig, Vec2\n"
|
||
"from .geometry.localization import smallest_enclosing_circle\n"
|
||
"from .protocol.rules import DIRECTION, NEAR, NO_SIGNAL\n", "")
|
||
PARTICLES = PARTICLES.replace(
|
||
"import math\nimport os\n\nVec2", "import math\nimport os\n\nVec2")
|
||
|
||
# ---- joint.py：跨模块私有符号改为同命名空间引用 --------------------------------
|
||
JOINT = JOINT.replace(
|
||
"from ..config import ProblemConfig, Vec2\n"
|
||
"from ..particles import ChannelParticles, _branch_radius_array\n"
|
||
"from ..protocol.rules import DIRECTION, NEAR, NO_SIGNAL\n", "")
|
||
JOINT = JOINT.replace(
|
||
"""try: # numpy 仅作可选加速后端（与 particles 一致）
|
||
import numpy as _np
|
||
from ..particles import _np as _particles_np
|
||
if _particles_np is None: # MCM26B_NO_NUMPY 或缺失：强制纯 Python
|
||
_np = None
|
||
except Exception: # pragma: no cover - 环境无 numpy
|
||
_np = None""",
|
||
"""# numpy 仅作可选加速后端（与 particles 同一开关：无 numpy 时走纯 Python 分支）
|
||
_np = _np""")
|
||
|
||
# ---- planner/__init__.py：单命名空间导入 --------------------------------------
|
||
PLANNER = PLANNER.replace(
|
||
"from ..belief import (BeliefState, CandidateRegion, ChannelStatus,\n"
|
||
" InconsistentBelief, clip_halfplane)\n"
|
||
"from ..belief.joint import JointBelief\n"
|
||
"from ..config import ProblemConfig, Vec2\n"
|
||
"from ..coverage import CoverageLedger\n"
|
||
"from ..evaluation import recompute_virtual_time, summarize\n"
|
||
"from ..guarantee import plan_instance, seven_plan, square_49_plan\n"
|
||
"from ..protocol.client import (DeadlineExceeded, ProtocolError,\n"
|
||
" SimulatorClient)\n"
|
||
"from ..protocol.rules import CLEAR_SUCCESS, DIRECTION, NEAR, NO_SIGNAL\n"
|
||
"from ..protocol.timing import move_time\n"
|
||
"from ..protocol.transport import TransportError\n"
|
||
"from ..routing import improve_open_path\n", "")
|
||
|
||
# ---- planner/bayes_loop.py ---------------------------------------------------
|
||
BAYES = BAYES.replace(
|
||
"from ..belief import CandidateRegion, ChannelStatus\n"
|
||
"from ..config import ProblemConfig, Vec2\n"
|
||
"from ..coverage import CoverageLedger\n"
|
||
"from ..particles import ChannelParticles, as_sample_array\n"
|
||
"from ..protocol.client import SimulatorClient\n"
|
||
"from ..protocol.rules import DIRECTION, NEAR, NO_SIGNAL\n"
|
||
"from ..protocol.timing import move_time\n"
|
||
"from . import (ClearCertificate, OmnidirectionalLoop, clear_certificate,\n"
|
||
" candidate_points, safe_disk_radius, select_measurement_point)\n"
|
||
"from .seven import validate_seven_coverage\n"
|
||
"from ..guarantee import seven_plan\n", "")
|
||
|
||
# ---- planner/seven.py：只保留被 CLI 采用的 SevenBatch --------------------------
|
||
SEVEN = SEVEN.replace(
|
||
"from ..belief import ChannelStatus\n"
|
||
"from ..config import ProblemConfig\n"
|
||
"from ..coverage import CoverageLedger\n"
|
||
"from ..guarantee import seven_plan\n"
|
||
"from ..protocol.rules import NEAR\n"
|
||
"from . import OmnidirectionalLoop, clear_certificate\n", "")
|
||
# SevenImmediate 只是"换站集"的中间类，直接把 SevenBatch 接到全向闭环骨架上；
|
||
# SevenSweep（全域扫完再统一服务）不是被采用的方案，整段删除。
|
||
SEVEN = drop(SEVEN, "\nclass SevenImmediate(", "\nclass SevenBatch(")
|
||
SEVEN = SEVEN.replace("class SevenBatch(SevenImmediate):", "class SevenBatch(OmnidirectionalLoop):")
|
||
SEVEN = SEVEN.replace("\n defer_all = False\n", "\n")
|
||
SEVEN = SEVEN.replace(" if not self.defer_all:\n"
|
||
" self._serve_queue()\n",
|
||
" self._serve_queue()\n")
|
||
SEVEN = drop(SEVEN, "\nclass SevenSweep")
|
||
SEVEN = re.sub(r"\n{3,}", "\n\n\n", SEVEN).rstrip() + "\n"
|
||
|
||
# ---- client.py：单命名空间导入 -------------------------------------------------
|
||
CLIENT = CLIENT.replace(
|
||
"from ..config import ProblemConfig, Vec2\n"
|
||
"from .log import SessionLog\n"
|
||
"from .rules import (CLEAR_RESULTS, DIRECTION, EXIT_USER, MEASURE_RESULTS,\n"
|
||
" POSITION_FIELDS, STATUS_OK, STATUS_RETRYABLE)\n"
|
||
"from .state import RobotState\n"
|
||
"from .timing import CostBreakdown\n"
|
||
"from .transport import RequestsTransport, Transport, TransportError\n", "")
|
||
|
||
LOCALIZATION = LOCALIZATION.replace("from dataclasses import dataclass\n",
|
||
"from dataclasses import dataclass\n")
|
||
PLANNING = PLANNING.replace("from math import cos, sin, radians, isfinite\n"
|
||
"from .localization import Measurement\n",
|
||
"from math import cos, sin, radians, isfinite\n")
|
||
|
||
# ---------------------------------------------------------------- 内嵌覆盖计划
|
||
PLAN_V2 = json.loads((SRC / "guarantee" / "plans" / "q4_mincover_v2.json")
|
||
.read_text(encoding="utf-8"))
|
||
PLAN_V1 = json.loads((SRC / "guarantee" / "plans" / "q4_mincover_v1.json")
|
||
.read_text(encoding="utf-8"))
|
||
|
||
|
||
def plan_to_literal(payload: dict) -> str:
|
||
"""把计划 JSON 渲染成可内嵌的 Python 字面量（点集按 6 点一行压缩）。"""
|
||
slim = dict(payload)
|
||
slim.pop("sha256", None)
|
||
points = slim.pop("points")
|
||
body = json.dumps(slim, ensure_ascii=False, indent=2)
|
||
body = body[:-1].rstrip() # 去掉 JSON 末尾的 "}"
|
||
body = "\n".join(" " + line for line in body.splitlines())
|
||
point_lines = []
|
||
for index in range(0, len(points), 6):
|
||
chunk = ", ".join(f"({x:g}, {y:g})" for x, y in points[index:index + 6])
|
||
point_lines.append(" " + chunk + ",")
|
||
literal = (body.rstrip() + ",\n"
|
||
' "points": [\n' + "\n".join(point_lines) + "\n ],\n}")
|
||
# JSON 的 true/false/null 在 Python 字面量里要写成 True/False/None
|
||
for source, target in (("true", "True"), ("false", "False"), ("null", "None")):
|
||
literal = re.sub(rf"\b{source}\b", target, literal)
|
||
return re.sub(r"=(\s*)\{", "= {", literal) # Q4_MINCOVER_V2 = {
|
||
|
||
|
||
EMBEDDED_SOURCE = '''# 覆盖计划以字面量内嵌（原实现从 guarantee/plans/*.json 读取；数值逐位一致）。
|
||
# v2 是现役计划：31 点、设计裕量 r_design=965 m / 前向角裕量 5°，
|
||
# 证书按 15 m / 72 档复核 3 256 200 个 (p, h) 目标，缺口 0、最坏命中 990.05 m。
|
||
# v1（75 m/24 档采样过拟合解）保留作对照，说明"粗采样假通过"的风险。
|
||
Q4_MINCOVER_V2 = ''' + plan_to_literal(PLAN_V2).lstrip() + '''
|
||
Q4_MINCOVER_V1 = ''' + plan_to_literal(PLAN_V1).lstrip() + '''
|
||
|
||
EMBEDDED_PLANS = {"q4_mincover_v1": Q4_MINCOVER_V1, "q4_mincover_v2": Q4_MINCOVER_V2}
|
||
|
||
|
||
def embedded_plan(name: str) -> CoveragePlan:
|
||
"""把内嵌计划字典还原成 :class:`CoveragePlan`（证书字段按数据类取值）。"""
|
||
payload = EMBEDDED_PLANS[name]
|
||
raw_cert = payload.get("certificate") or {}
|
||
known = {item.name for item in fields(CoverageCertificate)}
|
||
certificate = (CoverageCertificate(**{key: value for key, value in raw_cert.items()
|
||
if key in known}) if raw_cert else None)
|
||
return CoveragePlan(
|
||
name=payload["name"], problem=int(payload["problem"]),
|
||
kind=payload.get("kind", "certified"),
|
||
coverage=payload.get("coverage", "forward_half_disk"),
|
||
note=payload.get("note", ""), certificate=certificate,
|
||
generator=payload.get("generator", {}),
|
||
points=tuple((float(x), float(y)) for x, y in payload["points"]))
|
||
'''
|
||
|
||
|
||
def code(source: str) -> dict:
|
||
source = source.strip("\n") + "\n"
|
||
# 单命名空间下 __future__ 只能出现在文件最前：后续单元里的一律去掉
|
||
source = source.replace("from __future__ import annotations\n\n", "", 1)
|
||
return {"cell_type": "code", "execution_count": None, "metadata": {},
|
||
"outputs": [], "source": source.splitlines(keepends=True)}
|
||
|
||
|
||
def md(source: str) -> dict:
|
||
source = source.strip("\n") + "\n"
|
||
return {"cell_type": "markdown", "metadata": {},
|
||
"source": source.splitlines(keepends=True)}
|
||
|
||
|
||
CELLS: list[dict] = []
|
||
add = CELLS.append
|
||
|
||
# =============================================================== 0 说明与前置
|
||
add(md(r"""
|
||
# CUMCM 2026 B 题 · 无线电干扰源的快速自动定位与清除
|
||
## 单文件核心代码（问题一 ~ 问题四）
|
||
|
||
本 notebook 是四问**生产实现**的自包含整理版：按求解流程顺序排列的算法内核，
|
||
不含评测机、命令行外壳、采集/基准/绘图脚本、单元测试，也不含未被采用的方案。
|
||
|
||
### 1. 求解主线
|
||
|
||
| 问 | 任务 | 本文档中的实现 |
|
||
| --- | --- | --- |
|
||
| 一 | 由 ±1° 有界误差示向度交会定位，求定位区域、直径、直径圆覆盖判定、最小包围圆与 20 m 清除证书（并给出直径圆不覆盖区域的等边三角形反例） | `Measurement` / `halfplanes` / `convex_hull` / `diameter` / `smallest_enclosing_circle` / `solve` |
|
||
| 二 | 选择第二个检测点，使目标必被接收且交会角最优 | `first_bearing_candidate_disk` / `CandidateRegion` / `safe_disk_radius` / `select_second_point` |
|
||
| 三 | 全向源：发现全部干扰源、逐一定位清除，最小化虚拟总时间 | `SevenBatch`（对照）与 `BayesLoop`（默认：网格粒子滤波 + 负观测 + 信息增益选点 + 滚动 2-opt） |
|
||
| 四 | 含定向发射源（半盘覆盖）：不可见的静默必须与"距离不足"区分 | `DirectionalExtension`（定向覆盖最小命中集 + (位置×朝向) 联合信念 + ≤20 m 后备清除） |
|
||
|
||
### 2. 公共层（四问共用）
|
||
|
||
- **参数中心**（`ProblemConfig`）：题面常量与工程参数的唯一字面量来源；
|
||
- **协议层**：字段表/结果码（`rules`）、纯计时模型（`timing`）、本地状态机（`state`）、
|
||
传输抽象（`transport`）、会话日志（`log`）、串行幂等客户端（`SimulatorClient`）；
|
||
- **信念层**：逐频道保守凸候选集（`CandidateRegion`）、网格幸存集（`ChannelParticles`）、
|
||
(位置×朝向) 联合可行集（`JointBelief`）；
|
||
- **保证层**：覆盖账本（`CoverageLedger`）、覆盖计划与证书（`CoveragePlan`）；
|
||
- **调度层**：开放路径排序（`nearest_neighbor_order` / `two_opt_open` / `improve_open_path`）。
|
||
|
||
### 3. 三条硬约束（贯穿全部代码）
|
||
|
||
1. **保守性**：真源（第四问含朝向）永远不被观测更新删掉——候选集用外切多边形外包圆盘、
|
||
半平面裁剪带正容差、网格幸存集按 Voronoi 裕量内缩后再删格；
|
||
2. **幂等串行**：一个动作未决不发起下一个；网络失败时原样重发同一 `request_id` 与请求体；
|
||
`accepted=false` 不更新任何状态，其 `virtual_time_s=0` 不是当前时刻；
|
||
3. **计时一致**：预测（客户端）—执行（模拟器）—复算（日志）三方用同一计时模型，
|
||
按整数微秒累计。
|
||
|
||
### 4. 与仓库原文件的关系
|
||
|
||
各代码单元逐字取自 `src/mcm26b/**`，仅做机械化改写（去包内相对导入、去文件 docstring、
|
||
去掉覆盖计划的 JSON 文件读取改为内嵌字面量、去掉未被生产路径调用的函数）。
|
||
本文件可独立运行：**零第三方依赖**（`numpy` 仅作可选加速，缺失时自动回退纯 Python；
|
||
`requests` 仅在真正发起 HTTP 时才惰性导入）。最后的总自检单元会复算四问的核心结论，
|
||
15 项断言全部通过才算整理成功。
|
||
|
||
自检覆盖：问1 角域交会有界/直径圆反例、问2 保证接收与后验收缩、问3 七站解析覆盖证书与
|
||
Voronoi 真源保留、问4 定向最小命中集证书与 (位置×朝向) 联合信念保守性。
|
||
"""))
|
||
|
||
add(md("""
|
||
## 0. 环境与依赖
|
||
"""))
|
||
|
||
add(code(r'''
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import os
|
||
import uuid
|
||
from dataclasses import dataclass, field, fields, replace
|
||
from enum import Enum
|
||
from functools import lru_cache
|
||
from itertools import combinations
|
||
from math import cos, dist, hypot, isfinite, radians, sin
|
||
from pathlib import Path
|
||
from random import Random
|
||
from time import monotonic, sleep
|
||
|
||
try: # numpy 仅作可选加速后端；缺失时全部回退纯 Python
|
||
import numpy as _np
|
||
except Exception: # pragma: no cover - 环境无 numpy 时走纯 Python 分支
|
||
_np = None
|
||
|
||
print("numpy 加速后端:", "启用" if _np is not None else "缺失（纯 Python 回退）")
|
||
'''))
|
||
|
||
# =============================================================== 1 参数中心
|
||
add(md("""
|
||
# 第一部分 · 公共层
|
||
|
||
## 1. 参数中心（`config.py`）
|
||
|
||
题面常量标 `[题面]`，工程参数标 `[工程]`；后者可依演练实测调整，不代表官方规定。
|
||
任何魔法数字只在这里出现一次，其余模块通过 `ProblemConfig` 读取。
|
||
"""))
|
||
|
||
add(code(cut(CONFIG, "Vec2 = tuple[float, float]")))
|
||
add(code(cut(CONFIG, "@dataclass(frozen=True)\nclass ProblemConfig:")))
|
||
|
||
# =============================================================== 2 协议层
|
||
add(md("""
|
||
## 2. 协议层（`protocol/`）
|
||
|
||
### 2.1 规则表与字段表（`rules.py`）
|
||
|
||
四条指令的字段约束、结果码与 HTTP 错误语义，全部落成可执行数据：
|
||
客户端用它构造并预校验请求，模拟器用同一张表校验并拒绝未知字段。
|
||
"""))
|
||
|
||
add(code(cut(RULES, "# ---- 结果码")))
|
||
|
||
add(md("""
|
||
### 2.2 纯计时模型（`timing.py`）
|
||
|
||
附件示例（105 / 111 / 194 / 199 秒）是唯一权威锚点：
|
||
一次合法动作 = 移动（直线距离 / 5 m/s）+ 换频道（仅检测，1 s）+ 动作本身（检测 5 s、
|
||
清除成功 5 s / 未发现 3 s），逐段量化后**按整数微秒累计**。
|
||
"""))
|
||
|
||
add(code(cut(TIMING, "@dataclass(frozen=True)\nclass CostBreakdown")))
|
||
|
||
add(md("""
|
||
### 2.3 本地状态机（`state.py`）
|
||
|
||
位置 / 测向频道 / 虚拟时刻的唯一权威是最近一次 `accepted=true` 的响应；
|
||
`/clear` 不切换测向机频道，只有合法 `/measure` 才更新频道。
|
||
"""))
|
||
|
||
add(code(cut(STATE, "class StateError(RuntimeError):")))
|
||
|
||
add(md("""
|
||
### 2.4 传输抽象与会话日志（`transport.py` / `log.py`）
|
||
|
||
HTTP 细节与协议客户端解耦：正式运行用 `RequestsTransport`，
|
||
本地评测 / 故障注入 / 日志回放各自替换 `Transport` 实现即可，客户端代码零改动。
|
||
会话日志逐条追加 JSONL（题面要求机器狗程序自行留档指令与响应）。
|
||
"""))
|
||
|
||
add(code(cut(TRANSPORT, "class TransportError(Exception):")))
|
||
add(code(cut(LOG, "@dataclass\nclass SessionLog:")))
|
||
|
||
add(md("""
|
||
### 2.5 串行幂等客户端（`client.py`）
|
||
|
||
- 每个新动作使用新的 `request_id`；仅在网络失败后重试**完全相同的动作**时复用原 ID 与
|
||
原请求体（本类一次性序列化出 bytes 并原样重发）；
|
||
- 必须同时检查 HTTP 状态与 `accepted`；`accepted=false` 不更新状态；
|
||
- 逐次等待响应、不得并发；本类完全同步；
|
||
- 现实截止用 `/enter` 的 `remaining_real_duration_s` 建立本地单调时钟并预留退出余量。
|
||
"""))
|
||
|
||
add(code(cut(CLIENT, "class ProtocolError(RuntimeError):")))
|
||
|
||
# =============================================================== 3 几何内核
|
||
add(md("""
|
||
## 3. 几何内核（`geometry/`）
|
||
|
||
### 3.1 角域交会定位（`localization.py`）
|
||
|
||
每条示向度 = 三个半平面：角域下边界、上边界，以及"只能在前向"约束
|
||
（半角 < 90° 时交集是尖锥而非整条直线）。定位区域 = 全部半平面之交。
|
||
|
||
对偶枚举任意两个半平面的交点即得全部候选顶点，再按可行性筛选——
|
||
有界性由"是否存在非零衰退方向"判定，因此**不需要人为加包围盒**：
|
||
|
||
- 无交点 ⇒ 空集（数据矛盾或频道选错）；
|
||
- 有非零衰退方向 ⇒ 无界，直径为 ∞；
|
||
- 否则顶点凸包即定位多边形，直径由顶点对枚举给出，最小包围圆用 Welzl 随机增量算法。
|
||
"""))
|
||
|
||
add(code(cut(LOCALIZATION, "Point = tuple[float, float]")))
|
||
|
||
add(md("""
|
||
### 3.2 问题二的安全候选盘证书（`planning.py`）
|
||
|
||
首次示向度只给出"目标在角域 ∩ 距离 ≤ 1500 m 的扇形内"。该扇形的**最小包围圆**
|
||
半径 a = R / (2 cos δ)；再按保证接收半径（≥ 1000 m）内缩 s = 1000 − a 得到
|
||
保证接收盘 B(圆心, s)——盘内任意点到真源的距离 ≤ s + a = 1000 m，
|
||
因此第二点必收到 `direction`（或 `near`），不会浪费一次 `no_signal`。
|
||
"""))
|
||
|
||
add(code(cut(PLANNING, "def first_bearing_candidate_disk(")))
|
||
|
||
# =============================================================== 4 信念与覆盖
|
||
add(md("""
|
||
## 4. 信念层与保证层
|
||
|
||
### 4.1 逐频道保守凸候选集（`belief/__init__.py`）
|
||
|
||
候选集 K_c 用**圆盘外切多边形 ∩ 测向角域半平面**表示。两条不变式：
|
||
|
||
- 外切多边形严格包含圆盘、半平面裁剪带正容差 ⇒ 更新只会放大不会缩小真实候选集；
|
||
- `direction` 读数给出 K_c ⊆ Ω ∩ W(s, θ±δ_eff) ∩ B(s, 1500)，δ_eff 含示向度两位小数的
|
||
舍入裕量 0.005°；
|
||
- `no_signal` **不删空间候选**（对定向源它只说明被遮挡，见第四问），
|
||
"该频道无源"的证明由覆盖账本或网格幸存集给出；
|
||
- `near` 读数把候选集收缩到 B(s, 5)，可直接产生清除证书。
|
||
"""))
|
||
|
||
add(code(cut(BELIEF, "# 半平面裁剪的正容差（米）")))
|
||
|
||
add(md("""
|
||
### 4.2 逐频道覆盖账本（`coverage/__init__.py`）
|
||
|
||
"路过某点"不算"测过某频道"：覆盖必须按 (频道, 点) 二维记账，
|
||
这是"全点皆 `no_signal` ⇒ 该频道不存在源"这一缺席证明的唯一依据。
|
||
"""))
|
||
|
||
add(code(cut(COVERAGE, "def coverage_points(")))
|
||
|
||
add(md("""
|
||
### 4.3 覆盖计划与证书（`guarantee/cover.py`）
|
||
|
||
两类覆盖语义（题面约束不同，绝不可混用）：
|
||
|
||
| 语义 | 适用 | 缺席条件 |
|
||
| --- | --- | --- |
|
||
| `full_disk` | 第三问全向源 | ∪ B(c, 1000) ⊇ Ω，即 Ω 内每点都落在某个被访点的最小接收圆盘内 |
|
||
| `forward_half_disk` | 第四问定向源 | ∀(p, h) ∈ Ω × S¹，存在被访点 c 使 \\|c−p\\| ≤ 1000 且 (c−p)·u(h) ≥ 0 |
|
||
|
||
**保证 = 离线搜索 + 运行期校验**：点表与证书一起落盘并校验采样语义、缺口数、
|
||
最坏命中距离与生成参数哈希，运行期用同一套采样语义重新校核，未通过即拒绝启动。
|
||
证书硬化把"采样格覆盖"提升为"连续空间覆盖"：设计裕量 r_eff + step·√2/2 ≤ 1000
|
||
且角裕量 ≥ 180/headings/2。
|
||
"""))
|
||
|
||
add(code(cut(COVER, "R_EFF_MIN_M = 1000.0")))
|
||
|
||
add(md("""
|
||
### 4.4 第四问最小命中集（内嵌计划）
|
||
|
||
现役计划 `q4_mincover_v2`：31 点（700 m 方格 49 点的充分条件可压到 31 点），
|
||
设计裕量 r_design = 965 m、前向角裕量 5°，按 15 m / 72 档共 3 256 200 个 (p, h)
|
||
复核缺口为 0、最坏命中 990.05 m、最近邻巡回 23.88 km。
|
||
`q4_mincover_v1` 是同点数的粗采样（75 m / 24 档）解，细网格复检有 125 个缺口——
|
||
保留它正是"粗采样假通过"的反例。
|
||
"""))
|
||
|
||
add(code(cut(COVER, "def seven_points(")))
|
||
|
||
add(code(EMBEDDED_SOURCE))
|
||
|
||
add(md("""
|
||
### 4.5 第三问解析七站计划（`planner/seven.py` 覆盖部分）
|
||
|
||
原点 + 半径 r 的 6 等分环：域半径 R、环半径 r 时，域内任一点到最近站的距离
|
||
≤ max(r/√3, √(R² + r² − √3Rr))，只要该上界 < 1000 m（最小有效接收半径），
|
||
"7 站全 `no_signal`"就是严格的缺席证明。r ∈ [1122.9, 1732] 可行，巡检半径越小巡回越短。
|
||
"""))
|
||
|
||
add(code(cut(SEVEN, "RING_RADIUS_M = 1200.0", "\nclass SevenBatch")))
|
||
|
||
add(md("""
|
||
### 4.6 网格幸存集粒子滤波（`particles.py`）
|
||
|
||
保守凸外包无法表达"角域 ∩ 距离盘 ∩ 全部 no_signal 排除盘"这类**非凸（挖洞）**可行集。
|
||
网格幸存集可以，并额外买到三件事：
|
||
|
||
1. **缺席证明**：幸存集为空 ⇒ 可行集为空 ⇒ 该频道无源（严格，不必等覆盖账本跑完）；
|
||
2. **跳测规则**：幸存格到测点的最小距离 > r_max ⇒ 该次测量必 `no_signal` 且不产生新排除，
|
||
直接跳过（省 5 s 检测 + 1 s 换频）；
|
||
3. **负观测更新**：`no_signal` 直接删掉 B(s, 1000) 内的整格。
|
||
|
||
严格性由 **Voronoi 裕量**保证：格距 h 的裕量 m = √2/2·h 是格心到格内任一点的最大距离，
|
||
所有删除判定都内缩 m（"整格都在可行集外才删"），因此真源所在格永不被删。
|
||
"""))
|
||
|
||
add(code(cut(PARTICLES, "_ANG_EPS_DEG = 1e-9")))
|
||
|
||
add(md("""
|
||
### 4.7 (位置 × 朝向) 联合可行集（`belief/joint.py`）
|
||
|
||
第四问的失败全部来自定向源：策略知道"源可能在哪个空间区域"，却不知道"它朝着哪边"。
|
||
本模块给每个幸存格附一个 24 档（15°）朝向可行性掩码，把"看不看得见"变成可推理的量。
|
||
更新规则全部取保守侧并计入**格内角裕量** `asin(margin / max(dist − margin, 1))`
|
||
（格内任一点相对格心方位偏差的上界）：
|
||
|
||
| 观测 | 位置 | 朝向 |
|
||
| --- | --- | --- |
|
||
| `direction` @ s | 距离 ≤ 1500、角向 \\|gap\\| ≤ δ_eff + 格裕量 | 必须**可能**覆盖 s |
|
||
| `near` @ s | 距离 ≤ 5 | 必须**可能**覆盖 s（near 同样要求可见） |
|
||
| `no_signal` @ s | **不删**（可能是遮挡，也可能是距离不足） | 仅在整格必落在 1000 m 内时，删去"必然看得见"的档 |
|
||
|
||
由此得到二维真源保留定理：真源 (位置, 朝向) 永不被删。`prob_detect` 用可见性概率选点，
|
||
取代固定的 45° 阶梯试探；联合集为空同样就是缺席证明。
|
||
"""))
|
||
|
||
add(code(cut(JOINT, "HEADING_BINS = 24")))
|
||
|
||
# =============================================================== 5 调度层
|
||
add(md("""
|
||
## 5. 调度层：开放路径排序（`routing.py`）
|
||
|
||
本题的真实图问题是**开放路径**：起点固定在机器人当前位置、终点自由，
|
||
代价 = Σ 移动时间 + Σ 动作耗时。后备清除的走格顺序用最近邻打底 + 2-opt 改进
|
||
（O(1) 增量代价评估，终点自由故只反转内部片段、不做首尾相接）。
|
||
"""))
|
||
|
||
add(code(ROUTING))
|
||
|
||
# =============================================================== 6 策略层
|
||
add(md("""
|
||
# 第二部分 · 策略层（`planner/`）
|
||
|
||
## 6. 问题一/二/三共用的选点器与全向闭环
|
||
|
||
- **问题一**在几何内核里直接求解（第三部分给出算例）；
|
||
- **问题二**用 `select_second_point`：在保证接收盘内环形采样，以"最坏后验包围圆半径"
|
||
为主指标、移动代价为次指标；
|
||
- **问题三**用 `OmnidirectionalLoop` 骨架：覆盖 → 定位 → 清除 → 收尾，
|
||
子类只替换覆盖计划与调度细节。
|
||
"""))
|
||
|
||
add(code(cut(PLANNER, "\ndef bearing_deg(a: Vec2, b: Vec2) -> float:",
|
||
"\nclass OmnidirectionalLoop:")))
|
||
|
||
add(md("""
|
||
### 6.1 全向闭环骨架（`OmnidirectionalLoop`）
|
||
|
||
发现（覆盖账本）→ 定位（候选集收敛到 20 m 内）→ 清除 → 收尾 `/exit`。
|
||
任何异常都必须尽力 `/exit` 并产出 summary，不得静默丢会话：
|
||
断连、协议错误、守恒性破坏（候选集被剪空 / 保证接收盘内 `no_signal` /
|
||
清除证书失效）都会记入 `stop_reason` 与异常表。
|
||
`completed` 表示任务在证明意义上完成（每频道 CLEARED 或 ABSENT），
|
||
`accounted_virtual_time_s` 记录首次"找全证明"的虚拟时刻。
|
||
"""))
|
||
|
||
add(code(cut(PLANNER, "\nclass OmnidirectionalLoop:",
|
||
"\nclass DirectionalExtension(OmnidirectionalLoop):")))
|
||
|
||
add(md("""
|
||
### 6.2 第四问定向扩展（`DirectionalExtension`）
|
||
|
||
与第三问的三点差异：
|
||
|
||
1. 覆盖阶段换成定向覆盖最小命中集（`q4_mincover_v2`），发现保证对定向源同样成立；
|
||
2. 保证接收盘内的 `no_signal` 只能由**朝向遮挡**解释：该频道记为"观测到定向"，
|
||
对 MEC 圆心取对跖探点复核，并用联合信念按可见概率选点；
|
||
3. 定位停滞时对**有界**候选集构造 ≤20 m 有限圆盘覆盖，逐格 `/clear`
|
||
（清除与朝向无关，是完备后备）；格数超硬顶或预算不足即跳过并留痕，绝不盲扫。
|
||
"""))
|
||
|
||
add(code(cut(PLANNER, "\nclass DirectionalExtension(OmnidirectionalLoop):")))
|
||
|
||
add(md("""
|
||
### 6.3 第三问默认策略：贝叶斯网格滤波闭环（`planner/bayes_loop.py`）
|
||
|
||
三层重构：
|
||
|
||
1. **感知层**：每频道一个网格幸存集，全部读数（含 `no_signal` 负观测）即时更新，
|
||
幸存集为空即缺席证明；
|
||
2. **决策层**：站点扫频、定位测量、清除三类任务合成统一任务表，
|
||
开放路径最近邻 + 2-opt 排序，每执行一个任务滚动重排（动态 TSP）；
|
||
3. **选点层**：定位测量点在保证接收盘（凸证书）或 PF MEC 探索环（宽候选集）内
|
||
按"期望 log 幸存占比 + λ·移动时间"评分（信息增益贪心）。
|
||
|
||
相对基线的增益来自负观测跳站/跳测、PF 缺席证明提前证毕、全局任务排序。
|
||
清除证书仍以凸证书为主，PF 网格证书只在负观测挖洞使幸存集更紧时提前触发。
|
||
"""))
|
||
|
||
add(code(BAYES))
|
||
|
||
add(md("""
|
||
### 6.4 第三问对照策略：七站批量扫频（`planner/seven.py`）
|
||
|
||
同一骨架换覆盖计划：7 站解析覆盖 + 站内批量扫频 + 按"距候选集 MEC 圆心最近"服务队列。
|
||
它曾以官方案例 12/12 全清通过演练，是"覆盖证明 + 批量调度"的中间对照。
|
||
"""))
|
||
|
||
add(code(cut(SEVEN, "\nclass SevenBatch")))
|
||
|
||
# =============================================================== 7 问题一
|
||
add(md("""
|
||
# 第三部分 · 逐问算例与自检
|
||
|
||
以下单元用**合成观测**驱动上面的求解器（合成观测的生成规则单独写明，
|
||
不依赖评测机）。每个单元末尾用断言检查结论，运行通过即说明该问的代码链路正确。
|
||
|
||
## 7. 问题一：角域交会定位
|
||
|
||
三个检测点、±1° 误差的示向度。输出：定位区域多边形顶点、直径、直径圆是否覆盖整个区域、
|
||
最小包围圆（MEC）半径，以及"MEC 半径 ≤ 20 m"的清除证书。
|
||
"""))
|
||
|
||
add(code(r'''
|
||
Q1_EXAMPLES = {
|
||
"normal": { # 三点交会：区域有界、MEC 远小于 20 m
|
||
"target": (80.0, 90.0),
|
||
"measurements": [(-600.0, -250.0, 27.21505117707799),
|
||
(650.0, -100.0, 161.015051177078),
|
||
(100.0, 700.0, 268.4721225527146)],
|
||
},
|
||
"triangle": { # 直径圆反例：直径圆不覆盖区域
|
||
"target": (20.0, 11.547005383792515),
|
||
"measurements": [(-1000.0, 0.0, 1.0),
|
||
(540.0, -866.0254037844386, 121.0),
|
||
(520.0, 900.6664199358162, 241.0)],
|
||
},
|
||
"poor": { # 近共线：细长区域，MEC 远大于直径圆
|
||
"target": (0.0, 0.0),
|
||
"measurements": [(-900.0, -20.0, 1.29), (0.0, 0.0, 0.0), (900.0, 20.0, 358.73)],
|
||
},
|
||
}
|
||
|
||
def solve_q1(case: str, rounding_margin_deg: float = 0.0) -> dict:
|
||
"""复现 `mcm26b q1 examples/q1/<case>.json` 的求解口径。"""
|
||
rows = Q1_EXAMPLES[case]["measurements"]
|
||
measurements = [Measurement(x, y, bearing, 1.0 + rounding_margin_deg)
|
||
for x, y, bearing in rows]
|
||
return solve(measurements)
|
||
|
||
normal = solve_q1("normal")
|
||
print("== 三点交会 ==")
|
||
print("状态 :", normal["status"], "顶点数", len(normal["vertices"]))
|
||
print("直径 : %.3f m" % normal["diameter_m"])
|
||
print("直径圆覆盖区域 :", normal["diameter_circle"]["covers"],
|
||
"最大超出 %.3f m" % normal["diameter_circle"]["max_excess_m"])
|
||
print("最小包围圆 : 圆心 (%.3f, %.3f) 半径 %.3f m"
|
||
% (*normal["minimum_enclosing_circle"]["center"],
|
||
normal["minimum_enclosing_circle"]["radius_m"]))
|
||
print("可原地清除 :", normal["guaranteed_clearable_at_mec_center"])
|
||
assert normal["status"] == "polygon" and normal["diameter_m"] > 0
|
||
assert normal["diameter_infinite"] is False
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# 反例：直径圆的圆心是"最远顶点对"的中点，第二个最远顶点可能落在圆外。
|
||
tri = solve_q1("triangle")
|
||
print("== 直径圆反例（等边三角形） ==")
|
||
print("定位区域顶点：", [[round(v, 2) for v in p] for p in tri["vertices"]])
|
||
print("直径 %.3f m（最远顶点对 %s）" % (tri["diameter_m"],
|
||
[[round(v, 2) for v in p] for p in tri["diameter_pair"]]))
|
||
print("直径圆：圆心 %s 半径 %.3f m" % ([round(v, 2) for v in
|
||
tri["diameter_circle"]["center"]], tri["diameter_circle"]["radius_m"]))
|
||
print("圆外顶点下标 %s（见证顶点 %s），最大超出 %.3f m"
|
||
% (tri["diameter_circle"]["outside_vertex_indices"],
|
||
[round(v, 2) for v in tri["diameter_circle"]["witness_vertex"]],
|
||
tri["diameter_circle"]["max_excess_m"]))
|
||
print("直径圆覆盖整个定位区域：", tri["diameter_circle"]["covers"])
|
||
print("最小包围圆半径 %.3f m（>= 直径/2 = %.1f m，才是真正的最小覆盖半径）"
|
||
% (tri["minimum_enclosing_circle"]["radius_m"], tri["diameter_m"] / 2))
|
||
assert tri["diameter_circle"]["covers"] is False
|
||
assert tri["diameter_circle"]["outside_vertex_indices"] == [2]
|
||
# 一般地 直径/2 <= 最小包围圆半径；等边三角形取严格大于（23.09 > 20）
|
||
assert tri["minimum_enclosing_circle"]["radius_m"] >= tri["diameter_m"] / 2 - 1e-9
|
||
|
||
poor = solve_q1("poor")
|
||
print("\n== 近共线细长区域 ==")
|
||
print("状态 %s，直径 %.1f m，最小包围圆半径 %.1f m，可原地清除 %s"
|
||
% (poor["status"], poor["diameter_m"],
|
||
poor["minimum_enclosing_circle"]["radius_m"],
|
||
poor["guaranteed_clearable_at_mec_center"]))
|
||
assert poor["minimum_enclosing_circle"]["radius_m"] > 20.0
|
||
assert poor["guaranteed_clearable_at_mec_center"] is False
|
||
|
||
empty = solve([])
|
||
unbounded = solve([Measurement(-600.0, -250.0, 27.21505117707799, 1.0)])
|
||
print("\n无观测 ->", empty["status"], "| 单条示向度 ->", unbounded["status"],
|
||
"（衰退方向 %s）" % unbounded["recession_direction"])
|
||
assert empty["status"] == "unbounded" and unbounded["status"] == "unbounded"
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# 示向度二位小数的舍入敏感性：误差半角 1.0° vs 1.005°（含 0.005° 舍入裕量）
|
||
tight = solve_q1("normal", 0.0)
|
||
loose = solve_q1("normal", 0.005)
|
||
print("舍入裕量 0.000°：直径 %.3f m，MEC 半径 %.3f m"
|
||
% (tight["diameter_m"], tight["minimum_enclosing_circle"]["radius_m"]))
|
||
print("舍入裕量 0.005°：直径 %.3f m，MEC 半径 %.3f m"
|
||
% (loose["diameter_m"], loose["minimum_enclosing_circle"]["radius_m"]))
|
||
print("放宽 %.3f° 使 MEC 半径增加 %.3f m —— 因此库默认带该裕量，"
|
||
"否则合法真源可能被角域裁掉" % (0.005, loose["minimum_enclosing_circle"]["radius_m"]
|
||
- tight["minimum_enclosing_circle"]["radius_m"]))
|
||
assert loose["minimum_enclosing_circle"]["radius_m"] >= \
|
||
tight["minimum_enclosing_circle"]["radius_m"] - 1e-9
|
||
'''))
|
||
|
||
# =============================================================== 8 问题二
|
||
add(md("""
|
||
## 8. 问题二：第二检测点的选择
|
||
|
||
给定首次 `direction` 读数（检测点 s₁、示向度 θ₁±1°），构造
|
||
K₁ = Ω ∩ W(s₁, θ₁±δ_eff) ∩ B(s₁, 1500)，再在**保证接收盘**内选第二点：
|
||
|
||
- 目标函数：候选读数分支下的**最坏后验包围半径**（越小越好，直接决定能否进入 20 m 清除圈）；
|
||
- 次指标：从当前位置出发的移动时间（同分决胜）；
|
||
- 采样是启发式（候选点族 × 目标边界点 × 读数区间端点），选定后按 1/4 间距加密复核并如实上报。
|
||
"""))
|
||
|
||
add(code(r'''
|
||
class _StateStub:
|
||
"""最小状态替身：只提供选点器需要的当前位置与当前测向频道。"""
|
||
|
||
def __init__(self, position=(0.0, 0.0), channel: int = 1):
|
||
self.position = position
|
||
self.channel = channel
|
||
|
||
|
||
class _ClientStub:
|
||
def __init__(self, position=(0.0, 0.0), channel: int = 1):
|
||
self.state = _StateStub(position, channel)
|
||
|
||
|
||
CONFIG = ProblemConfig()
|
||
|
||
def plan_second_point(station, bearing_deg, from_position=(0.0, 0.0)):
|
||
"""问题二主入口：首次示向度 -> 保证接收的第二检测点方案。"""
|
||
client = _ClientStub(from_position, channel=1)
|
||
return select_second_point(client, station, bearing_deg, CONFIG)
|
||
|
||
|
||
def plan_second_point_from(region, from_position=(0.0, 0.0)):
|
||
"""已知候选集时的下一测点（同一选点器，供逐次追加测点演示）。"""
|
||
return select_measurement_point(region, from_position, CONFIG)
|
||
|
||
|
||
# 首次检测：真源在 (300, 400)，从原点测得示向度 53.13°，误差半角 ±1°
|
||
station1, bearing1 = (0.0, 0.0), 53.13010235415598
|
||
cert = first_bearing_candidate_disk(Measurement(*station1, bearing1, 1.0))
|
||
print("首次角域扇形的包围圆：圆心 (%.1f, %.1f) 半径 %.1f m"
|
||
% (*cert["sector_enclosing_circle"]["center"],
|
||
cert["sector_enclosing_circle"]["radius_m"]))
|
||
print("保证接收盘 ：圆心 (%.1f, %.1f) 半径 %.1f m"
|
||
% (*cert["sufficient_candidate_disk"]["center"],
|
||
cert["sufficient_candidate_disk"]["radius_m"]))
|
||
|
||
# 首次读数后的候选集 K1 = Ω ∩ W(s1, θ1±δ_eff) ∩ B(s1, 1500)
|
||
half = 1.0 + CONFIG.search.bearing_rounding_margin_deg
|
||
region = CandidateRegion.initial(station1, CONFIG)
|
||
region.clip_wedge(station1, bearing1, half)
|
||
|
||
plan = plan_second_point(station1, bearing1)
|
||
print("\n第二检测点 (%8.2f, %8.2f)" % plan.position)
|
||
print("最坏后验包围半径 %.2f m（当前候选集 MEC 半径 %.1f m，必须更小才叫有信息）"
|
||
% (plan.worst_posterior_radius_m, region.mec()[1]))
|
||
print("移动耗时 %.1f s（次指标：同样精度下取更近的点）" % plan.move_cost_s)
|
||
print("依据：", plan.reason)
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# 用同一套观测语义复核这个点：无论真源落在候选集何处，第二点都必收到信号。
|
||
# 候选集边界按 1 m 加密采样做最坏情形检验。
|
||
worst_distance, worst_radius, samples = 0.0, 0.0, 0
|
||
for truth in region.witnesses(1.0):
|
||
worst_distance = max(worst_distance, math.dist(plan.position, truth))
|
||
for reading in (bearing_deg(plan.position, truth) - half,
|
||
bearing_deg(plan.position, truth) + half):
|
||
probe = CandidateRegion(list(region.polygon), region.slack_m)
|
||
probe.clip_wedge(plan.position, reading, half)
|
||
if probe.polygon:
|
||
worst_radius = max(worst_radius, probe.mec()[1])
|
||
samples += 1
|
||
print("候选集边界采样点 %d 个" % samples)
|
||
print("真源到第二点的最远距离 %.2f m（<= 1000 m 才保证收到信号）" % worst_distance)
|
||
print("对候选集最坏点、最不利读数的后验包围半径 %.2f m" % worst_radius)
|
||
print("第二点相对候选集包围圆的保证接收余量：%.1f m"
|
||
% (CONFIG.field.r_eff_min_m - region.mec()[1] - math.dist(plan.position,
|
||
region.mec()[0])))
|
||
assert worst_distance <= CONFIG.field.r_eff_min_m # 保证接收：不会浪费 no_signal
|
||
assert worst_radius < region.mec()[1] # 一定带来收缩
|
||
|
||
# 逐次追加测点：最坏后验半径单调收缩，第三点后进入 20 m 清除圈
|
||
for step in range(2, 5):
|
||
if plan.worst_posterior_radius_m <= CONFIG.field.clear_radius_m:
|
||
break
|
||
region.clip_wedge(plan.position, bearing_deg(plan.position, (300.0, 400.0)), half)
|
||
plan = plan_second_point_from(region, (0.0, 0.0))
|
||
print("追加第 %d 个测点 (%8.2f, %8.2f)：最坏后验半径 %.2f m"
|
||
% (step, *plan.position, plan.worst_posterior_radius_m))
|
||
print("\n结论：第二点负责\"保证接收 + 最大化交会角\"，后验精度由后续测点继续压缩；"
|
||
"20 m 清除圈通常在第三点后达成。")
|
||
assert plan.worst_posterior_radius_m <= CONFIG.field.clear_radius_m
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# 对照：候选集在不同首次示向度下的选择结果（目标函数与移动代价的权衡）
|
||
print("首次示向度 第二点位置 最坏后验半径 移动耗时")
|
||
for bearing in (0.0, 45.0, 90.0, 135.0, 180.0, 225.0, 270.0, 315.0):
|
||
item = plan_second_point(station1, bearing, from_position=(-1500.0, 0.0))
|
||
print("%7.1f° (%8.2f, %8.2f) %8.2f m %6.1f s"
|
||
% (bearing, item.position[0], item.position[1],
|
||
item.worst_posterior_radius_m, item.move_cost_s))
|
||
'''))
|
||
|
||
# =============================================================== 9 问题三
|
||
add(md(r"""
|
||
## 9. 问题三：全向源闭环
|
||
|
||
第三问没有可直连的模拟器时，仍可自检全部**决策内核**：
|
||
|
||
1. 七站覆盖计划是否真的覆盖 Ω（每点到最近站 ≤ 1000 m）；
|
||
2. 网格幸存集在随机观测序列后是否始终保留真源（Voronoi 裕量定理）；
|
||
3. 空幸存集、跳测规则、清除证书是否按预期给出；
|
||
4. 任务表排序（最近邻 + 2-opt）是否真的缩短开放路径。
|
||
|
||
合成观测按下述语义生成，与题面一致（`r_eff` 未知，只在 [1000, 1500] 内取值）：
|
||
`dist > r_eff → no_signal`；`dist ≤ 5 → near`；否则 `direction`，读数为真实方位
|
||
加上 ≤1° 的误差。
|
||
"""))
|
||
|
||
add(code(r'''
|
||
def observe_omni(truth, station, r_eff, bearing_noise_deg=0.0):
|
||
"""全向源的观测语义（合成用）：返回 (result, svd_deg)。"""
|
||
distance = math.dist(truth, station)
|
||
if distance > r_eff:
|
||
return "no_signal", None
|
||
if distance <= 5.0:
|
||
return "near", None
|
||
return "direction", (bearing_deg(station, truth) + bearing_noise_deg) % 360.0
|
||
|
||
|
||
def observe_directional(truth, heading_deg, station, r_eff, bearing_noise_deg=0.0):
|
||
"""定向源的观测语义：源→读点方向与朝向夹角 > 90° 时同样静默。"""
|
||
dx, dy = station[0] - truth[0], station[1] - truth[1]
|
||
distance = math.hypot(dx, dy)
|
||
if distance > r_eff:
|
||
return "no_signal", None
|
||
forward = dx * math.cos(math.radians(heading_deg)) \
|
||
+ dy * math.sin(math.radians(heading_deg))
|
||
if forward < 0.0:
|
||
return "no_signal", None # 背向发射：接收不到
|
||
if distance <= 5.0:
|
||
return "near", None
|
||
return "direction", (bearing_deg(station, truth) + bearing_noise_deg) % 360.0
|
||
|
||
|
||
# ---- 9.1 七站解析覆盖计划 ---------------------------------------------------
|
||
CONFIG_Q3 = replace(CONFIG, bayes=replace(BayesConfig(), station_ring_radius_m=1200.0))
|
||
bound = validate_seven_coverage(CONFIG_Q3.bayes.station_ring_radius_m, CONFIG_Q3)
|
||
plan7 = seven_plan(CONFIG_Q3.bayes.station_ring_radius_m)
|
||
print("七站环半径 %.1f m，最近站最远距离解析上界 %.1f m（阈值 1000 m）"
|
||
% (CONFIG_Q3.bayes.station_ring_radius_m, bound))
|
||
print("站点：", [tuple(round(v, 1) for v in point) for point in plan7.points])
|
||
|
||
# 数值复核必须与证书同分辨率：这里用 300 m / 12 档演示（全分辨率证书见 4.4 节）
|
||
cert7 = verify_full_disk(plan7.points, step_m=300.0, r_eff=R_EFF_MIN_M)
|
||
print("粗采样复核：目标点 %d 个，未覆盖 %d 个，最坏命中 %.1f m，最近邻巡回 %.1f km"
|
||
% (cert7.targets, cert7.uncovered, cert7.worst_hit_distance_m, cert7.nn_tour_m / 1000))
|
||
assert cert7.uncovered == 0 and cert7.worst_hit_distance_m <= R_EFF_MIN_M
|
||
assert bound <= CONFIG_Q3.field.r_eff_min_m
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# ---- 9.2 网格幸存集的真源保留性（Voronoi 裕量定理的随机检验） ----------------
|
||
Random(20260913).seed(20260913)
|
||
source = (300.0, 400.0) # 真源
|
||
r_eff_true = 1200.0 # 真源的有效接收半径（策略端不可见）
|
||
stations = [(x, y) for x, y in
|
||
[(-900.0, 200.0), (0.0, 0.0), (600.0, -700.0), (1100.0, 800.0)]]
|
||
|
||
pf = ChannelParticles(CONFIG_Q3)
|
||
start_alive = len(pf)
|
||
retained = True
|
||
for index, station in enumerate(stations):
|
||
result, svd = observe_omni(source, station, r_eff_true,
|
||
bearing_noise_deg=0.8 if index % 2 else -0.9)
|
||
pf.apply_measure(station, result, svd)
|
||
cell = (min(pf.alive, key=lambda k: math.dist(pf.cell_center(k), source))
|
||
if pf.alive else None)
|
||
truth_offset = (math.dist(pf.cell_center(cell), source) if cell is not None
|
||
else float("nan"))
|
||
keeps_truth = (cell is not None
|
||
and truth_offset <= pf.margin_m + 1e-9)
|
||
retained = retained and keeps_truth
|
||
center, radius = pf.mec()
|
||
print("第 %d 次读数 %-9s 幸存格 %6d 格心距真源 %7.3f m（裕量 %.3f m） MEC 半径 %8.3f m"
|
||
% (index + 1, result, len(pf), truth_offset, pf.margin_m, radius))
|
||
print("\n初始幸存格 %d；真源始终落在某个幸存格内：%s" % (start_alive, retained))
|
||
print("Voronoi 裕量 m = %.3f m（格心到格内任一点的最大距离）" % pf.margin_m)
|
||
assert retained and pf.version > 0
|
||
|
||
# 跳测规则：必 no_signal 且不产生新排除的 (测点, 频道) 对应当被直接跳过
|
||
far_station = (4000.0, 4000.0)
|
||
print("远端测点 %s 是否必 no_signal：%s" % (far_station, pf.guaranteed_no_signal(far_station)))
|
||
assert pf.guaranteed_no_signal(far_station) is True
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# ---- 9.3 证书与缺席证明 -----------------------------------------------------
|
||
center, radius = pf.mec()
|
||
limit = CONFIG_Q3.field.clear_radius_m - pf.margin_m - CONFIG_Q3.search.clear_margin_m
|
||
print("幸存集 MEC：圆心 (%.2f, %.2f) 半径 %.3f m" % (*center, radius))
|
||
print("网格清除证书阈值 %.3f m（= 20 − 裕量 %.3f − 工程余量 %.1f）→ 可清除：%s"
|
||
% (limit, pf.margin_m, CONFIG_Q3.search.clear_margin_m, radius <= limit))
|
||
|
||
# near 读数把候选集收缩到 5 m 盘：凸口径直接给出清除证书
|
||
near_region = CandidateRegion.disk((source[0] + 3.0, source[1]), 5.0,
|
||
CONFIG_Q3.search.polygon_sides)
|
||
near_cert = clear_certificate(near_region, channel=7, config=CONFIG_Q3)
|
||
print("near 读数后的凸证书：半径 %.3f m ≤ %.3f m → %s"
|
||
% (near_cert.radius_m, near_cert.limit_m, near_cert.certified))
|
||
assert near_cert.certified
|
||
|
||
# 网格清除证书成立时，目标点可直接贴着来向利用 20 m 容差（三角不等式仍覆盖整集）
|
||
near_cert2 = loop_cert = None
|
||
center, radius = pf.mec()
|
||
shift = max(0.0, (CONFIG_Q3.field.clear_radius_m - pf.margin_m
|
||
- CONFIG_Q3.search.clear_margin_m) - radius)
|
||
shifted = (center[0] + shift * 0.6, center[1] + shift * 0.8)
|
||
print("沿来向偏移后清除点距真源 %.3f m <= 半径 %.3f + 偏移 %.3f = %.3f m < 20 m"
|
||
% (math.dist(shifted, source), radius, shift, radius + shift))
|
||
|
||
# 缺席证明：覆盖计划（七站）全 no_signal 后幸存集清空 —— 与解析证明互相印证
|
||
absent = ChannelParticles(CONFIG_Q3)
|
||
for station in plan7.points:
|
||
absent.apply_measure(station, "no_signal")
|
||
print("\n七站全部 no_signal 后幸存格 %d → 该频道判缺席（无需等覆盖账本跑完）"
|
||
% len(absent))
|
||
partial = ChannelParticles(CONFIG_Q3)
|
||
for station in plan7.points[:-1]:
|
||
partial.apply_measure(station, "no_signal")
|
||
print("只测其中 6 站仍有 %d 个幸存格 → 七站是刚好够用的最小解析覆盖"
|
||
% len(partial))
|
||
assert len(absent) == 0 and len(partial) > 0
|
||
assert absent.guaranteed_no_signal(tuple(plan7.points[0])) is True
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# ---- 9.4 统一任务表与滚动排序（信息增益选点 + 2-opt） ------------------------
|
||
class _LogStub:
|
||
def action_events(self):
|
||
return []
|
||
|
||
def step_events(self):
|
||
return []
|
||
|
||
|
||
client3 = SimulatorClient("OFFLINE", config=CONFIG_Q3, transport=None, log=_LogStub())
|
||
client3.state.position = (0.0, 0.0)
|
||
loop3 = BayesLoop(client3, CONFIG_Q3)
|
||
print("第三问策略：覆盖计划 %s（%d 站），网格间距 %.2f m"
|
||
% (loop3.coverage_plan.name, len(loop3.coverage_plan.points),
|
||
CONFIG_Q3.bayes.particle_spacing_m))
|
||
|
||
# 注入一次真实读数，建立该频道的信念与粒子集
|
||
loop3._record_measure(9, (0.0, 0.0), type("O", (), {
|
||
"result": "direction", "svd_deg": bearing_deg((0.0, 0.0), (300.0, 400.0))})())
|
||
tasks = loop3._build_tasks()
|
||
print("任务表：%d 项（站点扫频 %d / 定位 %d / 清除 %d）"
|
||
% (len(tasks), sum(t.kind == "station" for t in tasks),
|
||
sum(t.kind == "locate" for t in tasks), sum(t.kind == "clear" for t in tasks)))
|
||
order = loop3._order_tasks(tasks)
|
||
print("排序后前 5 项：", [(t.kind, t.channel) for t in order[:5]])
|
||
print("2-opt 前后路径耗时 %.2f s -> %.2f s"
|
||
% (loop3._path_cost(tasks), loop3._path_cost(order)))
|
||
assert loop3._path_cost(order) <= loop3._path_cost(tasks) + 1e-9
|
||
|
||
point = loop3._select_point(9)
|
||
print("信息增益选点：(%8.2f, %8.2f)" % point)
|
||
assert point is not None and 9 in loop3.particles
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# ---- 9.5 覆盖计划证书：离线全分辨率生成，运行期按同分辨率复核 ----------------------------------
|
||
plan_v2 = plan_instance("q4_mincover_v2", verify=False)
|
||
cert_v2 = plan_v2.certificate
|
||
print("第四问覆盖计划 %s：%d 点" % (plan_v2.name, len(plan_v2.points)))
|
||
print("证书采样分辨率 %g m / %d 档，目标 (p,h) %d 个，缺口 %d，最坏命中 %.3f m"
|
||
% (cert_v2.target_step_m, cert_v2.headings, cert_v2.targets,
|
||
cert_v2.uncovered, cert_v2.worst_hit_distance_m))
|
||
print("最近邻巡回 %.1f km（虚拟移动 %.1f s）→ 发现阶段的移动代价下界"
|
||
% (cert_v2.nn_tour_m / 1000, cert_v2.nn_tour_m / CONFIG.cost.speed_mps))
|
||
print("设计裕量：r_design %.0f m、前向角裕量 %.1f°；校核用 r_eff %.0f m、角裕量 %.1f°"
|
||
% (plan_v2.generator["design_r_eff_m"], plan_v2.generator["heading_margin_deg"],
|
||
cert_v2.r_eff_min_m, cert_v2.verify_heading_margin_deg))
|
||
assert cert_v2.uncovered == 0 and cert_v2.targets == 3256200
|
||
|
||
plan_v1 = plan_instance("q4_mincover_v1", verify=False)
|
||
print("\n对照 %s：同样 %d 点，但证书采样只有 %g m / %d 档，细网格复检有 125 个缺口"
|
||
% (plan_v1.name, len(plan_v1.points), plan_v1.generator["target_step_m"],
|
||
plan_v1.generator["headings"]))
|
||
print("→ 采样分辨率必须与证书一致，否则'删掉一个关键点仍看似覆盖'（校核退化为装饰）")
|
||
'''))
|
||
|
||
# =============================================================== 10 问题四
|
||
add(md(r"""
|
||
## 10. 问题四：含定向源的闭环
|
||
|
||
定向源的难点是**静默的歧义**：`no_signal` 既可能是距离不足，也可能是朝向遮挡。
|
||
本问的关键推理与对策：
|
||
|
||
1. **联合可行集**：给每个幸存格附 24 档朝向掩码。`direction`/`near` 要求朝向
|
||
*可能*覆盖读点；`no_signal`（且整格必在 1000 m 内）要求朝向*确定不*覆盖读点。
|
||
真源 (位置, 朝向) 永不被删；联合集为空即缺席证明。
|
||
2. **可见性选点**：`prob_detect` 给出"在该点能看见任一可行 (位置,朝向) 的概率质量"，
|
||
先在可见候选中过滤（`prob_detect ≥ ratio × max`），再取"看得见且少走路"的点，
|
||
取代固定 45° 阶梯试探。
|
||
3. **后备清除**：定位停滞时对有界候选集构造 ≤20 m 有限圆盘覆盖逐格 `/clear`——
|
||
清除只与距离有关、与朝向无关，是完备兜底（格数即请求上界，超限即拒绝盲扫）。
|
||
"""))
|
||
|
||
add(code(r'''
|
||
# ---- 10.1 定向覆盖最小命中集的覆盖语义 --------------------------------------
|
||
plan_q4 = plan_instance("q4_mincover_v2", verify=False)
|
||
coarse = verify_forward_half_disk(plan_q4.points, step_m=300.0, headings=12,
|
||
r_eff=R_EFF_MIN_M, heading_margin_deg=0.0)
|
||
print("最小命中集 %d 点，粗采样复核 (p,h) %d 个：未覆盖 %d，最坏命中 %.1f m"
|
||
% (len(plan_q4.points), coarse.targets, coarse.uncovered,
|
||
coarse.worst_hit_distance_m))
|
||
square49 = square_49_plan()
|
||
coarse49 = verify_forward_half_disk(square49.points, step_m=300.0, headings=12)
|
||
print("对照 700 m 方格 %d 点：同样未覆盖 %d，巡回 %.1f km vs %.1f km"
|
||
% (len(square49.points), coarse49.uncovered,
|
||
coarse49.nn_tour_m / 1000, coarse.nn_tour_m / 1000))
|
||
assert coarse.uncovered == 0 and coarse.targets > 0
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# ---- 10.2 (位置 × 朝向) 联合可行集的保守性随机检验 --------------------------
|
||
truth4 = (420.0, -260.0) # 真源位置
|
||
heading4 = 200.0 # 真源朝向（背向圆心发射的边界工况）
|
||
r_eff4 = 1100.0 # 真源的有效接收半径（策略端不可见）
|
||
|
||
|
||
def station_at(bearing_from_truth_deg, distance_m):
|
||
"""按"从真源看过去的方位"构造测点，便于控制可见/遮挡工况。"""
|
||
return (truth4[0] + distance_m * math.cos(math.radians(bearing_from_truth_deg)),
|
||
truth4[1] + distance_m * math.sin(math.radians(bearing_from_truth_deg)))
|
||
|
||
|
||
# 前向半球内的测点（|方位 − 朝向| ≤ 90°）：应收到 direction；
|
||
# 超过 1100 m 的测点：无论朝向如何都 no_signal。
|
||
CASES = [(heading4 - 85.0, 900.0), (heading4 + 85.0, 700.0), (heading4, 600.0),
|
||
(heading4 + 30.0, 1050.0), (heading4 - 60.0, 400.0), (heading4 + 130.0, 800.0),
|
||
(heading4, 1250.0)]
|
||
|
||
joint = JointBelief(CONFIG)
|
||
half_bin = 0.5 * 360.0 / HEADING_BINS # 档位半宽 7.5°：真实朝向必落在某一档内
|
||
expected = ["direction"] * 5 + ["no_signal"] * 2
|
||
for (bearing, dist_m), want in zip(CASES, expected):
|
||
station = station_at(bearing, dist_m)
|
||
result, svd = observe_directional(truth4, heading4, station, r_eff4)
|
||
joint.apply(station, result, svd)
|
||
cell = min(joint.position.alive,
|
||
key=lambda k: math.dist(joint.position.cell_center(k), truth4))
|
||
offset = math.dist(joint.position.cell_center(cell), truth4)
|
||
mask = joint._mask(cell)
|
||
keeps = any(heading_gap(360.0 * k / HEADING_BINS, heading4) <= half_bin
|
||
for k in range(HEADING_BINS) if mask >> k & 1)
|
||
print("测点方位 %6.1f° 距离 %5.0f m → %-9s（预期 %-9s）幸存格 %6d "
|
||
"真源格偏移 %6.3f m（裕量 %.3f m）朝向档 %2d 个 含真源朝向档：%s"
|
||
% (bearing % 360.0, dist_m, result, want, len(joint), offset,
|
||
joint.position.margin_m, bin(mask).count("1"), keeps))
|
||
assert result == want, "合成观测与题面语义不一致"
|
||
assert keeps, "联合信念删掉了真源的朝向档（保守性被破坏）"
|
||
assert offset <= joint.position.margin_m + 1e-9, \
|
||
"联合信念删掉了真源的位置格（Voronoi 裕量被破坏）"
|
||
|
||
print("\n联合集规模 %d 格，抽样平均朝向档 %.2f / %d"
|
||
% (len(joint), joint.summary()["sampled_mean_heading_bins"], HEADING_BINS))
|
||
assert not joint.is_absent()
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# ---- 10.3 可见性选点取代固定阶梯；后备清除兜底 ------------------------------
|
||
print("探点可见性概率（越高越可能在该点收到信号）：")
|
||
for candidate in [(-2000.0, 0.0), (0.0, 0.0), (420.0, -260.0), (1500.0, -900.0),
|
||
(420.0 + 800.0 * math.cos(math.radians(heading4)),
|
||
-260.0 + 800.0 * math.sin(math.radians(heading4)))]:
|
||
print(" (%8.1f, %8.1f) prob_detect = %.4f"
|
||
% (*candidate, joint.prob_detect(candidate)))
|
||
|
||
# 定位停滞时的后备：候选集有界 -> 20 m 网格覆盖，格心到格内任一点 <= 10√2 < 20 m
|
||
region4 = CandidateRegion.initial((0.0, 0.0), CONFIG)
|
||
region4.clip_wedge((0.0, 0.0), 45.0, 1.0 + CONFIG.search.bearing_rounding_margin_deg)
|
||
region4.clip_disk((900.0, 900.0), 900.0, CONFIG.search.polygon_sides)
|
||
cells = fallback_clear_cells(region4.polygon, CONFIG)
|
||
width = max(p[0] for p in region4.polygon) - min(p[0] for p in region4.polygon)
|
||
height = max(p[1] for p in region4.polygon) - min(p[1] for p in region4.polygon)
|
||
print("\n候选集外包框 %.0f m × %.0f m，20 m 网格覆盖 %d 格（外包框枚举上界 %d）"
|
||
% (width, height, len(cells), fallback_cell_bound(width, height, CONFIG)))
|
||
print("格内任意点到最近格心 <= %.2f m < 20 m → 逐格 /clear 至少命中一次"
|
||
% (CONFIG.search.fallback_clear_grid_m * math.sqrt(2) / 2))
|
||
assert cells and len(cells) <= CONFIG.search.fallback_max_cells
|
||
|
||
# 覆盖完备性抽查：在候选集外包框内随机撒点，每个点都必落在某个格心 20 m 内
|
||
rng = Random(7)
|
||
covered = 0
|
||
for _ in range(2000):
|
||
u, v = rng.random(), rng.random()
|
||
sample = (min(p[0] for p in region4.polygon) + u * width,
|
||
min(p[1] for p in region4.polygon) + v * height)
|
||
if not region4.contains(sample, tol_m=0.0):
|
||
continue # 框内但在候选集外：不计入
|
||
covered += min(math.dist(sample, cell) for cell in cells) <= 20.0
|
||
print("外包框内落在候选集中的随机点，被某个格心 20 m 覆盖：%d 个" % covered)
|
||
assert covered > 0
|
||
'''))
|
||
|
||
add(code(r'''
|
||
# ---- 10.4 第四问闭环装配（不连模拟器：只检查策略对象初始化） ----------------
|
||
client4 = SimulatorClient("OFFLINE", config=CONFIG, transport=None, log=_LogStub())
|
||
client4.state.position = (0.0, 0.0)
|
||
loop4 = DirectionalExtension(client4, CONFIG)
|
||
print("第四问策略：覆盖计划 %s（%d 点），联合信念开关 %s，探点口径 %s"
|
||
% (loop4.coverage_plan.name, len(loop4.coverage_plan.points),
|
||
CONFIG.search.directional_joint_belief, CONFIG.search.probe_score_mode))
|
||
print("后备清除硬顶 %d 格，单轮遮挡探点预算 %d 次，定位步数上限 %d"
|
||
% (CONFIG.search.fallback_max_cells, CONFIG.search.max_blocked_probes,
|
||
CONFIG.search.max_localize_steps))
|
||
|
||
# 保证接收盘内的 no_signal 只能由朝向遮挡解释：模拟"被遮挡"的梯子升级
|
||
blocked_point = (600.0, 600.0)
|
||
loop4._observe(3, blocked_point, "direction",
|
||
bearing_deg(blocked_point, truth4))
|
||
cert4 = clear_certificate(loop4.belief.candidate_set(3), 3, CONFIG)
|
||
print("\n频道 3 候选集半径 %.1f m，可清除：%s" % (cert4.radius_m, cert4.certified))
|
||
safe = safe_disk_radius(loop4.belief.candidate_set(3), CONFIG)
|
||
print("保证接收盘半径 %.1f m（盘内必收到信号 -> 静默即遮挡）" % safe)
|
||
loop4._observe(3, blocked_point, "no_signal", None)
|
||
next_probe = loop4._next_probe(loop4.belief.candidate_set(3), 3)
|
||
print("被遮挡后梯子升级到的探点：(%8.2f, %8.2f)"
|
||
% next_probe)
|
||
assert loop4.joint.get(3) is not None and not loop4.joint[3].is_absent()
|
||
'''))
|
||
|
||
# =============================================================== 11 收尾
|
||
add(md("""
|
||
# 第四部分 · 总览与自检
|
||
|
||
## 11. 本文件包含 / 不包含的内容
|
||
|
||
**包含**（四问生产路径）：
|
||
|
||
| 层 | 模块 | 内容 |
|
||
| --- | --- | --- |
|
||
| 参数 | `config.py` | 题面常量 + 工程参数 |
|
||
| 协议 | `protocol/` | 规则表、计时模型、状态机、传输、会话日志、串行幂等客户端 |
|
||
| 几何 | `geometry/` | 角域交会定位、直径圆、最小包围圆、保证接收盘证书 |
|
||
| 信念 | `belief/` | 保守凸候选集、(位置×朝向) 联合可行集 |
|
||
| 滤波 | `particles.py` | 网格幸存集、负观测更新、跳测规则、信息增益评分 |
|
||
| 保证 | `coverage/` `guarantee/` | 覆盖账本、覆盖计划与证书、最小命中集 |
|
||
| 调度 | `routing.py` | 开放路径排序（最近邻 + 2-opt） |
|
||
| 策略 | `planner/` | 问题二选点器、全向闭环骨架、贝叶斯闭环（第三问默认）、定向扩展（第四问） |
|
||
|
||
**不包含**（按要求剔除）：
|
||
|
||
- `referee/`：独立评测机（模拟器引擎 + 机器狗接口 + WebUI）——按附件协议独立复现，
|
||
与本队求解代码零耦合，不属求解方案；
|
||
- `src/mcm26b/harness/`：测试侧内存回环、故障注入、日志回放、场景生成；
|
||
- `src/mcm26b/cli.py`、`geometry/__main__.py`、`main.py`：命令行外壳；
|
||
- `scripts/`：采集、基准、Optuna 调参、可视化、论文插图与校核脚本；
|
||
- `project/`：1-4 问"最优实现 / 基线实现"对比框架；`tests/`：单元测试；
|
||
- 未被生产路径调用的函数：`routing.py` 的离线度量族（`held_karp_open` /
|
||
`optimal_open_path` / `or_opt_open` / `mst_weight` / `route_metrics`，
|
||
仅用于事后路由下界审计）、`guarantee/cover.py` 的计划落盘 IO、
|
||
`planner/seven.py` 的未采用变体（`SevenImmediate` / `SevenSweep`）。
|
||
这些在仓库中保留并有测试守护，默认关闭或不被实例化。
|
||
|
||
## 12. 运行口径
|
||
|
||
```bash
|
||
# 对接官方模拟器（演练/正式）
|
||
python -m mcm26b.cli run --robot-id <队号> --problem 3 # 第三问：bayes 闭环
|
||
python -m mcm26b.cli run --robot-id <队号> --problem 4 # 第四问：directional 闭环
|
||
```
|
||
|
||
成绩口径是**虚拟时间**：`avg_locate_clear_time_s = 全程虚拟时长 / 已清除源数`，
|
||
找全证明达成后立即 `/exit` 而不是复扫到限时。
|
||
"""))
|
||
|
||
add(code(r'''
|
||
# ---- 总自检：把四问的核心结论再跑一遍，全部通过才算整理成功 ----------------
|
||
checks = []
|
||
|
||
q1 = solve_q1("normal")
|
||
checks.append(("问1 角域交会有界且直径有限",
|
||
q1["status"] == "polygon" and q1["diameter_infinite"] is False))
|
||
checks.append(("问1 三点交会可原地清除（MEC ≤ 20 m）",
|
||
q1["guaranteed_clearable_at_mec_center"] is True))
|
||
checks.append(("问1 直径圆反例被识别（covers=False）",
|
||
solve_q1("triangle")["diameter_circle"]["covers"] is False))
|
||
|
||
plan2 = plan_second_point(station1, bearing1)
|
||
checks.append(("问2 第二点在保证接收盘内",
|
||
math.dist(plan2.position, region.mec()[0])
|
||
<= safe_disk_radius(region, CONFIG)))
|
||
checks.append(("问2 追加三点后最坏后验半径 ≤ 20 m",
|
||
plan.worst_posterior_radius_m <= CONFIG.field.clear_radius_m))
|
||
|
||
checks.append(("问3 七站解析覆盖上界 < 1000 m",
|
||
bound < CONFIG_Q3.field.r_eff_min_m))
|
||
checks.append(("问3 七站计划粗采样无缺口", cert7.uncovered == 0))
|
||
checks.append(("问3 粒子滤波保留真源", retained))
|
||
checks.append(("问3 空幸存集给出缺席证明", len(absent) == 0))
|
||
checks.append(("问3 2-opt 不劣于原任务序",
|
||
loop3._path_cost(order) <= loop3._path_cost(tasks) + 1e-9))
|
||
|
||
checks.append(("问4 最小命中集 31 点、证书 0 缺口",
|
||
len(plan_q4.points) == 31 and cert_v2.uncovered == 0))
|
||
checks.append(("问4 粗采样复核定向覆盖无缺口", coarse.uncovered == 0))
|
||
checks.append(("问4 联合信念保留真源朝向档", not joint.is_absent()))
|
||
checks.append(("问4 后备清除格数在硬顶内",
|
||
len(cells) <= CONFIG.search.fallback_max_cells))
|
||
checks.append(("问4 闭环装配成功（定向覆盖计划已加载）",
|
||
loop4.coverage_plan.name.startswith("q4_mincover")))
|
||
|
||
print("四问核心自检：")
|
||
for name, ok in checks:
|
||
print(" [%s] %s" % ("通过" if ok else "失败", name))
|
||
assert all(ok for _, ok in checks), "自检未全部通过"
|
||
print("\n全部 %d 项通过。" % len(checks))
|
||
'''))
|
||
|
||
|
||
def build(path: Path) -> None:
|
||
notebook = {
|
||
"cells": CELLS,
|
||
"metadata": {
|
||
"kernelspec": {"display_name": "Python 3", "language": "python",
|
||
"name": "python3"},
|
||
"language_info": {"name": "python", "version": "3.10",
|
||
"mimetype": "text/x-python",
|
||
"file_extension": ".py",
|
||
"pygments_lexer": "ipython3",
|
||
"nbconvert_exporter": "python",
|
||
"codemirror_mode": {"name": "ipython", "version": 3}},
|
||
},
|
||
"nbformat": 4,
|
||
"nbformat_minor": 5,
|
||
}
|
||
path.write_text(json.dumps(notebook, ensure_ascii=False, indent=1) + "\n",
|
||
encoding="utf-8")
|
||
code_lines = sum(len(cell["source"]) for cell in CELLS
|
||
if cell["cell_type"] == "code")
|
||
md_cells = sum(1 for cell in CELLS if cell["cell_type"] == "markdown")
|
||
print(f"已写出 {path}：{len(CELLS)} 个单元"
|
||
f"（markdown {md_cells}，代码 {len(CELLS) - md_cells}，代码行 {code_lines}）")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(description="生成四问核心代码单文件 notebook")
|
||
parser.add_argument("-o", "--output", type=Path,
|
||
default=ROOT / "mcm26b-核心代码.ipynb")
|
||
build(parser.parse_args().output)
|