90 lines
3.7 KiB
Python
90 lines
3.7 KiB
Python
"""附录2交会/串行换频约束和本轮优化的回归用例。"""
|
|
from dataclasses import replace
|
|
import math
|
|
|
|
import pytest
|
|
|
|
from mcm26b.config import ProblemConfig
|
|
from mcm26b.harness import LocalSimTransport, LocalSimulator
|
|
from mcm26b.planner import ClearCertificate, DirectionalExtension
|
|
from mcm26b.planner.bayes_loop import BayesLoop, Task
|
|
from mcm26b.planner.seven import validate_seven_coverage
|
|
from mcm26b.protocol.client import SimulatorClient
|
|
from mcm26b.protocol.rules import DIRECTION, NO_SIGNAL
|
|
from referee import Source
|
|
|
|
|
|
def setup(sources=(), cls=BayesLoop, config=ProblemConfig()):
|
|
sim = LocalSimulator(list(sources), robot_id='TEST', config=config)
|
|
client = SimulatorClient('TEST', config=config, transport=LocalSimTransport(sim))
|
|
return sim, client, cls(client, config)
|
|
|
|
|
|
def test_seven_coverage_checks_tuned_radius_and_rejects_holes():
|
|
config = ProblemConfig()
|
|
assert validate_seven_coverage(config.bayes.station_ring_radius_m, config) < 1000
|
|
for radius in (0, 500, 2000):
|
|
with pytest.raises(ValueError):
|
|
validate_seven_coverage(radius, config)
|
|
|
|
|
|
def test_seven_stations_cannot_prove_directional_absence():
|
|
# +x边界源朝外，七站全在源背面；第四问必须保留域外发现站。
|
|
source = Source(1, 1800, 0, 1000, directional=True, heading_deg=0)
|
|
sim, _, loop = setup([source])
|
|
assert all(sim._observe(p, 1)[0] == NO_SIGNAL for p in loop.ledger.points)
|
|
_, _, directional = setup([source], cls=DirectionalExtension)
|
|
assert any(sim._observe(p, 1)[0] == DIRECTION for p in directional.ledger.points)
|
|
|
|
|
|
def test_stopped_opportunistic_measure_is_serial_and_never_repeats_location():
|
|
sim, client, loop = setup([Source(1, 800, 200, 1500)])
|
|
client.enter()
|
|
loop._record_measure(1, (0, 0), client.measure((0, 0), 1))
|
|
client.measure((600, 0), 2)
|
|
before = client.state.virtual_time_s
|
|
loop._measure_at_stop()
|
|
assert loop.opportunistic_measures == 1
|
|
assert client.state.virtual_time_s - before == 6 # 1秒换频+5秒停靠检测
|
|
assert client.state.position == (600, 0)
|
|
assert loop.belief.candidate_set(1).contains((800, 200))
|
|
loop._measure_at_stop()
|
|
assert client.state.virtual_time_s - before == 6
|
|
|
|
|
|
def test_route_cost_accounts_for_channel_state_and_clear_preserves_it():
|
|
_, client, loop = setup()
|
|
client.state.channel = 3
|
|
at = (0., 0.)
|
|
clear = Task('clear', at, 5, channel=7)
|
|
locate = Task('locate', at, 6, channel=3)
|
|
station = Task('station', at, 18, station_index=0, station_channels=(1, 2, 3))
|
|
assert loop._path_cost([clear, locate]) == 10
|
|
assert loop._path_cost([locate, station]) == 5 + 17
|
|
assert loop._path_cost([station, locate]) == 17 + 6
|
|
|
|
|
|
def test_shorter_clear_target_still_covers_every_possible_source():
|
|
_, _, loop = setup()
|
|
cert = ClearCertificate(1, (100, 0), 5, 19.5, True)
|
|
result = loop._near_clear_target(cert)
|
|
assert math.dist((0, 0), result.center) < 100
|
|
assert result.radius_m <= result.limit_m
|
|
for k in range(360):
|
|
p = (100 + 5*math.cos(math.radians(k)), 5*math.sin(math.radians(k)))
|
|
assert math.dist(result.center, p) <= 19.5
|
|
|
|
|
|
def test_out_of_range_probe_does_not_certify_directionality():
|
|
base = ProblemConfig()
|
|
config = replace(base, search=replace(base.search, max_blocked_probes=1))
|
|
_, client, loop = setup([Source(1, 800, 0, 1000)], cls=DirectionalExtension, config=config)
|
|
client.enter()
|
|
outcome = client.measure((0, 0), 1)
|
|
loop.belief.update_measure(1, (0, 0), outcome.result, outcome.svd_deg)
|
|
loop._next_probe = lambda region, channel: (3000, 0)
|
|
loop._fallback_clear = lambda channel, region: False
|
|
assert not loop._localize_and_clear(1)
|
|
assert loop.directional_observed == []
|
|
assert loop.anomalies == []
|