323 KiB
323 KiB
In [ ]:
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 回退）")
In [ ]:
Vec2 = tuple[float, float]
@dataclass(frozen=True)
class FieldConfig:
"""题目物理世界与协议硬限制。[题面]"""
omega_radius_m: float = 1800.0 # 目标区域半径，圆心为原点
channel_min: int = 1 # 有效频道范围
channel_max: int = 20
source_count_min: int = 10 # 一局源总数范围（接口不直接告知）
source_count_max: int = 16
r_eff_min_m: float = 1000.0 # 有效接收半径下界：no_signal 只能排除该圆盘
r_eff_max_m: float = 1500.0 # 有效接收半径上界：direction 隐含距离上界
near_radius_m: float = 5.0 # "距离过近"阈值，不返回示向度
clear_radius_m: float = 20.0 # 清除半径；与定向朝向无关
bearing_error_half_deg: float = 1.0 # 示向度误差幅度（题设模型）
bearing_decimals: int = 2 # 返回示向度小数位
coord_abs_limit_m: float = 2_000_000.0 # 提交坐标的绝对值上限（超出返回 HTTP 400）
directional_half_angle_deg: float = 90.0 # 定向源朝向两侧各90度覆盖（含边界）
max_virtual_duration_s: float = 360_000.0 # 虚拟世界限时（100小时）
max_real_duration_s: float = 1200.0 # /enter 成功后程序运行上限（20分钟）
test_window_s: float = 1500.0 # 确认开始后的测试窗口（25分钟）
@dataclass(frozen=True)
class CostConfig:
"""虚拟世界动作耗时模型。[题面] 附件1表1、附件2第4节。"""
speed_mps: float = 5.0 # 移动速度；耗时=直线距离/speed
measure_s: float = 5.0 # 每次检测动作耗时（无论结果）
switch_s: float = 1.0 # 任意两频道间切换耗时，仅由合法 /measure 触发
clear_success_s: float = 5.0 # 清除成功耗时 = 光学定位3s + 激光清除2s
clear_fail_s: float = 3.0 # 未发现可清除目标 = 仅光学定位3s
@dataclass(frozen=True)
class SessionConfig:
"""会话与网络策略。arena/robot/request 规则为 [题面]，其余为 [工程]。"""
arena_id: str = "default" # [题面] 固定值
base_url: str = "http://127.0.0.1:2026" # [题面] 模拟器默认接口地址
robot_id_max_bytes: int = 64 # [题面] UTF-8 长度 1..64
request_id_max_bytes: int = 128 # [题面] UTF-8 长度 1..128
request_body_max_bytes: int = 65_536 # [题面] 超出返回 HTTP 413
connect_timeout_s: float = 5.0 # [工程] 连接超时
read_timeout_s: float = 15.0 # [工程] 读超时；覆盖"服务器已处理但响应丢失"
max_attempts: int = 3 # [工程] 单动作最多尝试次数（含首次）
retry_backoff_s: float = 0.25 # [工程] 重试退避基数，按 2^n 放大
reserve_exit_s: float = 60.0 # [工程] 预留退出与故障恢复的现实秒数
real_rtt_p95_s: float = 1.0 # [工程] 现实单请求高分位估计，预算核算用
# [工程] 自检/演练实机预算：题面正式测试程序运行上限 1200s 不可延长；
# 自检经 --real-budget-s 覆盖 LocalSimulator 返回的 remaining_real_duration_s，
# 默认 1800s（30 分钟实机），用于验证"限时内持续运行直到找全"的行为。
self_test_real_budget_s: float = 1800.0
@dataclass(frozen=True)
class SearchConfig:
"""搜索与覆盖规划参数。[工程] 推导依据见 docs/B题建模分析与实施计划.md。"""
grid_step_m: float = 700.0 # P4 覆盖网格步长；单元对角 700*sqrt(2)≈989.95<=1000
grid_half_cells: int = 3 # i,j ∈ [-3,3]，49 点覆盖 [-2100,2100]^2 ⊇ Ω
fallback_clear_grid_m: float = 20.0 # P5 后备清除网格；10*sqrt(2)<=20
# 舍入裕量：服务器对示向度保留两位小数，引入 <=0.005° 偏差 [题面]；
# 不加裕量时合法真源可能被角域裁掉，故库默认与 CLI run 默认一致取 0.005 [工程]。
bearing_rounding_margin_deg: float = 0.005
# ---- P3/P4 在线定位与闭环参数 [工程] ----
polygon_sides: int = 192 # 圆盘外切多边形边数；Ω 近似误差 1800(1-cos(π/192))≈0.24m
witness_spacing_m: float = 60.0 # 最坏后验评估的边界采样间距；选定后再按 1/4 加密复核
candidate_rings: tuple = (0.0, 0.4, 0.75, 1.0) # 安全接收盘内候选点的半径比例
candidate_angles: int = 12 # 每环候选方向数
clear_margin_m: float = 0.5 # 清除证书的额外工程余量（多边形近似误差之外）
max_localize_steps: int = 8 # 单频道定位阶段允许的追加测量上限
max_anomalies_per_channel: int = 3 # 单频道异常记录上限，防止异常刷屏
# ---- P5 定向扩展参数 [工程] ----
max_blocked_probes: int = 4 # 定位单轮内被朝向遮挡的探点预算
# （首次选点+对跖复核+2 次环探；达限转后备清除）
fallback_max_cells: int = 400 # 后备清除格数硬顶；超过则不盲扫（宽候选集继续定位）
# ---- 覆盖计划（保证层）[工程，2026-09-12 架构优化 P0-2 / S2 硬化] ----
# 第四问发现点集：
# square49 旧的充分条件表（700m 方格，巡回 36.4km）；
# q4_mincover_v1 75m/24 档贪心最小集（31 点，巡回 22.9km）——细网格复检
# 在 25m/48 档有 125 个 (p,h) 缺口，是采样过拟合解；
# q4_mincover_v2 **设计裕量**最小集（31 点，r_design=965m、前向角裕量 5°，
# 搜索分辨率 50m/48 档 + 证书分辨率精细剪枝），
# 15m/72 档（325.6 万 (p,h)）缺口 0，最坏命中 990m，巡回 23.9km。
# 运行期按同一采样语义复核，未通过即拒绝启动。
q4_coverage_plan: str = "q4_mincover_v2"
verify_coverage_plan: bool = True # 运行期抽样校验（进程内缓存一次）
# ---- 第四问定向源联合信念 [工程，2026-09-12 架构优化 P0-3] ----
# 为已发现的定向频道维护 (位置 × 朝向) 联合可行集：用可见性概率选点，
# 取代固定 45° 阶梯探点；联合集空即给出缺席证明。安全性仍由凸/网格
# 清除证书与覆盖证明负责，本开关只影响选点顺序。
directional_joint_belief: bool = True
# ---- 第四问定向源探点值函数 [工程，2026-09-12 图论与 DP 方案 S3/D1] ----
# 旧评分 ``prob_detect/(移动+6)`` 只问"这点看不看得见源"，不问"测完剩多少
# 不确定度"，会反复选中同一个近处被遮挡点。D1 口径：先按可见概率过滤
# （只保留 prob_detect ≥ ratio×最大值的候选），再最小化
# E[后验 MEC 半径](q) + λ·(当前位置→q→MEC 圆心 的移动秒数 + 检测)
# 9 场景 core 配对（修正联合信念朝向规则之后）：
# λ=0.05：后备 6.8 次 / T 9740 s（值函数主导，为信息多绕路）
# λ=5 ：后备 4.3 次 / T 8604 s（移动项主导，可见性过滤 + 就近抵近）
# 仅移动项（nearest_visible）：后备 4.3 次 / T 8602 s，墙钟最低
# 相对旧 visibility 口径（后备 125.6 次、T 9225 s）三者都更好，且 λ 越大
# 越优说明本题"看得见 + 少走路"比"信息量最大"更重要，故默认取极限口径。
probe_score_mode: str = "nearest_visible" # visibility | expected_radius | nearest_visible
probe_lambda_move_s: float = 5.0 # expected_radius 口径下的移动折算
# 可见概率过滤：只保留 prob_detect ≥ ratio×最大值的候选再比期望半径，
# 避免"期望上有点信息、实际必被朝向遮挡"的探点。
probe_visibility_ratio: float = 0.7
probe_exclude_blocked: bool = True # 已测过且被遮挡的探点不再重复选取
approach_fractions: tuple = () # D1 抵近点插值比例（0=圆心）；默认关闭
# ---- 后备清除路径优化 [工程，2026-09-12 S4/D3] ----
# 实测反例：后备清除在命中真源时立即结束，"整条走格路径最短"并不是正确
# 目标——Or-opt 会把真源推到序列更后面。9 场景配对：最近邻 125.7 次失败
# vs Or-opt 161.4 次（T 8676 → 8940 s）。默认关闭，保留代码与开关。
fallback_optimize_tour: bool = False
# ---- 持续搜索 [工程，2026-09-11 用户决策修订] ----
# 虚拟时间即评测成绩口径（表1：平均定位清除时间）：找全证明达成后
# 立即 /exit，不复扫。复扫保险模式（防缺席证明假设偏差）保留，经
# `mcm26b run --persistent` 显式开启：只复扫未清除频道，直到清除数
# 达 16、实机预算耗尽或虚拟安全余量不足。
persistent_search: bool = False
virtual_reserve_s: float = 10_000.0 # 虚拟时长安全余量：剩余不足时停止发起动作，
# 避免触及 100 小时上限被模拟器强制结束
@dataclass(frozen=True)
class BayesConfig:
"""P4b 贝叶斯网格滤波策略参数（问题3新策略 bayes）。[工程]
网格粒子滤波的严格性依赖 Voronoi 裕量：真实源所在格心与源的距离
<= sqrt(2)/2 * particle_spacing_m，所有 kill 判定都内缩该裕量，
因此幸存集永远包含真源（定理见 docs/algo-3-bayes.md §2）。
"""
particle_spacing_m: float = 14.65 # 网格间距；Voronoi 裕量 ≈ 10.35m；
# 全域约 4.7 万格/频道，查询走粗格索引
# ---- 信息增益评分 [工程，Optuna TPE 调参回填 2026-09-11] ----
# 原调参记录见 output/q3/bayes-tuning/；当前代码的配对验收记录见
# output/q3/optimization-20260911/，随机分布与信息评分仅为工程假设。
score_mode: str = "entropy" # entropy=期望 log 幸存占比（信息增益）|
# radius=期望后验半径（直指证书）
score_subsample: int = 800 # 期望幸存熵估计的粒子子采样规模
mc_representatives: int = 7 # 假设真值代表点数（均匀抽取幸存格）
lambda_move_s: float = 0.05 # 选点评分中移动时间的折算系数（分值/秒）；
# 2026-09-12 配对复跑：0.05 优于 0.0105（20 场均值 −2.8%），
# entropy 口径下信息项为 O(1) 对数值，λ须小型但不可过小
# ---- 路线感知探点 [工程，2026-09-12 P2] ----
# 旧评分只看 current->q + q->MEC，容易选到“信息增益略高但要多绕几百米”
# 的探点。开启后，在候选评分中加入 q 替换当前位置后对当前任务路线代价
# 的增量；A/B 以冻结基准集配对验收。
# 2026-09-12 冻结基准集复跑（18 场景）：关闭 3193.1 s vs 开启 3210.3 s，
# 移动 12339.8 m vs 12446.8 m —— 权重 1e-3 的路线项未带来净收益，默认关闭。
route_aware_probe: bool = False
route_aware_insert_weight: float = 0.001 # 路线代价增量的折算系数（小权重，仅微调）
route_aware_clear_weight: float = 0.0 # 额外 q->MEC 清除接近项（基础移动项已含）
wide_rings: tuple = (1.0, 1.5, 2.0) # 无保证接收盘时的探索环（× PF MEC 半径）
# ---- 站点与覆盖 [工程] ----
station_ring_radius_m: float = 1159.7 # 七点环半径；构造时检验全向覆盖解析上界
# ---- 发现-定位融合巡回 [工程，2026-09-12 P0-1，实测无效，默认关闭] ----
# 设想：站点扫频顺带取第二/第三条方位线，取代专用探点。
# 实测（output/suite/q3-fuse-ab*，20 场景配对，2026-09-12）：
# 仅追加站点方位线：3454.6s vs 3426.9s（+0.8%，检测 +10.8 次，位移仅 −188m）；
# 追加 + 抑制专用探点：4989.0s vs 3426.9s（+45.6%，20/20 场景更慢）。
# 原因：七站环上只有 2–3 个站能对同一源取得有效方位线，且站间交会角
# 不足以保证 20m 证书，专用信息增益探点仍然必要；抑制探点只会白等巡回。
# 代码保留以便复现与后续研究，成绩口径必须保持 False。
fuse_localize_into_tour: bool = False
# ---- 任务排序 [工程] ----
two_opt_enabled: bool = True # 开放路径 2-opt 改进（最近邻打底）
two_opt_passes: int = 4 # 2-opt 最大改进轮数
# ---- 精确任务序（Held-Karp，含换频状态）[工程，2026-09-12 S5/T1] ----
# 任务表典型 6–12 项：|T| ≤ exact_order_limit 时用精确 DP（状态=已访问集合
# × 结束时测向频道），换频代价精确计入；否则回退 NN + 2-opt。0 表示禁用。
# 实测（18 场景）：精确 DP 3203.2 s vs 2-opt 3193.1 s —— 逐位最优的"当前
# 任务表"在滚动时域下反而略差（贪心排序的次优性起到探索作用），默认 0；
# 代码与对拍测试保留，可随时开启复核。
exact_order_limit: int = 0
# ---- PF 清除证书 [工程] ----
grid_cert_enabled: bool = True # 允许用幸存集 MEC 颁发清除证书（凸证书之外）
opportunistic_measure: bool = True # 已停靠位置顺路交会；仍逐频道计费
# 预测后验半径至少减少 (1−ratio) 才补测；0.5 优于 0.65（18 场景 3205.4 vs 3210.3 s）
opportunistic_radius_ratio: float = 0.5
@dataclass(frozen=True)
class ProblemConfig:
"""聚合配置。各模块函数的第一个默认参数，便于单测替换。"""
field: FieldConfig = FieldConfig()
cost: CostConfig = CostConfig()
session: SessionConfig = SessionConfig()
search: SearchConfig = SearchConfig()
bayes: BayesConfig = BayesConfig()
DEFAULT_CONFIG = ProblemConfig()
In [ ]:
@dataclass(frozen=True)
class ProblemConfig:
"""聚合配置。各模块函数的第一个默认参数，便于单测替换。"""
field: FieldConfig = FieldConfig()
cost: CostConfig = CostConfig()
session: SessionConfig = SessionConfig()
search: SearchConfig = SearchConfig()
bayes: BayesConfig = BayesConfig()
DEFAULT_CONFIG = ProblemConfig()
In [ ]:
# ---- 结果码 [题面] ----------------------------------------------------------
NO_SIGNAL = "no_signal" # 收不到信号：无未清除源 / 超接收半径 / 定向覆盖角外
NEAR = "near" # 距离 <=5m 且在覆盖角内；不返回 svd_deg
DIRECTION = "direction" # 有信号；附 svd_deg（含 ±1° 误差，保留两位小数）
MEASURE_RESULTS = (NO_SIGNAL, NEAR, DIRECTION)
CLEAR_SUCCESS = "success" # 20m 内发现并清除；耗时 5s
CLEAR_NO_TARGET = "no_target_in_range" # 未发现（无源/已清除/超20m）；耗时 3s
CLEAR_RESULTS = (CLEAR_SUCCESS, CLEAR_NO_TARGET)
EXIT_USER = "user_exit" # /exit 成功响应中的固定退出原因
# accepted=false 的响应只含这 3 个字段；virtual_time_s 恒为 0，不是当前时刻 [题面]
BASE_RESPONSE_FIELDS = ("accepted", "real_timestamp_ms", "virtual_time_s")
# ---- 请求字段表 --------------------------------------------------------------
@dataclass(frozen=True)
class FieldSpec:
"""单个请求字段的约束。[题面] 附件2 5.1 节与各指令小节。"""
kind: type # str / dict；数值字段由 sim 按整型/浮点细则单独校验
required: bool = True
max_bytes: int | None = None # 字符串字段的 UTF-8 长度上限
COMMON_FIELDS: dict[str, FieldSpec] = {
"arena_id": FieldSpec(str), # 必须为 "default"
"robot_id": FieldSpec(str, max_bytes=64), # 逐字节等于登录参赛队号
"request_id": FieldSpec(str, max_bytes=128), # 会话内幂等键，新动作用新 ID
}
POSITION_FIELDS: dict[str, FieldSpec] = {
"x": FieldSpec(float), # 有限 JSON number，|x| <= 2_000_000
"y": FieldSpec(float),
}
# 路径 -> 除公共字段外的动作字段；/enter 与 /exit 无附加字段
ACTION_FIELDS: dict[str, dict[str, FieldSpec]] = {
"/enter": {},
"/measure": {"position": FieldSpec(dict), "channel": FieldSpec(float)},
"/clear": {"position": FieldSpec(dict), "channel": FieldSpec(float)},
"/exit": {},
}
ACTION_PATHS = tuple(ACTION_FIELDS)
# channel 写成 JSON number 1.0 且恰为整数也接受；1.5 返回 HTTP 400 [题面]
# ---- 响应字段表（accepted=true 时的附加字段） --------------------------------
RESPONSE_FIELDS: dict[str, tuple[str, ...]] = {
"/enter": ("max_virtual_duration_s", "max_real_duration_s", "remaining_real_duration_s"),
"/measure": ("measure_result",), # direction 时另有 svd_deg
"/clear": ("clear_result",),
"/exit": ("exit_reason",),
}
# ---- 错误语义 [题面] 附件2 5.3 节 --------------------------------------------
STATUS_OK = 200
STATUS_BAD_REQUEST = 400 # JSON 语法/重复键/缺字段/类型/频道或坐标范围
STATUS_CONFLICT = 409 # 同 request_id 对应不同动作，或并发不同动作
STATUS_RETRYABLE = (429, 500) # 流量保护/幂等记录上限、服务器内部错误：可原样重试
def http_status_meaning(status: int) -> str:
table = {
200: "动作已执行，或 JSON 合法但被业务拒绝（看 accepted）",
400: "请求结构错误；不占用 request_id，修正后可复用该 ID",
404: "路径未知或不精确（不接受尾随斜线与查询参数）",
405: "已知路径使用了非 POST 方法",
409: "request_id 幂等冲突或并发发送不同动作",
413: "请求体超过 65536 字节",
415: "Content-Type 或 Content-Encoding 不受支持",
429: "连接/流量保护触发，或本局幂等记录达到上限",
500: "模拟器内部错误",
}
return table.get(status, "未知状态码")
In [ ]:
@dataclass(frozen=True)
class CostBreakdown:
"""一次合法动作的三段耗时 [题面]：移动 + 换频道（仅检测）+ 动作本身。"""
move_s: float
switch_s: float
action_s: float
@property
def total_s(self) -> float:
return self.total_us / 1_000_000
@property
def total_us(self) -> int:
"""逐段量化后按整数微秒累计，与官方虚拟时钟一致。"""
return sum(round(value * 1_000_000)
for value in (self.move_s, self.switch_s, self.action_s))
def measure_cost(prev_pos: Vec2, pos: Vec2, prev_channel: int, channel: int,
cost: CostConfig = CostConfig()) -> CostBreakdown:
"""合法 /measure 的耗时。移动按直线距离；换频道只看前后频道是否不同。"""
return CostBreakdown(
move_s=dist(prev_pos, pos) / cost.speed_mps,
switch_s=0.0 if channel == prev_channel else cost.switch_s,
action_s=cost.measure_s,
)
def clear_cost(prev_pos: Vec2, pos: Vec2, success: bool,
cost: CostConfig = CostConfig()) -> CostBreakdown:
"""合法 /clear 的耗时。
/clear 的 channel 是目标干扰源频道，不切换测向机频道，也不产生换频道耗时 [题面]。
"""
return CostBreakdown(
move_s=dist(prev_pos, pos) / cost.speed_mps,
switch_s=0.0,
action_s=cost.clear_success_s if success else cost.clear_fail_s,
)
def move_time(a: Vec2, b: Vec2, cost: CostConfig = CostConfig()) -> float:
"""任意两点间的移动耗时（规划用）。"""
return dist(a, b) / cost.speed_mps
In [ ]:
class StateError(RuntimeError):
"""状态机被以非法顺序驱动（例如未 enter 就 apply_measure）。"""
@dataclass
class RobotState:
position: Vec2 = (0.0, 0.0) # /enter 成功后的初始位置 [题面]
channel: int = 1 # /enter 成功后的初始频道 [题面]
virtual_time_s: float = 0.0
entered: bool = False
exited: bool = False
action_count: int = 0
# 预测与实测的偏差记录；strict 模式下出现偏差会直接抛错。
mismatches: list = field(default_factory=list)
# -------------------------------------------------------------- enter/exit
def apply_enter(self, response: dict, *, strict: bool = False) -> None:
self._require_accepted(response)
if strict and (self.entered or self.exited):
raise StateError("重复 /enter 或已退出后再 enter")
self.entered = True
self.virtual_time_s = response["virtual_time_s"]
def apply_exit(self, response: dict) -> None:
self._require_accepted(response)
self.exited = True
self.virtual_time_s = response["virtual_time_s"]
# ----------------------------------------------------------------- measure
def apply_measure(self, request: dict, response: dict, *,
cost: CostConfig = CostConfig(),
strict: bool = False) -> CostBreakdown:
self._require_accepted(response)
if strict and (not self.entered or self.exited):
raise StateError("/measure 要求已 enter 且未退出")
pos = (request["position"]["x"], request["position"]["y"])
channel = request["channel"]
predicted = measure_cost(self.position, pos, self.channel, channel, cost)
predicted = self._advance(pos, channel, response, predicted, strict)
return predicted
# ------------------------------------------------------------------- clear
def apply_clear(self, request: dict, response: dict, *,
cost: CostConfig = CostConfig(),
strict: bool = False) -> CostBreakdown:
self._require_accepted(response)
if strict and (not self.entered or self.exited):
raise StateError("/clear 要求已 enter 且未退出")
pos = (request["position"]["x"], request["position"]["y"])
success = response["clear_result"] == CLEAR_SUCCESS
predicted = clear_cost(self.position, pos, success, cost)
# 频道保持不变：/clear 不切换测向机频道 [题面]
return self._advance(pos, self.channel, response, predicted, strict)
# ------------------------------------------------------------------ 共用
def _advance(self, pos: Vec2, channel: int, response: dict,
predicted: CostBreakdown, strict: bool) -> CostBreakdown:
# 先完成全部校验再落子：strict 模式下失败时不留半更新状态。
actual_delta = response["virtual_time_s"] - self.virtual_time_s
if abs(actual_delta - predicted.total_s) > 1e-6:
message = (f"耗时预测不一致: 预测 {predicted.total_s:.6f}s, "
f"实测 {actual_delta:.6f}s ({predicted})")
if strict:
raise StateError(message)
# 非严格模式：状态仍以响应为准，偏差记入日志供复盘。
self.mismatches.append(
{"action": self.action_count + 1, "predicted_s": predicted.total_s,
"actual_s": actual_delta, "message": message})
self.position = pos
self.channel = channel
self.virtual_time_s = response["virtual_time_s"]
self.action_count += 1
return predicted
@staticmethod
def _require_accepted(response: dict) -> None:
if response.get("accepted") is not True:
raise StateError("accepted=false 的响应不允许更新状态")
In [ ]:
class TransportError(Exception):
"""网络层失败：连接不上、读写超时。动作可能未送达，也可能结果未知。"""
@dataclass(frozen=True)
class TransportResponse:
status: int
body: bytes
class Transport:
"""最小 POST 接口。同步阻塞，一次一个请求（协议要求串行）。"""
def post(self, url: str, body: bytes, headers: dict[str, str],
timeout: tuple[float, float]) -> TransportResponse:
raise NotImplementedError
class RequestsTransport(Transport):
"""基于 requests 的默认实现。timeout=(连接, 读取)。"""
def post(self, url, body, headers, timeout):
import requests # 惰性导入：仅真正发起 HTTP 时才需要该依赖
try:
response = requests.post(url, data=body, headers=headers, timeout=timeout)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as exc:
raise TransportError(str(exc)) from exc
except requests.exceptions.RequestException as exc: # 其它请求级异常同样按网络失败处理
raise TransportError(str(exc)) from exc
return TransportResponse(response.status_code, response.content)
In [ ]:
@dataclass
class SessionLog:
path: Path | None = None
events: list = field(default_factory=list)
def __post_init__(self):
if self.path is not None:
self.path.parent.mkdir(parents=True, exist_ok=True)
def append(self, event: dict) -> None:
event = {**event, "logged_at_unix_s": time()}
self.events.append(event)
if self.path is not None:
with self.path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(event, ensure_ascii=False, allow_nan=False) + "\n")
def action_events(self) -> list[dict]:
"""仅最终被接受的动作事件（每个 request_id 一条，含重试次数）。"""
return [event for event in self.events if event.get("response") is not None
and event["response"].get("accepted") is True]
def step_events(self) -> list[dict]:
"""每步计时事件（客户端 _log_step 写入，按发生顺序）。"""
return [event for event in self.events if event.get("event") == "step"]
In [ ]:
class ProtocolError(RuntimeError):
"""协议层失败，含 HTTP 状态与响应体；默认不重试（确定性错误）。"""
def __init__(self, message: str, status: int | None = None, body=None):
super().__init__(message)
self.status = status
self.body = body
class ActionRejected(ProtocolError):
"""HTTP 200 但 accepted=false：动作未生效，状态不得更新。"""
def __init__(self, message: str, body=None):
super().__init__(message, status=STATUS_OK, body=body)
class ProtocolViolation(ProtocolError):
"""响应与协议约定不符（字段缺失、值非法、虚拟时刻回退等）。"""
class DeadlineExceeded(RuntimeError):
"""现实预算不足：预留退出余量后剩余时间不够执行新动作。"""
@dataclass(frozen=True)
class MeasureOutcome:
result: str
svd_deg: float | None
virtual_time_s: float
predicted: CostBreakdown
raw: dict
@dataclass(frozen=True)
class ClearOutcome:
result: str
virtual_time_s: float
predicted: CostBreakdown
raw: dict
@dataclass(frozen=True)
class ExitInfo:
reason: str
virtual_time_s: float
raw: dict
def _validate_identifier(name: str, value: str, max_bytes: int) -> None:
if not isinstance(value, str) or not value:
raise ValueError(f"{name} 必须是非空字符串")
encoded = value.encode("utf-8")
if not 1 <= len(encoded) <= max_bytes:
raise ValueError(f"{name} 的 UTF-8 长度须在 1..{max_bytes} 字节")
if any(code < 0x20 or code == 0x7F for code in encoded):
raise ValueError(f"{name} 不能包含控制字符")
def _validate_position(position: dict, limit: float) -> None:
if set(position) != set(POSITION_FIELDS):
raise ValueError("position 字段必须是且只能是 x、y")
for key in ("x", "y"):
value = position[key]
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"position.{key} 必须是数值")
if not math.isfinite(value):
raise ValueError(f"position.{key} 必须是有限数值")
if abs(value) > limit:
raise ValueError(f"position.{key} 绝对值不得超过 {limit} 米")
def _validate_channel(channel) -> int:
# 写成 JSON number 1.0 且恰为整数也接受 [题面]；1.5 拒绝。
if isinstance(channel, bool) or not isinstance(channel, (int, float)):
raise ValueError("channel 必须是数值")
if isinstance(channel, float) and not channel.is_integer():
raise ValueError("channel 必须是整数")
channel = int(channel)
if not 1 <= channel <= 20:
raise ValueError("channel 必须在 1..20 范围内")
return channel
class SimulatorClient:
"""四条指令的同步客户端。状态只在 accepted=true 后推进。"""
def __init__(self, robot_id: str, *,
config: ProblemConfig = ProblemConfig(),
transport: Transport | None = None,
log: SessionLog | None = None,
clock=monotonic):
_validate_identifier("robot_id", robot_id, config.session.robot_id_max_bytes)
self.robot_id = robot_id
self.config = config
self.session = config.session
self.transport = transport if transport is not None else RequestsTransport()
self.log = log if log is not None else SessionLog()
self.clock = clock
self.state = RobotState()
self._prefix = uuid.uuid4().hex[:12]
self._seq = 0
self._deadline: float | None = None # 本地单调时钟截止
self._server_deadline_ms: float | None = None # 服务器时间戳交叉核对
self._enter_monotonic: float | None = None # 实机计时起点（/enter 成功）
# ------------------------------------------------------------ 公共动作 API
def enter(self) -> dict:
started = self.clock()
payload = self._payload("/enter")
response = self._call("/enter", payload)
self.state.apply_enter(response)
remaining = response["remaining_real_duration_s"]
if not isinstance(remaining, (int, float)) or remaining < 0:
raise ProtocolViolation("remaining_real_duration_s 缺失或非法", body=response)
self._deadline = self.clock() + remaining
self._server_deadline_ms = (response["real_timestamp_ms"]
+ remaining * 1000.0)
self._enter_monotonic = self.clock()
self._log_step("/enter", payload, response, started,
predicted=CostBreakdown(0.0, 0.0, 0.0))
return response
def measure(self, position: Vec2, channel: int) -> MeasureOutcome:
started = self.clock()
payload = self._payload("/measure", position=position, channel=channel)
response = self._call("/measure", payload)
predicted = self.state.apply_measure(payload, response,
cost=self.config.cost)
result = response.get("measure_result")
if result not in MEASURE_RESULTS:
raise ProtocolViolation(f"measure_result 非法: {result!r}", body=response)
svd = response.get("svd_deg")
if result == DIRECTION:
if not isinstance(svd, (int, float)) or isinstance(svd, bool):
raise ProtocolViolation("direction 响应缺少数值 svd_deg", body=response)
self._log_step("/measure", payload, response, started, result=result,
predicted=predicted, channel=payload["channel"])
return MeasureOutcome(result, float(svd) if svd is not None else None,
response["virtual_time_s"], predicted, response)
def clear(self, position: Vec2, channel: int) -> ClearOutcome:
started = self.clock()
payload = self._payload("/clear", position=position, channel=channel)
response = self._call("/clear", payload)
predicted = self.state.apply_clear(payload, response,
cost=self.config.cost)
result = response.get("clear_result")
if result not in CLEAR_RESULTS:
raise ProtocolViolation(f"clear_result 非法: {result!r}", body=response)
self._log_step("/clear", payload, response, started, result=result,
predicted=predicted, channel=payload["channel"])
return ClearOutcome(result, response["virtual_time_s"], predicted, response)
def exit(self) -> ExitInfo:
started = self.clock()
payload = self._payload("/exit")
response = self._call("/exit", payload)
self.state.apply_exit(response)
reason = response.get("exit_reason")
if reason != EXIT_USER:
raise ProtocolViolation(f"exit_reason 非法: {reason!r}", body=response)
self._log_step("/exit", payload, response, started,
predicted=CostBreakdown(0.0, 0.0, 0.0))
return ExitInfo(reason, response["virtual_time_s"], response)
# ---------------------------------------------------------------- 预算控制
@property
def real_remaining_s(self) -> float | None:
"""按本地单调时钟估计的剩余现实秒数（未 enter 时为 None）。"""
return None if self._deadline is None else self._deadline - self.clock()
def _log_step(self, path: str, payload: dict, response: dict,
started: float, *, result: str | None = None,
predicted: CostBreakdown, channel: int | None = None) -> None:
"""每步计时留档 [用户决策 2026-09-11]：实机墙钟耗时 + 比赛虚拟耗时分项
（移动=直线距离/5、切频道 1s、检测 5s、清除成功 5s/未发现 3s）。
step 事件不含 response 字段，不会混入 action_events / 虚拟时刻复算。
"""
now = self.clock()
entry: dict = {"event": "step", "path": path,
"request_id": payload["request_id"],
"real_elapsed_s": round(now - started, 6),
"virtual_time_s": response["virtual_time_s"],
"virtual_cost": {"move_s": round(predicted.move_s, 6),
"switch_s": round(predicted.switch_s, 6),
"action_s": round(predicted.action_s, 6),
"total_s": round(predicted.total_s, 6)}}
if self._enter_monotonic is not None:
entry["real_since_enter_s"] = round(now - self._enter_monotonic, 6)
if "position" in payload:
entry["position"] = {"x": payload["position"]["x"],
"y": payload["position"]["y"]}
if channel is not None:
entry["channel"] = channel
if result is not None:
entry["result"] = result
self.log.append(entry)
def _check_budget(self, path: str) -> None:
if path == "/exit" or self._deadline is None:
return
margin = self.session.reserve_exit_s
if self.real_remaining_s <= margin:
raise DeadlineExceeded(
f"剩余现实时间 {self.real_remaining_s:.1f}s 不足以预留 {margin:.1f}s 退出余量")
# ---------------------------------------------------------------- 内部机制
def _payload(self, _path: str, *, position: Vec2 | None = None,
channel: int | None = None) -> dict:
payload = {"arena_id": self.session.arena_id,
"robot_id": self.robot_id,
"request_id": self._next_request_id()}
if position is not None:
x, y = float(position[0]), float(position[1])
_validate_position({"x": x, "y": y}, self.config.field.coord_abs_limit_m)
payload["position"] = {"x": x, "y": y}
if channel is not None:
payload["channel"] = _validate_channel(channel)
return payload
def _next_request_id(self) -> str:
self._seq += 1
request_id = f"{self._prefix}-{self._seq:04d}"
_validate_identifier("request_id", request_id,
self.session.request_id_max_bytes)
return request_id
def _call(self, path: str, payload: dict) -> dict:
"""串行执行一个动作；网络失败原样重发，HTTP 确定性错误立即抛出。"""
self._check_budget(path)
body = json.dumps(payload, ensure_ascii=False, allow_nan=False).encode("utf-8")
if len(body) > self.session.request_body_max_bytes:
raise ValueError("请求体超过 65536 字节")
headers = {"Content-Type": "application/json"}
url = self.session.base_url + path
attempts, last_error = 0, None
while attempts < self.session.max_attempts:
attempts += 1
started = self.clock()
self.log.append({"event": "attempt", "path": path,
"request_id": payload["request_id"],
"attempt": attempts, "request": payload})
try:
result = self.transport.post(
url, body, headers,
timeout=(self.session.connect_timeout_s,
self.session.read_timeout_s))
except TransportError as exc:
last_error = exc
self.log.append({"event": "transport_error", "path": path,
"request_id": payload["request_id"],
"attempt": attempts, "error": str(exc),
"elapsed_s": self.clock() - started})
if attempts < self.session.max_attempts:
sleep(self.session.retry_backoff_s * (2 ** (attempts - 1)))
continue
if result.status in STATUS_RETRYABLE:
self.log.append({"event": "retryable_status", "path": path,
"request_id": payload["request_id"],
"attempt": attempts, "status": result.status})
if attempts < self.session.max_attempts:
sleep(self.session.retry_backoff_s * (2 ** (attempts - 1)))
continue
if result.status != STATUS_OK:
raise ProtocolError(
f"{path} 返回 HTTP {result.status}", status=result.status,
body=self._safe_json(result.body))
response = self._safe_json(result.body)
if response is None or response.get("accepted") is not True:
raise ActionRejected(
f"{path} 未被接受: {response}", body=response)
self._verify_clock(response)
self.log.append({"event": "accepted", "path": path,
"request_id": payload["request_id"],
"request": payload, "response": response,
"attempts": attempts,
"elapsed_s": self.clock() - started})
return response
raise TransportError(
f"{path} 重试 {attempts} 次仍失败: {last_error}") from last_error
def _verify_clock(self, response: dict) -> None:
"""accepted=true 的虚拟时刻必须单调不减；现实时间戳用于漂移核对。"""
virtual = response.get("virtual_time_s")
if not isinstance(virtual, (int, float)) or isinstance(virtual, bool):
raise ProtocolViolation("accepted 响应缺少 virtual_time_s", body=response)
if virtual < self.state.virtual_time_s - 1e-9:
raise ProtocolViolation(
f"虚拟时刻回退: {self.state.virtual_time_s} -> {virtual}",
body=response)
if self._server_deadline_ms is not None:
drift = (self._server_deadline_ms - response["real_timestamp_ms"]) / 1000.0
if drift < self.session.reserve_exit_s / 2:
# 服务器视角余量告急（本地时钟漂移或请求排队）；仅记录，交由上层决策。
self.log.append({"event": "server_time_warning",
"server_remaining_s": drift})
@staticmethod
def _safe_json(body: bytes):
try:
return json.loads(body.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError):
return None
In [ ]:
Point = tuple[float, float]
EPS = 1e-9
PARALLEL_EPS = 1e-12
@dataclass(frozen=True)
class Measurement:
x: float
y: float
bearing_deg: float
half_angle_deg: float = 1.0
def __post_init__(self):
values = (self.x, self.y, self.bearing_deg, self.half_angle_deg)
if any(isinstance(v, bool) or not isinstance(v, (int, float))
or not isfinite(v) for v in values):
raise ValueError("坐标、示向度和误差半角必须是有限数值")
if not 0 <= self.half_angle_deg < 90:
raise ValueError("误差半角必须在 [0, 90) 度；题设为1度")
def cross(a: Point, b: Point) -> float:
return a[0] * b[1] - a[1] * b[0]
def distance(a: Point, b: Point) -> float:
return hypot(a[0] - b[0], a[1] - b[1])
def halfplanes(m: Measurement) -> list[tuple[float, float, float]]:
"""Return a*x+b*y<=c with unit normals, including the forward constraint."""
theta = m.bearing_deg % 360
lower, upper = radians(theta - m.half_angle_deg), radians(theta + m.half_angle_deg)
middle = radians(theta)
normals = [(sin(lower), -cos(lower)), (-sin(upper), cos(upper)),
(-cos(middle), -sin(middle))]
return [(a, b, a * m.x + b * m.y) for a, b in normals]
def convex_hull(points: list[Point], eps: float = EPS) -> list[Point]:
"""Monotone chain, CCW, no repeated final vertex; retain segment endpoints."""
ordered = sorted(points)
unique = []
for p in ordered:
if not unique or distance(p, unique[-1]) > eps:
unique.append(p)
if len(unique) <= 2:
return unique
def chain(seq):
out = []
for p in seq:
while len(out) >= 2:
a = (out[-1][0] - out[-2][0], out[-1][1] - out[-2][1])
b = (p[0] - out[-1][0], p[1] - out[-1][1])
# Distance-to-line tolerance, not an unscaled area tolerance.
if cross(a, b) > eps * max(hypot(*a), hypot(*b), eps):
break
out.pop()
out.append(p)
return out
return chain(unique)[:-1] + chain(reversed(unique))[:-1]
def diameter(points: list[Point]):
if not points:
raise ValueError("空点集没有本程序定义下的直径")
a = b = points[0]
value = 0.0
for p, q in combinations(points, 2):
d = distance(p, q)
if d > value:
value, a, b = d, p, q
return value, (a, b)
def _circumcircle(a: Point, b: Point, c: Point):
u = (b[0] - a[0], b[1] - a[1])
v = (c[0] - a[0], c[1] - a[1])
det = 2 * cross(u, v)
if abs(det) <= PARALLEL_EPS * max(hypot(*u) * hypot(*v), 1e-300):
return None
uu, vv = u[0] ** 2 + u[1] ** 2, v[0] ** 2 + v[1] ** 2
center = (a[0] + (uu * v[1] - vv * u[1]) / det,
a[1] + (u[0] * vv - v[0] * uu) / det)
return center, distance(center, a)
def _enclosing_circle_reference(points: list[Point], eps: float = EPS):
"""Enumerate all 1/2/3-point supports; O(m^4), appropriate for small m.
Radius is recomputed against every vertex to preserve numerical containment.
Coordinates should be normalized when there are very large offsets.
"""
if not points:
raise ValueError("空集不计算最小包围圆")
xs, ys = [p[0] for p in points], [p[1] for p in points]
best_c = ((min(xs) + max(xs)) / 2, (min(ys) + max(ys)) / 2)
best_r = max(distance(best_c, p) for p in points)
def consider(center, radius):
nonlocal best_c, best_r
if radius > best_r + eps:
return
actual = max(distance(center, p) for p in points)
if actual <= radius + eps and actual < best_r:
best_c, best_r = center, actual
for p in points:
consider(p, 0.0)
for a, b in combinations(points, 2):
consider(((a[0] + b[0]) / 2, (a[1] + b[1]) / 2), distance(a, b) / 2)
for a, b, c in combinations(points, 3):
circle = _circumcircle(a, b, c)
if circle is not None:
consider(*circle)
return best_c, best_r
def _circle_covers(circle, point: Point, tolerance: float) -> bool:
"""Whether the circle already covers point; None means no circle yet."""
if circle is None:
return False
return distance(circle[0], point) <= circle[1] + tolerance
def smallest_enclosing_circle(points: list[Point], eps: float = EPS):
"""Welzl's randomized incremental algorithm, using an explicit call stack.
The fixed shuffle makes results reproducible. Expected linear work refers
to randomized input order, not a worst-case guarantee for this fixed seed.
Normalize internally so this public helper also handles translated inputs.
The exhaustive reference is retained for independent regression comparisons.
"""
if not points:
raise ValueError("空集不计算最小包围圆")
if not isfinite(eps) or eps <= 0:
raise ValueError("eps必须为有限正数")
if any(len(p) != 2 or any(isinstance(x, bool) or not isinstance(x, (int, float))
or not isfinite(x) for x in p) for p in points):
raise ValueError("点集必须由有限二维坐标组成")
origin = points[0]
scale = max(distance(origin, p) for p in points)
if scale == 0:
return tuple(origin), 0.0
local = [((p[0]-origin[0])/scale, (p[1]-origin[1])/scale) for p in points]
tolerance = min(eps/scale, EPS)
Random(20260910).shuffle(local)
# Frame = (remaining prefix length, forced boundary, returning from child).
stack = [(len(local), (), False)]
circle = None
while stack:
n, boundary, returning = stack.pop()
if n == 0 or len(boundary) == 3:
# Forced support points must lie ON the boundary. For three points
# use their circumcircle, even when their unconstrained MEC is smaller.
circle = _circumcircle(*boundary) if len(boundary) == 3 else None
if circle is None and boundary:
circle = _enclosing_circle_reference(list(boundary), tolerance)
elif not returning:
stack.append((n, boundary, True))
stack.append((n-1, boundary, False))
elif not _circle_covers(circle, local[n-1], tolerance):
stack.append((n-1, boundary + (local[n-1],), False))
assert circle is not None
center = (origin[0]+circle[0][0]*scale, origin[1]+circle[0][1]*scale)
# Recompute after transforming back: returned circle contains all input points
# to distance() precision, including any roundoff from the large translation.
return center, max(distance(center, p) for p in points)
def solve(measurements: list[Measurement], eps: float = EPS) -> dict:
"""Solve Q1's pure angular intersection (no disk/range prior added).
Nonempty intersections of <180-degree forward wedges are pointed polyhedra,
so they possess a vertex, even if unbounded. Feasibility follows from vertex
enumeration; unboundedness follows from a nonzero recession direction.
Empty diameter is null, unbounded diameter is null with diameter_infinite=true.
Near-parallel plane pairs are intersected rather than skipped: their
intersection can be a legitimate vertex of a very long thin region
(e.g. stations 1cm apart with bearings 1 and 358.99999999999 deg give a
vertex ~5.7e10 m away). Only exactly-parallel pairs (det == 0) have no
intersection. The feasibility tolerance scales with the vertex magnitude,
since absolute residuals grow with coordinates.
"""
if not isfinite(eps) or eps <= 0:
raise ValueError("eps必须为有限正数")
if not measurements:
return {"status": "unbounded", "vertices": [], "diameter_m": None,
"diameter_infinite": True, "recession_direction": [1.0, 0.0],
"message": "没有观测，整个平面均可行"}
if len(measurements) > 100:
raise ValueError("参考实现最多接收100个检测点；更大规模应使用半平面交算法")
ox = sum(m.x for m in measurements) / len(measurements)
oy = sum(m.y for m in measurements) / len(measurements)
scale = max(1.0, max(hypot(m.x - ox, m.y - oy) for m in measurements))
local = [Measurement((m.x - ox) / scale, (m.y - oy) / scale,
m.bearing_deg, m.half_angle_deg) for m in measurements]
planes = [h for m in local for h in halfplanes(m)]
def inside(p):
tol = eps * max(1.0, abs(p[0]), abs(p[1]))
return all(a * p[0] + b * p[1] <= c + tol for a, b, c in planes)
feasible_vertices = []
for (a, b, c), (d, e, f) in combinations(planes, 2):
det = a * e - b * d
if det == 0.0:
continue # 精确平行：无交点（近平行对的交点是合法远端顶点，保留）
p = ((c * e - b * f) / det, (a * f - c * d) / det)
if isfinite(p[0]) and isfinite(p[1]) and inside(p):
feasible_vertices.append(p)
if not feasible_vertices:
return {"status": "empty", "vertices": [], "diameter_m": None,
"diameter_infinite": False, "tolerance_m": eps * scale,
"message": "测向角域没有公共点；检查频道、数据及误差界"}
# A nontrivial 2D recession cone has a ray tangent to an input halfplane.
# Use a much tighter angular threshold than the positional feasibility epsilon.
ray = None
for a, b, _ in planes:
for direction in ((b, -a), (-b, a)):
if all(d * direction[0] + e * direction[1] <= 2e-14
for d, e, _ in planes):
ray = direction
break
if ray is not None:
break
hull = convex_hull(feasible_vertices, eps)
def world(p):
return [ox + p[0] * scale, oy + p[1] * scale]
result = {"status": "unbounded" if ray else
("point" if len(hull) == 1 else "segment" if len(hull) == 2 else "polygon"),
"vertices": [world(p) for p in hull], "tolerance_m": eps * scale,
"diameter_infinite": ray is not None}
if ray is not None:
result.update(diameter_m=None, recession_direction=list(ray),
message="存在非零衰退方向，区域无界，直径为无穷大")
return result
d, pair = diameter(hull)
midpoint = ((pair[0][0] + pair[1][0]) / 2, (pair[0][1] + pair[1][1]) / 2)
distances = [distance(midpoint, p) for p in hull]
witness_index = max(range(len(hull)), key=distances.__getitem__)
max_distance = distances[witness_index]
center, radius = smallest_enclosing_circle(hull, eps)
area = abs(sum(cross(hull[i], hull[(i + 1) % len(hull)])
for i in range(len(hull)))) / 2
result.update(diameter_m=d * scale, diameter_pair=[world(p) for p in pair],
area_m2=area * scale ** 2,
diameter_circle={"center": world(midpoint), "radius_m": d * scale / 2,
"max_vertex_distance_m": max_distance * scale,
"covers": max_distance <= d / 2 + eps * 4,
"coverage_tolerance_m": 4 * eps * scale,
"max_excess_m": (max_distance-d/2) * scale,
"outside_vertex_indices": [i for i, value in enumerate(distances)
if value > d/2 + 4*eps],
"witness_vertex": world(hull[witness_index])
if max_distance > d/2 + 4*eps else None},
minimum_enclosing_circle={"center": world(center), "radius_m": radius * scale},
guaranteed_clearable_at_mec_center=radius * scale <= 20 - eps * scale * 4,
clearance_certificate={"model": "angular_intersection_only",
"clear_radius_m": 20.0,
"numerical_margin_m": 4*eps*scale,
"radius_with_margin_m": (radius+4*eps)*scale,
"remaining_margin_m": 20-(radius+4*eps)*scale,
"center": world(center)})
return result
In [ ]:
def first_bearing_candidate_disk(measurement: Measurement,
max_range_m: float = 1500,
guaranteed_range_m: float = 1000) -> dict:
"""Enclose the entire truncated first-bearing sector, then erode its disk.
For half-angle <=45 degrees the sector MEC has center distance and radius
a=R/(2*cos(delta)). Intersecting with Omega only shrinks the source set.
A missing sufficient disk is NOT proof that Q is empty. The certificate
guarantees distance only: directional transmitters may still be invisible.
"""
if any(isinstance(v, bool) or not isinstance(v, (int, float)) or not isfinite(v)
or v <= 0 for v in (max_range_m, guaranteed_range_m)):
raise ValueError("接收半径必须为有限正数")
if measurement.half_angle_deg > 45:
raise ValueError("此窄角域证书仅支持半角不超过45度；题设为1度")
delta, theta = radians(measurement.half_angle_deg), radians(measurement.bearing_deg % 360)
a = max_range_m/(2*cos(delta))
center = [measurement.x+a*cos(theta), measurement.y+a*sin(theta)]
available = guaranteed_range_m-a
return {"source_model": "first_bearing_sector_with_range_upper_bound",
"sector_enclosing_circle": {"center": center, "radius_m": a},
"sufficient_candidate_disk": {"center": center, "radius_m": available}
if available >= 0 else None,
"candidate_nonempty_certified": available >= 0,
"distance_only": True,
"max_range_m": max_range_m, "guaranteed_range_m": guaranteed_range_m}
In [ ]:
# 半平面裁剪的正容差（米）。坐标量级 ~2100，浮点误差远小于它；
# 容差只向"保留更多候选"的方向倾斜，不破坏保守性。
_CLIP_TOL_M = 1e-6
_DEDUPE_TOL_M = 1e-7
class InconsistentBelief(RuntimeError):
"""读数把候选集剪空等违背保守性的情形；须停机核查而非继续跑。"""
class ChannelStatus(Enum):
UNKNOWN = "unknown" # 尚无有效证据
DETECTED = "detected" # 收到过 direction/near，待定位
LOCATED = "located" # 候选集包围圆半径 <= 20，可清除
CLEARED = "cleared" # 清除成功
ABSENT = "absent" # 覆盖证明不存在（P4）
def circumscribed_polygon(center: Vec2, radius_m: float, sides: int) -> list[Vec2]:
"""圆的外切正多边形顶点（外接半径 r/cos(π/n)），严格包含圆盘。
近似误差（外接圆超出圆盘的部分）为 r(1/cos(π/n) - 1) ≤ r(1-cos(π/n))·sec²，
调用方用 (1-cos(π/n)) 口径记账即可（略保守偏大）。
"""
if sides < 8:
raise ValueError("外切多边形边数过少，近似误差不可控")
circumradius = radius_m / math.cos(math.pi / sides)
step = 2 * math.pi / sides
return [(center[0] + circumradius * math.cos(step * i + step / 2),
center[1] + circumradius * math.sin(step * i + step / 2))
for i in range(sides)]
def polygon_halfplanes(polygon: list[Vec2]) -> list[tuple[float, float, float]]:
"""CCW 多边形 -> 外法线半平面列表 (a, b, c)，含义 a·x + b·y <= c。"""
planes = []
count = len(polygon)
for index in range(count):
p, q = polygon[index], polygon[(index + 1) % count]
dx, dy = q[0] - p[0], q[1] - p[1]
norm = math.hypot(dx, dy)
if norm <= 1e-12:
continue
a, b = dy / norm, -dx / norm # CCW 的右法线朝外
planes.append((a, b, a * p[0] + b * p[1]))
return planes
def clip_halfplane(polygon: list[Vec2], a: float, b: float, c: float,
tol: float = _CLIP_TOL_M) -> list[Vec2]:
"""Sutherland–Hodgman 凸多边形半平面裁剪，保留 a·x + b·y <= c + tol。
边界交点取在无容差直线上；进入判定带正容差，保证"贴边"的真点不丢。
"""
if not polygon:
return []
output: list[Vec2] = []
count = len(polygon)
for index in range(count):
p, q = polygon[index], polygon[(index + 1) % count]
fp = a * p[0] + b * p[1] - c
fq = a * q[0] + b * q[1] - c
p_in, q_in = fp <= tol, fq <= tol
if p_in:
output.append(p)
if p_in != q_in:
ratio = fp / (fp - fq)
output.append((p[0] + ratio * (q[0] - p[0]),
p[1] + ratio * (q[1] - p[1])))
return _dedupe(output)
def _dedupe(points: list[Vec2], tol: float = _DEDUPE_TOL_M) -> list[Vec2]:
unique: list[Vec2] = []
for point in points:
if not unique or math.dist(point, unique[-1]) > tol:
unique.append(point)
if len(unique) > 1 and math.dist(unique[0], unique[-1]) <= tol:
unique.pop()
return unique
class CandidateRegion:
"""某频道候选集合的保守凸外包多边形（顶点按 CCW；允许退化为 1-2 点）。"""
def __init__(self, polygon: list[Vec2], slack_m: float = 0.0):
self.polygon = list(polygon)
self.slack_m = float(slack_m) # 外包近似引入的半径向误差记账
self._mec: tuple[Vec2, float] | None = None
# ------------------------------------------------------------ 构造
@classmethod
def disk(cls, center: Vec2, radius_m: float, sides: int,
slack_m: float | None = None) -> "CandidateRegion":
if slack_m is None:
slack_m = radius_m * (1 - math.cos(math.pi / sides))
return cls(circumscribed_polygon(center, radius_m, sides), slack_m)
@classmethod
def initial(cls, station: Vec2, config: ProblemConfig) -> "CandidateRegion":
"""首个 direction 后的基础集：Ω ∩ B(station, r_max) 的保守外包。"""
field_cfg, sides = config.field, config.search.polygon_sides
slack = field_cfg.omega_radius_m * (1 - math.cos(math.pi / sides))
region = cls.disk((0.0, 0.0), field_cfg.omega_radius_m, sides, slack)
region.clip_disk(station, field_cfg.r_eff_max_m, sides)
return region
# ------------------------------------------------------------ 收缩
def _clip(self, a: float, b: float, c: float) -> None:
polygon = clip_halfplane(self.polygon, a, b, c)
if not polygon:
raise InconsistentBelief(
f"候选区域被半平面 ({a:.6f},{b:.6f},{c:.6f}) 剪空：读数与保守集矛盾")
self.polygon = polygon
self._mec = None
def clip_disk(self, center: Vec2, radius_m: float, sides: int) -> None:
"""与圆盘 B(center, radius) 的外切多边形求交（近似误差入账）。"""
for a, b, c in polygon_halfplanes(circumscribed_polygon(center, radius_m, sides)):
self._clip(a, b, c)
self.slack_m += radius_m * (1 - math.cos(math.pi / sides))
def clip_wedge(self, station: Vec2, bearing_deg: float,
half_angle_deg: float) -> None:
"""与测向角域 W(station, bearing±half_angle) 求交（半角须 <90°）。
两个边界半平面与 localization.halfplanes 同口径，0°/360° 跨界由
sin/cos 直接处理，无需分支。
"""
if not 0 < half_angle_deg < 90:
raise ValueError("角域半角须在 (0, 90) 度")
lower = math.radians(bearing_deg - half_angle_deg)
upper = math.radians(bearing_deg + half_angle_deg)
sx, sy = float(station[0]), float(station[1])
self._clip(math.sin(lower), -math.cos(lower),
math.sin(lower) * sx - math.cos(lower) * sy)
self._clip(-math.sin(upper), math.cos(upper),
-math.sin(upper) * sx + math.cos(upper) * sy)
# ------------------------------------------------------------ 查询
def mec(self) -> tuple[Vec2, float]:
"""最小包围圆（圆心, 半径）。半径是真实候选集 ρ 的上界。"""
if self._mec is None:
if not self.polygon:
raise InconsistentBelief("候选区域为空")
self._mec = smallest_enclosing_circle(self.polygon)
return self._mec
def witnesses(self, spacing_m: float) -> list[Vec2]:
"""顶点 + 边界按间距细分的采样点（最坏后验评估用）。"""
if len(self.polygon) < 3:
return list(self.polygon)
points = list(self.polygon)
count = len(self.polygon)
for index in range(count):
p, q = self.polygon[index], self.polygon[(index + 1) % count]
length = math.dist(p, q)
steps = int(length // spacing_m)
for k in range(1, steps + 1):
ratio = k / (steps + 1)
points.append((p[0] + ratio * (q[0] - p[0]),
p[1] + ratio * (q[1] - p[1])))
return _dedupe(points, tol=1e-9)
def contains(self, point: Vec2, tol_m: float = 0.05) -> bool:
"""点是否在候选集内（带容差；用于真源保留性检验，不做删点依据）。"""
if not self.polygon:
return False
if len(self.polygon) == 1:
return math.dist(point, self.polygon[0]) <= tol_m + self.slack_m
if len(self.polygon) == 2:
p, q = self.polygon
return _point_segment_distance(point, p, q) <= tol_m + self.slack_m
for index, p in enumerate(self.polygon):
q = self.polygon[(index + 1) % len(self.polygon)]
if (q[0] - p[0]) * (point[1] - p[1]) - (q[1] - p[1]) * (point[0] - p[0]) \
< -tol_m - self.slack_m:
return False
return True
def _point_segment_distance(point: Vec2, p: Vec2, q: Vec2) -> float:
vx, vy = q[0] - p[0], q[1] - p[1]
length_sq = vx * vx + vy * vy
if length_sq <= 1e-18:
return math.dist(point, p)
t = max(0.0, min(1.0, ((point[0] - p[0]) * vx + (point[1] - p[1]) * vy) / length_sq))
return math.dist(point, (p[0] + t * vx, p[1] + t * vy))
@dataclass
class ChannelBelief:
channel: int
status: ChannelStatus = ChannelStatus.UNKNOWN
region: CandidateRegion | None = None
measurements: list = field(default_factory=list) # 逐步观测记录（审计用）
class BeliefState:
"""全部 20 个频道的信念集合。策略端唯一的事实来源。"""
def __init__(self, config: ProblemConfig = ProblemConfig()):
self.config = config
self.channels = {channel: ChannelBelief(channel)
for channel in range(config.field.channel_min,
config.field.channel_max + 1)}
@property
def effective_half_angle_deg(self) -> float:
"""示向度误差半角 + 舍入裕量：服务器保留两位小数引入 ≤0.005° 偏差。"""
return (self.config.field.bearing_error_half_deg
+ self.config.search.bearing_rounding_margin_deg)
def update_measure(self, channel: int, position: Vec2, result: str,
svd_deg: float | None) -> None:
"""按读数收敛候选集合；真源保留性由保守表示保证。"""
belief = self.channels[channel]
belief.measurements.append({"position": tuple(position), "result": result,
"svd_deg": svd_deg})
if result == DIRECTION:
if svd_deg is None:
raise ValueError("direction 读数必须携带 svd_deg")
sides = self.config.search.polygon_sides
if belief.region is None:
belief.region = CandidateRegion.initial(position, self.config)
else:
# 后续 direction 同样隐含距离 <= r_max [题面]
belief.region.clip_disk(position, self.config.field.r_eff_max_m, sides)
belief.region.clip_wedge(position, svd_deg, self.effective_half_angle_deg)
belief.status = ChannelStatus.DETECTED
elif result == NEAR:
sides = self.config.search.polygon_sides
if belief.region is None:
belief.region = CandidateRegion.disk(
tuple(position), self.config.field.near_radius_m, sides)
else:
belief.region.clip_disk(position, self.config.field.near_radius_m, sides)
belief.status = ChannelStatus.DETECTED
elif result == NO_SIGNAL:
# P3/P4 基线：不据此删空间候选（缺席证明由覆盖账本给出）。
pass
else:
raise ValueError(f"未知测向结果: {result!r}")
def candidate_set(self, channel: int) -> CandidateRegion | None:
"""该频道当前候选集合（保守外包形式）；尚无 direction/near 时为 None。"""
return self.channels[channel].region
def mark_cleared(self, channel: int) -> None:
self.channels[channel].status = ChannelStatus.CLEARED
def mark_absent(self, channel: int) -> None:
self.channels[channel].status = ChannelStatus.ABSENT
def channels_by_status(self, *statuses: ChannelStatus) -> list[int]:
return sorted(channel for channel, belief in self.channels.items()
if belief.status in statuses)
In [ ]:
def coverage_points(config: ProblemConfig = ProblemConfig()) -> list[Vec2]:
"""49 点覆盖网格生成（纯几何，可独立测试）。"""
step = config.search.grid_step_m
half = config.search.grid_half_cells
return [(step * i, step * j)
for j in range(half, -half - 1, -1)
for i in range(-half, half + 1)]
@dataclass
class CoverageLedger:
"""逐频道覆盖账本：(频道, 网格点) -> 是否已测。"""
config: ProblemConfig = field(default_factory=ProblemConfig)
points: list[Vec2] = field(default_factory=coverage_points)
measured: set = field(default_factory=set) # {(channel, point_index)}
skipped: dict = field(default_factory=dict) # channel -> 原因
def mark_measured(self, channel: int, point_index: int) -> None:
self.measured.add((channel, point_index))
def measured_at(self, channel: int, point_index: int) -> bool:
return (channel, point_index) in self.measured
def skip_channel(self, channel: int, reason: str) -> None:
"""记录整频道跳过原因（如 already_cleared / located_pending_clear）。"""
self.skipped[channel] = reason
def pending_points(self, channel: int) -> list[int]:
"""该频道尚未测量的网格点下标。扫描调度策略未实现（P4）。"""
return [index for index in range(len(self.points))
if not self.measured_at(channel, index)]
def coverage_complete(self, channel: int) -> bool:
"""该频道覆盖是否完成（含显式跳过）。"""
return channel in self.skipped or not self.pending_points(channel)
In [ ]:
R_EFF_MIN_M = 1000.0 # [题面] 最小有效接收半径
OMEGA_RADIUS_M = 1800.0 # [题面] 目标圆域半径
SPEED_MPS = 5.0 # [题面] 用于把巡回长度折算成虚拟移动秒数
# 采样分辨率。**校核必须与证书同分辨率**：更粗的采样会放过"删掉一个关键点
# 仍然看似覆盖"的点集，使校核退化为装饰（实测 100 m/12 档即漏判）。
# 75 m / 24 档下一次全域校核约 1 s，且按计划名缓存，每进程只付一次。
CERTIFY_STEP_M = 75.0
CERTIFY_HEADINGS = 24
VERIFY_STEP_M = CERTIFY_STEP_M
VERIFY_HEADINGS = CERTIFY_HEADINGS
@dataclass(frozen=True)
class CoverageCertificate:
"""覆盖证书：采样语义 + 缺口 + 最坏命中距离。"""
kind: str # "full_disk" | "forward_half_disk"
target_step_m: float
headings: int
r_eff_min_m: float
targets: int
uncovered: int
worst_hit_distance_m: float
nn_tour_m: float
generator_sha256: str = ""
# 证书硬化：校核时使用的前向角裕量（度）。与 r_eff_min_m 的设计裕量
# 一起把"采样格覆盖"提升为"连续空间覆盖"，见 verify_forward_half_disk。
verify_heading_margin_deg: float = 0.0
@property
def valid(self) -> bool:
return self.uncovered == 0 and self.targets > 0
def as_dict(self) -> dict:
return {"kind": self.kind, "target_step_m": self.target_step_m,
"headings": self.headings, "r_eff_min_m": self.r_eff_min_m,
"targets": self.targets, "uncovered": self.uncovered,
"worst_hit_distance_m": round(self.worst_hit_distance_m, 3),
"nn_tour_m": round(self.nn_tour_m, 1),
"nn_tour_virtual_s": round(self.nn_tour_m / SPEED_MPS, 1),
"generator_sha256": self.generator_sha256,
"verify_heading_margin_deg": self.verify_heading_margin_deg}
@dataclass(frozen=True)
class CoveragePlan:
"""一份覆盖计划：点集 + 语义 + 证书（可为解析证明）。"""
name: str
problem: int
points: tuple[Vec2, ...]
kind: str # "analytic" | "certified"
coverage: str # "full_disk" | "forward_half_disk"
note: str = ""
certificate: CoverageCertificate | None = None
generator: dict = field(default_factory=dict)
def __len__(self) -> int:
return len(self.points)
def as_dict(self) -> dict:
return {"name": self.name, "problem": self.problem, "kind": self.kind,
"coverage": self.coverage, "note": self.note,
"generator": self.generator,
"certificate": (self.certificate.as_dict()
if self.certificate is not None else None),
"points": [[p[0], p[1]] for p in self.points]}
# --------------------------------------------------------------------- 采样
def omega_positions(step_m: float) -> list[Vec2]:
"""Ω 内均匀方格位置采样。"""
span = int(OMEGA_RADIUS_M / step_m) + 1
return [(step_m * i, step_m * j)
for i in range(-span, span + 1) for j in range(-span, span + 1)
if (step_m * i) ** 2 + (step_m * j) ** 2 <= OMEGA_RADIUS_M ** 2]
def forward_targets(step_m: float, headings: int) -> list[tuple[float, float, float]]:
"""(p, h) 目标采样：位置在 Ω 内，朝向均分整圆。"""
return [(x, y, 2.0 * math.pi * k / headings)
for (x, y) in omega_positions(step_m) for k in range(headings)]
def candidate_lattice(spacing_m: float, radius_m: float) -> list[Vec2]:
"""半径 radius_m 圆盘内的方格候选点（离线搜索用）。"""
span = int(radius_m / spacing_m) + 1
return [(spacing_m * i, spacing_m * j)
for i in range(-span, span + 1) for j in range(-span, span + 1)
if (spacing_m * i) ** 2 + (spacing_m * j) ** 2 <= radius_m ** 2]
def _bucket(points: list[Vec2], cell: float) -> dict[tuple[int, int], list[int]]:
table: dict[tuple[int, int], list[int]] = {}
for index, (x, y) in enumerate(points):
table.setdefault((int(x // cell), int(y // cell)), []).append(index)
return table
def coverage_masks(targets, candidates: list[Vec2],
r_eff: float = R_EFF_MIN_M,
heading_margin_deg: float = 0.0) -> list[int]:
"""每个候选点覆盖的目标位掩码（Python 大整数，按位）。
只在与候选点空间桶相邻的格内做距离/前向判定，避免 |T|·|C| 笛卡尔积。
``r_eff`` / ``heading_margin_deg`` 是**设计裕量**：离线搜索时把"命中"
定义得比物理要求更严（距离 ≤ r_eff、前向夹角 ≤ 90°−margin），使采样格
上的覆盖自动蕴含相邻未采样 (p,h) 也被覆盖（S2 证书硬化）。
"""
table = _bucket(candidates, r_eff)
masks = [0] * len(candidates)
sin_margin = math.sin(math.radians(max(heading_margin_deg, 0.0)))
for t_index, (px, py, heading) in enumerate(targets):
ux, uy = math.cos(heading), math.sin(heading)
ci, cj = int(px // r_eff), int(py // r_eff)
bit = 1 << t_index
for di in (-1, 0, 1):
for dj in (-1, 0, 1):
for c_index in table.get((ci + di, cj + dj), ()):
cx, cy = candidates[c_index]
dx, dy = cx - px, cy - py
distance_sq = dx * dx + dy * dy
if distance_sq > r_eff * r_eff:
continue
if dx * ux + dy * uy < sin_margin * math.sqrt(distance_sq) - 1e-9:
continue
masks[c_index] |= bit
return masks
def greedy_cover(masks: list[int], n_targets: int) -> list[int]:
"""贪心集合覆盖：每次取"新增覆盖目标数"最大的候选，返回索引表。"""
full = (1 << n_targets) - 1
covered = 0
chosen: list[int] = []
remaining = list(range(len(masks)))
while covered != full and remaining:
best_index, best_gain = -1, 0
for c_index in remaining:
gain = (masks[c_index] & ~covered).bit_count()
if gain > best_gain:
best_index, best_gain = c_index, gain
if best_index < 0:
break
chosen.append(best_index)
covered |= masks[best_index]
remaining.remove(best_index)
return chosen
def nn_tour_length(points: list[Vec2], start: Vec2 = (0.0, 0.0)) -> float:
"""最近邻开放路径长度（巡回长度上界）。"""
remaining = list(points)
position = start
total = 0.0
while remaining:
nxt = min(remaining, key=lambda q: (q[0] - position[0]) ** 2
+ (q[1] - position[1]) ** 2)
remaining.remove(nxt)
total += math.dist(position, nxt)
position = nxt
return total
# --------------------------------------------------------------------- 校验
def _hit_distance(points: list[Vec2], position: Vec2, heading: float,
r_eff: float) -> float | None:
"""最近的前向命中距离；没有任何前向命中点则返回 None。"""
ux, uy = math.cos(heading), math.sin(heading)
best = None
for (cx, cy) in points:
dx, dy = cx - position[0], cy - position[1]
if dx * dx + dy * dy > r_eff * r_eff:
continue
if dx * ux + dy * uy < -1e-9:
continue
distance = math.hypot(dx, dy)
if best is None or distance < best:
best = distance
return best
def _nearest_distance(points: list[Vec2], position: Vec2,
r_eff: float) -> float | None:
"""最近距离（不要求前向）；超过 r_eff 视作未命中。"""
best = None
for (cx, cy) in points:
distance = math.hypot(cx - position[0], cy - position[1])
if distance <= r_eff and (best is None or distance < best):
best = distance
return best
def verify_forward_half_disk(points: list[Vec2], step_m: float = VERIFY_STEP_M,
headings: int = VERIFY_HEADINGS,
r_eff: float = R_EFF_MIN_M,
heading_margin_deg: float = 0.0) -> CoverageCertificate:
"""第四问语义：∀(p,h) 存在前向半盘内的被访点。
``heading_margin_deg`` 是**证书硬化**用的前向角裕量：把"命中"收紧为
"前向夹角 ≤ 90° − margin"。与 ``r_eff`` 的设计裕量合用时，采样格上的
覆盖即可推出**连续空间**覆盖：真实 (p',h') 距采样格 ≤ step·√2/2、
方位差 ≤ 180/headings 度，只要
r_eff + step·√2/2 ≤ 1000 且 margin ≥ 180/headings/2
就能保证 |c−p'| ≤ 1000 且 (c−p')·u(h') ≥ 0。这正是 S2 取代"粗采样
假通过"的方式。默认 0 表示按物理条件校核。
"""
points = list(points)
targets = forward_targets(step_m, headings)
table = _bucket(points, r_eff)
missing, worst = 0, 0.0
sin_margin = math.sin(math.radians(max(heading_margin_deg, 0.0)))
for (px, py, heading) in targets:
ux, uy = math.cos(heading), math.sin(heading)
ci, cj = int(px // r_eff), int(py // r_eff)
best = None
for di in (-1, 0, 1):
for dj in (-1, 0, 1):
for c_index in table.get((ci + di, cj + dj), ()):
cx, cy = points[c_index]
dx, dy = cx - px, cy - py
distance_sq = dx * dx + dy * dy
if distance_sq > r_eff * r_eff:
continue
if dx * ux + dy * uy < sin_margin * math.sqrt(distance_sq) - 1e-9:
continue
distance = math.sqrt(distance_sq)
if best is None or distance < best:
best = distance
if best is None:
missing += 1
else:
worst = max(worst, best)
return CoverageCertificate("forward_half_disk", step_m, headings, r_eff,
len(targets), missing, worst,
nn_tour_length(points))
def verify_full_disk(points: list[Vec2], step_m: float = VERIFY_STEP_M,
r_eff: float = R_EFF_MIN_M) -> CoverageCertificate:
"""第三问语义：Ω 内每点都在某个被访点的 r_eff 圆盘内。"""
points = list(points)
positions = omega_positions(step_m)
table = _bucket(points, r_eff)
missing, worst = 0, 0.0
for (px, py) in positions:
ci, cj = int(px // r_eff), int(py // r_eff)
best = None
for di in (-1, 0, 1):
for dj in (-1, 0, 1):
for c_index in table.get((ci + di, cj + dj), ()):
cx, cy = points[c_index]
distance = math.hypot(cx - px, cy - py)
if distance <= r_eff and (best is None or distance < best):
best = distance
if best is None:
missing += 1
else:
worst = max(worst, best)
return CoverageCertificate("full_disk", step_m, 1, r_eff, len(positions),
missing, worst, nn_tour_length(points))
def verify_plan(plan: CoveragePlan, step_m: float | None = None,
headings: int | None = None) -> CoverageCertificate:
"""按计划声明的覆盖语义校核。
默认使用**证书自带的采样分辨率**（离线 75 m / 24 档）：更粗的运行期采样
会放过"删掉一个关键点仍看似覆盖"的点集，使校核失去意义。校核按计划名
缓存（``load_plan`` 的 lru_cache），每进程只付一次成本。
"""
certificate = plan.certificate
if step_m is None:
step_m = certificate.target_step_m if certificate else VERIFY_STEP_M
if headings is None:
headings = (certificate.headings if certificate
else VERIFY_HEADINGS)
if plan.coverage == "forward_half_disk":
# 硬化证书：按证书自带的距离/角度设计裕量重新校核（旧证书默认
# r_eff=1000、margin=0，与历史口径逐位一致）。
return verify_forward_half_disk(
plan.points, step_m, headings,
r_eff=(certificate.r_eff_min_m if certificate else R_EFF_MIN_M),
heading_margin_deg=(certificate.verify_heading_margin_deg
if certificate else 0.0))
return verify_full_disk(plan.points, step_m, certificate.r_eff_min_m
if certificate else R_EFF_MIN_M)
# --------------------------------------------------------------------- 计划
def seven_points(ring_radius_m: float) -> list[Vec2]:
"""解析七站：原点 + 环上 6 等分点。"""
return [(0.0, 0.0)] + [(ring_radius_m * math.cos(k * math.pi / 3),
ring_radius_m * math.sin(k * math.pi / 3))
for k in range(6)]
def seven_plan(ring_radius_m: float) -> CoveragePlan:
"""第三问解析计划（覆盖证明由 validate_seven_coverage 给出）。"""
return CoveragePlan(
name=f"seven-r{ring_radius_m:.1f}", problem=3, kind="analytic",
coverage="full_disk",
note=("解析证明 max(r/√3, sqrt(R²+r²−√3Rr)) < 1000；"
"r∈[1122.9, 1732] 时成立，巡检半径越小巡回越短"),
points=tuple(seven_points(ring_radius_m)))
def square_49_plan() -> CoveragePlan:
"""第四问旧计划：700 m 方格 49 点（充分条件，保留作回退与对照）。"""
return CoveragePlan(
name="square49", problem=4, kind="certified",
coverage="forward_half_disk",
note="单元对角 700√2≈989.95≤1000 的充分条件；点数为最小命中集的 1.6 倍",
# 顺序与 coverage.coverage_points() 完全一致（y 由大到小），保持历史
# 基线 grid49_immediate 的逐位可复现。
points=tuple((700.0 * i, 700.0 * j)
for j in range(3, -4, -1) for i in range(-3, 4)))
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)
In [ ]:
def seven_points(ring_radius_m: float) -> list[Vec2]:
"""解析七站：原点 + 环上 6 等分点。"""
return [(0.0, 0.0)] + [(ring_radius_m * math.cos(k * math.pi / 3),
ring_radius_m * math.sin(k * math.pi / 3))
for k in range(6)]
def seven_plan(ring_radius_m: float) -> CoveragePlan:
"""第三问解析计划（覆盖证明由 validate_seven_coverage 给出）。"""
return CoveragePlan(
name=f"seven-r{ring_radius_m:.1f}", problem=3, kind="analytic",
coverage="full_disk",
note=("解析证明 max(r/√3, sqrt(R²+r²−√3Rr)) < 1000；"
"r∈[1122.9, 1732] 时成立，巡检半径越小巡回越短"),
points=tuple(seven_points(ring_radius_m)))
def square_49_plan() -> CoveragePlan:
"""第四问旧计划：700 m 方格 49 点（充分条件，保留作回退与对照）。"""
return CoveragePlan(
name="square49", problem=4, kind="certified",
coverage="forward_half_disk",
note="单元对角 700√2≈989.95≤1000 的充分条件；点数为最小命中集的 1.6 倍",
# 顺序与 coverage.coverage_points() 完全一致（y 由大到小），保持历史
# 基线 grid49_immediate 的逐位可复现。
points=tuple((700.0 * i, 700.0 * j)
for j in range(3, -4, -1) for i in range(-3, 4)))
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)
In [ ]:
# 覆盖计划以字面量内嵌（原实现从 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 = {
"name": "q4_mincover_v2",
"problem": 4,
"kind": "certified",
"coverage": "forward_half_disk",
"note": "设计裕量最小命中集：31 点，r_design=965 m / 前向角裕量 5.0°；证书按 15 m/72 档、r_verify=1000 m、角裕量 0.0° 校核缺口 0（同一证书口径复核，含精细剪枝）；最坏命中 990 m，NN 巡回 23.9 km",
"generator": {
"script": "scripts/verify_q4_cover.py",
"spacing_m": 100.0,
"radius_m": 2850.0,
"target_step_m": 50.0,
"headings": 48,
"design_r_eff_m": 965.0,
"heading_margin_deg": 5.0,
"pruned": True,
"verify_step_m": 15.0,
"verify_headings": 72,
"verify_r_eff_m": 1000.0,
"verify_heading_margin_deg": 0.0,
"code_sha256": "adf4fe6e9853c47314fac2c630fc6846ea6922102c2e83b507e5d397434c0e1d"
},
"certificate": {
"kind": "forward_half_disk",
"target_step_m": 15.0,
"headings": 72,
"r_eff_min_m": 1000.0,
"targets": 3256200,
"uncovered": 0,
"worst_hit_distance_m": 990.051,
"nn_tour_m": 23883.8,
"nn_tour_virtual_s": 4776.8,
"generator_sha256": "adf4fe6e9853c47314fac2c630fc6846ea6922102c2e83b507e5d397434c0e1d",
"verify_heading_margin_deg": 0.0
},
"points": [
(-800, -200), (800, 200), (-200, 900), (200, -900), (-1200, 700), (1200, -700),
(800, 1200), (-1800, -300), (1800, 300), (-400, 1800), (0, 0), (1400, -1400),
(-1500, -1300), (1500, 1300), (-400, -1800), (400, 1800), (-1800, 400), (1800, -400),
(-500, 700), (-700, -700), (700, 700), (1300, 1100), (-1300, 1500), (500, -1800),
(-900, -2100), (600, 1900), (-2500, 500), (1900, -900), (-1700, -500), (-700, -1200),
(900, -700),
],
}
Q4_MINCOVER_V1 = {
"name": "q4_mincover_v1",
"problem": 4,
"kind": "certified",
"coverage": "forward_half_disk",
"note": "贪心击中集等价覆盖：31 点 vs 现役 49 点，NN 巡回 22.9 km vs 36.4 km",
"generator": {
"script": "scripts/verify_q4_cover.py",
"spacing_m": 200.0,
"radius_m": 2850.0,
"target_step_m": 75.0,
"headings": 24,
"code_sha256": "dce9f0e189439bbea26f6bcbebf7e7f7bb99021a9035fca0c8d67bb67a3a529a"
},
"certificate": None,
"points": [
(-800, 0), (800, 0), (0, -1000), (0, 1000), (-1200, -1000), (-1000, 1000),
(1000, -1000), (1000, 1000), (1800, 0), (-1800, 200), (-400, -1800), (-200, 1800),
(0, 0), (800, -1800), (1200, 1600), (-1400, 1400), (1800, -800), (-1800, -800),
(1800, 600), (-600, -1000), (400, 1800), (-1400, -1800), (-1200, 800), (-1200, 1800),
(400, -800), (-200, 400), (600, -1400), (-2200, 600), (1000, -1600), (-2000, -400),
(-200, -2200),
],
}
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"]))
In [ ]:
RING_RADIUS_M = 1200.0 # [工程] 解析证明见 docs/algo-3.md §5 与 geometry_check
def seven_points(ring_radius_m: float = RING_RADIUS_M) -> list[tuple[float, float]]:
"""原点 + 环上 6 等分点的七站覆盖集。"""
return [(0.0, 0.0)] + [(ring_radius_m * math.cos(k * math.pi / 3),
ring_radius_m * math.sin(k * math.pi / 3)) for k in range(6)]
def validate_seven_coverage(ring_radius_m: float, config: ProblemConfig) -> float:
"""六个60°扇区内的最近站最远距离上界；仅证明全向源覆盖。
原点/环站Voronoi交点给出r/√3，目标圆边界的扇区接缝给出另一项。
环半径满足0<r<=√3R时这两项覆盖全部极值。
"""
radius = config.field.omega_radius_m
if not 0 < ring_radius_m <= math.sqrt(3)*radius:
raise ValueError("七站环半径不满足全域覆盖证明的前提")
bound = max(ring_radius_m / math.sqrt(3), math.sqrt(
radius**2 + ring_radius_m**2 - math.sqrt(3)*radius*ring_radius_m))
if bound > config.field.r_eff_min_m - 1e-6:
raise ValueError("七站未覆盖整个目标域，不能据此证明频道缺席")
return bound
In [ ]:
_ANG_EPS_DEG = 1e-9 # 角向 kill 比较的浮点余量
_DIST_EPS_M = 1e-9
_COARSE_FACTOR = 4 # 粗格每边聚合的细格数
try: # numpy 仅作**可选**加速后端；缺失时全部回退纯 Python
import numpy as _np
except Exception: # pragma: no cover - 环境无 numpy 时走纯 Python 分支
_np = None
if os.environ.get("MCM26B_NO_NUMPY"): # 强制纯 Python：对拍与回退验证用
_np = None
def angular_gap(a_deg: float, b_deg: float) -> float:
"""两个方位角（度）的最小夹角。"""
return abs((a_deg - b_deg + 180.0) % 360.0 - 180.0)
def as_sample_array(samples):
"""把 [(x,y), ...] 采样转成 (n,2) float64 数组；numpy 缺失时原样返回。
评分为 O(代表点 × 子采样) 的向量运算，是单局 CPU 的主要开销
（实测占 72%）。调用方在每次选点前转换一次，评分函数按类型分派，
两条实现（numpy / 纯 Python）语义逐位对齐，见 tests/test_particles.py。
"""
if _np is None or isinstance(samples, _np.ndarray) or len(samples) == 0:
return samples
return _np.asarray(samples, dtype=float)
def _branch_radius_array(members) -> float:
"""分支后验半径的数组版：到质心的最大距离。"""
centroid = members.mean(axis=0)
return float(_np.max(_np.hypot(members[:, 0] - centroid[0],
members[:, 1] - centroid[1])))
class ChannelParticles:
"""单个频道的网格幸存集。
网格为覆盖 Ω 外扩一格的轴对齐方格（边长 h），索引 ``i*W + j``；
``alive`` 是幸存格索引集合。初始集 = 格心距原点 <= Ω 半径 + 裕量的
全部格子，因此真源（|p| <= 1800 [题面]）所在格必在初始集内。
``version`` 随每次删除自增，调用方可据此缓存派生量（MEC、跳测判定）。
"""
def __init__(self, config: ProblemConfig = ProblemConfig(),
spacing_m: float | None = None):
field = config.field
self.config = config
self.spacing_m = float(spacing_m if spacing_m is not None
else config.bayes.particle_spacing_m)
if not 1.0 <= self.spacing_m <= 200.0:
raise ValueError("网格间距须在 [1, 200] 米")
self.margin_m = self.spacing_m * math.sqrt(2) / 2 # Voronoi 裕量
self.r_eff_min_m = field.r_eff_min_m
self.r_eff_max_m = field.r_eff_max_m
self.near_radius_m = field.near_radius_m
self.half_angle_deg = (field.bearing_error_half_deg
+ config.search.bearing_rounding_margin_deg)
# 覆盖 [-R-m, R+m]² 的格点阵列；格 (i,j) 的中心在
# (x0 + (i+0.5)h, x0 + (j+0.5)h)，中心到格内任一点 <= h·√2/2 = 裕量
span = field.omega_radius_m + self.margin_m
self.x0 = -span
self.width = int(math.ceil(2 * span / self.spacing_m)) + 1
limit_sq = (span + 1e-9) ** 2
self.alive: set[int] = set()
self._coarse: dict[int, int] = {} # 粗格索引 -> 幸存细格数
coarse_w = self._coarse_width()
for i in range(self.width):
x = self.x0 + (i + 0.5) * self.spacing_m
for j in range(self.width):
if x * x + (self.x0 + (j + 0.5) * self.spacing_m) ** 2 <= limit_sq:
index = i * self.width + j
self.alive.add(index)
key = (i // _COARSE_FACTOR) * coarse_w + j // _COARSE_FACTOR
self._coarse[key] = self._coarse.get(key, 0) + 1
self.version = 0
self._mec: tuple[Vec2, float] | None = None
# ------------------------------------------------------------ 索引与坐标
def _coarse_width(self) -> int:
return (self.width + _COARSE_FACTOR - 1) // _COARSE_FACTOR
def cell_center(self, index: int) -> Vec2:
i, j = divmod(index, self.width)
return (self.x0 + (i + 0.5) * self.spacing_m,
self.x0 + (j + 0.5) * self.spacing_m)
def discard(self, indices: list[int]) -> None:
"""公开删除入口：联合信念把"朝向不可能"的格一并删除时使用。"""
self._discard(indices)
def _discard(self, indices: list[int]) -> None:
if not indices:
return
coarse_w = self._coarse_width()
for index in indices:
if index not in self.alive:
continue
self.alive.discard(index)
key = (index // self.width // _COARSE_FACTOR) * coarse_w \
+ (index % self.width) // _COARSE_FACTOR
count = self._coarse.get(key, 0) - 1
if count <= 0:
self._coarse.pop(key, None)
else:
self._coarse[key] = count
self.version += 1
self._mec = None
# ------------------------------------------------------------ 观测更新
def _kill_disk(self, center: Vec2, radius_m: float) -> None:
"""删除格心距 center <= radius_m 的全部幸存格（整格判据见模块注释）。"""
if not self.alive:
return
reach = radius_m + _DIST_EPS_M
ilo = max(0, int(math.floor((center[0] - reach - self.x0) / self.spacing_m)))
ihi = min(self.width - 1,
int(math.ceil((center[0] + reach - self.x0) / self.spacing_m)))
jlo = max(0, int(math.floor((center[1] - reach - self.x0) / self.spacing_m)))
jhi = min(self.width - 1,
int(math.ceil((center[1] + reach - self.x0) / self.spacing_m)))
limit_sq = reach * reach
doomed = []
for i in range(ilo, ihi + 1):
dx = self.x0 + (i + 0.5) * self.spacing_m - center[0]
base = i * self.width
for j in range(jlo, jhi + 1):
index = base + j
if index not in self.alive:
continue
dy = self.x0 + (j + 0.5) * self.spacing_m - center[1]
if dx * dx + dy * dy <= limit_sq:
doomed.append(index)
self._discard(doomed)
def _kill_outside_disk(self, center: Vec2, radius_m: float) -> None:
"""删除格心距 center > radius_m 的全部幸存格（保留盘内）。"""
if not self.alive:
return
reach = radius_m + _DIST_EPS_M
doomed = [index for index in self.alive
if math.dist(self.cell_center(index), center) > reach]
self._discard(doomed)
def _kill_outside_wedge(self, station: Vec2, reading_deg: float) -> None:
"""删除角向偏离读数超出 δ_eff + 角向格裕量的幸存格。
角向格裕量 = asin(m / max(dist−m, 1))：格内任一点相对格心的方位
偏差不超过它（点距测点还可再近 m，故分母再内缩 m）。
"""
if not self.alive:
return
sx, sy = float(station[0]), float(station[1])
doomed = []
for index in self.alive:
gx, gy = self.cell_center(index)
dist = math.hypot(gx - sx, gy - sy)
if dist <= self.margin_m + 1.0:
continue # 格贴测点：格内方位可指向任意方向，不可删
angular_margin = math.degrees(
math.asin(min(0.9999, self.margin_m / (dist - self.margin_m))))
gap = angular_gap(math.degrees(math.atan2(gy - sy, gx - sx)), reading_deg)
if gap > self.half_angle_deg + angular_margin + _ANG_EPS_DEG:
doomed.append(index)
self._discard(doomed)
def apply_measure(self, station: Vec2, result: str,
svd_deg: float | None = None, *,
directional: bool = False) -> None:
"""按读数更新幸存集。真源保留性由 Voronoi 裕量保证（模块注释）。
``directional=True``（第四问）：no_signal **不删空间候选**——定向源
的静默还可能来自朝向遮挡，单独删空间候选会误删真源；朝向维度由
:class:`~mcm26b.belief.joint.JointBelief` 维护。
"""
if result == NO_SIGNAL:
if directional:
return
# no_signal ⟹ dist > r_eff >= 1000：整格落入 B(s, 1000−m) 的删。
self._kill_disk(station, self.r_eff_min_m - self.margin_m)
elif result == NEAR:
self._kill_outside_disk(station, self.near_radius_m + self.margin_m)
elif result == DIRECTION:
if svd_deg is None:
raise ValueError("direction 读数必须携带 svd_deg")
# direction 隐含 dist <= r_max [题面]
self._kill_outside_disk(station, self.r_eff_max_m + self.margin_m)
self._kill_outside_wedge(station, svd_deg)
else:
raise ValueError(f"未知测向结果: {result!r}")
# ------------------------------------------------------------ 查询
def __len__(self) -> int:
return len(self.alive)
def alive_points(self, limit: int | None = None) -> list[Vec2]:
sorted_alive = sorted(self.alive)
if limit is not None and len(sorted_alive) > limit:
step = len(sorted_alive) / limit
indices = {sorted_alive[int(k * step) % len(sorted_alive)]
for k in range(limit)}
indices = sorted(indices)
else:
indices = sorted_alive
return [self.cell_center(index) for index in indices]
def nearest_alive_distance(self, point: Vec2, limit_m: float) -> float | None:
"""最近幸存格心距离；> limit_m 时返回 None（粗格环扫 + 下界剪枝）。
返回值语义：恰为最近幸存格心距离（若 <= limit_m）；None 表示所有
幸存格心都超出 limit_m。粗格聚合半径已计入扫描范围，不漏判。
"""
if not self.alive:
return None
coarse_cell_m = self.spacing_m * _COARSE_FACTOR
coarse_reach = coarse_cell_m * math.sqrt(2) / 2
cw = self._coarse_width()
ci_base = int(math.floor((point[0] - self.x0) / coarse_cell_m))
cj_base = int(math.floor((point[1] - self.x0) / coarse_cell_m))
ring_max = int(math.ceil((limit_m + coarse_reach) / coarse_cell_m)) + 1
best: float | None = None
for ring in range(ring_max + 1):
lower_bound = max(0.0, ring - 1) * coarse_cell_m - coarse_reach
if best is not None and lower_bound >= best:
break # 更外环的所有格心都 >= 下界 >= 当前最优
for ci in range(max(0, ci_base - ring), min(cw - 1, ci_base + ring) + 1):
for cj in range(max(0, cj_base - ring), min(cw - 1, cj_base + ring) + 1):
if max(abs(ci - ci_base), abs(cj - cj_base)) != ring:
continue # 只扫当前环（Chebyshev 环）
if (ci * cw + cj) not in self._coarse:
continue
ccx = self.x0 + (ci + 0.5) * coarse_cell_m
ccy = self.x0 + (cj + 0.5) * coarse_cell_m
if math.hypot(ccx - point[0], ccy - point[1]) \
> limit_m + coarse_reach:
continue
i0, j0 = ci * _COARSE_FACTOR, cj * _COARSE_FACTOR
for i in range(i0, min(i0 + _COARSE_FACTOR, self.width)):
base = i * self.width
for j in range(j0, min(j0 + _COARSE_FACTOR, self.width)):
index = base + j
if index not in self.alive:
continue
dist = math.dist(self.cell_center(index), point)
if dist <= limit_m and (best is None or dist < best):
best = dist
return best
def guaranteed_no_signal(self, station: Vec2) -> bool:
"""真源与该测点距离必 > r_max（>= r_eff）⟹ 必 no_signal，测了也无信息。"""
return self.nearest_alive_distance(
station, self.r_eff_max_m + self.margin_m) is None
def mec(self) -> tuple[Vec2, float]:
"""幸存集最小包围圆（圆心, 半径）；按 version 缓存。"""
if self._mec is None:
if not self.alive:
raise ValueError("幸存集为空（该频道已可证明缺席）")
points = [self.cell_center(index) for index in self.alive]
self._mec = smallest_enclosing_circle(points)
return self._mec
# ------------------------------------------------------------ 信息增益
def subsample(self, size: int) -> list[Vec2]:
"""按索引序均匀抽取的子采样（索引序≈空间序，代表点空间铺开）。"""
return self.alive_points(limit=size)
# ---------------------------------------------------- 向量化评分（可选）
def _expected_survival_vector(self, point: Vec2, sub, representatives) -> float:
"""``expected_survival`` 的向量化实现（与纯 Python 版逐分支同语义）。"""
count = len(sub)
eps = 1.0 / count
dx, dy = sub[:, 0] - point[0], sub[:, 1] - point[1]
dist = _np.hypot(dx, dy)
within = int(_np.count_nonzero(
dist <= self.r_eff_min_m - self.margin_m + _DIST_EPS_M))
log_keep_ns = math.log(max(1.0 - within / count, eps))
bearing = _np.degrees(_np.arctan2(dy, dx))
in_range = dist <= self.r_eff_max_m + self.margin_m
rdx, rdy = representatives[:, 0] - point[0], representatives[:, 1] - point[1]
rdist = _np.hypot(rdx, rdy)
rbearing = _np.degrees(_np.arctan2(rdy, rdx))
gaps = _np.abs((bearing[None, :] - rbearing[:, None] + 180.0) % 360.0 - 180.0)
survivors = _np.count_nonzero(
(gaps <= self.half_angle_deg + _ANG_EPS_DEG) & in_range[None, :], axis=1)
p_ns = _np.clip((rdist - self.r_eff_min_m)
/ (self.r_eff_max_m - self.r_eff_min_m), 0.0, 1.0)
keep_dir = _np.maximum(survivors / count, eps)
values = p_ns * log_keep_ns + (1.0 - p_ns) * _np.log(keep_dir)
near = rdist <= self.near_radius_m + self.margin_m
values = _np.where(near, math.log(eps), values)
return float(values.mean())
def _expected_posterior_radius_vector(self, point: Vec2, sub,
representatives) -> float:
"""``expected_posterior_radius`` 的向量化实现。"""
eps_radius = self.spacing_m / 2.0
dx, dy = sub[:, 0] - point[0], sub[:, 1] - point[1]
dist = _np.hypot(dx, dy)
bearing = _np.degrees(_np.arctan2(dy, dx))
keep_ns = dist > self.r_eff_min_m - self.margin_m
r_ns = (_branch_radius_array(sub[keep_ns]) if keep_ns.any()
else eps_radius)
rdx, rdy = representatives[:, 0] - point[0], representatives[:, 1] - point[1]
rdist = _np.hypot(rdx, rdy)
rbearing = _np.degrees(_np.arctan2(rdy, rdx))
gaps = _np.abs((bearing[None, :] - rbearing[:, None] + 180.0) % 360.0 - 180.0)
dir_mask = ((gaps <= self.half_angle_deg + _ANG_EPS_DEG)
& (dist[None, :] <= self.r_eff_max_m + self.margin_m))
r_dir = _np.array([_branch_radius_array(sub[row]) if row.any()
else eps_radius for row in dir_mask])
p_ns = _np.clip((rdist - self.r_eff_min_m)
/ (self.r_eff_max_m - self.r_eff_min_m), 0.0, 1.0)
near = rdist <= self.near_radius_m + self.margin_m
values = _np.where(near, self.near_radius_m + self.margin_m,
p_ns * r_ns + (1.0 - p_ns) * r_dir)
return float(values.mean())
def expected_survival(self, point: Vec2, subsample,
representatives) -> float:
"""E[log 幸存占比]：越小该测点的期望信息增益越大。
r_eff 不确定性解析积分：dist∈(1000,1500] 时按 r_eff~U(1000,1500)
取 p_ns=(dist−1000)/500 为 no_signal 分支概率。
- no_signal 分支：删 B(point, 1000−m)，幸存 = 子采样中盘外比例；
- direction 分支：以代表点方位为期望读数，角域 ∩ B(point,1500+m) 内幸存；
- near 分支（dist <= 5+m）：可行集收缩到 5m 盘，幸存占比趋近 0。
"""
if not self.alive or len(subsample) == 0 or len(representatives) == 0:
return 0.0
if _np is not None and isinstance(subsample, _np.ndarray) \
and isinstance(representatives, _np.ndarray):
return self._expected_survival_vector(point, subsample, representatives)
eps = 1.0 / len(subsample)
keep_ns = max(1.0 - self.fraction_within(subsample, point,
self.r_eff_min_m - self.margin_m),
eps)
log_keep_ns = math.log(keep_ns)
total = 0.0
for rep in representatives:
dist = math.hypot(rep[0] - point[0], rep[1] - point[1])
if dist <= self.near_radius_m + self.margin_m:
total += math.log(eps)
continue
p_ns = min(1.0, max(0.0, (dist - self.r_eff_min_m)
/ (self.r_eff_max_m - self.r_eff_min_m)))
reading = math.atan2(rep[1] - point[1], rep[0] - point[0])
survivors = 0
for p in subsample:
if math.dist(p, point) > self.r_eff_max_m + self.margin_m:
continue
gap = angular_gap(math.degrees(
math.atan2(p[1] - point[1], p[0] - point[0])),
math.degrees(reading))
if gap <= self.half_angle_deg + _ANG_EPS_DEG:
survivors += 1
keep_dir = max(survivors / len(subsample), eps)
total += p_ns * log_keep_ns + (1.0 - p_ns) * math.log(keep_dir)
return total / max(len(representatives), 1)
def fraction_within(self, samples, point: Vec2, radius_m: float) -> float:
if len(samples) == 0:
return 0.0
reach = radius_m + _DIST_EPS_M
if _np is not None and isinstance(samples, _np.ndarray):
distance = _np.hypot(samples[:, 0] - point[0], samples[:, 1] - point[1])
return float(_np.count_nonzero(distance <= reach)) / len(samples)
count = sum(1 for p in samples if math.dist(p, point) <= reach)
return count / len(samples)
def _branch_radius(self, members) -> float:
"""分支后验的半径代理：幸存子采样到其质心的最大距离（O(n)，免 Welzl）。"""
if len(members) == 0:
return 0.0
if _np is not None and isinstance(members, _np.ndarray):
return _branch_radius_array(members)
mx = sum(p[0] for p in members) / len(members)
my = sum(p[1] for p in members) / len(members)
return max(math.hypot(p[0] - mx, p[1] - my) for p in members)
def expected_posterior_radius(self, point: Vec2, subsample,
representatives) -> float:
"""一次测量后期望后验半径（米）：选点主指标，直指清除证书阈值。
对每个代表假设（真值候选）按 r_eff~U(1000,1500) 解析积分
no_signal/direction 分支，near 分支（dist <= 5+m）后验为 5m 盘；
各分支半径用幸存子采样的质心最大距离近似（偏差 << 格距，排序用）。
与几何版 worst-posterior 同族，但作用于含负观测的非凸幸存集，
且以期望代替最坏（代表点覆盖幸存集主轴）。
"""
if not self.alive or len(subsample) == 0 or len(representatives) == 0:
return 0.0
if _np is not None and isinstance(subsample, _np.ndarray) \
and isinstance(representatives, _np.ndarray):
return self._expected_posterior_radius_vector(
point, subsample, representatives)
eps_radius = self.spacing_m / 2.0
total = 0.0
for rep in representatives:
dist = math.hypot(rep[0] - point[0], rep[1] - point[1])
if dist <= self.near_radius_m + self.margin_m:
total += self.near_radius_m + self.margin_m
continue
p_ns = min(1.0, max(0.0, (dist - self.r_eff_min_m)
/ (self.r_eff_max_m - self.r_eff_min_m)))
members_ns = [p for p in subsample
if math.dist(p, point) > self.r_eff_min_m - self.margin_m]
reading = math.atan2(rep[1] - point[1], rep[0] - point[0])
members_dir = []
for p in subsample:
if math.dist(p, point) > self.r_eff_max_m + self.margin_m:
continue
gap = angular_gap(math.degrees(
math.atan2(p[1] - point[1], p[0] - point[0])),
math.degrees(reading))
if gap <= self.half_angle_deg + _ANG_EPS_DEG:
members_dir.append(p)
r_ns = self._branch_radius(members_ns) if members_ns else eps_radius
r_dir = self._branch_radius(members_dir) if members_dir else eps_radius
total += p_ns * r_ns + (1.0 - p_ns) * r_dir
return total / max(len(representatives), 1)
In [ ]:
HEADING_BINS = 24 # 15° 一档
_BIN_HALF_DEG = 180.0 / HEADING_BINS # 7.5°
_FULL_MASK = (1 << HEADING_BINS) - 1
_GRID_DEG = 3.0 # 角度查表分辨率
_GRID_N = int(round(360.0 / _GRID_DEG))
# numpy 仅作可选加速后端（与 particles 同一开关：无 numpy 时走纯 Python 分支）
_np = _np
def heading_gap(a_deg: float, b_deg: float) -> float:
"""两个朝向角的最小夹角（度）。"""
return abs((a_deg - b_deg + 180.0) % 360.0 - 180.0)
def _cover_mask(bearing_deg: float, half_deg: float) -> int:
"""与该方位夹角不超过 ``half_deg`` 的朝向档掩码。"""
mask = 0
for k in range(HEADING_BINS):
if heading_gap(360.0 * k / HEADING_BINS, bearing_deg) <= half_deg:
mask |= 1 << k
return mask
# 查表：以 bearing 为轴、半角 90° 的可见半平面。
# - COVER：**可能**覆盖该方位（含半档与查表舍入的保守侧）；
# - SILENT：**确定**覆盖该方位（内缩半档与舍入）——离散对照表，供审计与
# 文档比对；实际裁剪走 _cover_mask / _far_mask，因为还必须逐格计入
# "格内角裕量"（格心到格内任一点的方位偏差上界），无法预先制表。
_COVER_TABLE = tuple(
_cover_mask(_GRID_DEG * i, 90.0 + _BIN_HALF_DEG + _GRID_DEG / 2)
for i in range(_GRID_N))
_SILENT_TABLE = tuple(
_cover_mask(_GRID_DEG * i, 90.0 - _BIN_HALF_DEG - _GRID_DEG / 2)
for i in range(_GRID_N))
def _far_mask(bearing_deg: float, exclude_half_deg: float) -> int:
"""与方位夹角**大于** ``exclude_half_deg`` 的朝向档掩码（no_signal 用）。
与 :func:`_cover_mask` 互补：no_signal 只允许保留"确定看不见"的朝向档。
"""
mask = 0
for k in range(HEADING_BINS):
if heading_gap(360.0 * k / HEADING_BINS, bearing_deg) > exclude_half_deg:
mask |= 1 << k
return mask
if _np is not None: # 12 位半字 popcount 查表（朝向掩码 24 位）与半平面查表数组
_POP16 = _np.array([bin(value).count("1") for value in range(1 << 12)],
dtype=_np.int64)
_COVER_ARR = _np.array(_COVER_TABLE, dtype=_np.int64)
else: # pragma: no cover - 无 numpy 分支
_POP16 = _COVER_ARR = None
def _popcount(mask: int) -> int:
return bin(mask).count("1")
def _table_index(bearing_deg: float) -> int:
return int(round((bearing_deg % 360.0) / _GRID_DEG)) % _GRID_N
class JointBelief:
"""单频道 (位置 × 朝向) 联合可行集。"""
def __init__(self, config: ProblemConfig = ProblemConfig()):
self.config = config
self.field = config.field
self.position = ChannelParticles(config)
self.masks: dict[int, int] = {} # 格索引 -> 朝向掩码（缺省为全档）
self._sample_cache: tuple[int, list[int]] | None = None
# ------------------------------------------------------------ 查询
def __len__(self) -> int:
return len(self.position)
def is_absent(self) -> bool:
"""联合集空 ⟹ 该频道不存在源（真源保留定理的二维推广）。"""
return not self.position.alive
def _mask(self, index: int) -> int:
return self.masks.get(index, _FULL_MASK)
def sample(self, size: int = 256) -> list[int]:
"""均匀抽样幸存格（评分用；证书始终用完整集合）。"""
cached = self._sample_cache
if cached is not None and cached[0] == self.position.version:
return cached[1]
alive = sorted(self.position.alive)
if len(alive) > size:
step = len(alive) / size
alive = [alive[int(k * step) % len(alive)] for k in range(size)]
self._sample_cache = (self.position.version, alive)
return alive
def _mask_popcount(self, index: int) -> int:
return _popcount(self._mask(index))
def expected_posterior_radius(self, reader: Vec2, size: int | None = None) -> float:
"""一次测量后的期望后验 MEC 半径（米）；D1 值函数。
与 :meth:`ChannelParticles.expected_posterior_radius` 同族，差别只在
no_signal 分支概率：定向源不仅会因距离 > r_eff 静默，还会因朝向遮挡
静默，因此
p_vis(p) = P(r_eff ≥ |p−q|) × cover_ratio(p, q)
其中 ``cover_ratio`` 是该格朝向掩码中"可能覆盖 q 方位"的档位占比。
分支半径语义：
- direction（可见，非 near）：读数为 q 处的方位 ±δ，后验收缩到该方位
楔形与当前可行集之交，半径取幸存子采样的质心最大距离；
- no_signal：定向源的静默**不删空间候选**，位置后验半径保持当前值；
- near：后验为 5m 盘。
返回值只用于**选点排序**，不参与任何安全判定；证书仍由凸/网格口径给出。
"""
pf = self.position
if not pf.alive:
return 0.0
if size is None:
size = self.config.bayes.score_subsample
indices = self.sample(max(2, size))
if not indices:
return 0.0
stride = max(len(indices) // max(self.config.bayes.mc_representatives, 1), 1)
representatives = indices[::stride]
if _np is not None:
return self._epr_vector(reader, indices, representatives)
return self._epr_python(reader, indices, representatives) # pragma: no cover
def _epr_vector(self, reader: Vec2, indices: list[int],
representatives: list[int]) -> float:
pf = self.position
points = _np.array([pf.cell_center(index) for index in indices], dtype=float)
masks = _np.array([self._mask(index) for index in indices], dtype=_np.int64)
reps = _np.array([pf.cell_center(index) for index in representatives],
dtype=float)
rep_masks = _np.array([self._mask(index) for index in representatives],
dtype=_np.int64)
dx, dy = points[:, 0] - reader[0], points[:, 1] - reader[1]
dist = _np.hypot(dx, dy)
bearing = _np.degrees(_np.arctan2(dy, dx))
rdx, rdy = reps[:, 0] - reader[0], reps[:, 1] - reader[1]
rdist = _np.hypot(rdx, rdy)
rbearing = _np.degrees(_np.arctan2(rdy, rdx))
cover = _COVER_ARR[_np.round(_np.mod(rbearing, 360.0) / _GRID_DEG)
.astype(_np.int64) % _GRID_N]
visible_mask = rep_masks & cover
bits = (_POP16[visible_mask & 0xFFF]
+ _POP16[(visible_mask >> 12) & 0xFFF])
cover_ratio = bits / float(HEADING_BINS)
p_silent_range = _np.clip((rdist - self.field.r_eff_min_m)
/ (self.field.r_eff_max_m
- self.field.r_eff_min_m), 0.0, 1.0)
p_vis = (1.0 - p_silent_range) * cover_ratio
p_ns = 1.0 - p_vis
gaps = _np.abs((bearing[None, :] - rbearing[:, None] + 180.0) % 360.0 - 180.0)
in_wedge = ((gaps <= self.config.field.bearing_error_half_deg
+ self.config.search.bearing_rounding_margin_deg + 1e-9)
& (dist[None, :] <= self.field.r_eff_max_m + pf.margin_m))
eps_radius = pf.spacing_m / 2.0
radii = []
for row in in_wedge:
member = points[row]
radii.append(_branch_radius_array(member)
if len(member) else eps_radius)
radii = _np.array(radii, dtype=float)
current = pf.mec()[1]
near = rdist <= self.field.near_radius_m + pf.margin_m
values = _np.where(near, self.field.near_radius_m + pf.margin_m,
p_ns * current + p_vis * radii)
return float(values.mean())
def _epr_python(self, reader: Vec2, indices: list[int],
representatives: list[int]) -> float: # pragma: no cover
"""纯 Python 回退（与向量版逐分支同语义，供无 numpy 环境与对拍）。"""
pf = self.position
points = [pf.cell_center(index) for index in indices]
masks = [self._mask(index) for index in indices]
eps_radius = pf.spacing_m / 2.0
current = pf.mec()[1]
half_angle = (self.config.field.bearing_error_half_deg
+ self.config.search.bearing_rounding_margin_deg)
total = 0.0
for rep_index in representatives:
rx, ry = pf.cell_center(rep_index)
distance = math.hypot(rx - reader[0], ry - reader[1])
if distance <= self.field.near_radius_m + pf.margin_m:
total += self.field.near_radius_m + pf.margin_m
continue
reading = math.degrees(math.atan2(ry - reader[1], rx - reader[0]))
cover = _COVER_TABLE[_table_index(reading)]
bits = _popcount(self._mask(rep_index) & cover)
ratio = bits / float(HEADING_BINS)
if distance <= self.field.r_eff_min_m:
p_range = 1.0
elif distance >= self.field.r_eff_max_m:
p_range = 0.0
else:
p_range = ((self.field.r_eff_max_m - distance)
/ (self.field.r_eff_max_m - self.field.r_eff_min_m))
p_vis = p_range * ratio
members = []
for point, mask in zip(points, masks):
if math.dist(point, reader) > self.field.r_eff_max_m + pf.margin_m:
continue
bearing = math.degrees(math.atan2(point[1] - reader[1],
point[0] - reader[0]))
if heading_gap(bearing, reading) <= half_angle + 1e-9:
members.append(point)
if members:
mx = sum(p[0] for p in members) / len(members)
my = sum(p[1] for p in members) / len(members)
radius = max(math.hypot(p[0] - mx, p[1] - my) for p in members)
else:
radius = eps_radius
total += (1.0 - p_vis) * current + p_vis * radius
return total / max(len(representatives), 1)
def prob_detect(self, reader: Vec2, size: int = 256) -> float:
"""在 ``reader`` 处能看见任一可行 (位置, 朝向) 的概率质量（启发式）。
r_eff ~ U[1000, 1500] 解析积分；朝向按可行档位占比计权。
只用于选点排序，不参与任何安全判定。
"""
cells = self.sample(size)
if not cells:
return 0.0
total = 0.0
span = self.field.r_eff_max_m - self.field.r_eff_min_m
for index in cells:
gx, gy = self.position.cell_center(index)
distance = math.hypot(reader[0] - gx, reader[1] - gy)
if distance > self.field.r_eff_max_m:
continue
if distance <= self.field.r_eff_min_m:
p_range = 1.0
else:
p_range = (self.field.r_eff_max_m - distance) / span
if p_range <= 0.0:
continue
bearing = math.degrees(math.atan2(reader[1] - gy, reader[0] - gx))
cover = _COVER_TABLE[_table_index(bearing)]
bits = bin(self._mask(index) & cover).count("1")
if bits:
total += p_range * bits / HEADING_BINS
return total / len(cells)
# ------------------------------------------------------------ 观测更新
def _apply_heading(self, station: Vec2, keep: bool,
within_m: float | None = None) -> None:
"""按可见性约束裁剪朝向档（保守：只删"必然不可能"的档）。
``keep=True``（direction / near）⟹ 朝向必须**可能**覆盖 station：
保留与方位夹角 ≤ 90° + 半档 + 查表舍入 + 格内角裕量的档。
``keep=False``（no_signal 且整格必在 r_eff_min 内）⟹ 朝向必须
**确定不覆盖** station：只保留与方位夹角 > 90° − 半档 − 查表舍入
− 格内角裕量的档。
``格内角裕量 = asin(margin / max(dist − margin, 1))``：格内任一点
相对格心的方位偏差上界。两个方向都必须计入它，否则会误删真源的
朝向档（keep=True 漏计）或保留必然可见的档（keep=False 漏计）。
"""
margin_m = self.position.margin_m
doomed: list[int] = []
for index in self.position.alive:
gx, gy = self.position.cell_center(index)
dx, dy = station[0] - gx, station[1] - gy
distance = math.hypot(dx, dy)
if within_m is not None and distance > within_m:
continue
bearing = math.degrees(math.atan2(dy, dx))
if distance <= margin_m + 1.0:
# 格贴测点：格内方位可指向任意方向，任何一侧都不可据此裁剪
continue
angular_margin = math.degrees(
math.asin(min(0.9999, margin_m / (distance - margin_m))))
if keep:
updated = self._mask(index) & _cover_mask(
bearing, 90.0 + _BIN_HALF_DEG + _GRID_DEG / 2 + angular_margin)
else:
updated = self._mask(index) & _far_mask(
bearing, 90.0 - _BIN_HALF_DEG - _GRID_DEG / 2 - angular_margin)
if updated == 0:
doomed.append(index)
self.masks.pop(index, None)
elif updated != _FULL_MASK:
self.masks[index] = updated
if doomed:
self.position.discard(doomed)
def apply(self, station: Vec2, result: str,
svd_deg: float | None = None) -> None:
"""按读数更新联合可行集（保守，真源 (位置, 朝向) 永不被删）。"""
if result == DIRECTION:
self.position.apply_measure(station, DIRECTION, svd_deg,
directional=True)
self._apply_heading(station, keep=True)
elif result == NEAR:
# near 同样要求可见（评测机先判朝向再判距离）
self.position.apply_measure(station, NEAR, directional=True)
self._apply_heading(station, keep=True)
elif result == NO_SIGNAL:
# 距离阈值必须取**保守侧**：只有"整格都必然落在 r_eff_min 圆盘内"
# 时，no_signal 才一定由朝向遮挡解释。格心到格内任一点 ≤ margin，
# 故阈值取 r_eff_min − margin；取 + margin 会把 1000~1010m 的模糊带
# 也当成"必须不覆盖"，从而可能删掉真源的朝向档（破坏真源保留性）。
self._apply_heading(station, keep=False,
within_m=max(
0.0, self.field.r_eff_min_m
- self.position.margin_m))
else:
raise ValueError(f"未知测向结果: {result!r}")
# ------------------------------------------------------------ 诊断
def summary(self) -> dict:
"""联合集规模快照（写入策略 summary，供审计与论文复算）。"""
cells = self.sample(256)
bins = [bin(self._mask(index)).count("1") for index in cells]
return {"cells": len(self.position),
"sampled_mean_heading_bins": (round(sum(bins) / len(bins), 2)
if bins else 0.0),
"heading_bins_total": HEADING_BINS}
In [ ]:
import math
Vec2 = tuple[float, float]
def path_length(order: list[Vec2], start: Vec2 = (0.0, 0.0)) -> float:
"""开放路径长度（起点 start，终点自由）。"""
total = 0.0
position = start
for point in order:
total += math.dist(position, point)
position = point
return total
def nearest_neighbor_order(points: list[Vec2],
start: Vec2 = (0.0, 0.0)) -> list[Vec2]:
"""最近邻开放路径（O(n²)，启发式上界）。"""
remaining = list(points)
order: list[Vec2] = []
position = start
while remaining:
nxt = min(remaining, key=lambda q: math.dist(position, q))
remaining.remove(nxt)
order.append(nxt)
position = nxt
return order
def two_opt_open(order: list[Vec2], start: Vec2 = (0.0, 0.0),
max_passes: int = 8) -> list[Vec2]:
"""开放路径 2-opt（终点自由：只反转内部片段，不做首尾相接）。
增量代价 O(1)：反转 [i..j] 只影响 (prev, i) 与 (j, next) 两条边。
"""
best = list(order)
n = len(best)
if n < 3:
return best
for _ in range(max_passes):
improved = False
for i in range(n - 1):
prev = best[i - 1] if i > 0 else start
pi = best[i]
d_prev_i = math.dist(prev, pi)
for j in range(i + 1, n):
pj = best[j]
nxt = best[j + 1] if j + 1 < n else None
d_prev_j = math.dist(prev, pj)
removed = d_prev_i
added = d_prev_j
if nxt is not None:
removed += math.dist(pj, nxt)
added += math.dist(pi, nxt)
if added < removed - 1e-9:
best[i:j + 1] = best[i:j + 1][::-1]
improved = True
pi = best[i]
d_prev_i = math.dist(prev, pi)
if not improved:
break
return best
def improve_open_path(points: list[Vec2], start: Vec2 = (0.0, 0.0),
max_passes: int = 8) -> list[Vec2]:
"""最近邻打底 + 2-opt 改进的开放路径（后备清除走格顺序用）。"""
order = nearest_neighbor_order(points, start)
return two_opt_open(order, start, max_passes)
In [ ]:
def bearing_deg(a: Vec2, b: Vec2) -> float:
"""从 a 指向 b 的方位角（0-360，0 为 x 正方向，逆时针）。"""
return math.degrees(math.atan2(b[1] - a[1], b[0] - a[0])) % 360.0
def _error_brief(exc: BaseException) -> dict:
"""异常摘要（summary 留档用）：类型名 + 截断消息。"""
return {"type": type(exc).__name__, "message": str(exc)[:500]}
@dataclass(frozen=True)
class SecondPointPlan:
"""下一检测点的候选与依据（P3 输出）。"""
position: Vec2
worst_posterior_radius_m: float
move_cost_s: float
reason: str
@dataclass(frozen=True)
class ClearCertificate:
"""清除证书：候选集外包半径（含近似误差）不超过 20m 阈值。"""
channel: int
center: Vec2
radius_m: float
limit_m: float
certified: bool
def clear_certificate(region: CandidateRegion, channel: int,
config: ProblemConfig = ProblemConfig()) -> ClearCertificate:
"""ρ_poly <= 20 - 外包近似误差 - 工程余量 才允许发起清除。"""
center, radius = region.mec()
limit = (config.field.clear_radius_m - region.slack_m
- config.search.clear_margin_m)
return ClearCertificate(channel, center, radius, limit, radius <= limit)
def candidate_points(center: Vec2, safe_radius_m: float, rings, angles: int) -> list[Vec2]:
"""保证接收盘内的候选点：盘心 + 各比例半径 × 均匀方向。"""
points = [tuple(center)]
for fraction in rings:
radius = fraction * safe_radius_m
if radius <= 1e-9:
continue
step = 2 * math.pi / angles
for k in range(angles):
angle = step * k + step / 2
points.append((center[0] + radius * math.cos(angle),
center[1] + radius * math.sin(angle)))
return points
def _worst_posterior_radius(region: CandidateRegion, q: Vec2,
witnesses: list[Vec2], half_angle_deg: float,
near_radius_m: float, disk_sides: int) -> float:
"""对候选点 q，遍历"目标边界点 × 读数分支"的最坏后验包围半径。
读数分支：direction 的区间端点 φ(g)±δ（端点是每个 g 的最不利读数），
以及 near——源落在 q 的 near_radius_m 内时后验为 B(q, near)∩候选集。
保证接收盘内 no_signal 不可能发生，故不枚举该分支。
这是启发式近似，不构成最优性证明 [工程]。
"""
posterior_radii = []
for g in witnesses:
phi = bearing_deg(q, g)
for reading in (phi - half_angle_deg, phi + half_angle_deg):
probe = CandidateRegion(list(region.polygon), region.slack_m)
probe.clip_wedge(q, reading, half_angle_deg)
if probe.polygon:
posterior_radii.append(probe.mec()[1])
near_probe = CandidateRegion(list(region.polygon), region.slack_m)
try:
near_probe.clip_disk(q, near_radius_m, disk_sides)
except InconsistentBelief:
pass # B(q, near) 与候选集不相交：near 分支不可能发生，跳过
else:
if near_probe.polygon:
posterior_radii.append(near_probe.mec()[1])
if not posterior_radii:
return region.mec()[1] # 无有效采样点时退回当前半径
return max(posterior_radii)
def safe_disk_radius(region: CandidateRegion,
config: ProblemConfig = ProblemConfig()) -> float:
"""保证接收盘 B(mec, safe) 的半径；<0 表示无保证（ρ 超过最小接收半径）。
盘内任意点到候选集每点距离 <= safe + ρ + 1e-3 < 1000 <= 最小接收半径，
故必然收到 direction/near。1mm 数值余量避免浮点贴边被 r_eff=1000 的源
判为 no_signal。
"""
return config.field.r_eff_min_m - region.mec()[1] - 1e-3
def select_measurement_point(region: CandidateRegion, from_position: Vec2,
config: ProblemConfig = ProblemConfig()) -> SecondPointPlan:
"""在保证接收盘内选择下一检测点（P3 核心；对任意候选集通用）。
保证接收盘 B(mec_center, 1000-ρ)：盘内任意点到候选集每点距离
<= (1000-ρ) + ρ = 1000 <= 最小接收半径，故必然收到 direction/near，
不会浪费一次 no_signal。ρ>1000 时退化为探索 MEC 圆心（可能 no_signal）。
"""
field, search = config.field, config.search
center, radius = region.mec()
half_angle = field.bearing_error_half_deg + search.bearing_rounding_margin_deg
def move_cost(point: Vec2) -> float:
return move_time(from_position, point, config.cost)
# 1mm 数值余量在 safe_disk_radius 内记账。
safe_radius = safe_disk_radius(region, config)
if safe_radius < 0:
return SecondPointPlan(tuple(center), radius, move_cost(center),
"无保证接收盘（ρ>1000）：探索 MEC 圆心，可能 no_signal")
coarse = region.witnesses(search.witness_spacing_m)
samples = candidate_points(center, safe_radius,
search.candidate_rings, search.candidate_angles)
near_radius, sides = field.near_radius_m, search.polygon_sides
best_key, best_point, best_worst = None, None, None
for point in samples:
worst = _worst_posterior_radius(region, point, coarse, half_angle,
near_radius, sides)
key = (round(worst, 6), round(move_cost(point), 6))
if best_key is None or key < best_key:
best_key, best_point, best_worst = key, point, worst
dense = _worst_posterior_radius(region, best_point,
region.witnesses(search.witness_spacing_m / 4),
half_angle, near_radius, sides)
reason = (f"安全盘 B(centre,{safe_radius:.0f}m) 内 {len(samples)} 候选；"
f"粗采样最坏后验 {best_worst:.1f}m，1/4 间距加密复核 {dense:.1f}m")
return SecondPointPlan(tuple(best_point), dense, move_cost(best_point), reason)
def select_second_point(client: SimulatorClient, first_position: Vec2,
first_bearing_deg: float,
config: ProblemConfig = ProblemConfig()) -> SecondPointPlan:
"""P3 入口：由首次 direction 构造 K1，返回保证接收的第二检测点方案。"""
region = CandidateRegion.initial(first_position, config)
half_angle = config.field.bearing_error_half_deg \
+ config.search.bearing_rounding_margin_deg
region.clip_wedge(first_position, first_bearing_deg, half_angle)
return select_measurement_point(region, client.state.position, config)
In [ ]:
class OmnidirectionalLoop:
"""问题3全向"发现—定位—清除—持续复扫"闭环（P4 基线 + 持续搜索）。
串行驱动 SimulatorClient：覆盖阶段在 49 点网格上对每个 UNKNOWN 频道
检测；发现 direction 即进入该频道的定位子循环（选点→测量→收敛→
证书清除），near 直接清除。全部频道"已清除或证明不存在"后按
``search.persistent_search``（默认关闭）可选择只复扫
未清除频道，直到清除数达 16、实机预算耗尽或虚拟安全余量不足。
stop_reason 记录循环结束原因；``completed`` 表示任务证明完成
（结束时每频道均 CLEARED/ABSENT 且无中断），与"复扫到限时才停"
相互独立：accounted_virtual_time_s 记录首次证明找全的虚拟时刻。
"""
def __init__(self, client: SimulatorClient,
config: ProblemConfig = ProblemConfig()):
self.client = client
self.config = config
self.belief = BeliefState(config)
# 覆盖计划是"可离线搜索、可在运行期校验"的一等对象；基类默认保持
# 历史 49 点表（grid49_immediate 基线口径不变），各问题策略自行覆盖。
self.coverage_plan = square_49_plan()
self.ledger = CoverageLedger(config, points=list(self.coverage_plan.points))
self.cleared: list[int] = []
self.anomalies: list[dict] = []
self.stop_reason: str | None = None
# 逐频道缺席证据来源（pf_empty / coverage），供审计与论文逐条追溯。
self.absence_evidence: dict[int, str] = {}
# "同一 0.5m 测点 × 同一频道"的重复检测计数：历史上一次官方演练
# 有 98.84% 的虚拟时间耗在空复扫上，本计数是该问题的回归指标。
self.redundant_rescan_measures = 0
# 首次达成"每频道均已清除或证明不存在"的虚拟时刻（找全证明）。
self.accounted_virtual_time_s: float | None = None
# 证书有效但 /clear 失败后必须先补测再重试，避免在同一证书上
# 反复消耗 3 秒/次的失败清除。
self._may_clear: set[int] = set()
# 断连/协议错误/守恒性破坏等中断：留档异常，收尾时写入 summary。
self._abort_detail: BaseException | None = None
# ------------------------------------------------------------ 主流程
def run(self) -> dict:
"""主闭环。任何异常都必须尽力 /exit 并产出 summary，不得静默丢会话。"""
try:
self.client.enter()
self._coverage_phase()
if self.stop_reason is None:
self._mark_absent_after_coverage()
self._resolve_detected()
self._record_accounted()
if self.config.search.persistent_search:
self._persistent_phase()
if self.stop_reason is None:
if len(self.cleared) >= self.config.field.source_count_max:
self.stop_reason = "sixteen_cleared"
elif self._all_accounted():
self.stop_reason = "all_resolved"
else:
self.stop_reason = "unresolved_channels"
except DeadlineExceeded:
self.stop_reason = "real_budget_exhausted"
except (TransportError, ProtocolError, InconsistentBelief) as exc:
self._abort_detail = exc
if isinstance(exc, TransportError):
self.stop_reason = "transport_failure"
elif isinstance(exc, InconsistentBelief):
self.stop_reason = "inconsistent_belief"
else:
self.stop_reason = "protocol_error"
except Exception as exc: # 意外异常同样收尾，不阻断 /exit 与结果汇总
self._abort_detail = exc
self.stop_reason = "unexpected_error"
info = None
exit_error = None
try:
info = self.client.exit()
except Exception as exc: # 接口不可用：记录后仍产出未完成状态
exit_error = exc
summary = self._summary(info)
if exit_error is not None:
summary["exit_error"] = _error_brief(exit_error)
return summary
# ------------------------------------------------------------ 预算与证明
def _min_virtual_reserve(self) -> float:
"""单动作耗时上界的保守估计，用于虚拟安全余量。
上界 = 2 × 最远站点距离 / 速度 + 检测 + 换频（机器人不会跑到覆盖
计划包围区之外执行任务）。配置值通常更宽松；本函数保证即使配置被
调小，余量也足以完成最后一个动作而不触及 100 小时上限。
"""
reach = max((math.dist((0.0, 0.0), point) for point in self.ledger.points),
default=0.0)
action_bound = (2.0 * reach / self.config.cost.speed_mps
+ self.config.cost.measure_s + self.config.cost.switch_s)
return max(self.config.search.virtual_reserve_s, action_bound)
def _virtual_cap_reached(self) -> bool:
"""虚拟安全余量不足：置 stop_reason 并返回 True，调用方立即收尾。
余量须大于单次动作耗时上界（见 :meth:`_min_virtual_reserve`），
`search.virtual_reserve_s=10000s` [工程] 提供充分裕度，避免触及
附件2 §4.5 的 100 小时上限被模拟器强制结束。
"""
if (self.client.state.virtual_time_s + self._min_virtual_reserve()
< self.config.field.max_virtual_duration_s):
return False
self.stop_reason = "virtual_time_cap"
return True
def _all_accounted(self) -> bool:
return all(belief.status in (ChannelStatus.CLEARED, ChannelStatus.ABSENT)
for belief in self.belief.channels.values())
def _record_accounted(self) -> None:
"""首次"每频道均已清除或证明不存在"的时刻留档（找全证明）。"""
if self.accounted_virtual_time_s is None and self._all_accounted():
self.accounted_virtual_time_s = self.client.state.virtual_time_s
# ------------------------------------------------------------ 覆盖阶段
def _pending_channels_at(self, index: int) -> list[int]:
return [channel for channel, belief in self.belief.channels.items()
if belief.status is ChannelStatus.UNKNOWN
and not self.ledger.measured_at(channel, index)]
def _next_point_index(self) -> int | None:
"""尚有 UNKNOWN 频道待测的最近网格点（贪心最近邻 [工程]）。"""
position = self.client.state.position
best_index, best_distance = None, 0.0
for index, point in enumerate(self.ledger.points):
if not self._pending_channels_at(index):
continue
distance = math.dist(position, point)
if best_index is None or distance < best_distance:
best_index, best_distance = index, distance
return best_index
def _coverage_phase(self) -> None:
while True:
if self._count_limit_reached():
return
index = self._next_point_index()
if index is None:
return
point = self.ledger.points[index]
for channel in self._pending_channels_at(index):
if self._virtual_cap_reached():
return
outcome = self.client.measure(point, channel)
self.ledger.mark_measured(channel, index)
self.belief.update_measure(channel, point, outcome.result,
outcome.svd_deg)
if outcome.result == DIRECTION:
self._localize_and_clear(channel)
elif outcome.result == NEAR:
region = self.belief.candidate_set(channel)
cert = clear_certificate(region, channel, self.config)
if cert.certified:
self._attempt_clear(channel, cert.center, cert)
else: # 理论上不发生（near 候选半径 5m）；保守记录
self._anomaly("near_without_certificate", channel,
position=point)
if self.stop_reason is not None or self._count_limit_reached():
return
# ------------------------------------------------------------ 持续复扫
def _persistent_phase(self) -> None:
"""全部频道已清除/证明不存在后的持续复扫 [用户决策 2026-09-11]。
只扫未清除频道（DETECTED 继续定位，ABSENT 全网格复测，防模拟器
或覆盖证明假设偏差），直到清除数达 16 或虚拟安全余量不足；
实机预算由客户端 DeadlineExceeded 兜底（扫到限时为止）。
"""
while self.stop_reason is None:
if self._count_limit_reached():
return
if self._virtual_cap_reached():
return
self._persistent_round()
def _persistent_round(self) -> None:
"""一轮全网格复扫：网格点最近邻序，逐点测量全部未清除频道。"""
measured: set[tuple[int, int]] = set()
while self.stop_reason is None:
index = self._next_resweep_point(measured)
if index is None:
return
point = self.ledger.points[index]
for channel in self._resweep_channels_at(index, measured):
if self._virtual_cap_reached():
return
measured.add((channel, index))
was_absent = (self.belief.channels[channel].status
is ChannelStatus.ABSENT)
outcome = self.client.measure(point, channel)
self.belief.update_measure(channel, point, outcome.result,
outcome.svd_deg)
if was_absent and outcome.result in (DIRECTION, NEAR):
# 复测推翻了覆盖缺席证明：守恒性留痕，按新证据定位清除。
self._anomaly("absent_detected_on_resweep", channel,
position=point, result=outcome.result)
if outcome.result == DIRECTION:
self._localize_and_clear(channel)
elif outcome.result == NEAR:
region = self.belief.candidate_set(channel)
cert = clear_certificate(region, channel, self.config)
if cert.certified:
self._attempt_clear(channel, cert.center, cert)
else:
self._anomaly("near_without_certificate", channel,
position=point)
if self.stop_reason is not None:
return
def _next_resweep_point(self, measured: set[tuple[int, int]]) -> int | None:
"""尚有未清除频道待复测的最近网格点（贪心最近邻 [工程]）。"""
position = self.client.state.position
best_index, best_distance = None, 0.0
for index, point in enumerate(self.ledger.points):
if not self._resweep_channels_at(index, measured):
continue
distance = math.dist(position, point)
if best_index is None or distance < best_distance:
best_index, best_distance = index, distance
return best_index
def _resweep_channels_at(self, index: int,
measured: set[tuple[int, int]]) -> list[int]:
return [channel for channel, belief in self.belief.channels.items()
if belief.status is not ChannelStatus.CLEARED
and (channel, index) not in measured]
def _count_limit_reached(self) -> bool:
if len(self.cleared) >= self.config.field.source_count_max:
self.stop_reason = "sixteen_cleared"
return True
return False
# ------------------------------------------------------------ 定位与清除
def _localize_and_clear(self, channel: int) -> bool:
"""定位子循环：选点测量收敛候选集，拿到清除证书后清除。
返回是否已清除；步数耗尽仍未收敛则保持 DETECTED 留待二轮。
进入本函数前该频道刚有一次 direction 读数，允许立即清除。
"""
search = self.config.search
self._may_clear.add(channel)
for _ in range(search.max_localize_steps):
if self._virtual_cap_reached():
return False
region = self.belief.candidate_set(channel)
if region is None:
return False
cert = clear_certificate(region, channel, self.config)
if cert.certified and channel in self._may_clear:
if self._attempt_clear(channel, cert.center, cert):
return True
self._may_clear.discard(channel)
continue
plan = select_measurement_point(region, self.client.state.position,
self.config)
outcome = self.client.measure(plan.position, channel)
self.belief.update_measure(channel, plan.position, outcome.result,
outcome.svd_deg)
self._may_clear.add(channel)
if outcome.result == NO_SIGNAL:
self._anomaly("no_signal_in_safe_disk", channel,
position=plan.position, detail=plan.reason)
return self.belief.channels[channel].status is ChannelStatus.CLEARED
def _attempt_clear(self, channel: int, target: Vec2,
cert: ClearCertificate) -> bool:
outcome = self.client.clear(target, channel)
if outcome.result == CLEAR_SUCCESS:
self.belief.mark_cleared(channel)
self.cleared.append(channel)
return True
if cert.certified:
self._anomaly("clear_failed_with_certificate", channel, position=target,
radius_m=round(cert.radius_m, 3), limit_m=cert.limit_m)
return False
def _anomaly(self, kind: str, channel: int, **detail) -> None:
"""异常只在守恒性被打破或证书失效时出现；每频道限量记录防刷屏。"""
recorded = sum(1 for item in self.anomalies if item["channel"] == channel)
if recorded >= self.config.search.max_anomalies_per_channel:
return
self.anomalies.append({"kind": kind, "channel": channel, **detail})
# ------------------------------------------------------------ 收尾
def _mark_absent_after_coverage(self) -> None:
"""全向源缺席证明：全网格无非 no_signal 读数的频道不存在源 [题面]。"""
for channel, belief in self.belief.channels.items():
if belief.status is ChannelStatus.UNKNOWN \
and self.ledger.coverage_complete(channel):
self.belief.mark_absent(channel)
self.absence_evidence[channel] = "coverage"
def _resolve_detected(self) -> None:
"""覆盖完成后，对仍未清除的 DETECTED 频道做第二轮定位。
不再在此决定停止：持续搜索模式下由 _persistent_phase 接力，
非持续模式由 run() 依据 _all_accounted 判定 all_resolved。
"""
for channel in self.belief.channels_by_status(ChannelStatus.DETECTED):
if self._count_limit_reached() or self._virtual_cap_reached():
return
self._localize_and_clear(channel)
if self.stop_reason is not None:
return
def _summary(self, info) -> dict:
"""结果汇总。info 为 None 表示 /exit 未能完成（断连等），按未完成上报。
completed = 无中断且结束时每频道均 CLEARED/ABSENT（任务证明完成）；
stop_reason 另行记录循环结束原因（持续复扫模式下通常是
virtual_time_cap / real_budget_exhausted / sixteen_cleared）。
"""
events = self.client.log.action_events()
stats = summarize(events)
report = recompute_virtual_time(events)
steps = self._step_report()
cleared = self.belief.channels_by_status(ChannelStatus.CLEARED)
absent = self.belief.channels_by_status(ChannelStatus.ABSENT)
resolved = set(cleared) | set(absent)
unresolved = [channel for channel in sorted(self.belief.channels)
if channel not in resolved]
virtual_time_s = round((info.virtual_time_s if info is not None
else self.client.state.virtual_time_s), 3)
# 清除数达 16 即数量上界证明（其余频道无论状态如何都不可能有源）；
# 否则要求每频道均 CLEARED/ABSENT。
completed = (self._abort_detail is None
and (len(self.cleared) >= self.config.field.source_count_max
or not unresolved))
# 完成依据必须显式区分"数量上界"与"逐频道缺席/清除证明"，
# 否则论文与审计会把 16 源截断误读为逐频道证明。
if self._abort_detail is not None:
completion_basis = "incomplete"
elif len(self.cleared) >= self.config.field.source_count_max:
completion_basis = "count_cap"
elif not unresolved:
completion_basis = "per_channel_proof"
else:
completion_basis = "incomplete"
summary = {
"completed": completed,
"completion_basis": completion_basis,
"stop_reason": self.stop_reason,
"exit_reason": "exit_unreachable" if info is None else info.reason,
"virtual_time_s": virtual_time_s,
"accounted_virtual_time_s": (round(self.accounted_virtual_time_s, 3)
if self.accounted_virtual_time_s is not None
else None),
"persistent_search": self.config.search.persistent_search,
"virtual_reserve_s": round(self._min_virtual_reserve(), 3),
"coverage_plan": self.coverage_plan.name,
"coverage_points": len(self.coverage_plan.points),
"coverage_plan_points": [[p[0], p[1]] for p in self.coverage_plan.points],
"coverage_plan_certificate": (
self.coverage_plan.certificate.as_dict()
if self.coverage_plan.certificate is not None else None),
"cleared_channels": cleared,
"absent_channels": absent,
"unresolved_channels": unresolved,
"absence_evidence": {str(channel): self.absence_evidence[channel]
for channel in sorted(absent)},
"redundant_rescan_measures": self.redundant_rescan_measures,
"measures": stats["measures"],
"clears": stats["clears"],
"clear_success": stats["cleared"],
"total_requests": stats["measures"] + stats["clears"],
"move_distance_m": round(stats["distance_m"], 1),
# 题面口径：定位清除总时间（全程虚拟时长）/ 被清除干扰源个数。
"avg_locate_clear_time_s": (round(virtual_time_s / len(cleared), 3)
if cleared else None),
"step_totals": steps["totals"],
"steps": steps["rows"],
"anomalies": self.anomalies,
"virtual_recompute_mismatches": len(report["mismatches"]),
"channel_status": self._channel_status_snapshot(),
}
if self._abort_detail is not None:
summary["abort_detail"] = _error_brief(self._abort_detail)
return summary
def _step_report(self) -> dict:
"""每步计时汇总 [用户决策 2026-09-11]：实机墙钟 + 比赛虚拟耗时分项。"""
totals = {"count": 0, "real_elapsed_total_s": 0.0,
"real_since_enter_total_s": 0.0,
"virtual_move_s": 0.0, "virtual_switch_s": 0.0,
"virtual_measure_action_s": 0.0, "virtual_clear_action_s": 0.0,
"virtual_total_s": 0.0}
rows = []
for index, step in enumerate(self.client.log.step_events()):
cost = step["virtual_cost"]
totals["count"] += 1
totals["real_elapsed_total_s"] += step["real_elapsed_s"]
totals["real_since_enter_total_s"] = step.get(
"real_since_enter_s", totals["real_since_enter_total_s"])
totals["virtual_move_s"] += cost["move_s"]
totals["virtual_switch_s"] += cost["switch_s"]
if step["path"] == "/measure":
totals["virtual_measure_action_s"] += cost["action_s"]
elif step["path"] == "/clear":
totals["virtual_clear_action_s"] += cost["action_s"]
totals["virtual_total_s"] += cost["total_s"]
rows.append({"index": index, "action": step["path"].lstrip("/"),
"result": step.get("result"),
"channel": step.get("channel"),
"position": step.get("position"),
"real_elapsed_ms": round(step["real_elapsed_s"] * 1000.0, 3),
"real_since_enter_s": step.get("real_since_enter_s"),
"move_s": round(cost["move_s"], 3),
"switch_s": round(cost["switch_s"], 3),
"action_s": round(cost["action_s"], 3),
"step_virtual_s": round(cost["total_s"], 3),
"virtual_time_s": step["virtual_time_s"]})
for key, value in totals.items():
if isinstance(value, float):
totals[key] = round(value, 3)
return {"totals": totals, "rows": rows}
def _channel_status_snapshot(self) -> dict:
"""逐频道状态快照：中断收尾时的未完成状态留档（每频道一条）。"""
snapshot = {}
for channel, belief in sorted(self.belief.channels.items()):
entry = {"status": belief.status.value}
if belief.region is not None:
entry["posterior_radius_m"] = round(belief.region.mec()[1], 3)
snapshot[str(channel)] = entry
return snapshot
In [ ]:
class DirectionalExtension(OmnidirectionalLoop):
"""问题4混合源闭环（P5）：全向基线 + 定向观测处理 + 后备清除。
与 P4 的三点差异（协议衔接见 docs/协议规则表与状态机.md §7）：
- 49 点覆盖的发现保证对定向源同样成立，覆盖阶段不区分源类型；
- 保证接收盘内的 no_signal 只能由朝向遮挡解释 [观测推理]：该频道
记为"观测到定向"，随后对 MEC 圆心取对跖探点复核；联合约束基线
不据此删除空间候选（真源不误删）；
- 定位停滞（候选集有界但未达清除证书）时构造 <=20m 有限圆盘覆盖，
预算核对后逐格 /clear：清除与朝向无关 [题面]，是定向源不依赖
可见性的完备后备。格数超硬顶或现实预算不足则跳过并留痕，绝不盲扫。
"""
def __init__(self, client: SimulatorClient,
config: ProblemConfig = ProblemConfig()):
super().__init__(client, config)
# P0-2：第四问切换到同一保证下的最小命中集（默认 q4_mincover_v1）。
self.coverage_plan = plan_instance(config.search.q4_coverage_plan,
verify=config.search.verify_coverage_plan)
self.ledger = CoverageLedger(config, points=list(self.coverage_plan.points))
self.directional_observed: list[int] = [] # 观测到定向性的频道
self.blocked_probes: dict[int, list[Vec2]] = {}
self.fallback_log: list[dict] = [] # 后备清除逐次留痕
# P0-3：(位置 × 朝向) 联合可行集，仅在频道首次被看见时惰性建立，
# 使其初始化与更新成本只落在"确实存在定向源"的频道上。
self.joint: dict[int, JointBelief] = {}
self.joint_absent_channels: list[int] = []
self.visibility_probes = 0 # 由联合信念选出的探点数
def _coverage_phase(self) -> None:
"""49站的定向覆盖证明不变；每站先串行扫频，再就近服务已发现源。
避免检测一个源后远行清除，再返回同站检测下一个频道。
no_signal不用于删空间候选，也不使用第三问的七站缺席证明。
"""
while not self._count_limit_reached() and self.stop_reason is None:
index = self._next_point_index()
if index is None:
return
point = self.ledger.points[index]
pending = self._pending_channels_at(index)
pending.sort(key=lambda c: (c != self.client.state.channel, c))
for channel in pending:
if self._virtual_cap_reached():
return
outcome = self.client.measure(point, channel)
self.ledger.mark_measured(channel, index)
self._observe(channel, point, outcome.result, outcome.svd_deg)
if outcome.result == NEAR:
cert = clear_certificate(
self.belief.candidate_set(channel), channel, self.config)
if cert.certified:
self._attempt_clear(channel, point, cert)
if self._count_limit_reached():
return
queue = self.belief.channels_by_status(ChannelStatus.DETECTED)
while queue and self.stop_reason is None:
channel = min(queue, key=lambda c: (
math.dist(self.client.state.position,
self.belief.candidate_set(c).mec()[0]), c))
queue.remove(channel)
self._localize_and_clear(channel)
if self._count_limit_reached():
return
# ------------------------------------------------------------ 定位与清除
def _localize_and_clear(self, channel: int) -> bool:
"""定位子循环：选点测量收敛候选集，拿到清除证书后清除。
与 P4 的差异：保证接收盘内的 no_signal 说明探点被朝向遮挡
[观测推理]，走"对跖复核 -> 环探"升级梯子（不计入定位步数，
由 max_blocked_probes 限流），之后仍未收敛再转后备清除。
"""
search = self.config.search
self._may_clear.add(channel)
steps = 0
while steps < search.max_localize_steps:
if self._virtual_cap_reached():
return False
region = self.belief.candidate_set(channel)
if region is None:
return False
cert = clear_certificate(region, channel, self.config)
if cert.certified and channel in self._may_clear:
if self._attempt_clear(channel, cert.center, cert):
return True
self._may_clear.discard(channel) # 证书失败：补测后才可重试
continue
if len(self.blocked_probes.get(channel, [])) >= search.max_blocked_probes:
break # 梯子用尽仍被遮挡：定位停滞，转后备清除
point = self._next_probe(region, channel)
within_range = (math.dist(point, region.mec()[0])
<= safe_disk_radius(region, self.config))
outcome = self.client.measure(point, channel)
self._observe(channel, point, outcome.result, outcome.svd_deg)
self._may_clear.add(channel)
if outcome.result == NO_SIGNAL:
if within_range and channel not in self.directional_observed:
self.directional_observed.append(channel)
self.blocked_probes.setdefault(channel, []).append(point)
continue # 盘外无信号也可能是距离不足，不据此断言定向
self.blocked_probes.pop(channel, None)
steps += 1
if self.belief.channels[channel].status is ChannelStatus.CLEARED:
return True
region = self.belief.candidate_set(channel)
if region is None:
return False
cert = clear_certificate(region, channel, self.config)
if cert.certified and channel in self._may_clear:
if self._attempt_clear(channel, cert.center, cert):
return True
self._may_clear.discard(channel)
return self._fallback_clear(channel, region)
def _observe(self, channel: int, point: Vec2, result: str,
svd_deg: float | None) -> None:
"""同步更新凸候选集与 (位置 × 朝向) 联合可行集。
联合集只在频道首次被看见（direction/near）时建立：从未被看见的
频道由覆盖证明负责缺席，不需要朝向维度。联合集一旦为空，直接给出
该频道的缺席证明（严格，与覆盖证明同一条件），可提前停止对其检测。
"""
self.belief.update_measure(channel, point, result, svd_deg)
if not self.config.search.directional_joint_belief:
return
joint = self.joint.get(channel)
if joint is None:
if result not in (DIRECTION, NEAR):
return
joint = JointBelief(self.config)
self.joint[channel] = joint
joint.apply(point, result, svd_deg)
if joint.is_absent() and self.belief.channels[channel].status \
is not ChannelStatus.CLEARED:
self.belief.mark_absent(channel)
self.absence_evidence[channel] = "joint_empty"
if channel not in self.joint_absent_channels:
self.joint_absent_channels.append(channel)
def _probe_candidates(self, region: CandidateRegion, channel: int) -> list[Vec2]:
"""可见性选点的候选点族：保证接收点 + 对跖点 + 环探 + **抵近点**。
D1 关键结构性改动：旧点族全部围绕 MEC 圆心固定环采样，远距离不收敛时
没有"主动靠近"这一动作，只能退化到 20m 网格盲清。现在补上
``center + α·(当前位置 − center)``（α∈approach_fractions，α=0 即圆心）
的抵近序列，使策略可以在保持可见的前提下逐次逼近真源，最终以正常
角域证书或 near 收尾。
"""
center, radius = region.mec()
position = self.client.state.position
candidates = [select_measurement_point(
region, position, self.config).position]
safe = safe_disk_radius(region, self.config)
blocked = self.blocked_probes.get(channel, [])
if blocked and safe >= 0:
dx, dy = center[0] - blocked[0][0], center[1] - blocked[0][1]
norm = math.hypot(dx, dy)
if norm > 1e-9:
scale = min(1.0, safe / norm)
candidates.append((center[0] + dx * scale, center[1] + dy * scale))
# 抵近点：从 MEC 圆心到当前点的插值序列（含圆心本身）。
for fraction in self.config.search.approach_fractions:
candidates.append((center[0] + (position[0] - center[0]) * fraction,
center[1] + (position[1] - center[1]) * fraction))
for factor in (0.5, 1.0, 1.5):
ring = max(radius * factor, 2.0 * self.config.field.near_radius_m)
for step in range(8):
angle = math.pi * step / 4.0
candidates.append((center[0] + ring * math.cos(angle),
center[1] + ring * math.sin(angle)))
unique: list[Vec2] = []
seen: set[tuple[int, int]] = set()
blocked_keys = {(int(round(point[0] * 2)), int(round(point[1] * 2)))
for point in blocked} \
if self.config.search.probe_exclude_blocked else set()
for point in candidates:
key = (int(round(point[0] * 2)), int(round(point[1] * 2)))
if key in seen or key in blocked_keys:
continue # 已测过且被遮挡的点：再测不会有任何新信息
seen.add(key)
unique.append(point)
return unique
def _visibility_probe(self, region: CandidateRegion, channel: int) -> Vec2 | None:
"""按联合信念选探点；联合信念不可用时返回 None（退回梯子）。
三种口径（只影响选点顺序，不承担任何安全责任）：
- ``nearest_visible``（默认）：先按可见概率过滤（只保留
``prob_detect ≥ ratio × max`` 的候选，避免选中"实际必然被朝向遮挡"的
点），再取 ``当前位置→q→MEC 圆心`` 移动最省者。9 场景 core 实测：
后备盲清 125.7 → 4.3 次/局、T 9225 → 8602 s、墙钟最低。
- ``expected_radius``：在同样过滤后最小化
``E[后验 MEC 半径](q) + λ·移动``。λ=5 时与上面几乎同分（8604 s），
λ 很小时（0.05）值函数主导，为信息多绕路（9740 s）——本题
"看得见且少走路"比"单步信息量最大"更重要。
- ``visibility``（旧口径）：最大化 ``prob_detect/(移动+6)``，会反复选中
同一个近处被遮挡点（实测 9225 s、后备 125.6 次）。
"""
joint = self.joint.get(channel)
if joint is None or joint.is_absent():
return None
position = self.client.state.position
service_s = self.config.cost.measure_s + self.config.cost.switch_s
candidates = self._probe_candidates(region, channel)
if not candidates:
return None
mode = self.config.search.probe_score_mode
if mode == "visibility":
best_point, best_score = None, None
for point in candidates:
probability = joint.prob_detect(point)
if probability <= 0.0:
continue
score = -(probability / (move_time(position, point, self.config.cost)
+ service_s))
if best_score is None or score < best_score:
best_point, best_score = point, score
if best_point is not None:
self.visibility_probes += 1
return best_point
scored = [(joint.prob_detect(point), point) for point in candidates]
peak = max(probability for probability, _ in scored)
ratio = self.config.search.probe_visibility_ratio
keep = [point for probability, point in scored
if probability >= ratio * peak] or [point for _, point in scored]
center = region.mec()[0]
weight = self.config.search.probe_lambda_move_s
def approach_cost(point: Vec2) -> float:
return (move_time(position, point, self.config.cost)
+ move_time(point, center, self.config.cost) + service_s)
if mode == "nearest_visible":
# λ 很大时值函数项对 argmin 已无影响：直接取"可见候选中移动最省"，
# 省掉每个候选一次期望后验半径计算（单局墙钟 −3 s）。
best = min(keep, key=lambda point: (approach_cost(point), point))
self.visibility_probes += 1
return best
best = min(keep, key=lambda point: (
joint.expected_posterior_radius(point) + weight * approach_cost(point),
point))
self.visibility_probes += 1
return best
def _next_probe(self, region: CandidateRegion, channel: int) -> Vec2:
"""定向遮挡下的探点升级梯子 [工程]。
- 未被遮挡过：worst-posterior 选点器（保证接收盘内）；
- 第 1 次被遮挡：对 MEC 圆心取对跖方向，偏移裁进保证接收盘；
- 之后：绕 MEC 圆心、半径 = ρ 的环探，方位自首个被遮挡点起
每次 +45°。几何边界必须如实陈述：环上可见弧没有下界——
源可位于环上（|g-c| = ρ）且朝向沿外切线，可见集退化为单个
切点，固定步进环探可以全部漏过。梯子只提高提前命中概率，
不构成接收或命中保证；完备性完全由后备清除兜底。
"""
visibility = self._visibility_probe(region, channel)
if visibility is not None:
return visibility
blocked = self.blocked_probes.get(channel, [])
center, radius = region.mec()
if not blocked:
return select_measurement_point(region, self.client.state.position,
self.config).position
safe = safe_disk_radius(region, self.config)
first = blocked[0]
if len(blocked) == 1 and safe >= 0:
dx, dy = center[0] - first[0], center[1] - first[1]
norm = math.hypot(dx, dy)
if norm <= 1e-9:
# 首个被遮挡探点恰在圆心：方向信息为零，沿 +x 走满安全盘。
return (center[0] + safe, center[1])
scale = min(1.0, safe / norm)
# 对跖方向，偏移模长 = min(norm, safe)：圆心对称点裁进保证接收盘。
return (center[0] + dx * scale, center[1] + dy * scale)
base = math.atan2(first[1] - center[1], first[0] - center[0])
azimuth = base + math.radians(45.0 * (len(blocked) - 1))
return (center[0] + radius * math.cos(azimuth),
center[1] + radius * math.sin(azimuth))
def _fallback_clear(self, channel: int, region: CandidateRegion) -> bool:
"""定位停滞的后备：候选集有限圆盘覆盖逐格清除（格数即请求上界）。"""
search, session = self.config.search, self.config.session
width = max(p[0] for p in region.polygon) - min(p[0] for p in region.polygon)
height = max(p[1] for p in region.polygon) - min(p[1] for p in region.polygon)
cells = fallback_clear_cells(region.polygon, self.config)
entry = {"channel": channel, "cells": len(cells),
"bound": fallback_cell_bound(width, height, self.config),
"posterior_radius_m": round(region.mec()[1], 3)}
self.fallback_log.append(entry)
if len(cells) > search.fallback_max_cells:
# 宽候选集盲扫违背预算纪律：留待定位收敛，不得硬扫 [计划§4]。
entry["outcome"] = "skipped_cell_cap"
self._anomaly("fallback_cell_cap_exceeded", channel,
cells=len(cells), cap=search.fallback_max_cells)
return False
# 现实预算核对：N 次请求 × 单请求高分位估计 + 退出余量，不足不启动。
remaining = self.client.real_remaining_s
if remaining is None or len(cells) * session.real_rtt_p95_s \
+ session.reserve_exit_s > remaining:
entry["outcome"] = "skipped_budget"
return False
attempts = 0
order = self._fallback_order(cells)
for cell in order:
if self._virtual_cap_reached():
entry["outcome"] = "skipped_virtual_budget"
entry["attempts"] = attempts
return False
attempts += 1
outcome = self.client.clear(cell, channel)
if outcome.result == CLEAR_SUCCESS:
self.belief.mark_cleared(channel)
self.cleared.append(channel)
entry["outcome"] = "cleared"
entry["attempts"] = attempts
return True
# 候选集保守包含真源，格心距格内任一点 <= 10*sqrt(2) < 20m：全部
# 失败说明守恒性被打破，按异常留痕，不得据此删除候选。
entry["outcome"] = "exhausted"
entry["attempts"] = attempts
self._anomaly("fallback_exhausted_without_success", channel,
cells=len(cells))
return False
def _fallback_order(self, cells: list[Vec2]) -> list[Vec2]:
"""后备清除的走格顺序：最近邻打底 + 2-opt（可选 Or-opt）改进。
旧实现每步重新 ``min(remaining, key=dist)``，等价于最近邻开放路径
（典型为最优的 1.2–1.3 倍）且每次重扫 O(k²)。这里一次性求解改进序
（D3），把格间移动压到接近最优；格数不变，故完备性不变。
"""
if not cells:
return []
if not self.config.search.fallback_optimize_tour or len(cells) < 4:
position = self.client.state.position
remaining = list(cells)
order: list[Vec2] = []
while remaining:
nxt = min(remaining, key=lambda c: (math.dist(position, c), c))
remaining.remove(nxt)
order.append(nxt)
position = nxt
return order
return improve_open_path([tuple(c) for c in cells],
self.client.state.position)
# ------------------------------------------------------------ 收尾
def _summary(self, info) -> dict:
report = super()._summary(info)
report["directional_observed_channels"] = list(self.directional_observed)
report["fallback_clears"] = self.fallback_log
# 后备 /clear 请求计入总请求口径，仅统计检测次数会低估现实耗时。
report["fallback_clear_attempts"] = sum(
entry.get("attempts", 0) for entry in self.fallback_log)
report["visibility_probes"] = self.visibility_probes
report["joint_absent_channels"] = list(self.joint_absent_channels)
report["joint_final"] = {str(channel): belief.summary()
for channel, belief in sorted(self.joint.items())}
return report
def fallback_cell_bound(width_m: float, height_m: float,
config: ProblemConfig = ProblemConfig()) -> int:
"""外包框枚举的格数上界 (⌈w/cell⌉+2)(⌈h/cell⌉+2)（计划§4 规模检查）。"""
cell = config.search.fallback_clear_grid_m
return (math.ceil(width_m / cell) + 2) * (math.ceil(height_m / cell) + 2)
def fallback_clear_cells(polygon: list[Vec2],
config: ProblemConfig = ProblemConfig()) -> list[Vec2]:
"""候选外包多边形的 <=20m 有限圆盘覆盖格心列表（P5 后备清除）。
全局半开 20m 网格索引；多边形与格子的相交判定带正容差（宁可多测），
多边形边界恰在格线上时两侧格子都保留，因此格内任意点距某格心
<= 10*sqrt(2) < 20m，逐格 /clear 至少一次命中。列表长度即 /clear
请求预算上界；先核对 fallback_cell_bound，超预算不得逐格盲扫。
"""
if not polygon:
return []
cell = config.search.fallback_clear_grid_m
xs = [p[0] for p in polygon]
ys = [p[1] for p in polygon]
i0, i1 = math.floor(min(xs) / cell), math.floor(max(xs) / cell)
j0, j1 = math.floor(min(ys) / cell), math.floor(max(ys) / cell)
eps = 1e-9
centers: list[Vec2] = []
for i in range(i0, i1 + 1):
for j in range(j0, j1 + 1):
clipped = list(polygon)
for a, b, c in ((1.0, 0.0, (i + 1) * cell + eps),
(-1.0, 0.0, -(i * cell - eps)),
(0.0, 1.0, (j + 1) * cell + eps),
(0.0, -1.0, -(j * cell - eps))):
clipped = clip_halfplane(clipped, a, b, c, tol=eps)
if not clipped:
break
if clipped:
centers.append(((i + 0.5) * cell, (j + 0.5) * cell))
return centers
In [ ]:
import math
from dataclasses import dataclass
@dataclass(frozen=True)
class Task:
"""统一任务表的一项。service_s 为不含移动的固有耗时（检测+换频/清除）。"""
kind: str # "station" | "locate" | "clear"
position: Vec2
service_s: float
channel: int | None = None
station_index: int | None = None
station_channels: tuple[int, ...] = ()
def _point_key(point: Vec2) -> tuple[int, int]:
"""候选点去重键：0.5m 网格取整（同一物理点不重复测量）。"""
return (int(round(point[0] * 2)), int(round(point[1] * 2)))
class BayesLoop(OmnidirectionalLoop):
"""问题3贝叶斯闭环：负观测感知 + 信息增益选点 + 滚动任务排序。"""
def __init__(self, client: SimulatorClient,
config: ProblemConfig = ProblemConfig()):
super().__init__(client, config)
validate_seven_coverage(config.bayes.station_ring_radius_m, config)
self.coverage_plan = seven_plan(config.bayes.station_ring_radius_m)
self.ledger = CoverageLedger(
config, points=list(self.coverage_plan.points))
self.particles = {channel: ChannelParticles(config)
for channel in self.belief.channels}
self._measured_points: dict[int, set[tuple[int, int]]] = {}
self._clear_failed: set[int] = set() # 证书清除失败后须先补测再重试
self._stall: dict[int, int] = {} # 连续无收缩计数（异常留痕用）
self._skip_cache: dict[tuple[int, int], tuple[int, bool]] = {}
self.skipped_measures = 0 # 跳测规则省下的检测次数
self.no_shrink_measures = 0 # 未收缩幸存集的定位检测（诊断用）
self.pf_absent_channels: list[int] = [] # 由幸存集为空证明缺席的频道
self.opportunistic_measures = 0
# ------------------------------------------------------------ 主循环
def _coverage_phase(self) -> None:
"""统一任务循环：发现 + 定位 + 清除滚动调度，直到任务表为空。"""
while self.stop_reason is None:
if self._count_limit_reached() or self._virtual_cap_reached():
return
if not self._step_once():
return
def _step_once(self) -> bool:
"""构建任务表、排序并执行首任务；无可执行任务返回 False。"""
if self.config.bayes.opportunistic_measure:
self._measure_at_stop()
tasks = self._build_tasks()
if not tasks:
return False
tasks = self._refine_locate_tasks(tasks)
order = self._order_tasks(tasks)
self._execute_task(order[0])
return True
def _measure_at_stop(self) -> None:
"""利用已停靠位置补交会方位；评分只作选测，证书仍用完整角域。
只测尚未在此处测过、有望显著收缩且保证距离可接收的已发现频道。
逐频道调用measure，换频与5秒检测照常计费；不在移动中检测。
此距离保证仅适用于第三问，不能用于未知定向源。
"""
point = self.client.state.position
channels = self.belief.channels_by_status(ChannelStatus.DETECTED)
channels.sort(key=lambda c: (c != self.client.state.channel, c))
for channel in channels:
if self._virtual_cap_reached():
return
if _point_key(point) in self._measured_points.get(channel, set()):
continue
cert = self._best_certificate(channel)
if cert is not None and cert.certified:
continue
region = self.belief.candidate_set(channel)
farthest = max(math.dist(point, v) for v in region.polygon)
if farthest > self.config.field.r_eff_min_m - 1e-3:
continue
radius = region.mec()[1]
vertices = region.polygon
stride = max(1, len(vertices) // self.config.bayes.mc_representatives)
radii = []
delta = (self.config.field.bearing_error_half_deg
+ self.config.search.bearing_rounding_margin_deg)
for target in vertices[::stride]:
theta = math.degrees(math.atan2(target[1]-point[1], target[0]-point[0]))
probe = CandidateRegion(list(vertices), region.slack_m)
probe.clip_wedge(point, theta, delta)
radii.append(probe.mec()[1] if probe.polygon else radius)
if sum(radii) / len(radii) > radius * self.config.bayes.opportunistic_radius_ratio:
continue
outcome = self.client.measure(point, channel)
self._record_measure(channel, point, outcome)
self._may_clear.add(channel)
self.opportunistic_measures += 1
# ------------------------------------------------------------ 任务构建
def _station_pending(self, index: int) -> list[int]:
"""站点待测频道：UNKNOWN、未测过、非"必 no_signal"且未证缺席。"""
point = self.ledger.points[index]
pending = []
for channel, belief in self.belief.channels.items():
if belief.status is not ChannelStatus.UNKNOWN:
continue
if self.ledger.measured_at(channel, index):
continue
if self._skip_station_channel(channel, index, point):
continue
pending.append(channel)
return pending
def _fused_channels_at(self, index: int) -> list[int]:
"""本站要测的频道 = 未发现(UNKNOWN) + 已发现但尚未定证(DETECTED)。
P0-1：把定位并入覆盖巡回。站点三角化可行性（方案 §1.3 F11）：
99.28% 的源位置有 ≥2 个站在 r_eff_max=1500m 内，最佳交会角中位
122.9°、5 分位 58.5°，因此"顺路取第二条方位线"对绝大多数源成立。
已有证书、已在同一 0.5m 测点测过、或该点必 no_signal 的频道跳过。
实测结论（2026-09-12，`output/suite/q3-fuse-ab*`）：本机制**不划算**。
七站环上通常只有 2–3 个站能对该源取得有效方位线，交会角也不足以
直接换到 20m 证书，专用信息增益探点仍然必要；仅在站点顺带测量等于
平白多花约 11 次检测（+0.8%），若同时抑制专用探点则 +45.6%（20/20
场景更慢）。默认关闭，保留代码以便复现与后续研究。
"""
point = self.ledger.points[index]
channels = self._station_pending(index)
selected = set(channels)
if not self.config.bayes.fuse_localize_into_tour:
return channels
for channel in self.belief.channels_by_status(ChannelStatus.DETECTED):
if channel in selected or self.ledger.measured_at(channel, index):
continue
if _point_key(point) in self._measured_points.get(channel, set()):
continue
cert = self._best_certificate(channel)
if cert is not None and cert.certified:
continue # 已有证书：交给清除任务，不再多测
if self._skip_station_channel(channel, index, point):
continue # 幸存集都在 1500m 外：该点必 no_signal
channels.append(channel)
selected.add(channel)
return channels
def _skip_station_channel(self, channel: int, index: int, point: Vec2) -> bool:
"""跳测规则（按 pf.version 缓存；幸存集单调收缩，判定单调）。"""
pf = self.particles[channel]
if not pf:
return True
cached = self._skip_cache.get((channel, index))
if cached is not None and cached[0] == pf.version:
return cached[1]
value = pf.guaranteed_no_signal(point)
self._skip_cache[(channel, index)] = (pf.version, value)
return value
def _build_tasks(self) -> list[Task]:
tasks: list[Task] = []
for index, point in enumerate(self.ledger.points):
pending = self._fused_channels_at(index)
if not pending:
continue
switches = len(pending) - (1 if self.client.state.channel in pending else 0)
tasks.append(Task("station", tuple(point),
self.config.cost.measure_s * len(pending)
+ self.config.cost.switch_s * max(switches, 0),
station_index=index, station_channels=tuple(pending)))
for channel in self.belief.channels_by_status(ChannelStatus.DETECTED):
cert = self._best_certificate(channel)
if cert is not None and cert.certified \
and channel not in self._clear_failed:
tasks.append(Task("clear", cert.center,
self.config.cost.clear_success_s,
channel=channel))
continue
if self._defer_to_station(channel):
continue # 下一条方位线可由"本就要去的站点"顺路提供
point = self._select_point(channel)
if point is None:
continue
tasks.append(Task("locate", point,
self.config.cost.measure_s + self.config.cost.switch_s,
channel=channel))
return tasks
def _defer_to_station(self, channel: int) -> bool:
"""该频道能否由"本就要去的站点"提供下一条方位线。
P0-1 的关键：仅仅在站点顺带测量是不够的（那只是多做一次检测），
必须同时**抑制专用探点任务**，否则滚动排序仍会优先飞向更近的探点。
判据要求该站本来就有 UNKNOWN 频道待测（覆盖巡回必须去），因此不会
为了等一条方位线而制造新的绕行，也不会出现"无任务可做"的死锁：
站点扫完后 ``_station_pending`` 为空，探点任务自然恢复。
"""
if not self.config.bayes.fuse_localize_into_tour:
return False
for index, point in enumerate(self.ledger.points):
if not self._station_pending(index):
continue
if self.ledger.measured_at(channel, index):
continue
if _point_key(point) in self._measured_points.get(channel, set()):
continue
if self._skip_station_channel(channel, index, point):
continue
return True
return False
# ------------------------------------------------------------ 任务排序
def _order_tasks(self, tasks: list[Task]) -> list[Task]:
"""开放路径排序：小任务表用精确 DP，否则最近邻 + 2-opt。
Held-Karp（S5/T1）把"已访问集合 × 结束时测向频道"并入状态，因此换频
代价是精确的，而不是 :meth:`_path_cost` 里"按频道排序"的贪心近似。
"""
if len(tasks) <= 2:
ordered = sorted(tasks, key=lambda t: (
move_time(self.client.state.position, t.position, self.config.cost),
t.kind, t.channel or 0))
return min((ordered, ordered[::-1]), key=self._path_cost)
limit = self.config.bayes.exact_order_limit
if 0 < limit and len(tasks) <= limit:
exact = self._exact_order(tasks)
if exact is not None:
return exact
order = self._nearest_order(tasks)
if not self.config.bayes.two_opt_enabled:
return order
return self._two_opt(order)
def _service_and_exit(self, task: Task,
channel: int) -> tuple[float, int]:
"""任务在进入频道 ``channel`` 时的固有耗时与离开后的测向频道。"""
cost = self.config.cost
if task.kind == "locate":
return (cost.measure_s + (cost.switch_s if channel != task.channel else 0.0),
task.channel)
if task.kind == "station" and task.station_channels:
pending = sorted(task.station_channels,
key=lambda c: (c != channel, c))
switches = len(pending) - (1 if channel in pending else 0)
return (cost.measure_s * len(pending) + cost.switch_s * switches,
pending[-1])
return task.service_s, channel
def _exact_order(self, tasks: list[Task]) -> list[Task] | None:
"""Held-Karp 精确开放路径（起点固定、终点自由，含换频状态）。
状态 ``(mask, c)`` = 已访问任务集 mask、离开最后一个任务后测向频道为 c；
频道取值域只需 {当前频道} ∪ {locate 频道} ∪ {station 离开频道}。
规模 ``2^n·n·|C|``；n 超过 :attr:`BayesConfig.exact_order_limit` 时由
调用方回退启发式。返回 ``None`` 表示状态过大（防御性）。
"""
n = len(tasks)
if n <= 1:
return list(tasks)
start = self.client.state.position
init = self.client.state.channel
channels = {init}
for task in tasks:
if task.kind == "locate":
channels.add(task.channel)
elif task.kind == "station" and task.station_channels:
# 离开频道 = 按"当前频道优先"排序后的最后一个：通常是最大值；
# 进入频道恰为最大值时退化为次大值，故两者都要进状态空间。
ordered = sorted(task.station_channels)
channels.add(ordered[-1])
if len(ordered) > 1:
channels.add(ordered[-2])
channels = sorted(channels)
width = len(channels)
if width * (1 << n) * n > 4_000_000:
return None
index_of = {channel: i for i, channel in enumerate(channels)}
c_init = index_of[init]
service: list[list[float]] = []
exits: list[list[int]] = []
for task in tasks:
row_service, row_exit = [], []
for channel in channels:
value, exit_channel = self._service_and_exit(task, channel)
row_service.append(value)
row_exit.append(index_of[exit_channel])
service.append(row_service)
exits.append(row_exit)
move = [[move_time(tasks[i].position, tasks[j].position, self.config.cost)
for j in range(n)] for i in range(n)]
from_start = [move_time(start, task.position, self.config.cost)
for task in tasks]
inf = float("inf")
# dp[(mask, ci)] = 最小代价；parent[(mask, ci)] = (上一 mask, 上一 ci)
dp: dict[tuple[int, int], float] = {}
last_task: dict[tuple[int, int], int] = {}
parent: dict[tuple[int, int], tuple[int, int] | None] = {}
for k in range(n):
key = (1 << k, exits[k][c_init])
value = from_start[k] + service[k][c_init]
if value < dp.get(key, inf):
dp[key] = value
last_task[key] = k
parent[key] = None
for mask in range(1, 1 << n):
for ci in range(width):
state = (mask, ci)
cost = dp.get(state)
if cost is None:
continue
for k in range(n):
if (mask >> k) & 1:
continue
nxt = (mask | (1 << k), exits[k][ci])
value = cost + move[last_task[state]][k] + service[k][ci]
if value < dp.get(nxt, inf) - 1e-12:
dp[nxt] = value
last_task[nxt] = k
parent[nxt] = state
full = (1 << n) - 1
best_key, best_cost = None, inf
for ci in range(width):
state = (full, ci)
if state in dp and dp[state] < best_cost:
best_key, best_cost = state, dp[state]
if best_key is None:
return None
order: list[Task] = []
key: tuple[int, int] | None = best_key
while key is not None:
order.append(tasks[last_task[key]])
key = parent[key]
order.reverse()
return order
def _nearest_order(self, tasks: list[Task]) -> list[Task]:
remaining = list(tasks)
order = []
position = self.client.state.position
while remaining:
best = min(remaining, key=lambda t: (
move_time(position, t.position, self.config.cost),
t.kind, t.channel or 0))
remaining.remove(best)
order.append(best)
position = best.position
return order
def _path_cost(self, order: list[Task]) -> float:
total = 0.0
position = self.client.state.position
channel = self.client.state.channel
for task in order:
total += move_time(position, task.position, self.config.cost)
if task.kind == "locate":
total += self.config.cost.measure_s
total += self.config.cost.switch_s * (channel != task.channel)
channel = task.channel
elif task.kind == "station" and task.station_channels:
pending = sorted(
task.station_channels,
key=lambda c, channel=channel: (c != channel, c))
total += self.config.cost.measure_s * len(pending)
total += self.config.cost.switch_s * (len(pending) - (channel in pending))
channel = pending[-1]
else:
total += task.service_s # 清除不改变测向频道
position = task.position
return total
def _two_opt(self, order: list[Task]) -> list[Task]:
best = list(order)
best_cost = self._path_cost(best)
for _ in range(self.config.bayes.two_opt_passes):
improved = False
for i in range(len(best) - 1):
for j in range(i + 1, len(best)):
candidate = best[:i] + best[i:j + 1][::-1] + best[j + 1:]
cost = self._path_cost(candidate)
if cost < best_cost - 1e-6:
best, best_cost, improved = candidate, cost, True
if not improved:
break
return best
# ------------------------------------------------------------ 任务执行
def _execute_task(self, task: Task) -> None:
if task.kind == "station":
self._execute_station(task.station_index)
elif task.kind == "clear":
cert = self._best_certificate(task.channel)
if cert is not None and cert.certified:
if not self._attempt_clear(task.channel, cert.center, cert):
self._clear_failed.add(task.channel)
self._may_clear.discard(task.channel)
else:
self._clear_failed.discard(task.channel)
elif task.kind == "locate":
self._execute_locate(task.channel, task.position)
def _execute_station(self, index: int) -> None:
"""单站批量扫频：只测待测频道（跳测规则已过滤），扫完即返回重排。
P0-1：本站同时测量"已发现未定证"频道，使第二/第三条方位线在巡回
途中顺路取得；证书一旦成立即就地清除，不额外绕行。
"""
point = self.ledger.points[index]
pending = self._fused_channels_at(index)
unknown = [c for c, b in self.belief.channels.items()
if b.status is ChannelStatus.UNKNOWN]
self.skipped_measures += len(unknown) - len(
[c for c in pending if self.belief.channels[c].status
is ChannelStatus.UNKNOWN])
# 当前频道优先，减少换频；同站 near 直接清除不打断扫频
pending.sort(key=lambda c: (c != self.client.state.channel, c))
for channel in pending:
if self.stop_reason is not None or self._count_limit_reached() \
or self._virtual_cap_reached():
return
detected_before = (self.belief.channels[channel].status
is ChannelStatus.DETECTED)
outcome = self.client.measure(point, channel)
self.ledger.mark_measured(channel, index)
self._record_measure(channel, point, outcome)
if outcome.result in (DIRECTION, NEAR):
self._may_clear.add(channel)
if outcome.result == NEAR:
region = self.belief.candidate_set(channel)
cert = clear_certificate(region, channel, self.config)
if cert.certified:
if self._attempt_clear(channel, point, cert):
self._clear_failed.discard(channel)
else:
self._clear_failed.add(channel)
self._may_clear.discard(channel)
else:
self._anomaly("near_without_certificate", channel, position=point)
continue
if not detected_before:
continue
if outcome.result == NO_SIGNAL:
# 该频道已被发现：保证接收盘内不可能 no_signal，属守恒性异常
region = self.belief.candidate_set(channel)
safe = (safe_disk_radius(region, self.config)
if region is not None else -1.0)
if safe >= 0 and math.dist(point, region.mec()[0]) <= safe:
self._anomaly("no_signal_in_safe_disk", channel,
position=point)
continue
cert = self._best_certificate(channel)
if cert is not None and cert.certified:
if self._attempt_clear(channel, cert.center, cert):
self._clear_failed.discard(channel)
else:
self._clear_failed.add(channel)
self._may_clear.discard(channel)
def _execute_locate(self, channel: int, point: Vec2) -> None:
"""单步定位测量：执行后立即重排（滚动时域），收敛则顺路清除。"""
pf = self.particles[channel]
before = len(pf)
outcome = self.client.measure(point, channel)
self._record_measure(channel, point, outcome)
self._may_clear.add(channel)
if outcome.result == NO_SIGNAL:
region = self.belief.candidate_set(channel)
if region is not None and safe_disk_radius(region, self.config) >= 0 \
and math.dist(point, region.mec()[0]) \
<= safe_disk_radius(region, self.config):
self._anomaly("no_signal_in_safe_disk", channel, position=point)
after = len(pf)
cert = self._best_certificate(channel)
if cert is not None and cert.certified:
if self._attempt_clear(channel, cert.center, cert):
self._clear_failed.discard(channel)
self._stall.pop(channel, None)
return
self._clear_failed.add(channel)
self._may_clear.discard(channel)
if after == before:
self._stall[channel] = self._stall.get(channel, 0) + 1
self.no_shrink_measures += 1 # 只是浪费不破坏守恒性，不进异常表
else:
self._stall[channel] = 0
# ------------------------------------------------------------ 感知与证书
def _record_measure(self, channel: int, position: Vec2, outcome) -> None:
"""belief（凸证书）与 PF（幸存集）同步更新；PF 空即证缺席。"""
self.belief.update_measure(channel, position, outcome.result,
outcome.svd_deg)
# 同一 0.5m 测点 × 同一频道的重复检测：历史上一次官方演练有 98.84%
# 的虚拟时间耗在空复扫上，本计数是该问题的回归指标。
key = _point_key(position)
seen = self._measured_points.setdefault(channel, set())
if key in seen:
self.redundant_rescan_measures += 1
seen.add(key)
self._clear_failed.discard(channel) # 新证据落地：允许再次尝试清除
belief = self.belief.channels[channel]
if belief.status in (ChannelStatus.CLEARED, ChannelStatus.ABSENT):
return # 已证毕：清除/缺席后的 no_signal 与距离无关，不进 PF
pf = self.particles[channel]
pf.apply_measure(position, outcome.result, outcome.svd_deg)
if not pf:
if belief.status is ChannelStatus.UNKNOWN:
self.belief.mark_absent(channel)
self.absence_evidence[channel] = "pf_empty"
if channel not in self.pf_absent_channels:
self.pf_absent_channels.append(channel)
else:
self._anomaly("pf_empty_while_detected", channel, position=position)
self._apply_count_bound()
def _apply_count_bound(self) -> None:
"""存在证据达到题面源数上界时，其余 UNKNOWN 可由计数直接判缺席。
依据 [题面]：每频道至多一个源、全局限定 10..16 个源。一旦
DETECTED/CLEARED 的不同频道数达到 16，任何尚未出现有效信号的
频道都不可能是源，无须等 7 站覆盖账本跑完。该判定与位置无关，
是严格的计数证明；evidence 与 coverage/pf_empty 分开留痕。
"""
present = [channel for channel, belief in self.belief.channels.items()
if belief.status in (ChannelStatus.DETECTED,
ChannelStatus.CLEARED)]
if len(present) < self.config.field.source_count_max:
return
for channel, belief in self.belief.channels.items():
if belief.status is ChannelStatus.UNKNOWN:
belief.status = ChannelStatus.ABSENT
self.absence_evidence[channel] = "count_bound"
def _best_certificate(self, channel: int) -> ClearCertificate | None:
"""凸证书优先；负观测挖洞使 PF 幸存集更紧时用网格证书补位。"""
region = self.belief.candidate_set(channel)
cert = clear_certificate(region, channel, self.config) \
if region is not None else None
if cert is not None and cert.certified:
return self._near_clear_target(cert)
if self.config.bayes.grid_cert_enabled:
pf = self.particles[channel]
if len(pf):
center, radius = pf.mec()
limit = (self.config.field.clear_radius_m - pf.margin_m
- self.config.search.clear_margin_m)
if radius <= limit:
return self._near_clear_target(
ClearCertificate(channel, center, radius, limit, True))
return cert
def _near_clear_target(self, cert: ClearCertificate) -> ClearCertificate:
"""沿当前来向利用20m清除容差；三角不等式保留整集覆盖证明。"""
position = self.client.state.position
distance = math.dist(position, cert.center)
shift = min(distance, max(0.0, cert.limit_m - cert.radius_m - 1e-6))
if distance <= 1e-9 or shift <= 0:
return cert
target = tuple(c + (p-c)*shift/distance for p, c in zip(position, cert.center))
return ClearCertificate(cert.channel, target, cert.radius_m+shift, cert.limit_m, True)
# ------------------------------------------------------------ 信息增益选点
def _probe_context(self, channel: int):
"""构造某频道下一次定位测量的候选集与信息增益评分函数。
返回 ``(candidates, info_score, mec_center, current_position)``；
候选为空返回 ``None``。``_select_point`` 与路线感知精修共用此逻辑，
保证两者只差“移动代价如何折算”。
"""
bayes, search = self.config.bayes, self.config.search
pf = self.particles[channel]
if not pf:
return None
region = self.belief.candidate_set(channel)
center, radius = pf.mec()
measured = self._measured_points.get(channel, set())
position = self.client.state.position
candidates: list[Vec2] = []
safe = safe_disk_radius(region, self.config) if region is not None else -1.0
if region is not None and safe >= 0:
# 保证接收盘（凸证书口径）：任意候选必产生 direction/near
candidates = candidate_points(tuple(region.mec()[0]), safe,
search.candidate_rings,
search.candidate_angles)
# 以实际来向生成对称交会候选，避免固定全局角度网格的旋转偏差。
geometric_center = region.mec()[0]
base_angle = math.atan2(position[1] - geometric_center[1],
position[0] - geometric_center[0])
for fraction in search.candidate_rings:
if fraction <= 0:
continue
for offset in (-math.pi/3, -math.pi/6, 0, math.pi/6, math.pi/3):
angle = base_angle + offset
candidates.append((geometric_center[0] + safe * fraction * math.cos(angle),
geometric_center[1] + safe * fraction * math.sin(angle)))
# 已停靠位置若本身在安全接收盘内，是零移动的合法测点；显式加入
# 可避免为了固定角度环而白白横移。
if math.dist(position, geometric_center) <= safe + 1e-9:
candidates.append(tuple(position))
else:
# 宽候选集：PF MEC 探索环（负观测挖洞后幸存集的包围圆）
for fraction in bayes.wide_rings:
ring = max(radius * fraction, 2.0 * pf.margin_m)
candidates.extend(candidate_points(center, ring, (1.0,),
search.candidate_angles))
candidates.append(tuple(center))
candidates.append(tuple(position))
candidates = [tuple(c) for c in candidates
if _point_key(c) not in measured]
if not candidates:
return None
# 采样一次转成数组（numpy 可用时），使候选评分走向量路径；
# numpy 缺失时 as_sample_array 原样返回列表，语义不变。
subsample = as_sample_array(pf.subsample(bayes.score_subsample))
stride = max(len(subsample) // bayes.mc_representatives, 1)
representatives = subsample[::stride]
def info_score(q: Vec2) -> float:
if bayes.score_mode == "entropy":
return pf.expected_survival(q, subsample, representatives)
return pf.expected_posterior_radius(q, subsample, representatives)
return candidates, info_score, center, position
def _select_point(self, channel: int) -> Vec2 | None:
"""定位测量点：期望 log 幸存占比 + λ·移动时间 最小的候选。"""
context = self._probe_context(channel)
if context is None:
return None
candidates, info_score, center, position = context
bayes = self.config.bayes
def score(q: Vec2) -> float:
# 一次定位后仍需接近源才能清除；仅计到检测点的移动会偏好绕行。
move = bayes.lambda_move_s * (move_time(position, q, self.config.cost)
+ move_time(q, center, self.config.cost))
return info_score(q) + move
return min(candidates, key=lambda q: (score(q), _point_key(q)))
def _refine_locate_tasks(self, tasks: list[Task]) -> list[Task]:
"""路线感知精修定位任务点（P2，2026-09-12）。
旧评分只比较 ``current->q + q->MEC``，是局部代理；滚动 2-opt 的真实
代价还包含 ``q->next_task``。这里对每个 locate 任务在候选点上计算
“替换该任务后整条当前路线代价的增量”，再与信息增益、MEC 接近项合并
选点。只改变选点启发式，不改变覆盖/清除证明。
"""
bayes = self.config.bayes
if not bayes.route_aware_probe:
return tasks
locate_indices = [index for index, task in enumerate(tasks)
if task.kind == "locate"]
if not locate_indices:
return tasks
base_order = self._order_tasks(tasks)
base_cost = self._path_cost(base_order)
refined = list(tasks)
for task_index in locate_indices:
task = tasks[task_index]
context = self._probe_context(task.channel)
if context is None:
continue
candidates, info_score, center, position = context
try:
order_index = next(index for index, item in enumerate(base_order)
if item is task)
except StopIteration: # pragma: no cover - 同对象应始终存在
order_index = next(index for index, item in enumerate(base_order)
if item.kind == "locate"
and item.channel == task.channel)
best_score: float | None = None
best_point: Vec2 | None = None
for point in candidates:
trial_order = list(base_order)
trial_order[order_index] = Task(
"locate", tuple(point), task.service_s, channel=task.channel)
route_delta = self._path_cost(trial_order) - base_cost
move_proxy = bayes.lambda_move_s * (
move_time(position, point, self.config.cost)
+ move_time(point, center, self.config.cost))
clear_near = move_time(point, center, self.config.cost)
value = (info_score(point)
+ move_proxy
+ bayes.route_aware_insert_weight * route_delta
+ bayes.route_aware_clear_weight * clear_near)
if best_score is None or value < best_score:
best_score, best_point = value, tuple(point)
if best_point is not None:
refined[task_index] = Task(
"locate", best_point, task.service_s, channel=task.channel)
return refined
# ------------------------------------------------------------ 安全网与复扫
def _localize_and_clear(self, channel: int) -> bool:
"""阻塞式定位子循环（resolve 阶段安全网）：PF 选点，失败退凸选点。"""
search = self.config.search
self._may_clear.add(channel)
for _ in range(search.max_localize_steps):
if self._virtual_cap_reached():
return False
cert = self._best_certificate(channel)
if cert is not None and cert.certified and channel in self._may_clear:
if self._attempt_clear(channel, cert.center, cert):
self._clear_failed.discard(channel)
return True
self._clear_failed.add(channel)
self._may_clear.discard(channel)
continue
point = self._select_point(channel)
if point is None:
region = self.belief.candidate_set(channel)
if region is None:
return False
point = select_measurement_point(
region, self.client.state.position, self.config).position
outcome = self.client.measure(point, channel)
self._record_measure(channel, point, outcome)
self._may_clear.add(channel)
if outcome.result == NO_SIGNAL:
region = self.belief.candidate_set(channel)
if region is not None and safe_disk_radius(region, self.config) >= 0 \
and math.dist(point, region.mec()[0]) \
<= safe_disk_radius(region, self.config):
self._anomaly("no_signal_in_safe_disk", channel, position=point)
cert = self._best_certificate(channel)
if cert is not None and cert.certified:
if self._attempt_clear(channel, cert.center, cert):
self._clear_failed.discard(channel)
return True
self._clear_failed.add(channel)
self._may_clear.discard(channel)
return self.belief.channels[channel].status is ChannelStatus.CLEARED
def _persistent_round(self) -> None:
"""一轮复扫：与基线一致的全站复测，但读数同步进 PF（防假设偏差）。"""
measured: set[tuple[int, int]] = set()
while self.stop_reason is None:
index = self._next_resweep_point(measured)
if index is None:
return
point = self.ledger.points[index]
for channel in self._resweep_channels_at(index, measured):
if self._virtual_cap_reached():
return
measured.add((channel, index))
was_absent = (self.belief.channels[channel].status
is ChannelStatus.ABSENT)
outcome = self.client.measure(point, channel)
self._record_measure(channel, point, outcome)
if was_absent and outcome.result in (DIRECTION, NEAR):
# 复测推翻了缺席证明：守恒性留痕，按新证据定位清除。
self._anomaly("absent_detected_on_resweep", channel,
position=point, result=outcome.result)
if outcome.result == DIRECTION:
self._localize_and_clear(channel)
elif outcome.result == NEAR:
cert = self._best_certificate(channel)
if cert is not None and cert.certified:
if not self._attempt_clear(channel, cert.center, cert):
self._clear_failed.add(channel)
else:
self._anomaly("near_without_certificate", channel,
position=point)
if self.stop_reason is not None:
return
# ------------------------------------------------------------ 收尾
def _summary(self, info) -> dict:
report = super()._summary(info)
report["strategy"] = "bayes"
report["opportunistic_measures"] = self.opportunistic_measures
report["skipped_measures"] = self.skipped_measures
report["no_shrink_measures"] = self.no_shrink_measures
report["pf_absent_channels"] = sorted(self.pf_absent_channels)
report["pf_alive_final"] = {
str(channel): len(self.particles[channel])
for channel in sorted(self.particles)
if self.belief.channels[channel].status is ChannelStatus.DETECTED}
return report
In [ ]:
class SevenBatch(OmnidirectionalLoop):
"""站内只扫频，扫完按最近候选集圆心顺序服务检测队列（A2 主策略）。"""
def _serve_queue(self):
# Each detected channel gets at most one localization call in this pass.
queue = self.belief.channels_by_status(ChannelStatus.DETECTED)
while queue:
c = min(queue, key=lambda c: (math.dist(self.client.state.position,
self.belief.candidate_set(c).mec()[0]), c))
queue.remove(c)
self._localize_and_clear(c)
if self._count_limit_reached():
break
def _coverage_phase(self):
while not self._count_limit_reached():
index = self._next_point_index()
if index is None:
return
point = self.ledger.points[index]
pending = self._pending_channels_at(index)
# Keeping the current measuring channel first saves a switch when possible.
pending.sort(key=lambda c: (c != self.client.state.channel, c))
for c in pending:
outcome = self.client.measure(point, c)
self.ledger.mark_measured(c, index)
self.belief.update_measure(c, point, outcome.result, outcome.svd_deg)
if outcome.result == NEAR:
cert = clear_certificate(self.belief.candidate_set(c), c, self.config)
# Same-station near clear; do not displace the station scan.
if cert.certified:
self._attempt_clear(c, point, cert)
if self._count_limit_reached():
return
self._serve_queue()
def _resolve_detected(self):
self._serve_queue()
unresolved = [c for c, b in self.belief.channels.items()
if b.status not in (ChannelStatus.CLEARED, ChannelStatus.ABSENT)]
return "all_resolved" if not unresolved else "unresolved_channels"
In [ ]:
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
In [ ]:
# 反例：直径圆的圆心是"最远顶点对"的中点，第二个最远顶点可能落在圆外。
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"
In [ ]:
# 示向度二位小数的舍入敏感性：误差半角 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
In [ ]:
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)
In [ ]:
# 用同一套观测语义复核这个点：无论真源落在候选集何处，第二点都必收到信号。
# 候选集边界按 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
In [ ]:
# 对照：候选集在不同首次示向度下的选择结果（目标函数与移动代价的权衡）
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))
In [ ]:
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
In [ ]:
# ---- 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
In [ ]:
# ---- 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
In [ ]:
# ---- 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
In [ ]:
# ---- 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("→ 采样分辨率必须与证书一致，否则'删掉一个关键点仍看似覆盖'（校核退化为装饰）")
In [ ]:
# ---- 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
In [ ]:
# ---- 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()
In [ ]:
# ---- 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
In [ ]:
# ---- 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()
In [ ]:
# ---- 总自检：把四问的核心结论再跑一遍，全部通过才算整理成功 ----------------
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))