"""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()