"""Lab043: программная часть тракта Pluto+ -> независимый RTL-SDR. Текущий этап содержит только детерминированные синтетические проверки и чистые функции подготовки обработки. Модуль не импортирует аппаратные библиотеки, не открывает SDR и не включает передатчик. Аппаратная постановка описана в docs/lab043_pluto_to_rtlsdr_spec.md. """ from __future__ import annotations import csv import hashlib import json import math from dataclasses import asdict, dataclass from pathlib import Path from typing import Iterable import numpy as np from matplotlib.figure import Figure from scipy.signal import fftconvolve, welch from experiments.lab042_pluto_image_loopback import ( GUARD_SYMBOL_COUNT as LAB042_GUARD_SYMBOL_COUNT, RRC_ROLLOFF as LAB042_RRC_ROLLOFF, RRC_SPAN_SYMBOLS as LAB042_RRC_SPAN_SYMBOLS, TX_AMPLITUDE as LAB042_TX_AMPLITUDE, ) from protocol import bpsk_radio as radio from protocol.image_fragments import MissingFragmentsError, reassemble_image from protocol.packet import MESSAGE_TYPE_TEXT, build_packet, parse_packet SYMBOL_RATE = 20_000 SAMPLE_RATE_HZ = 2_400_000 SAMPLES_PER_SYMBOL = SAMPLE_RATE_HZ // SYMBOL_RATE CARRIER_HZ = 435_000_000 F_CAL_HZ = 50_000.0 CALIBRATION_NFFT = 65_536 CALIBRATION_ANALYSIS_HALF_BAND_HZ = 100_000.0 CALIBRATION_SEARCH_HALF_WIDTH_HZ = 30_000.0 CALIBRATION_GUARD_HALF_WIDTH_HZ = 2_000.0 MINIMUM_TONE_EXCESS_DB = 10.0 MAXIMUM_ABS_SAMPLE_CLOCK_ERROR_PPM = 1_000.0 RX_LEADING_MARGIN_SECONDS = 0.5 RX_TRAILING_MARGIN_SECONDS = 0.5 DEFAULT_READ_BLOCK_SAMPLES = 262_144 RTL_ASYNC_BUFFER_BYTES = 262_144 RTL_ASYNC_BUFFER_COUNT = 15 RTL_ASYNC_WARMUP_CALLBACK_COUNT = 1 RTL_BYTES_PER_COMPLEX_SAMPLE = 2 PRBS11_LENGTH = 2_047 PRBS11_INITIAL_STATE = 0x7FF PRBS11_POLYNOMIAL = "x^11 + x^9 + 1" PRBS_SHORT_DURATION_SECONDS = 0.25 PRBS_LONG_DURATION_SECONDS = 2.0 PILOT_SYMBOL_COUNT = 1_280 PILOT_INITIAL_STATE = 0x7F PILOT_POLYNOMIAL = "x^7 + x^6 + 1" CALIBRATION_TONE_DURATION_SECONDS = 0.25 CALIBRATION_TO_BPSK_GUARD_SECONDS = 0.10 PRBS_MARKER_MINIMUM_CORRELATION = 0.65 EXPECTED_FRAME_COUNT = 10 @dataclass(frozen=True) class CalibrationResult: """Единый явный контракт полной двухтоновой калибровки.""" valid: bool invalid_reason: str f_low_hz: float f_high_hz: float carrier_offset_hz: float carrier_offset_ppm: float clock_scale: float sample_clock_error_ppm: float residual_cfo_hz: float phase_fit_rmse_rad: float peak_margin_low_db: float peak_margin_high_db: float noise_power: float @property def low_peak_excess_db(self) -> float: """Совместимое имя для старых отчётных потребителей.""" return self.peak_margin_low_db @property def high_peak_excess_db(self) -> float: """Совместимое имя для старых отчётных потребителей.""" return self.peak_margin_high_db @property def failure_reason(self) -> str: """Совместимое имя причины отказа.""" return self.invalid_reason # Старое имя оставлено только как импортная совместимость. Объект и контракт # один: новые функции и отчёты используют CalibrationResult. CalibrationEstimate = CalibrationResult @dataclass(frozen=True) class ModeResult: """Результат обработки одной записи одним из режимов A/B/C.""" mode: str bit_error_rate: float bit_errors: int bit_count: int marker_correlation: float symbol_sample_phase: int fine_cfo_applied: bool estimated_fine_cfo_hz: float @dataclass(frozen=True) class PrbsTransmissionPlan: """Заранее фиксированный состав одной калибровочно-PRBS11 посылки.""" label: str useful_duration_seconds: float payload_bits: np.ndarray marker_bits: np.ndarray pilot_bits: np.ndarray tx_samples: np.ndarray calibration_sample_count: int fixed_guard_sample_count: int bpsk_start_sample: int @dataclass(frozen=True) class PrbsModeMetrics: """Метрики одного режима обработки той же непрерывной IQ-записи.""" mode: str detected: bool failure_reason: str transmitted_bit_count: int matched_bit_count: int bit_errors: int bit_error_rate: float quarter_bit_error_rates: tuple[float, float, float, float] evm_percent: float marker_correlation: float symbol_phase_start_samples: float symbol_phase_end_samples: float accumulated_timing_drift_samples: float accumulated_timing_drift_symbols: float residual_cfo_hz: float maximum_correct_run_bits: int fine_cfo_applied: bool estimated_fine_cfo_hz: float timing_sro_ppm: float coarse_cfo_applied: bool pilot_cfo_applied: bool pilot_estimate_valid: bool estimated_pilot_cfo_hz: float residual_pilot_cfo_hz: float pilot_phase_fit_rmse_rad: float pilot_mean_block_coherence: float legacy_prbs_adjacent_phase_hz: float bit_pattern_diagnostics: dict @dataclass(frozen=True) class AcceptanceResult: """Раздельные критерии радиотракта и прикладного уровня.""" frames_found: int packets_parsed: int packets_crc_valid: int fragments_recovered: int image_reassembled: bool image_size_matches: bool image_sha256_matches: bool software_tests_passed: bool @property def radio_passed(self) -> bool: return ( self.frames_found == EXPECTED_FRAME_COUNT and self.packets_parsed == EXPECTED_FRAME_COUNT and self.packets_crc_valid == EXPECTED_FRAME_COUNT ) @property def application_passed(self) -> bool: return ( self.fragments_recovered == EXPECTED_FRAME_COUNT and self.image_reassembled and self.image_size_matches and self.image_sha256_matches ) @property def passed(self) -> bool: return self.radio_passed and self.application_passed and self.software_tests_passed @dataclass(frozen=True) class FunctionalTestResult: """Результат одной встроенной программной проверки.""" name: str passed: bool detail: str @dataclass(frozen=True) class AsyncCallbackBoundary: """Граница одного callback непрерывного RTL-SDR-захвата.""" callback_index: int source_sample_count: int accepted_start_sample: int accepted_end_sample: int accepted_sample_count: int warmup_discarded: bool trimmed_at_target: bool class ContinuousAsyncIqCollector: """Собрать одну запись из последовательно пронумерованных callback-блоков. Коллектор ничего не знает об аппаратуре и проверяет программную часть непрерывного захвата: повтор, пропуск или перестановка номера callback приводят к ошибке. Первый заранее заданный callback можно отбросить как разогревочный, не перезапуская асинхронный сеанс. """ def __init__( self, expected_sample_count: int, warmup_callback_count: int = RTL_ASYNC_WARMUP_CALLBACK_COUNT, ) -> None: if expected_sample_count <= 0: raise ValueError("Ожидаемое число отсчётов должно быть положительным") if warmup_callback_count < 0: raise ValueError("Число разогревочных callback не может быть отрицательным") self.expected_sample_count = int(expected_sample_count) self.warmup_callback_count = int(warmup_callback_count) self._expected_callback_index = 0 self._accepted_sample_count = 0 self._blocks: list[np.ndarray] = [] self._boundaries: list[AsyncCallbackBoundary] = [] @property def complete(self) -> bool: return self._accepted_sample_count == self.expected_sample_count @property def accepted_sample_count(self) -> int: return self._accepted_sample_count @property def callback_boundaries(self) -> tuple[AsyncCallbackBoundary, ...]: return tuple(self._boundaries) def add_callback(self, callback_index: int, samples: np.ndarray) -> bool: """Добавить ровно следующий callback и вернуть признак завершения.""" if self.complete: raise RuntimeError("Захват уже набрал заданное число отсчётов") if callback_index != self._expected_callback_index: if callback_index < self._expected_callback_index: raise ValueError("Callback продублирован или переставлен назад") raise ValueError("В последовательности callback обнаружен пропуск") block = np.asarray(samples, dtype=np.complex64) if block.ndim != 1 or block.size == 0: raise ValueError("Callback должен содержать непустой одномерный IQ-блок") warmup = callback_index < self.warmup_callback_count start = self._accepted_sample_count if warmup: accepted = 0 else: accepted = min(block.size, self.expected_sample_count - start) self._blocks.append(block[:accepted].copy()) self._accepted_sample_count += accepted end = self._accepted_sample_count self._boundaries.append( AsyncCallbackBoundary( callback_index=callback_index, source_sample_count=int(block.size), accepted_start_sample=int(start), accepted_end_sample=int(end), accepted_sample_count=int(accepted), warmup_discarded=warmup, trimmed_at_target=bool(not warmup and accepted < block.size), ) ) self._expected_callback_index += 1 return self.complete def finalize(self) -> np.ndarray: """Вернуть запись только после получения точного целевого размера.""" if not self.complete: raise RuntimeError("Асинхронный захват короче заданной длины") combined = np.concatenate(self._blocks).astype(np.complex64, copy=False) if len(combined) != self.expected_sample_count: raise RuntimeError("Сумма принятых callback не совпала с целевой длиной") return combined def capture_continuous_rtlsdr_async( sdr, expected_sample_count: int, capture_ready_event=None, buffer_bytes: int = RTL_ASYNC_BUFFER_BYTES, buffer_count: int = RTL_ASYNC_BUFFER_COUNT, warmup_callback_count: int = RTL_ASYNC_WARMUP_CALLBACK_COUNT, ) -> tuple[np.ndarray, tuple[AsyncCallbackBoundary, ...], dict]: """Принять RTL-SDR IQ одним непрерывным ``rtlsdr_read_async``. Используется существующая ctypes-привязка pyrtlsdr. На установленной Windows-сборке высокоуровневый ``read_bytes_async`` после длительной штатной отмены пытается повторно закрыть уже недоступный USB-дескриптор. Поэтому здесь напрямую вызывается уже объявленная функция librtlsdr, без собственной DLL-обёртки. Вызывающий код не должен повторно использовать объект SDR, если ``read_async_result`` отрицателен. """ if buffer_bytes <= 0 or buffer_bytes % 512 != 0: raise ValueError("Размер async-буфера должен быть положительным и кратным 512 байтам") if buffer_bytes % 16_384 != 0: raise ValueError("Размер async-буфера должен быть кратным 16384 байтам") if buffer_bytes % RTL_BYTES_PER_COMPLEX_SAMPLE != 0: raise ValueError("Async-буфер должен содержать целое число комплексных отсчётов") if buffer_count <= 0: raise ValueError("Число async-буферов должно быть положительным") from ctypes import py_object import threading import time from rtlsdr import librtlsdr from rtlsdr.librtlsdr import rtlsdr_read_async_cb_t collector = ContinuousAsyncIqCollector( expected_sample_count, warmup_callback_count=warmup_callback_count, ) done = threading.Event() cancel_errors: list[str] = [] callback_index = 0 started = time.perf_counter() def receive_bytes(raw_bytes, _context) -> None: nonlocal callback_index iq = sdr.packed_bytes_to_iq(raw_bytes) complete = collector.add_callback(callback_index, iq) callback_index += 1 if callback_index == warmup_callback_count and capture_ready_event is not None: capture_ready_event.set() if complete: done.set() def cancel_when_complete() -> None: done.wait() try: sdr.cancel_read_async() except Exception as error: # pragma: no cover - зависит от USB-библиотеки cancel_errors.append(f"{type(error).__name__}: {error}") sdr.DEFAULT_ASYNC_BUF_NUMBER = int(buffer_count) sdr._callback_bytes = receive_bytes sdr.read_async_canceling = False callback = rtlsdr_read_async_cb_t(sdr._bytes_converter_callback) cancel_thread = threading.Thread(target=cancel_when_complete, daemon=True) cancel_thread.start() result = librtlsdr.rtlsdr_read_async( sdr.dev_p, callback, py_object(sdr), int(buffer_count), int(buffer_bytes), ) elapsed = time.perf_counter() - started cancel_thread.join(timeout=5.0) if not collector.complete: raise RuntimeError( f"Асинхронное чтение завершилось до целевого размера, код {result}" ) if result not in (0, -5): raise RuntimeError(f"Асинхронное чтение завершилось с кодом {result}") samples = collector.finalize() diagnostics = { "read_async_result": int(result), "elapsed_seconds": float(elapsed), "callback_count": int(callback_index), "callback_boundary_count": max(0, int(callback_index) - 1), "buffer_bytes": int(buffer_bytes), "buffer_count": int(buffer_count), "warmup_callback_count": int(warmup_callback_count), "cancel_errors": tuple(cancel_errors), } return samples, collector.callback_boundaries, diagnostics def receive_duration_seconds(tx_sample_count: int, actual_tx_sample_rate_hz: float) -> float: """Рассчитать длительность единого приёма из фактического TX-сигнала.""" if tx_sample_count < 0: raise ValueError("Число TX-отсчётов не может быть отрицательным") if not np.isfinite(actual_tx_sample_rate_hz) or actual_tx_sample_rate_hz <= 0.0: raise ValueError("Фактическая частота TX должна быть положительной") return ( RX_LEADING_MARGIN_SECONDS + tx_sample_count / actual_tx_sample_rate_hz + RX_TRAILING_MARGIN_SECONDS ) def receive_sample_count( tx_sample_count: int, actual_tx_sample_rate_hz: float, actual_rx_sample_rate_hz: float, ) -> int: """Вернуть число RX-отсчётов для единого непрерывного захвата.""" if not np.isfinite(actual_rx_sample_rate_hz) or actual_rx_sample_rate_hz <= 0.0: raise ValueError("Фактическая частота RX должна быть положительной") duration = receive_duration_seconds(tx_sample_count, actual_tx_sample_rate_hz) return math.ceil(duration * actual_rx_sample_rate_hz) def combine_continuous_blocks( blocks: Iterable[np.ndarray], expected_sample_count: int, ) -> np.ndarray: """Объединить последовательные блоки чтения в одну временную запись.""" if expected_sample_count < 0: raise ValueError("Ожидаемое число отсчётов не может быть отрицательным") arrays = [np.asarray(block, dtype=np.complex64) for block in blocks] if any(array.ndim != 1 for array in arrays): raise ValueError("Каждый блок должен быть одномерным") combined = np.concatenate(arrays) if arrays else np.empty(0, dtype=np.complex64) if len(combined) < expected_sample_count: raise ValueError("Непрерывный захват короче рассчитанной длительности") return combined[:expected_sample_count] def shape_frame_like_lab042( symbols: np.ndarray, samples_per_symbol: int = SAMPLES_PER_SYMBOL, ) -> np.ndarray: """Сформировать кадр с существующими защитными интервалами Lab042.""" symbols = np.asarray(symbols, dtype=np.complex128) if symbols.ndim != 1 or symbols.size == 0: raise ValueError("Нужна непустая одномерная последовательность символов") if samples_per_symbol <= 0: raise ValueError("Число отсчётов на символ должно быть положительным") taps = radio.root_raised_cosine_taps( LAB042_RRC_ROLLOFF, samples_per_symbol, LAB042_RRC_SPAN_SYMBOLS, ) upsampled = np.zeros(len(symbols) * samples_per_symbol, dtype=np.complex128) upsampled[::samples_per_symbol] = symbols guard = np.zeros( LAB042_GUARD_SYMBOL_COUNT * samples_per_symbol, dtype=np.complex128, ) return ( fftconvolve(np.concatenate((guard, upsampled, guard)), taps, mode="full") * LAB042_TX_AMPLITUDE ) def concatenate_frames_without_new_gaps(frame_waveforms: Iterable[np.ndarray]) -> np.ndarray: """Соединить готовые кадры без добавочного интервала Lab043.""" frames = [np.asarray(frame, dtype=np.complex128) for frame in frame_waveforms] if any(frame.ndim != 1 for frame in frames): raise ValueError("Каждый кадр должен быть одномерным") return np.concatenate(frames) if frames else np.empty(0, dtype=np.complex128) def prbs11( length: int = PRBS11_LENGTH, initial_state: int = PRBS11_INITIAL_STATE, ) -> np.ndarray: """Сформировать детерминированную PRBS11 с полиномом x^11+x^9+1.""" if length <= 0: raise ValueError("Длина PRBS11 должна быть положительной") if not 1 <= initial_state <= 0x7FF: raise ValueError("Начальное состояние PRBS11 должно быть ненулевым 11-битным") state = initial_state bits = np.empty(length, dtype=np.uint8) for index in range(length): bits[index] = (state >> 10) & 1 feedback = ((state >> 10) ^ (state >> 8)) & 1 state = ((state << 1) & 0x7FF) | feedback return bits def pilot_prbs7( length: int = PILOT_SYMBOL_COUNT, initial_state: int = PILOT_INITIAL_STATE, ) -> np.ndarray: """Сформировать отдельный детерминированный пилот x^7+x^6+1. Последовательность и начальное состояние фиксированы до аппаратного опыта и не зависят от полезной PRBS11 либо сохранённого IQ. """ if length <= 0: raise ValueError("Длина пилота должна быть положительной") if not 1 <= initial_state <= 0x7F: raise ValueError("Начальное состояние пилота должно быть ненулевым 7-битным") state = initial_state bits = np.empty(length, dtype=np.uint8) for index in range(length): bits[index] = (state >> 6) & 1 feedback = ((state >> 6) ^ (state >> 5)) & 1 state = ((state << 1) & 0x7F) | feedback return bits def build_prbs11_transmission_plan( label: str, useful_duration_seconds: float, sample_rate_hz: int = SAMPLE_RATE_HZ, samples_per_symbol: int = SAMPLES_PER_SYMBOL, ) -> PrbsTransmissionPlan: """Собрать один непрерывный TX: два тона, фиксированный guard и PRBS11. Перед PRBS используется ровно один известный маркер для начальной синхронизации. Внутри полезной PRBS периодических маркеров и повторных запусков синхронизации нет. """ if label not in {"S", "L"}: raise ValueError("Метка опыта должна быть S или L") if useful_duration_seconds <= 0.0: raise ValueError("Полезная длительность PRBS11 должна быть положительной") if sample_rate_hz != samples_per_symbol * SYMBOL_RATE: raise ValueError("Fs должна быть целым числом отсчётов на символ") payload_bit_count = round(useful_duration_seconds * SYMBOL_RATE) if not math.isclose( payload_bit_count / SYMBOL_RATE, useful_duration_seconds, rel_tol=0.0, abs_tol=1e-12, ): raise ValueError("Полезная длительность должна содержать целое число символов") payload_bits = prbs11(payload_bit_count, PRBS11_INITIAL_STATE) marker_bits, marker_symbols = radio.build_frame_marker() pilot_bits = ( pilot_prbs7(PILOT_SYMBOL_COUNT, PILOT_INITIAL_STATE) if label == "S" else np.empty(0, dtype=np.uint8) ) bpsk_symbols = np.concatenate( ( marker_symbols, radio.bpsk_modulate(pilot_bits), radio.bpsk_modulate(payload_bits), ) ) bpsk_samples = shape_frame_like_lab042(bpsk_symbols, samples_per_symbol) calibration_sample_count = round(CALIBRATION_TONE_DURATION_SECONDS * sample_rate_hz) indexes = np.arange(calibration_sample_count, dtype=np.float64) calibration = 0.35 * ( np.exp(-1j * 2.0 * np.pi * F_CAL_HZ * indexes / sample_rate_hz) + np.exp(1j * 2.0 * np.pi * F_CAL_HZ * indexes / sample_rate_hz) ) fixed_guard_sample_count = round(CALIBRATION_TO_BPSK_GUARD_SECONDS * sample_rate_hz) fixed_guard = np.zeros(fixed_guard_sample_count, dtype=np.complex128) tx_samples = np.concatenate((calibration, fixed_guard, bpsk_samples)) peak = float(np.max(np.abs(tx_samples))) if peak > 1.0: raise ValueError(f"Пик TX-последовательности {peak:.6f} превышает единицу") return PrbsTransmissionPlan( label=label, useful_duration_seconds=float(useful_duration_seconds), payload_bits=payload_bits, marker_bits=marker_bits, pilot_bits=pilot_bits, tx_samples=tx_samples, calibration_sample_count=calibration_sample_count, fixed_guard_sample_count=fixed_guard_sample_count, bpsk_start_sample=calibration_sample_count + fixed_guard_sample_count, ) def to_pluto_tx_samples(samples: np.ndarray) -> np.ndarray: """Преобразовать нормированные комплексные отсчёты в формат Pluto+.""" waveform = np.asarray(samples, dtype=np.complex128) if waveform.ndim != 1 or waveform.size == 0: raise ValueError("TX-последовательность должна быть непустой и одномерной") peak = float(np.max(np.abs(waveform))) if peak > 1.0: raise ValueError("Амплитуда TX-последовательности превышает единицу") return (waveform * (2**14)).astype(np.complex64) def locate_calibration_tone_start( samples: np.ndarray, sample_rate_hz: float, block_samples: int = 8_192, hop_samples: int = 4_096, ) -> int: """Найти первый сильный участок заранее первой двухтоновой посылки.""" received = np.asarray(samples, dtype=np.complex128) if received.ndim != 1 or len(received) < 4 * block_samples: raise ValueError("Захват слишком короток для поиска калибровки") if sample_rate_hz <= 0.0 or block_samples <= 0 or hop_samples <= 0: raise ValueError("Параметры поиска должны быть положительными") centered = received - np.mean(received[: min(len(received), round(0.3 * sample_rate_hz))]) starts = np.arange(0, len(centered) - block_samples + 1, hop_samples, dtype=np.int64) powers = np.asarray( [np.mean(np.abs(centered[start : start + block_samples]) ** 2) for start in starts], dtype=np.float64, ) baseline_limit = max(3, int(0.30 * sample_rate_hz / hop_samples)) baseline = float(np.median(powers[:baseline_limit])) threshold = max(baseline * 10.0, np.finfo(np.float64).tiny) active = powers >= threshold for index in range(len(active) - 2): if bool(np.all(active[index : index + 3])): return int(starts[index]) raise RuntimeError("Начало двухтоновой калибровки не найдено по фиксированному порогу 10 дБ") def split_calibration_and_bpsk( samples: np.ndarray, plan: PrbsTransmissionPlan, actual_rx_sample_rate_hz: float, actual_tx_sample_rate_hz: float, ) -> tuple[np.ndarray, np.ndarray, dict]: """Выделить тоны и BPSK по одной временной шкале одного захвата.""" received = np.asarray(samples, dtype=np.complex128) tone_start = locate_calibration_tone_start(received, actual_rx_sample_rate_hz) tone_duration_rx = round( plan.calibration_sample_count / actual_tx_sample_rate_hz * actual_rx_sample_rate_hz ) edge = round(0.03 * actual_rx_sample_rate_hz) calibration_start = tone_start + edge calibration_end = tone_start + tone_duration_rx - edge if calibration_end <= calibration_start: raise RuntimeError("После удаления переходных краёв калибровочный участок пуст") bpsk_nominal_start = tone_start + round( plan.bpsk_start_sample / actual_tx_sample_rate_hz * actual_rx_sample_rate_hz ) search_margin = round(0.025 * actual_rx_sample_rate_hz) bpsk_tx_count = len(plan.tx_samples) - plan.bpsk_start_sample bpsk_rx_count = round(bpsk_tx_count / actual_tx_sample_rate_hz * actual_rx_sample_rate_hz) bpsk_start = max(0, bpsk_nominal_start - search_margin) bpsk_end = min(len(received), bpsk_nominal_start + bpsk_rx_count + search_margin) if bpsk_end <= bpsk_start: raise RuntimeError("BPSK-участок не помещается в захват") return ( received[calibration_start:calibration_end], received[bpsk_start:bpsk_end], { "tone_start_sample": int(tone_start), "calibration_start_sample": int(calibration_start), "calibration_end_sample": int(calibration_end), "bpsk_nominal_start_sample": int(bpsk_nominal_start), "bpsk_slice_start_sample": int(bpsk_start), "bpsk_slice_end_sample": int(bpsk_end), }, ) def estimate_refined_calibration( calibration_samples: np.ndarray, actual_sample_rate_hz: float, ) -> CalibrationResult: """Оценить тоны спектром и уточнить обе частоты по фазовому наклону.""" coarse = estimate_calibration(calibration_samples, actual_sample_rate_hz) if not coarse.valid: return coarse try: refined = radio.refine_two_tone_frequencies_from_phase( calibration_samples, actual_sample_rate_hz, coarse.f_low_hz, coarse.f_high_hz, block_samples=max(256, round(actual_sample_rate_hz / 10_000.0)), hop_samples=max(128, round(actual_sample_rate_hz / 20_000.0)), ) except (ValueError, RuntimeError) as error: return _invalid_calibration( f"фазовое уточнение не выполнено: {error}", peak_margin_low_db=coarse.peak_margin_low_db, peak_margin_high_db=coarse.peak_margin_high_db, noise_power=coarse.noise_power, ) offsets = radio.estimate_two_tone_offsets( refined.low_frequency_hz, refined.high_frequency_hz, F_CAL_HZ, ) valid = refined.valid and all(np.isfinite(value) for value in offsets.values()) if not valid: return _invalid_calibration( refined.invalid_reason or "фазовое уточнение двух тонов недостоверно", peak_margin_low_db=coarse.peak_margin_low_db, peak_margin_high_db=coarse.peak_margin_high_db, noise_power=coarse.noise_power, ) carrier_offset_hz = float(offsets["carrier_offset_hz"]) return CalibrationResult( valid=True, invalid_reason="", f_low_hz=refined.low_frequency_hz, f_high_hz=refined.high_frequency_hz, carrier_offset_hz=carrier_offset_hz, carrier_offset_ppm=carrier_offset_hz / CARRIER_HZ * 1e6, clock_scale=float(offsets["clock_scale"]), sample_clock_error_ppm=float(offsets["sample_clock_error_ppm"]), residual_cfo_hz=refined.residual_cfo_hz, phase_fit_rmse_rad=refined.phase_fit_rmse_rad, peak_margin_low_db=coarse.peak_margin_low_db, peak_margin_high_db=coarse.peak_margin_high_db, noise_power=coarse.noise_power, ) def _invalid_calibration( reason: str, *, peak_margin_low_db: float = float("nan"), peak_margin_high_db: float = float("nan"), noise_power: float = float("nan"), ) -> CalibrationResult: """Вернуть полный недействительный контракт, не опуская поля.""" nan = float("nan") return CalibrationResult( valid=False, invalid_reason=str(reason), f_low_hz=nan, f_high_hz=nan, carrier_offset_hz=nan, carrier_offset_ppm=nan, clock_scale=nan, sample_clock_error_ppm=nan, residual_cfo_hz=nan, phase_fit_rmse_rad=nan, peak_margin_low_db=float(peak_margin_low_db), peak_margin_high_db=float(peak_margin_high_db), noise_power=float(noise_power), ) def bit_error_rate(expected_bits: np.ndarray, received_bits: np.ndarray) -> tuple[int, float]: """Измерить ошибки битов до пакетного разбора и CRC.""" expected = np.asarray(expected_bits, dtype=np.uint8) received = np.asarray(received_bits, dtype=np.uint8) if expected.ndim != 1 or received.ndim != 1: raise ValueError("Битовые последовательности должны быть одномерными") if expected.size == 0 or len(expected) != len(received): return 0, float("nan") if not np.all((expected <= 1) & (received <= 1)): raise ValueError("Последовательности должны содержать только нули и единицы") errors = int(np.count_nonzero(expected != received)) return errors, errors / len(expected) def _parabolic_peak_frequency( frequencies_hz: np.ndarray, powers: np.ndarray, peak_index: int, ) -> float: """Уточнить частоту максимума параболой по логарифму мощности.""" if peak_index <= 0 or peak_index >= len(powers) - 1: return float(frequencies_hz[peak_index]) local = np.maximum(powers[peak_index - 1 : peak_index + 2], np.finfo(float).tiny) left, center, right = np.log(local) denominator = left - 2.0 * center + right if not np.isfinite(denominator) or abs(denominator) < 1e-15: return float(frequencies_hz[peak_index]) offset_bins = 0.5 * (left - right) / denominator offset_bins = float(np.clip(offset_bins, -1.0, 1.0)) bin_width_hz = float(frequencies_hz[peak_index + 1] - frequencies_hz[peak_index]) return float(frequencies_hz[peak_index] + offset_bins * bin_width_hz) def estimate_calibration_from_spectrum( frequency_axis_hz: np.ndarray, power_spectrum: np.ndarray, ) -> CalibrationResult: """Применить зафиксированную методику Lab043 к готовому спектру мощности.""" frequencies = np.asarray(frequency_axis_hz, dtype=np.float64) powers = np.asarray(power_spectrum, dtype=np.float64) if frequencies.ndim != 1 or powers.ndim != 1 or len(frequencies) != len(powers): raise ValueError("Ось частот и спектр мощности должны быть одномерными и равными") analysis = ( (np.abs(frequencies) <= CALIBRATION_ANALYSIS_HALF_BAND_HZ) & np.isfinite(frequencies) & np.isfinite(powers) & (powers >= 0.0) ) candidates = radio.find_known_tone_peaks( frequencies[analysis], powers[analysis], F_CAL_HZ, CALIBRATION_SEARCH_HALF_WIDTH_HZ, ) low_candidate = float(candidates["low_frequency_hz"]) high_candidate = float(candidates["high_frequency_hz"]) if np.isfinite(low_candidate): low_index = int(np.nanargmin(np.abs(frequencies - low_candidate))) low_candidate = _parabolic_peak_frequency(frequencies, powers, low_index) if np.isfinite(high_candidate): high_index = int(np.nanargmin(np.abs(frequencies - high_candidate))) high_candidate = _parabolic_peak_frequency(frequencies, powers, high_index) noise_mask = analysis.copy() if np.isfinite(low_candidate): noise_mask &= np.abs(frequencies - low_candidate) > CALIBRATION_GUARD_HALF_WIDTH_HZ if np.isfinite(high_candidate): noise_mask &= np.abs(frequencies - high_candidate) > CALIBRATION_GUARD_HALF_WIDTH_HZ noise_values = powers[noise_mask] noise_power = float(np.median(noise_values)) if noise_values.size else float("nan") def excess_db(peak_power: float) -> float: if not np.isfinite(peak_power) or not np.isfinite(noise_power) or noise_power <= 0.0: return float("nan") return float(10.0 * np.log10(peak_power / noise_power)) low_excess = excess_db(float(candidates["low_power"])) high_excess = excess_db(float(candidates["high_power"])) peaks_valid = ( np.isfinite(low_candidate) and np.isfinite(high_candidate) and np.isfinite(low_excess) and np.isfinite(high_excess) and low_excess >= MINIMUM_TONE_EXCESS_DB and high_excess >= MINIMUM_TONE_EXCESS_DB ) if not peaks_valid: reason = "два тона не превышают медианный фон на 10 дБ" return _invalid_calibration( reason, peak_margin_low_db=low_excess, peak_margin_high_db=high_excess, noise_power=noise_power, ) offsets = radio.estimate_two_tone_offsets(low_candidate, high_candidate, F_CAL_HZ) valid = all(np.isfinite(offsets[name]) for name in offsets) if valid and abs(float(offsets["sample_clock_error_ppm"])) > MAXIMUM_ABS_SAMPLE_CLOCK_ERROR_PPM: return _invalid_calibration( "разнос тонов означает ошибку такта более 1000 ppm", peak_margin_low_db=low_excess, peak_margin_high_db=high_excess, noise_power=noise_power, ) carrier_offset_hz = float(offsets["carrier_offset_hz"]) if not valid: return _invalid_calibration( "невычислимая геометрия двух тонов", peak_margin_low_db=low_excess, peak_margin_high_db=high_excess, noise_power=noise_power, ) return CalibrationResult( valid=True, invalid_reason="", f_low_hz=low_candidate, f_high_hz=high_candidate, carrier_offset_hz=carrier_offset_hz, carrier_offset_ppm=carrier_offset_hz / CARRIER_HZ * 1e6, clock_scale=float(offsets["clock_scale"]), sample_clock_error_ppm=float(offsets["sample_clock_error_ppm"]), residual_cfo_hz=float("nan"), phase_fit_rmse_rad=float("nan"), peak_margin_low_db=low_excess, peak_margin_high_db=high_excess, noise_power=noise_power, ) def estimate_calibration(samples: np.ndarray, actual_sample_rate_hz: float) -> CalibrationResult: """Оценить калибровку Lab043 по комплексным отсчётам.""" received = np.asarray(samples, dtype=np.complex128) if received.ndim != 1 or len(received) < CALIBRATION_NFFT: raise ValueError(f"Нужно не менее {CALIBRATION_NFFT} комплексных отсчётов") if not np.isfinite(actual_sample_rate_hz) or actual_sample_rate_hz <= 0.0: raise ValueError("Фактическая частота дискретизации должна быть положительной") centered = received - np.mean(received) frequencies, powers = welch( centered, fs=actual_sample_rate_hz, window="hann", nperseg=CALIBRATION_NFFT, noverlap=CALIBRATION_NFFT // 2, nfft=CALIBRATION_NFFT, return_onesided=False, scaling="density", ) order = np.argsort(frequencies) return estimate_calibration_from_spectrum(frequencies[order], powers[order]) def summarize_calibrations(estimates: Iterable[CalibrationResult]) -> dict: """Свести отдельные оценки без замены невычислимых значений нулём.""" rows = tuple(estimates) if len(rows) != 3: raise ValueError("Для Lab043 нужны ровно три калибровочных захвата") fields = ( "f_low_hz", "f_high_hz", "peak_margin_low_db", "peak_margin_high_db", "carrier_offset_hz", "carrier_offset_ppm", "clock_scale", "sample_clock_error_ppm", ) summary: dict[str, object] = { "capture_count": len(rows), "failure_count": sum(not row.valid for row in rows), } for field_name in fields: values = np.asarray([getattr(row, field_name) for row in rows], dtype=np.float64) finite = values[np.isfinite(values)] summary[field_name] = { "mean": float(np.mean(finite)) if finite.size else float("nan"), "minimum": float(np.min(finite)) if finite.size else float("nan"), "maximum": float(np.max(finite)) if finite.size else float("nan"), "standard_deviation": float(np.std(finite, ddof=0)) if finite.size else float("nan"), } return summary def synthesize_known_bpsk_capture( carrier_offset_hz: float, clock_scale: float, samples_per_symbol: int = 16, initial_sample_phase: int = 5, ) -> tuple[np.ndarray, np.ndarray]: """Синтетически сформировать одну запись маркера и полной PRBS11.""" _marker_bits, marker_symbols = radio.build_frame_marker() payload_bits = prbs11() symbols = np.concatenate((marker_symbols, radio.bpsk_modulate(payload_bits))) prefix = np.zeros(3 * samples_per_symbol + initial_sample_phase, dtype=np.complex128) suffix = np.zeros(3 * samples_per_symbol, dtype=np.complex128) transmitter_samples = np.concatenate((prefix, np.repeat(symbols, samples_per_symbol), suffix)) received_length = max(1, math.floor(len(transmitter_samples) / clock_scale)) receive_indexes = np.arange(received_length, dtype=np.float64) transmitter_positions = np.minimum( receive_indexes * clock_scale, len(transmitter_samples) - 1.0, ) source_indexes = np.arange(len(transmitter_samples), dtype=np.float64) received = np.interp(transmitter_positions, source_indexes, transmitter_samples.real).astype( np.complex128 ) sample_rate_hz = SYMBOL_RATE * samples_per_symbol received *= np.exp( 1j * 2.0 * np.pi * carrier_offset_hz * receive_indexes / sample_rate_hz ) return received, payload_bits def process_bpsk_modes( received_samples: np.ndarray, expected_payload_bits: np.ndarray, coarse_carrier_offset_hz: float, clock_scale: float, samples_per_symbol: int, ) -> dict[str, ModeResult]: """Обработать одну запись режимами A, B и C без изменения исходника.""" source = np.asarray(received_samples, dtype=np.complex128) marker_bits, marker_symbols = radio.build_frame_marker() expected_payload = np.asarray(expected_payload_bits, dtype=np.uint8) required_symbol_count = len(marker_bits) + len(expected_payload) sample_rate_hz = SYMBOL_RATE * samples_per_symbol results: dict[str, ModeResult] = {} for mode in ("A", "B", "C"): processed = source.copy() if mode in ("B", "C"): processed = radio.apply_coarse_frequency_correction( processed, coarse_carrier_offset_hz, sample_rate_hz, ) if mode == "C": processed = radio.resample_for_clock_scale(processed, clock_scale) try: found = radio.find_radio_frame(processed, marker_symbols, samples_per_symbol) start = int(found["start_symbol_index"]) symbols = found["symbol_samples"][start : start + required_symbol_count] if len(symbols) != required_symbol_count: raise RuntimeError("Известная последовательность обрезана") estimate = radio.estimate_carrier_parameters(symbols[: len(marker_bits)], marker_symbols) fine_cfo_hz = float(estimate["estimated_frequency_hz"]) fine_applied = mode == "C" and radio.should_apply_cfo_correction( fine_cfo_hz, float(estimate["phase_consistency"]), float(estimate["coherence_gain"]), ) if fine_applied: corrected_symbols = radio.correct_phase_and_frequency( symbols, float(estimate["initial_phase_after_cfo"]), float(estimate["phase_increment"]), ) else: corrected_symbols = radio.correct_phase_and_frequency( symbols, float(estimate["constant_phase"]), 0.0, ) received_payload = radio.bpsk_demodulate(corrected_symbols[len(marker_bits) :]) errors, ber = bit_error_rate(expected_payload, received_payload) results[mode] = ModeResult( mode, ber, errors, len(expected_payload), float(found["score"]), int(found["sample_phase"]), fine_applied, fine_cfo_hz, ) except (RuntimeError, ValueError): results[mode] = ModeResult( mode, float("nan"), 0, 0, 0.0, 0, False, float("nan"), ) return results def _maximum_true_run(values: np.ndarray) -> int: best = current = 0 for value in np.asarray(values, dtype=bool): current = current + 1 if value else 0 best = max(best, current) return int(best) def _best_local_timing_offset( matched_samples: np.ndarray, nominal_start_sample: int, expected_symbols: np.ndarray, samples_per_symbol: int, ) -> float: """Оценить локальное смещение символьной сетки без повторной синхронизации.""" expected = np.asarray(expected_symbols, dtype=np.complex128) if expected.size == 0: return float("nan") count = min(len(expected), 2_048) expected = expected[:count] best_offset = 0 best_score = -1.0 for offset in range(-samples_per_symbol // 2, samples_per_symbol // 2 + 1): indexes = nominal_start_sample + offset + np.arange(count) * samples_per_symbol if indexes[0] < 0 or indexes[-1] >= len(matched_samples): continue observed = matched_samples[indexes] denominator = math.sqrt( float(np.sum(np.abs(observed) ** 2) * np.sum(np.abs(expected) ** 2)) ) + 1e-12 score = float(abs(np.vdot(expected, observed)) / denominator) if score > best_score: best_score = score best_offset = offset if best_score < 0.0: return float("nan") return float(best_offset) def _failed_prbs_metrics(mode: str, transmitted_bit_count: int, reason: str) -> PrbsModeMetrics: nan = float("nan") return PrbsModeMetrics( mode=mode, detected=False, failure_reason=reason, transmitted_bit_count=int(transmitted_bit_count), matched_bit_count=0, bit_errors=0, bit_error_rate=nan, quarter_bit_error_rates=(nan, nan, nan, nan), evm_percent=nan, marker_correlation=0.0, symbol_phase_start_samples=nan, symbol_phase_end_samples=nan, accumulated_timing_drift_samples=nan, accumulated_timing_drift_symbols=nan, residual_cfo_hz=nan, maximum_correct_run_bits=0, fine_cfo_applied=False, estimated_fine_cfo_hz=nan, timing_sro_ppm=nan, coarse_cfo_applied=False, pilot_cfo_applied=False, pilot_estimate_valid=False, estimated_pilot_cfo_hz=nan, residual_pilot_cfo_hz=nan, pilot_phase_fit_rmse_rad=nan, pilot_mean_block_coherence=nan, legacy_prbs_adjacent_phase_hz=nan, bit_pattern_diagnostics={}, ) def _prbs_bit_pattern_diagnostics( payload_symbols: np.ndarray, expected_bits: np.ndarray, ) -> dict: """Сгруппировать диагностические ошибки по тройкам соседних битов.""" symbols = np.asarray(payload_symbols, dtype=np.complex128) bits = np.asarray(expected_bits, dtype=np.uint8) expected_symbols = radio.bpsk_modulate(bits) if len(symbols) != len(bits) or len(bits) < 3: return {} gain = np.vdot(expected_symbols, symbols) / ( np.vdot(expected_symbols, expected_symbols) + 1e-12 ) if abs(gain) <= 1e-12: return {} normalized_despread = symbols / gain * expected_symbols received_bits = radio.bpsk_demodulate(symbols) error_mask = received_bits != bits groups: dict[str, dict] = {} indexes = np.arange(1, len(bits) - 1) for previous in (0, 1): for current in (0, 1): for following in (0, 1): selected = indexes[ (bits[indexes - 1] == previous) & (bits[indexes] == current) & (bits[indexes + 1] == following) ] values = normalized_despread[selected] errors = int(np.count_nonzero(error_mask[selected])) groups[f"{previous}{current}{following}"] = { "count": int(len(selected)), "bit_errors": errors, "bit_error_rate": errors / len(selected) if len(selected) else float("nan"), "mean_real": float(np.mean(values.real)) if len(values) else float("nan"), "mean_imag": float(np.mean(values.imag)) if len(values) else float("nan"), "mean_magnitude": float(np.mean(np.abs(values))) if len(values) else float("nan"), } return { "gain_real": float(gain.real), "gain_imag": float(gain.imag), "groups": groups, } def analyze_prbs_modes( bpsk_samples: np.ndarray, expected_payload_bits: np.ndarray, coarse_carrier_offset_hz: float, clock_scale: float, actual_sample_rate_hz: float, samples_per_symbol: int = SAMPLES_PER_SYMBOL, known_pilot_bits: np.ndarray | None = None, ) -> dict[str, PrbsModeMetrics]: """Измерить A/B/C/D на одной записи без знания полезной PRBS при коррекции. A: без грубой CFO и SRO; B: только грубая CFO; C: грубая CFO и отдельный пилот; D: то же, что C, плюс компенсация SRO. Известная PRBS используется только после рабочей коррекции для расчёта метрик. """ source = np.asarray(bpsk_samples, dtype=np.complex128) expected_bits = np.asarray(expected_payload_bits, dtype=np.uint8) pilot_bits = ( pilot_prbs7() if known_pilot_bits is None else np.asarray(known_pilot_bits, dtype=np.uint8) ) if pilot_bits.ndim != 1 or len(pilot_bits) == 0: raise ValueError("Для короткого S нужен отдельный непустой известный пилот") marker_bits, marker_symbols = radio.build_frame_marker() pilot_symbols = radio.bpsk_modulate(pilot_bits) expected_payload_symbols = radio.bpsk_modulate(expected_bits) pilot_start_symbol = len(marker_bits) payload_start_symbol = pilot_start_symbol + len(pilot_bits) required_symbol_count = payload_start_symbol + len(expected_bits) taps = radio.root_raised_cosine_taps( LAB042_RRC_ROLLOFF, samples_per_symbol, LAB042_RRC_SPAN_SYMBOLS, ) results: dict[str, PrbsModeMetrics] = {} for mode in ("A", "B", "C", "D"): try: processed = source.copy() coarse_applied = mode in ("B", "C", "D") pilot_applied = mode in ("C", "D") if coarse_applied: processed = radio.apply_coarse_frequency_correction( processed, coarse_carrier_offset_hz, actual_sample_rate_hz, ) if mode == "D": processed = radio.resample_for_clock_scale(processed, clock_scale) matched = fftconvolve(processed, taps, mode="full") found = radio.find_radio_frame(matched, marker_symbols, samples_per_symbol) marker_correlation = float(found["score"]) if marker_correlation < PRBS_MARKER_MINIMUM_CORRELATION: raise RuntimeError( f"корреляция маркера {marker_correlation:.4f} ниже фиксированного порога " f"{PRBS_MARKER_MINIMUM_CORRELATION:.2f}" ) start_symbol = int(found["start_symbol_index"]) symbols = found["symbol_samples"][start_symbol : start_symbol + required_symbol_count] if len(symbols) != required_symbol_count: raise RuntimeError("маркер, пилот или полезная PRBS11 обрезаны") marker_carrier = radio.estimate_carrier_parameters( symbols[: len(marker_bits)], marker_symbols, ) constant_corrected = radio.correct_phase_and_frequency( symbols, float(marker_carrier["constant_phase"]), 0.0, ) pilot_estimate = None if coarse_applied: pilot_estimate = radio.estimate_known_pilot_carrier( constant_corrected[pilot_start_symbol:payload_start_symbol], pilot_symbols, ) if pilot_applied and (pilot_estimate is None or not pilot_estimate.valid): reason = ( "пилотная оценка отсутствует" if pilot_estimate is None else pilot_estimate.invalid_reason ) raise RuntimeError(f"пилотная оценка недостоверна: {reason}") if pilot_applied and pilot_estimate is not None: corrected_tail = radio.correct_phase_and_frequency( constant_corrected[pilot_start_symbol:], pilot_estimate.initial_phase_rad, pilot_estimate.phase_increment_rad_per_symbol, ) corrected_pilot = corrected_tail[: len(pilot_bits)] payload_symbols = corrected_tail[len(pilot_bits) :] residual_estimate = radio.estimate_known_pilot_carrier( corrected_pilot, pilot_symbols, ) else: corrected_pilot = constant_corrected[ pilot_start_symbol:payload_start_symbol ] payload_symbols = constant_corrected[payload_start_symbol:] residual_estimate = pilot_estimate pilot_estimate_valid = bool( pilot_estimate is not None and pilot_estimate.valid ) estimated_pilot_cfo_hz = ( float(pilot_estimate.frequency_hz) if pilot_estimate_valid and pilot_estimate is not None else float("nan") ) residual_pilot_cfo_hz = ( float(residual_estimate.frequency_hz) if residual_estimate is not None and residual_estimate.valid else float("nan") ) pilot_phase_fit_rmse_rad = ( float(pilot_estimate.phase_fit_rmse_rad) if pilot_estimate is not None else float("nan") ) pilot_mean_block_coherence = ( float(pilot_estimate.mean_block_coherence) if pilot_estimate is not None else float("nan") ) if len(payload_symbols) != len(expected_bits): raise RuntimeError("полезная PRBS11 обрезана после коррекции") received_bits = radio.bpsk_demodulate(payload_symbols) errors, ber = bit_error_rate(expected_bits, received_bits) quarter_bers: list[float] = [] for quarter in np.array_split(np.arange(len(expected_bits)), 4): quarter_errors = int( np.count_nonzero(expected_bits[quarter] != received_bits[quarter]) ) quarter_bers.append(quarter_errors / len(quarter)) gain = np.vdot(expected_payload_symbols, payload_symbols) / ( np.vdot(expected_payload_symbols, expected_payload_symbols) + 1e-12 ) reference = gain * expected_payload_symbols evm_percent = float( 100.0 * math.sqrt( float(np.mean(np.abs(payload_symbols - reference) ** 2)) / (float(np.mean(np.abs(reference) ** 2)) + 1e-12) ) ) # Историческая величина оставлена только с явным названием. Она # не используется как CFO и не участвует ни в одной коррекции. diagnostic_despread = payload_symbols * expected_payload_symbols diagnostic_adjacent = ( diagnostic_despread[1:] * np.conj(diagnostic_despread[:-1]) ) legacy_prbs_adjacent_phase_hz = float( np.angle(np.sum(diagnostic_adjacent)) * SYMBOL_RATE / (2.0 * np.pi) ) marker_start_sample = int(found["sample_phase"]) + start_symbol * samples_per_symbol payload_start_sample = marker_start_sample + payload_start_symbol * samples_per_symbol timing_window = min(2_048, max(1, len(expected_bits) // 4)) start_offset = _best_local_timing_offset( matched, payload_start_sample, expected_payload_symbols[:timing_window], samples_per_symbol, ) end_bit = len(expected_bits) - timing_window end_offset = _best_local_timing_offset( matched, payload_start_sample + end_bit * samples_per_symbol, expected_payload_symbols[end_bit:], samples_per_symbol, ) timing_drift = end_offset - start_offset timing_denominator = max(1, end_bit) * samples_per_symbol timing_sro_ppm = -timing_drift / timing_denominator * 1e6 correct = received_bits == expected_bits results[mode] = PrbsModeMetrics( mode=mode, detected=True, failure_reason="", transmitted_bit_count=len(expected_bits), matched_bit_count=len(received_bits), bit_errors=errors, bit_error_rate=ber, quarter_bit_error_rates=tuple(float(value) for value in quarter_bers), evm_percent=evm_percent, marker_correlation=marker_correlation, symbol_phase_start_samples=float(found["sample_phase"]) + start_offset, symbol_phase_end_samples=float(found["sample_phase"]) + end_offset, accumulated_timing_drift_samples=timing_drift, accumulated_timing_drift_symbols=timing_drift / samples_per_symbol, residual_cfo_hz=residual_pilot_cfo_hz, maximum_correct_run_bits=_maximum_true_run(correct), fine_cfo_applied=bool(pilot_applied), estimated_fine_cfo_hz=estimated_pilot_cfo_hz, timing_sro_ppm=timing_sro_ppm, coarse_cfo_applied=coarse_applied, pilot_cfo_applied=pilot_applied, pilot_estimate_valid=pilot_estimate_valid, estimated_pilot_cfo_hz=estimated_pilot_cfo_hz, residual_pilot_cfo_hz=residual_pilot_cfo_hz, pilot_phase_fit_rmse_rad=pilot_phase_fit_rmse_rad, pilot_mean_block_coherence=pilot_mean_block_coherence, legacy_prbs_adjacent_phase_hz=legacy_prbs_adjacent_phase_hz, bit_pattern_diagnostics=_prbs_bit_pattern_diagnostics( payload_symbols, expected_bits, ), ) except (RuntimeError, ValueError) as error: results[mode] = _failed_prbs_metrics(mode, len(expected_bits), str(error)) return results def control_packet_crc_roundtrip(payload: bytes = b"Lab043 control packet") -> bool: """Проверить один настоящий пакет с CRC внутри радиокадра.""" packet = build_packet(payload, MESSAGE_TYPE_TEXT, sequence_number=43) bits, _, _ = radio.build_radio_frame(packet) recovered_packet = radio.parse_radio_frame(bits) return recovered_packet is not None and parse_packet(recovered_packet).payload == payload def try_reassemble_complete_image(fragments: Iterable) -> bytes | None: """Собрать изображение либо явно вернуть отсутствие полного результата.""" try: return reassemble_image(fragments) except MissingFragmentsError: return None def exit_code_for_acceptance(result: AcceptanceResult) -> int: """Вернуть ноль только по полному критерию Lab043.""" return 0 if result.passed else 1 def save_reference_iq_capture( base_path: Path, samples: np.ndarray, metadata: dict, ) -> tuple[Path, Path]: """Сохранить эталонный IQ и воспроизводимые метаданные. Функция только подготавливает поддержку формата. Аппаратный файл в текущем программном этапе не создаётся. """ iq = np.asarray(samples, dtype=np.complex64) if iq.ndim != 1: raise ValueError("IQ-захват должен быть одномерным") required = { "actual_sample_rate_hz", "center_frequency_hz", "rx_gain_db", "capture_started_utc", "capture_order", "tx_parameters", "calibration", } missing = sorted(required - set(metadata)) if missing: raise ValueError("Не хватает метаданных: " + ", ".join(missing)) base_path = Path(base_path) base_path.parent.mkdir(parents=True, exist_ok=True) iq_path = base_path.with_suffix(".npy") metadata_path = base_path.with_suffix(".json") np.save(iq_path, iq, allow_pickle=False) file_sha256 = hashlib.sha256(iq_path.read_bytes()).hexdigest() document = dict(metadata) document.update( { "sample_count": len(iq), "dtype": "complex64", "iq_file": iq_path.name, "iq_sha256": file_sha256, } ) metadata_path.write_text( json.dumps(document, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) return iq_path, metadata_path def _write_json_atomically(path: Path, document: dict) -> None: temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( json.dumps(document, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) temporary.replace(path) def save_raw_iq_before_processing( base_path: Path, samples: np.ndarray, metadata: dict, ) -> tuple[Path, Path, dict]: """Сохранить и повторно проверить raw до запуска любого анализа.""" iq = np.asarray(samples, dtype=np.complex64) if iq.ndim != 1 or iq.size == 0: raise ValueError("Raw IQ должен быть непустым одномерным массивом") if not np.all(np.isfinite(iq)): raise ValueError("Raw IQ содержит нечисловые значения") required = { "capture_status", "processing_status", "processing_error", "capture_started_utc", "tx_parameters", "rx_parameters", "waveform_sha256", "callback_count", "callback_boundaries", "tx_waveform_duration_seconds", "expected_intervals", } missing = sorted(required - set(metadata)) if missing: raise ValueError("Не хватает raw-метаданных: " + ", ".join(missing)) if metadata["capture_status"] != "captured": raise ValueError("До обработки capture_status должен быть captured") if metadata["processing_status"] != "pending": raise ValueError("До обработки processing_status должен быть pending") base_path = Path(base_path) base_path.parent.mkdir(parents=True, exist_ok=True) iq_path = base_path.with_suffix(".npy") metadata_path = base_path.with_suffix(".json") # Порядок обязателен: NPY -> первичный JSON -> SHA-256 -> проверка NPY. np.save(iq_path, iq, allow_pickle=False) document = dict(metadata) document.update( { "sample_count": int(len(iq)), "dtype": "complex64", "iq_file": iq_path.name, "iq_sha256": None, "raw_verified": False, } ) _write_json_atomically(metadata_path, document) iq_sha256 = hashlib.sha256(iq_path.read_bytes()).hexdigest() restored = np.load(iq_path, allow_pickle=False) verified = ( restored.dtype == np.dtype(np.complex64) and restored.shape == iq.shape and np.array_equal(restored, iq) and hashlib.sha256(iq_path.read_bytes()).hexdigest() == iq_sha256 ) if not verified: raise RuntimeError("Повторная проверка сохранённого raw IQ не прошла") document["iq_sha256"] = iq_sha256 document["raw_verified"] = True _write_json_atomically(metadata_path, document) return iq_path, metadata_path, document def update_processing_metadata( metadata_path: Path, *, processing_status: str, processing_error: str | None, analysis: dict | None = None, ) -> dict: """Безопасно обновить только состояние обработки уже сохранённого raw.""" if processing_status not in {"success", "processing_failed"}: raise ValueError("Неизвестный processing_status") path = Path(metadata_path) document = json.loads(path.read_text(encoding="utf-8")) document["processing_status"] = processing_status document["processing_error"] = processing_error if analysis is not None: document["analysis"] = analysis _write_json_atomically(path, document) return document def save_then_process_capture( base_path: Path, samples: np.ndarray, metadata: dict, processor, ) -> tuple[Path, Path, object | None, str | None]: """Зафиксировать raw, затем обработать; ошибку анализа оставить в JSON.""" iq_path, metadata_path, _document = save_raw_iq_before_processing( base_path, samples, metadata, ) try: result = processor(np.asarray(samples, dtype=np.complex64)) except Exception as error: message = f"{type(error).__name__}: {error}" update_processing_metadata( metadata_path, processing_status="processing_failed", processing_error=message, ) return iq_path, metadata_path, None, message update_processing_metadata( metadata_path, processing_status="success", processing_error=None, analysis=asdict(result) if hasattr(result, "__dataclass_fields__") else result, ) return iq_path, metadata_path, result, None def save_diagnostic_artifacts( output_directory: Path, calibrations: Iterable[CalibrationResult], mode_results: dict[str, ModeResult], acceptance: AcceptanceResult, ) -> tuple[Path, ...]: """Сохранить CSV, TXT и PNG диагностики без обращения к аппаратуре.""" output_directory = Path(output_directory) output_directory.mkdir(parents=True, exist_ok=True) calibration_rows = tuple(calibrations) calibration_csv = output_directory / "lab043_calibration.csv" with calibration_csv.open("w", encoding="utf-8", newline="") as stream: field_names = list(CalibrationResult.__dataclass_fields__) writer = csv.DictWriter(stream, fieldnames=field_names) writer.writeheader() for row in calibration_rows: writer.writerow(asdict(row)) modes_csv = output_directory / "lab043_modes.csv" with modes_csv.open("w", encoding="utf-8", newline="") as stream: field_names = list(ModeResult.__dataclass_fields__) writer = csv.DictWriter(stream, fieldnames=field_names) writer.writeheader() for mode in ("A", "B", "C"): if mode in mode_results: writer.writerow(asdict(mode_results[mode])) summary_csv = output_directory / "lab043_summary.csv" summary_row = asdict(acceptance) | { "radio_passed": acceptance.radio_passed, "application_passed": acceptance.application_passed, "passed": acceptance.passed, "exit_code": exit_code_for_acceptance(acceptance), } with summary_csv.open("w", encoding="utf-8", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=list(summary_row)) writer.writeheader() writer.writerow(summary_row) report_path = output_directory / "lab043_report.txt" report_lines = [ "Lab043: программная диагностика", f"Калибровочных захватов: {len(calibration_rows)}.", f"Радиокритерий: {'пройден' if acceptance.radio_passed else 'не пройден'}.", f"Прикладной критерий: {'пройден' if acceptance.application_passed else 'не пройден'}.", f"Программные проверки: {'пройдены' if acceptance.software_tests_passed else 'не пройдены'}.", f"Код возврата: {exit_code_for_acceptance(acceptance)}.", "Аппаратный тракт этим отчётом не подтверждается.", ] report_path.write_text("\n".join(report_lines) + "\n", encoding="utf-8") plot_path = output_directory / "lab043_calibration.png" figure = Figure(figsize=(7, 4)) axis = figure.subplots() capture_indexes = np.arange(1, len(calibration_rows) + 1) low = [row.f_low_hz for row in calibration_rows] high = [row.f_high_hz for row in calibration_rows] axis.plot(capture_indexes, low, "o-", label="Нижний тон") axis.plot(capture_indexes, high, "o-", label="Верхний тон") axis.set_xlabel("Номер калибровочного захвата") axis.set_ylabel("Частота относительно несущей, Гц") axis.grid(True, alpha=0.3) axis.legend() figure.tight_layout() figure.savefig(plot_path, dpi=140) return calibration_csv, modes_csv, summary_csv, report_path, plot_path def _run_check(name: str, function) -> FunctionalTestResult: try: detail = function() return FunctionalTestResult(name, True, str(detail)) except Exception as error: return FunctionalTestResult(name, False, f"{type(error).__name__}: {error}") def run_functional_tests() -> tuple[FunctionalTestResult, ...]: """Выполнить быстрые синтетические проверки без аппаратуры и файлов.""" def duration_is_derived() -> str: duration = receive_duration_seconds(2_400_000, 2_400_000.0) assert duration == 2.0 assert receive_sample_count(2_400_000, 2_400_000.0, 2_400_000.0) == 4_800_000 return "длительность получена из числа TX-отсчётов" def blocks_are_continuous() -> str: blocks = (np.arange(4), np.arange(4, 9)) combined = combine_continuous_blocks(blocks, 8) assert np.array_equal(combined.real, np.arange(8)) return "последовательные блоки образуют одну запись" def prbs_has_full_period() -> str: sequence = prbs11() assert len(sequence) == PRBS11_LENGTH assert not np.array_equal(sequence, np.roll(sequence, 1)) return "PRBS11 содержит 2047 детерминированных битов" def separate_pilot_is_fixed() -> str: pilot = pilot_prbs7() assert len(pilot) == PILOT_SYMBOL_COUNT assert np.array_equal(pilot[:127], pilot[127:254]) assert int(np.count_nonzero(pilot[:127])) == 64 assert len(pilot) / SYMBOL_RATE == 0.064 return "отдельный PRBS7-пилот содержит 1280 символов и длится 64 мс" def cfo_signs_are_correct() -> str: sample_rate = 10_000.0 indexes = np.arange(10_000) for offset in (-700.0, 700.0): impaired = np.exp(1j * 2.0 * np.pi * offset * indexes / sample_rate) corrected = radio.apply_coarse_frequency_correction(impaired, offset, sample_rate) assert np.max(np.abs(corrected - 1.0)) < 1e-9 return "грубая CFO обоих знаков компенсируется правильным знаком" def clock_scale_cases_are_correct() -> str: for ppm in (20.0, -20.0, 100.0, -100.0): scale = 1.0 + ppm * 1e-6 low = -F_CAL_HZ * scale high = F_CAL_HZ * scale estimate = radio.estimate_two_tone_offsets(low, high, F_CAL_HZ) assert math.isclose(estimate["sample_clock_error_ppm"], ppm, abs_tol=1e-6) source = np.arange(100_000, dtype=np.complex128) corrected = radio.resample_for_clock_scale(source, scale) assert len(corrected) == round(len(source) * scale) return "знак, величина и направление проверены для +/-20 и +/-100 ppm" def weak_tones_are_rejected() -> str: frequencies = np.linspace(-100_000.0, 100_000.0, 4001) powers = np.ones_like(frequencies) powers[np.argmin(np.abs(frequencies + F_CAL_HZ))] = 5.0 powers[np.argmin(np.abs(frequencies - F_CAL_HZ))] = 5.0 estimate = estimate_calibration_from_spectrum(frequencies, powers) assert not estimate.valid and math.isnan(estimate.carrier_offset_hz) return "слабые тоны дают явную невычислимую оценку" def packet_crc_is_valid() -> str: assert control_packet_crc_roundtrip() return "контрольный пакет прошёл CRC" def modes_use_one_capture() -> str: samples_per_symbol = 16 sample_rate = SYMBOL_RATE * samples_per_symbol plan = build_prbs11_transmission_plan( "S", 0.025, sample_rate_hz=sample_rate, samples_per_symbol=samples_per_symbol, ) capture = plan.tx_samples[plan.bpsk_start_sample :] indexes = np.arange(len(capture), dtype=np.float64) capture = capture * np.exp( 1j * 2.0 * np.pi * 701.2 * indexes / sample_rate ) results = analyze_prbs_modes( capture, plan.payload_bits, 700.0, 1.0, sample_rate, samples_per_symbol=samples_per_symbol, known_pilot_bits=plan.pilot_bits, ) assert set(results) == {"A", "B", "C", "D"} assert results["C"].pilot_estimate_valid assert results["C"].bit_error_rate == 0.0 return "режим C восстановил PRBS11 по маркеру и отдельному пилоту" def incomplete_acceptance_fails() -> str: complete = AcceptanceResult(10, 10, 10, 10, True, True, True, True) incomplete = AcceptanceResult(10, 10, 10, 9, False, False, False, True) assert exit_code_for_acceptance(complete) == 0 assert exit_code_for_acceptance(incomplete) == 1 return "код 0 выдаётся только по полному критерию" checks = ( ("01. Расчёт непрерывного захвата", duration_is_derived), ("02. Склейка блоков без разрывов", blocks_are_continuous), ("03. Известная PRBS11", prbs_has_full_period), ("04. Отдельный пилот", separate_pilot_is_fixed), ("05. Знак грубой CFO", cfo_signs_are_correct), ("06. Направление clock_scale", clock_scale_cases_are_correct), ("07. Порог двух тонов", weak_tones_are_rejected), ("08. Один пакет с CRC", packet_crc_is_valid), ("09. Режимы A/B/C/D", modes_use_one_capture), ("10. Полный критерий успеха", incomplete_acceptance_fails), ) return tuple(_run_check(name, function) for name, function in checks) def main() -> int: """Запустить только программные проверки Lab043.""" results = run_functional_tests() for result in results: print(f"{'PASS' if result.passed else 'FAIL'} | {result.name} | {result.detail}") passed = all(result.passed for result in results) print("Аппаратный тракт не запускался.") return 0 if passed else 1 if __name__ == "__main__": raise SystemExit(main())