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

feat: 3

Browse Source
This commit is contained in:
Horldsence
2026-09-11 03:32:05 +08:00
parent 846b45e420
commit fc70d6edfd
14 changed files with 46032 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_loop_q3.py
+138 -10
View File
@@ -1,11 +1,16 @@
"""P4 全向闭环端到端验收：10/16/0 源、边界源、预算耗尽与故障注入。
对应 docs/B题建模分析与实施计划.md P4 行验收门槛（自建部分）：
- 10/16 源场景正确退出，全清或数量上界证明；
- 10/16 源场景全部清除；16 源触发数量上界停止；
- 近圆域边界源不漏搜、可清除；
- 持续复扫 [用户决策 2026-09-11]：找全后只复扫未清除频道直到虚拟上限；
- 每步计时（实机 + 比赛虚拟耗时分项）逐步留档可汇总；
- 虚拟时刻逐步复算与服务器响应一致；
- 现实预算耗尽退出记为未完成，不与成功混记；
- 现实预算耗尽（找全前）退出记为未完成，不与成功混记；
- 网络故障重试（幂等回放）不污染闭环结果。
SMALL_CAP 把虚拟限时压到 60k 秒：持续复扫在数轮内触及安全余量，
避免测试扫满 360k 秒上限（等价行为，仅缩短复扫轮数）。
"""
import unittest
@@ -24,16 +29,18 @@ from mcm26b.sim.scenes import random_omni_scene
CONFIG = replace(ProblemConfig(),
search=replace(ProblemConfig().search,
bearing_rounding_margin_deg=0.005))
SMALL_CAP = replace(CONFIG,
field=replace(CONFIG.field, max_virtual_duration_s=60_000.0))
def _run_scene(sources, fault_script=None, clock=None):
simulator = LocalSimulator(sources, robot_id="T001", config=CONFIG)
def _run_scene(sources, fault_script=None, clock=None, config=SMALL_CAP):
simulator = LocalSimulator(sources, robot_id="T001", config=config)
transport = LocalSimTransport(simulator)
if fault_script is not None:
transport = ScriptedFaultTransport(transport, fault_script)
kwargs = {"clock": clock} if clock is not None else {}
client = SimulatorClient("T001", config=CONFIG, transport=transport, **kwargs)
loop = OmnidirectionalLoop(client, CONFIG)
client = SimulatorClient("T001", config=config, transport=transport, **kwargs)
loop = OmnidirectionalLoop(client, config)
summary = loop.run()
report = recompute_virtual_time(client.log.action_events())
return summary, report, simulator
@@ -45,10 +52,16 @@ class LoopAcceptanceTests(unittest.TestCase):
summary, report, simulator = _run_scene(sources)
truth = sorted(s.channel for s in sources)
self.assertTrue(summary["completed"])
self.assertEqual(summary["stop_reason"], "all_resolved")
self.assertEqual(summary["stop_reason"], "virtual_time_cap")
self.assertEqual(sorted(summary["cleared_channels"]), truth)
self.assertEqual(len(summary["absent_channels"]), 10)
self.assertEqual(summary["unresolved_channels"], [])
self.assertIsNotNone(summary["accounted_virtual_time_s"],
"找全证明时刻必须留档")
self.assertLessEqual(summary["accounted_virtual_time_s"],
summary["virtual_time_s"])
self.assertGreater(summary["measures"], 49 * 20,
"找全后应继续复扫未清除频道（扫到限时为止）")
self.assertEqual(summary["anomalies"], [], "守恒性或证书不应被打破")
self.assertEqual(report["mismatches"], [])
self.assertEqual(summary["virtual_recompute_mismatches"], 0)
@@ -70,10 +83,12 @@ class LoopAcceptanceTests(unittest.TestCase):
def test_empty_field_proves_absence_everywhere(self):
summary, report, _ = _run_scene([])
self.assertTrue(summary["completed"])
self.assertEqual(summary["stop_reason"], "all_resolved")
self.assertEqual(summary["stop_reason"], "virtual_time_cap")
self.assertEqual(summary["cleared_channels"], [])
self.assertEqual(summary["absent_channels"], list(range(1, 21)))
self.assertEqual(summary["measures"], 49 * 20, "缺席证明要求全网格逐频道检测")
self.assertGreaterEqual(summary["measures"], 49 * 20,
"缺席证明要求全网格逐频道检测")
self.assertIsNotNone(summary["accounted_virtual_time_s"])
self.assertEqual(report["mismatches"], [])
def test_boundary_source_close_to_domain_edge(self):
@@ -91,7 +106,7 @@ class LoopAcceptanceTests(unittest.TestCase):
self.assertEqual(sorted(summary["cleared_channels"]), [20])
def test_real_budget_exhaustion_exits_cleanly(self):
"""现实预算耗尽：立即退出并记为未证明完成。"""
"""现实预算在找全前耗尽：立即退出并记为未证明完成。"""
ticks = count(0.0, 200.0) # 每次查询时钟前进 200s，迅速耗尽 1200s 预算
sources = random_omni_scene(seed=8, count=4)
summary, _, _ = _run_scene(sources, clock=lambda: next(ticks))
@@ -112,6 +127,99 @@ class LoopAcceptanceTests(unittest.TestCase):
self.assertEqual(sorted(simulator.cleared), truth)
class StepTimingTests(unittest.TestCase):
"""每步计时 [用户决策 2026-09-11]：实机耗时 + 比赛虚拟耗时分项。"""
def test_steps_recorded_with_virtual_breakdown(self):
sources = random_omni_scene(seed=5, count=2)
summary, report, _ = _run_scene(sources)
steps = summary["steps"]
self.assertEqual(steps[0]["action"], "enter")
self.assertEqual(steps[-1]["action"], "exit")
self.assertTrue(all(step["real_elapsed_ms"] >= 0 for step in steps))
first_measure = next(step for step in steps if step["action"] == "measure")
self.assertEqual(first_measure["step_virtual_s"],
first_measure["move_s"] + first_measure["switch_s"]
+ first_measure["action_s"])
totals = summary["step_totals"]
self.assertEqual(totals["count"], len(steps))
# 虚拟分项之和 = 虚拟总时长（比赛口径：移动+切频道+检测+清除）
self.assertAlmostEqual(
totals["virtual_move_s"] + totals["virtual_switch_s"]
+ totals["virtual_measure_action_s"] + totals["virtual_clear_action_s"],
totals["virtual_total_s"], delta=0.5)
self.assertAlmostEqual(totals["virtual_total_s"],
summary["virtual_time_s"], delta=0.5)
self.assertGreaterEqual(totals["real_elapsed_total_s"], 0.0)
# 清除成功一步 5s、未发现 3s：步级记录可复核
clear_step = next(step for step in steps if step["action"] == "clear")
self.assertIn(clear_step["action_s"], (3.0, 5.0))
self.assertEqual(clear_step["switch_s"], 0.0, "/clear 不切频道")
def test_avg_locate_clear_time_metric(self):
"""题面口径：平均定位清除时间 = 虚拟总时长 / 被清除个数。"""
sources = random_omni_scene(seed=5, count=2)
summary, _, _ = _run_scene(sources)
cleared = len(summary["cleared_channels"])
self.assertEqual(summary["avg_locate_clear_time_s"],
round(summary["virtual_time_s"] / cleared, 3))
class PersistentSearchTests(unittest.TestCase):
def test_persistence_disabled_stops_when_resolved(self):
"""persistent_search=False：回到"证明找全即退出"的基线行为。"""
config = replace(CONFIG, search=replace(
CONFIG.search, persistent_search=False, bearing_rounding_margin_deg=0.005))
summary, _, _ = _run_scene(random_omni_scene(seed=5, count=2),
config=config)
self.assertTrue(summary["completed"])
self.assertEqual(summary["stop_reason"], "all_resolved")
self.assertLessEqual(summary["measures"], 49 * 20 + 60)
def test_persistence_resweeps_only_uncleared_channels(self):
"""复扫只测未清除频道：已清除频道不再消耗检测。"""
sources = random_omni_scene(seed=5, count=2)
summary, _, _ = _run_scene(sources)
cleared = set(summary["cleared_channels"])
resweep_channels = {step["channel"] for step in summary["steps"]
if step["index"] > 0
and step["virtual_time_s"]
> summary["accounted_virtual_time_s"]
and step["action"] == "measure"}
self.assertTrue(resweep_channels)
self.assertFalse(resweep_channels & cleared,
"已清除频道不得在找全后被复扫")
def test_virtual_budget_stops_before_simulator_cap(self):
"""计划层在虚拟安全余量处停下，不触发模拟器 accepted=false 兜底。"""
sources = random_omni_scene(seed=5, count=2)
summary, report, _ = _run_scene(sources)
self.assertEqual(summary["stop_reason"], "virtual_time_cap")
field = SMALL_CAP.field
self.assertLessEqual(summary["virtual_time_s"],
field.max_virtual_duration_s)
self.assertGreaterEqual(summary["virtual_time_s"],
field.max_virtual_duration_s
- SMALL_CAP.search.virtual_reserve_s - 5_000)
self.assertEqual(summary["anomalies"], [])
self.assertEqual(report["mismatches"], [])
class SimulatorVirtualCapTests(unittest.TestCase):
def test_simulator_rejects_actions_beyond_virtual_cap(self):
"""附件2 §4.5 兜底：虚拟时长达限后本地模拟器不再执行检测。"""
simulator = LocalSimulator([], robot_id="T001", config=CONFIG)
simulator.entered = True
simulator.virtual_time_s = CONFIG.field.max_virtual_duration_s
status, body = simulator.handle(
"/measure", {"arena_id": "default", "robot_id": "T001",
"request_id": "x1",
"position": {"x": 0.0, "y": 0.0}, "channel": 1})
self.assertEqual(status, 200)
self.assertFalse(body["accepted"])
self.assertEqual(body["virtual_time_s"], 0)
class AbortFinalizationTests(unittest.TestCase):
"""断连超过重试上限：仍须 /exit、产出未完成 summary，不得静默崩溃。"""
@@ -161,6 +269,26 @@ class CoverageLedgerIntegrationTests(unittest.TestCase):
self.assertLessEqual(worst, 700.0 * 2 ** 0.5 + 1e-9)
class SceneGenerationTests(unittest.TestCase):
"""场景生成 [题面]：默认 10~16 源、Ω 内均匀、频道互异、r_eff∈[1000,1500]。"""
def test_default_count_within_problem_range(self):
for seed in range(20):
sources = random_omni_scene(seed=seed)
self.assertTrue(10 <= len(sources) <= 16, f"seed={seed}")
def test_scene_respects_problem_constraints(self):
for seed in range(10):
sources = random_omni_scene(seed=seed)
channels = [s.channel for s in sources]
self.assertEqual(len(channels), len(set(channels)))
self.assertTrue(all(1 <= c <= 20 for c in channels))
for s in sources:
self.assertLessEqual((s.x ** 2 + s.y ** 2) ** 0.5, 1800.0)
self.assertGreaterEqual(s.r_eff_m, 1000.0)
self.assertLessEqual(s.r_eff_m, 1500.0)
def math_dist(a, b):
return ((a[0] - b[0]) ** 2 + (a[1] - b[1]) ** 2) ** 0.5
tests/test_loop_q4.py
+15 -5
View File
@@ -26,6 +26,9 @@ from mcm26b.sim.scenes import mixed_scene
CONFIG = replace(ProblemConfig(),
search=replace(ProblemConfig().search,
bearing_rounding_margin_deg=0.005))
# 持续复扫扫到虚拟上限为止 [用户决策]；SMALL_CAP 压到 60k 秒缩短复扫轮数。
SMALL_CAP = replace(CONFIG,
field=replace(CONFIG.field, max_virtual_duration_s=60_000.0))
def _run_scene(sources, config=CONFIG, clock=None):
@@ -124,9 +127,10 @@ class MixedLoopAcceptanceTests(unittest.TestCase):
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(summary["stop_reason"], "virtual_time_cap")
self.assertEqual(sorted(summary["cleared_channels"]), truth)
self.assertEqual(summary["unresolved_channels"], [])
self.assertIsNotNone(summary["accounted_virtual_time_s"])
self.assertEqual(summary["anomalies"], [], "守恒性或证书不应被打破")
self.assertEqual(report["mismatches"], [])
self.assertEqual(summary["virtual_recompute_mismatches"], 0)
@@ -152,7 +156,7 @@ class MixedLoopAcceptanceTests(unittest.TestCase):
"""朝向背离圆心、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])
summary, _, simulator, _, _ = _run_scene([source], config=SMALL_CAP)
self.assertTrue(summary["completed"])
self.assertEqual(summary["cleared_channels"], [7])
self.assertEqual(sorted(simulator.cleared), [7])
@@ -162,7 +166,8 @@ class MixedLoopAcceptanceTests(unittest.TestCase):
"""固定批量种子回归：任何种子下不得全清失败或破坏守恒性。"""
for seed in range(5, 10):
sources = mixed_scene(seed=seed, count=10, directional=4)
summary, report, simulator, _, _ = _run_scene(sources)
summary, report, simulator, _, _ = _run_scene(sources,
config=SMALL_CAP)
truth = sorted(s.channel for s in sources)
self.assertTrue(summary["completed"], f"seed={seed}")
self.assertEqual(sorted(summary["cleared_channels"]), truth,
@@ -190,7 +195,7 @@ class JointConstraintBaselineTests(unittest.TestCase):
"""端到端保留性：定向源被遮挡若干次后仍被清除，不因 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])
summary, _, simulator, _, _ = _run_scene([source], config=SMALL_CAP)
self.assertTrue(summary["completed"])
self.assertEqual(summary["cleared_channels"], [3])
self.assertEqual(sorted(simulator.cleared), [3])
@@ -198,11 +203,16 @@ class JointConstraintBaselineTests(unittest.TestCase):
class FallbackDisciplineTests(unittest.TestCase):
def _stall_config(self, **search_updates):
"""迫使定位立即停滞进入后备清除的最小配置。"""
"""迫使定位立即停滞进入后备清除的最小配置。
persistent_search=False：本组验证的是后备清除纪律（格数/预算门槛），
与持续复扫无关，维持"证明找全即退出"的基线收尾便于断言。
"""
return replace(CONFIG,
search=replace(CONFIG.search,
max_localize_steps=1,
max_blocked_probes=0,
persistent_search=False,
**search_updates))
def test_cell_cap_skips_instead_of_blind_scan(self):