92 lines
4.4 KiB
Python
92 lines
4.4 KiB
Python
"""P1-4 安全阀与退出可解释性验收。
|
||
|
||
覆盖 论文“实验结果与分析”一节 §P1-4：
|
||
- summary 必须显式给出完成依据 completion_basis（数量上界 vs 逐频道证明）；
|
||
- 逐频道缺席必须带证据来源 absence_evidence（pf_empty / coverage）；
|
||
- 重复复扫计数 redundant_rescan_measures 是历史 98.84% 空复扫问题的回归指标；
|
||
- 虚拟安全余量不小于单动作耗时上界（配置被调小时仍成立）。
|
||
"""
|
||
|
||
import unittest
|
||
from dataclasses import replace
|
||
|
||
from mcm26b.config import ProblemConfig
|
||
from mcm26b.planner import DirectionalExtension, OmnidirectionalLoop
|
||
from mcm26b.planner.bayes_loop import BayesLoop
|
||
from mcm26b.protocol.client import SimulatorClient
|
||
from mcm26b.harness import LocalSimTransport, LocalSimulator
|
||
from mcm26b.harness import mixed_scene, random_omni_scene
|
||
|
||
CONFIG = replace(ProblemConfig(),
|
||
search=replace(ProblemConfig().search,
|
||
bearing_rounding_margin_deg=0.005,
|
||
persistent_search=False))
|
||
|
||
|
||
def _run(loop_cls, sources, config=CONFIG):
|
||
simulator = LocalSimulator(sources, robot_id="T-SAFE", config=config)
|
||
client = SimulatorClient("T-SAFE", config=config,
|
||
transport=LocalSimTransport(simulator))
|
||
return loop_cls(client, config).run(), simulator
|
||
|
||
|
||
class SafetyValveTests(unittest.TestCase):
|
||
def test_summary_completion_and_absence_evidence(self):
|
||
sources = random_omni_scene(seed=5, count=10, config=CONFIG)
|
||
summary, _ = _run(BayesLoop, sources)
|
||
truth = sorted(s.channel for s in sources)
|
||
self.assertEqual(sorted(summary["cleared_channels"]), truth)
|
||
self.assertEqual(summary["completion_basis"], "per_channel_proof")
|
||
self.assertEqual(sorted(summary["absent_channels"]),
|
||
sorted(set(range(1, 21)) - set(truth)))
|
||
evidence = summary["absence_evidence"]
|
||
self.assertEqual(sorted(int(c) for c in evidence),
|
||
sorted(summary["absent_channels"]))
|
||
self.assertTrue(set(evidence.values()) <= {"pf_empty", "coverage"})
|
||
self.assertIsInstance(summary["redundant_rescan_measures"], int)
|
||
self.assertGreaterEqual(summary["virtual_reserve_s"], 10_000.0)
|
||
|
||
def test_redundant_rescan_measure_is_counted(self):
|
||
"""同一测点重复检测必须被计数（正常模式应为 0）。"""
|
||
sources = random_omni_scene(seed=7, count=10, config=CONFIG)
|
||
summary, _ = _run(BayesLoop, sources)
|
||
self.assertEqual(summary["redundant_rescan_measures"], 0)
|
||
|
||
def test_count_cap_is_distinguished_from_proof(self):
|
||
"""16 源时按数量上界停止，completion_basis 不得写成逐频道证明。"""
|
||
sources = random_omni_scene(seed=11, count=16, config=CONFIG)
|
||
summary, _ = _run(BayesLoop, sources)
|
||
self.assertGreaterEqual(len(summary["cleared_channels"]), 16)
|
||
self.assertEqual(summary["completion_basis"], "count_cap")
|
||
|
||
def test_q4_completion_basis_present(self):
|
||
sources = mixed_scene(seed=3, count=10, directional=3, config=CONFIG)
|
||
summary, simulator = _run(DirectionalExtension, sources)
|
||
truth = sorted(s.channel for s in sources)
|
||
self.assertEqual(sorted(summary["cleared_channels"]), truth)
|
||
self.assertIn(summary["completion_basis"],
|
||
{"per_channel_proof", "count_cap"})
|
||
self.assertEqual(sorted(simulator.cleared), truth)
|
||
|
||
|
||
class VirtualReserveTests(unittest.TestCase):
|
||
def test_reserve_covers_single_action_bound(self):
|
||
"""配置余量被调小到 0 时，有效余量仍不得小于单动作上界。"""
|
||
tight = replace(ProblemConfig(),
|
||
search=replace(ProblemConfig().search,
|
||
virtual_reserve_s=0.0))
|
||
sources = random_omni_scene(seed=1, count=1, config=tight)
|
||
simulator = LocalSimulator(sources, robot_id="T-RES", config=tight)
|
||
client = SimulatorClient("T-RES", config=tight,
|
||
transport=LocalSimTransport(simulator))
|
||
loop = OmnidirectionalLoop(client, tight)
|
||
reach = max((x * x + y * y) ** 0.5 for x, y in loop.ledger.points)
|
||
bound = 2.0 * reach / tight.cost.speed_mps \
|
||
+ tight.cost.measure_s + tight.cost.switch_s
|
||
self.assertGreaterEqual(loop._min_virtual_reserve(), bound)
|
||
self.assertLess(loop._min_virtual_reserve(), 10_000.0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|