"""Синтетические 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()