217 lines
9.6 KiB
Python
217 lines
9.6 KiB
Python
import asyncio
|
|
import json
|
|
from unittest.mock import AsyncMock
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from defense_sim.radar.console import RadarConsole
|
|
from defense_sim.sensors.tracking import RadarTracker
|
|
from defense_sim.server import socketio_server as server
|
|
from defense_sim.simulation.scenarios import basic_intercept
|
|
from test_audit_regressions import measurement
|
|
|
|
|
|
def polar_measurement(position, velocity, sensor, t, name='r', rng=None):
|
|
relative = position-sensor
|
|
radius = np.linalg.norm(relative)
|
|
azimuth = np.arctan2(relative[1], relative[0])
|
|
elevation = np.arctan2(-relative[2], np.linalg.norm(relative[:2]))
|
|
radial = float(velocity @ (relative/radius))
|
|
if rng is not None:
|
|
radius += rng.normal(0, 2)
|
|
azimuth += rng.normal(0, np.deg2rad(.5))
|
|
elevation += rng.normal(0, np.deg2rad(.5))
|
|
radial += rng.normal(0, 1)
|
|
direction = np.array([np.cos(elevation)*np.cos(azimuth), np.cos(elevation)*np.sin(azimuth), -np.sin(elevation)])
|
|
return {'target_id': 'p', 'radar_id': name, 'measurement_time_s': t,
|
|
'position_ned_m': sensor+radius*direction, 'sensor_position_ned_m': sensor,
|
|
'range_m': radius, 'azimuth_rad': azimuth, 'elevation_rad': elevation,
|
|
'range_sigma_m': 2., 'azimuth_sigma_rad': np.deg2rad(.5),
|
|
'elevation_sigma_rad': np.deg2rad(.5), 'radial_velocity_m_s': radial, 'doppler_sigma_m_s': 1.}
|
|
|
|
|
|
def test_radial_measurement_leaves_tangential_velocity_uncertain():
|
|
tracker = RadarTracker()
|
|
point = np.array([1000., 0., -10.])
|
|
estimate = tracker.update(polar_measurement(point, np.array([-40., 30., 0.]), np.zeros(3), 0.))
|
|
assert estimate.velocity[0] == pytest.approx(-40., abs=.01)
|
|
assert abs(estimate.velocity[1]) < .001
|
|
assert estimate.covariance[4, 4] > 80000
|
|
assert not estimate.velocity_observable
|
|
|
|
|
|
def test_polar_ekf_fuses_multisite_evidence_in_ned_without_truth_inputs():
|
|
rng = np.random.default_rng(9)
|
|
tracker = RadarTracker(acceleration_diffusion=.01)
|
|
start = np.array([12000., 2000., -1200.])
|
|
velocity = np.array([-100., 20., 1.])
|
|
sites = [np.zeros(3), np.array([0., 10000., 0.]), np.array([10000., 0., 0.])]
|
|
estimate = None
|
|
for t in range(20):
|
|
for i, site in enumerate(sites):
|
|
updated = tracker.update(polar_measurement(start+t*velocity, velocity, site, float(t), str(i), rng))
|
|
if updated is not None:
|
|
estimate = updated
|
|
assert estimate.velocity_observable
|
|
assert np.linalg.norm(estimate.velocity-velocity) < 2.
|
|
assert np.linalg.norm(estimate.position-(start+19*velocity)) < 15.
|
|
assert np.linalg.eigvalsh(estimate.covariance).min() > 0
|
|
|
|
|
|
def test_outlier_is_recorded_and_cannot_corrupt_track_or_sample_count():
|
|
tracker = RadarTracker(acceleration_diffusion=.01)
|
|
for t in (0., 1., 2., 3.):
|
|
original = tracker.update(measurement(t))
|
|
outlier = measurement(4.)
|
|
outlier['position_ned_m'] = outlier['position_ned_m']+np.array([10000., 10000., -10000.])
|
|
assert tracker.update(outlier) is None
|
|
assert tracker.counts['p'] == 4
|
|
assert tracker.decisions[('p', 'r')]['reason'] == 'innovation_gate'
|
|
clean = tracker.update(measurement(5.))
|
|
assert clean is not None
|
|
assert np.linalg.norm(clean.velocity-original.velocity) < .2
|
|
|
|
|
|
def test_polar_filter_handles_azimuth_wrap():
|
|
tracker = RadarTracker(acceleration_diffusion=.01)
|
|
velocity = np.array([0., -1., 0.])
|
|
for t in (0., 1., 2., 3.):
|
|
updated = tracker.update(polar_measurement(np.array([-1000., 1.-t, -10.]), velocity, np.zeros(3), t))
|
|
assert updated is not None
|
|
assert abs(updated.position[1]) < 5.
|
|
|
|
|
|
@pytest.mark.parametrize('mode', ['demo', 'replay', 'scene'])
|
|
def test_console_views_are_finite_bounded_and_have_matching_resolution(mode):
|
|
console = RadarConsole(mode=mode, seed=4)
|
|
console.configure({'range_bins': 2048, 'doppler_bins': 8, 'range_bin': 2000})
|
|
packet = console.sample(1.)
|
|
assert len(packet['heatmap']) <= 256
|
|
assert len(packet['heatmap'][0]) == 8
|
|
assert len(packet['range_profile_db']) == 2048
|
|
assert len(packet['doppler_spectrum_db']) == 8
|
|
assert all(0 <= value <= 255 for row in packet['heatmap'] for value in row)
|
|
json.dumps(packet, allow_nan=False)
|
|
|
|
|
|
def test_v6_demo_is_seeded_and_controls_change_working_engine():
|
|
a, b = RadarConsole(seed=7), RadarConsole(seed=7)
|
|
assert a.sample(0.)['heatmap'] == b.sample(0.)['heatmap']
|
|
a.configure({'num_targets': 10, 'noise_floor': 0., 'clutter_level': 0., 'range_bins': 256, 'doppler_bins': 64})
|
|
packet = a.sample(1.)
|
|
assert len(a.processor.targets) == 10
|
|
assert a.processor.noise_floor == 0.
|
|
assert len(packet['heatmap'][0]) == 64
|
|
assert packet['waveform']['unambiguous_velocity_m_s'] > 0
|
|
a.configure({'range_bins': 256, 'range_bin': 255})
|
|
a.configure({'range_bins': 512})
|
|
assert a.settings['range_bin'] == 255
|
|
|
|
|
|
@pytest.mark.parametrize('changes', [ {'range_bins': 255}, {'doppler_bins': 8.5},
|
|
{'noise_floor': float('nan')}, {'show_grid': 'false'}, {'color_map': 'unknown'},
|
|
{'unexpected': 1}, {'max_velocity': True} ])
|
|
def test_settings_validation_is_atomic(changes):
|
|
console = RadarConsole(seed=1)
|
|
previous = console.settings.copy()
|
|
with pytest.raises(ValueError):
|
|
console.configure(changes)
|
|
assert console.settings == previous
|
|
|
|
|
|
def test_recording_pause_and_reset():
|
|
console = RadarConsole(seed=2)
|
|
console.configure({'range_bins': 256, 'doppler_bins': 8})
|
|
console.recording = True
|
|
for t in range(205):
|
|
console.sample(float(t))
|
|
assert len(console.recorded) == 200
|
|
assert console.recorded[0]['frame'] == 6
|
|
console.running = False
|
|
packet = console.sample(999.)
|
|
assert packet['frame'] == 205 and not packet['running']
|
|
assert len(console.recorded) == 200
|
|
positions = [(t['range'], t['velocity']) for t in console.processor.targets]
|
|
console.configure({'color_map': 'plasma'})
|
|
paused = console.sample(1000.)
|
|
assert paused['frame'] == 205
|
|
assert paused['settings']['color_map'] == 'plasma'
|
|
assert positions == [(t['range'], t['velocity']) for t in console.processor.targets]
|
|
console.reset()
|
|
assert console.frame == 0 and not console.recorded
|
|
|
|
|
|
def test_csv_replay_detects_bins_but_does_not_invent_angles():
|
|
packet = RadarConsole(mode='replay').sample(0.)
|
|
assert len(packet['targets']) > 0
|
|
assert all(target['azimuth'] is None and target['elevation'] is None for target in packet['targets'])
|
|
assert 'synthetic' in packet['provenance'] and 'illustrative' in packet['provenance']
|
|
|
|
|
|
def test_scene_calibration_is_site_specific_and_frames_do_not_change_truth():
|
|
sim = basic_intercept(seed=4)
|
|
a, b = sim.radars[:2]
|
|
console = RadarConsole(a.radar_id, 'scene', seed=7)
|
|
console.configure({'azimuth_sigma_deg': .25, 'forecast_horizon_s': 15.})
|
|
console.apply_scene_settings(sim, a)
|
|
assert a.config.azimuth_sigma_deg == .25 and b.config.azimuth_sigma_deg == .5
|
|
assert sim.forecast_horizon_s == 15.
|
|
before = sim.projectiles[0].position_ned_m.copy()
|
|
console.sample(0.)
|
|
assert np.array_equal(before, sim.projectiles[0].position_ned_m)
|
|
assert not sim.radar_tracks
|
|
|
|
|
|
def test_site_status_and_forecast_uncertainty_are_strict_json():
|
|
from test_audit_regressions import launch_fixture
|
|
sim = launch_fixture()
|
|
sim.radar_tracks['p'].covariance[3:, 3:] *= 400
|
|
sim.forecast_horizon_s = 30.
|
|
paths, details = sim._display_forecasts()
|
|
assert paths and 0 < details['p']['horizon_s'] < 30.
|
|
assert details['p']['end_position_sigma_bound_m']*3 <= 750.+.01
|
|
populated = basic_intercept(seed=5)
|
|
sites = populated.snapshot()['devices']['radars']
|
|
assert len(sites) == 10
|
|
assert sites[0]['usb_state'] == 'not_connected'
|
|
assert sites[0]['central_link'] == 'awaiting_report'
|
|
assert sites[0]['processing_latency_s'] == .0071
|
|
json.dumps(populated.snapshot(), allow_nan=False)
|
|
|
|
|
|
def test_console_events_are_socket_isolated_and_reject_other_runs(monkeypatch):
|
|
emit = AsyncMock()
|
|
monkeypatch.setattr(server.sio, 'emit', emit)
|
|
monkeypatch.setattr(server, 'authenticated', AsyncMock(return_value=True))
|
|
monkeypatch.setattr(server, 'engines', {'a': basic_intercept(seed=1)})
|
|
monkeypatch.setattr(server, 'radar_consoles', {})
|
|
monkeypatch.setattr(server, 'radar_console_tasks', {})
|
|
|
|
async def run():
|
|
await server.open_radar_console('b', {'radar_id': 'aeris10_01', 'mode': 'scene'})
|
|
assert 'b' not in server.radar_consoles
|
|
await server.open_radar_console('a', {'radar_id': 'aeris10_01', 'mode': 'scene'})
|
|
await asyncio.sleep(.001)
|
|
await server.command_radar_console('a', {'action': 'configure', 'settings': {'noise_floor': -1}})
|
|
await server.close_radar_console('a')
|
|
assert not server.radar_console_tasks
|
|
|
|
asyncio.run(run())
|
|
errors = [c for c in emit.call_args_list if c.args[0] == 'radar_console_error']
|
|
assert {c.kwargs['to'] for c in errors} == {'a', 'b'}
|
|
states = [c for c in emit.call_args_list if c.args[0] == 'radar_console_state']
|
|
assert states and all(c.kwargs['to'] == 'a' for c in states)
|
|
|
|
|
|
def test_heatmap_downsampling_preserves_sparse_scene_return():
|
|
console = RadarConsole('site', 'scene', seed=2)
|
|
console.configure({'noise_floor': 0., 'range_bins': 2048, 'max_range': 20000.})
|
|
row = {'target_id': 'target', 'measurement_time_s': 0., 'receive_time_s': 0.,
|
|
'range_m': 20000.*1001.5/2047, 'radial_velocity_m_s': 0.,
|
|
'azimuth_deg': 10., 'elevation_deg': 5., 'accepted': True, 'reason': 'accepted'}
|
|
packet = console.sample(0., [row])
|
|
# An isolated return lies between rows selected by naive stride decimation.
|
|
assert max(max(line) for line in packet['heatmap']) == 255
|
|
assert sum(value > 0 for line in packet['heatmap'] for value in line) == 1
|