"""把 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/.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("2026001", 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("2026001", 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.run --robot-id <队号> --problem 3 # 第三问：bayes 闭环 python -m mcm26b.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)