292 lines
9.8 KiB
Python
292 lines
9.8 KiB
Python
"""Проверки Lab046B: несколько полных BPSK/FEC кадров в одном IQ stream."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import inspect
|
||
from functools import lru_cache
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
from experiments import lab046b_continuous_iq as lab046b
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def full_validation() -> lab046b.Lab046bValidation:
|
||
"""Один дорогой интеграционный прогон для всего test-модуля."""
|
||
|
||
return lab046b.run_lab046b_validation()
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def two_frame_ideal() -> lab046b.IqScenarioResult:
|
||
plans = lab046b.build_frame_plans(2)
|
||
return lab046b.run_iq_scenario(
|
||
"TWO_FRAME_IDEAL",
|
||
plans,
|
||
leading_background_samples=137,
|
||
inter_frame_gaps=(211,),
|
||
trailing_background_samples=233,
|
||
)
|
||
|
||
|
||
def test_two_sequential_iq_frames_found_without_known_boundaries() -> None:
|
||
result = two_frame_ideal()
|
||
assert result.frame_count == 2
|
||
assert len(result.receive.accepted_frames) == 2
|
||
assert result.receive.crc_pass_count == 2
|
||
assert all(row.sample_offset_error == 0 for row in result.diagnostics)
|
||
|
||
|
||
def test_five_full_iq_frames_pass_complete_receiver_chain() -> None:
|
||
result = full_validation().ideal
|
||
assert result.frame_count == 5
|
||
assert len(result.receive.accepted_frames) == 5
|
||
assert result.receive.phy_header_pass_count == 5
|
||
assert result.receive.fec_pass_count == 5
|
||
assert result.receive.crc_pass_count == 5
|
||
assert result.receive.crc_fail_count == 0
|
||
assert result.post_fec_protocol_bit_errors == 0
|
||
|
||
|
||
def test_first_frame_offset_and_inter_frame_gaps_are_not_symbol_aligned() -> None:
|
||
result = full_validation().ideal
|
||
assert result.leading_background_samples == 137
|
||
assert result.frame_starts[0] == 137
|
||
assert result.frame_starts[0] % lab046b.lab045.SAMPLES_PER_SYMBOL != 0
|
||
assert result.inter_frame_gaps == (211, 353, 587, 809)
|
||
assert any(
|
||
gap % lab046b.lab045.SAMPLES_PER_SYMBOL != 0
|
||
for gap in result.inter_frame_gaps
|
||
)
|
||
assert [row.symbol_sample_phase for row in result.diagnostics] == [137, 348, 301, 88, 97]
|
||
|
||
|
||
def test_each_real_frame_is_accepted_exactly_once_and_tail_adds_none() -> None:
|
||
result = full_validation().ideal
|
||
accepted = result.receive.accepted_frames
|
||
starts = [frame.refined_start_sample for frame in accepted]
|
||
assert len(starts) == len(set(starts)) == 5
|
||
assert starts == list(result.frame_starts)
|
||
assert result.trailing_background_samples == 233
|
||
assert result.receive.crc_pass_count == 5
|
||
|
||
|
||
def test_multiple_strong_real_peaks_do_not_fail_peak_over_second_gate() -> None:
|
||
result = full_validation().ideal
|
||
assert result.receive.preamble_candidates >= 5
|
||
assert len(result.receive.accepted_frames) == 5
|
||
# Другие настоящие кадры дают почти такой же пик: старый mandatory gate
|
||
# peak/second здесь не прошёл бы, но многокадровое решение его не использует.
|
||
assert max(abs(row.over_second_db) for row in result.diagnostics) < 0.05
|
||
|
||
|
||
def test_local_candidate_finder_keeps_two_equal_strong_peaks() -> None:
|
||
scores = np.asarray([0.0, 0.91, 0.0, 0.90, 0.0], dtype=np.float32)
|
||
assert lab046b._local_candidate_indexes(scores, 0.65).tolist() == [1, 3]
|
||
|
||
|
||
def test_cfo_phase_and_amplitude_recovered_without_truth_in_scanner_api() -> None:
|
||
result = full_validation().impaired
|
||
parameters = set(inspect.signature(lab046b.scan_continuous_iq).parameters)
|
||
assert parameters == {
|
||
"iq_samples",
|
||
"preamble_threshold",
|
||
"max_frames",
|
||
"max_search_iterations",
|
||
}
|
||
assert result.cfo_hz == pytest.approx(0.2)
|
||
assert result.phase_rad == pytest.approx(0.7)
|
||
assert result.amplitude == pytest.approx(0.65)
|
||
assert result.estimated_cfo_hz == pytest.approx(0.2, abs=0.01)
|
||
assert result.estimated_phase_rad == pytest.approx(0.7, abs=0.02)
|
||
assert result.receive.crc_pass_count == 5
|
||
assert all(
|
||
frame.bpsk_polarity == "resolved from preamble"
|
||
for frame in result.receive.accepted_frames
|
||
)
|
||
|
||
|
||
def test_complex_iq_noise_is_deterministic_and_crc_clean() -> None:
|
||
first = np.zeros(4096, dtype=np.complex64)
|
||
second = np.zeros(4096, dtype=np.complex64)
|
||
arguments = {
|
||
"cfo_hz": 0.0,
|
||
"phase_rad": 0.0,
|
||
"amplitude": 1.0,
|
||
"noise_sigma": lab046b.NOISE_SIGMA,
|
||
"noise_seed": lab046b.NOISE_SEED,
|
||
}
|
||
lab046b._apply_impairments_in_place(first, **arguments)
|
||
lab046b._apply_impairments_in_place(second, **arguments)
|
||
assert np.array_equal(first, second)
|
||
|
||
result = full_validation().noisy
|
||
assert result.noise_seed == 46002
|
||
assert result.noise_sigma == pytest.approx(0.17)
|
||
assert result.receive.crc_pass_count == 5
|
||
assert result.receive.crc_fail_count == 0
|
||
assert result.pre_fec_coded_bit_errors == 0
|
||
assert result.post_fec_protocol_bit_errors == 0
|
||
|
||
|
||
def test_malformed_iq_candidate_rejected_then_real_frame_found() -> None:
|
||
result = full_validation().malformed
|
||
assert result.malformed_candidate_start == 137
|
||
assert result.receive.preamble_candidates > 1
|
||
assert result.receive.rejected_candidates >= 1
|
||
assert any(reason.startswith("PHY_HEADER_") for reason in result.receive.rejected_reasons)
|
||
assert len(result.receive.accepted_frames) == 1
|
||
assert result.receive.crc_pass_count == 1
|
||
assert result.diagnostics[0].sample_offset_error == 0
|
||
|
||
|
||
def _fake_frame(start: int, end: int) -> lab046b.IqFrameReception:
|
||
reception = lab046b.lab045.IndependentReception(
|
||
{
|
||
"fec_decode_pass": True,
|
||
"crc": lab046b.lab045.CRC_PASS,
|
||
},
|
||
b"",
|
||
b"",
|
||
)
|
||
return lab046b.IqFrameReception(
|
||
detection_number=1,
|
||
coarse_start_sample=start,
|
||
refined_start_sample=start,
|
||
frame_end_sample=end,
|
||
preamble_score=0.9,
|
||
over_median_db=20.0,
|
||
over_second_db=0.0,
|
||
timing_offset_samples=0,
|
||
symbol_sample_phase=0,
|
||
nominal_sample=0,
|
||
estimated_cfo_hz=0.0,
|
||
estimated_phase_rad=0.0,
|
||
bpsk_polarity="resolved from preamble",
|
||
preamble_errors=0,
|
||
snr_db=20.0,
|
||
post_preamble_soft=np.empty(0),
|
||
reception=reception,
|
||
)
|
||
|
||
|
||
def _fake_candidates(*starts: int) -> tuple[lab046b.IqPreambleCandidate, ...]:
|
||
return tuple(
|
||
lab046b.IqPreambleCandidate(start, 0.9, 20.0, 0.0)
|
||
for start in starts
|
||
)
|
||
|
||
|
||
def test_rejected_candidate_continues_and_cursor_is_monotonic(monkeypatch) -> None:
|
||
monkeypatch.setattr(
|
||
lab046b,
|
||
"_detect_iq_preamble_candidates",
|
||
lambda samples, threshold: (_fake_candidates(10, 30), 0),
|
||
)
|
||
|
||
def decode(samples, candidate, detection_number):
|
||
if candidate.coarse_start_sample == 10:
|
||
return lab046b.CandidateDecode(
|
||
10,
|
||
11,
|
||
False,
|
||
False,
|
||
lab046b.lab045.CRC_NOT_REACHED,
|
||
False,
|
||
"PHY_HEADER_BAD_MAGIC",
|
||
None,
|
||
)
|
||
frame = _fake_frame(30, 60)
|
||
return lab046b.CandidateDecode(
|
||
30,
|
||
60,
|
||
True,
|
||
True,
|
||
lab046b.lab045.CRC_PASS,
|
||
True,
|
||
"CRC_PASS",
|
||
frame,
|
||
)
|
||
|
||
monkeypatch.setattr(lab046b, "_decode_iq_candidate", decode)
|
||
result = lab046b.scan_continuous_iq(np.ones(100, dtype=np.complex64))
|
||
assert result.rejected_candidates == 1
|
||
assert len(result.accepted_frames) == 1
|
||
assert all(
|
||
after > before
|
||
for before, after in zip(result.cursor_history, result.cursor_history[1:])
|
||
)
|
||
|
||
|
||
def test_max_search_iterations_is_fail_closed(monkeypatch) -> None:
|
||
monkeypatch.setattr(
|
||
lab046b,
|
||
"_detect_iq_preamble_candidates",
|
||
lambda samples, threshold: (_fake_candidates(10, 20, 30), 0),
|
||
)
|
||
monkeypatch.setattr(
|
||
lab046b,
|
||
"_decode_iq_candidate",
|
||
lambda samples, candidate, detection_number: lab046b.CandidateDecode(
|
||
candidate.coarse_start_sample,
|
||
candidate.coarse_start_sample + 1,
|
||
False,
|
||
False,
|
||
lab046b.lab045.CRC_NOT_REACHED,
|
||
False,
|
||
"BAD",
|
||
None,
|
||
),
|
||
)
|
||
result = lab046b.scan_continuous_iq(
|
||
np.ones(100, dtype=np.complex64),
|
||
max_search_iterations=1,
|
||
)
|
||
assert result.search_iterations == 1
|
||
assert result.max_search_iterations_reached is True
|
||
assert result.termination_reason == "MAX_SEARCH_ITERATIONS_REACHED"
|
||
|
||
|
||
def test_max_frames_is_fail_closed(monkeypatch) -> None:
|
||
monkeypatch.setattr(
|
||
lab046b,
|
||
"_detect_iq_preamble_candidates",
|
||
lambda samples, threshold: (_fake_candidates(10, 40), 0),
|
||
)
|
||
|
||
def decode(samples, candidate, detection_number):
|
||
start = candidate.coarse_start_sample
|
||
frame = _fake_frame(start, start + 20)
|
||
return lab046b.CandidateDecode(
|
||
start,
|
||
start + 20,
|
||
True,
|
||
True,
|
||
lab046b.lab045.CRC_PASS,
|
||
True,
|
||
"CRC_PASS",
|
||
frame,
|
||
)
|
||
|
||
monkeypatch.setattr(lab046b, "_decode_iq_candidate", decode)
|
||
result = lab046b.scan_continuous_iq(
|
||
np.ones(100, dtype=np.complex64),
|
||
max_frames=1,
|
||
)
|
||
assert len(result.accepted_frames) == 1
|
||
assert result.max_frames_reached is True
|
||
assert result.termination_reason == "MAX_FRAMES_REACHED"
|
||
|
||
|
||
def test_main_iq_memory_accounting_is_complex64() -> None:
|
||
result = full_validation().ideal
|
||
assert result.iq_dtype == "complex64"
|
||
assert result.sample_count == 11_534_330
|
||
assert result.iq_memory_bytes == result.sample_count * np.dtype(np.complex64).itemsize
|
||
assert result.iq_memory_bytes == 92_274_640
|
||
|
||
|
||
def test_small_functional_checks() -> None:
|
||
lab046b.run_functional_tests()
|