287 lines
13 KiB
Python
287 lines
13 KiB
Python
"""HRL 验收：槽位布局、观测/掩码同源性、整场环境闭环与奖励恒等式。
|
|
|
|
HRL 依赖属于可选组 [dependency-groups] rl；未安装 numpy/gymnasium 时
|
|
全部跳过（运行闭环不依赖它们）。
|
|
"""
|
|
|
|
import math
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
pytest = None
|
|
try:
|
|
import pytest
|
|
numpy = pytest.importorskip("numpy")
|
|
pytest.importorskip("gymnasium")
|
|
except Exception: # pragma: no cover - 无 RL 依赖环境
|
|
numpy = None
|
|
|
|
if numpy is not None:
|
|
from mcm26b.config import ProblemConfig
|
|
from mcm26b.hrl.constants import (CHANNEL_COUNT, CLEAR_OFFSET, CLEAR_SLOTS,
|
|
FALLBACK_OFFSET, FALLBACK_SLOTS, HIGH_GLOBAL_DIM,
|
|
HIGH_OBS_DIM, LOCATE_OFFSET,
|
|
LOCATE_SLOTS,
|
|
N_CANDIDATES, PER_CANDIDATE_DIM,
|
|
PER_CHANNEL_DIM, STATION_OFFSET, STATION_SLOTS)
|
|
from mcm26b.hrl.curriculum import STAGES, make_scene, training_config
|
|
from mcm26b.hrl.executor import LowLevelExecutor, build_low_observation
|
|
from mcm26b.hrl.hierarchical_env import HierarchicalEnv
|
|
from mcm26b.hrl.low_level_env import LowLevelEnv
|
|
from mcm26b.hrl.scheduler import HRLControllerMixin
|
|
|
|
CONFIG = ProblemConfig()
|
|
|
|
|
|
def _absent(info: dict) -> int:
|
|
return info.get("absent_proven", 0)
|
|
|
|
|
|
def _env(problem: int = 3, stage_index: int = 0, **kwargs) -> HierarchicalEnv:
|
|
return HierarchicalEnv(problem=problem, stage=STAGES[stage_index], **kwargs)
|
|
|
|
|
|
if numpy is not None:
|
|
class _CandidateHost(HRLControllerMixin):
|
|
"""只提供候选所需感知原语，槽位分配和掩码均调用生产实现。"""
|
|
|
|
def __init__(self, detected, certified=(), blocked=()):
|
|
self.config = CONFIG
|
|
self.problem = 4 if blocked else 3
|
|
self._certified = set(certified)
|
|
self._clear_failed = set()
|
|
self.blocked_probes = {
|
|
channel: [(0.0, 0.0)] * CONFIG.search.max_blocked_probes
|
|
for channel in blocked
|
|
}
|
|
region = SimpleNamespace(
|
|
polygon=[(1.0, 1.0), (9.0, 1.0), (9.0, 9.0), (1.0, 9.0)],
|
|
mec=lambda: ((5.0, 5.0), math.sqrt(32.0)))
|
|
self.belief = SimpleNamespace(
|
|
channels_by_status=lambda status: list(detected),
|
|
candidate_set=lambda channel: region)
|
|
self.ledger = SimpleNamespace(points=[])
|
|
|
|
def _best_certificate(self, channel):
|
|
return SimpleNamespace(certified=channel in self._certified,
|
|
center=(float(channel), 0.0))
|
|
|
|
def _select_point(self, channel):
|
|
return (float(channel), 1.0)
|
|
|
|
def _hrl_probe_point(self, region, channel):
|
|
return (float(channel), 2.0)
|
|
|
|
|
|
@unittest.skipIf(numpy is None, "RL 依赖组未安装")
|
|
class LayoutTests(unittest.TestCase):
|
|
def test_high_obs_dim_matches_component_contract(self):
|
|
expected = (HIGH_GLOBAL_DIM + CHANNEL_COUNT * PER_CHANNEL_DIM
|
|
+ N_CANDIDATES * PER_CANDIDATE_DIM)
|
|
self.assertEqual(HIGH_OBS_DIM, expected)
|
|
self.assertEqual(HIGH_OBS_DIM, 1248)
|
|
self.assertEqual(N_CANDIDATES, 120)
|
|
|
|
def test_observation_finite_and_deterministic(self):
|
|
env = _env(problem=3)
|
|
obs, _ = env.reset(seed=2026)
|
|
self.assertEqual(obs.shape, (HIGH_OBS_DIM,))
|
|
self.assertTrue(numpy.isfinite(obs).all())
|
|
again = env._loop.hrl_observation(env._current_candidates)
|
|
self.assertTrue(numpy.array_equal(obs, again))
|
|
|
|
def test_mask_nonempty_and_matches_candidate_slots(self):
|
|
env = _env(problem=3)
|
|
env.reset(seed=7)
|
|
for _ in range(6):
|
|
mask = env.action_masks()
|
|
self.assertTrue(mask.any())
|
|
slots = {cand.slot for cand in env._current_candidates}
|
|
self.assertEqual({int(i) for i in numpy.flatnonzero(mask)}, slots)
|
|
_obs, _r, term, trunc, _info = env.step(
|
|
int(env._loop.heuristic_action(env._current_candidates)))
|
|
if term or trunc:
|
|
break
|
|
|
|
def test_q4_probe_and_fallback_slots_exist_in_layout(self):
|
|
self.assertEqual(CLEAR_SLOTS, CHANNEL_COUNT)
|
|
self.assertEqual(LOCATE_SLOTS, CHANNEL_COUNT)
|
|
self.assertEqual(FALLBACK_SLOTS, CHANNEL_COUNT)
|
|
self.assertEqual(STATION_SLOTS, 49)
|
|
self.assertEqual(CLEAR_OFFSET, 0)
|
|
self.assertEqual(LOCATE_OFFSET, CLEAR_OFFSET + CLEAR_SLOTS)
|
|
self.assertEqual(STATION_OFFSET, LOCATE_OFFSET + LOCATE_SLOTS)
|
|
self.assertEqual(FALLBACK_OFFSET, STATION_OFFSET + STATION_SLOTS)
|
|
self.assertEqual(N_CANDIDATES - (FALLBACK_OFFSET + FALLBACK_SLOTS), 11)
|
|
|
|
def test_build_candidates_keeps_clear_and_locate(self):
|
|
host = _CandidateHost(detected=[1, 2, 3, 4, 17, 18, 19, 20],
|
|
certified=[17, 18, 19, 20])
|
|
candidates = host.build_candidates()
|
|
expected = {(CLEAR_OFFSET + channel - 1, "clear", channel)
|
|
for channel in range(17, 21)}
|
|
expected.update((LOCATE_OFFSET + channel - 1, "locate", channel)
|
|
for channel in range(1, 5))
|
|
self.assertEqual({(c.slot, c.kind, c.channel) for c in candidates}, expected)
|
|
self.assertEqual(len(candidates), 8)
|
|
mask = host.hrl_action_mask(candidates)
|
|
self.assertEqual(set(numpy.flatnonzero(mask)), {c.slot for c in candidates})
|
|
for channel in range(17, 21):
|
|
selected = host.candidate_by_slot(candidates, CLEAR_OFFSET + channel - 1)
|
|
self.assertEqual((selected.kind, selected.channel), ("clear", channel))
|
|
|
|
def test_build_candidates_fallback_covers_slots(self):
|
|
channels = list(range(17, 21))
|
|
host = _CandidateHost(detected=channels, blocked=channels)
|
|
candidates = host.build_candidates()
|
|
fallbacks = [c for c in candidates if c.kind == "fallback"]
|
|
self.assertEqual({c.channel for c in fallbacks}, set(channels))
|
|
self.assertEqual(len(candidates), 8) # 四个 probe 与四个 fallback 并存
|
|
for candidate in fallbacks:
|
|
self.assertEqual(candidate.slot, FALLBACK_OFFSET + candidate.channel - 1)
|
|
self.assertGreaterEqual(candidate.slot, FALLBACK_OFFSET)
|
|
self.assertLess(candidate.slot, FALLBACK_OFFSET + FALLBACK_SLOTS)
|
|
self.assertGreater(candidate.targets, 0)
|
|
mask = host.hrl_action_mask(candidates)
|
|
self.assertTrue(mask[FALLBACK_OFFSET + 19])
|
|
self.assertFalse(mask[FALLBACK_OFFSET + FALLBACK_SLOTS:].any())
|
|
|
|
|
|
@unittest.skipIf(numpy is None, "RL 依赖组未安装")
|
|
class HeuristicLoopTests(unittest.TestCase):
|
|
def test_q3_full_episode_clears_all_sources(self):
|
|
env = _env(problem=3, stage_index=3, max_episode_steps=500)
|
|
_, _ = env.reset(seed=11)
|
|
steps = 0
|
|
while True:
|
|
action = env._loop.heuristic_action(env._current_candidates)
|
|
_, _, term, trunc, info = env.step(action)
|
|
steps += 1
|
|
if term or trunc:
|
|
break
|
|
self.assertTrue(term)
|
|
self.assertGreaterEqual(info["cleared"], 10)
|
|
self.assertEqual(info["cleared"] + _absent(info), 20)
|
|
self.assertEqual(info["accounted"], 20)
|
|
self.assertEqual(info["anomalies"], [])
|
|
|
|
def test_q4_episode_accounts_all_channels(self):
|
|
env = _env(problem=4, stage_index=3, max_episode_steps=600)
|
|
env.reset(seed=21)
|
|
while True:
|
|
action = env._loop.heuristic_action(env._current_candidates)
|
|
_, _, term, trunc, info = env.step(action)
|
|
if term or trunc:
|
|
break
|
|
self.assertTrue(term)
|
|
self.assertEqual(info["accounted"], 20)
|
|
self.assertEqual(info["anomalies"], [])
|
|
self.assertGreaterEqual(info["cleared"], 10)
|
|
|
|
def test_reward_identity_from_outcome_terms(self):
|
|
"""非终止步奖励恒等于各分项加权和（prompt §5.2 的可折算实现）。"""
|
|
env = _env(problem=3, stage_index=1, max_episode_steps=50)
|
|
env.reset(seed=5)
|
|
reward_cfg = env.reward
|
|
for _ in range(8):
|
|
action = env._loop.heuristic_action(env._current_candidates)
|
|
_virtual_before = env._client.state.virtual_time_s
|
|
_, r, term, trunc, info = env.step(action)
|
|
outcome = env._last_outcome
|
|
if outcome.invalid:
|
|
self.assertAlmostEqual(r, -reward_cfg.invalid, places=6)
|
|
else:
|
|
expected = (-outcome.delta_virtual_s / reward_cfg.time_scale
|
|
+ reward_cfg.clear * len(outcome.clear_success)
|
|
+ reward_cfg.detect * len(outcome.new_detections)
|
|
+ reward_cfg.info_weight
|
|
* sum(outcome.info_gains.values()))
|
|
if outcome.redundant:
|
|
expected -= reward_cfg.redundant
|
|
if term:
|
|
expected += (reward_cfg.terminal_accounted
|
|
if info["accounted"] == 20
|
|
else -reward_cfg.terminal_unresolved)
|
|
self.assertAlmostEqual(r, expected, places=6)
|
|
if term or trunc:
|
|
break
|
|
|
|
def test_invalid_action_penalized_without_sim_call(self):
|
|
env = _env(problem=3, stage_index=0, max_episode_steps=50)
|
|
env.reset(seed=3)
|
|
mask = env.action_masks()
|
|
invalid_action = int(numpy.argmin(mask))
|
|
virtual_before = env._client.state.virtual_time_s
|
|
_, r, _, _, _ = env.step(invalid_action)
|
|
self.assertAlmostEqual(r, -env.reward.invalid, places=6)
|
|
self.assertAlmostEqual(env._client.state.virtual_time_s,
|
|
virtual_before, places=6)
|
|
|
|
|
|
@unittest.skipIf(numpy is None, "RL 依赖组未安装")
|
|
class LowLevelTests(unittest.TestCase):
|
|
def test_low_env_shapes_and_expert_reaches_goal(self):
|
|
env = LowLevelEnv(seed=99)
|
|
obs, info = env.reset(seed=99)
|
|
self.assertEqual(obs.shape, (10,))
|
|
start = tuple(info["position"])
|
|
goal = tuple(info["goal"])
|
|
total = 0
|
|
for _ in range(60):
|
|
obs, _, term, trunc, _ = env.step(env.expert_action())
|
|
total += 1
|
|
if term or trunc:
|
|
break
|
|
self.assertTrue(term, "直线专家应在步数上限内到达")
|
|
straight = math.dist(start, goal)
|
|
efficiency = straight / (total * env.step_m)
|
|
self.assertGreater(efficiency, 0.7, "专家路径效率应接近直线")
|
|
|
|
def test_executor_straightline_is_goal_only(self):
|
|
executor = LowLevelExecutor(policy=None)
|
|
self.assertEqual(executor.roll((0.0, 0.0), (500.0, 200.0)),
|
|
[(500.0, 200.0)])
|
|
|
|
def test_low_observation_source_shared(self):
|
|
obs = build_low_observation((0.0, 0.0), (300.0, 0.0), None, 0)
|
|
self.assertEqual(obs.shape, (10,))
|
|
self.assertAlmostEqual(float(obs[4]), 1.0, places=6) # cos(bearing)=1
|
|
|
|
|
|
@unittest.skipIf(numpy is None, "RL 依赖组未安装")
|
|
class CurriculumTests(unittest.TestCase):
|
|
def test_stages_are_monotonic_in_difficulty(self):
|
|
for prev, nxt in zip(STAGES, STAGES[1:]):
|
|
self.assertLessEqual(prev.sources_max, nxt.sources_max)
|
|
self.assertLessEqual(prev.channel_max, nxt.channel_max)
|
|
self.assertLessEqual(prev.bearing_error_half_deg,
|
|
nxt.bearing_error_half_deg)
|
|
|
|
def test_training_config_defaults_to_production(self):
|
|
config = training_config(CONFIG, STAGES[3])
|
|
self.assertEqual(config.field.channel_max, 20)
|
|
self.assertEqual(config.field.bearing_error_half_deg, 1.0)
|
|
self.assertEqual(config.bayes, CONFIG.bayes)
|
|
|
|
def test_fast_config_preserves_certificate_grid(self):
|
|
config = training_config(CONFIG, STAGES[3], fidelity="fast")
|
|
self.assertEqual(config.bayes.particle_spacing_m,
|
|
CONFIG.bayes.particle_spacing_m)
|
|
self.assertLess(config.bayes.particle_spacing_m * math.sqrt(2) / 2,
|
|
config.field.clear_radius_m)
|
|
self.assertLessEqual(config.bayes.score_subsample,
|
|
CONFIG.bayes.score_subsample)
|
|
|
|
def test_scene_source_count_and_channel_bounds(self):
|
|
from random import Random
|
|
rng = Random(1)
|
|
sources, channels = make_scene(rng, training_config(CONFIG, STAGES[1]),
|
|
3, STAGES[1])
|
|
self.assertEqual(len(sources), STAGES[1].sources_max)
|
|
self.assertTrue(all(1 <= c <= STAGES[1].channel_max for c in channels))
|
|
self.assertEqual(len(set(channels)), len(channels))
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
unittest.main()
|