Lab046C: analyze weak RF capture offline

2026-09-01 17:09:27 +03:00
parent b6654671b4
commit 9b906c622d
4 changed files with 1682 additions and 1 deletions
--- a/tests/test_lab046c_offline_recovery.py
+++ b/tests/test_lab046c_offline_recovery.py
@@ -0,0 +1,471 @@
+from __future__ import annotations
+
+import inspect
+from pathlib import Path
+import struct
+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_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
+
+
+@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)
+
+
+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_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()