87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
"""HRL 协议执行的分支回归；所有模拟器交互仅在本地进行。"""
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("numpy")
|
|
|
|
from mcm26b.config import ProblemConfig
|
|
from mcm26b.harness import LocalSimTransport
|
|
from mcm26b.hrl.constants import FALLBACK_OFFSET, N_CANDIDATES, STATION_OFFSET
|
|
from mcm26b.hrl.scheduler import Candidate
|
|
from mcm26b.planner.hrl_loop import HRLLoop
|
|
from mcm26b.protocol import SimulatorClient
|
|
from referee import LocalSimulator, Source
|
|
|
|
|
|
def make_loop(*, problem=4, high_policy=None):
|
|
simulator = LocalSimulator(
|
|
[Source(channel=20, x=45.0, y=0.0, r_eff_m=1000.0)],
|
|
robot_id="hrl-branch-test")
|
|
client = SimulatorClient("hrl-branch-test", transport=LocalSimTransport(simulator))
|
|
client.enter()
|
|
return HRLLoop(client, ProblemConfig(), problem=problem, high_policy=high_policy)
|
|
|
|
|
|
def test_fallback_failure_then_success_records_actual_time_position_and_clear():
|
|
loop = make_loop()
|
|
try:
|
|
observation = loop.client.measure((0.0, 0.0), 20)
|
|
loop._record_measure(20, (0.0, 0.0), observation)
|
|
loop.blocked_probes[20] = [(0.0, 0.0)]
|
|
candidate = Candidate(FALLBACK_OFFSET + 19, "fallback", (999.0, 999.0),
|
|
channel=20)
|
|
with patch("mcm26b.hrl.scheduler.fallback_clear_cells",
|
|
return_value=[(20.0, 0.0), (40.0, 0.0)]):
|
|
outcome = loop.hrl_step(candidate.slot, [candidate])
|
|
assert outcome.clear_success == (20,)
|
|
assert loop.cleared == [20]
|
|
assert outcome.position_reached == (40.0, 0.0)
|
|
# 40m / 5m/s + 3s失败 + 5s成功；clear不产生换频时间。
|
|
assert outcome.delta_virtual_s == pytest.approx(16.0)
|
|
assert loop.client.state.channel == 20
|
|
assert loop.fallback_log[-1]["attempts"] == 2
|
|
assert 20 not in loop.blocked_probes
|
|
assert loop.anomalies == []
|
|
finally:
|
|
loop.client.exit()
|
|
|
|
|
|
def test_invalid_policy_output_is_logged_and_replaced_by_valid_action():
|
|
class InvalidPolicy:
|
|
def select(self, observation, mask):
|
|
return N_CANDIDATES + 1
|
|
|
|
loop = make_loop(problem=3, high_policy=InvalidPolicy())
|
|
candidate = Candidate(STATION_OFFSET, "station", (0.0, 0.0), station_index=0)
|
|
executed = []
|
|
|
|
def execute(action, candidates):
|
|
executed.append(action)
|
|
loop.stop_reason = "test_complete"
|
|
|
|
try:
|
|
with patch.object(loop, "build_candidates", return_value=[candidate]), \
|
|
patch.object(loop, "hrl_observation", return_value=None), \
|
|
patch.object(loop, "heuristic_action", return_value=candidate.slot), \
|
|
patch.object(loop, "hrl_step", side_effect=execute):
|
|
loop._coverage_phase()
|
|
assert executed == [candidate.slot]
|
|
assert loop.anomalies[0]["kind"] == "hrl_invalid_policy_action"
|
|
finally:
|
|
loop.client.exit()
|
|
|
|
|
|
def test_opportunistic_measure_precedes_candidate_snapshot():
|
|
loop = make_loop(problem=3)
|
|
order = []
|
|
try:
|
|
assert loop.config.bayes.opportunistic_measure
|
|
with patch.object(loop, "_measure_at_stop", side_effect=lambda: order.append("measure")), \
|
|
patch.object(loop, "build_candidates", side_effect=lambda: order.append("snapshot") or []):
|
|
loop._coverage_phase()
|
|
assert order == ["measure", "snapshot"]
|
|
finally:
|
|
loop.client.exit()
|