Lab059: establish limits of preamble-only carrier estimation

tests/test_lab059_preamble_carrier_limits.py

@@ -0,0 +1,155 @@
+"""Проверки математики и информационных границ Lab059."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from experiments import lab059_preamble_carrier_limits as lab059
+
+
+@pytest.mark.parametrize("frequency_hz", [-100.0, -5.0, -0.25, 0.0, 0.5, 20.0])
+def test_phase_slope_conversion_sign_units_and_square_factor(frequency_hz):
+    ordinary = 2 * np.pi * frequency_hz
+    squared = 4 * np.pi * frequency_hz
+    assert lab059.phase_slope_to_hz(ordinary, squared_bpsk=False) == pytest.approx(frequency_hz)
+    assert lab059.phase_slope_to_hz(squared, squared_bpsk=True) == pytest.approx(frequency_hz)
+    if frequency_hz:
+        assert lab059.phase_slope_to_hz(squared, squared_bpsk=False) == pytest.approx(2 * frequency_hz)
+
+
+def test_phase_drift_and_frequency_limit_are_inverse():
+    for phase_limit in (15, 30, 45, 60, 90):
+        frequency = lab059.maximum_frequency_error_hz(
+            phase_limit, lab059.BURST_DURATION_SECONDS
+        )
+        assert lab059.phase_drift_degrees(
+            frequency, lab059.BURST_DURATION_SECONDS
+        ) == pytest.approx(phase_limit)
+    assert lab059.maximum_frequency_error_hz(30, 0.2048) == pytest.approx(0.4069010416666667)
+
+
+def test_awgn_crlb_scale_matches_rate_and_observed_b_quality():
+    slow = lab059.awgn_cfo_crlb_standard_deviation_hz(0.845350, 1280, 20_000)
+    fast = lab059.awgn_cfo_crlb_standard_deviation_hz(0.845350, 1280, 500_000)
+    assert fast / slow == pytest.approx(25.0)
+    assert fast == pytest.approx(3.5982182026)
+
+
+@pytest.mark.parametrize("symbol_rate", [20_000, 500_000])
+@pytest.mark.parametrize("method", lab059.METHODS)
+def test_clean_known_cfo_reference_time_and_compensation(symbol_rate, method):
+    frequency_hz = 7.25 if symbol_rate == 20_000 else -37.5
+    phase_at_zero = -0.73
+    indexes = np.arange(lab059.PREAMBLE_SYMBOLS)
+    times = (indexes + 0.5) / symbol_rate
+    symbols = lab059.lab057.EXPECTED * np.exp(
+        1j * (phase_at_zero + 2 * np.pi * frequency_hz * times)
+    )
+    estimate = lab059.estimate_carrier(symbols, symbol_rate, method)
+    assert estimate.frequency_hz == pytest.approx(frequency_hz, abs=1e-8)
+    assert np.angle(estimate.gain * np.exp(-1j * phase_at_zero)) == pytest.approx(0.0, abs=1e-10)
+    corrected = lab059.compensate(symbols, symbol_rate, estimate)
+    assert np.max(abs(corrected - lab059.lab057.EXPECTED)) < 1e-9
+
+
+def test_p0_matches_active_two_block_formula_and_ambiguity():
+    symbol_rate = 500_000
+    frequency_hz = 41.25
+    times = (np.arange(lab059.PREAMBLE_SYMBOLS) + 0.5) / symbol_rate
+    symbols = lab059.lab057.EXPECTED * np.exp(2j * np.pi * frequency_hz * times)
+    expected = lab059.lab057.squared_cfo_statistics(symbols, symbol_rate)
+    estimate = lab059.estimate_carrier(symbols, symbol_rate, "P0")
+    assert estimate.frequency_hz == pytest.approx(expected["cfo_hz"], abs=1e-12)
+    assert estimate.estimation_symbols == 1024
+    assert estimate.unambiguous_abs_hz == pytest.approx(symbol_rate / 2048)
+
+
+def test_p2_reference_time_is_global_burst_zero():
+    symbol_rate = 500_000
+    frequency_hz = -12.75
+    phase_at_zero = 1.12
+    times = (np.arange(lab059.PREAMBLE_SYMBOLS) + 0.5) / symbol_rate
+    symbols = lab059.lab057.EXPECTED * np.exp(
+        1j * (phase_at_zero + 2 * np.pi * frequency_hz * times)
+    )
+    estimate = lab059.estimate_carrier(symbols, symbol_rate, "P2")
+    assert estimate.frequency_hz == pytest.approx(frequency_hz, abs=1e-9)
+    assert np.angle(estimate.gain * np.exp(-1j * phase_at_zero)) == pytest.approx(0.0, abs=1e-10)
+    assert estimate.validation_errors == 0
+
+
+@pytest.mark.parametrize("method", lab059.METHODS)
+def test_payload_truth_cannot_change_carrier_estimate(method):
+    symbol_rate = 500_000
+    indexes = np.arange(4096)
+    expected = np.resize(lab059.lab057.EXPECTED, len(indexes))
+    values = expected * np.exp(1j * (0.4 + 2 * np.pi * 3.5 * (indexes + 0.5) / symbol_rate))
+    changed = values.copy()
+    random_generator = np.random.default_rng(5907)
+    changed[lab059.PREAMBLE_SYMBOLS:] = (
+        random_generator.normal(size=len(changed) - lab059.PREAMBLE_SYMBOLS)
+        + 1j * random_generator.normal(size=len(changed) - lab059.PREAMBLE_SYMBOLS)
+    ) * 100
+    first = lab059.estimate_from_burst(values, symbol_rate, method)
+    second = lab059.estimate_from_burst(changed, symbol_rate, method)
+    assert first == second
+
+
+def test_raw_payload_cannot_change_new_preamble_only_estimates():
+    symbol_rate = 500_000
+    values, _truth = lab059.lab057.synthetic_waveform(
+        symbol_rate, delay_symbols=0.0, cfo_hz=3.5, phase_rad=0.4, es_n0_db=None
+    )
+    frontend = lab059.lab057.Frontend(symbol_rate)
+    probe = values[: (4096 + 42) * frontend.samples_per_symbol]
+    stock = frontend.extract(probe, 4096)
+    matched = frontend.filter(probe)
+    symbols = lab059._extract_symbols(
+        matched, frontend.nominal, stock.timing_offset_samples,
+        4096, frontend.samples_per_symbol,
+    )
+    changed = probe.copy()
+    payload_start = (
+        frontend.nominal + stock.timing_offset_samples
+        + lab059.PREAMBLE_SYMBOLS * frontend.samples_per_symbol
+    )
+    changed[payload_start + len(frontend.taps):] += 10 + 7j
+    first = lab059.real_carrier_estimates(
+        probe, frontend, stock.timing_offset_samples, symbols, symbol_rate, ("P1", "P2", "P3")
+    )
+    second = lab059.real_carrier_estimates(
+        changed, frontend, stock.timing_offset_samples, symbols, symbol_rate, ("P1", "P2", "P3")
+    )
+    assert first == second
+
+
+def test_phase_extrapolation_uses_one_compensation_only():
+    symbol_rate = 500_000
+    count = 102_400
+    frequency_hz = 4.7
+    phase = 0.8
+    indexes = np.arange(count)
+    expected = np.resize(lab059.lab057.EXPECTED, count)
+    symbols = expected * np.exp(1j * (phase + 2 * np.pi * frequency_hz * (indexes + 0.5) / symbol_rate))
+    estimate = lab059.estimate_from_burst(symbols, symbol_rate, "P1")
+    corrected = lab059.compensate(symbols, symbol_rate, estimate)
+    assert np.max(abs(corrected - expected)) < 1e-8
+
+
+def test_fixed_split_is_disjoint_and_not_result_dependent():
+    assert len(lab059.P2_TRAIN) == len(lab059.P2_VALIDATION) == 576
+    assert lab059.P2_TRAIN[0] == 32 and lab059.P2_TRAIN[-1] == 607
+    assert lab059.P2_VALIDATION[0] == 672 and lab059.P2_VALIDATION[-1] == 1247
+    assert set(lab059.P2_TRAIN).isdisjoint(lab059.P2_VALIDATION)
+
+
+def test_partial_baseline_summary_does_not_create_nan_for_unrun_methods():
+    row = {
+        "dataset": "Lab056", "label": "B1", "method": "P0",
+        "frequency_error_hz": 1.0, "symbol_error_rate": 0.5,
+        "evm_rms": 1.0, "matched_residual_db": 0.0,
+    }
+    summary = lab059.summarize_real([row])
+    assert set(summary["Lab056"]) == {"P0"}
+    assert summary["Lab054"] == {} and summary["Lab047"] == {}