"""Проверки protocol/bpsk_radio.py. Модуль был собран переносом функций из Lab018, Lab019 и Lab023. Главная задача этих проверок — доказать, что перенос ничего не изменил: результаты сравниваются с оригиналами, загруженными из лабораторных в изоляции, без выполнения их модульного кода. """ from __future__ import annotations import ast import pathlib import struct import numpy as np import pytest from protocol import bpsk_radio as radio from protocol.packet import MESSAGE_TYPE_IMAGE_FRAGMENT, build_packet, parse_packet PROJECT_ROOT = pathlib.Path(__file__).resolve().parent.parent SHARED_CONSTANTS = { "RADIO_SYNC_WORD", "PREAMBLE_BIT_COUNT", "SYMBOL_RATE", "CFO_REFINEMENT_HALF_WIDTH_HZ", "CFO_REFINEMENT_STEP_HZ", } def _load_isolated(relative_path: str, function_names: set[str]) -> dict: """Выполнить только заданные функции лабораторной, без кода модуля. Лабораторные при импорте создают каталоги и пишут файлы, поэтому импортировать их в тестах нельзя. """ path = PROJECT_ROOT / relative_path tree = ast.parse(path.read_text(encoding="utf-8")) kept = [ node for node in tree.body if (isinstance(node, ast.FunctionDef) and node.name in function_names) or (isinstance(node, ast.Assign) and ast.unparse(node.targets[0]) in SHARED_CONSTANTS) ] namespace: dict = {"np": np, "struct": struct} exec(compile(ast.Module(body=kept, type_ignores=[]), str(path), "exec"), namespace) return namespace @pytest.fixture(scope="module") def original_lab023() -> dict: return _load_isolated( "experiments/lab023_guarded_cfo_correction.py", { "bytes_to_bits", "bits_to_bytes", "bpsk_modulate", "bpsk_demodulate", "estimate_carrier_parameters", "correct_phase_and_frequency", }, ) @pytest.fixture(scope="module") def original_lab019() -> dict: return _load_isolated( "experiments/lab019_bpsk_receiver.py", { "build_frame_marker", "find_radio_frame", "bytes_to_bits", "bits_to_bytes", "bpsk_modulate", "bpsk_demodulate", "root_raised_cosine_taps", }, ) @pytest.fixture(scope="module") def original_lab018() -> dict: return _load_isolated("experiments/lab018_bpsk_radio_frame.py", {"root_raised_cosine_taps"}) def _identical(left, right) -> bool: if isinstance(left, np.ndarray) or isinstance(right, np.ndarray): return np.array_equal(np.asarray(left), np.asarray(right)) if isinstance(left, dict): return sorted(left) == sorted(right) and all(_identical(left[k], right[k]) for k in left) if isinstance(left, (tuple, list)): return len(left) == len(right) and all(_identical(a, b) for a, b in zip(left, right)) return left == right # ---------------------------------------------------------------- эквивалентность @pytest.mark.parametrize("length", [1, 7, 64, 255]) def test_bit_conversion_matches_lab023(original_lab023: dict, length: int) -> None: data = bytes(np.random.default_rng(length).integers(0, 256, size=length, dtype=np.uint8)) ours = radio.bytes_to_bits(data) assert _identical(ours, original_lab023["bytes_to_bits"](data)) assert _identical(radio.bits_to_bytes(ours), original_lab023["bits_to_bytes"](ours)) @pytest.mark.parametrize("length", [8, 64, 256]) def test_modulation_matches_lab023(original_lab023: dict, length: int) -> None: bits = np.random.default_rng(length).integers(0, 2, size=length).astype(np.uint8) ours = radio.bpsk_modulate(bits) assert _identical(ours, original_lab023["bpsk_modulate"](bits)) assert _identical(radio.bpsk_demodulate(ours), original_lab023["bpsk_demodulate"](ours)) @pytest.mark.parametrize("rolloff,sps,span", [(0.35, 32, 10), (0.35, 128, 10), (0.5, 64, 8)]) def test_shaping_filter_matches_lab018(original_lab018: dict, rolloff: float, sps: int, span: int) -> None: assert _identical( radio.root_raised_cosine_taps(rolloff, sps, span), original_lab018["root_raised_cosine_taps"](rolloff, sps, span), ) def test_frame_marker_matches_lab019(original_lab019: dict) -> None: """Маркер возвращает пару: биты и символы. Совпасть должны обе.""" assert _identical(radio.build_frame_marker(), original_lab019["build_frame_marker"]()) @pytest.mark.parametrize("seed", [1, 2, 3]) def test_carrier_estimation_matches_lab023(original_lab023: dict, seed: int) -> None: """Оценка ухода частоты обязана совпасть при произвольных фазе и уходе.""" rng = np.random.default_rng(seed) _, marker = radio.build_frame_marker() phase = float(rng.uniform(-3.0, 3.0)) increment = float(rng.uniform(-0.005, 0.005)) impaired = marker * np.exp(1j * (phase + increment * np.arange(len(marker)))) impaired = impaired + 0.02 * rng.standard_normal(len(marker)) assert _identical( radio.estimate_carrier_parameters(impaired, marker), original_lab023["estimate_carrier_parameters"](impaired, marker), ) assert _identical( radio.correct_phase_and_frequency(impaired, phase, increment), original_lab023["correct_phase_and_frequency"](impaired, phase, increment), ) # ---------------------------------------------------------------- новые функции кадра @pytest.mark.parametrize("payload_size", [1, 17, 512, radio.MAXIMUM_PROTOCOL_PACKET_BYTES]) def test_frame_roundtrip(payload_size: int) -> None: packet = bytes(np.random.default_rng(payload_size).integers(0, 256, size=payload_size, dtype=np.uint8)) bits, symbols, marker = radio.build_radio_frame(packet) assert len(bits) == radio.radio_frame_bit_count(payload_size) assert len(symbols) == len(bits) assert len(marker) == radio.MARKER_BIT_COUNT assert radio.parse_radio_frame(bits) == packet def test_frame_rejects_sizes_outside_range() -> None: with pytest.raises(ValueError): radio.build_radio_frame(b"") with pytest.raises(ValueError): radio.build_radio_frame(b"\x00" * (radio.MAXIMUM_PROTOCOL_PACKET_BYTES + 1)) with pytest.raises(TypeError): radio.build_radio_frame("не байты") def test_frame_with_broken_sync_word_is_rejected() -> None: bits, _, _ = radio.build_radio_frame(b"payload") broken = bits.copy() broken[radio.PREAMBLE_BIT_COUNT] ^= 1 assert radio.parse_radio_frame(broken) is None def test_truncated_frame_is_rejected() -> None: bits, _, _ = radio.build_radio_frame(b"payload") assert radio.parse_radio_frame(bits[:10]) is None assert radio.parse_radio_frame(bits[:-8]) is None def test_frame_carries_a_protocol_packet() -> None: """Кадр должен переносить настоящий пакет так, чтобы CRC32 сошлась.""" packet = build_packet(b"fragment payload", MESSAGE_TYPE_IMAGE_FRAGMENT, 5) bits, _, _ = radio.build_radio_frame(packet) assert parse_packet(radio.parse_radio_frame(bits)).payload == b"fragment payload" def test_frame_search_finds_the_start_in_a_shaped_signal() -> None: """Сквозная проверка: формирование, фильтрация и поиск кадра.""" samples_per_symbol = 32 taps = radio.root_raised_cosine_taps(0.35, samples_per_symbol, 10) packet = build_packet(b"HELLO SDR", MESSAGE_TYPE_IMAGE_FRAGMENT, 1) bits, symbols, marker = radio.build_radio_frame(packet) upsampled = np.zeros(len(symbols) * samples_per_symbol, dtype=complex) upsampled[::samples_per_symbol] = symbols matched = np.convolve(np.convolve(upsampled, taps, mode="full"), taps, mode="full") found = radio.find_radio_frame(matched, marker, samples_per_symbol) start = found["start_symbol_index"] recovered = found["symbol_samples"][start : start + len(bits)] assert radio.parse_radio_frame(radio.bpsk_demodulate(recovered)) == packet