from __future__ import annotations import inspect from pathlib import Path import struct from types import SimpleNamespace import zlib import numpy as np import pytest from experiments import lab046c_offline_recovery as offline def _capture_file(path: Path, sample_count: int = 200) -> Path: path.write_bytes(bytes(2 * sample_count)) return path def _header_soft( *, magic: int | None = None, version: int | None = None, length: int = 268, checksum_delta: int = 0, ) -> np.ndarray: lab045 = offline.lab046c.lab045 magic = lab045.PHY_MAGIC if magic is None else magic version = lab045.PHY_HEADER_VERSION if version is None else version checksum = lab045._phy_header_checksum(magic, version, length) ^ checksum_delta fields = struct.pack(lab045.PHY_HEADER_FIELDS_FORMAT, magic, version, length, checksum) bits = offline.lab046c.lab044.bytes_to_bits(fields) repeated = np.repeat(bits, lab045.HEADER_REPETITION) return 2.0 * repeated.astype(np.float64) - 1.0 def _soft_window(normalized: np.ndarray) -> offline.lab046b.SoftWindow: return offline.lab046b.SoftWindow( True, normalized, 0, 0, 0, { "abs_h": 1.0, "cfo_hz": 0.0, "phase_rad": 0.0, "polarity": "resolved from preamble", "snr_db": 20.0, }, None, ) def _candidate_outcome( candidate: offline.CoarseCandidate, *, accepted: bool, next_sample: int, crc: str | None = None, ) -> offline.CandidateOutcome: crc = crc or ( offline.lab046c.lab045.CRC_PASS if accepted else offline.lab046c.lab045.CRC_NOT_REACHED ) frame = None if accepted: reception = offline.lab046c.lab045.IndependentReception( { "fec_decode_pass": True, "protocol_header_pass": True, "crc": crc, }, b"packet", b"payload", ) frame = offline.lab046b.IqFrameReception( detection_number=1, coarse_start_sample=candidate.sample_offset, refined_start_sample=candidate.sample_offset, frame_end_sample=next_sample, preamble_score=candidate.score, over_median_db=0.0, over_second_db=0.0, timing_offset_samples=0, symbol_sample_phase=0, nominal_sample=0, estimated_cfo_hz=0.0, estimated_phase_rad=0.0, bpsk_polarity="resolved from preamble", preamble_errors=0, snr_db=20.0, post_preamble_soft=np.empty(0), reception=reception, ) return offline.CandidateOutcome( candidate_offset=candidate.sample_offset, refined_start_sample=candidate.sample_offset, next_sample=next_sample, refinement_pass=True, timing_pass=True, carrier_pass=True, phy_header_attempted=True, phy_header_pass=accepted, fec_attempted=accepted, fec_pass=accepted, protocol_parse_pass=accepted, crc=crc, accepted=accepted, reason="CRC_PASS" if accepted else "PHY_HEADER_BAD", frame=frame, maximum_local_window_samples=10, elapsed_seconds=0.001, ) def test_calibration_uses_only_fixed_local_peak_quantile(monkeypatch) -> None: scores = np.linspace(0.001, 0.03, 1000, dtype=np.float32) peaks = np.linspace(0.01, 0.04, 101, dtype=np.float32) monkeypatch.setattr( offline, "_background_peak_scores", lambda path, interval: (scores, peaks, 0.1), ) rule, statistics = offline.calibrate_candidate_rule(Path("unused")) assert rule.local_peak_quantile == 0.999 assert rule.threshold == pytest.approx(np.quantile(peaks, 0.999)) assert statistics.candidate_count == int(np.count_nonzero(peaks >= rule.threshold)) def test_1400mhz_threshold_formula_is_reproducible() -> None: peaks = np.asarray([0.01, 0.02, 0.021, 0.022, 0.024], dtype=np.float32) first = offline.candidate_threshold_from_local_peaks( peaks, quantile=offline.BACKGROUND_1400MHZ_LOCAL_PEAK_QUANTILE, mad_multiplier=offline.BACKGROUND_1400MHZ_MAD_MULTIPLIER, ) second = offline.candidate_threshold_from_local_peaks( peaks.copy(), quantile=offline.BACKGROUND_1400MHZ_LOCAL_PEAK_QUANTILE, mad_multiplier=offline.BACKGROUND_1400MHZ_MAD_MULTIPLIER, ) assert first == second def test_1400mhz_control_background_creates_no_candidates(monkeypatch) -> None: scores = np.asarray([0.001, 0.01, 0.02], dtype=np.float32) peaks = np.asarray([0.02, 0.02711654081940651], dtype=np.float32) monkeypatch.setattr( offline, "_background_peak_scores", lambda path, interval: (scores, peaks, 0.1), ) rule = offline.CandidateRule( "fixed before independent decoding", 0.999, offline.FIXED_1400MHZ_CANDIDATE_THRESHOLD, ) statistics = offline.evaluate_1400mhz_control_background(Path("unused"), rule) assert statistics.candidate_count == 0 assert max(peaks) < rule.threshold def test_control_background_cannot_change_fixed_rule(monkeypatch) -> None: peaks = np.asarray([0.01, 0.02, 0.20], dtype=np.float32) monkeypatch.setattr( offline, "_background_peak_scores", lambda path, interval: (peaks.copy(), peaks, 0.1), ) rule = offline.CandidateRule("fixed", 0.999, 0.025) statistics = offline.evaluate_control_background(Path("unused"), rule) assert rule.threshold == 0.025 assert statistics.candidate_count == 1 def test_deterministic_weak_prbs7_candidate_is_detected() -> None: reference = offline.lab046c.lab044._preamble_reference_samples() reference_decimated = offline.lab046b._decimate_then_downconvert(reference) random_generator = np.random.Generator(np.random.PCG64(46003)) noise = ( random_generator.normal(0.0, 0.03, reference.size + 8_000) + 1j * random_generator.normal(0.0, 0.03, reference.size + 8_000) ).astype(np.complex64) first = noise.copy() second = noise.copy() first[4_000 : 4_000 + reference.size] += np.complex64(0.08) * reference second[4_000 : 4_000 + reference.size] += np.complex64(0.08) * reference first_scores = offline._correlation_scores(first, reference_decimated) second_scores = offline._correlation_scores(second, reference_decimated) assert np.array_equal(first_scores, second_scores) indexes = offline._all_local_peak_indexes(first_scores) assert np.max(first_scores[indexes]) > 0.025 def test_multiple_equal_weak_peaks_need_no_global_peak_second_gate() -> None: scores = np.asarray([0.0, 0.08, 0.0, 0.08, 0.0], dtype=np.float32) indexes = offline._all_local_peak_indexes(scores) assert indexes[scores[indexes] >= 0.025].tolist() == [1, 3] @pytest.mark.parametrize("peak_index", [3, 4]) def test_candidate_at_block_boundary_is_owned_once( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, peak_index: int, ) -> None: capture = _capture_file(tmp_path / f"boundary_{peak_index}.cs8", 160) raw = np.memmap(capture, dtype=np.int8, mode="r") monkeypatch.setattr(offline, "_aligned_overlap", lambda reference_samples: 20) calls = 0 def score_function(samples, reference): nonlocal calls del reference calls += 1 scores = np.zeros(max(5, samples.size // 20), dtype=np.float32) if calls == 1: scores[peak_index] = 0.1 elif peak_index == 4: scores[0] = 0.1 return scores starts = [] for window_start, scores, indexes, _, _ in offline._score_blocks( raw, offline.SampleInterval("test", 0, 160), chunk_samples=100, score_function=score_function, ): starts.extend(window_start + int(index) * 20 for index in indexes if scores[index] >= 0.025) assert len(starts) == len(set(starts)) == 1 def test_noise_candidates_never_become_valid_without_crc( tmp_path: Path, ) -> None: capture = _capture_file(tmp_path / "noise.cs8") candidates = tuple(offline.CoarseCandidate(index * 10, 0.03) for index in range(5)) def reject(raw, sample_count, candidate, number): del raw, sample_count, number return _candidate_outcome(candidate, accepted=False, next_sample=candidate.sample_offset + 1) result = offline.strict_process_candidates(capture, candidates, decoder=reject) assert result.counters.valid_frames == 0 assert result.counters.crc_pass == 0 def test_inconsistent_accepted_flag_cannot_bypass_crc(tmp_path: Path) -> None: capture = _capture_file(tmp_path / "inconsistent_accept.cs8") candidate = offline.CoarseCandidate(10, 0.04) def inconsistent(raw, sample_count, row, number): del raw, sample_count, number return _candidate_outcome( row, accepted=True, next_sample=20, crc=offline.lab046c.lab045.CRC_FAIL, ) result = offline.strict_process_candidates( capture, (candidate,), decoder=inconsistent, ) assert result.counters.crc_pass == 0 assert result.counters.crc_fail == 1 assert result.counters.valid_frames == 0 assert result.accepted_frames == () def test_weak_peak_only_enters_first_stage_without_crc( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: capture = _capture_file(tmp_path / "weak_candidate.cs8") def blocks(raw, interval, chunk_samples): del raw, interval, chunk_samples scores = np.asarray([0.0, 0.04, 0.0], dtype=np.float32) yield 0, scores, np.asarray([1]), 3, 100 monkeypatch.setattr(offline, "_score_blocks", blocks) rule = offline.CandidateRule( "fixed", 0.999, offline.FIXED_1400MHZ_CANDIDATE_THRESHOLD, ) coarse = offline.coarse_search_capture(capture, rule) def reject(raw, sample_count, candidate, number): del raw, sample_count, number return _candidate_outcome( candidate, accepted=False, next_sample=candidate.sample_offset + 1, ) strict = offline.strict_process_candidates( capture, coarse.candidates, decoder=reject, ) assert len(coarse.candidates) == 1 assert strict.counters.valid_frames == 0 assert strict.counters.crc_pass == 0 @pytest.mark.parametrize( ("header", "reason"), [ (_header_soft(magic=0x0000), "PHY_HEADER_MAGIC_MISMATCH"), (_header_soft(version=2), "PHY_HEADER_VERSION_MISMATCH"), (_header_soft(checksum_delta=1), "PHY_HEADER_CHECKSUM_MISMATCH"), (_header_soft(length=1), "PHY_HEADER_LENGTH_OUT_OF_RANGE"), ], ) def test_invalid_phy_header_is_rejected( monkeypatch: pytest.MonkeyPatch, header: np.ndarray, reason: str, ) -> None: preamble = np.ones(offline.lab046c.lab045.PREAMBLE_BIT_COUNT) monkeypatch.setattr(offline.lab046c, "_cs8_window", lambda raw, start, end: np.ones(100, np.complex64)) monkeypatch.setattr( offline.lab046c.lab044.lab043, "_refine_frame_start", lambda samples, reference, coarse: coarse, ) monkeypatch.setattr( offline.lab046b, "_extract_soft_window", lambda samples, start, count: _soft_window(np.concatenate((preamble, header))), ) outcome = offline.decode_candidate_from_memmap( np.zeros(2, dtype=np.int8), 10_000_000, offline.CoarseCandidate(10, 0.03), 1, ) assert outcome.accepted is False assert outcome.reason == reason assert outcome.crc == offline.lab046c.lab045.CRC_NOT_REACHED @pytest.mark.parametrize( ("fec_pass", "crc"), [ (False, offline.lab046c.lab045.CRC_NOT_REACHED), (True, offline.lab046c.lab045.CRC_FAIL), ], ) def test_fec_or_crc_failure_never_becomes_valid( monkeypatch: pytest.MonkeyPatch, fec_pass: bool, crc: str, ) -> None: preamble = np.ones(offline.lab046c.lab045.PREAMBLE_BIT_COUNT) header = _header_soft() normalized = np.concatenate((preamble, header, np.ones(100))) monkeypatch.setattr(offline.lab046c, "_cs8_window", lambda raw, start, end: np.ones(100, np.complex64)) monkeypatch.setattr( offline.lab046c.lab044.lab043, "_refine_frame_start", lambda samples, reference, coarse: 0, ) monkeypatch.setattr( offline.lab046b, "_extract_soft_window", lambda samples, start, count: _soft_window(normalized), ) reception = offline.lab046c.lab045.IndependentReception( { "fec_decode_pass": fec_pass, "protocol_header_pass": fec_pass, "crc": crc, }, None, None, ) monkeypatch.setattr( offline.lab046c.lab045, "analyze_post_preamble_soft", lambda soft: reception, ) outcome = offline.decode_candidate_from_memmap( np.zeros(2, dtype=np.int8), 10_000_000, offline.CoarseCandidate(10, 0.03), 1, ) assert outcome.accepted is False assert outcome.crc == crc def test_rejected_candidate_does_not_hide_later_true_frame(tmp_path: Path) -> None: capture = _capture_file(tmp_path / "later.cs8") candidates = ( offline.CoarseCandidate(10, 0.03), offline.CoarseCandidate(30, 0.04), ) def decoder(raw, sample_count, candidate, number): del raw, sample_count, number return _candidate_outcome( candidate, accepted=candidate.sample_offset == 30, next_sample=60 if candidate.sample_offset == 30 else 11, ) result = offline.strict_process_candidates(capture, candidates, decoder=decoder) assert result.counters.valid_frames == 1 assert result.accepted_frames[0].refined_start_sample == 30 def test_cursor_is_strictly_monotonic(tmp_path: Path) -> None: capture = _capture_file(tmp_path / "cursor.cs8") candidates = tuple(offline.CoarseCandidate(index, 0.03) for index in (10, 20, 30)) def decoder(raw, sample_count, candidate, number): del raw, sample_count, number return _candidate_outcome(candidate, accepted=False, next_sample=candidate.sample_offset + 1) result = offline.strict_process_candidates(capture, candidates, decoder=decoder) assert all(after > before for before, after in zip(result.cursor_history, result.cursor_history[1:])) def test_max_candidates_fails_closed( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: capture = _capture_file(tmp_path / "candidate_limit.cs8") def blocks(raw, interval, chunk_samples): del raw, interval, chunk_samples scores = np.asarray([0.1, 0.2, 0.3], dtype=np.float32) yield 0, scores, np.arange(3), 3, 100 monkeypatch.setattr(offline, "_score_blocks", blocks) result = offline.coarse_search_capture( capture, offline.CandidateRule("test", 0.999, 0.01), max_candidates=2, ) assert result.max_candidates_reached is True assert result.termination_reason == "MAX_CANDIDATES_REACHED" assert len(result.candidates) == 2 def test_coarse_memory_estimate_is_explicitly_conservative( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: capture = _capture_file(tmp_path / "memory.cs8", 600_000) raw = np.memmap(capture, dtype=np.int8, mode="r") monkeypatch.setattr(offline, "_aligned_overlap", lambda reference_samples: 20) blocks = list(offline._score_blocks( raw, offline.SampleInterval("memory", 0, 600_000), chunk_samples=600_020, score_function=lambda samples, reference: np.zeros(1, dtype=np.float32), )) reported = blocks[0][4] input_bytes = 600_000 * np.dtype(np.complex64).itemsize assert reported >= offline.COARSE_TEMPORARY_MEMORY_FACTOR * input_bytes def test_max_search_iterations_fails_closed(tmp_path: Path) -> None: capture = _capture_file(tmp_path / "iteration_limit.cs8") candidates = tuple(offline.CoarseCandidate(index, 0.03) for index in (10, 20, 30)) def decoder(raw, sample_count, candidate, number): del raw, sample_count, number return _candidate_outcome(candidate, accepted=False, next_sample=candidate.sample_offset + 1) result = offline.strict_process_candidates( capture, candidates, max_search_iterations=1, decoder=decoder, ) assert result.max_search_iterations_reached is True assert result.counters.candidates_processed == 1 def test_receiver_api_has_no_tx_or_expected_count_oracle() -> None: parameters = set(inspect.signature(offline.run_independent_recovery).parameters) assert parameters == {"capture_path"} forbidden = { "tx_offsets", "frame_starts", "expected_frame_count", "tx_object", "tx_stream", "fragment_order", } assert parameters.isdisjoint(forbidden) short_parameters = set( inspect.signature(offline.run_1400mhz_independent_recovery).parameters ) assert short_parameters == {"capture_path"} assert short_parameters.isdisjoint(forbidden) def test_post_verdict_diagnostics_refuses_incomplete_verdict(tmp_path: Path) -> None: result = object.__new__(offline.IndependentRecoveryResult) object.__setattr__(result, "completed", False) with pytest.raises(RuntimeError, match="verdict"): offline.diagnose_after_independent_verdict(result, tmp_path) def test_1400mhz_missing_frames_report_error_counts_as_unavailable( tmp_path: Path, ) -> None: payload = bytes(256) plan = offline.lab046c.lab045.build_transmission_plan(payload) (tmp_path / "tx_payload_0.bin").write_bytes(payload) (tmp_path / "tx_packet_0.bin").write_bytes(plan.packet) independent = object.__new__(offline.IndependentRecoveryResult) object.__setattr__(independent, "completed", True) object.__setattr__( independent, "strict", SimpleNamespace(accepted_frames=()), ) diagnostics = offline.diagnose_1400mhz_after_independent_verdict( independent, tmp_path, ) assert diagnostics.matched_valid_frames == 0 assert diagnostics.missed_tx_frames == (0,) assert diagnostics.total_pre_fec_coded_bit_errors is None assert diagnostics.total_post_fec_protocol_bit_errors is None assert diagnostics.total_payload_bit_errors is None def test_1400mhz_main_persists_independent_result_before_tx_reference() -> None: source = inspect.getsource(offline.main) write_position = source.index("write_1400mhz_independent_result") diagnostic_position = source.index("diagnose_1400mhz_after_independent_verdict") assert write_position < diagnostic_position def test_safety_limited_independent_run_does_not_unlock_tx_reference( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: capture = _capture_file(tmp_path / "limited.cs8") rule = offline.CandidateRule("fixed", 0.999, 0.025) background = offline.BackgroundStatistics( offline.SampleInterval("background", 0, 100), 10, 3, 0.01, 0.001, {"0.99": 0.02}, 1, 1.0, 0.01, ) coarse = offline.CoarseSearchResult( 200, 10, (), 100, 1000, 0.1, "MAX_CANDIDATES_REACHED", True, ) monkeypatch.setattr(offline, "calibrate_candidate_rule", lambda path: (rule, background)) monkeypatch.setattr(offline, "evaluate_control_background", lambda path, fixed: background) monkeypatch.setattr(offline, "coarse_search_capture", lambda path, fixed: coarse) result = offline.run_independent_recovery(capture) assert result.completed is False with pytest.raises(RuntimeError, match="verdict"): offline.diagnose_after_independent_verdict(result, tmp_path) def test_lab045_and_lab046b_historical_behavior_is_unchanged() -> None: assert offline.lab046b.IQ_PREAMBLE_THRESHOLD == 0.65 parameters = inspect.signature(offline.lab046b.scan_continuous_iq).parameters assert parameters["preamble_threshold"].default == 0.65 plan = offline.lab046c.lab045.build_transmission_plan(bytes(256)) reception = offline.lab046c.lab045.analyze_post_preamble_soft( offline.lab046c.lab045.simulate_soft_channel(plan) ) assert reception.analysis["crc"] == offline.lab046c.lab045.CRC_PASS def test_false_sequences_are_not_part_of_receiver_api() -> None: parameters = set(inspect.signature(offline.run_independent_recovery).parameters) assert "false_preamble" not in parameters assert "reference_bits" not in parameters def test_direct_invocation_is_rf_free() -> None: offline.run_functional_tests() source = inspect.getsource(offline.run_independent_recovery) assert "hackrf" not in source.lower()