diff --git a/PROJECT_LOG.md b/PROJECT_LOG.md index eb71c18..7b72bec 100644 --- a/PROJECT_LOG.md +++ b/PROJECT_LOG.md @@ -883,3 +883,79 @@ tools/quick_gate.py` PASS, `python -m pytest tests/ -q` = 482 passed, Следующий этап определяется отдельной постановкой после анализа RESULT C; повторный RF TX, изменение PRBS7 threshold, gain или detector автоматически не разрешены. + +# Запись 030 + +## Дата + +1 сентября 2026 года + +## Тема + +Lab046C: автономная слабосигнальная обработка исходного RX capture, offline +результат FAIL без нового RF. + +## Выполнено + +Исходный аппаратный результат сохранён без изменения: **ORIGINAL LAB046C +HARDWARE RESULT = C, RF TX COUNT = 1, ORIGINAL HARDWARE RESULT CHANGED = NO.** +HackRF не запускался, RF TX в этом этапе равен 0, второй аппаратный или offline +независимый прогон не выполнялся. + +Добавлен отдельный `experiments/lab046c_offline_recovery.py`. Исторический +Lab046B threshold `0.65` и поведение Lab045 не менялись. Новый первый каскад +зафиксирован до полного прогона только по фону первых 0,5 с: 99,9-й процентиль +локальных PRBS7-максимумов, threshold `0.025011632591485977`, median +`0.00838765874505043`, MAD `0.003039387986063957`, 5193 peaks, 6 candidates, +12,0 candidates/s. На независимом контрольном фоне 21--24 с уже фиксированное +правило дало 35501 peak, 301 candidate, 100,333333 candidates/s; после этого +порог не изменялся. + +Выполнен ровно один полный independent pass сохранённых 192 000 000 cs8 +samples. API не получал TX stream/object, offsets, ожидаемые 68, fragment order, +packets или coded bits. Coarse stage: 291446 local peaks и 3025 candidates. +Strict stage обработал все 3025: refinement/timing/carrier PASS 3025, +PHY HEADER PASS 0, PHY HEADER FAIL 3025 (`MAGIC_MISMATCH`), FEC attempts 0, +CRC 0/0, valid frames 0. `VALID FRAGMENTS = 0`, `OBJECT COMPLETE = NO`, +`recovered_object.bin` не создан. **OFFLINE REPROCESSING OF ORIGINAL RX CAPTURE += FAIL; OBJECT RECOVERED = NO.** + +Только после независимого verdict выполнена TX-reference диагностика. Все 68 +реальных стартов имели candidate в пределах +-40 samples; корреляция +min/median/max `0.059379209/0.088661019/0.126557365`, все выше фиксированного +threshold. Однако preamble errors min/median/max `341/430/554` из 1280, +PHY-header errors `30/54/81` из 144, SNR +`-19.282130/-12.829480/-9.195846 dB`. Все 68 диагностических заголовков дали +MAGIC_MISMATCH. `FALSE CRC PASS = 0`, independently valid frames matching TX = +0, missed TX frames = 68. Ни один пропущенный кадр не добавлялся обратно в +`ObjectAssembler`. + +Производительность: background calibration 0,269617 с, control 1,031204 с, +coarse search 7,721316 с, strict processing 836,340048 с, среднее 0,276476 с на +кандидат, полный прогон 845,427589 с, то есть 35,226150 длительности 24-секундного +capture. RX использован как memmap, coarse block 16 000 000 complex64 samples; +консервативная оценка временной памяти 1 545 705 764 байта, strict local window +не более 594400 samples / 4 755 200 байт. Полный complex64 capture в память не +загружался. + +Добавлены 24 теста statistical calibration/control, deterministic weak PRBS7, +CRC-only acceptance, ошибок magic/version/checksum/length, FEC/CRC fail, +block-boundary ownership, monotonic cursor, safety limits, отсутствия TX oracle +и регрессии Lab045/Lab046B. До полного прогона verifier нашёл и помог закрыть +fail-open status safety-limit и заниженную оценку памяти; повторная проверка +дала чистый pre-run verdict. + +Финальный software gate после фиксации offline FAIL: `python +tools/quick_gate.py` PASS, 506 проверок и 9 функциональных лабораторий; +`python -m pytest tests/ -q` = 506 passed. Предупреждение `.pytest_cache` не +повлияло на результат. `git diff --check` чист, кроме информационных LF/CRLF +notices. Финальный read-only verifier независимо сверил JSON, документацию, +68/68 post-verdict mapping, счётчики, времена и память; научных расхождений не +найдено. + +## Следующий этап + +Сначала анализируется честный offline FAIL. Порог этого independent pass не +менять задним числом и второй улучшенный pass автоматически не запускать. +Повторный RF TX, gain ladder, JPEG, video, QPSK, erasure FEC и ARQ без новой +отдельной постановки запрещены. diff --git a/docs/lab046_current_state.md b/docs/lab046_current_state.md index 1ca7727..15c860f 100644 --- a/docs/lab046_current_state.md +++ b/docs/lab046_current_state.md @@ -20,6 +20,17 @@ **OFFICIAL RESULT CHANGED BY POST-RUN DIAGNOSTICS = NO.** +**OFFLINE DETECTOR STUDY = COMPLETE.** + +**OFFLINE REPROCESSING OF ORIGINAL RX CAPTURE = FAIL.** + +**OFFLINE VALID FRAMES = 0. OFFLINE OBJECT COMPLETE = NO.** + +**ORIGINAL HARDWARE RESULT CHANGED = NO.** + +**CURRENT SOFTWARE GATE = PASS: quick_gate 506, pytest 506 passed, +git diff --check clean.** + Lab046A/B остаются только программными лабами. Для Lab046C выполнен ровно один заранее разрешённый непрерывный RF-сеанс без повтора. Ни JPEG, ни видео, ни QPSK не запускались. @@ -105,6 +116,69 @@ read-only диагностика с TX reference. Она обнаружила с исход остаётся `LAB046C HARDWARE RESULT = C`, `RF TX COUNT = 1`, `OBJECT COMPLETE = NO`. +## Lab046C: автономная слабосигнальная обработка исходного RX capture + +После аппаратного RESULT C реализован отдельный offline-модуль +`experiments/lab046c_offline_recovery.py`. Lab045 и строгий многокадровый +порог Lab046B `IQ_PREAMBLE_THRESHOLD = 0.65` не изменялись. Новый режим не +содержит HackRF-команд и читает только ранее сохранённый `rx_stream.cs8` через +`np.memmap`. + +Первый каскад был зафиксирован до полного прогона по калибровочному фону +`0.0--0.5 s`: порог равен 99,9-му процентилю локальных PRBS7-максимумов. +Получено `threshold = 0.025011632591485977`, correlation median +`0.00838765874505043`, MAD `0.003039387986063957`, 5193 локальных максимума, +6 кандидатов, `12.0 candidates/s`. После фиксации правило проверено на +отдельном контрольном фоне `21.0--24.0 s` и больше не менялось: 35501 локальный +максимум, 301 кандидат, `100.333333 candidates/s`. Временные границы TX +использовались только для выбора участков, где передачи заведомо не было, и не +передавались поиску кадров. + +Выполнен ровно один независимый полный offline-прогон 192 000 000 samples без +`tx_stream.cs8`, `tx_object.bin`, TX offsets, ожидаемого числа 68, fragment +order, packets или coded bits. Первый каскад насчитал 291446 локальных +максимумов и 3025 кандидатов выше порога. Все 3025 прошли refinement, timing и +оценку carrier/CFO/phase, но все 3025 были отвергнуты как +`PHY_HEADER_MAGIC_MISMATCH`. `PHY HEADER PASS = 0`, `FEC ATTEMPTS = 0`, +`CRC PASS = 0`, `VALID FRAMES = 0`, `VALID FRAGMENTS = 0`, +`OBJECT COMPLETE = NO`. `recovered_object.bin` не создавался. + +Только после завершённого independent verdict была открыта TX reference. +Диагностика показала, что каждый из 68 настоящих стартов имел first-stage +кандидат в пределах `+-40 samples`; корреляция на этих кандидатах +min/median/max `0.059379209/0.088661019/0.126557365`, то есть все они были выше +зафиксированного порога. `FALSE CRC PASS = 0`, но совпавших independently valid +frames также 0, пропущены все 68 TX frames. На известных только post-verdict +позициях локальная физическая экстракция прошла для 68/68, однако ошибки +min/median/max составили `341/430/554` из 1280 бит преамбулы и `30/54/81` из +144 бит PHY header. Оценочный SNR min/median/max +`-19.282130/-12.829480/-9.195846 dB`; все заголовки дали `MAGIC_MISMATCH`. +Следовательно, новый first-stage detector обнаружил слабые следы, но качество +демодуляции оказалось недостаточным даже для защищённого PHY header. + +Производительность: calibration `0.269617 s`, control `1.031204 s`, coarse +search `7.721316 s`, strict candidate processing `836.340048 s`, в среднем +`0.276476 s/candidate`, полный прогон `845.427589 s`. Capture длительностью +24 s обработан за `35.226150` его длительности. RX остаётся `memmap`; самый +большой coarse `complex64` block содержит 16 000 000 samples (128 000 000 +байт), консервативная оценка временных массивов SciPy `1 545 705 764` байта. +Максимальное strict local window: 594400 samples, 4 755 200 байт. Полный +192M-sample `complex64` массив не создавался. + +Артефакты результата находятся в +`data/processed/lab046c_offline_recovery/`: `independent_result.json`, +`post_verdict_diagnostics.json`, `tx_reference_stage_diagnostics.json`. +Правильная итоговая формулировка: **ORIGINAL LAB046C HARDWARE RESULT = C; +OFFLINE REPROCESSING OF ORIGINAL RX CAPTURE = FAIL; OBJECT RECOVERED = NO; +ORIGINAL HARDWARE RESULT CHANGED = NO.** Второй независимый прогон с новым +порогом не выполнялся. + +Финальный software gate после результата: `python tools/quick_gate.py` PASS, +506 проверок и 9 функциональных лабораторий; `python -m pytest tests/ -q` = +506 passed. Единственное предупреждение связано с недоступным `.pytest_cache` +и не является отказом. `git diff --check` чист, кроме информационных LF/CRLF +notices. + ## Назначение Lab046 переходит от одиночной 256-байтной полезной нагрузки к передаче @@ -398,4 +472,6 @@ Goodput 6,69 кбит/с — это честная baseline при 20 ксимв Для Lab046A/B RF TX = 0; первый аппаратный Lab046C завершён с RESULT C и одним RF TX без повтора. Erasure-код 12+3 GF(256) не задействован. Порог PRBS7, gain, PHY, FEC и detector после результата не менялись. Следующий этап -определяется отдельной постановкой; новый RF-сеанс автоматически не разрешён. +определяется отдельной постановкой. Offline weak-signal прогон с заранее +зафиксированным правилом уже завершён как FAIL и не повторяется с другим +порогом. Новый RF-сеанс автоматически не разрешён. diff --git a/experiments/lab046c_offline_recovery.py b/experiments/lab046c_offline_recovery.py new file mode 100644 index 0000000..25807bc --- /dev/null +++ b/experiments/lab046c_offline_recovery.py @@ -0,0 +1,1058 @@ +"""Lab046C: автономное восстановление объекта из сохранённого слабого IQ. + +Модуль не содержит аппаратных команд. Первый каскад калибруется только по +заведомо фоновому участку RX, второй каскад принимает кадр исключительно после +PHY header, soft Viterbi, protocol parser и CRC PASS. TX reference доступен +только отдельной функции диагностики после завершённого независимого verdict. +""" + +from __future__ import annotations + +import argparse +from dataclasses import asdict, dataclass, replace +import hashlib +import json +import math +from pathlib import Path +import sys +import time +from typing import Callable + +import numpy as np + +if __package__ in (None, ""): + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from experiments import lab046_bpsk_object_transport as lab046 +from experiments import lab046b_continuous_iq as lab046b +from experiments import lab046c_hackrf_bpsk_object as lab046c + + +DEFAULT_CAPTURE_PATH = Path( + r"C:\Users\user\Documents\SDR_Rover_Lab046_raw" + r"\20260901_145339_bpsk_object\rx_stream.cs8" +) +DEFAULT_REFERENCE_DIRECTORY = DEFAULT_CAPTURE_PATH.parent +DEFAULT_OUTPUT_DIRECTORY = Path("data/processed/lab046c_offline_recovery") + +# По плану аппаратного опыта RX стартовал за 0.75 s до TX, а после TX оставалось +# около 3.65 s. Более узкие интервалы оставляют защитный запас с обеих сторон. +CALIBRATION_START_SAMPLE = 0 +CALIBRATION_END_SAMPLE = 4_000_000 # 0.0 .. 0.5 s +CONTROL_START_SAMPLE = 168_000_000 # 21.0 .. 24.0 s +CONTROL_END_SAMPLE = 192_000_000 + +BACKGROUND_LOCAL_PEAK_QUANTILE = 0.999 +DETECTION_CHUNK_SAMPLES = 16_000_000 +MAX_CANDIDATES = 10_000 +MAX_SEARCH_ITERATIONS = 10_000 +MAX_VALID_FRAMES = 256 +REFINEMENT_MARGIN_SAMPLES = 4_000 +# resample_poly/fftconvolve одновременно создают несколько крупных временных +# массивов. Измеренный tracemalloc synthetic peak был 9.61 размера complex64 +# блока; коэффициент 12 сохраняет явный консервативный запас. +COARSE_TEMPORARY_MEMORY_FACTOR = 12 + + +@dataclass(frozen=True) +class SampleInterval: + name: str + start_sample: int + end_sample: int + + @property + def sample_count(self) -> int: + return self.end_sample - self.start_sample + + @property + def duration_seconds(self) -> float: + return self.sample_count / lab046c.SAMPLE_RATE_HZ + + +CALIBRATION_INTERVAL = SampleInterval( + "CALIBRATION_BACKGROUND", + CALIBRATION_START_SAMPLE, + CALIBRATION_END_SAMPLE, +) +CONTROL_INTERVAL = SampleInterval( + "CONTROL_BACKGROUND", + CONTROL_START_SAMPLE, + CONTROL_END_SAMPLE, +) + + +@dataclass(frozen=True) +class CandidateRule: + method: str + local_peak_quantile: float + threshold: float + + +@dataclass(frozen=True) +class BackgroundStatistics: + interval: SampleInterval + correlation_score_count: int + local_peak_count: int + correlation_median: float + correlation_mad: float + local_peak_quantiles: dict[str, float] + candidate_count: int + candidates_per_second: float + elapsed_seconds: float + + +@dataclass(frozen=True) +class CoarseCandidate: + sample_offset: int + score: float + + +@dataclass(frozen=True) +class CoarseSearchResult: + sample_count: int + total_local_peaks: int + candidates: tuple[CoarseCandidate, ...] + maximum_window_samples: int + estimated_maximum_temporary_bytes: int + elapsed_seconds: float + termination_reason: str + max_candidates_reached: bool + + +@dataclass(frozen=True) +class CandidateOutcome: + candidate_offset: int + refined_start_sample: int + next_sample: int + refinement_pass: bool + timing_pass: bool + carrier_pass: bool + phy_header_attempted: bool + phy_header_pass: bool + fec_attempted: bool + fec_pass: bool + protocol_parse_pass: bool + crc: str + accepted: bool + reason: str + frame: lab046b.IqFrameReception | None + maximum_local_window_samples: int + elapsed_seconds: float + + +@dataclass(frozen=True) +class StrictCounters: + candidates_processed: int + rejected_during_refinement: int + refinement_pass: int + timing_pass: int + carrier_pass: int + phy_header_pass: int + phy_header_fail: int + fec_attempts: int + fec_pass: int + fec_fail: int + protocol_parse_pass: int + crc_pass: int + crc_fail: int + valid_frames: int + + +@dataclass(frozen=True) +class StrictSearchResult: + counters: StrictCounters + accepted_frames: tuple[lab046b.IqFrameReception, ...] + outcomes: tuple[CandidateOutcome, ...] + cursor_history: tuple[int, ...] + elapsed_seconds: float + average_candidate_seconds: float + maximum_local_window_samples: int + maximum_local_window_bytes: int + termination_reason: str + max_search_iterations_reached: bool + max_valid_frames_reached: bool + + +@dataclass(frozen=True) +class ObjectVerdict: + valid_fragments: int + duplicates: int + conflicting_duplicates: int + rejected_fragments: int + object_id: int | None + declared_fragment_count: int | None + received_unique_fragments: int + missing_fragments: tuple[int, ...] + object_complete: bool + recovered_object: bytes | None + + +@dataclass(frozen=True) +class IndependentRecoveryResult: + completed: bool + capture_path: Path + capture_samples: int + capture_duration_seconds: float + rule: CandidateRule + calibration: BackgroundStatistics + control: BackgroundStatistics + coarse: CoarseSearchResult + strict: StrictSearchResult + object_verdict: ObjectVerdict + full_elapsed_seconds: float + processing_to_capture_ratio: float + + +@dataclass(frozen=True) +class PostVerdictDiagnostics: + stage: str + expected_tx_frames: int + independent_valid_frames_matching_tx: int + missed_tx_frames: tuple[int, ...] + false_crc_pass: int + sample_offset_errors: tuple[int, ...] + total_pre_fec_coded_bit_errors: int + total_post_fec_protocol_bit_errors: int + object_size: int | None + recovered_sha256: str | None + tx_sha256: str + sha_match: bool + object_byte_errors: int | None + object_bit_errors: int | None + offline_recovery: str + + +ScoreFunction = Callable[[np.ndarray, np.ndarray], np.ndarray] + + +def _sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _amplitude_db(ratio: float) -> float: + return 20.0 * math.log10(max(float(ratio), 1e-300)) + + +def _validate_capture(capture_path: Path) -> tuple[np.memmap, int]: + path = Path(capture_path) + file_bytes = path.stat().st_size + if file_bytes <= 0 or file_bytes % 2: + raise ValueError("RX capture должен иметь положительный чётный размер") + raw = np.memmap(path, dtype=np.int8, mode="r") + return raw, file_bytes // 2 + + +def _correlation_scores( + samples: np.ndarray, + reference_decimated: np.ndarray, +) -> np.ndarray: + rx_decimated = lab046b._decimate_then_downconvert(samples) + if rx_decimated.size < reference_decimated.size: + return np.empty(0, dtype=np.float32) + scores = np.asarray( + lab046c.lab044.lab043._normalized_sliding_correlation( + rx_decimated, + reference_decimated, + ), + dtype=np.float32, + ) + scores[~np.isfinite(scores)] = 0.0 + np.clip(scores, 0.0, 1.0, out=scores) + return scores + + +def _all_local_peak_indexes(scores: np.ndarray) -> np.ndarray: + values = np.asarray(scores, dtype=np.float32) + if values.ndim != 1 or values.size == 0: + return np.empty(0, dtype=np.int64) + left = np.empty_like(values) + right = np.empty_like(values) + left[0] = -np.inf + left[1:] = values[:-1] + right[-1] = -np.inf + right[:-1] = values[1:] + return np.flatnonzero((values >= left) & (values > right)) + + +def _aligned_overlap(reference_samples: int) -> int: + decimation = lab046c.lab044.lab043.DETECTION_DECIMATION + required = reference_samples + decimation + return ((required + decimation - 1) // decimation) * decimation + + +def _score_blocks( + raw: np.memmap, + interval: SampleInterval, + *, + chunk_samples: int = DETECTION_CHUNK_SAMPLES, + score_function: ScoreFunction = _correlation_scores, +): + reference = lab046c.lab044._preamble_reference_samples() + reference_decimated = lab046b._decimate_then_downconvert(reference) + decimation = lab046c.lab044.lab043.DETECTION_DECIMATION + overlap = _aligned_overlap(reference.size) + chunk_samples = (chunk_samples // decimation) * decimation + if chunk_samples <= overlap: + raise ValueError("chunk_samples должен быть больше PRBS7 reference с overlap") + if not 0 <= interval.start_sample < interval.end_sample <= raw.size // 2: + raise ValueError("Фоновый интервал выходит за границы capture") + step = chunk_samples - overlap + for window_start in range(interval.start_sample, interval.end_sample, step): + window_end = min(interval.end_sample, window_start + chunk_samples) + ownership_end = ( + interval.end_sample + if window_end == interval.end_sample + else window_start + step + ) + samples = lab046c._cs8_window(raw, window_start, window_end) + scores = score_function(samples, reference_decimated) + owned_count = min( + scores.size, + (ownership_end - window_start + decimation - 1) // decimation, + ) + local_indexes = _all_local_peak_indexes(scores) + local_indexes = local_indexes[local_indexes < owned_count] + yield ( + window_start, + scores[:owned_count], + local_indexes, + int(samples.size), + int( + COARSE_TEMPORARY_MEMORY_FACTOR * samples.nbytes + + reference_decimated.nbytes + + scores.nbytes + + math.ceil(samples.size / decimation) + * np.dtype(np.complex64).itemsize + ), + ) + if window_end == interval.end_sample: + break + + +def _background_peak_scores( + capture_path: Path, + interval: SampleInterval, +) -> tuple[np.ndarray, np.ndarray, float]: + raw, _ = _validate_capture(capture_path) + started = time.perf_counter() + score_blocks: list[np.ndarray] = [] + peak_blocks: list[np.ndarray] = [] + for _, scores, indexes, _, _ in _score_blocks(raw, interval): + score_blocks.append(np.asarray(scores, dtype=np.float32).copy()) + peak_blocks.append(np.asarray(scores[indexes], dtype=np.float32).copy()) + all_scores = np.concatenate(score_blocks) if score_blocks else np.empty(0, np.float32) + all_peaks = np.concatenate(peak_blocks) if peak_blocks else np.empty(0, np.float32) + if all_scores.size == 0 or all_peaks.size == 0: + raise RuntimeError("Фоновый интервал не дал корреляционной статистики") + return all_scores, all_peaks, time.perf_counter() - started + + +def _background_statistics( + interval: SampleInterval, + scores: np.ndarray, + peaks: np.ndarray, + threshold: float, + elapsed_seconds: float, +) -> BackgroundStatistics: + median = float(np.median(scores)) + mad = float(np.median(np.abs(scores - median))) + quantiles = { + "0.99": float(np.quantile(peaks, 0.99)), + "0.999": float(np.quantile(peaks, 0.999)), + "0.9995": float(np.quantile(peaks, 0.9995)), + "0.9999": float(np.quantile(peaks, 0.9999)), + "1.0": float(np.max(peaks)), + } + candidate_count = int(np.count_nonzero(peaks >= threshold)) + return BackgroundStatistics( + interval=interval, + correlation_score_count=int(scores.size), + local_peak_count=int(peaks.size), + correlation_median=median, + correlation_mad=mad, + local_peak_quantiles=quantiles, + candidate_count=candidate_count, + candidates_per_second=candidate_count / interval.duration_seconds, + elapsed_seconds=elapsed_seconds, + ) + + +def calibrate_candidate_rule( + capture_path: Path, + interval: SampleInterval = CALIBRATION_INTERVAL, +) -> tuple[CandidateRule, BackgroundStatistics]: + """Зафиксировать threshold только по локальным максимумам calibration background.""" + + scores, peaks, elapsed = _background_peak_scores(capture_path, interval) + threshold = float(np.quantile(peaks, BACKGROUND_LOCAL_PEAK_QUANTILE)) + rule = CandidateRule( + method="calibration local-peak quantile", + local_peak_quantile=BACKGROUND_LOCAL_PEAK_QUANTILE, + threshold=threshold, + ) + return rule, _background_statistics(interval, scores, peaks, threshold, elapsed) + + +def evaluate_control_background( + capture_path: Path, + rule: CandidateRule, + interval: SampleInterval = CONTROL_INTERVAL, +) -> BackgroundStatistics: + """Проверить уже зафиксированное правило, не изменяя threshold.""" + + scores, peaks, elapsed = _background_peak_scores(capture_path, interval) + return _background_statistics(interval, scores, peaks, rule.threshold, elapsed) + + +def coarse_search_capture( + capture_path: Path, + rule: CandidateRule, + *, + max_candidates: int = MAX_CANDIDATES, + chunk_samples: int = DETECTION_CHUNK_SAMPLES, +) -> CoarseSearchResult: + """Найти first-stage candidates во всём capture без TX offsets/count.""" + + if max_candidates < 0: + raise ValueError("max_candidates не может быть отрицательным") + raw, sample_count = _validate_capture(capture_path) + whole = SampleInterval("FULL_CAPTURE", 0, sample_count) + started = time.perf_counter() + candidates: list[CoarseCandidate] = [] + total_local_peaks = 0 + maximum_window = 0 + maximum_temporary = 0 + termination = "END_OF_CAPTURE" + max_reached = False + decimation = lab046c.lab044.lab043.DETECTION_DECIMATION + for window_start, scores, indexes, window_samples, temporary_bytes in _score_blocks( + raw, + whole, + chunk_samples=chunk_samples, + ): + total_local_peaks += int(indexes.size) + maximum_window = max(maximum_window, window_samples) + maximum_temporary = max(maximum_temporary, temporary_bytes) + selected = indexes[scores[indexes] >= np.float32(rule.threshold)] + for index in selected: + if len(candidates) >= max_candidates: + termination = "MAX_CANDIDATES_REACHED" + max_reached = True + break + candidates.append(CoarseCandidate( + sample_offset=window_start + int(index) * decimation, + score=float(scores[index]), + )) + if max_reached: + break + candidates.sort(key=lambda row: row.sample_offset) + return CoarseSearchResult( + sample_count=sample_count, + total_local_peaks=total_local_peaks, + candidates=tuple(candidates), + maximum_window_samples=maximum_window, + estimated_maximum_temporary_bytes=maximum_temporary, + elapsed_seconds=time.perf_counter() - started, + termination_reason=termination, + max_candidates_reached=max_reached, + ) + + +def _failed_outcome( + candidate: CoarseCandidate, + *, + refined: int, + next_sample: int, + refinement_pass: bool, + timing_pass: bool, + carrier_pass: bool, + phy_header_attempted: bool, + phy_header_pass: bool, + reason: str, + maximum_window: int, + started: float, +) -> CandidateOutcome: + return CandidateOutcome( + candidate_offset=candidate.sample_offset, + refined_start_sample=refined, + next_sample=next_sample, + refinement_pass=refinement_pass, + timing_pass=timing_pass, + carrier_pass=carrier_pass, + phy_header_attempted=phy_header_attempted, + phy_header_pass=phy_header_pass, + fec_attempted=False, + fec_pass=False, + protocol_parse_pass=False, + crc=lab046c.lab045.CRC_NOT_REACHED, + accepted=False, + reason=reason, + frame=None, + maximum_local_window_samples=maximum_window, + elapsed_seconds=time.perf_counter() - started, + ) + + +def decode_candidate_from_memmap( + raw: np.memmap, + sample_count: int, + candidate: CoarseCandidate, + detection_number: int, +) -> CandidateOutcome: + """Строго проверить один candidate локальными IQ-окнами.""" + + started = time.perf_counter() + reference = lab046c.lab044._preamble_reference_samples() + probe_symbols = lab046c.lab045.PREAMBLE_BIT_COUNT + lab046c.lab045.PHY_HEADER_BIT_COUNT + probe_waveform = ( + (probe_symbols + 2 * lab046c.lab045.GUARD_SYMBOL_COUNT) + * lab046c.lab045.SAMPLES_PER_SYMBOL + + lab046c.lab045.RRC_SPAN_SYMBOLS * lab046c.lab045.SAMPLES_PER_SYMBOL + ) + probe_start = max(0, candidate.sample_offset - REFINEMENT_MARGIN_SAMPLES) + probe_end = min( + sample_count, + candidate.sample_offset + probe_waveform + REFINEMENT_MARGIN_SAMPLES, + ) + probe_iq = lab046c._cs8_window(raw, probe_start, probe_end) + maximum_window = int(probe_iq.size) + local_coarse = candidate.sample_offset - probe_start + try: + local_refined = lab046c.lab044.lab043._refine_frame_start( + probe_iq, + reference, + local_coarse, + ) + except (ValueError, IndexError): + return _failed_outcome( + candidate, + refined=candidate.sample_offset, + next_sample=candidate.sample_offset + 1, + refinement_pass=False, + timing_pass=False, + carrier_pass=False, + phy_header_attempted=False, + phy_header_pass=False, + reason="REFINEMENT_FAILED", + maximum_window=maximum_window, + started=started, + ) + refined_global = probe_start + int(local_refined) + probe = lab046b._extract_soft_window(probe_iq, int(local_refined), probe_symbols) + if not probe.passed: + return _failed_outcome( + candidate, + refined=refined_global, + next_sample=candidate.sample_offset + 1, + refinement_pass=True, + timing_pass=False, + carrier_pass=False, + phy_header_attempted=False, + phy_header_pass=False, + reason=probe.reason or "PROBE_FAILED", + maximum_window=maximum_window, + started=started, + ) + header_soft = probe.normalized_real[ + lab046c.lab045.PREAMBLE_BIT_COUNT : probe_symbols + ] + header = lab046c.lab045.decode_phy_header_soft(header_soft) + if not header["passed"]: + return _failed_outcome( + candidate, + refined=refined_global, + next_sample=candidate.sample_offset + 1, + refinement_pass=True, + timing_pass=True, + carrier_pass=True, + phy_header_attempted=True, + phy_header_pass=False, + reason=f"PHY_HEADER_{header['reason']}", + maximum_window=maximum_window, + started=started, + ) + + packet_bytes = int(header["packet_bytes"]) + coded_length = lab046c.lab045.conv.encoded_bit_count(packet_bytes * 8) + frame_symbols = ( + lab046c.lab045.PREAMBLE_BIT_COUNT + + lab046c.lab045.PHY_HEADER_BIT_COUNT + + coded_length + ) + waveform_samples = ( + (frame_symbols + 2 * lab046c.lab045.GUARD_SYMBOL_COUNT) + * lab046c.lab045.SAMPLES_PER_SYMBOL + + lab046c.lab045.RRC_SPAN_SYMBOLS * lab046c.lab045.SAMPLES_PER_SYMBOL + ) + full_end = refined_global + waveform_samples + if refined_global < 0 or full_end > sample_count: + return _failed_outcome( + candidate, + refined=refined_global, + next_sample=candidate.sample_offset + 1, + refinement_pass=True, + timing_pass=True, + carrier_pass=True, + phy_header_attempted=True, + phy_header_pass=True, + reason="TRUNCATED_FULL_IQ_WINDOW", + maximum_window=maximum_window, + started=started, + ) + full_iq = lab046c._cs8_window(raw, refined_global, full_end) + maximum_window = max(maximum_window, int(full_iq.size)) + full = lab046b._extract_soft_window(full_iq, 0, frame_symbols) + if not full.passed: + return _failed_outcome( + candidate, + refined=refined_global, + next_sample=candidate.sample_offset + 1, + refinement_pass=True, + timing_pass=True, + carrier_pass=True, + phy_header_attempted=True, + phy_header_pass=True, + reason=full.reason or "FULL_EXTRACTION_FAILED", + maximum_window=maximum_window, + started=started, + ) + + post_preamble = full.normalized_real[lab046c.lab045.PREAMBLE_BIT_COUNT:] + reception = lab046c.lab045.analyze_post_preamble_soft(post_preamble) + crc = str(reception.analysis["crc"]) + fec_pass = bool(reception.analysis["fec_decode_pass"]) + protocol_pass = bool(reception.analysis["protocol_header_pass"]) + accepted = crc == lab046c.lab045.CRC_PASS + preamble_hard = ( + full.normalized_real[: lab046c.lab045.PREAMBLE_BIT_COUNT] >= 0.0 + ).astype(np.uint8) + preamble_errors = int(np.count_nonzero( + preamble_hard != lab046c.lab044.preamble_bits() + )) + frame = lab046b.IqFrameReception( + detection_number=detection_number, + coarse_start_sample=candidate.sample_offset, + refined_start_sample=refined_global, + frame_end_sample=full_end, + preamble_score=candidate.score, + over_median_db=float("nan"), + over_second_db=float("nan"), + timing_offset_samples=full.timing_offset_samples, + symbol_sample_phase=full.symbol_sample_phase, + nominal_sample=full.nominal_sample, + estimated_cfo_hz=float(full.carrier["cfo_hz"]), + estimated_phase_rad=float(full.carrier["phase_rad"]), + bpsk_polarity=str(full.carrier["polarity"]), + preamble_errors=preamble_errors, + snr_db=float(full.carrier["snr_db"]), + post_preamble_soft=post_preamble, + reception=reception, + ) + return CandidateOutcome( + candidate_offset=candidate.sample_offset, + refined_start_sample=refined_global, + next_sample=full_end if accepted else candidate.sample_offset + 1, + refinement_pass=True, + timing_pass=True, + carrier_pass=True, + phy_header_attempted=True, + phy_header_pass=True, + fec_attempted=True, + fec_pass=fec_pass, + protocol_parse_pass=protocol_pass, + crc=crc, + accepted=accepted, + reason="CRC_PASS" if accepted else f"CRC_{crc}", + frame=frame, + maximum_local_window_samples=maximum_window, + elapsed_seconds=time.perf_counter() - started, + ) + + +CandidateDecoder = Callable[ + [np.memmap, int, CoarseCandidate, int], + CandidateOutcome, +] + + +def strict_process_candidates( + capture_path: Path, + candidates: tuple[CoarseCandidate, ...], + *, + max_search_iterations: int = MAX_SEARCH_ITERATIONS, + max_valid_frames: int = MAX_VALID_FRAMES, + decoder: CandidateDecoder = decode_candidate_from_memmap, +) -> StrictSearchResult: + """Проверить кандидатов по offset; cursor всегда движется только вперёд.""" + + if max_search_iterations < 0 or max_valid_frames < 0: + raise ValueError("Safety limits не могут быть отрицательными") + raw, sample_count = _validate_capture(capture_path) + started = time.perf_counter() + cursor = 0 + cursor_history = [0] + outcomes: list[CandidateOutcome] = [] + accepted_frames: list[lab046b.IqFrameReception] = [] + termination = "END_OF_CANDIDATES" + max_iterations_reached = False + max_frames_reached = False + for candidate in sorted(candidates, key=lambda row: row.sample_offset): + if candidate.sample_offset < cursor: + continue + if len(outcomes) >= max_search_iterations: + termination = "MAX_SEARCH_ITERATIONS_REACHED" + max_iterations_reached = True + break + if len(accepted_frames) >= max_valid_frames: + termination = "MAX_VALID_FRAMES_REACHED" + max_frames_reached = True + break + outcome = decoder(raw, sample_count, candidate, len(outcomes) + 1) + outcomes.append(outcome) + if outcome.accepted and outcome.frame is not None: + accepted_frames.append(outcome.frame) + cursor = max(cursor + 1, int(outcome.next_sample)) + cursor_history.append(cursor) + + counters = StrictCounters( + candidates_processed=len(outcomes), + rejected_during_refinement=sum(not row.refinement_pass for row in outcomes), + refinement_pass=sum(row.refinement_pass for row in outcomes), + timing_pass=sum(row.timing_pass for row in outcomes), + carrier_pass=sum(row.carrier_pass for row in outcomes), + phy_header_pass=sum(row.phy_header_pass for row in outcomes), + phy_header_fail=sum( + row.phy_header_attempted and not row.phy_header_pass for row in outcomes + ), + fec_attempts=sum(row.fec_attempted for row in outcomes), + fec_pass=sum(row.fec_pass for row in outcomes), + fec_fail=sum(row.fec_attempted and not row.fec_pass for row in outcomes), + protocol_parse_pass=sum(row.protocol_parse_pass for row in outcomes), + crc_pass=sum(row.crc == lab046c.lab045.CRC_PASS for row in outcomes), + crc_fail=sum(row.crc == lab046c.lab045.CRC_FAIL for row in outcomes), + valid_frames=len(accepted_frames), + ) + elapsed = time.perf_counter() - started + maximum_window = max( + (row.maximum_local_window_samples for row in outcomes), + default=0, + ) + return StrictSearchResult( + counters=counters, + accepted_frames=tuple(accepted_frames), + outcomes=tuple(outcomes), + cursor_history=tuple(cursor_history), + elapsed_seconds=elapsed, + average_candidate_seconds=elapsed / len(outcomes) if outcomes else 0.0, + maximum_local_window_samples=maximum_window, + maximum_local_window_bytes=maximum_window * np.dtype(np.complex64).itemsize, + termination_reason=termination, + max_search_iterations_reached=max_iterations_reached, + max_valid_frames_reached=max_frames_reached, + ) + + +def assemble_independent_object( + frames: tuple[lab046b.IqFrameReception, ...], +) -> ObjectVerdict: + """Собрать объект только по payload независимо принятых CRC PASS frames.""" + + assembler = lab046.ObjectAssembler() + object_id: int | None = None + declared_count: int | None = None + for frame in frames: + payload = frame.reception.payload_bytes + fragment = lab046.recover_fragment_from_payload(payload) if payload is not None else None + if fragment is not None and object_id is None: + object_id = int(fragment.image_id) + declared_count = int(fragment.fragment_count) + assembler.offer(fragment) + recovered = assembler.recovered_object() + return ObjectVerdict( + valid_fragments=assembler.counters.accepted, + duplicates=assembler.counters.duplicates, + conflicting_duplicates=assembler.counters.conflicting, + rejected_fragments=assembler.counters.rejected, + object_id=object_id, + declared_fragment_count=declared_count, + received_unique_fragments=assembler.counters.accepted, + missing_fragments=tuple(assembler.missing_indexes()), + object_complete=assembler.object_complete(), + recovered_object=recovered, + ) + + +def run_independent_recovery(capture_path: Path) -> IndependentRecoveryResult: + """Один полный RX-only прогон; API не принимает ни одного TX oracle.""" + + full_started = time.perf_counter() + capture_path = Path(capture_path) + _, sample_count = _validate_capture(capture_path) + rule, calibration = calibrate_candidate_rule(capture_path) + control = evaluate_control_background(capture_path, rule) + coarse = coarse_search_capture(capture_path, rule) + if coarse.max_candidates_reached: + strict = strict_process_candidates( + capture_path, + (), + max_search_iterations=0, + max_valid_frames=0, + ) + verdict = assemble_independent_object(()) + else: + strict = strict_process_candidates(capture_path, coarse.candidates) + verdict = assemble_independent_object(strict.accepted_frames) + elapsed = time.perf_counter() - full_started + completed = bool( + not coarse.max_candidates_reached + and not strict.max_search_iterations_reached + and not strict.max_valid_frames_reached + and strict.termination_reason == "END_OF_CANDIDATES" + ) + return IndependentRecoveryResult( + completed=completed, + capture_path=capture_path, + capture_samples=sample_count, + capture_duration_seconds=sample_count / lab046c.SAMPLE_RATE_HZ, + rule=rule, + calibration=calibration, + control=control, + coarse=coarse, + strict=strict, + object_verdict=verdict, + full_elapsed_seconds=elapsed, + processing_to_capture_ratio=( + elapsed / (sample_count / lab046c.SAMPLE_RATE_HZ) + ), + ) + + +def diagnose_after_independent_verdict( + independent: IndependentRecoveryResult, + reference_directory: Path, +) -> PostVerdictDiagnostics: + """Открыть TX reference только после неизменяемого independent verdict.""" + + if not independent.completed: + raise RuntimeError("Независимый verdict ещё не завершён") + reference_directory = Path(reference_directory) + tx_object = (reference_directory / "tx_object.bin").read_bytes() + metadata = json.loads( + (reference_directory / "tx_metadata.json").read_text(encoding="utf-8") + ) + tx_frames = lab046.build_object_frames(tx_object, lab046c.OBJECT_ID) + by_packet = {frame.plan.packet: (index, frame) for index, frame in enumerate(tx_frames)} + matched_indexes: list[int] = [] + starts_by_index: list[tuple[int, int]] = [] + false_crc_pass = 0 + pre_fec_errors = 0 + post_fec_errors = 0 + for received in independent.strict.accepted_frames: + packet = received.reception.packet_bytes + matched = by_packet.get(packet) if packet is not None else None + if matched is None: + false_crc_pass += 1 + continue + index, frame = matched + matched_indexes.append(index) + starts_by_index.append((index, received.refined_start_sample)) + comparison = lab046c.lab045.diagnostic_pre_and_post_fec( + received.reception, + received.post_preamble_soft, + frame.plan, + ) + pre_fec_errors += int(comparison["pre_fec_coded"]["bit_errors"] or 0) + post_fec_errors += int(comparison["post_fec_packet"]["bit_errors"] or 0) + expected_indexes = set(range(len(tx_frames))) + missed = tuple(sorted(expected_indexes - set(matched_indexes))) + sample_errors: tuple[int, ...] = () + if starts_by_index: + frame_samples = tx_frames[0].plan.waveform_sample_count + base_estimates = [start - index * frame_samples for index, start in starts_by_index] + base = int(round(float(np.median(base_estimates)))) + sample_errors = tuple( + start - (base + index * frame_samples) + for index, start in starts_by_index + ) + + recovered = independent.object_verdict.recovered_object + recovered_sha = _sha256(recovered) if recovered is not None else None + tx_sha = str(metadata["object"]["sha256"]) + if recovered is not None and len(recovered) == len(tx_object): + received_array = np.frombuffer(recovered, dtype=np.uint8) + tx_array = np.frombuffer(tx_object, dtype=np.uint8) + xor = np.bitwise_xor(received_array, tx_array) + byte_errors = int(np.count_nonzero(xor)) + bit_errors = int(np.unpackbits(xor).sum()) + else: + byte_errors = None + bit_errors = None + sha_match = recovered_sha == tx_sha if recovered_sha is not None else False + if ( + independent.object_verdict.object_complete + and false_crc_pass == 0 + and len(recovered or b"") == lab046c.OBJECT_SIZE_BYTES + and sha_match + and byte_errors == 0 + ): + offline_recovery = "PASS" + elif independent.strict.counters.valid_frames: + offline_recovery = "PARTIAL" + else: + offline_recovery = "FAIL" + return PostVerdictDiagnostics( + stage="AFTER_INDEPENDENT_OBJECT_VERDICT", + expected_tx_frames=len(tx_frames), + independent_valid_frames_matching_tx=len(matched_indexes), + missed_tx_frames=missed, + false_crc_pass=false_crc_pass, + sample_offset_errors=sample_errors, + total_pre_fec_coded_bit_errors=pre_fec_errors, + total_post_fec_protocol_bit_errors=post_fec_errors, + object_size=len(recovered) if recovered is not None else None, + recovered_sha256=recovered_sha, + tx_sha256=tx_sha, + sha_match=sha_match, + object_byte_errors=byte_errors, + object_bit_errors=bit_errors, + offline_recovery=offline_recovery, + ) + + +def _independent_summary(result: IndependentRecoveryResult) -> dict: + return { + "completed": result.completed, + "capture_path": str(result.capture_path), + "capture_samples": result.capture_samples, + "capture_duration_seconds": result.capture_duration_seconds, + "rule": asdict(result.rule), + "calibration": asdict(result.calibration), + "control": asdict(result.control), + "coarse": { + **asdict(result.coarse), + "candidates": [asdict(row) for row in result.coarse.candidates], + }, + "strict": { + "counters": asdict(result.strict.counters), + "cursor_history": result.strict.cursor_history, + "elapsed_seconds": result.strict.elapsed_seconds, + "average_candidate_seconds": result.strict.average_candidate_seconds, + "maximum_local_window_samples": result.strict.maximum_local_window_samples, + "maximum_local_window_bytes": result.strict.maximum_local_window_bytes, + "termination_reason": result.strict.termination_reason, + "max_search_iterations_reached": result.strict.max_search_iterations_reached, + "max_valid_frames_reached": result.strict.max_valid_frames_reached, + "outcomes": [ + { + key: value + for key, value in asdict(row).items() + if key != "frame" + } + for row in result.strict.outcomes + ], + }, + "object_verdict": { + key: value + for key, value in asdict(result.object_verdict).items() + if key != "recovered_object" + }, + "full_elapsed_seconds": result.full_elapsed_seconds, + "processing_to_capture_ratio": result.processing_to_capture_ratio, + "receiver_oracles": { + "tx_stream": False, + "tx_offsets": False, + "expected_frame_count": False, + "tx_object": False, + "fragment_order": False, + "tx_packets": False, + "tx_coded_bits": False, + }, + "rf_tx": 0, + } + + +def write_results( + output_directory: Path, + independent: IndependentRecoveryResult, + diagnostics: PostVerdictDiagnostics, +) -> tuple[Path, Path, Path | None]: + output_directory = Path(output_directory) + output_directory.mkdir(parents=True, exist_ok=True) + independent_path = output_directory / "independent_result.json" + diagnostic_path = output_directory / "post_verdict_diagnostics.json" + independent_path.write_text( + json.dumps(_independent_summary(independent), ensure_ascii=False, indent=2), + encoding="utf-8", + ) + diagnostic_path.write_text( + json.dumps(asdict(diagnostics), ensure_ascii=False, indent=2), + encoding="utf-8", + ) + recovered_path: Path | None = None + if independent.object_verdict.object_complete: + recovered = independent.object_verdict.recovered_object + if recovered is None: + raise RuntimeError("OBJECT COMPLETE без recovered bytes") + recovered_path = output_directory / "recovered_object.bin" + recovered_path.write_bytes(recovered) + return independent_path, diagnostic_path, recovered_path + + +def run_functional_tests() -> None: + assert lab046b.IQ_PREAMBLE_THRESHOLD == 0.65 + assert BACKGROUND_LOCAL_PEAK_QUANTILE == 0.999 + assert MAX_VALID_FRAMES > 68 + assert CALIBRATION_INTERVAL.end_sample < CONTROL_INTERVAL.start_sample + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--capture", type=Path, default=DEFAULT_CAPTURE_PATH) + parser.add_argument( + "--reference-directory", + type=Path, + default=DEFAULT_REFERENCE_DIRECTORY, + ) + parser.add_argument("--output-directory", type=Path, default=DEFAULT_OUTPUT_DIRECTORY) + parser.add_argument("--run-original-capture", action="store_true") + arguments = parser.parse_args() + if not arguments.run_original_capture: + run_functional_tests() + print("LAB046C OFFLINE SOFTWARE PREFLIGHT = PASS") + print("RF TX = 0") + return 0 + + independent = run_independent_recovery(arguments.capture) + if not independent.completed: + print(json.dumps({ + "independent": _independent_summary(independent), + "post_verdict": None, + "reason": "INDEPENDENT_SCAN_INCOMPLETE_FAIL_CLOSED", + "original_hardware_result": "C", + "original_hardware_result_changed": False, + "rf_tx": 0, + }, ensure_ascii=False, indent=2)) + return 2 + # Следующая строка исполняется только после возвращённого completed verdict. + diagnostics = diagnose_after_independent_verdict( + independent, + arguments.reference_directory, + ) + paths = write_results( + arguments.output_directory, + independent, + diagnostics, + ) + print(json.dumps({ + "independent": _independent_summary(independent), + "post_verdict": asdict(diagnostics), + "written_paths": [str(path) if path is not None else None for path in paths], + "original_hardware_result": "C", + "original_hardware_result_changed": False, + "rf_tx": 0, + }, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_lab046c_offline_recovery.py b/tests/test_lab046c_offline_recovery.py new file mode 100644 index 0000000..597b909 --- /dev/null +++ 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()