"""P5 定向扩展端到端验收：朝外边界源不漏、真源不误删、格数与预算检查。 对应 docs/B题建模分析与实施计划.md P5 行验收门槛（自建部分）： - 混合源（全向+定向）场景全部清除并正常退出；16 源触发数量上界停止； - 朝向背离圆心的近边界定向源仍被 49 点覆盖发现并清除； - no_signal 联合约束基线不删除空间候选（真源保留性）； - 后备清除 <=20m 覆盖无空洞、格数有上界、格数/现实预算不足时不盲扫； - 虚拟时刻逐步复算与服务器响应一致。 """ import math import unittest from dataclasses import replace from itertools import count from mcm26b.belief import BeliefState, CandidateRegion from mcm26b.config import ProblemConfig from mcm26b.evaluation import recompute_virtual_time from mcm26b.planner import (DirectionalExtension, fallback_cell_bound, fallback_clear_cells, safe_disk_radius) from mcm26b.protocol.client import SimulatorClient from mcm26b.protocol.rules import DIRECTION, NO_SIGNAL from mcm26b.sim import LocalSimTransport, LocalSimulator, Source from mcm26b.sim.scenes import mixed_scene CONFIG = replace(ProblemConfig(), search=replace(ProblemConfig().search, bearing_rounding_margin_deg=0.005)) def _run_scene(sources, config=CONFIG, clock=None): simulator = LocalSimulator(sources, robot_id="T001", config=config) client = SimulatorClient("T001", config=config, transport=LocalSimTransport(simulator), **({"clock": clock} if clock else {})) loop = DirectionalExtension(client, config) summary = loop.run() report = recompute_virtual_time(client.log.action_events()) return summary, report, simulator, client, loop class FallbackClearCellsTests(unittest.TestCase): def test_cells_cover_every_polygon_witness(self): """覆盖无空洞：多边形边界采样点到最近格心 <= 10*sqrt(2)。""" region = CandidateRegion.disk((0.0, 0.0), 200.0, 192) cells = fallback_clear_cells(region.polygon, CONFIG) self.assertTrue(cells) limit = 10.0 * 2 ** 0.5 + 1e-6 for witness in region.witnesses(40.0): nearest = min(math.dist(witness, cell) for cell in cells) self.assertLessEqual(nearest, limit, f"格心覆盖在 {witness} 附近出现空洞") def test_grid_line_polygon_keeps_both_sides(self): """多边形横跨格线时两侧格子都保留（宁可多测）。""" square = [(10.0, -10.0), (30.0, -10.0), (30.0, 10.0), (10.0, 10.0)] cells = fallback_clear_cells(square, CONFIG) columns = {cell[0] for cell in cells} self.assertIn(10.0, columns) # 格线 x=20 左侧 self.assertIn(30.0, columns) # 右侧 def test_degenerate_polygon_single_cell(self): """点与线段候选集：覆盖格数最小化且无空洞。""" point = [(500.0, 500.0)] self.assertEqual(len(fallback_clear_cells(point, CONFIG)), 1) segment = [(490.0, 500.0), (510.0, 500.0)] cells = fallback_clear_cells(segment, CONFIG) self.assertTrue(cells) limit = 10.0 * 2 ** 0.5 + 1e-6 for step in range(21): sample = (490.0 + step, 500.0) nearest = min(math.dist(sample, cell) for cell in cells) self.assertLessEqual(nearest, limit) def test_count_within_plan_bound(self): region = CandidateRegion.disk((0.0, 0.0), 200.0, 192) xs = [p[0] for p in region.polygon] ys = [p[1] for p in region.polygon] bound = fallback_cell_bound(max(xs) - min(xs), max(ys) - min(ys), CONFIG) cells = fallback_clear_cells(region.polygon, CONFIG) self.assertLessEqual(len(cells), bound) class MixedLoopAcceptanceTests(unittest.TestCase): def test_mixed_ten_sources_all_cleared(self): sources = mixed_scene(seed=5, count=10, directional=3) summary, report, simulator, _, loop = _run_scene(sources) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(summary["stop_reason"], "all_resolved") self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertEqual(summary["unresolved_channels"], []) self.assertEqual(summary["anomalies"], [], "守恒性或证书不应被打破") self.assertEqual(report["mismatches"], []) self.assertEqual(summary["virtual_recompute_mismatches"], 0) self.assertEqual(sorted(simulator.cleared), truth) self.assertEqual(summary["exit_reason"], "user_exit") # 观测到的定向频道必须来自真定向源（观测推理不误标） truth_directional = sorted(s.channel for s in sources if s.directional) self.assertTrue(summary["directional_observed_channels"]) self.assertTrue(set(summary["directional_observed_channels"]) <= set(truth_directional)) def test_mixed_sixteen_sources_stop_by_count_bound(self): sources = mixed_scene(seed=11, count=16, directional=6) summary, report, simulator, _, _ = _run_scene(sources) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"]) self.assertEqual(summary["stop_reason"], "sixteen_cleared") self.assertEqual(sorted(summary["cleared_channels"]), truth) self.assertEqual(sorted(simulator.cleared), truth) self.assertEqual(report["mismatches"], []) def test_outward_boundary_directional_source_not_missed(self): """朝向背离圆心、r_eff=1000 的近边界源：覆盖保证仍发现并清除。""" source = Source(channel=7, x=1790.0, y=0.0, r_eff_m=1000.0, directional=True, heading_deg=0.0) summary, _, simulator, _, _ = _run_scene([source]) self.assertTrue(summary["completed"]) self.assertEqual(summary["cleared_channels"], [7]) self.assertEqual(sorted(simulator.cleared), [7]) self.assertEqual(summary["directional_observed_channels"], [7]) def test_mixed_batch_seeds_never_degrade(self): """固定批量种子回归：任何种子下不得全清失败或破坏守恒性。""" for seed in range(5, 10): sources = mixed_scene(seed=seed, count=10, directional=4) summary, report, simulator, _, _ = _run_scene(sources) truth = sorted(s.channel for s in sources) self.assertTrue(summary["completed"], f"seed={seed}") self.assertEqual(sorted(summary["cleared_channels"]), truth, f"seed={seed}") self.assertEqual(summary["anomalies"], [], f"seed={seed}") self.assertEqual(report["mismatches"], [], f"seed={seed}") class JointConstraintBaselineTests(unittest.TestCase): def test_no_signal_keeps_candidates_and_true_source(self): """联合约束基线：保证接收盘内 no_signal 不删空间候选（真源不误删）。""" belief = BeliefState(CONFIG) belief.update_measure(7, (0.0, 0.0), DIRECTION, 30.0) region = belief.candidate_set(7) snapshot = list(region.polygon) self.assertGreater(safe_disk_radius(region, CONFIG), 0.0) center, _ = region.mec() belief.update_measure(7, tuple(center), NO_SIGNAL, None) self.assertEqual(region.polygon, snapshot, "no_signal 不得收缩候选集") # 候选集中任取的代表点（含生成读数的角域内部点）仍被保留 for witness in region.witnesses(300.0): self.assertTrue(region.contains(witness, tol_m=0.05)) def test_directional_source_retained_through_blocked_probes(self): """端到端保留性：定向源被遮挡若干次后仍被清除，不因 no_signal 误删。""" source = Source(channel=3, x=600.0, y=800.0, r_eff_m=1500.0, directional=True, heading_deg=200.0) summary, _, simulator, _, _ = _run_scene([source]) self.assertTrue(summary["completed"]) self.assertEqual(summary["cleared_channels"], [3]) self.assertEqual(sorted(simulator.cleared), [3]) class FallbackDisciplineTests(unittest.TestCase): def _stall_config(self, **search_updates): """迫使定位立即停滞进入后备清除的最小配置。""" return replace(CONFIG, search=replace(CONFIG.search, max_localize_steps=1, max_blocked_probes=0, **search_updates)) def test_cell_cap_skips_instead_of_blind_scan(self): """格数超硬顶：后备清除全部记录 skipped_cell_cap，频道保持未解决。""" config = self._stall_config(fallback_max_cells=5) sources = mixed_scene(seed=5, count=6, directional=2) summary, report, _, _, _ = _run_scene(sources, config=config) self.assertFalse(summary["completed"]) self.assertEqual(summary["stop_reason"], "unresolved_channels") self.assertTrue(summary["fallback_clears"]) self.assertTrue(all(entry["outcome"] == "skipped_cell_cap" for entry in summary["fallback_clears"])) self.assertEqual(report["mismatches"], []) def test_real_budget_check_skips_fallback(self): """现实预算核对：N*p95 + 退出余量超过剩余时不启动后备清除。""" config = replace(self._stall_config(), session=replace(CONFIG.session, real_rtt_p95_s=600.0)) sources = mixed_scene(seed=6, count=6, directional=2) summary, _, _, _, _ = _run_scene(sources, config=config) self.assertFalse(summary["completed"]) self.assertTrue(summary["fallback_clears"]) self.assertTrue(all(entry["outcome"] == "skipped_budget" for entry in summary["fallback_clears"])) self.assertTrue(summary["unresolved_channels"]) def test_fallback_clear_succeeds_on_bounded_candidate(self): """后备清除正路径：有界候选集逐格 /clear 命中真源。""" source = Source(channel=7, x=500.0, y=500.0, r_eff_m=1500.0, directional=True, heading_deg=180.0) simulator = LocalSimulator([source], robot_id="T001", config=CONFIG) client = SimulatorClient("T001", config=CONFIG, transport=LocalSimTransport(simulator)) loop = DirectionalExtension(client, CONFIG) client.enter() region = CandidateRegion.disk((500.0, 500.0), 30.0, 192) self.assertTrue(loop._fallback_clear(7, region)) self.assertEqual(loop.cleared, [7]) self.assertEqual(simulator.cleared, {7}) entry = loop.fallback_log[0] self.assertEqual(entry["outcome"], "cleared") self.assertLessEqual(entry["attempts"], entry["cells"]) cells = fallback_clear_cells(region.polygon, CONFIG) self.assertLessEqual(len(cells), CONFIG.search.fallback_max_cells) def test_fallback_exhaustion_records_anomaly_without_deletion(self): """后备全失败（空域）：守恒性异常留痕，且不删除空间候选。""" simulator = LocalSimulator([], robot_id="T001", config=CONFIG) client = SimulatorClient("T001", config=CONFIG, transport=LocalSimTransport(simulator)) loop = DirectionalExtension(client, CONFIG) client.enter() region = CandidateRegion.disk((0.0, 0.0), 30.0, 192) snapshot = list(region.polygon) self.assertFalse(loop._fallback_clear(7, region)) entry = loop.fallback_log[0] self.assertEqual(entry["outcome"], "exhausted") kinds = [item["kind"] for item in loop.anomalies] self.assertIn("fallback_exhausted_without_success", kinds) self.assertEqual(region.polygon, snapshot, "清除失败不得删除候选") class RealBudgetExitTests(unittest.TestCase): def test_budget_exhaustion_exits_incomplete(self): """混合场景现实预算耗尽：立即退出且记为未证明完成。""" ticks = count(0.0, 100.0) # 每次查询前进 100s，约 12 个动作内耗尽预算 sources = mixed_scene(seed=7, count=8, directional=2) summary, _, _, _, _ = _run_scene(sources, config=CONFIG, clock=lambda: next(ticks)) self.assertFalse(summary["completed"]) self.assertEqual(summary["stop_reason"], "real_budget_exhausted") self.assertEqual(summary["exit_reason"], "user_exit") if __name__ == "__main__": unittest.main()