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