706 lines
28 KiB
Python
706 lines
28 KiB
Python
"""Синтетические end-to-end тесты Lab045 со свёрточной FEC поверх BPSK.
|
||
|
||
Без аппаратуры и данных эксперимента. Свёрточный код проверяется вторым,
|
||
независимо записанным encoder и вручную рассчитанными reference vectors, а не
|
||
только своим decoder. Полный тракт проверяется на идеальном, ошибочном,
|
||
пакетном и noisy soft каналах.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import struct
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
from experiments import lab044_hackrf_bpsk_packet as lab044
|
||
from experiments import lab045_hackrf_bpsk_fec as lab045
|
||
from protocol import bit_interleaver as interleaver
|
||
from protocol import bpsk_radio as radio
|
||
from protocol import convolutional_fec as conv
|
||
from protocol.packet import HEADER_SIZE, build_packet
|
||
|
||
|
||
def deterministic_payload() -> bytes:
|
||
return bytes(range(256))
|
||
|
||
|
||
# ---- Независимый reference encoder с другой реализацией ----------------------
|
||
|
||
|
||
def independent_convolutional_encode(information_bits: list[int]) -> list[int]:
|
||
"""Второй вручную записанный encoder для независимой проверки.
|
||
|
||
Список state содержит шесть предыдущих входных битов, начиная с самого
|
||
свежего. Регистр шага равен [current, prev0..prev4, prev5], а биты
|
||
многочлена читаются MSB-first.
|
||
"""
|
||
|
||
generators = (0o171, 0o133)
|
||
bits = list(information_bits) + [0] * conv.TAIL_BIT_COUNT
|
||
state = [0] * conv.MEMORY
|
||
output: list[int] = []
|
||
for current in bits:
|
||
register = [current] + state # 7 entries, index 0 = current (bit 6)
|
||
for generator in generators:
|
||
generator_bits = [(generator >> k) & 1 for k in range(conv.CONSTRAINT_LENGTH - 1, -1, -1)]
|
||
output.append(sum(a & b for a, b in zip(register, generator_bits)) & 1)
|
||
state = register[: conv.MEMORY] # drop oldest -> new state
|
||
return output
|
||
|
||
|
||
# ---- Корректность свёрточного кода -------------------------------------------
|
||
|
||
|
||
def test_encode_matches_hand_computed_impulse_response() -> None:
|
||
# Импульсная характеристика (171,133) и 6 tail zeros, рассчитана вручную.
|
||
expected = [1, 1, 1, 0, 1, 1, 1, 1, 0, 0, 0, 1, 1, 1]
|
||
result = conv.encode(np.array([1], dtype=np.uint8))
|
||
assert result.tolist() == expected
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("information", "expected_coded"),
|
||
[
|
||
([0], "00000000000000"),
|
||
([1, 0], "1110111100011100"),
|
||
([0, 1], "0011101111000111"),
|
||
([1, 1], "1101010011011011"),
|
||
([1, 0, 1], "111000011110110111"),
|
||
],
|
||
)
|
||
def test_encode_matches_short_reference_vectors(
|
||
information: list[int],
|
||
expected_coded: str,
|
||
) -> None:
|
||
"""Короткие фиксированные векторы защищают соглашение независимо от decoder."""
|
||
|
||
result = conv.encode(np.asarray(information, dtype=np.uint8))
|
||
assert "".join(str(bit) for bit in result.tolist()) == expected_coded
|
||
|
||
|
||
def test_encode_all_zero_is_all_zero() -> None:
|
||
result = conv.encode(np.zeros(20, dtype=np.uint8))
|
||
assert result.tolist() == [0] * conv.encoded_bit_count(20)
|
||
|
||
|
||
def test_encode_all_one_matches_independent_reference() -> None:
|
||
information = [1] * 17
|
||
assert conv.encode(np.array(information, dtype=np.uint8)).tolist() == independent_convolutional_encode(information)
|
||
|
||
|
||
def test_encode_random_matches_independent_reference() -> None:
|
||
generator = np.random.default_rng(45)
|
||
for _ in range(8):
|
||
information = generator.integers(0, 2, generator.integers(1, 300)).astype(np.uint8)
|
||
assert conv.encode(information).tolist() == independent_convolutional_encode(information.tolist())
|
||
|
||
|
||
def test_encode_rejects_empty_and_invalid() -> None:
|
||
with pytest.raises(ValueError):
|
||
conv.encode(np.array([], dtype=np.uint8))
|
||
with pytest.raises(ValueError):
|
||
conv.encode(np.array([2, 0, 1], dtype=np.uint8))
|
||
with pytest.raises(ValueError):
|
||
conv.encode(np.zeros((2, 3), dtype=np.uint8))
|
||
with pytest.raises(ValueError):
|
||
conv.encode(np.array([256, 0, 1]))
|
||
with pytest.raises(ValueError):
|
||
conv.encode(np.array([-1, 0, 1]))
|
||
|
||
|
||
def test_ideal_soft_round_trip_is_exact() -> None:
|
||
generator = np.random.default_rng(7)
|
||
information = generator.integers(0, 2, 2144).astype(np.uint8)
|
||
coded = conv.encode(information)
|
||
soft = 2.0 * coded.astype(np.float64) - 1.0
|
||
decoded = conv.viterbi_decode_soft(soft, len(information))
|
||
assert np.array_equal(decoded, information)
|
||
|
||
|
||
def test_tail_termination_lengths_are_exact() -> None:
|
||
information = np.ones(100, dtype=np.uint8)
|
||
coded = conv.encode(information)
|
||
assert len(coded) == conv.encoded_bit_count(100) == 2 * (100 + 6)
|
||
decoded = conv.viterbi_decode_soft(2.0 * coded - 1.0, 100)
|
||
assert len(decoded) == 100
|
||
|
||
|
||
def test_viterbi_rejects_wrong_length() -> None:
|
||
with pytest.raises(ValueError):
|
||
conv.viterbi_decode_soft(np.zeros(10, dtype=np.float64), 100)
|
||
with pytest.raises(ValueError):
|
||
conv.viterbi_decode_soft(np.zeros(12, dtype=np.float64), 0)
|
||
with pytest.raises(ValueError):
|
||
conv.viterbi_decode_soft(np.array([np.nan] * 14), 1)
|
||
with pytest.raises(ValueError):
|
||
conv.viterbi_decode_hard(np.array([256] * 14), 1)
|
||
|
||
|
||
def test_bpsk_mapping_matches_existing_radio_primitive() -> None:
|
||
bits = np.array([0, 1, 1, 0], dtype=np.uint8)
|
||
expected = np.array([-1.0, 1.0, 1.0, -1.0])
|
||
assert np.array_equal(lab045.modulate_bits_to_ideal_soft(bits), expected)
|
||
assert np.array_equal(radio.bpsk_modulate(bits), expected)
|
||
|
||
|
||
def test_single_and_multiple_coded_errors_are_corrected() -> None:
|
||
generator = np.random.default_rng(11)
|
||
information = generator.integers(0, 2, 512).astype(np.uint8)
|
||
coded = conv.encode(information)
|
||
soft = 2.0 * coded.astype(np.float64) - 1.0
|
||
for error_count in (1, 3, 8):
|
||
corrupted = soft.copy()
|
||
positions = generator.choice(len(corrupted), error_count, replace=False)
|
||
corrupted[positions] = -corrupted[positions]
|
||
assert np.array_equal(conv.viterbi_decode_soft(corrupted, len(information)), information)
|
||
|
||
|
||
# ---- Перемежитель ------------------------------------------------------------
|
||
|
||
|
||
def test_interleaver_inverse_and_permutation_are_exact() -> None:
|
||
for length, columns in [(1, 1), (2, 3), (7, 3), (97, 10), (101, 13), (4300, 64)]:
|
||
permutation = interleaver.build_permutation(length, columns)
|
||
assert np.array_equal(np.sort(permutation), np.arange(length))
|
||
values = np.arange(length, dtype=np.float64) - 0.5
|
||
restored = interleaver.deinterleave(interleaver.interleave(values, permutation), permutation)
|
||
assert np.array_equal(restored, values)
|
||
|
||
|
||
def test_interleaver_rejects_malformed_permutations() -> None:
|
||
values = np.arange(4)
|
||
with pytest.raises(ValueError):
|
||
interleaver.interleave(values, np.array([0, 1, 1, 3]))
|
||
with pytest.raises(ValueError):
|
||
interleaver.deinterleave(values, np.array([0, 1, 2, 4]))
|
||
with pytest.raises(ValueError):
|
||
interleaver.interleave(values, np.array([0.0, 1.0, 2.0, 3.0]))
|
||
|
||
|
||
# ---- Полный тракт Lab045 -----------------------------------------------------
|
||
|
||
|
||
def test_full_chain_ideal_recovers_payload_and_crc() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
assert plan.frame_symbol_count == 1280 + 144 + 4300 # 5724
|
||
soft = lab045.simulate_soft_channel(plan)
|
||
reception = lab045.analyze_post_preamble_soft(soft)
|
||
assert reception.analysis["classification"] == "A"
|
||
assert reception.analysis["crc"] == lab045.CRC_PASS
|
||
assert reception.payload_bytes == plan.payload
|
||
assert plan.waveform_sample_count == 2_306_400
|
||
assert plan.airtime_seconds == pytest.approx(0.2883)
|
||
assert plan.frame_symbol_count / lab045.SYMBOL_RATE == pytest.approx(0.2862)
|
||
assert (
|
||
plan.frame_symbol_count
|
||
+ 2 * lab045.GUARD_SYMBOL_COUNT
|
||
+ lab045.RRC_SPAN_SYMBOLS
|
||
) == 5766
|
||
|
||
|
||
def test_actual_lab044_waveform_builder_has_exact_length() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
tx_cs8 = lab045.build_tx_cs8(plan)
|
||
assert len(tx_cs8) == 2 * 2_306_400
|
||
assert np.max(np.abs(np.frombuffer(tx_cs8, dtype=np.int8).astype(np.int16))) < 127
|
||
|
||
|
||
def test_ideal_iq_adapter_reuses_lab044_sync_and_recovers_crc() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
waveform = lab044.signed_iq_bytes_to_complex64(lab045.build_tx_cs8(plan))
|
||
capture = np.concatenate(
|
||
(
|
||
np.zeros(50_000, dtype=np.complex64),
|
||
waveform,
|
||
np.zeros(50_000, dtype=np.complex64),
|
||
)
|
||
)
|
||
reception = lab045.analyze_capture_independently(capture)
|
||
assert reception.analysis["phy_sync_pass"] is True
|
||
assert reception.analysis["crc"] == lab045.CRC_PASS
|
||
assert reception.payload_bytes == plan.payload
|
||
|
||
|
||
def test_distributed_errors_are_corrected_and_prove_fec() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
soft = lab045.simulate_soft_channel(plan, bit_error_probability=0.006, seed=7)
|
||
reception = lab045.analyze_post_preamble_soft(soft)
|
||
diagnostic = lab045.diagnostic_pre_and_post_fec(reception, soft, plan)
|
||
assert diagnostic["pre_fec_coded"]["bit_errors"] > 0 # raw channel has errors
|
||
assert reception.analysis["crc"] == lab045.CRC_PASS
|
||
assert diagnostic["post_fec_payload"]["bit_perfect"] is True # FEC fixed them
|
||
|
||
|
||
def test_short_burst_survives_interleaver() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
burst = tuple(range(lab045.PHY_HEADER_BIT_COUNT + 500, lab045.PHY_HEADER_BIT_COUNT + 500 + 14))
|
||
soft = lab045.simulate_soft_channel(plan, burst_positions=burst)
|
||
reception = lab045.analyze_post_preamble_soft(soft)
|
||
assert reception.analysis["crc"] == lab045.CRC_PASS
|
||
assert reception.payload_bytes == plan.payload
|
||
|
||
|
||
def test_soft_ambiguous_symbols_near_zero_are_handled() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
soft = lab045.simulate_soft_channel(plan)
|
||
generator = np.random.default_rng(3)
|
||
weak = generator.choice(len(soft), 40, replace=False)
|
||
soft[weak] = 1e-6 * np.sign(soft[weak] + 1e-9) # near-zero magnitude, defined behavior
|
||
reception = lab045.analyze_post_preamble_soft(soft)
|
||
assert reception.analysis["crc"] == lab045.CRC_PASS
|
||
|
||
|
||
def test_exact_zero_soft_values_are_deterministic() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
soft = lab045.simulate_soft_channel(plan)
|
||
soft[lab045.PHY_HEADER_BIT_COUNT + 100 : lab045.PHY_HEADER_BIT_COUNT + 106] = 0.0
|
||
first = lab045.analyze_post_preamble_soft(soft)
|
||
second = lab045.analyze_post_preamble_soft(soft.copy())
|
||
assert first.analysis == second.analysis
|
||
assert first.packet_bytes == second.packet_bytes
|
||
|
||
|
||
def test_extra_trailing_soft_is_ignored_deterministically() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
soft = lab045.simulate_soft_channel(plan)
|
||
padded = np.concatenate((soft, np.zeros(1000, dtype=np.float64)))
|
||
reception = lab045.analyze_post_preamble_soft(padded)
|
||
assert reception.analysis["crc"] == lab045.CRC_PASS
|
||
assert reception.payload_bytes == plan.payload
|
||
|
||
|
||
def test_truncated_coded_frame_fails_closed() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
soft = lab045.simulate_soft_channel(plan)
|
||
truncated = soft[: lab045.PHY_HEADER_BIT_COUNT + 200]
|
||
reception = lab045.analyze_post_preamble_soft(truncated)
|
||
assert reception.analysis["phy_header_pass"] is True
|
||
assert reception.analysis["packet_recovered"] is False
|
||
assert reception.analysis["crc"] == lab045.CRC_NOT_REACHED
|
||
assert reception.packet_bytes is None
|
||
|
||
|
||
def test_crc_corruption_after_valid_fec_decode_is_crc_fail() -> None:
|
||
payload = deterministic_payload()
|
||
packet = bytearray(build_packet(payload, lab045.MESSAGE_TYPE, lab045.SEQUENCE_NUMBER))
|
||
# Меняется payload bit; sync/version/length остаются верны, CRC расходится.
|
||
packet[HEADER_SIZE + 5] ^= 0x01
|
||
information_bits = lab044.bytes_to_bits(bytes(packet))
|
||
coded = conv.encode(information_bits)
|
||
permutation = interleaver.build_permutation(len(coded), lab045.INTERLEAVER_COLUMNS)
|
||
interleaved = interleaver.interleave(coded, permutation)
|
||
header = lab045.encode_phy_header(len(packet))
|
||
soft = lab045.modulate_bits_to_ideal_soft(np.concatenate((header, interleaved)))
|
||
reception = lab045.analyze_post_preamble_soft(soft)
|
||
assert reception.analysis["crc"] == lab045.CRC_FAIL
|
||
assert reception.analysis["classification"] == "B"
|
||
assert reception.payload_bytes is None
|
||
|
||
|
||
# ---- Защищённый PHY header ---------------------------------------------------
|
||
|
||
|
||
def _header_soft(packet_length: int) -> np.ndarray:
|
||
return lab045.modulate_bits_to_ideal_soft(lab045.encode_phy_header(packet_length))
|
||
|
||
|
||
def test_header_majority_survives_one_error_per_repetition_group() -> None:
|
||
soft = _header_soft(268)
|
||
# В каждой побитовой тройке b,b,b повреждена ровно одна копия.
|
||
soft[0::lab045.HEADER_REPETITION] *= -1.0
|
||
header = lab045.decode_phy_header_soft(soft)
|
||
assert header["passed"] is True
|
||
assert header["packet_bytes"] == 268
|
||
|
||
|
||
def test_header_two_errors_in_one_group_are_caught_by_checksum() -> None:
|
||
soft = _header_soft(268)
|
||
# Две из трёх копий одного бита повреждены: majority меняет этот бит.
|
||
target = 32 # бит внутри поля длины
|
||
group_start = target * lab045.HEADER_REPETITION
|
||
soft[group_start] = -soft[group_start]
|
||
soft[group_start + 1] = -soft[group_start + 1]
|
||
header = lab045.decode_phy_header_soft(soft)
|
||
assert header["passed"] is False
|
||
assert header["reason"] in {"CHECKSUM_MISMATCH", "LENGTH_OUT_OF_RANGE", "MAGIC_MISMATCH"}
|
||
|
||
|
||
def test_header_bad_magic_fails_closed() -> None:
|
||
fields = struct.pack(lab045.PHY_HEADER_FIELDS_FORMAT, 0x1234, lab045.PHY_HEADER_VERSION, 268,
|
||
lab045._phy_header_checksum(0x1234, lab045.PHY_HEADER_VERSION, 268))
|
||
bits = np.repeat(lab044.bytes_to_bits(fields), lab045.HEADER_REPETITION)
|
||
header = lab045.decode_phy_header_soft(lab045.modulate_bits_to_ideal_soft(bits))
|
||
assert header["passed"] is False
|
||
assert header["reason"] == "MAGIC_MISMATCH"
|
||
|
||
|
||
def test_header_impossible_length_fails_closed() -> None:
|
||
bad_length = lab045.MAX_PACKET_BYTES + 500
|
||
fields = struct.pack(lab045.PHY_HEADER_FIELDS_FORMAT, lab045.PHY_MAGIC, lab045.PHY_HEADER_VERSION,
|
||
bad_length, lab045._phy_header_checksum(lab045.PHY_MAGIC, lab045.PHY_HEADER_VERSION, bad_length))
|
||
bits = np.repeat(lab044.bytes_to_bits(fields), lab045.HEADER_REPETITION)
|
||
header = lab045.decode_phy_header_soft(lab045.modulate_bits_to_ideal_soft(bits))
|
||
assert header["passed"] is False
|
||
assert header["reason"] == "LENGTH_OUT_OF_RANGE"
|
||
|
||
|
||
def test_header_truncated_fails_closed() -> None:
|
||
header = lab045.decode_phy_header_soft(np.zeros(10, dtype=np.float64))
|
||
assert header["passed"] is False
|
||
assert header["reason"] == "TRUNCATED"
|
||
|
||
|
||
def test_header_too_short_protocol_length_fails_closed() -> None:
|
||
short_length = lab045.MIN_PACKET_BYTES - 1
|
||
fields = struct.pack(
|
||
lab045.PHY_HEADER_FIELDS_FORMAT,
|
||
lab045.PHY_MAGIC,
|
||
lab045.PHY_HEADER_VERSION,
|
||
short_length,
|
||
lab045._phy_header_checksum(
|
||
lab045.PHY_MAGIC,
|
||
lab045.PHY_HEADER_VERSION,
|
||
short_length,
|
||
),
|
||
)
|
||
bits = np.repeat(lab044.bytes_to_bits(fields), lab045.HEADER_REPETITION)
|
||
header = lab045.decode_phy_header_soft(lab045.modulate_bits_to_ideal_soft(bits))
|
||
assert header["passed"] is False
|
||
assert header["reason"] == "LENGTH_OUT_OF_RANGE"
|
||
with pytest.raises(ValueError):
|
||
lab045.encode_phy_header(short_length)
|
||
|
||
|
||
def test_header_unknown_version_fails_closed() -> None:
|
||
version = lab045.PHY_HEADER_VERSION + 1
|
||
fields = struct.pack(
|
||
lab045.PHY_HEADER_FIELDS_FORMAT,
|
||
lab045.PHY_MAGIC,
|
||
version,
|
||
268,
|
||
lab045._phy_header_checksum(lab045.PHY_MAGIC, version, 268),
|
||
)
|
||
bits = np.repeat(lab044.bytes_to_bits(fields), lab045.HEADER_REPETITION)
|
||
header = lab045.decode_phy_header_soft(lab045.modulate_bits_to_ideal_soft(bits))
|
||
assert header["passed"] is False
|
||
assert header["reason"] == "VERSION_MISMATCH"
|
||
|
||
|
||
# ---- Детерминизм и отсутствие утечки TX data --------------------------------
|
||
|
||
|
||
def test_frame_build_is_deterministic() -> None:
|
||
payload = deterministic_payload()
|
||
first = lab045.build_transmission_plan(payload)
|
||
second = lab045.build_transmission_plan(payload)
|
||
assert np.array_equal(first.coded_bits, second.coded_bits)
|
||
assert np.array_equal(first.interleaved_bits, second.interleaved_bits)
|
||
assert np.array_equal(first.frame_bits, second.frame_bits)
|
||
|
||
|
||
def test_analyzer_does_not_receive_tx_payload() -> None:
|
||
# Независимый analyzer принимает только soft values. TX data поступают
|
||
# лишь в отдельную диагностику ПОСЛЕ CRC-решения.
|
||
import inspect
|
||
|
||
analyze_parameters = list(inspect.signature(lab045.analyze_post_preamble_soft).parameters)
|
||
assert analyze_parameters == ["post_preamble_soft"]
|
||
capture_parameters = list(inspect.signature(lab045.analyze_capture_independently).parameters)
|
||
assert capture_parameters == ["samples"]
|
||
diagnostic_parameters = list(inspect.signature(lab045.diagnostic_pre_and_post_fec).parameters)
|
||
assert "plan" in diagnostic_parameters # TX reference only here
|
||
|
||
|
||
def test_no_leakage_order_independent_then_diagnostic() -> None:
|
||
events: list[str] = []
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
soft = lab045.simulate_soft_channel(plan, bit_error_probability=0.006, seed=5)
|
||
|
||
events.append("independent_decode")
|
||
reception = lab045.analyze_post_preamble_soft(soft)
|
||
assert reception.analysis["crc"] in {lab045.CRC_PASS, lab045.CRC_FAIL, lab045.CRC_NOT_REACHED}
|
||
|
||
events.append("diagnostic_read")
|
||
diagnostic = lab045.diagnostic_pre_and_post_fec(reception, soft, plan)
|
||
assert diagnostic["receiver_knew_payload_before_decode"] is False
|
||
assert events == ["independent_decode", "diagnostic_read"]
|
||
|
||
|
||
# ---- Научная проверка soft против hard decision ------------------------------
|
||
|
||
|
||
def test_soft_decision_beats_hard_on_a_real_noisy_case() -> None:
|
||
"""Детерминированный AWGN-случай, где амплитуды дают реальную пользу."""
|
||
|
||
information = np.random.default_rng(123).integers(0, 2, 128, dtype=np.uint8)
|
||
coded = conv.encode(information)
|
||
permutation = interleaver.build_permutation(len(coded), 32)
|
||
interleaved_bits = interleaver.interleave(coded, permutation)
|
||
soft = (
|
||
lab045.modulate_bits_to_ideal_soft(interleaved_bits)
|
||
+ np.random.default_rng(0).normal(0.0, 0.7, size=len(interleaved_bits))
|
||
)
|
||
|
||
soft_decoded = conv.viterbi_decode_soft(
|
||
interleaver.deinterleave(soft, permutation),
|
||
len(information),
|
||
)
|
||
hard_decoded = conv.viterbi_decode_hard(
|
||
interleaver.deinterleave((soft >= 0.0).astype(np.uint8), permutation),
|
||
len(information),
|
||
)
|
||
|
||
assert np.array_equal(soft_decoded, information)
|
||
assert not np.array_equal(hard_decoded, information)
|
||
|
||
|
||
def test_full_chain_with_deterministic_gaussian_noise_passes_crc() -> None:
|
||
plan = lab045.build_transmission_plan(deterministic_payload())
|
||
soft = lab045.simulate_soft_channel(plan, noise_sigma=0.45, seed=45)
|
||
reception = lab045.analyze_post_preamble_soft(soft)
|
||
assert reception.analysis["crc"] == lab045.CRC_PASS
|
||
assert reception.payload_bytes == plan.payload
|
||
|
||
|
||
def test_direct_invocation_smoke() -> None:
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
project_root = Path(__file__).resolve().parents[1]
|
||
result = subprocess.run(
|
||
[sys.executable, str(project_root / "experiments" / "lab045_hackrf_bpsk_fec.py")],
|
||
cwd=project_root,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60,
|
||
)
|
||
assert result.returncode == 0, result.stderr
|
||
assert "total symbols = 5724" in result.stdout
|
||
assert "waveform samples = 2306400" in result.stdout
|
||
assert "synthetic CRC = PASS" in result.stdout
|
||
|
||
|
||
def test_hardware_entry_point_is_fail_closed() -> None:
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
project_root = Path(__file__).resolve().parents[1]
|
||
result = subprocess.run(
|
||
[
|
||
sys.executable,
|
||
str(project_root / "experiments" / "lab045_hackrf_bpsk_fec.py"),
|
||
"--hardware",
|
||
],
|
||
cwd=project_root,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60,
|
||
)
|
||
assert result.returncode == 2
|
||
assert "RF TX = 0" in result.stdout
|
||
|
||
|
||
# ---- Fail-closed аппаратная обвязка без реального RF -------------------------
|
||
|
||
|
||
class HardwareFakeRunner:
|
||
def __init__(self) -> None:
|
||
self.commands: list[list[str]] = []
|
||
|
||
def __call__(self, command: list[str]):
|
||
import subprocess
|
||
|
||
self.commands.append(command)
|
||
if command[0].lower().endswith("hackrf_info.exe"):
|
||
output = (
|
||
f"Serial number: {lab044.lab043.MASTER_SERIAL}\n"
|
||
f"Serial number: {lab044.lab043.SLAVE_SERIAL}\n"
|
||
)
|
||
elif "-t" in command:
|
||
output = "0 underruns\n"
|
||
else:
|
||
output = ""
|
||
return subprocess.CompletedProcess(command, 0, output, "")
|
||
|
||
|
||
class HardwareFakeProcess:
|
||
def __init__(self, startup_failed: bool = False) -> None:
|
||
self.startup_failed = startup_failed
|
||
self.terminated = False
|
||
|
||
def poll(self) -> int | None:
|
||
return 1 if self.startup_failed else (0 if self.terminated else None)
|
||
|
||
def wait(self, timeout: float) -> int:
|
||
del timeout
|
||
return 0
|
||
|
||
def terminate(self) -> None:
|
||
self.terminated = True
|
||
|
||
def close_logs(self) -> None:
|
||
pass
|
||
|
||
|
||
class HardwareFakeStarter:
|
||
def __init__(self, startup_failed: bool = False) -> None:
|
||
self.startup_failed = startup_failed
|
||
self.processes: list[HardwareFakeProcess] = []
|
||
|
||
def __call__(self, command: list[str], stdout_path, stderr_path):
|
||
from pathlib import Path
|
||
|
||
stdout_path.write_text("0 overruns\n", encoding="utf-8")
|
||
stderr_path.write_text("", encoding="utf-8")
|
||
raw_path = Path(command[command.index("-r") + 1])
|
||
raw_path.write_bytes(bytes(2 * lab045.CAPTURE_SAMPLE_COUNT))
|
||
process = HardwareFakeProcess(self.startup_failed)
|
||
self.processes.append(process)
|
||
return process
|
||
|
||
|
||
def allow_hardware_preflight(runner, guard) -> dict:
|
||
del runner
|
||
guard.allow()
|
||
return {
|
||
"passed": True,
|
||
"master_reg29": 0x26,
|
||
"slave_reg29": 0x26,
|
||
"slave_clkin_status": "DETECTED",
|
||
}
|
||
|
||
|
||
def prepare_small_hardware_mock(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr(lab045, "CAPTURE_SAMPLE_COUNT", 16)
|
||
monkeypatch.setattr(lab045, "SAMPLES_PER_SYMBOL", 1)
|
||
monkeypatch.setattr(lab045, "GUARD_SYMBOL_COUNT", 0)
|
||
monkeypatch.setattr(lab045, "RRC_SPAN_SYMBOLS", 0)
|
||
monkeypatch.setattr(
|
||
lab045,
|
||
"build_tx_cs8",
|
||
lambda plan: bytes(2 * plan.waveform_sample_count),
|
||
)
|
||
monkeypatch.setattr(
|
||
lab044.lab043,
|
||
"common_clock_preflight",
|
||
allow_hardware_preflight,
|
||
)
|
||
monkeypatch.setattr(lab044.lab043, "_disable_clock_outputs", lambda runner: [])
|
||
|
||
|
||
def test_hardware_common_clock_failure_means_zero_tx(
|
||
tmp_path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
prepare_small_hardware_mock(monkeypatch)
|
||
runner = HardwareFakeRunner()
|
||
monkeypatch.setattr(
|
||
lab044.lab043,
|
||
"common_clock_preflight",
|
||
lambda runner, guard: (_ for _ in ()).throw(RuntimeError("clock")),
|
||
)
|
||
with pytest.raises(RuntimeError, match="clock"):
|
||
lab045.run_hardware_experiment(
|
||
tmp_path,
|
||
runner=runner,
|
||
payload_factory=lambda count: bytes(count),
|
||
process_checker=lambda: [],
|
||
)
|
||
assert not any("-t" in command for command in runner.commands)
|
||
assert (tmp_path / "failure.json").is_file()
|
||
|
||
|
||
def test_hardware_rx_startup_failure_means_zero_tx(
|
||
tmp_path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
prepare_small_hardware_mock(monkeypatch)
|
||
runner = HardwareFakeRunner()
|
||
starter = HardwareFakeStarter(startup_failed=True)
|
||
with pytest.raises(RuntimeError, match="RX завершился"):
|
||
lab045.run_hardware_experiment(
|
||
tmp_path,
|
||
runner=runner,
|
||
process_starter=starter,
|
||
sleep_function=lambda seconds: None,
|
||
payload_factory=lambda count: bytes(count),
|
||
process_checker=lambda: [],
|
||
)
|
||
assert not any("-t" in command for command in runner.commands)
|
||
assert starter.processes[0].terminated is True
|
||
|
||
|
||
def test_hardware_one_tx_cleanup_and_artifacts(
|
||
tmp_path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
import hashlib
|
||
import json
|
||
|
||
prepare_small_hardware_mock(monkeypatch)
|
||
runner = HardwareFakeRunner()
|
||
starter = HardwareFakeStarter()
|
||
payload = deterministic_payload()
|
||
packet = build_packet(payload, lab045.MESSAGE_TYPE, lab045.SEQUENCE_NUMBER)
|
||
reception = lab045.IndependentReception(
|
||
{
|
||
"classification": "A",
|
||
"crc": lab045.CRC_PASS,
|
||
"phy_sync_pass": True,
|
||
"phy_header_pass": True,
|
||
"fec_decode_pass": True,
|
||
"protocol_header_pass": True,
|
||
},
|
||
packet,
|
||
payload,
|
||
)
|
||
events: list[str] = []
|
||
|
||
def analyzer(samples: np.ndarray) -> lab045.IndependentReception:
|
||
del samples
|
||
events.append("independent_decode")
|
||
return reception
|
||
|
||
def comparer(result, samples, tx_payload_path, tx_packet_path) -> dict:
|
||
del result, samples
|
||
assert events == ["independent_decode"]
|
||
assert tx_payload_path.read_bytes() == payload
|
||
assert tx_packet_path.read_bytes() == packet
|
||
events.append("diagnostic_read")
|
||
return {
|
||
"stage": "AFTER_INDEPENDENT_CRC_DECISION",
|
||
"receiver_knew_payload_before_decode": False,
|
||
"pre_fec_coded": {"bit_errors": 3, "ber": 3 / 4300},
|
||
"post_fec_packet": {"bit_errors": 0, "ber": 0.0},
|
||
"post_fec_payload": {
|
||
"bit_errors": 0,
|
||
"byte_errors": 0,
|
||
"bit_perfect": True,
|
||
},
|
||
}
|
||
|
||
result = lab045.run_hardware_experiment(
|
||
tmp_path,
|
||
runner=runner,
|
||
process_starter=starter,
|
||
sleep_function=lambda seconds: None,
|
||
payload_factory=lambda count: payload,
|
||
analyzer=analyzer,
|
||
comparer=comparer,
|
||
process_checker=lambda: [],
|
||
)
|
||
assert result["classification"] == "A"
|
||
assert result["analysis"]["real_rf_fec_correction_demonstrated"] is True
|
||
assert sum("-t" in command for command in runner.commands) == 1
|
||
assert events == ["independent_decode", "diagnostic_read"]
|
||
assert starter.processes[0].terminated is True
|
||
metadata = json.loads((tmp_path / "metadata.json").read_text(encoding="utf-8"))
|
||
assert metadata["rf_tx_count"] == 1
|
||
assert metadata["cleanup"]["clock_outputs_off"] is True
|
||
assert metadata["cleanup"]["no_lingering_hackrf_transfer"] is True
|
||
assert metadata["artifact_sha256_verified"] is True
|
||
for name, digest in metadata["artifact_sha256"].items():
|
||
path = tmp_path / name
|
||
if path.is_file():
|
||
assert digest == hashlib.sha256(path.read_bytes()).hexdigest()
|