98 lines
4.4 KiB
Python
98 lines
4.4 KiB
Python
"""CLOCK-001: отдельная частота АЦП, строгий приём и область метрик."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from scipy.ndimage import map_coordinates
|
|
|
|
from protocol.video003_modem import SAMPLE_RATE_HZ
|
|
from protocol.video005_receiver import IqStreamResult, receive_iq_stream
|
|
from tools import clock001
|
|
from tools.qpsk_rf_common import SEED, payload_for_sequence
|
|
|
|
|
|
def test_sampling_rate_changes_time_base_and_keeps_cfo_independent() -> None:
|
|
tone_hz = 125_000
|
|
cfo_hz = 100
|
|
indexes = np.arange(200_000)
|
|
transmitted = np.exp(2j * np.pi * tone_hz * indexes / SAMPLE_RATE_HZ).astype(np.complex64)
|
|
for ppm in (-50, 0, 50):
|
|
received = clock001.sample_independent_clock(
|
|
transmitted, ppm, cfo_hz=cfo_hz, initial_phase_degrees=45,
|
|
)
|
|
scale = 1 + ppm * 1e-6
|
|
assert len(received) == math.floor((len(transmitted) - 1) * scale) + 1
|
|
measured_hz = (np.angle(np.vdot(received[:-1].astype(np.complex128),
|
|
received[1:].astype(np.complex128)))
|
|
* SAMPLE_RATE_HZ / (2 * np.pi))
|
|
assert measured_hz == pytest.approx((tone_hz + cfo_hz) / scale, abs=0.2)
|
|
assert np.angle(received[0]) == pytest.approx(np.pi / 4, abs=1e-6)
|
|
assert clock001.payload_for_case(256, 7) == payload_for_sequence(SEED, 7)
|
|
|
|
|
|
@pytest.mark.parametrize("ppm", (-50, 50))
|
|
def test_long_packets_reacquire_and_decode_at_primary_limit(ppm: int) -> None:
|
|
fixture = clock001.build_fixture(1012)
|
|
received_iq = clock001.sample_independent_clock(fixture.iq, ppm)
|
|
result = receive_iq_stream(received_iq, decoder_backend="legacy")
|
|
summary = clock001.summarize_case(
|
|
result, fixture, sample_clock_offset_ppm=ppm, cfo_hz=0,
|
|
initial_phase_degrees=0, ebn0_db=None, receiver_time_s=1.0,
|
|
received_samples=len(received_iq),
|
|
)
|
|
assert summary["exact_packets"] == 3
|
|
assert summary["exact_sequences"] == [0, 1, 2]
|
|
assert summary["reacquisition_per_packet"] is True
|
|
assert summary["injected_timing_drift_across_packet_symbols"] == pytest.approx(-ppm * 1e-6 * 7572)
|
|
|
|
|
|
@pytest.mark.parametrize("ppm", (-100, 100))
|
|
def test_long_packet_stress_failure_is_not_declared_exact(ppm: int) -> None:
|
|
fixture = clock001.build_fixture(1012, packet_count=1)
|
|
received_iq = clock001.sample_independent_clock(fixture.iq, ppm)
|
|
result = receive_iq_stream(received_iq, decoder_backend="legacy")
|
|
summary = clock001.summarize_case(
|
|
result, fixture, sample_clock_offset_ppm=ppm, cfo_hz=0,
|
|
initial_phase_degrees=0, ebn0_db=None, receiver_time_s=1.0,
|
|
received_samples=len(received_iq),
|
|
)
|
|
assert summary["exact_packets"] == 0
|
|
assert summary["missing_sequences"] == [0]
|
|
assert summary["initial_timing_error_symbols_median"] is None
|
|
assert summary["final_timing_error_symbols_median"] is None
|
|
assert abs(summary["injected_timing_drift_across_packet_symbols"]) > 0.75
|
|
|
|
|
|
def test_cubic_resampling_confirms_long_packet_stress_boundary() -> None:
|
|
"""Проверяет вывод вторым интерполятором, без TX-подсказок приёмнику."""
|
|
fixture = clock001.build_fixture(1012, packet_count=1)
|
|
for ppm, expected_packets in ((-100, 0), (-50, 1), (50, 1), (100, 0)):
|
|
scale = 1 + ppm * 1e-6
|
|
count = math.floor((len(fixture.iq) - 1) * scale) + 1
|
|
positions = np.arange(count, dtype=np.float64) / scale
|
|
received_iq = (
|
|
map_coordinates(fixture.iq.real, [positions], order=3, mode="nearest")
|
|
+ 1j * map_coordinates(fixture.iq.imag, [positions], order=3, mode="nearest")
|
|
).astype(np.complex64)
|
|
result = receive_iq_stream(received_iq, decoder_backend="legacy")
|
|
assert len(result.packets) == expected_packets
|
|
|
|
|
|
def test_reference_is_used_only_after_receiver(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
fixture = clock001.build_fixture(256, packet_count=1)
|
|
calls = []
|
|
|
|
def fake_receiver(iq: np.ndarray) -> IqStreamResult:
|
|
calls.append(iq)
|
|
return IqStreamResult((), 0, 0, {})
|
|
|
|
monkeypatch.setattr(clock001, "receive_iq_stream", fake_receiver)
|
|
summary = clock001.evaluate_case(fixture, ppm=20)
|
|
assert len(calls) == 1
|
|
assert isinstance(calls[0], np.ndarray)
|
|
assert summary["exact_packets"] == 0
|
|
assert summary["missing_sequences"] == [0]
|