yes fuking shit

This commit is contained in:
Horldsence
2026-09-12 14:48:43 +08:00
parent 61fd22598a
commit 45e48d4c91
64 changed files with 11540 additions and 57 deletions
+388
View File
@@ -0,0 +1,388 @@
"""位置×朝向联合可行集(第四问定向源)。
现状问题(docs/第三四问架构优化与运行提速方案.md §1.3 F3):第四问的失败**全部**
来自定向源(评测机两局:全向 17/17 清除,定向仅 2/6)。原因不是覆盖不足,而是
策略只知道"源可能在某个空间区域",不知道"它朝着哪边"——于是选点只能靠固定
45° 阶梯试探,一旦试探失败就退化到 ≤400 格 20 m 盲清。
本模块把"看不看得见"变成可推理的量:为已在 Ω 上维护的网格幸存集
(:class:`~mcm26b.particles.ChannelParticles`)附加每格 24 档朝向可行性掩码,
得到 (位置 × 朝向) 联合可行集。观测语义严格对齐评测机
``referee.simulator._observe``:
- 距离 > r_eff → no_signal;定向且源→读点方向与朝向夹角 > 90° → no_signal;
- 距离 ≤ 5 m → near(**仍需可见**:3 m 处背向的定向源同样返回 no_signal);
- 其余 → direction(读数 = 源→读点方位 + 有界误差)。
更新规则(全部保守,只删"必然不可能"的格/档):
=============== ============================================ ==================
观测 位置 朝向
=============== ============================================ ==================
``direction``@s 距离 ≤ 1500、角向 |gap| ≤ δ_eff + 格裕量 必须**可能**覆盖 s
``near``@s 距离 ≤ 5 必须**可能**覆盖 s
``no_signal``@s **不删**(可能是朝向遮挡或距离不足) 必须**确定不**覆盖 s
(仅当整格必在 r_eff_min 内才约束朝向)
=============== ============================================ ==================
阈值全部取保守侧,并计入**格内角裕量**
``asin(margin / max(dist − margin, 1))``(格内任一点相对格心方位偏差上界):
- "必须可能覆盖":``gap ≤ 90° + 半档 + 查表舍入 + 格内角裕量``(宁可保留);
- "必须确定不覆盖":先删去 ``gap ≤ 90° − 半档 − 查表舍入 − 格内角裕量`` 的档,
只保留确定看不见 s 的档(同样宁可保留)。
因此真源 (位置, 朝向) 永不被删——这是网格幸存集真源保留定理的二维推广。
(2026-09-12 修正:no_signal 分支原先误用"确定覆盖"掩码做交集,等价于把
规则反向;keep=True 分支也漏计了格内角裕量。两处都会删掉真源朝向档,
现由 ``tests/test_joint_belief.py`` 的随机观测序列保留性测试固化。)
用途:
1. ``prob_detect``:该点能看见该频道任一可行 (位置, 朝向) 的概率质量,
用于**可见性选点**(取代固定阶梯)——只影响选点顺序,不承担安全责任;
2. ``expected_posterior_radius``:一次测量后的期望后验 MEC 半径(D1 值函数);
3. ``is_absent``:联合集空 ⟹ 该频道不存在源(严格,与覆盖证明同一条件);
4. 位置边际仍由凸证书负责 20 m 清除判定,本模块不改变任何证书口径。
"""
from __future__ import annotations
import math
from ..config import ProblemConfig, Vec2
from ..particles import ChannelParticles, _branch_radius_array
from ..protocol.rules import DIRECTION, NEAR, NO_SIGNAL
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))
try: # numpy 仅作可选加速后端(与 particles 一致)
import numpy as _np
from ..particles import _np as _particles_np
if _particles_np is None: # MCM26B_NO_NUMPY 或缺失:强制纯 Python
_np = None
except Exception: # pragma: no cover - 环境无 numpy
_np = None
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}
+11
View File
@@ -206,6 +206,17 @@ def _run(args) -> int:
search=replace(config.search,
bearing_rounding_margin_deg=args.bearing_margin_deg,
persistent_search=args.persistent))
if args.persistent:
# 历史教训:一次官方演练在 persistent 复扫下把 98.84% 的虚拟时间
# (346170 s)花在空复扫上。这里给出单轮上界,明确这是"扫到限时"模式。
stations = 49 if args.problem == 4 else 7
per_round = config.field.channel_max * stations * (
config.cost.measure_s + config.cost.switch_s)
print(f"[警告] --persistent:找全后仍复扫未清除频道直到限时。"
f"单轮上界约 {per_round:.0f}s 虚拟时间"
f"({config.field.channel_max} 频道 × {stations} 站点 × "
f"{config.cost.measure_s + config.cost.switch_s:.0f}s),"
f"且会持续多轮;成绩口径请保持默认(关闭)。", file=sys.stderr)
truth = None
transport = None
real_budget_s = None
+69 -3
View File
@@ -91,6 +91,45 @@ class SearchConfig:
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,不复扫。复扫保险模式(防缺席证明假设偏差)保留,经
@@ -119,18 +158,45 @@ class BayesConfig:
# radius=期望后验半径(直指证书)
score_subsample: int = 800 # 期望幸存熵估计的粒子子采样规模
mc_representatives: int = 7 # 假设真值代表点数(均匀抽取幸存格)
lambda_move_s: float = 0.0105 # 选点评分中移动时间的折算系数(分值/秒);
# entropy 口径下信息项为 O(1) 对数值,λ须极小
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 # 已停靠位置顺路交会;仍逐频道计费
opportunistic_radius_ratio: float = 0.65 # 预测后验半径至少减少35%才补测
# 预测后验半径至少减少 (1−ratio) 才补测;0.5 优于 0.65(18 场景 3205.4 vs 3210.3 s)
opportunistic_radius_ratio: float = 0.5
@dataclass(frozen=True)
+23
View File
@@ -0,0 +1,23 @@
"""保证层:覆盖计划(CoveragePlan)、证书与运行期校验。
对外只暴露覆盖计划与校验入口;`coverage` 包仍负责逐频道覆盖账本。
"""
from .cover import (CERTIFY_HEADINGS, CERTIFY_STEP_M, OMEGA_RADIUS_M,
R_EFF_MIN_M, CoverageCertificate,
CoveragePlan, PLAN_DIR, VERIFY_HEADINGS, VERIFY_STEP_M,
candidate_lattice, coverage_masks, forward_targets,
greedy_cover, load_plan, nn_tour_length, omega_positions,
plan_instance, plan_names, seven_plan, seven_points,
square_49_plan, verify_full_disk, verify_forward_half_disk,
verify_plan, write_plan)
__all__ = [
"CERTIFY_HEADINGS", "CERTIFY_STEP_M", "OMEGA_RADIUS_M", "R_EFF_MIN_M",
"CoverageCertificate", "CoveragePlan",
"PLAN_DIR", "VERIFY_HEADINGS", "VERIFY_STEP_M", "candidate_lattice",
"coverage_masks", "forward_targets", "greedy_cover", "load_plan",
"nn_tour_length", "omega_positions", "plan_instance", "plan_names",
"seven_plan", "seven_points", "square_49_plan", "verify_full_disk",
"verify_forward_half_disk", "verify_plan", "write_plan",
]
+428
View File
@@ -0,0 +1,428 @@
"""保证层:可离线搜索、可在运行期校验的覆盖计划(CoveragePlan)。
两类覆盖语义(题面约束不同,绝不可混用):
- ``full_disk``(第三问,全向源):在 s 处 no_signal 排除 B(s, r_eff_min),
因此"全点皆 no_signal ⟹ 不存在源"要求
``∪ B(c, 1000) ⊇ Ω``。解析最优解是"原点 + 半径 r 的 6 等分环",
``r`` 的可行区间由 ``validate_seven_coverage`` 给出(本模块提供数值复核)。
- ``forward_half_disk``(第四问,定向源):源可能背向读点,no_signal 不再
排除整个圆盘,缺席证明要求
``∀(p,h) ∈ Ω×S¹, ∃c ∈ S: |c−p| ≤ 1000 ∧ (c−p)·u(h) ≥ 0``。
现役 700 m 方格 49 点是充分条件(单元对角 ≤ 1000),但远非必要;
本模块用贪心击中集把点数压到 31,并在同一采样空间上校核。
设计原则(docs/第三四问架构优化与运行提速方案.md §2.3):
**保证 = 离线搜索 + 运行期校验**。点表以 JSON 落盘并携带证书(采样分辨率、
缺口数、最坏命中距离、生成参数哈希),运行期用同一套采样语义重新校核,
未通过即拒绝启动。证书只能用于选点优化,不改变任何安全性证明。
"""
from __future__ import annotations
import hashlib
import json
import math
from dataclasses import dataclass, field, fields
from functools import lru_cache
from pathlib import Path
from ..config import Vec2
PLAN_DIR = Path(__file__).resolve().parent / "plans"
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 _load_plan_file(path: Path) -> CoveragePlan:
payload = json.loads(path.read_text(encoding="utf-8"))
raw_cert = payload.get("certificate") or {}
if raw_cert:
# as_dict() 含 nn_tour_virtual_s 等派生列:只取数据类实际字段。
known = {field.name for field in fields(CoverageCertificate)}
certificate = CoverageCertificate(
**{key: value for key, value in raw_cert.items() if key in known})
else:
certificate = 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"]))
@lru_cache(maxsize=None)
def load_plan(name: str, verify: bool = True) -> CoveragePlan:
"""按名称加载覆盖计划;``verify`` 为真时做运行期抽样校验(进程内缓存)。"""
path = PLAN_DIR / f"{name}.json"
if not path.exists():
raise FileNotFoundError(f"未找到覆盖计划 {name!r}({path})")
plan = _load_plan_file(path)
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_instance(name: str, verify: bool = True) -> CoveragePlan:
"""统一入口:支持内置计划名与磁盘上的证书计划。"""
if name == "square49":
return square_49_plan()
if name == "seven":
return seven_plan(1200.0)
return load_plan(name, verify)
def plan_names() -> list[str]:
"""全部可加载的计划名(含内置)。"""
builtin = ["square49", "seven"]
return builtin + sorted(p.stem for p in PLAN_DIR.glob("*.json"))
def write_plan(plan: CoveragePlan, path: Path) -> None:
"""落盘计划 + 证书(生成器用)。"""
path.parent.mkdir(parents=True, exist_ok=True)
payload = plan.as_dict()
payload["generator"] = dict(payload.get("generator") or {})
payload["sha256"] = hashlib.sha256(
json.dumps(payload["points"]).encode("utf-8")).hexdigest()
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8")
@@ -0,0 +1,143 @@
{
"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": null,
"points": [
[
-800.0,
0.0
],
[
800.0,
0.0
],
[
0.0,
-1000.0
],
[
0.0,
1000.0
],
[
-1200.0,
-1000.0
],
[
-1000.0,
1000.0
],
[
1000.0,
-1000.0
],
[
1000.0,
1000.0
],
[
1800.0,
0.0
],
[
-1800.0,
200.0
],
[
-400.0,
-1800.0
],
[
-200.0,
1800.0
],
[
0.0,
0.0
],
[
800.0,
-1800.0
],
[
1200.0,
1600.0
],
[
-1400.0,
1400.0
],
[
1800.0,
-800.0
],
[
-1800.0,
-800.0
],
[
1800.0,
600.0
],
[
-600.0,
-1000.0
],
[
400.0,
1800.0
],
[
-1400.0,
-1800.0
],
[
-1200.0,
800.0
],
[
-1200.0,
1800.0
],
[
400.0,
-800.0
],
[
-200.0,
400.0
],
[
600.0,
-1400.0
],
[
-2200.0,
600.0
],
[
1000.0,
-1600.0
],
[
-2000.0,
-400.0
],
[
-200.0,
-2200.0
]
],
"sha256": "000e0c1c8642bbe9fb15ad854b22a5df03d4385f7834135a129062822442ccd1"
}
@@ -0,0 +1,162 @@
{
"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.0,
-200.0
],
[
800.0,
200.0
],
[
-200.0,
900.0
],
[
200.0,
-900.0
],
[
-1200.0,
700.0
],
[
1200.0,
-700.0
],
[
800.0,
1200.0
],
[
-1800.0,
-300.0
],
[
1800.0,
300.0
],
[
-400.0,
1800.0
],
[
0.0,
0.0
],
[
1400.0,
-1400.0
],
[
-1500.0,
-1300.0
],
[
1500.0,
1300.0
],
[
-400.0,
-1800.0
],
[
400.0,
1800.0
],
[
-1800.0,
400.0
],
[
1800.0,
-400.0
],
[
-500.0,
700.0
],
[
-700.0,
-700.0
],
[
700.0,
700.0
],
[
1300.0,
1100.0
],
[
-1300.0,
1500.0
],
[
500.0,
-1800.0
],
[
-900.0,
-2100.0
],
[
600.0,
1900.0
],
[
-2500.0,
500.0
],
[
1900.0,
-900.0
],
[
-1700.0,
-500.0
],
[
-700.0,
-1200.0
],
[
900.0,
-700.0
]
],
"sha256": "a7fc25beb70bc159bf3a9d309813677fb72277a78250374fcd8039e5699748fe"
}
+3 -2
View File
@@ -11,8 +11,9 @@
from .playback import PlaybackTransport
from .scenes import mixed_scene, random_omni_scene, referee_spec
from .simulator import LocalSimulator
from .suites import SceneSpec, suite, suite_manifest
from .transport import LocalSimTransport, ScriptedFaultTransport
__all__ = ["LocalSimTransport", "LocalSimulator", "PlaybackTransport",
"ScriptedFaultTransport", "mixed_scene", "random_omni_scene",
"referee_spec"]
"ScriptedFaultTransport", "SceneSpec", "mixed_scene",
"random_omni_scene", "referee_spec", "suite", "suite_manifest"]
+132
View File
@@ -0,0 +1,132 @@
"""冻结基准集:第三/四问的固定场景清单与真值构造(测试端专用)。
存在意义(docs/第三四问架构优化与运行提速方案.md §1.4 B6 / §P1-3):
历史评估集是 20 个**单源全向**场景,使 `avg_locate_clear_time_s` 恒等于
`virtual_time_s`,也无法支撑任何第四问结论。基准集必须覆盖题面声明的
源数区间与定向源存在性,并且**版本化、可复现、不可随意增删**——
任何"更快"的声明都要在同一份清单上做配对比较。
真值只在本模块与评测机之间流动;策略端只通过 Transport 看到响应。
"""
from __future__ import annotations
import math
from dataclasses import dataclass
from referee import Source
from ..config import ProblemConfig
from .scenes import mixed_scene, random_omni_scene
SUITE_VERSION = 1
# 第三问:沿用历史验证集的种子与源数(seeds 100–105 × 10/13/16)
Q3_SEEDS = (100, 101, 102, 103, 104, 105)
Q3_COUNTS = (10, 13, 16)
# 第四问:三种"定向占比"组成(少量 / 约半数 / 多数)
Q4_SEEDS = (200, 201, 202)
Q4_CONFIGS = ((10, 3), (13, 6), (16, 8))
# 误差场种子偏移:与 benchmark_q3_bayes.py 同口径,保证逐点固定误差可复现
ERROR_SEED_BASE = 20260910
@dataclass(frozen=True)
class SceneSpec:
"""一个冻结场景:标识、题号、源数、定向数与种子。"""
scene_id: str
problem: int
kind: str # "random" | "boundary"
seed: int
error_seed: int
count: int
directional: int = 0
def as_dict(self) -> dict:
return {"scene_id": self.scene_id, "problem": self.problem,
"kind": self.kind, "seed": self.seed,
"error_seed": self.error_seed, "count": self.count,
"directional": self.directional}
def sources(self, config: ProblemConfig | None = None) -> list[Source]:
"""构造真值(仅测试端调用)。"""
config = config or ProblemConfig()
if self.kind == "random" and self.problem == 3:
return random_omni_scene(seed=self.seed, count=self.count,
config=config)
if self.kind == "random":
return mixed_scene(seed=self.seed, count=self.count,
directional=self.directional, config=config)
return _boundary_sources(self, config)
def _boundary_sources(spec: SceneSpec, config: ProblemConfig) -> list[Source]:
"""圆周边界压力场景:源全部位于 Ω 边界、r_eff 取下界。
第三问:全向,扇区接缝角度,检验七站覆盖的边界情形;
第四问:全部定向且朝向**背离圆心**(最不利于发现),检验
"朝外发射的边界源不漏"。
"""
field = config.field
radius = field.omega_radius_m
sources: list[Source] = []
for index in range(spec.count):
angle = math.pi / 6 + 2 * math.pi * index / spec.count
x, y = radius * math.cos(angle), radius * math.sin(angle)
directional = index < spec.directional
sources.append(Source(
channel=index + 1, x=x, y=y, r_eff_m=field.r_eff_min_m,
directional=directional,
heading_deg=(math.degrees(angle) % 360.0) if directional else 0.0))
return sources
def core_scenes(problem: int) -> list[SceneSpec]:
"""主体场景:覆盖题面源数区间与定向占比组成。"""
scenes: list[SceneSpec] = []
if problem == 3:
for seed in Q3_SEEDS:
for count in Q3_COUNTS:
scenes.append(SceneSpec(
scene_id=f"q3-core-s{seed}-n{count}", problem=3,
kind="random", seed=seed, error_seed=ERROR_SEED_BASE + seed,
count=count))
else:
for seed in Q4_SEEDS:
for count, directional in Q4_CONFIGS:
scenes.append(SceneSpec(
scene_id=f"q4-core-s{seed}-n{count}-d{directional}",
problem=4, kind="random", seed=seed,
error_seed=ERROR_SEED_BASE + seed, count=count,
directional=directional))
return scenes
def boundary_scenes(problem: int) -> list[SceneSpec]:
"""圆周边界压力场景(每问 2 场:10 源与 16 源)。"""
if problem == 3:
return [SceneSpec(f"q3-boundary-n{count}", 3, "boundary", 0,
ERROR_SEED_BASE, count)
for count in (10, 16)]
return [SceneSpec(f"q4-boundary-n{count}-d{directional}", 4, "boundary", 0,
ERROR_SEED_BASE, count, directional)
for count, directional in ((10, 5), (16, 8))]
def suite(name: str, problem: int) -> list[SceneSpec]:
"""按名称取场景清单:core / boundary / all。"""
if name == "core":
return core_scenes(problem)
if name == "boundary":
return boundary_scenes(problem)
if name == "all":
return core_scenes(problem) + boundary_scenes(problem)
raise ValueError(f"未知基准集 {name!r}(可选 core/boundary/all)")
def suite_manifest(name: str, problem: int) -> dict:
"""清单头:版本、问题、场景表(写入 manifest.json)。"""
scenes = suite(name, problem)
return {"suite_version": SUITE_VERSION, "suite": name, "problem": problem,
"scenes": [scene.as_dict() for scene in scenes]}
+114 -13
View File
@@ -28,6 +28,7 @@ m = sqrt(2)/2·h 是格心到格内任一点的最大距离。所有 kill 判定
from __future__ import annotations
import math
import os
from .config import ProblemConfig, Vec2
from .geometry.localization import smallest_enclosing_circle
@@ -37,12 +38,38 @@ _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:
"""单个频道的网格幸存集。
@@ -95,6 +122,10 @@ class ChannelParticles:
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
@@ -171,9 +202,17 @@ class ChannelParticles:
self._discard(doomed)
def apply_measure(self, station: Vec2, result: str,
svd_deg: float | None = None) -> None:
"""按读数更新幸存集。真源保留性由 Voronoi 裕量保证(模块注释)。"""
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:
@@ -263,8 +302,59 @@ class ChannelParticles:
"""按索引序均匀抽取的子采样(索引序≈空间序,代表点空间铺开)。"""
return self.alive_points(limit=size)
def expected_survival(self, point: Vec2, subsample: list[Vec2],
representatives: list[Vec2]) -> float:
# ---------------------------------------------------- 向量化评分(可选)
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)
@@ -273,8 +363,11 @@ class ChannelParticles:
- direction 分支:以代表点方位为期望读数,角域 ∩ B(point,1500+m) 内幸存;
- near 分支(dist <= 5+m):可行集收缩到 5m 盘,幸存占比趋近 0。
"""
if not self.alive or not subsample:
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),
@@ -302,24 +395,28 @@ class ChannelParticles:
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: list[Vec2], point: Vec2,
radius_m: float) -> float:
if not samples:
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: list[Vec2]) -> float:
def _branch_radius(self, members) -> float:
"""分支后验的半径代理:幸存子采样到其质心的最大距离(O(n),免 Welzl)。"""
if not members:
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: list[Vec2],
representatives: list[Vec2]) -> float:
def expected_posterior_radius(self, point: Vec2, subsample,
representatives) -> float:
"""一次测量后期望后验半径(米):选点主指标,直指清除证书阈值。
对每个代表假设(真值候选)按 r_eff~U(1000,1500) 解析积分
@@ -328,8 +425,12 @@ class ChannelParticles:
与几何版 worst-posterior 同族,但作用于含负观测的非凸幸存集,
且以期望代替最坏(代表点覆盖幸存集主轴)。
"""
if not self.alive or not subsample:
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:
+225 -11
View File
@@ -36,14 +36,17 @@ from dataclasses import dataclass
from ..belief import (BeliefState, CandidateRegion, ChannelStatus,
InconsistentBelief, clip_halfplane)
from ..belief.joint import JointBelief
from ..config import ProblemConfig, Vec2
from ..coverage import CoverageLedger
from ..evaluation import recompute_virtual_time, summarize
from ..guarantee import plan_instance, seven_plan, square_49_plan
from ..protocol.client import (DeadlineExceeded, ProtocolError,
SimulatorClient)
from ..protocol.rules import CLEAR_SUCCESS, DIRECTION, NEAR, NO_SIGNAL
from ..protocol.timing import move_time
from ..protocol.transport import TransportError
from ..routing import improve_open_path
def bearing_deg(a: Vec2, b: Vec2) -> float:
@@ -211,10 +214,18 @@ class OmnidirectionalLoop:
self.client = client
self.config = config
self.belief = BeliefState(config)
self.ledger = CoverageLedger(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 失败后必须先补测再重试,避免在同一证书上
@@ -267,14 +278,27 @@ class OmnidirectionalLoop:
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,调用方立即收尾。
余量须大于单次动作耗时上界(网格最远移动 ~1200s + 检测 5s),
search.virtual_reserve_s=10000s [工程] 提供充分裕度,避免触及
余量须大于单次动作耗时上界(见 :meth:`_min_virtual_reserve`),
`search.virtual_reserve_s=10000s` [工程] 提供充分裕度,避免触及
附件2 §4.5 的 100 小时上限被模拟器强制结束。
"""
if (self.client.state.virtual_time_s + self.config.search.virtual_reserve_s
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"
@@ -466,6 +490,7 @@ class OmnidirectionalLoop:
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 频道做第二轮定位。
@@ -503,8 +528,19 @@ class OmnidirectionalLoop:
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,
@@ -512,9 +548,19 @@ class OmnidirectionalLoop:
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"],
@@ -597,9 +643,18 @@ class DirectionalExtension(OmnidirectionalLoop):
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站的定向覆盖证明不变;每站先串行扫频,再就近服务已发现源。
@@ -619,7 +674,7 @@ class DirectionalExtension(OmnidirectionalLoop):
return
outcome = self.client.measure(point, channel)
self.ledger.mark_measured(channel, index)
self.belief.update_measure(channel, point, outcome.result, outcome.svd_deg)
self._observe(channel, point, outcome.result, outcome.svd_deg)
if outcome.result == NEAR:
cert = clear_certificate(
self.belief.candidate_set(channel), channel, self.config)
@@ -666,8 +721,7 @@ class DirectionalExtension(OmnidirectionalLoop):
within_range = (math.dist(point, region.mec()[0])
<= safe_disk_radius(region, self.config))
outcome = self.client.measure(point, channel)
self.belief.update_measure(channel, point, outcome.result,
outcome.svd_deg)
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:
@@ -688,6 +742,139 @@ class DirectionalExtension(OmnidirectionalLoop):
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:
"""定向遮挡下的探点升级梯子 [工程]。
@@ -699,6 +886,9 @@ class DirectionalExtension(OmnidirectionalLoop):
切点,固定步进环探可以全部漏过。梯子只提高提前命中概率,
不构成接收或命中保证;完备性完全由后备清除兜底。
"""
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:
@@ -743,14 +933,12 @@ class DirectionalExtension(OmnidirectionalLoop):
entry["outcome"] = "skipped_budget"
return False
attempts = 0
remaining_cells = list(cells)
while remaining_cells:
order = self._fallback_order(cells)
for cell in order:
if self._virtual_cap_reached():
entry["outcome"] = "skipped_virtual_budget"
entry["attempts"] = attempts
return False
cell = min(remaining_cells, key=lambda c: (math.dist(self.client.state.position, c), c))
remaining_cells.remove(cell)
attempts += 1
outcome = self.client.clear(cell, channel)
if outcome.result == CLEAR_SUCCESS:
@@ -767,6 +955,28 @@ class DirectionalExtension(OmnidirectionalLoop):
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)
@@ -775,6 +985,10 @@ class DirectionalExtension(OmnidirectionalLoop):
# 后备 /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
+328 -25
View File
@@ -24,13 +24,14 @@ from dataclasses import dataclass
from ..belief import CandidateRegion, ChannelStatus
from ..config import ProblemConfig, Vec2
from ..coverage import CoverageLedger
from ..particles import ChannelParticles
from ..particles import ChannelParticles, as_sample_array
from ..protocol.client import SimulatorClient
from ..protocol.rules import DIRECTION, NEAR, NO_SIGNAL
from ..protocol.timing import move_time
from . import (ClearCertificate, OmnidirectionalLoop, clear_certificate,
candidate_points, safe_disk_radius, select_measurement_point)
from .seven import seven_points, validate_seven_coverage
from .seven import validate_seven_coverage
from ..guarantee import seven_plan
@dataclass(frozen=True)
@@ -57,8 +58,9 @@ class BayesLoop(OmnidirectionalLoop):
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=seven_points(config.bayes.station_ring_radius_m))
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]]] = {}
@@ -86,6 +88,7 @@ class BayesLoop(OmnidirectionalLoop):
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
@@ -145,6 +148,39 @@ class BayesLoop(OmnidirectionalLoop):
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]
@@ -160,7 +196,7 @@ class BayesLoop(OmnidirectionalLoop):
def _build_tasks(self) -> list[Task]:
tasks: list[Task] = []
for index, point in enumerate(self.ledger.points):
pending = self._station_pending(index)
pending = self._fused_channels_at(index)
if not pending:
continue
switches = len(pending) - (1 if self.client.state.channel in pending else 0)
@@ -176,6 +212,8 @@ class BayesLoop(OmnidirectionalLoop):
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
@@ -184,19 +222,155 @@ class BayesLoop(OmnidirectionalLoop):
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]:
"""开放路径排序:最近邻打底,2-opt 改进(终点自由,不回起点)。"""
"""开放路径排序:小任务表用精确 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 = []
@@ -263,18 +437,26 @@ class BayesLoop(OmnidirectionalLoop):
self._execute_locate(task.channel, task.position)
def _execute_station(self, index: int) -> None:
"""单站批量扫频:只测待测频道(跳测规则已过滤),扫完即返回重排。"""
"""单站批量扫频:只测待测频道(跳测规则已过滤),扫完即返回重排。
P0-1:本站同时测量"已发现未定证"频道,使第二/第三条方位线在巡回
途中顺路取得;证书一旦成立即就地清除,不额外绕行。
"""
point = self.ledger.points[index]
pending = self._station_pending(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(pending)
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)
@@ -291,6 +473,25 @@ class BayesLoop(OmnidirectionalLoop):
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:
"""单步定位测量:执行后立即重排(滚动时域),收敛则顺路清除。"""
@@ -325,7 +526,13 @@ class BayesLoop(OmnidirectionalLoop):
"""belief(凸证书)与 PF(幸存集)同步更新;PF 空即证缺席。"""
self.belief.update_measure(channel, position, outcome.result,
outcome.svd_deg)
self._measured_points.setdefault(channel, set()).add(_point_key(position))
# 同一 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):
@@ -335,10 +542,30 @@ class BayesLoop(OmnidirectionalLoop):
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 幸存集更紧时用网格证书补位。"""
@@ -369,8 +596,13 @@ class BayesLoop(OmnidirectionalLoop):
return ClearCertificate(cert.channel, target, cert.radius_m+shift, cert.limit_m, True)
# ------------------------------------------------------------ 信息增益选点
def _select_point(self, channel: int) -> Vec2 | None:
"""定位测量点:期望 log 幸存占比 + λ·移动时间 最小的候选。"""
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:
@@ -378,6 +610,7 @@ class BayesLoop(OmnidirectionalLoop):
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:
@@ -387,15 +620,19 @@ class BayesLoop(OmnidirectionalLoop):
search.candidate_angles)
# 以实际来向生成对称交会候选,避免固定全局角度网格的旋转偏差。
geometric_center = region.mec()[0]
base_angle = math.atan2(self.client.state.position[1]-geometric_center[1],
self.client.state.position[0]-geometric_center[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)))
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:
@@ -403,27 +640,93 @@ class BayesLoop(OmnidirectionalLoop):
candidates.extend(candidate_points(center, ring, (1.0,),
search.candidate_angles))
candidates.append(tuple(center))
candidates = [c for c in candidates if _point_key(c) not in measured]
candidates.append(tuple(position))
candidates = [tuple(c) for c in candidates
if _point_key(c) not in measured]
if not candidates:
return None
subsample = pf.subsample(bayes.score_subsample)
# 采样一次转成数组(numpy 可用时),使候选评分走向量路径;
# numpy 缺失时 as_sample_array 原样返回列表,语义不变。
subsample = as_sample_array(pf.subsample(bayes.score_subsample))
stride = max(len(subsample) // bayes.mc_representatives, 1)
representatives = [subsample[k]
for k in range(0, len(subsample), stride)]
position = self.client.state.position
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))
if bayes.score_mode == "entropy":
return (pf.expected_survival(q, subsample, representatives)
+ move)
return (pf.expected_posterior_radius(q, subsample, representatives)
+ move)
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 选点,失败退凸选点。"""
+4 -1
View File
@@ -20,6 +20,7 @@ import math
from ..belief import ChannelStatus
from ..config import ProblemConfig
from ..coverage import CoverageLedger
from ..guarantee import seven_plan
from ..protocol.rules import NEAR
from . import OmnidirectionalLoop, clear_certificate
@@ -55,7 +56,9 @@ class SevenImmediate(OmnidirectionalLoop):
ring_radius_m: float = RING_RADIUS_M):
super().__init__(client, config)
validate_seven_coverage(ring_radius_m, config)
self.ledger = CoverageLedger(config=config, points=seven_points(ring_radius_m))
self.coverage_plan = seven_plan(ring_radius_m)
self.ledger = CoverageLedger(config=config,
points=list(self.coverage_plan.points))
class SevenBatch(SevenImmediate):
+254
View File
@@ -0,0 +1,254 @@
"""图论/动态规划工具箱:开放路径排序、下界与路由度量(S1/S5)。
设计原则(docs/第三四问图论与动态规划优化方案.md §1、§3.2、§3.4):
- **开放路径**(起点固定为原点、终点自由)是本题的真实图问题:一次作业回合
= 依次停靠若干作业点,代价 = Σ 移动时间 + Σ 动作耗时;
- 小规模(n ≤ ``EXACT_LIMIT``)直接用 Held-Karp 精确 DP,规模更大时用
最近邻打底 + 2-opt / Or-opt 改进,并始终给出 **MST 下界**用于在线后悔值审计;
- 所有函数只做"排序与度量",不承担任何安全责任,也不改变覆盖/清除语义。
约定:边权为**欧氏距离**(米);时间折算由调用方按 ``speed_mps`` 处理。
2-opt / Or-opt 用 O(1) 增量代价评估,使 n ≈ 10² 量级仍可在毫秒内完成。
"""
from __future__ import annotations
import math
from .config import Vec2
EXACT_LIMIT = 13 # Held-Karp 状态数 2^n·n:n=13 → 106 496,纯 Python 可接受
OR_OPT_LIMIT = 60 # 作业点数超过此值时跳过高阶 Or-opt(见 improve_open_path)
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 or_opt_open(order: list[Vec2], start: Vec2 = (0.0, 0.0),
max_passes: int = 4, max_segment: int = 3) -> list[Vec2]:
"""Or-opt:把长度 ≤ max_segment 的片段搬家(开放路径,增量代价 O(1))。"""
best = list(order)
n = len(best)
if n < 4:
return best
for _ in range(max_passes):
improved = False
for length in range(1, min(max_segment, n - 1) + 1):
for i in range(0, n - length + 1):
segment = best[i:i + length]
rest = best[:i] + best[i + length:]
for j in range(len(rest) + 1):
if j == i:
continue
candidate = rest[:j] + segment + rest[j:]
if _segment_move_gain(best, i, length, j, start) > 1e-9:
best = candidate
improved = True
break
if improved:
break
if improved:
break
if not improved:
break
return best
def _segment_move_gain(order: list[Vec2], i: int, length: int, j: int,
start: Vec2) -> float:
"""把 order[i:i+length] 插到去掉该段后的第 j 位所省下的长度(正=更省)。"""
rest = order[:i] + order[i + length:]
segment = order[i:i + length]
def length_of(seq: list[Vec2]) -> float:
return path_length(seq, start)
return length_of(order) - length_of(rest[:j] + segment + rest[j:])
def improve_open_path(points: list[Vec2], start: Vec2 = (0.0, 0.0),
max_passes: int = 8) -> list[Vec2]:
"""最近邻 + 2-opt + Or-opt 的开放路径改进(大规模兜底)。
Or-opt 的搬移评估是 O(n³),作业点数超过 :data:`OR_OPT_LIMIT` 时只做
O(n²) 的 2-opt(度量口径下 Or-opt 的相对收益远小于其墙钟代价)。
"""
order = nearest_neighbor_order(points, start)
order = two_opt_open(order, start, max_passes)
if len(order) > OR_OPT_LIMIT:
return order
return or_opt_open(order, start, max_passes)
def held_karp_open(points: list[Vec2], start: Vec2 = (0.0, 0.0)) -> list[Vec2]:
"""Held-Karp 精确开放路径(起点固定、终点自由)。
状态 ``dp[mask][j]`` = 访问集合 mask、当前停在 j 的最小长度;
规模 2^n·n,仅用于 n ≤ :data:`EXACT_LIMIT`。返回最优访问顺序。
"""
n = len(points)
if n == 0:
return []
if n == 1:
return list(points)
dist = [[math.dist(points[i], points[j]) for j in range(n)] for i in range(n)]
from_start = [math.dist(start, points[j]) for j in range(n)]
full = (1 << n) - 1
inf = float("inf")
dp: list[list[float]] = [[inf] * n for _ in range(1 << n)]
parent: list[list[int]] = [[-1] * n for _ in range(1 << n)]
for j in range(n):
dp[1 << j][j] = from_start[j]
for mask in range(1 << n):
row = dp[mask]
for j in range(n):
current = row[j]
if current == inf or not (mask >> j) & 1:
continue
rest = full ^ mask
while rest:
bit = rest & -rest
k = bit.bit_length() - 1
rest ^= bit
candidate = current + dist[j][k]
nxt = mask | bit
if candidate < dp[nxt][k]:
dp[nxt][k] = candidate
parent[nxt][k] = j
best_end = min(range(n), key=lambda j: dp[full][j])
order: list[Vec2] = []
mask, j = full, best_end
while j != -1:
order.append(points[j])
previous = parent[mask][j]
mask ^= 1 << j
j = previous
order.reverse()
return order
def optimal_open_path(points: list[Vec2], start: Vec2 = (0.0, 0.0),
exact_limit: int = EXACT_LIMIT) -> list[Vec2]:
"""离线最优(或近优)开放路径:小规模精确,大规模 2-opt + Or-opt。"""
unique = _dedupe(points)
if len(unique) <= exact_limit:
return held_karp_open(unique, start)
return improve_open_path(unique, start)
def mst_weight(points: list[Vec2], start: Vec2 | None = (0.0, 0.0)) -> float:
"""最小生成树权重(Prim,O(n²));含 start 时给出开放路径最优值的下界。
开放路径删除其最后一条边即得一条生成树,故 MST 权重 ≤ 最优路径长度。
"""
nodes = ([start] if start is not None else []) + list(points)
n = len(nodes)
if n <= 1:
return 0.0
in_tree = [False] * n
best = [float("inf")] * n
best[0] = 0.0
total = 0.0
for _ in range(n):
u, u_cost = -1, float("inf")
for i in range(n):
if not in_tree[i] and best[i] < u_cost:
u, u_cost = i, best[i]
if u < 0:
break
in_tree[u] = True
total += u_cost
ux, uy = nodes[u]
for v in range(n):
if in_tree[v]:
continue
distance = math.hypot(nodes[v][0] - ux, nodes[v][1] - uy)
if distance < best[v]:
best[v] = distance
return total
def _dedupe(points, cell_m: float = 0.5) -> list[Vec2]:
"""按 ``cell_m`` 网格去重(同一物理点只算一次作业)。"""
seen: set[tuple[int, int]] = set()
unique: list[Vec2] = []
for point in points:
key = (int(round(point[0] / cell_m)), int(round(point[1] / cell_m)))
if key in seen:
continue
seen.add(key)
unique.append((float(point[0]), float(point[1])))
return unique
def route_metrics(service_points, actual_move_m: float,
speed_mps: float = 5.0) -> dict:
"""路由度量:离线最优开放回路、MST 下界与相对 gap(S1 必报列)。"""
unique = _dedupe(service_points)
offline = path_length(optimal_open_path(unique)) if unique else 0.0
lower = mst_weight(unique)
gap = ((actual_move_m - offline) / actual_move_m) if actual_move_m > 0 else 0.0
return {
"distinct_service_points": len(unique),
"offline_opt_open_path_m": round(offline, 1),
"offline_opt_open_path_s": round(offline / speed_mps, 1),
"route_lower_bound_s": round(lower / speed_mps, 1),
"route_gap_vs_offline": round(gap, 4),
}