131 lines
5.9 KiB
Python
131 lines
5.9 KiB
Python
"""Быстрые проверки интеграции T2 в строгий приёмник Lab061."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from experiments import lab045_hackrf_bpsk_fec as lab045
|
|
from experiments import lab049_bpsk_symbol_rate as lab049
|
|
from experiments import lab053_packet_processing as packet
|
|
from experiments import lab053_streaming_receiver as streaming
|
|
from experiments import lab059_preamble_carrier_limits as lab059
|
|
from experiments import lab060_decision_directed_carrier_tracking as lab060
|
|
from protocol import bpsk_carrier_tracking as tracking
|
|
|
|
|
|
def full_packet(payload: bytes = bytes(range(256))) -> np.ndarray:
|
|
plan = lab045.build_transmission_plan(payload)
|
|
mode = lab049.PhyMode(500_000)
|
|
raw = np.frombuffer(lab049.build_parameterized_tx_cs8(plan, mode), np.int8)
|
|
return raw[::2].astype(np.float32) + 1j * raw[1::2].astype(np.float32)
|
|
|
|
|
|
def receive(values: np.ndarray, mode: str):
|
|
receiver = streaming.StreamingReceiver(
|
|
processor=packet.PacketProcessor(carrier_tracking_mode=mode)
|
|
)
|
|
output = []
|
|
for start in range(0, len(values), receiver.block_samples):
|
|
output.extend(receiver.push_iq_block(values[start:start + receiver.block_samples]))
|
|
output.extend(receiver.finish())
|
|
return output, receiver
|
|
|
|
|
|
def impaired(cfo_hz: float, seed: int = 61001) -> np.ndarray:
|
|
clean = full_packet()
|
|
indexes = np.arange(len(clean))
|
|
signal = clean * np.exp(1j * (0.37 + 2 * np.pi * cfo_hz * indexes / 8_000_000))
|
|
random_generator = np.random.default_rng(seed)
|
|
noise = random_generator.normal(0, 2.0, len(clean)) + 1j * random_generator.normal(0, 2.0, len(clean))
|
|
return np.concatenate((np.zeros(20_000, np.complex64), signal + noise, np.zeros(20_000, np.complex64))).astype(np.complex64)
|
|
|
|
|
|
def test_t2_coefficients_and_frequency_units():
|
|
parameters = tracking.t2_parameters()
|
|
assert parameters.phase_coefficient == pytest.approx(0.0013324447407406753)
|
|
assert parameters.frequency_coefficient == pytest.approx(8.882964938271169e-7)
|
|
|
|
|
|
def test_p3_and_t2_are_numerically_equivalent_to_lab060():
|
|
symbols, expected, metadata = lab060.clean_rrc_symbols(500_000)
|
|
p3 = lab059.estimate_from_burst(symbols, 500_000, "P3")
|
|
production = tracking.P3CarrierEstimate(
|
|
p3.frequency_hz, p3.gain, p3.phase_fit_rmse_rad, p3.mean_block_coherence
|
|
)
|
|
actual = tracking.track_t2(symbols, 500_000, production)
|
|
reference = lab060.track_symbols(symbols, 500_000, p3.frequency_hz, p3.gain, "T2", lab060.loop_parameters("T2", 0.0005))
|
|
assert metadata["symbols"] == len(symbols)
|
|
assert np.allclose(actual.corrected, reference.corrected, rtol=0, atol=3e-15)
|
|
assert np.array_equal(actual.hard_decisions, reference.hard_decisions)
|
|
|
|
|
|
def test_default_is_explicit_legacy_and_invalid_mode_is_rejected():
|
|
assert packet.PacketProcessor().carrier_tracking_mode == "legacy"
|
|
with pytest.raises(ValueError):
|
|
packet.PacketProcessor(carrier_tracking_mode="automatic")
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ["legacy", "t2"])
|
|
def test_clean_full_500k_packet_passes_header_fec_crc(mode):
|
|
output, receiver = receive(impaired(0.0), mode)
|
|
assert len(output) == receiver.packet_count == 1
|
|
assert output[0].payload == bytes(range(256))
|
|
|
|
|
|
@pytest.mark.parametrize("cfo_hz", [-10.0, -2.0, 2.0, 10.0])
|
|
def test_t2_full_packet_with_positive_and_negative_residual_frequency(cfo_hz):
|
|
output, receiver = receive(impaired(cfo_hz), "t2")
|
|
assert len(output) == receiver.packet_count == 1
|
|
|
|
|
|
def test_t2_is_causal_and_payload_truth_is_not_an_input():
|
|
symbols, expected, unused = lab060.clean_rrc_symbols(500_000)
|
|
p3 = tracking.estimate_p3(symbols[:1280], expected[:1280], 500_000)
|
|
first = tracking.track_t2(symbols, 500_000, p3)
|
|
changed = symbols.copy()
|
|
changed[-1] *= -1
|
|
second = tracking.track_t2(changed, 500_000, p3)
|
|
assert np.array_equal(first.hard_decisions[:-1], second.hard_decisions[:-1])
|
|
assert np.array_equal(first.phase_rad[:-1], second.phase_rad[:-1])
|
|
|
|
|
|
def test_p3_frontend_does_not_read_raw_header_or_payload():
|
|
processor = packet.PacketProcessor(carrier_tracking_mode="t2")
|
|
original = full_packet().astype(np.complex128)
|
|
changed = original.copy()
|
|
boundary = 16 * processor.samples_per_symbol + processor.taps.size + 1279 * processor.samples_per_symbol
|
|
changed[boundary:] += 10.0 + 7.0j
|
|
def p3(values):
|
|
matched = processor.filter(processor.shift(values))
|
|
nominal = 16 * processor.samples_per_symbol + processor.taps.size - 1
|
|
received = matched[nominal:nominal + 1280 * processor.samples_per_symbol:processor.samples_per_symbol]
|
|
return tracking.estimate_p3(received, processor.expected, processor.symbol_rate)
|
|
first, second = p3(original), p3(changed)
|
|
assert first.frequency_hz == pytest.approx(second.frequency_hz, abs=1e-8)
|
|
assert first.gain == pytest.approx(second.gain, abs=1e-8)
|
|
|
|
|
|
def test_error_bursts_fade_and_noise_remain_finite():
|
|
symbols, expected, unused = lab060.clean_rrc_symbols(500_000)
|
|
p3 = tracking.estimate_p3(symbols[:1280], expected[:1280], 500_000)
|
|
damaged = symbols.astype(np.complex128)
|
|
damaged[2000] *= -1
|
|
damaged[2600:2610] *= -1
|
|
damaged[3200:3280] *= -1
|
|
damaged[4000:4200] *= 0.02
|
|
random_generator = np.random.default_rng(61002)
|
|
damaged[4500:4700] = random_generator.normal(size=200) + 1j * random_generator.normal(size=200)
|
|
result = tracking.track_t2(damaged, 500_000, p3)
|
|
assert np.all(np.isfinite(result.corrected))
|
|
assert np.all(np.isfinite(result.frequency_hz))
|
|
assert np.max(np.abs(result.frequency_hz)) < 100.0
|
|
|
|
|
|
def test_background_produces_no_header_or_packet_and_seed_is_deterministic():
|
|
one = np.random.default_rng(61003).normal(size=180_000).astype(np.float32).astype(np.complex64)
|
|
two = np.random.default_rng(61003).normal(size=180_000).astype(np.float32).astype(np.complex64)
|
|
assert np.array_equal(one, two)
|
|
output, receiver = receive(one, "t2")
|
|
assert output == [] and receiver.packet_count == 0
|