diff --git a/PROJECT_LOG.md b/PROJECT_LOG.md index 31e85a1..078c42d 100644 --- a/PROJECT_LOG.md +++ b/PROJECT_LOG.md @@ -749,3 +749,33 @@ Verifier read-only нашёл реальный дефект первого ва ## Следующий этап Сначала ChatGPT смотрит Lab046A software result. После отдельного явного разрешения можно выбирать следующий шаг: аппаратная проверка многокадровой передачи, оптимизация накладных расходов серии кадров или erasure-FEC для потери целых фрагментов. RF TX, QPSK, JPEG и video автоматически не запускать. + +# Запись 027 + +## Дата + +1 сентября 2026 года + +## Тема + +Lab046B: программная проверка нескольких полных Lab045-кадров в одном непрерывном сыром IQ-потоке, RF TX = 0, аппаратная проверка не выполнялась. + +## Выполнено + +Статус: **ПРОГРАММНАЯ ПРОВЕРКА НЕПРЕРЫВНОГО СЫРОГО IQ-ПОТОКА = PASS. RF TX = 0. АППАРАТНАЯ ПРОВЕРКА = НЕ ВЫПОЛНЯЛАСЬ.** HackRF, hackrf_transfer, common-clock операции, RF RX/TX, JPEG, video, QPSK, erasure FEC и ARQ не запускались. + +Добавлен software-only эксперимент `experiments/lab046b_continuous_iq.py`. Пять кадров формируются существующим Lab045: protocol packet, convolutional FEC R=1/2 K=7, interleaver, защищённый PHY header, PRBS7, BPSK, RRC и guards. Они размещаются в одном `complex64` IQ stream. Приёмный `scan_continuous_iq` получает только IQ и известные PHY/FEC параметры; TX offsets, длины waveform, payload, packets и coded bits в его API отсутствуют. TX reference используется лишь в отдельной диагностике после независимого CRC PASS. + +Первичный поиск переиспользует PRBS7 reference и normalized correlation Lab043/044, но не применяет глобальный mandatory peak/second gate. Все локальные candidates рассматриваются по положению. Для каждого кандидата локально переиспользуются refine start, matched filter, preamble-only timing, CFO, phase/polarity и `normalized.real`; затем Lab045 выполняет header-driven deinterleave, soft Viterbi и CRC. Cursor монотонен, accepted frame повторно не декодируется, rejected candidate не останавливает поиск, `MAX_FRAMES` и `MAX_SEARCH_ITERATIONS` ограничивают работу. + +Фактический идеальный stream: 5 waveform, leading 137 samples, gaps 211/353/587/809, trailing 233; всего 11 534 330 `complex64` samples = 92 274 640 bytes (88.0 MiB). PRBS7 candidates 32, processed 5, accepted 5, rejected 0, PHY header PASS 5, FEC PASS 5, CRC 5/0. Все обнаруженные sample offsets совпали с TX при постфактум-сравнении; sampling phase modulo 400 = 137/348/301/88/97. Peak/second настоящих кадров около 0 dB, что воспроизводимо показывает непригодность старого глобального критерия для многокадрового решения. + +Сценарий CFO/phase/amplitude: внесено 0.2 Hz, 0.7 rad и scale 0.65 без передачи этих истинных значений scanner-у. Медианная независимая оценка 0.200003768 Hz и 0.699634288 rad, полярность восстановлена по PRBS7, CRC 5/0. Комплексный AWGN: seed 46002, финальная sigma=0.17, accepted 5, pre-FEC 0, post-FEC 0, CRC 5/0. Начальная разумная sigma=0.12 также дала 0 pre-FEC ошибок; после одной разрешённой корректировки дальнейшего подбора не было. Результат шума фиксируется честно как успешный IQ receive без нового события FEC correction. + +Ложный IQ-кандидат содержит настоящую PRBS7 waveform и неверный protected PHY magic. Получено candidates 11, processed 8, rejected 7; поиск продолжился и следующий настоящий кадр прошёл PHY/FEC/CRC. Добавлены 14 тестов: два и пять реальных IQ-кадров, arbitrary/non-grid start, разные gaps, ровно один приём кадра, tail background/noise, malformed candidate, продолжение поиска, отсутствие TX-oracle API, CFO/phase, deterministic complex noise, monotonic cursor, оба safety limits и регрессия нескольких сильных настоящих пиков. + +Независимый read-only verifier подтвердил тракт, отсутствие TX oracle, multi-peak решение, ложный кандидат, численные CFO/phase и размер основного массива; 88.0 MiB относится именно к main IQ array, не к полному peak RSS с FFT/RRC workspace. Software gate: `python tools/quick_gate.py` PASS (468 тестов, 7 функциональных лаб, 13 штатно пропущено); `python -m pytest tests/ -q` = 468 passed; `git diff --check` чист, кроме информационных LF/CRLF notices. Документация обновлена в `docs/lab046_current_state.md`. Git add/commit/push не выполнялись. + +## Следующий этап + +Сначала ChatGPT анализирует Lab046B. Аппаратную Lab046, JPEG, video, QPSK, 12+3 erasure FEC, ARQ и адаптивную модуляцию автоматически не начинать. diff --git a/docs/lab046_current_state.md b/docs/lab046_current_state.md index 2846b6e..6d2ccd6 100644 --- a/docs/lab046_current_state.md +++ b/docs/lab046_current_state.md @@ -4,9 +4,13 @@ **LAB046 SOFTWARE OBJECT TRANSPORT = COMPLETE.** -**CONTINUOUS STREAM VALIDATION = PASS (NORMALIZED SOFT-SYMBOL LEVEL). RF TX = 0. -CONTINUOUS SAMPLE/IQ VALIDATION = NOT PERFORMED. -HARDWARE VALIDATION = NOT PERFORMED.** +**CONTINUOUS STREAM VALIDATION = PASS (NORMALIZED SOFT-SYMBOL LEVEL).** + +**ПРОГРАММНАЯ ПРОВЕРКА НЕПРЕРЫВНОГО СЫРОГО IQ-ПОТОКА = PASS.** + +**RF TX = 0.** + +**АППАРАТНАЯ ПРОВЕРКА = НЕ ВЫПОЛНЯЛАСЬ.** Lab046 — только программная лаба. HackRF и `hackrf_transfer` не запускались, эфирной передачи нет. Ни JPEG, ни видео, ни QPSK. @@ -110,6 +114,61 @@ Verifier нашёл опасный вариант прежнего глобал кадра. Поэтому suppression убран из списка кандидатов, а добавленный регрессионный тест проверяет, что ложный peak не скрывает настоящий кадр. +## Lab046B: непрерывный сырой IQ-поток + +Lab046B закрывает только программный sample-level пробел Lab046A. Новый +`experiments/lab046b_continuous_iq.py` формирует пять полных Lab045 waveform с +PRBS7, защищённым PHY header, interleaved convolutional body, BPSK, RRC и +guards, размещает их в одном `complex64` массиве и передаёт scanner-у только +этот массив. TX start offsets, длины TX waveform, номера кадров, payload, +packet и coded bits в `scan_continuous_iq` не передаются. + +Переиспользованы существующие функции Lab043-Lab045: BPSK/RRC waveform, +PRBS7 reference, normalized sliding correlation, локальное уточнение sample +start, matched filter, preamble-only timing, non-data-aided CFO и комплексный +gain, `normalized.real`, защищённый PHY header, deinterleave, soft Viterbi и +protocol CRC. Новый слой только перечисляет локальные IQ-кандидаты, вызывает +этот тракт на окне кандидата и монотонно продолжает поиск. + +Глобальное требование «главный пик значительно выше второго» не используется. +В идеальном пятикадровом опыте peak/second для настоящих кадров лежит около +0 dB, поскольку вторым пиком является другой настоящий кадр; тем не менее все +пять кадров приняты. Все локальные максимумы выше порога сохраняются, а после +CRC PASS cursor переносится за вычисленный по принятому header конец кадра. +После плохого кандидата cursor также строго растёт, но следующий кандидат не +скрывается. Выход ограничен `MAX_FRAMES` и `MAX_SEARCH_ITERATIONS`. + +### Фактические сценарии Lab046B + +| Сценарий | Получено | +|---|---| +| Идеальный continuous IQ | PRBS7 candidates 32, processed 5, accepted 5, rejected 0, PHY/FEC/CRC 5/5/5, CRC FAIL 0, sample offset error 0 для всех | +| CFO + phase + amplitude | внесено 0.2 Hz / 0.7 rad / 0.65; медианная оценка 0.200003768 Hz / 0.699634288 rad; CRC 5/0 | +| Комплексный AWGN | sigma=0.17, seed=46002; accepted 5, pre-FEC 0, post-FEC 0, CRC 5/0 | +| PRBS7 + неверный PHY header перед настоящим кадром | candidates 11, processed 8, rejected 7, следующий настоящий кадр найден, CRC 1/0 | + +Начало первого waveform: sample 137, то есть не sample 0 и не граница сетки +400 samples/symbol. Дополнительные gaps: 211/353/587/809 samples. Найденные +фазы символьной выборки: 137/348/301/88/97 samples modulo 400. Точные TX +offsets используются только постфактум после CRC PASS; ошибки обнаруженного +start во всех пяти случаях равны нулю. + +Шумовая точка зафиксирована честно. Первый разумный однокадровый запуск при +sigma=0.12 дал CRC PASS, но 0 pre-FEC ошибок. Выполнена одна разрешённая +обоснованная корректировка до sigma=0.17; она также дала 0 pre-FEC ошибок и +CRC PASS. Дальнейший подбор не выполнялся. Поэтому этот сценарий доказывает +устойчивый многокадровый IQ receive с комплексным детерминированным шумом, но +не является новым доказательством реального исправления ошибок FEC; такое +доказательство уже относится к аппаратной Lab045. + +Основной массив содержит 11 534 330 `complex64` отсчётов и занимает +92 274 640 байт (88.0 MiB). Пять waveform занимают 11 532 000 samples, остальное +— leading/gaps/trailing. Полные CFO/phase-corrected копии не создаются: +первичный поиск работает с decimated stream, а matched filtering и soft +extraction выполняются на локальном окне одного кандидата. Это размер именно +основного IQ-массива, а не peak RSS процесса: RRC/FFT, локальный matched stream +и временные массивы SciPy требуют дополнительную рабочую память. + ## Память continuous stream Основной 68-кадровый soft stream содержит 389 232 значения `float32` и занимает @@ -235,9 +294,9 @@ Goodput 6,69 кбит/с — это честная baseline при 20 ксимв ## Software gate -- `python tools/quick_gate.py`: PASS; 454 теста ядра, 6 функциональных +- `python tools/quick_gate.py`: PASS; 468 тестов ядра, 7 функциональных проверок лабораторий выполнено, 13 штатно пропущено. -- `python -m pytest tests/ -q`: 454 passed, 0 failed, 0 errors, 0 skipped. +- `python -m pytest tests/ -q`: 468 passed, 0 failed, 0 errors, 0 skipped. Единственное предупреждение связано с отсутствием прав на создание `.pytest_cache`; на выполнение и результат тестов оно не повлияло. - `git diff --check`: чист. diff --git a/experiments/lab046b_continuous_iq.py b/experiments/lab046b_continuous_iq.py new file mode 100644 index 0000000..94e8834 --- /dev/null +++ b/experiments/lab046b_continuous_iq.py @@ -0,0 +1,926 @@ +"""Lab046B: несколько Lab045-кадров в одном непрерывном IQ-потоке. + +Эксперимент программный. Передатчик формирует настоящие Lab045 waveform, +а приёмник получает только один ``complex64`` массив. Известные TX-границы, +payload, packet и coded bits не входят в API приёмника. Сравнение с TX +выполняется отдельно и только после независимого CRC PASS. + +RF TX = 0. Аппаратура не используется. +""" + +from __future__ import annotations + +import argparse +import math +import sys +from dataclasses import dataclass +from pathlib import Path + +if __package__ in (None, ""): + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import numpy as np + +from experiments import lab045_hackrf_bpsk_fec as lab045 +from protocol import bpsk_radio as radio + + +FRAME_COUNT = 5 +LEADING_BACKGROUND_SAMPLES = 137 +INTER_FRAME_GAPS = (211, 353, 587, 809) +TRAILING_BACKGROUND_SAMPLES = 233 + +IMPAIRED_CFO_HZ = 0.2 +IMPAIRED_PHASE_RAD = 0.7 +IMPAIRED_AMPLITUDE = 0.65 + +NOISE_SEED = 46002 +NOISE_SIGMA = 0.17 + +IQ_PREAMBLE_THRESHOLD = 0.65 +MAX_FRAMES = 32 +MAX_SEARCH_ITERATIONS = 256 +IMPAIRMENT_BLOCK_SAMPLES = 262_144 + + +def _amplitude_db(ratio: float) -> float: + return 20.0 * math.log10(max(float(ratio), 1e-300)) + + +def _wrap_phase(value: float) -> float: + return float((value + math.pi) % (2.0 * math.pi) - math.pi) + + +def deterministic_payload(frame_index: int) -> bytes: + """Создать различимый детерминированный payload одного Lab045-кадра.""" + + if frame_index < 0: + raise ValueError("frame_index не может быть отрицательным") + random_generator = np.random.Generator(np.random.PCG64(46_000 + frame_index)) + payload = random_generator.integers( + 0, + 256, + lab045.PAYLOAD_SIZE_BYTES, + dtype=np.uint8, + ) + payload[:4] = np.frombuffer(frame_index.to_bytes(4, "big"), dtype=np.uint8) + return payload.tobytes() + + +def build_frame_plans(frame_count: int = FRAME_COUNT) -> tuple[lab045.TransmissionPlan, ...]: + if frame_count <= 0: + raise ValueError("нужен хотя бы один кадр") + return tuple( + lab045.build_transmission_plan(deterministic_payload(frame_index)) + for frame_index in range(frame_count) + ) + + +def malformed_candidate_waveform() -> np.ndarray: + """Сформировать настоящий IQ-prefix PRBS7 с заведомо неверным PHY magic.""" + + header = lab045.encode_phy_header(lab045.MIN_PACKET_BYTES).copy() + header[: lab045.HEADER_REPETITION] ^= 1 + bits = np.concatenate((lab045.lab044.preamble_bits(), header)) + shaped = lab045.lab044._scale_and_carrier(radio.bpsk_modulate(bits)) + raw = lab045.lab044.lab043.to_cs8_bytes(shaped) + return lab045.lab044.signed_iq_bytes_to_complex64(raw) + + +@dataclass(frozen=True) +class ContinuousIqPlan: + """TX-side описание; ``frame_starts`` запрещено передавать scanner-у.""" + + samples: np.ndarray + frame_plans: tuple[lab045.TransmissionPlan, ...] + frame_starts: tuple[int, ...] + leading_background_samples: int + inter_frame_gaps: tuple[int, ...] + trailing_background_samples: int + cfo_hz: float + phase_rad: float + amplitude: float + noise_sigma: float + noise_seed: int + malformed_candidate_start: int | None + malformed_candidate_sample_count: int + + @property + def sample_count(self) -> int: + return int(self.samples.size) + + @property + def memory_bytes(self) -> int: + return int(self.samples.nbytes) + + +def _apply_impairments_in_place( + samples: np.ndarray, + *, + cfo_hz: float, + phase_rad: float, + amplitude: float, + noise_sigma: float, + noise_seed: int, +) -> None: + """Применить канал блоками без постоянной full-size complex128 копии.""" + + if amplitude <= 0.0 or not np.isfinite(amplitude): + raise ValueError("amplitude должна быть положительной и конечной") + if noise_sigma < 0.0 or not np.isfinite(noise_sigma): + raise ValueError("noise_sigma должна быть неотрицательной и конечной") + if not np.isfinite(cfo_hz) or not np.isfinite(phase_rad): + raise ValueError("CFO и phase должны быть конечными") + + phase_step = 2.0 * math.pi * cfo_hz / lab045.SAMPLE_RATE_HZ + random_generator = np.random.Generator(np.random.PCG64(noise_seed)) + for start in range(0, samples.size, IMPAIRMENT_BLOCK_SAMPLES): + end = min(samples.size, start + IMPAIRMENT_BLOCK_SAMPLES) + indexes = np.arange(start, end, dtype=np.float64) + rotation = np.exp(1j * (phase_rad + phase_step * indexes)).astype(np.complex64) + samples[start:end] *= np.complex64(amplitude) * rotation + if noise_sigma > 0.0: + real_noise = random_generator.standard_normal(end - start, dtype=np.float32) + imaginary_noise = random_generator.standard_normal(end - start, dtype=np.float32) + samples.real[start:end] += np.float32(noise_sigma) * real_noise + samples.imag[start:end] += np.float32(noise_sigma) * imaginary_noise + + +def build_continuous_iq_stream( + frame_plans: tuple[lab045.TransmissionPlan, ...] | list[lab045.TransmissionPlan], + *, + leading_background_samples: int = LEADING_BACKGROUND_SAMPLES, + inter_frame_gaps: tuple[int, ...] | list[int] | None = None, + trailing_background_samples: int = TRAILING_BACKGROUND_SAMPLES, + cfo_hz: float = 0.0, + phase_rad: float = 0.0, + amplitude: float = 1.0, + noise_sigma: float = 0.0, + noise_seed: int = NOISE_SEED, + prepend_malformed_candidate: bool = False, + malformed_to_frame_gap_samples: int = 317, +) -> ContinuousIqPlan: + """Собрать один непрерывный ``complex64`` stream из полных Lab045 waveform.""" + + plans = tuple(frame_plans) + if not plans: + raise ValueError("нужен хотя бы один кадр") + if leading_background_samples < 0 or trailing_background_samples < 0: + raise ValueError("фоновые интервалы не могут быть отрицательными") + if malformed_to_frame_gap_samples < 0: + raise ValueError("интервал после ложного кандидата не может быть отрицательным") + if inter_frame_gaps is None: + gaps = tuple(0 for _ in range(len(plans) - 1)) + else: + gaps = tuple(int(value) for value in inter_frame_gaps) + if len(gaps) != len(plans) - 1 or any(value < 0 for value in gaps): + raise ValueError("нужен один неотрицательный gap между каждой парой кадров") + + malformed = malformed_candidate_waveform() if prepend_malformed_candidate else None + malformed_count = 0 if malformed is None else int(malformed.size) + total_samples = ( + leading_background_samples + + malformed_count + + (malformed_to_frame_gap_samples if malformed is not None else 0) + + sum(plan.waveform_sample_count for plan in plans) + + sum(gaps) + + trailing_background_samples + ) + samples = np.zeros(total_samples, dtype=np.complex64) + cursor = leading_background_samples + malformed_start = None + if malformed is not None: + malformed_start = cursor + samples[cursor : cursor + malformed_count] = malformed + cursor += malformed_count + malformed_to_frame_gap_samples + + frame_starts: list[int] = [] + for frame_index, plan in enumerate(plans): + frame_starts.append(cursor) + waveform = lab045.lab044.signed_iq_bytes_to_complex64(lab045.build_tx_cs8(plan)) + if waveform.dtype != np.complex64 or waveform.size != plan.waveform_sample_count: + raise ValueError("Lab045 waveform имеет неожиданную форму") + samples[cursor : cursor + waveform.size] = waveform + cursor += int(waveform.size) + if frame_index < len(gaps): + cursor += gaps[frame_index] + if cursor + trailing_background_samples != total_samples: + raise RuntimeError("внутренняя ошибка учёта длины IQ stream") + + _apply_impairments_in_place( + samples, + cfo_hz=cfo_hz, + phase_rad=phase_rad, + amplitude=amplitude, + noise_sigma=noise_sigma, + noise_seed=noise_seed, + ) + return ContinuousIqPlan( + samples=samples, + frame_plans=plans, + frame_starts=tuple(frame_starts), + leading_background_samples=leading_background_samples, + inter_frame_gaps=gaps, + trailing_background_samples=trailing_background_samples, + cfo_hz=cfo_hz, + phase_rad=phase_rad, + amplitude=amplitude, + noise_sigma=noise_sigma, + noise_seed=noise_seed, + malformed_candidate_start=malformed_start, + malformed_candidate_sample_count=malformed_count, + ) + + +@dataclass(frozen=True) +class IqPreambleCandidate: + coarse_start_sample: int + score: float + over_median_db: float + over_second_db: float + + +def _decimate_then_downconvert(samples: np.ndarray) -> np.ndarray: + """Получить компактный detection stream без full-size baseband копии.""" + + decimation = lab045.lab044.lab043.DETECTION_DECIMATION + decimated = lab045.lab044.lab043.resample_poly( + samples, + 1, + decimation, + window=("kaiser", 8), + ).astype(np.complex64, copy=False) + decimated -= np.mean(decimated, dtype=np.complex128).astype(np.complex64) + sample_rate = lab045.SAMPLE_RATE_HZ / decimation + indexes = np.arange(decimated.size, dtype=np.float64) + rotation = np.exp( + -2j * math.pi * lab045.CARRIER_OFFSET_HZ * indexes / sample_rate + ).astype(np.complex64) + decimated *= rotation + return decimated + + +def _local_candidate_indexes(scores: np.ndarray, threshold: float) -> np.ndarray: + """Вернуть все локальные максимумы выше порога без global peak/second gate.""" + + values = np.asarray(scores, dtype=np.float32) + if values.ndim != 1 or values.size == 0: + return np.empty(0, dtype=np.int64) + if not 0.0 < threshold <= 1.0: + raise ValueError("threshold должен лежать в (0, 1]") + 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 >= np.float32(threshold)) + & (values >= left) + & (values > right) + ) + + +def _detect_iq_preamble_candidates( + samples: np.ndarray, + threshold: float, +) -> tuple[tuple[IqPreambleCandidate, ...], int]: + reference = lab045.lab044._preamble_reference_samples() + rx_decimated = _decimate_then_downconvert(samples) + reference_decimated = _decimate_then_downconvert(reference) + if rx_decimated.size < reference_decimated.size: + return (), int(rx_decimated.nbytes + reference_decimated.nbytes) + scores = np.asarray( + lab045.lab044.lab043._normalized_sliding_correlation( + rx_decimated, + reference_decimated, + ), + dtype=np.float32, + ) + scores[~np.isfinite(scores)] = 0.0 + scores = np.clip(scores, 0.0, 1.0) + indexes = _local_candidate_indexes(scores, threshold) + median = float(np.median(scores)) + candidate_scores = scores[indexes] + decimation = lab045.lab044.lab043.DETECTION_DECIMATION + candidates: list[IqPreambleCandidate] = [] + for position, index in enumerate(indexes): + if candidate_scores.size > 1: + others = np.delete(candidate_scores, position) + second = float(np.max(others)) + else: + second = 0.0 + score = float(scores[index]) + candidates.append(IqPreambleCandidate( + coarse_start_sample=int(index) * decimation, + score=score, + over_median_db=( + _amplitude_db(score / median) if median > 0.0 else float("inf") + ), + over_second_db=( + _amplitude_db(score / second) if second > 0.0 else float("inf") + ), + )) + storage_bytes = int( + rx_decimated.nbytes + reference_decimated.nbytes + scores.nbytes + ) + return tuple(candidates), storage_bytes + + +@dataclass(frozen=True) +class SoftWindow: + passed: bool + normalized_real: np.ndarray + timing_offset_samples: int + symbol_sample_phase: int + nominal_sample: int + carrier: dict + reason: str | None + + +def _extract_soft_window( + samples: np.ndarray, + frame_start: int, + symbol_count: int, +) -> SoftWindow: + waveform_samples = ( + (symbol_count + 2 * lab045.GUARD_SYMBOL_COUNT) * lab045.SAMPLES_PER_SYMBOL + + lab045.RRC_SPAN_SYMBOLS * lab045.SAMPLES_PER_SYMBOL + ) + if frame_start < 0 or frame_start + waveform_samples > samples.size: + return SoftWindow(False, np.empty(0), 0, 0, 0, {}, "TRUNCATED_IQ_WINDOW") + segment = samples[frame_start : frame_start + waveform_samples] + matched, nominal = lab045.lab044.lab043._matched_waveform( + segment, + 0, + waveform_samples, + ) + preamble_bits = lab045.lab044.preamble_bits() + preamble_symbols = radio.bpsk_modulate(preamble_bits) + timing, _ = lab045.lab044.lab043._timing_offsets( + matched, + nominal, + preamble_symbols, + ) + carrier = lab045.lab044.lab043._evaluate_symbols( + matched, + nominal, + timing, + preamble_bits, + lab045.PREAMBLE_BIT_COUNT, + ) + if carrier["abs_h"] == 0.0: + return SoftWindow(False, np.empty(0), timing, 0, nominal, carrier, "ZERO_GAIN") + symbols = matched[ + nominal + timing : nominal + timing + symbol_count * lab045.SAMPLES_PER_SYMBOL : lab045.SAMPLES_PER_SYMBOL + ].astype(np.complex128) + if symbols.size != symbol_count: + return SoftWindow(False, np.empty(0), timing, 0, nominal, carrier, "TRUNCATED_SYMBOLS") + symbol_time = (np.arange(symbol_count, dtype=np.float64) + 0.5) / lab045.SYMBOL_RATE + normalized = ( + symbols + * np.exp(-2j * math.pi * carrier["cfo_hz"] * symbol_time) + / (carrier["abs_h"] * np.exp(1j * carrier["phase_rad"])) + ) + sample_phase = int((frame_start + nominal + timing) % lab045.SAMPLES_PER_SYMBOL) + return SoftWindow( + True, + normalized.real.astype(np.float64, copy=False), + int(timing), + sample_phase, + int(nominal), + carrier, + None, + ) + + +@dataclass(frozen=True) +class IqFrameReception: + detection_number: int + coarse_start_sample: int + refined_start_sample: int + frame_end_sample: int + preamble_score: float + over_median_db: float + over_second_db: float + timing_offset_samples: int + symbol_sample_phase: int + nominal_sample: int + estimated_cfo_hz: float + estimated_phase_rad: float + bpsk_polarity: str + preamble_errors: int + snr_db: float + post_preamble_soft: np.ndarray + reception: lab045.IndependentReception + + +@dataclass(frozen=True) +class CandidateDecode: + refined_start_sample: int + next_sample: int + phy_header_pass: bool + fec_pass: bool + crc: str + accepted: bool + reason: str + frame: IqFrameReception | None + + +def _decode_iq_candidate( + samples: np.ndarray, + candidate: IqPreambleCandidate, + detection_number: int, +) -> CandidateDecode: + reference = lab045.lab044._preamble_reference_samples() + try: + refined = lab045.lab044.lab043._refine_frame_start( + samples, + reference, + candidate.coarse_start_sample, + ) + except (ValueError, IndexError): + return CandidateDecode( + candidate.coarse_start_sample, + candidate.coarse_start_sample + 1, + False, + False, + lab045.CRC_NOT_REACHED, + False, + "REFINEMENT_FAILED", + None, + ) + + probe_symbols = lab045.PREAMBLE_BIT_COUNT + lab045.PHY_HEADER_BIT_COUNT + probe = _extract_soft_window(samples, refined, probe_symbols) + if not probe.passed: + return CandidateDecode( + refined, + candidate.coarse_start_sample + 1, + False, + False, + lab045.CRC_NOT_REACHED, + False, + probe.reason or "PROBE_FAILED", + None, + ) + header = lab045.decode_phy_header_soft( + probe.normalized_real[ + lab045.PREAMBLE_BIT_COUNT : lab045.PREAMBLE_BIT_COUNT + lab045.PHY_HEADER_BIT_COUNT + ] + ) + if not header["passed"]: + return CandidateDecode( + refined, + candidate.coarse_start_sample + 1, + False, + False, + lab045.CRC_NOT_REACHED, + False, + f"PHY_HEADER_{header['reason']}", + None, + ) + + packet_bytes = int(header["packet_bytes"]) + coded_length = lab045.conv.encoded_bit_count(packet_bytes * 8) + frame_symbols = lab045.PREAMBLE_BIT_COUNT + lab045.PHY_HEADER_BIT_COUNT + coded_length + full = _extract_soft_window(samples, refined, frame_symbols) + if not full.passed: + return CandidateDecode( + refined, + candidate.coarse_start_sample + 1, + True, + False, + lab045.CRC_NOT_REACHED, + False, + full.reason or "FULL_EXTRACTION_FAILED", + None, + ) + post_preamble = full.normalized_real[lab045.PREAMBLE_BIT_COUNT:] + reception = lab045.analyze_post_preamble_soft(post_preamble) + crc = str(reception.analysis["crc"]) + accepted = crc == lab045.CRC_PASS + waveform_samples = ( + (frame_symbols + 2 * lab045.GUARD_SYMBOL_COUNT) * lab045.SAMPLES_PER_SYMBOL + + lab045.RRC_SPAN_SYMBOLS * lab045.SAMPLES_PER_SYMBOL + ) + frame_end = refined + waveform_samples + preamble_hard = ( + full.normalized_real[: lab045.PREAMBLE_BIT_COUNT] >= 0.0 + ).astype(np.uint8) + preamble_errors = int(np.count_nonzero( + preamble_hard != lab045.lab044.preamble_bits() + )) + carrier = full.carrier + frame = IqFrameReception( + detection_number=detection_number, + coarse_start_sample=candidate.coarse_start_sample, + refined_start_sample=refined, + frame_end_sample=frame_end, + preamble_score=candidate.score, + over_median_db=candidate.over_median_db, + over_second_db=candidate.over_second_db, + timing_offset_samples=full.timing_offset_samples, + symbol_sample_phase=full.symbol_sample_phase, + nominal_sample=full.nominal_sample, + estimated_cfo_hz=float(carrier["cfo_hz"]), + estimated_phase_rad=float(carrier["phase_rad"]), + bpsk_polarity=str(carrier["polarity"]), + preamble_errors=preamble_errors, + snr_db=float(carrier["snr_db"]), + post_preamble_soft=post_preamble, + reception=reception, + ) + reason = "CRC_PASS" if accepted else f"CRC_{crc}" + return CandidateDecode( + refined, + frame_end if accepted else candidate.coarse_start_sample + 1, + True, + bool(reception.analysis["fec_decode_pass"]), + crc, + accepted, + reason, + frame, + ) + + +@dataclass(frozen=True) +class ContinuousIqReceiveResult: + iq_sample_count: int + iq_dtype: str + preamble_candidates: int + processed_candidates: int + rejected_candidates: int + phy_header_pass_count: int + fec_pass_count: int + crc_pass_count: int + crc_fail_count: int + accepted_frames: tuple[IqFrameReception, ...] + rejected_reasons: tuple[str, ...] + candidate_starts: tuple[int, ...] + cursor_history: tuple[int, ...] + search_iterations: int + termination_reason: str + max_frames_reached: bool + max_search_iterations_reached: bool + detection_storage_bytes: int + + +def scan_continuous_iq( + iq_samples: np.ndarray, + *, + preamble_threshold: float = IQ_PREAMBLE_THRESHOLD, + max_frames: int = MAX_FRAMES, + max_search_iterations: int = MAX_SEARCH_ITERATIONS, +) -> ContinuousIqReceiveResult: + """Найти несколько кадров только по одному IQ stream, без TX offsets.""" + + try: + samples = np.asarray(iq_samples, dtype=np.complex64) + except (TypeError, ValueError): + samples = np.empty((0, 0), dtype=np.complex64) + if not 0.0 < preamble_threshold <= 1.0: + raise ValueError("preamble_threshold должен лежать в (0, 1]") + if max_frames < 0 or max_search_iterations < 0: + raise ValueError("safety limits не могут быть отрицательными") + if samples.ndim != 1 or samples.size == 0 or not np.all(np.isfinite(samples)): + return ContinuousIqReceiveResult( + int(samples.size), + str(samples.dtype), + 0, + 0, + 0, + 0, + 0, + 0, + 0, + (), + ("INVALID_IQ_STREAM",), + (), + (0,), + 0, + "INVALID_IQ_STREAM", + False, + False, + 0, + ) + + candidates, storage_bytes = _detect_iq_preamble_candidates( + samples, + preamble_threshold, + ) + cursor = 0 + cursor_history = [cursor] + accepted_frames: list[IqFrameReception] = [] + rejected_reasons: list[str] = [] + candidate_starts: list[int] = [] + phy_header_pass = 0 + fec_pass = 0 + crc_pass = 0 + crc_fail = 0 + iterations = 0 + termination = "END_OF_STREAM" + max_frames_reached = False + max_iterations_reached = False + + for candidate in candidates: + if candidate.coarse_start_sample < cursor: + continue + if len(accepted_frames) >= max_frames: + termination = "MAX_FRAMES_REACHED" + max_frames_reached = True + break + if iterations >= max_search_iterations: + termination = "MAX_SEARCH_ITERATIONS_REACHED" + max_iterations_reached = True + break + iterations += 1 + candidate_starts.append(candidate.coarse_start_sample) + decoded = _decode_iq_candidate(samples, candidate, iterations) + phy_header_pass += int(decoded.phy_header_pass) + fec_pass += int(decoded.fec_pass) + crc_pass += int(decoded.crc == lab045.CRC_PASS) + crc_fail += int(decoded.crc == lab045.CRC_FAIL) + if decoded.accepted and decoded.frame is not None: + accepted_frames.append(decoded.frame) + else: + rejected_reasons.append(decoded.reason) + next_cursor = max(cursor + 1, int(decoded.next_sample)) + cursor = next_cursor + cursor_history.append(cursor) + + return ContinuousIqReceiveResult( + iq_sample_count=int(samples.size), + iq_dtype=str(samples.dtype), + preamble_candidates=len(candidates), + processed_candidates=iterations, + rejected_candidates=len(rejected_reasons), + phy_header_pass_count=phy_header_pass, + fec_pass_count=fec_pass, + crc_pass_count=crc_pass, + crc_fail_count=crc_fail, + accepted_frames=tuple(accepted_frames), + rejected_reasons=tuple(rejected_reasons), + candidate_starts=tuple(candidate_starts), + cursor_history=tuple(cursor_history), + search_iterations=iterations, + termination_reason=termination, + max_frames_reached=max_frames_reached, + max_search_iterations_reached=max_iterations_reached, + detection_storage_bytes=storage_bytes, + ) + + +@dataclass(frozen=True) +class FrameDiagnostic: + detection_number: int + tx_frame_index: int + sample_offset: int + sample_offset_error: int + symbol_sample_phase: int + estimated_cfo_hz: float + estimated_channel_phase_rad: float + preamble_errors: int + over_median_db: float + over_second_db: float + snr_db: float + pre_fec_coded_bit_errors: int + post_fec_protocol_bit_errors: int + crc: str + + +def diagnostics_after_crc( + receive: ContinuousIqReceiveResult, + stream_plan: ContinuousIqPlan, +) -> tuple[FrameDiagnostic, ...]: + """Сопоставить RX с TX только после независимого CRC PASS.""" + + plan_by_packet = { + plan.packet: (frame_index, plan) + for frame_index, plan in enumerate(stream_plan.frame_plans) + } + rows: list[FrameDiagnostic] = [] + for frame in receive.accepted_frames: + packet = frame.reception.packet_bytes + if packet not in plan_by_packet: + continue + frame_index, plan = plan_by_packet[packet] + comparison = lab045.diagnostic_pre_and_post_fec( + frame.reception, + frame.post_preamble_soft, + plan, + ) + pre_fec_errors = comparison["pre_fec_coded"]["bit_errors"] + post_fec_errors = comparison["post_fec_packet"]["bit_errors"] + if pre_fec_errors is None or post_fec_errors is None: + raise RuntimeError("CRC PASS frame оказался несопоставим с TX diagnostic") + phase_reference_sample = ( + frame.refined_start_sample + + frame.nominal_sample + + frame.timing_offset_samples + - 0.5 * lab045.SAMPLES_PER_SYMBOL + ) + estimated_channel_phase = _wrap_phase( + frame.estimated_phase_rad + - 2.0 + * math.pi + * frame.estimated_cfo_hz + * phase_reference_sample + / lab045.SAMPLE_RATE_HZ + ) + rows.append(FrameDiagnostic( + detection_number=frame.detection_number, + tx_frame_index=frame_index, + sample_offset=frame.refined_start_sample, + sample_offset_error=( + frame.refined_start_sample - stream_plan.frame_starts[frame_index] + ), + symbol_sample_phase=frame.symbol_sample_phase, + estimated_cfo_hz=frame.estimated_cfo_hz, + estimated_channel_phase_rad=estimated_channel_phase, + preamble_errors=frame.preamble_errors, + over_median_db=frame.over_median_db, + over_second_db=frame.over_second_db, + snr_db=frame.snr_db, + pre_fec_coded_bit_errors=int(pre_fec_errors), + post_fec_protocol_bit_errors=int(post_fec_errors), + crc=str(frame.reception.analysis["crc"]), + )) + return tuple(rows) + + +@dataclass(frozen=True) +class IqScenarioResult: + name: str + frame_count: int + sample_count: int + iq_dtype: str + iq_memory_bytes: int + frame_starts: tuple[int, ...] + inter_frame_gaps: tuple[int, ...] + leading_background_samples: int + trailing_background_samples: int + cfo_hz: float + phase_rad: float + amplitude: float + noise_sigma: float + noise_seed: int + malformed_candidate_start: int | None + receive: ContinuousIqReceiveResult + diagnostics: tuple[FrameDiagnostic, ...] + + @property + def pre_fec_coded_bit_errors(self) -> int: + return sum(row.pre_fec_coded_bit_errors for row in self.diagnostics) + + @property + def post_fec_protocol_bit_errors(self) -> int: + return sum(row.post_fec_protocol_bit_errors for row in self.diagnostics) + + @property + def estimated_cfo_hz(self) -> float | None: + if not self.diagnostics: + return None + return float(np.median([row.estimated_cfo_hz for row in self.diagnostics])) + + @property + def estimated_phase_rad(self) -> float | None: + if not self.diagnostics: + return None + values = np.asarray( + [row.estimated_channel_phase_rad for row in self.diagnostics], + dtype=np.float64, + ) + return float(np.angle(np.mean(np.exp(1j * values)))) + + +def run_iq_scenario( + name: str, + frame_plans: tuple[lab045.TransmissionPlan, ...], + **stream_arguments, +) -> IqScenarioResult: + stream_plan = build_continuous_iq_stream(frame_plans, **stream_arguments) + receive = scan_continuous_iq(stream_plan.samples) + diagnostics = diagnostics_after_crc(receive, stream_plan) + return IqScenarioResult( + name=name, + frame_count=len(frame_plans), + sample_count=stream_plan.sample_count, + iq_dtype=str(stream_plan.samples.dtype), + iq_memory_bytes=stream_plan.memory_bytes, + frame_starts=stream_plan.frame_starts, + inter_frame_gaps=stream_plan.inter_frame_gaps, + leading_background_samples=stream_plan.leading_background_samples, + trailing_background_samples=stream_plan.trailing_background_samples, + cfo_hz=stream_plan.cfo_hz, + phase_rad=stream_plan.phase_rad, + amplitude=stream_plan.amplitude, + noise_sigma=stream_plan.noise_sigma, + noise_seed=stream_plan.noise_seed, + malformed_candidate_start=stream_plan.malformed_candidate_start, + receive=receive, + diagnostics=diagnostics, + ) + + +@dataclass(frozen=True) +class Lab046bValidation: + ideal: IqScenarioResult + impaired: IqScenarioResult + noisy: IqScenarioResult + malformed: IqScenarioResult + + +def run_lab046b_validation(frame_count: int = FRAME_COUNT) -> Lab046bValidation: + plans = build_frame_plans(frame_count) + gaps = INTER_FRAME_GAPS[: frame_count - 1] + if len(gaps) != frame_count - 1: + gaps = tuple(137 + 2 * index for index in range(frame_count - 1)) + common = { + "leading_background_samples": LEADING_BACKGROUND_SAMPLES, + "inter_frame_gaps": gaps, + "trailing_background_samples": TRAILING_BACKGROUND_SAMPLES, + } + ideal = run_iq_scenario("IDEAL", plans, **common) + impaired = run_iq_scenario( + "CFO_PHASE_AMPLITUDE", + plans, + **common, + cfo_hz=IMPAIRED_CFO_HZ, + phase_rad=IMPAIRED_PHASE_RAD, + amplitude=IMPAIRED_AMPLITUDE, + ) + noisy = run_iq_scenario( + "COMPLEX_GAUSSIAN_NOISE", + plans, + **common, + noise_sigma=NOISE_SIGMA, + noise_seed=NOISE_SEED, + ) + malformed = run_iq_scenario( + "MALFORMED_CANDIDATE", + plans[:1], + leading_background_samples=LEADING_BACKGROUND_SAMPLES, + inter_frame_gaps=(), + trailing_background_samples=TRAILING_BACKGROUND_SAMPLES, + prepend_malformed_candidate=True, + ) + return Lab046bValidation(ideal, impaired, noisy, malformed) + + +def run_functional_tests() -> None: + scores = np.asarray([0.0, 0.8, 0.1, 0.9, 0.0], dtype=np.float32) + assert _local_candidate_indexes(scores, 0.65).tolist() == [1, 3] + assert scan_continuous_iq(np.empty(0, dtype=np.complex64)).termination_reason == "INVALID_IQ_STREAM" + + +def _print_scenario(result: IqScenarioResult) -> None: + print(f"{result.name}:") + print(f" samples = {result.sample_count}") + print(f" dtype = {result.iq_dtype}") + print(f" preamble candidates = {result.receive.preamble_candidates}") + print(f" processed candidates = {result.receive.processed_candidates}") + print(f" accepted frames = {len(result.receive.accepted_frames)}") + print(f" rejected candidates = {result.receive.rejected_candidates}") + print(f" PHY header PASS = {result.receive.phy_header_pass_count}") + print(f" FEC PASS = {result.receive.fec_pass_count}") + print(f" CRC PASS/FAIL = {result.receive.crc_pass_count}/{result.receive.crc_fail_count}") + print(f" pre/post-FEC bit errors = {result.pre_fec_coded_bit_errors}/{result.post_fec_protocol_bit_errors}") + print(f" estimated CFO = {result.estimated_cfo_hz}") + print(f" estimated phase = {result.estimated_phase_rad}") + for row in result.diagnostics: + print( + " frame " + f"{row.tx_frame_index}: offset={row.sample_offset}, " + f"offset_error={row.sample_offset_error}, " + f"sample_phase={row.symbol_sample_phase}, " + f"CFO={row.estimated_cfo_hz:.9f} Hz, " + f"phase={row.estimated_channel_phase_rad:.9f} rad, " + f"preamble_errors={row.preamble_errors}, " + f"peak/median={row.over_median_db:.3f} dB, " + f"peak/second={row.over_second_db:.3f} dB, " + f"SNR={row.snr_db:.3f} dB, " + f"pre/post-FEC={row.pre_fec_coded_bit_errors}/" + f"{row.post_fec_protocol_bit_errors}, CRC={row.crc}" + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--frame-count", type=int, default=FRAME_COUNT) + arguments = parser.parse_args() + result = run_lab046b_validation(arguments.frame_count) + print("LAB046B SOFTWARE CONTINUOUS IQ VALIDATION") + print("RF TX = 0") + print("known TX frame boundaries at RX = NO") + print( + f"main IQ = {result.ideal.sample_count} samples, " + f"{result.ideal.iq_dtype}, {result.ideal.iq_memory_bytes} bytes" + ) + for scenario in (result.ideal, result.impaired, result.noisy, result.malformed): + _print_scenario(scenario) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_lab046b_continuous_iq.py b/tests/test_lab046b_continuous_iq.py new file mode 100644 index 0000000..f3dbd15 --- /dev/null +++ b/tests/test_lab046b_continuous_iq.py @@ -0,0 +1,291 @@ +"""Проверки Lab046B: несколько полных BPSK/FEC кадров в одном IQ stream.""" + +from __future__ import annotations + +import inspect +from functools import lru_cache + +import numpy as np +import pytest + +from experiments import lab046b_continuous_iq as lab046b + + +@lru_cache(maxsize=1) +def full_validation() -> lab046b.Lab046bValidation: + """Один дорогой интеграционный прогон для всего test-модуля.""" + + return lab046b.run_lab046b_validation() + + +@lru_cache(maxsize=1) +def two_frame_ideal() -> lab046b.IqScenarioResult: + plans = lab046b.build_frame_plans(2) + return lab046b.run_iq_scenario( + "TWO_FRAME_IDEAL", + plans, + leading_background_samples=137, + inter_frame_gaps=(211,), + trailing_background_samples=233, + ) + + +def test_two_sequential_iq_frames_found_without_known_boundaries() -> None: + result = two_frame_ideal() + assert result.frame_count == 2 + assert len(result.receive.accepted_frames) == 2 + assert result.receive.crc_pass_count == 2 + assert all(row.sample_offset_error == 0 for row in result.diagnostics) + + +def test_five_full_iq_frames_pass_complete_receiver_chain() -> None: + result = full_validation().ideal + assert result.frame_count == 5 + assert len(result.receive.accepted_frames) == 5 + assert result.receive.phy_header_pass_count == 5 + assert result.receive.fec_pass_count == 5 + assert result.receive.crc_pass_count == 5 + assert result.receive.crc_fail_count == 0 + assert result.post_fec_protocol_bit_errors == 0 + + +def test_first_frame_offset_and_inter_frame_gaps_are_not_symbol_aligned() -> None: + result = full_validation().ideal + assert result.leading_background_samples == 137 + assert result.frame_starts[0] == 137 + assert result.frame_starts[0] % lab046b.lab045.SAMPLES_PER_SYMBOL != 0 + assert result.inter_frame_gaps == (211, 353, 587, 809) + assert any( + gap % lab046b.lab045.SAMPLES_PER_SYMBOL != 0 + for gap in result.inter_frame_gaps + ) + assert [row.symbol_sample_phase for row in result.diagnostics] == [137, 348, 301, 88, 97] + + +def test_each_real_frame_is_accepted_exactly_once_and_tail_adds_none() -> None: + result = full_validation().ideal + accepted = result.receive.accepted_frames + starts = [frame.refined_start_sample for frame in accepted] + assert len(starts) == len(set(starts)) == 5 + assert starts == list(result.frame_starts) + assert result.trailing_background_samples == 233 + assert result.receive.crc_pass_count == 5 + + +def test_multiple_strong_real_peaks_do_not_fail_peak_over_second_gate() -> None: + result = full_validation().ideal + assert result.receive.preamble_candidates >= 5 + assert len(result.receive.accepted_frames) == 5 + # Другие настоящие кадры дают почти такой же пик: старый mandatory gate + # peak/second здесь не прошёл бы, но многокадровое решение его не использует. + assert max(abs(row.over_second_db) for row in result.diagnostics) < 0.05 + + +def test_local_candidate_finder_keeps_two_equal_strong_peaks() -> None: + scores = np.asarray([0.0, 0.91, 0.0, 0.90, 0.0], dtype=np.float32) + assert lab046b._local_candidate_indexes(scores, 0.65).tolist() == [1, 3] + + +def test_cfo_phase_and_amplitude_recovered_without_truth_in_scanner_api() -> None: + result = full_validation().impaired + parameters = set(inspect.signature(lab046b.scan_continuous_iq).parameters) + assert parameters == { + "iq_samples", + "preamble_threshold", + "max_frames", + "max_search_iterations", + } + assert result.cfo_hz == pytest.approx(0.2) + assert result.phase_rad == pytest.approx(0.7) + assert result.amplitude == pytest.approx(0.65) + assert result.estimated_cfo_hz == pytest.approx(0.2, abs=0.01) + assert result.estimated_phase_rad == pytest.approx(0.7, abs=0.02) + assert result.receive.crc_pass_count == 5 + assert all( + frame.bpsk_polarity == "resolved from preamble" + for frame in result.receive.accepted_frames + ) + + +def test_complex_iq_noise_is_deterministic_and_crc_clean() -> None: + first = np.zeros(4096, dtype=np.complex64) + second = np.zeros(4096, dtype=np.complex64) + arguments = { + "cfo_hz": 0.0, + "phase_rad": 0.0, + "amplitude": 1.0, + "noise_sigma": lab046b.NOISE_SIGMA, + "noise_seed": lab046b.NOISE_SEED, + } + lab046b._apply_impairments_in_place(first, **arguments) + lab046b._apply_impairments_in_place(second, **arguments) + assert np.array_equal(first, second) + + result = full_validation().noisy + assert result.noise_seed == 46002 + assert result.noise_sigma == pytest.approx(0.17) + assert result.receive.crc_pass_count == 5 + assert result.receive.crc_fail_count == 0 + assert result.pre_fec_coded_bit_errors == 0 + assert result.post_fec_protocol_bit_errors == 0 + + +def test_malformed_iq_candidate_rejected_then_real_frame_found() -> None: + result = full_validation().malformed + assert result.malformed_candidate_start == 137 + assert result.receive.preamble_candidates > 1 + assert result.receive.rejected_candidates >= 1 + assert any(reason.startswith("PHY_HEADER_") for reason in result.receive.rejected_reasons) + assert len(result.receive.accepted_frames) == 1 + assert result.receive.crc_pass_count == 1 + assert result.diagnostics[0].sample_offset_error == 0 + + +def _fake_frame(start: int, end: int) -> lab046b.IqFrameReception: + reception = lab046b.lab045.IndependentReception( + { + "fec_decode_pass": True, + "crc": lab046b.lab045.CRC_PASS, + }, + b"", + b"", + ) + return lab046b.IqFrameReception( + detection_number=1, + coarse_start_sample=start, + refined_start_sample=start, + frame_end_sample=end, + preamble_score=0.9, + over_median_db=20.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, + ) + + +def _fake_candidates(*starts: int) -> tuple[lab046b.IqPreambleCandidate, ...]: + return tuple( + lab046b.IqPreambleCandidate(start, 0.9, 20.0, 0.0) + for start in starts + ) + + +def test_rejected_candidate_continues_and_cursor_is_monotonic(monkeypatch) -> None: + monkeypatch.setattr( + lab046b, + "_detect_iq_preamble_candidates", + lambda samples, threshold: (_fake_candidates(10, 30), 0), + ) + + def decode(samples, candidate, detection_number): + if candidate.coarse_start_sample == 10: + return lab046b.CandidateDecode( + 10, + 11, + False, + False, + lab046b.lab045.CRC_NOT_REACHED, + False, + "PHY_HEADER_BAD_MAGIC", + None, + ) + frame = _fake_frame(30, 60) + return lab046b.CandidateDecode( + 30, + 60, + True, + True, + lab046b.lab045.CRC_PASS, + True, + "CRC_PASS", + frame, + ) + + monkeypatch.setattr(lab046b, "_decode_iq_candidate", decode) + result = lab046b.scan_continuous_iq(np.ones(100, dtype=np.complex64)) + assert result.rejected_candidates == 1 + assert len(result.accepted_frames) == 1 + assert all( + after > before + for before, after in zip(result.cursor_history, result.cursor_history[1:]) + ) + + +def test_max_search_iterations_is_fail_closed(monkeypatch) -> None: + monkeypatch.setattr( + lab046b, + "_detect_iq_preamble_candidates", + lambda samples, threshold: (_fake_candidates(10, 20, 30), 0), + ) + monkeypatch.setattr( + lab046b, + "_decode_iq_candidate", + lambda samples, candidate, detection_number: lab046b.CandidateDecode( + candidate.coarse_start_sample, + candidate.coarse_start_sample + 1, + False, + False, + lab046b.lab045.CRC_NOT_REACHED, + False, + "BAD", + None, + ), + ) + result = lab046b.scan_continuous_iq( + np.ones(100, dtype=np.complex64), + max_search_iterations=1, + ) + assert result.search_iterations == 1 + assert result.max_search_iterations_reached is True + assert result.termination_reason == "MAX_SEARCH_ITERATIONS_REACHED" + + +def test_max_frames_is_fail_closed(monkeypatch) -> None: + monkeypatch.setattr( + lab046b, + "_detect_iq_preamble_candidates", + lambda samples, threshold: (_fake_candidates(10, 40), 0), + ) + + def decode(samples, candidate, detection_number): + start = candidate.coarse_start_sample + frame = _fake_frame(start, start + 20) + return lab046b.CandidateDecode( + start, + start + 20, + True, + True, + lab046b.lab045.CRC_PASS, + True, + "CRC_PASS", + frame, + ) + + monkeypatch.setattr(lab046b, "_decode_iq_candidate", decode) + result = lab046b.scan_continuous_iq( + np.ones(100, dtype=np.complex64), + max_frames=1, + ) + assert len(result.accepted_frames) == 1 + assert result.max_frames_reached is True + assert result.termination_reason == "MAX_FRAMES_REACHED" + + +def test_main_iq_memory_accounting_is_complex64() -> None: + result = full_validation().ideal + assert result.iq_dtype == "complex64" + assert result.sample_count == 11_534_330 + assert result.iq_memory_bytes == result.sample_count * np.dtype(np.complex64).itemsize + assert result.iq_memory_bytes == 92_274_640 + + +def test_small_functional_checks() -> None: + lab046b.run_functional_tests()