Lab057: diagnose Lab054 symbol synchronization failures offline

tests/test_lab057_symbol_diagnosis.py

@@ -0,0 +1,128 @@
+"""Проверки диагностических оценок без сохранённых IQ и оборудования."""
+
+from __future__ import annotations
+
+import json
+
+import numpy as np
+import pytest
+
+from experiments import lab057_symbol_diagnosis as diagnosis
+
+
+@pytest.mark.parametrize("symbol_rate", [20_000, 500_000])
+@pytest.mark.parametrize("sign", [-1, 1])
+def test_frequency_phase_units_and_compensation(symbol_rate, sign):
+    frequency = sign * 37.5 * symbol_rate / 500_000
+    phase = -.8
+    times = (np.arange(1280) + .5) / symbol_rate
+    symbols = diagnosis.EXPECTED * np.exp(1j * (phase + 2*np.pi*frequency*times))
+    found, gain, score, boundary = diagnosis.frequency_search(symbols[diagnosis.TRAIN], symbol_rate)
+    assert abs(found - frequency) < .01
+    assert abs(np.angle(gain) - phase) < .001
+    assert score > .99999 and not boundary
+    estimate = diagnosis.Estimate(0, found, gain, score)
+    corrected = diagnosis.compensate(symbols, symbol_rate, estimate)
+    assert np.count_nonzero((corrected.real[diagnosis.VALIDATION] >= 0) != diagnosis.PREAMBLE[diagnosis.VALIDATION]) == 0
+
+
+@pytest.mark.parametrize("symbol_rate", [20_000, 500_000])
+def test_squared_cfo_alias_and_factor_two(symbol_rate):
+    bound = symbol_rate / 2048
+    times = (np.arange(1280) + .5) / symbol_rate
+    for frequency, expected in [(bound*.4, bound*.4), (bound*1.2, -bound*.8)]:
+        symbols = diagnosis.EXPECTED * np.exp(2j*np.pi*frequency*times)
+        result = diagnosis.squared_cfo_statistics(symbols, symbol_rate)
+        assert result["cfo_hz"] == pytest.approx(expected, abs=1e-9)
+        assert result["delta_time_seconds"] == 512 / symbol_rate
+
+
+def test_frequency_boundary_is_reported_without_expanding_search():
+    times = (diagnosis.TRAIN + .5) / 500_000
+    symbols = diagnosis.EXPECTED[diagnosis.TRAIN] * np.exp(2j*np.pi*400*times)
+    frequency, _, _, boundary = diagnosis.frequency_search(symbols, 500_000)
+    assert boundary and frequency == 250
+
+
+def test_fractional_interpolator_preserves_integer_samples_and_complex_phase():
+    indexes = np.arange(200)
+    values = np.exp(2j*np.pi*.03*indexes)
+    assert np.allclose(diagnosis.interpolate(values, indexes[20:180]), values[20:180], atol=1e-14)
+    positions = np.arange(20, 180) + .35
+    assert np.max(abs(diagnosis.interpolate(values, positions) - np.exp(2j*np.pi*.03*positions))) < .001
+
+
+@pytest.mark.parametrize("symbol_rate", [20_000, 500_000])
+def test_fractional_timing_recovery_from_waveform(symbol_rate):
+    values, truth = diagnosis.synthetic_waveform(symbol_rate)
+    frontend = diagnosis.Frontend(symbol_rate)
+    probe = values[:(1424+42)*frontend.samples_per_symbol]
+    estimate = diagnosis.estimate_variant(frontend.estimation_filter(probe), frontend.nominal,
+                                          symbol_rate, "D3", 700, 999)
+    assert abs(estimate.timing_samples / frontend.samples_per_symbol - truth["delay_symbols"]) <= 1/32
+    assert abs(estimate.cfo_hz) < symbol_rate/500_000 * 2
+
+
+@pytest.mark.parametrize("variant", ["D0", "D1", "D2", "D3"])
+def test_raw_validation_header_and_body_cannot_change_new_estimates(variant):
+    frontend = diagnosis.Frontend(500_000)
+    random_generator = np.random.default_rng(5717)
+    original = (random_generator.normal(size=23456) + 1j*random_generator.normal(size=23456)).astype(np.complex64)
+    altered = original.copy()
+    low, high = frontend.train_support()
+    altered[:low] += 10+7j
+    altered[high:] += 10+7j
+    first = diagnosis.estimate_variant(frontend.estimation_filter(original), frontend.nominal, 500_000, variant, 0, 17)
+    second = diagnosis.estimate_variant(frontend.estimation_filter(altered), frontend.nominal, 500_000, variant, 0, 17)
+    assert first == second
+
+
+def test_d3_does_not_inherit_stock_timing_or_cfo():
+    values, _ = diagnosis.synthetic_waveform(500_000)
+    frontend = diagnosis.Frontend(500_000)
+    matched = frontend.estimation_filter(values[:23456])
+    assert diagnosis.estimate_variant(matched, frontend.nominal, 500_000, "D3", 0, 0) == diagnosis.estimate_variant(
+        matched, frontend.nominal, 500_000, "D3", 800, -200)
+
+
+def test_reference_file_changes_only_post_decision_error_counts(tmp_path):
+    values, _ = diagnosis.synthetic_waveform(500_000, delay_symbols=0)
+    reader = diagnosis.ArrayFile(values)
+    path = tmp_path / "expected_header_and_jpeg.json"
+    expected = diagnosis.historical.plans()[0].plan.phy_header_bits
+
+    def analyze(bits, jpeg):
+        path.write_text(json.dumps({"header": bits.tolist(), "jpeg": jpeg}), encoding="utf-8")
+        truth = json.loads(path.read_text(encoding="utf-8"))
+        return diagnosis.analyze_frame(reader, 0, 500_000, "TEST", 1,
+                                       np.array(truth["header"]), write_packets=False)[0]
+
+    first = analyze(expected, "original")
+    second = analyze(1-expected, "different bytes")
+    keys = ("timing_samples", "cfo_hz", "phase_rad", "gain_abs", "HEADER_ACCEPTED", "CRC_RESULT")
+    assert all(all(a[key] == b[key] for key in keys) for a, b in zip(first, second))
+    assert first[0]["header_voted_errors"] != second[0]["header_voted_errors"]
+
+
+def test_failed_header_never_starts_fec_or_crc():
+    result, output = diagnosis.decode_body(np.zeros(144))
+    assert not result["HEADER_ACCEPTED"]
+    assert not result["FEC_STARTED"] and not result["CRC_STARTED"]
+    assert result["CRC_RESULT"] == "NOT_REACHED" and output is None
+
+
+def test_truncated_stock_noise_probe_is_not_repaired():
+    random_generator = np.random.default_rng(5730)
+    values = (random_generator.normal(size=92256) + 1j*random_generator.normal(size=92256)).astype(np.complex64)
+    rows, _, _ = diagnosis.analyze_frame(diagnosis.ArrayFile(values), 0, 500_000, "NOISE", 1, None, write_packets=False)
+    assert not any(row["PACKET_OUTPUT"] for row in rows)
+    assert all(not row["HEADER_ACCEPTED"] for row in rows)
+
+
+def test_equal_es_n0_is_not_equal_noise_variance_across_rates():
+    _, slow = diagnosis.synthetic_waveform(20_000, es_n0_db=12)
+    _, fast = diagnosis.synthetic_waveform(500_000, es_n0_db=12)
+    for truth in [slow, fast]:
+        ratio = truth["signal_power"] * (diagnosis.FS / truth["symbol_rate"]) / truth["complex_noise_variance"]
+        assert 10*np.log10(ratio) == pytest.approx(12)
+    assert slow["complex_noise_variance"] / fast["complex_noise_variance"] == pytest.approx(25, rel=.01)