sloray/mcm-26-b
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'q3-bayes' of https://git.simba-robotics.com/HelixCopex/mcm-26-b into q3-bayes

Browse Source
This commit is contained in:
Horldsence
2026-09-11 11:37:22 +08:00
parent a18a139a1d fd48064000
commit 2bc4c7ed82
218 changed files with 280849 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mcm26b/config.py
+5 -9
View File
@@ -113,8 +113,8 @@ class BayesConfig:
particle_spacing_m: float = 14.65 # 网格间距；Voronoi 裕量 ≈ 10.35m；
# 全域约 4.7 万格/频道，查询走粗格索引
# ---- 信息增益评分 [工程，Optuna TPE 调参回填 2026-09-11] ----
# 全量消融（14 场景，对齐找全口径）：均值 3650s、配对降幅 31.0%、
# 14/14 全清零异常；调参记录见 output/q3/bayes-tuning/。
# 原调参记录见 output/q3/bayes-tuning/；当前代码的配对验收记录见
# output/q3/optimization-20260911/，随机分布与信息评分仅为工程假设。
score_mode: str = "entropy" # entropy=期望 log 幸存占比（信息增益）|
# radius=期望后验半径（直指证书）
score_subsample: int = 800 # 期望幸存熵估计的粒子子采样规模
@@ -123,18 +123,14 @@ class BayesConfig:
# entropy 口径下信息项为 O(1) 对数值，λ须极小
wide_rings: tuple = (1.0, 1.5, 2.0) # 无保证接收盘时的探索环（× PF MEC 半径）
# ---- 站点与覆盖 [工程] ----
station_ring_radius_m: float = 1200.0 # 七点环半径 [调参后保留 1200 而非
# Optuna 的 1159.7]：环 1200 时七站
# 排除盘对外圈格心（|c| ≤ ω+裕量）
# 的最大距离 977.3m ≤ 1000 − 裕量
# 10.35m，PF 可独立证明缺席；
# 1159.7 虽均值再快 ~60s 但外圈差
# 3.3m，缺席只能靠覆盖账本兜底
station_ring_radius_m: float = 1159.7 # 七点环半径；构造时检验全向覆盖解析上界
# ---- 任务排序 [工程] ----
two_opt_enabled: bool = True # 开放路径 2-opt 改进（最近邻打底）
two_opt_passes: int = 4 # 2-opt 最大改进轮数
# ---- PF 清除证书 [工程] ----
grid_cert_enabled: bool = True # 允许用幸存集 MEC 颁发清除证书（凸证书之外）
opportunistic_measure: bool = True # 已停靠位置顺路交会；仍逐频道计费
opportunistic_radius_ratio: float = 0.65 # 预测后验半径至少减少35%才补测
@dataclass(frozen=True)
src/mcm26b/planner/__init__.py
+51 -13
View File
@@ -11,8 +11,8 @@ P3 第二检测点（验收：保证收到信号；Q 空时退化；near 分支
P4 全向闭环（验收：10/16/0 源场景、边界源、预算耗尽与故障注入）：
发现（coverage.CoverageLedger）→ 定位（belief 收敛）→ 清除 → 持续复扫
→ /exit。全部频道"已清除或证明不存在"后按 persistent_search（默认开
[用户决策 2026-09-11]）只复扫未清除频道，扫到限时为止（清除数达 16 /
→ /exit。全部频道"已清除或证明不存在"后默认立即退出；显式开启
persistent_search时只复扫未清除频道，扫到限时为止（清除数达 16 /
实机预算耗尽 / 虚拟安全余量不足）。completed 表示任务证明完成（结束
时每频道均 CLEARED/ABSENT 且无中断）；accounted_virtual_time_s 记录
首次证明找全的虚拟时刻，与"复扫到限时才停"相互独立。
@@ -199,7 +199,7 @@ class OmnidirectionalLoop:
串行驱动 SimulatorClient：覆盖阶段在 49 点网格上对每个 UNKNOWN 频道
检测；发现 direction 即进入该频道的定位子循环（选点→测量→收敛→
证书清除），near 直接清除。全部频道"已清除或证明不存在"后按
``search.persistent_search``（默认开 [用户决策 2026-09-11]）只复扫
``search.persistent_search``（默认关闭）可选择只复扫
未清除频道，直到清除数达 16、实机预算耗尽或虚拟安全余量不足。
stop_reason 记录循环结束原因；``completed`` 表示任务证明完成
（结束时每频道均 CLEARED/ABSENT 且无中断），与"复扫到限时才停"
@@ -598,9 +598,43 @@ class DirectionalExtension(OmnidirectionalLoop):
config: ProblemConfig = ProblemConfig()):
super().__init__(client, config)
self.directional_observed: list[int] = [] # 观测到定向性的频道
self.blocked_probes: dict[int, Vec2] = {} # channel -> 被遮挡的探点
self.blocked_probes: dict[int, list[Vec2]] = {}
self.fallback_log: list[dict] = [] # 后备清除逐次留痕
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.belief.update_measure(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:
"""定位子循环：选点测量收敛候选集，拿到清除证书后清除。
@@ -627,19 +661,17 @@ class DirectionalExtension(OmnidirectionalLoop):
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.belief.update_measure(channel, point, outcome.result,
outcome.svd_deg)
self._may_clear.add(channel)
if outcome.result == NO_SIGNAL \
and safe_disk_radius(region, self.config) >= 0:
if channel not in self.directional_observed:
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 # 被遮挡探点不消耗定位步数，由遮挡预算限流
if outcome.result == NO_SIGNAL:
self._anomaly("no_signal_outside_safe_disk", channel,
position=point)
continue # 盘外无信号也可能是距离不足，不据此断言定向
else:
self.blocked_probes.pop(channel, None)
steps += 1
@@ -709,9 +741,15 @@ class DirectionalExtension(OmnidirectionalLoop):
+ session.reserve_exit_s > remaining:
entry["outcome"] = "skipped_budget"
return False
position = self.client.state.position
attempts = 0
for cell in sorted(cells, key=lambda c: math.dist(position, c)):
remaining_cells = list(cells)
while remaining_cells:
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:
src/mcm26b/planner/bayes_loop.py
+87 -8
View File
@@ -21,7 +21,7 @@ from __future__ import annotations
import math
from dataclasses import dataclass
from ..belief import ChannelStatus
from ..belief import CandidateRegion, ChannelStatus
from ..config import ProblemConfig, Vec2
from ..coverage import CoverageLedger
from ..particles import ChannelParticles
@@ -30,7 +30,7 @@ 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
from .seven import seven_points, validate_seven_coverage
@dataclass(frozen=True)
@@ -42,6 +42,7 @@ class Task:
service_s: float
channel: int | None = None
station_index: int | None = None
station_channels: tuple[int, ...] = ()
def _point_key(point: Vec2) -> tuple[int, int]:
@@ -55,6 +56,7 @@ class BayesLoop(OmnidirectionalLoop):
def __init__(self, client: SimulatorClient,
config: ProblemConfig = ProblemConfig()):
super().__init__(client, config)
validate_seven_coverage(config.bayes.station_ring_radius_m, config)
self.ledger = CoverageLedger(
config, points=seven_points(config.bayes.station_ring_radius_m))
self.particles = {channel: ChannelParticles(config)
@@ -66,6 +68,7 @@ class BayesLoop(OmnidirectionalLoop):
self.skipped_measures = 0 # 跳测规则省下的检测次数
self.no_shrink_measures = 0 # 未收缩幸存集的定位检测（诊断用）
self.pf_absent_channels: list[int] = [] # 由幸存集为空证明缺席的频道
self.opportunistic_measures = 0
# ------------------------------------------------------------ 主循环
def _coverage_phase(self) -> None:
@@ -78,6 +81,8 @@ class BayesLoop(OmnidirectionalLoop):
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
@@ -85,6 +90,44 @@ class BayesLoop(OmnidirectionalLoop):
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)
if max(math.dist(point, v) for v in region.polygon) > 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"且未证缺席。"""
@@ -122,7 +165,7 @@ class BayesLoop(OmnidirectionalLoop):
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_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 \
@@ -143,9 +186,10 @@ class BayesLoop(OmnidirectionalLoop):
def _order_tasks(self, tasks: list[Task]) -> list[Task]:
"""开放路径排序：最近邻打底，2-opt 改进（终点自由，不回起点）。"""
if len(tasks) <= 2:
return sorted(tasks, key=lambda t: (
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)
order = self._nearest_order(tasks)
if not self.config.bayes.two_opt_enabled:
return order
@@ -167,9 +211,20 @@ class BayesLoop(OmnidirectionalLoop):
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)
total += task.service_s
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: (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
@@ -287,7 +342,7 @@ class BayesLoop(OmnidirectionalLoop):
cert = clear_certificate(region, channel, self.config) \
if region is not None else None
if cert is not None and cert.certified:
return cert
return self._near_clear_target(cert)
if self.config.bayes.grid_cert_enabled:
pf = self.particles[channel]
if len(pf):
@@ -295,9 +350,19 @@ class BayesLoop(OmnidirectionalLoop):
limit = (self.config.field.clear_radius_m - pf.margin_m
- self.config.search.clear_margin_m)
if radius <= limit:
return ClearCertificate(channel, center, radius, limit, True)
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 _select_point(self, channel: int) -> Vec2 | None:
"""定位测量点：期望 log 幸存占比 + λ·移动时间 最小的候选。"""
@@ -315,6 +380,17 @@ class BayesLoop(OmnidirectionalLoop):
candidates = candidate_points(tuple(region.mec()[0]), safe,
search.candidate_rings,
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])
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)))
else:
# 宽候选集：PF MEC 探索环（负观测挖洞后幸存集的包围圆）
for fraction in bayes.wide_rings:
@@ -332,7 +408,9 @@ class BayesLoop(OmnidirectionalLoop):
position = self.client.state.position
def score(q: Vec2) -> float:
move = bayes.lambda_move_s * move_time(position, q, self.config.cost)
# 一次定位后仍需接近源才能清除；仅计到检测点的移动会偏好绕行。
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)
@@ -419,6 +497,7 @@ class BayesLoop(OmnidirectionalLoop):
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)
src/mcm26b/planner/seven.py
+19 -2
View File
@@ -2,8 +2,8 @@
从 scripts/benchmark_q3_search.py 提升为库代码（A2 基线，官方演练
12/12 全清、虚拟 5232s）。覆盖几何：原点 + 半径 RING_RADIUS_M 环上
6 等分点；切换最近站的半径 a/sqrt(3)，域内任意点到最近站距离的解析
上界 max(R/sqrt(3), sqrt(Rom³/(sqrt(3)·Rring)+...)) < 1000m [题面
6 等分点；域半径R、环半径r时，最近站距离的解析
上界 max(r/√3, sqrt(R²+r²−√3Rr)) < 1000m [题面
r_eff 下界]，故任意合法源必被至少一站检出——缺席证明只需该频道
7 站全 no_signal（CoverageLedger.coverage_complete）。
@@ -32,12 +32,29 @@ def seven_points(ring_radius_m: float = RING_RADIUS_M) -> list[tuple[float, floa
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
class SevenImmediate(OmnidirectionalLoop):
"""七点覆盖 + 站内即时定位（基类覆盖阶段逻辑不变，仅换站集）。"""
def __init__(self, client, config: ProblemConfig = ProblemConfig(),
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))