"""Lab043: воспроизводимый BPSK-опыт HackRF → HackRF при общей опоре. Инструмент превращает уже выполненный вручную аппаратный опыт в повторяемую процедуру внутри проекта. Он НЕ вводит новую архитектуру: форма сигнала и примитивы взяты из уже проверенного кода проекта. BPSK (Binary Phase Shift Keying) — двоичная фазовая манипуляция, два состояния фазы, один бит на символ. Два режима: --analyze-existing offline-анализ сохранённого захвата, оборудование не требуется; --hardware --lna L --vga V один аппаратный опыт (передача + приём); --hardware-ladder последовательная разрешённая gain ladder. Оба аппаратных режима по умолчанию заблокированы и требуют явного --i-understand-hardware. Восстановление формы (см. build_tx_cs8) проверяется по SHA-256; при несовпадении аппаратный режим останавливается до передачи. """ from __future__ import annotations import argparse import hashlib import json import math import re import subprocess import sys import time from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Callable, TextIO if __package__ in (None, ""): sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import numpy as np from scipy.signal import fftconvolve, resample_poly from experiments.lab043_hackrf_hardware import ( signed_iq_bytes_to_complex64, validate_hackrf_rx_gains, ) from experiments.lab043_pluto_to_rtlsdr import pilot_prbs7, shape_frame_like_lab042 from protocol import bpsk_radio as radio # ------------------------------------------------------------------ оборудование MASTER_SERIAL = "000000000000000097c063dc323b97d3" # TX SLAVE_SERIAL = "000000000000000097c063dc2e5327d3" # RX HACKRF_TOOLS_DIR = Path(r"C:\Users\user\Tools\HackRF\Library\bin") # ------------------------------------------------------------------ радиотракт RF_HZ = 1_080_000_000 SAMPLE_RATE_HZ = 8_000_000 SYMBOL_RATE = 20_000 SAMPLES_PER_SYMBOL = SAMPLE_RATE_HZ // SYMBOL_RATE # 400 CARRIER_OFFSET_HZ = 50_000.0 TX_VGA_DB = 10 RF_AMP_ENABLED = False ANTENNA_POWER_ENABLED = False # ------------------------------------------------------------------ форма BPSK PILOT_SYMBOL_COUNT = 1_280 PILOT_INITIAL_STATE = 0x7F PAYLOAD_SEED = 0x043 PAYLOAD_SYMBOL_COUNT = 16_384 TOTAL_SYMBOL_COUNT = PILOT_SYMBOL_COUNT + PAYLOAD_SYMBOL_COUNT # 17664 TARGET_RMS_CODE = 44.0 INT8_LIMIT = 127 # SHA-256 эталонной cs8-формы. Восстанавливается из примитивов, не # подгоняется под это значение; при несовпадении аппаратный режим STOP. EXPECTED_TX_SHA256 = "fed634fb2a4fbf41ea5a1fce547447b8e3e6dbe8110b8429c507637c6d2d1baa" # ------------------------------------------------------------------ пороги анализа PREAMBLE_OVER_MEDIAN_DB_MIN = 12.0 PREAMBLE_OVER_SECOND_DB_MIN = 3.0 DETECTION_DECIMATION = 20 # Восстановлено из фактического анализатора первой gain ladder: 0.05 с # при 20 ксимв/с. Это исключает автокорреляционные повторы одной и той же # 1280-символьной PRBS7-преамбулы, но не соседний независимый пакет. SECOND_PEAK_GUARD_SYMBOLS = 1_000 # Si5351 register 29 после запуска RX при внешней 10 МГц опоре. REG29_EXTERNAL_CLOCK = 0x26 SHORT_RX_SAMPLE_COUNT = 65_536 CAPTURE_SAMPLE_COUNT = 32_000_000 RX_LEAD_SECONDS = 0.75 RX_WAIT_TIMEOUT_SECONDS = 15 PASS_BER_MAX = 0.01 PASS_SNR_DB_MIN = 6.0 PASS_SEPARATION_MIN = 2.0 NEAR_INT8_CODE = 120 MIN_H_CHANGE_DB = -1.0 BASELINE_24_16_ABS_H = 2.1859e-3 GAIN_LADDER = ((24, 24), (32, 24), (32, 32), (40, 32)) CLKIN_STATUS_DETECTED = "DETECTED" CLKIN_STATUS_NOT_DETECTED = "NOT_DETECTED" CLKIN_STATUS_UNKNOWN = "UNKNOWN" CLOCK_WAIT_TIMEOUT = "TIMEOUT" CLOCK_WAIT_ERROR = "ERROR" # Первый измерительный цикл 2026-08-31: NOT_DETECTED до 115.3 мс, # DETECTED на 171.0 мс после T0. Предел 500 мс почти втрое больше # измеренного времени, но остаётся коротким и жёстко ограниченным. CLOCK_INITIAL_SETTLE_SECONDS = 0.050 CLOCK_POLL_INTERVAL_SECONDS = 0.050 CLOCK_TIMEOUT_SECONDS = 0.500 CLOCK_INTER_CYCLE_SECONDS = 0.200 CLOCK_SETTLE_CYCLE_COUNT = 5 # ================================================================== форма сигнала def reference_bits() -> tuple[np.ndarray, np.ndarray, np.ndarray]: """Вернуть биты пилота, payload и их объединение. Пилот — детерминированный PRBS7 проекта. Payload — фиксированная последовательность NumPy PCG64 с зафиксированным семенем. Обе части восстанавливаются из примитивов, а не из сохранённого IQ. """ pilot = pilot_prbs7(PILOT_SYMBOL_COUNT, PILOT_INITIAL_STATE) payload = np.random.Generator(np.random.PCG64(PAYLOAD_SEED)).integers( 0, 2, PAYLOAD_SYMBOL_COUNT, dtype=np.uint8 ) return pilot, payload, np.concatenate((pilot, payload)) def build_tx_complex(all_bits: np.ndarray | None = None) -> np.ndarray: """Собрать комплексную форму BPSK до перевода в int8. Символы формируются существующим формирователем Lab042 (RRC и защитные интервалы), затем переносятся на цифровую несущую +50 кГц. Возвращаются отсчёты с целевым среднеквадратичным уровнем TARGET_RMS_CODE. """ if all_bits is None: _, _, all_bits = reference_bits() symbols = radio.bpsk_modulate(all_bits) shaped = shape_frame_like_lab042(symbols, SAMPLES_PER_SYMBOL) indexes = np.arange(len(shaped)) carriered = shaped * np.exp(1j * 2.0 * np.pi * CARRIER_OFFSET_HZ * indexes / SAMPLE_RATE_HZ) rms = float(np.sqrt(np.mean(np.abs(carriered) ** 2))) if rms <= 0.0: raise ValueError("Нулевая мощность формы") return carriered * (TARGET_RMS_CODE / rms) def to_cs8_bytes(waveform: np.ndarray) -> bytes: """Перевести комплексную форму в interleaved signed int8 I/Q.""" real = np.clip(np.round(waveform.real), -128, INT8_LIMIT).astype(np.int8) imag = np.clip(np.round(waveform.imag), -128, INT8_LIMIT).astype(np.int8) interleaved = np.empty(2 * len(real), dtype=np.int8) interleaved[0::2] = real interleaved[1::2] = imag return interleaved.tobytes() def build_tx_cs8() -> bytes: """Собрать cs8-форму и проверить её SHA-256 против эталона.""" payload = to_cs8_bytes(build_tx_complex()) digest = hashlib.sha256(payload).hexdigest() if digest != EXPECTED_TX_SHA256: raise ValueError( "SHA-256 восстановленной формы не совпал с эталоном. " f"Получено {digest}, ожидалось {EXPECTED_TX_SHA256}. Аппаратный режим запрещён." ) return payload # ================================================================== чтение raw def load_cs8(path: Path) -> np.ndarray: """Прочитать сохранённый interleaved signed int8 I/Q в complex64. Файл читается один раз; повторных загрузок 64 МБ не выполняется. """ raw = Path(path).read_bytes() return signed_iq_bytes_to_complex64(raw) def raw_statistics(samples: np.ndarray) -> dict: """Базовая статистика сырого потока: уровень, перегрузка, проценты.""" codes = samples * 128.0 # обратно в шкалу кодов int8 magnitude = np.abs(codes) centered = codes - np.mean(codes) component_peak = np.maximum(np.abs(codes.real), np.abs(codes.imag)) return { "sample_count": int(len(samples)), "mean_i_code": float(np.mean(codes.real)), "mean_q_code": float(np.mean(codes.imag)), "rms_code": float(np.sqrt(np.mean(magnitude**2))), "rms_after_dc_removal_code": float(np.sqrt(np.mean(np.abs(centered) ** 2))), "i_min_code": float(np.min(codes.real)), "i_max_code": float(np.max(codes.real)), "q_min_code": float(np.min(codes.imag)), "q_max_code": float(np.max(codes.imag)), "amplitude_p95_code": float(np.percentile(magnitude, 95.0)), "amplitude_p99_code": float(np.percentile(magnitude, 99.0)), "amplitude_p999_code": float(np.percentile(magnitude, 99.9)), "peak_component_code": float(np.max(component_peak)), "peak_complex_code": float(np.max(magnitude)), "clipping_count": int(np.sum(component_peak >= INT8_LIMIT)), "clipping_fraction": float(np.mean(component_peak >= INT8_LIMIT)), "near_int8_count": int(np.sum(component_peak >= NEAR_INT8_CODE)), "near_int8_fraction": float(np.mean(component_peak >= NEAR_INT8_CODE)), "adc_headroom_code": float(INT8_LIMIT - np.max(component_peak)), } # ================================================================== анализ BPSK def _db(ratio: float) -> float: return 10.0 * math.log10(max(float(ratio), 1e-300)) def _amplitude_db(ratio: float) -> float: """Перевести отношение амплитуд корреляции в dB через 20 log10.""" return 20.0 * math.log10(max(float(ratio), 1e-300)) def _tx_cs8_complex() -> np.ndarray: """Вернуть точную проверенную TX-форму в той же шкале, что и RX.""" return signed_iq_bytes_to_complex64(build_tx_cs8()) def _normalized_sliding_correlation(signal: np.ndarray, reference: np.ndarray) -> np.ndarray: """Нормированная комплексная корреляция для всех допустимых стартов.""" correlation = fftconvolve(signal, np.conj(reference[::-1]), mode="valid") cumulative_energy = np.concatenate(( np.zeros(1, dtype=np.float64), np.cumsum(np.abs(signal) ** 2, dtype=np.float64), )) window_energy = cumulative_energy[len(reference) :] - cumulative_energy[: -len(reference)] reference_energy = float(np.sum(np.abs(reference) ** 2)) denominator = np.sqrt(np.maximum(window_energy * reference_energy, 1e-30)) return np.abs(correlation) / denominator def _detection_streams(samples: np.ndarray, tx_samples: np.ndarray) -> tuple[np.ndarray, np.ndarray]: """DC removal, перенос -50 кГц и decimation как в исходном анализаторе.""" rx_indexes = np.arange(len(samples)) centered = samples - np.mean(samples) rx_baseband = centered * np.exp( -2j * np.pi * CARRIER_OFFSET_HZ * rx_indexes / SAMPLE_RATE_HZ ) rx_decimated = resample_poly( rx_baseband, 1, DETECTION_DECIMATION, window=("kaiser", 8) ).astype(np.complex64) tx_indexes = np.arange(len(tx_samples)) tx_baseband = tx_samples * np.exp( -2j * np.pi * CARRIER_OFFSET_HZ * tx_indexes / SAMPLE_RATE_HZ ) tx_decimated = resample_poly( tx_baseband, 1, DETECTION_DECIMATION, window=("kaiser", 8) ).astype(np.complex64) tx_decimated -= np.mean(tx_decimated) return rx_decimated, tx_decimated def _peak_metrics(scores: np.ndarray, guard_samples: int) -> dict: peak_index = int(np.argmax(scores)) peak = float(scores[peak_index]) median = float(np.median(scores)) mask = np.ones(len(scores), dtype=bool) low = max(0, peak_index - guard_samples) high = min(len(mask), peak_index + guard_samples + 1) mask[low:high] = False remaining_indexes = np.flatnonzero(mask) if len(remaining_indexes): second_relative = int(np.argmax(scores[mask])) second_index = int(remaining_indexes[second_relative]) second = float(scores[second_index]) else: second_index = -1 second = 0.0 return { "peak_index": peak_index, "peak": peak, "median": median, "second_index": second_index, "second": second, "over_median_db": _amplitude_db(peak / median) if median > 0 else float("inf"), "over_second_db": _amplitude_db(peak / second) if second > 0 else float("inf"), } def detect_preamble(rx_decimated: np.ndarray, tx_decimated: np.ndarray) -> dict: """Глобально найти кадр только по известной преамбуле, без payload. Reference повторяет фактический анализ 2026-08-27: ведущий guard, 1280 символов PRBS7 и десять символов хвоста RRC. Полная 17664-битная форма здесь принципиально не используется. """ samples_per_symbol = SAMPLES_PER_SYMBOL // DETECTION_DECIMATION reference_symbol_count = 16 + PILOT_SYMBOL_COUNT + 10 reference_length = reference_symbol_count * samples_per_symbol scores = _normalized_sliding_correlation(rx_decimated, tx_decimated[:reference_length]) guard = SECOND_PEAK_GUARD_SYMBOLS * samples_per_symbol metrics = _peak_metrics(scores, guard) detected = bool( metrics["over_median_db"] >= PREAMBLE_OVER_MEDIAN_DB_MIN and metrics["over_second_db"] >= PREAMBLE_OVER_SECOND_DB_MIN ) sample_rate_decimated = SAMPLE_RATE_HZ / DETECTION_DECIMATION return { "preamble_detected": detected, "method": "global preamble-only normalized correlation", "payload_used_for_start": False, "peak_index_decimated": metrics["peak_index"], "start_sample": metrics["peak_index"] * DETECTION_DECIMATION, "start_seconds": metrics["peak_index"] / sample_rate_decimated, "peak": metrics["peak"], "median": metrics["median"], "second_index_decimated": metrics["second_index"], "second_start_seconds": ( metrics["second_index"] / sample_rate_decimated if metrics["second_index"] >= 0 else None ), "over_median_db": metrics["over_median_db"], "over_second_db": metrics["over_second_db"], "threshold_over_median_db": PREAMBLE_OVER_MEDIAN_DB_MIN, "threshold_over_second_db": PREAMBLE_OVER_SECOND_DB_MIN, "second_peak_guard_symbols": SECOND_PEAK_GUARD_SYMBOLS, } def _refine_frame_start(samples: np.ndarray, tx_samples: np.ndarray, coarse_start: int) -> int: """Уточнить raw-start локальной корреляцией только с prefix-преамбулой.""" reference_length = (16 + PILOT_SYMBOL_COUNT + 10) * SAMPLES_PER_SYMBOL low = max(0, coarse_start - 2_000) high = min(len(samples), coarse_start + reference_length + 2_000) local = samples[low:high] - np.mean(samples) reference = tx_samples[:reference_length] - np.mean(tx_samples[:reference_length]) correlation = fftconvolve(local, np.conj(reference[::-1]), mode="valid") return low + int(np.argmax(np.abs(correlation))) def _matched_waveform(samples: np.ndarray, start: int, tx_length: int) -> tuple[np.ndarray, int]: segment = samples[start : start + tx_length] - np.mean(samples) indexes = start + np.arange(len(segment)) baseband = segment * np.exp( -2j * np.pi * CARRIER_OFFSET_HZ * indexes / SAMPLE_RATE_HZ ) taps = radio.root_raised_cosine_taps(0.35, SAMPLES_PER_SYMBOL, 10) matched = fftconvolve(baseband, taps, mode="full") nominal = 16 * SAMPLES_PER_SYMBOL + 2 * ((len(taps) - 1) // 2) return matched, nominal def _timing_offsets(matched: np.ndarray, nominal: int, expected_symbols: np.ndarray) -> tuple[int, int]: offsets = np.arange(-800, 801) preamble_scores: list[float] = [] full_scores: list[float] = [] for offset in offsets: symbols = matched[ nominal + offset : nominal + offset + len(expected_symbols) * SAMPLES_PER_SYMBOL : SAMPLES_PER_SYMBOL ] if len(symbols) != len(expected_symbols): preamble_scores.append(0.0) full_scores.append(0.0) continue preamble = symbols[:PILOT_SYMBOL_COUNT] preamble_scores.append( abs(np.vdot(expected_symbols[:PILOT_SYMBOL_COUNT], preamble)) / math.sqrt(PILOT_SYMBOL_COUNT * np.sum(np.abs(preamble) ** 2) + 1e-30) ) full_scores.append( abs(np.vdot(expected_symbols, symbols)) / math.sqrt(len(expected_symbols) * np.sum(np.abs(symbols) ** 2) + 1e-30) ) preamble_offset = int(offsets[int(np.argmax(preamble_scores))]) oracle_offset = int(offsets[int(np.argmax(full_scores))]) return preamble_offset, oracle_offset def _estimate_cfo_gain( symbols: np.ndarray, expected_symbols: np.ndarray, estimation_count: int, ) -> tuple[float, complex]: """Оценить CFO и комплексный gain по заданному prefix символов.""" count = min(len(symbols), len(expected_symbols), estimation_count) if count <= 0: return 0.0, 0j if estimation_count == PILOT_SYMBOL_COUNT: # Non-data-aided BPSK carrier recovery: squaring removes unknown # payload signs because (+1)^2 == (-1)^2. The phase slope is doubled. squared = symbols**2 block_length = 512 block_count = len(squared) // block_length if block_count >= 2: block_values = squared[: block_count * block_length].reshape( block_count, block_length ).mean(1) block_time = ( np.arange(block_count) * block_length + (block_length - 1) / 2 + 0.5 ) / SYMBOL_RATE phase = np.unwrap(np.angle(block_values)) weights = np.abs(block_values) ** 2 matrix = np.column_stack((np.ones(block_count), block_time - block_time[0])) root_weights = np.sqrt(weights / max(float(np.max(weights)), 1e-30)) slope = np.linalg.lstsq( matrix * root_weights[:, None], phase * root_weights, rcond=None )[0][1] cfo_hz = float(slope / (4 * np.pi)) else: cfo_hz = 0.0 preamble_time = (np.arange(count) + 0.5) / SYMBOL_RATE corrected_data = symbols[:count] * expected_symbols[:count] gain = complex(np.mean( corrected_data * np.exp(-2j * np.pi * cfo_hz * preamble_time) )) return cfo_hz, gain corrected_data = symbols[:count] * expected_symbols[:count] time = (np.arange(count) + 0.5) / SYMBOL_RATE frequencies = np.fft.fftfreq(count, 1 / SYMBOL_RATE) spectrum = np.abs(np.fft.fft(corrected_data * np.hanning(count))) ** 2 candidates = np.flatnonzero(np.abs(frequencies) <= 100.0) peak_index = int(candidates[int(np.argmax(spectrum[candidates]))]) neighbours = np.array([(peak_index - 1) % count, peak_index, (peak_index + 1) % count]) logarithmic_power = np.log(np.maximum(spectrum[neighbours], 1e-300)) denominator = logarithmic_power[0] - 2 * logarithmic_power[1] + logarithmic_power[2] interpolation = ( 0.0 if abs(denominator) < 1e-18 else 0.5 * (logarithmic_power[0] - logarithmic_power[2]) / denominator ) coarse_cfo = float(frequencies[peak_index] + interpolation * SYMBOL_RATE / count) block_length = 512 block_count = count // block_length derotated = corrected_data * np.exp(-2j * np.pi * coarse_cfo * time) if block_count >= 2: block_values = derotated[: block_count * block_length].reshape(block_count, block_length).mean(1) block_time = time[: block_count * block_length].reshape(block_count, block_length).mean(1) phase = np.unwrap(np.angle(block_values)) weights = np.abs(block_values) ** 2 matrix = np.column_stack((np.ones(block_count), block_time - block_time[0])) root_weights = np.sqrt(weights / max(float(np.max(weights)), 1e-30)) slope = np.linalg.lstsq( matrix * root_weights[:, None], phase * root_weights, rcond=None )[0][1] cfo_hz = float(coarse_cfo + slope / (2 * np.pi)) else: cfo_hz = coarse_cfo gain = complex(np.mean(corrected_data * np.exp(-2j * np.pi * cfo_hz * time))) return cfo_hz, gain def _evaluate_symbols( matched: np.ndarray, nominal: int, timing_offset: int, bits: np.ndarray, estimation_count: int, ) -> dict: expected = radio.bpsk_modulate(bits).astype(np.complex128) symbols = matched[ nominal + timing_offset : nominal + timing_offset + len(bits) * SAMPLES_PER_SYMBOL : SAMPLES_PER_SYMBOL ].astype(np.complex128) count = min(len(symbols), len(expected)) symbols = symbols[:count] expected = expected[:count] bits = bits[:count] cfo_hz, gain = _estimate_cfo_gain(symbols, expected, estimation_count) time = (np.arange(count) + 0.5) / SYMBOL_RATE derotation = np.exp(-2j * np.pi * cfo_hz * time) if gain == 0 or count == 0: return { "timing_offset_samples": timing_offset, "cfo_hz": cfo_hz, "abs_h": 0.0, "phase_rad": 0.0, "phase_deg": 0.0, "errors": count, "ber": 1.0 if count else float("nan"), "preamble_errors": min(count, PILOT_SYMBOL_COUNT), "symbols_compared": count, "snr_db": float("-inf"), "centers": [], "distance": 0.0, "within_rms": 0.0, "separation": 0.0, } normalized = symbols * derotation / gain decisions = normalized.real >= 0.0 errors = int(np.count_nonzero(decisions != bits)) preamble_count = min(PILOT_SYMBOL_COUNT, count) preamble_errors = int(np.count_nonzero(decisions[:preamble_count] != bits[:preamble_count])) residual = symbols - gain * expected * np.exp(2j * np.pi * cfo_hz * time) snr_db = float(_db(abs(gain) ** 2 / np.mean(np.abs(residual) ** 2))) centers: list[complex] = [] squared_residual = 0.0 for bit in (0, 1): group = normalized[bits == bit] center = complex(np.mean(group)) centers.append(center) squared_residual += float(np.sum(np.abs(group - center) ** 2)) within_rms = math.sqrt(squared_residual / count) distance = float(abs(centers[1] - centers[0])) return { "timing_offset_samples": int(timing_offset), "cfo_hz": cfo_hz, "abs_h": float(abs(gain)), "phase_rad": float(np.angle(gain)), "phase_deg": float(np.degrees(np.angle(gain))), "polarity": "resolved from preamble" if estimation_count == PILOT_SYMBOL_COUNT else "oracle full frame", "errors": errors, "ber": errors / count, "preamble_errors": preamble_errors, "symbols_compared": count, "snr_db": snr_db, "centers": [{"i": center.real, "q": center.imag} for center in centers], "distance": distance, "within_rms": within_rms, "separation": distance / within_rms, } def analyze_capture(samples: np.ndarray, include_oracle: bool = True) -> dict: """Offline-анализ с независимой preamble-only синхронизацией. Полная известная форма используется только в явно помеченной oracle диагностике и никогда не влияет на официальный preamble PASS. """ _, _, all_bits = reference_bits() expected_symbols = radio.bpsk_modulate(all_bits) tx_samples = _tx_cs8_complex() raw = raw_statistics(samples) rx_decimated, tx_decimated = _detection_streams(samples, tx_samples) detection = detect_preamble(rx_decimated, tx_decimated) independent_start = _refine_frame_start(samples, tx_samples, detection["start_sample"]) independent_matched, nominal = _matched_waveform(samples, independent_start, len(tx_samples)) preamble_timing, _ = _timing_offsets(independent_matched, nominal, expected_symbols) formal = _evaluate_symbols( independent_matched, nominal, preamble_timing, all_bits, PILOT_SYMBOL_COUNT, ) # После независимого start можно оценить |h| по всей известной форме как # измерительную характеристику. Эта оценка не участвует в синхронизации. full_measurement = _evaluate_symbols( independent_matched, nominal, preamble_timing, all_bits, TOTAL_SYMBOL_COUNT, ) oracle: dict | None = None oracle_detection: dict | None = None if include_oracle: full_scores = _normalized_sliding_correlation(rx_decimated, tx_decimated) full_guard = int(0.15 * SAMPLE_RATE_HZ / DETECTION_DECIMATION) oracle_detection = _peak_metrics(full_scores, full_guard) oracle_coarse = oracle_detection["peak_index"] * DETECTION_DECIMATION oracle_start = _refine_frame_start(samples, tx_samples, oracle_coarse) oracle_matched, oracle_nominal = _matched_waveform(samples, oracle_start, len(tx_samples)) _, oracle_timing = _timing_offsets(oracle_matched, oracle_nominal, expected_symbols) oracle = _evaluate_symbols( oracle_matched, oracle_nominal, oracle_timing, all_bits, TOTAL_SYMBOL_COUNT, ) oracle["start_sample"] = oracle_start oracle["data_leakage"] = "full 17664-bit waveform used for start/timing" preamble_detected = bool(detection["preamble_detected"]) selected_h = full_measurement["abs_h"] if preamble_detected else (oracle or formal)["abs_h"] selected_h_status = "INDEPENDENT_SYNC" if preamble_detected else "ORACLE_DIAGNOSTIC" return { "raw": raw, "timing": { "detected_start_sample": int(detection["start_sample"]), "refined_start_sample": int(independent_start), "start_seconds": independent_start / SAMPLE_RATE_HZ, "nominal_symbol_sample": int(nominal), "symbol_timing_phase_samples": int(preamble_timing), }, "detection": detection, "synchronization": { "independent": preamble_detected, "payload_used_for_start_or_timing": False, "estimated_carrier_phase_rad": formal["phase_rad"], "estimated_carrier_phase_deg": formal["phase_deg"], "cfo_hz": formal["cfo_hz"], "polarity": formal["polarity"], "preamble_errors": formal["preamble_errors"], "preamble_symbols_compared": min(PILOT_SYMBOL_COUNT, formal["symbols_compared"]), }, "formal": formal if preamble_detected else None, "formal_candidate": formal, "full_frame_measurement_after_independent_sync": ( full_measurement if preamble_detected else None ), "oracle_diagnostic": oracle, "oracle_detection": oracle_detection, "channel": { "abs_h": selected_h, "status": selected_h_status, "delta_definition": "20*log10(abs_h_new/abs_h_previous)", }, "official_ber": formal["ber"] if preamble_detected else None, "official_ser": formal["ber"] if preamble_detected else None, "official_ber_note": ( "по независимой preamble-only синхронизации" if preamble_detected else "НЕ ВЫЧИСЛЕНО: только oracle diagnostic" ), "diagnostic_ber": (oracle or formal)["ber"], "symbol_snr_db": ( full_measurement["snr_db"] if preamble_detected else (oracle or formal)["snr_db"] ), "separation": ( full_measurement["separation"] if preamble_detected else (oracle or formal)["separation"] ), "clipping_count": raw["clipping_count"], } def h_ladder_table(points: list[tuple[int, int, Path]]) -> list[dict]: """Свести |h| и производные показатели по нескольким сохранённым точкам.""" rows: list[dict] = [] previous_abs_h: float | None = None for lna, vga, path in points: analysis = analyze_capture(load_cs8(path)) abs_h = analysis["channel"]["abs_h"] delta_db = None if previous_abs_h in (None, 0.0) else 20.0 * math.log10(abs_h / previous_abs_h) rows.append({ "lna_db": lna, "vga_db": vga, "abs_h": abs_h, "delta_abs_h_db": delta_db, "sync_status": analysis["channel"]["status"], "preamble_detected": analysis["detection"]["preamble_detected"], "symbol_snr_db": analysis["symbol_snr_db"], "separation": analysis["separation"], "official_ber": analysis["official_ber"], "diagnostic_ber": analysis["diagnostic_ber"], "clipping_count": analysis["clipping_count"], }) previous_abs_h = abs_h return rows # ================================================================== аппаратный режим @dataclass class TransmitGuard: """Гарантия ровно одной передачи за аппаратный опыт.""" transmit_calls: int = 0 tx_allowed: bool = False def allow(self) -> None: self.tx_allowed = True def register_transmit(self) -> None: if not self.tx_allowed: raise RuntimeError("Передача запрещена: общий clock-gate не пройден") if self.transmit_calls >= 1: raise RuntimeError("Повторная передача запрещена: разрешён ровно один TX") self.transmit_calls += 1 CommandRunner = Callable[[list[str]], subprocess.CompletedProcess] def _default_command_runner(command: list[str]) -> subprocess.CompletedProcess: # pragma: no cover - железо return subprocess.run(command, capture_output=True, text=True, timeout=30) @dataclass class RunningCommand: """Фоновый процесс с принадлежащими ему файлами stdout/stderr.""" process: subprocess.Popen stdout_handle: TextIO stderr_handle: TextIO def poll(self) -> int | None: return self.process.poll() def wait(self, timeout: float) -> int: try: return self.process.wait(timeout=timeout) finally: self.close_logs() def terminate(self) -> None: if self.process.poll() is None: self.process.terminate() try: self.process.wait(timeout=5) except subprocess.TimeoutExpired: self.process.kill() self.process.wait(timeout=5) self.close_logs() def close_logs(self) -> None: if not self.stdout_handle.closed: self.stdout_handle.close() if not self.stderr_handle.closed: self.stderr_handle.close() ProcessStarter = Callable[[list[str], Path, Path], RunningCommand] SleepFunction = Callable[[float], None] MonotonicFunction = Callable[[], float] def _default_process_starter( command: list[str], stdout_path: Path, stderr_path: Path ) -> RunningCommand: # pragma: no cover - железо stdout_handle = stdout_path.open("w", encoding="utf-8") stderr_handle = stderr_path.open("w", encoding="utf-8") try: process = subprocess.Popen( command, stdout=stdout_handle, stderr=stderr_handle, text=True, ) except Exception: stdout_handle.close() stderr_handle.close() raise return RunningCommand(process, stdout_handle, stderr_handle) def parse_clkin_status(output: str) -> str: """Строго разобрать единственную допустимую строку состояния CLKIN. Проверка намеренно fail-closed: положительное и отрицательное состояния распознаются только по полному совпадению. Пустой, многострочный или любой неожиданный вывод возвращает UNKNOWN и не может открыть clock-gate. """ normalized_output = output.strip() if normalized_output == "CLKIN status: clock signal detected": return CLKIN_STATUS_DETECTED if normalized_output == "CLKIN status: no clock signal detected": return CLKIN_STATUS_NOT_DETECTED return CLKIN_STATUS_UNKNOWN def wait_for_slave_clock( runner: CommandRunner, *, initial_settle_s: float = CLOCK_INITIAL_SETTLE_SECONDS, poll_interval_s: float = CLOCK_POLL_INTERVAL_SECONDS, timeout_s: float = CLOCK_TIMEOUT_SECONDS, sleep_function: SleepFunction = time.sleep, monotonic_function: MonotonicFunction = time.monotonic, started_at: float | None = None, ) -> dict: """Ограниченно ждать точного DETECTED от SLAVE, сохраняя каждый poll. UNKNOWN, ненулевой exit code и исключение subprocess немедленно закрывают gate. NOT_DETECTED разрешает следующий запрос только пока не исчерпан монотонный timeout. """ if initial_settle_s < 0 or poll_interval_s <= 0 or timeout_s <= 0: raise ValueError("Параметры ожидания clock должны быть положительными") if initial_settle_s > timeout_s: raise ValueError("initial_settle_s не может превышать timeout_s") start = monotonic_function() if started_at is None else started_at settle_remaining = initial_settle_s - (monotonic_function() - start) if settle_remaining > 0: sleep_function(settle_remaining) clock_tool = str(HACKRF_TOOLS_DIR / "hackrf_clock.exe") polls: list[dict] = [] while True: query_started_s = monotonic_function() - start if query_started_s > timeout_s: return { "state": CLOCK_WAIT_TIMEOUT, "queries": len(polls), "time_to_detect_ms": None, "elapsed_ms": 1000.0 * query_started_s, "polls": polls, } command = [clock_tool, "-d", SLAVE_SERIAL, "-i"] try: result = runner(command) except Exception as error: elapsed_s = monotonic_function() - start polls.append({ "query": len(polls) + 1, "start_ms": 1000.0 * query_started_s, "end_ms": 1000.0 * elapsed_s, "raw": f"{type(error).__name__}: {error}", "state": CLOCK_WAIT_ERROR, "exit_code": None, }) return { "state": CLOCK_WAIT_ERROR, "queries": len(polls), "time_to_detect_ms": None, "elapsed_ms": 1000.0 * elapsed_s, "polls": polls, } raw_output = "\n".join( part.strip() for part in ((result.stdout or ""), (result.stderr or "")) if part.strip() ) parsed_state = ( parse_clkin_status(raw_output) if result.returncode == 0 else CLOCK_WAIT_ERROR ) elapsed_s = monotonic_function() - start polls.append({ "query": len(polls) + 1, "start_ms": 1000.0 * query_started_s, "end_ms": 1000.0 * elapsed_s, "raw": raw_output, "state": parsed_state, "exit_code": result.returncode, }) if parsed_state == CLKIN_STATUS_DETECTED: return { "state": CLKIN_STATUS_DETECTED, "queries": len(polls), "time_to_detect_ms": 1000.0 * elapsed_s, "elapsed_ms": 1000.0 * elapsed_s, "polls": polls, } if parsed_state in {CLKIN_STATUS_UNKNOWN, CLOCK_WAIT_ERROR}: return { "state": parsed_state, "queries": len(polls), "time_to_detect_ms": None, "elapsed_ms": 1000.0 * elapsed_s, "polls": polls, } remaining_s = timeout_s - (monotonic_function() - start) if remaining_s <= 1e-9: return { "state": CLOCK_WAIT_TIMEOUT, "queries": len(polls), "time_to_detect_ms": None, "elapsed_ms": 1000.0 * (monotonic_function() - start), "polls": polls, } sleep_function(min(poll_interval_s, remaining_s)) def _read_si5351_reg29(serial: str, runner: CommandRunner) -> int: """Прочитать Si5351 register 29 указанной платы (read-only).""" tool = str(HACKRF_TOOLS_DIR / "hackrf_debug.exe") result = runner([tool, "-d", serial, "--si5351c", "-n", "29", "-r"]) if result.returncode != 0: raise RuntimeError(f"hackrf_debug reg29 не выполнен для {serial}: {result.stderr.strip()}") text = (result.stdout or "") + (result.stderr or "") for token in text.replace(":", " ").replace("=", " ").split(): try: return int(token, 0) except ValueError: continue raise RuntimeError(f"reg29 не разобран для {serial}: {text.strip()}") def _set_clkout(serial: str, enabled: bool, runner: CommandRunner) -> None: tool = str(HACKRF_TOOLS_DIR / "hackrf_clock.exe") result = runner([tool, "-d", serial, "-o", "1" if enabled else "0"]) if result.returncode != 0: output = ((result.stdout or "") + (result.stderr or "")).strip() raise RuntimeError(f"CLKOUT {serial} не установлен в {enabled}: {output}") def _short_rx(serial: str, runner: CommandRunner) -> None: """Один конечный RX-start для штатного transceiver_startup, без TX.""" tool = str(HACKRF_TOOLS_DIR / "hackrf_transfer.exe") command = [ tool, "-d", serial, "-r", "NUL", "-f", str(RF_HZ), "-s", str(SAMPLE_RATE_HZ), "-n", str(SHORT_RX_SAMPLE_COUNT), "-l", "0", "-g", "0", "-a", "0", "-p", "0", "-B", ] result = runner(command) if result.returncode != 0: output = ((result.stdout or "") + (result.stderr or "")).strip() raise RuntimeError(f"Короткий RX {serial} не выполнен: {output}") def _disable_clock_outputs(runner: CommandRunner) -> list[str]: """Best-effort cleanup обоих CLKOUT; вернуть описания ошибок.""" errors: list[str] = [] for serial in (MASTER_SERIAL, SLAVE_SERIAL): try: _set_clkout(serial, False, runner) except RuntimeError as error: errors.append(str(error)) return errors def common_clock_preflight( runner: CommandRunner, guard: TransmitGuard, *, sleep_function: SleepFunction = time.sleep, monotonic_function: MonotonicFunction = time.monotonic, ) -> dict: """Fail-closed проверка общей опоры до разрешения передачи. Любая ошибка оставляет guard.tx_allowed = False, поэтому передача остаётся запрещённой. TX разрешается только при полном PASS. """ guard.tx_allowed = False try: _set_clkout(MASTER_SERIAL, False, runner) _set_clkout(SLAVE_SERIAL, False, runner) _short_rx(MASTER_SERIAL, runner) master_reg29 = _read_si5351_reg29(MASTER_SERIAL, runner) if master_reg29 != REG29_EXTERNAL_CLOCK: raise RuntimeError(f"MASTER reg29={master_reg29:#04x} != {REG29_EXTERNAL_CLOCK:#04x}") clock_wait_started = monotonic_function() _set_clkout(MASTER_SERIAL, True, runner) clock_wait = wait_for_slave_clock( runner, sleep_function=sleep_function, monotonic_function=monotonic_function, started_at=clock_wait_started, ) clkin_status = clock_wait["state"] if clkin_status != CLKIN_STATUS_DETECTED: last_output = clock_wait["polls"][-1]["raw"] if clock_wait["polls"] else "<нет запросов>" raise RuntimeError(f"SLAVE CLKIN={clkin_status}: {last_output}") _short_rx(SLAVE_SERIAL, runner) slave_reg29 = _read_si5351_reg29(SLAVE_SERIAL, runner) if slave_reg29 != REG29_EXTERNAL_CLOCK: raise RuntimeError(f"SLAVE reg29={slave_reg29:#04x} != {REG29_EXTERNAL_CLOCK:#04x}") except Exception: _disable_clock_outputs(runner) raise guard.allow() return { "passed": True, "master_reg29": master_reg29, "slave_reg29": slave_reg29, "slave_clkin_status": clkin_status, "slave_clock_wait": clock_wait, "master_serial": MASTER_SERIAL, "slave_serial": SLAVE_SERIAL, } def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: while block := stream.read(1024 * 1024): digest.update(block) return digest.hexdigest() def _transfer_log_is_clean(text: str) -> bool: """Отклонить явный ненулевой overrun/underrun и неясную ошибку USB.""" lowered = text.lower() for match in re.finditer(r"(\d+)\s+(?:rx\s+)?overruns?|(\d+)\s+(?:tx\s+)?underruns?", lowered): values = [value for value in match.groups() if value is not None] if any(int(value) != 0 for value in values): return False suspicious = ("overrun", "underrun") for word in suspicious: if word in lowered and not re.search(rf"0\s+(?:rx\s+|tx\s+)?{word}s?", lowered): return False return True def _capture_intervals(samples: np.ndarray, frame_start: int, frame_length: int) -> dict: """Статистика независимых областей до, во время и после кадра.""" frame_start = max(0, min(int(frame_start), len(samples))) frame_end = max(frame_start, min(frame_start + frame_length, len(samples))) def statistics_or_none(region: np.ndarray) -> dict | None: return raw_statistics(region) if len(region) else None return { "before": statistics_or_none(samples[:frame_start]), "during": statistics_or_none(samples[frame_start:frame_end]), "after": statistics_or_none(samples[frame_end:]), "frame_start_sample": frame_start, "frame_end_sample": frame_end, } def _point_passes(analysis: dict) -> bool: return bool( analysis["detection"]["preamble_detected"] and analysis["official_ber"] is not None and analysis["official_ber"] <= PASS_BER_MAX and analysis["symbol_snr_db"] >= PASS_SNR_DB_MIN and analysis["separation"] >= PASS_SEPARATION_MIN and analysis["raw"]["clipping_count"] == 0 and analysis["raw"]["near_int8_count"] == 0 ) def _point_stop_reason(analysis: dict, delta_abs_h_db: float | None) -> str | None: raw = analysis["raw"] if raw["clipping_count"]: return "ADC_CLIPPING" if raw["near_int8_count"]: return "NEAR_SIGNED_INT8_LIMIT" if delta_abs_h_db is not None and delta_abs_h_db < MIN_H_CHANGE_DB: return "ABS_H_ANOMALOUS_DECLINE" return None def run_one_gain_point( lna_gain_db: int, vga_gain_db: int, output_directory: Path, runner: CommandRunner | None = None, process_starter: ProcessStarter | None = None, sleep_function: SleepFunction = time.sleep, previous_abs_h: float | None = None, ) -> dict: """Один аппаратный опыт: preflight, RX до TX, ровно один TX, анализ. Порядок fail-closed: сначала общий clock-gate, только затем передача. При любой ошибке после передачи второй TX не выполняется. """ validate_hackrf_rx_gains(lna_gain_db, vga_gain_db) runner = runner or _default_command_runner process_starter = process_starter or _default_process_starter guard = TransmitGuard() tx_payload = build_tx_cs8() # SHA-проверка формы до любого обращения к железу waveform_sha256 = hashlib.sha256(tx_payload).hexdigest() rx_process: RunningCommand | None = None try: preflight = common_clock_preflight(runner, guard) output_directory.mkdir(parents=True, exist_ok=True) tx_path = output_directory / "tx_bpsk_cs8.iq" tx_path.write_bytes(tx_payload) rx_path = output_directory / "rx_cs8.iq" rx_stdout_path = output_directory / "rx_stdout.txt" rx_stderr_path = output_directory / "rx_stderr.txt" tx_stdout_path = output_directory / "tx_stdout.txt" tx_stderr_path = output_directory / "tx_stderr.txt" transfer_tool = str(HACKRF_TOOLS_DIR / "hackrf_transfer.exe") rx_command = [ transfer_tool, "-d", SLAVE_SERIAL, "-r", str(rx_path), "-f", str(RF_HZ), "-s", str(SAMPLE_RATE_HZ), "-n", str(CAPTURE_SAMPLE_COUNT), "-l", str(lna_gain_db), "-g", str(vga_gain_db), "-a", "0", "-p", "0", "-B", ] rx_process = process_starter(rx_command, rx_stdout_path, rx_stderr_path) sleep_function(RX_LEAD_SECONDS) if rx_process.poll() is not None: rx_process.close_logs() details = rx_stderr_path.read_text(encoding="utf-8", errors="replace") raise RuntimeError(f"RX завершился до разрешения TX: {details.strip()}") tx_command = [ transfer_tool, "-d", MASTER_SERIAL, "-t", str(tx_path), "-f", str(RF_HZ), "-s", str(SAMPLE_RATE_HZ), "-n", str(len(tx_payload) // 2), "-x", str(TX_VGA_DB), "-a", "0", "-p", "0", "-B", ] guard.register_transmit() tx_result = runner(tx_command) tx_stdout_path.write_text(tx_result.stdout or "", encoding="utf-8") tx_stderr_path.write_text(tx_result.stderr or "", encoding="utf-8") if tx_result.returncode != 0: raise RuntimeError(f"Единственный TX завершился с кодом {tx_result.returncode}") rx_returncode = rx_process.wait(RX_WAIT_TIMEOUT_SECONDS) if rx_returncode != 0: raise RuntimeError(f"RX завершился с кодом {rx_returncode}") expected_size = CAPTURE_SAMPLE_COUNT * 2 size_bytes = rx_path.stat().st_size if size_bytes != expected_size: raise RuntimeError(f"Размер raw {size_bytes}, ожидалось {expected_size}") raw_sha256_first = _sha256_file(rx_path) raw_sha256_second = _sha256_file(rx_path) if raw_sha256_first != raw_sha256_second: raise RuntimeError("SHA-256 raw изменился после закрытия файла") rx_log = rx_stdout_path.read_text(encoding="utf-8", errors="replace") + rx_stderr_path.read_text( encoding="utf-8", errors="replace" ) tx_log = (tx_result.stdout or "") + (tx_result.stderr or "") if not _transfer_log_is_clean(rx_log): raise RuntimeError("RX log содержит overrun/underrun") if not _transfer_log_is_clean(tx_log): raise RuntimeError("TX log содержит overrun/underrun") samples = load_cs8(rx_path) analysis = analyze_capture(samples) frame_start = analysis["timing"]["refined_start_sample"] analysis["capture_intervals"] = _capture_intervals( samples, frame_start, len(tx_payload) // 2 ) abs_h = analysis["channel"]["abs_h"] delta_abs_h_db = ( None if previous_abs_h in (None, 0.0) else 20.0 * math.log10(abs_h / previous_abs_h) ) passed = _point_passes(analysis) stop_reason = "BPSK_PASS" if passed else _point_stop_reason(analysis, delta_abs_h_db) point_result = { "lna_db": lna_gain_db, "vga_db": vga_gain_db, "preflight": preflight, "waveform_sha256": waveform_sha256, "raw": { "path": str(rx_path), "size_bytes": size_bytes, "sample_count": CAPTURE_SAMPLE_COUNT, "duration_seconds": CAPTURE_SAMPLE_COUNT / SAMPLE_RATE_HZ, "sha256_first": raw_sha256_first, "sha256_second": raw_sha256_second, }, "tx_calls": guard.transmit_calls, "delta_abs_h_db": delta_abs_h_db, "passed": passed, "stop_reason": stop_reason, "analysis": analysis, } metadata = { key: value for key, value in point_result.items() if key != "analysis" } (output_directory / "metadata.json").write_text( json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8" ) (output_directory / "analysis.json").write_text( json.dumps(analysis, ensure_ascii=False, indent=2), encoding="utf-8" ) return point_result finally: if rx_process is not None: try: rx_process.terminate() except Exception: pass _disable_clock_outputs(runner) def default_ladder_output_directory() -> Path: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") return ( Path.home() / "Documents" / "SDR_Rover_Lab043_raw" / f"{timestamp}_common_clock_bpsk_gain_ladder_2" ) def default_settle_output_directory() -> Path: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") return ( Path.home() / "Documents" / "SDR_Rover_Lab043_raw" / f"{timestamp}_common_clock_settle_test" ) def run_common_clock_settle_test( output_directory: Path, cycle_count: int = CLOCK_SETTLE_CYCLE_COUNT, runner: CommandRunner | None = None, sleep_function: SleepFunction = time.sleep, monotonic_function: MonotonicFunction = time.monotonic, ) -> dict: """Выполнить bounded common-clock-only циклы и сохранить доказательства.""" if cycle_count <= 0: raise ValueError("cycle_count должен быть положительным") runner = runner or _default_command_runner cycles: list[dict] = [] for cycle_number in range(1, cycle_count + 1): cycle = { "cycle": cycle_number, "t0_utc": None, "master_reg29": None, "slave_reg29": None, "slave_clkin": None, "time_to_detect_ms": None, "number_of_queries": 0, "polls": [], "passed": False, "error": None, "cleanup_errors": [], } try: _set_clkout(MASTER_SERIAL, False, runner) _set_clkout(SLAVE_SERIAL, False, runner) _short_rx(MASTER_SERIAL, runner) cycle["master_reg29"] = _read_si5351_reg29(MASTER_SERIAL, runner) if cycle["master_reg29"] != REG29_EXTERNAL_CLOCK: raise RuntimeError( f"MASTER reg29={cycle['master_reg29']:#04x} != {REG29_EXTERNAL_CLOCK:#04x}" ) cycle["t0_utc"] = datetime.now(timezone.utc).isoformat() clock_wait_started = monotonic_function() _set_clkout(MASTER_SERIAL, True, runner) clock_wait = wait_for_slave_clock( runner, sleep_function=sleep_function, monotonic_function=monotonic_function, started_at=clock_wait_started, ) cycle["slave_clkin"] = clock_wait["state"] cycle["time_to_detect_ms"] = clock_wait["time_to_detect_ms"] cycle["number_of_queries"] = clock_wait["queries"] cycle["polls"] = clock_wait["polls"] if clock_wait["state"] != CLKIN_STATUS_DETECTED: raise RuntimeError(f"SLAVE CLKIN={clock_wait['state']}") _short_rx(SLAVE_SERIAL, runner) cycle["slave_reg29"] = _read_si5351_reg29(SLAVE_SERIAL, runner) if cycle["slave_reg29"] != REG29_EXTERNAL_CLOCK: raise RuntimeError( f"SLAVE reg29={cycle['slave_reg29']:#04x} != {REG29_EXTERNAL_CLOCK:#04x}" ) cycle["passed"] = True except Exception as error: cycle["error"] = f"{type(error).__name__}: {error}" finally: cycle["cleanup_errors"] = _disable_clock_outputs(runner) cycles.append(cycle) if not cycle["passed"]: break if cycle_number < cycle_count: sleep_function(CLOCK_INTER_CYCLE_SECONDS) report = { "test": "Lab043 common clock bounded settle", "created_utc": datetime.now(timezone.utc).isoformat(), "parameters": { "initial_settle_s": CLOCK_INITIAL_SETTLE_SECONDS, "poll_interval_s": CLOCK_POLL_INTERVAL_SECONDS, "timeout_s": CLOCK_TIMEOUT_SECONDS, "inter_cycle_s": CLOCK_INTER_CYCLE_SECONDS, "requested_cycles": cycle_count, }, "completed_cycles": len(cycles), "passed_cycles": sum(int(cycle["passed"]) for cycle in cycles), "stable": len(cycles) == cycle_count and all(cycle["passed"] for cycle in cycles), "rf_tx_count": 0, "cycles": cycles, } output_directory.mkdir(parents=True, exist_ok=True) (output_directory / "settle_report.json").write_text( json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" ) lines = [ "Lab043 common-clock settle test", f"parameters: initial={CLOCK_INITIAL_SETTLE_SECONDS:.3f}s, " f"interval={CLOCK_POLL_INTERVAL_SECONDS:.3f}s, timeout={CLOCK_TIMEOUT_SECONDS:.3f}s", f"cycles: {report['passed_cycles']}/{cycle_count} PASS", f"common-clock gate: {'STABLE' if report['stable'] else 'UNSTABLE'}", "RF TX count: 0", "", ] for cycle in cycles: lines.append( f"cycle {cycle['cycle']}: {'PASS' if cycle['passed'] else 'FAIL'}, " f"detect_ms={cycle['time_to_detect_ms']}, queries={cycle['number_of_queries']}, " f"MASTER reg29={cycle['master_reg29']}, SLAVE CLKIN={cycle['slave_clkin']}, " f"SLAVE reg29={cycle['slave_reg29']}, error={cycle['error']}" ) (output_directory / "settle_report.txt").write_text("\n".join(lines) + "\n", encoding="utf-8") return report def run_gain_ladder(output_directory: Path) -> dict: # pragma: no cover - железо """Последовательно выполнить только нужные точки и остановиться fail-closed.""" rows: list[dict] = [] previous_abs_h: float | None = BASELINE_24_16_ABS_H classification = "B" stop_reason = "LADDER_EXHAUSTED_WITHOUT_PASS" for lna_gain_db, vga_gain_db in GAIN_LADDER: point_directory = output_directory / f"lna{lna_gain_db:02d}_vga{vga_gain_db:02d}" try: result = run_one_gain_point( lna_gain_db, vga_gain_db, point_directory, previous_abs_h=previous_abs_h, ) except Exception as error: classification = "D" stop_reason = f"{type(error).__name__}: {error}" break rows.append(result) previous_abs_h = result["analysis"]["channel"]["abs_h"] if result["passed"]: classification = "A" stop_reason = "BPSK_PASS" break if result["stop_reason"] is not None: classification = "C" stop_reason = result["stop_reason"] break report = { "classification": classification, "baseline": { "lna_db": 24, "vga_db": 16, "abs_h": BASELINE_24_16_ABS_H, }, "stop_reason": stop_reason, "recommended_rx_gain": ( {"lna_db": rows[-1]["lna_db"], "vga_db": rows[-1]["vga_db"]} if rows and rows[-1]["passed"] else None ), "tx_count": sum(row["tx_calls"] for row in rows), "points": rows, "output_directory": str(output_directory), } output_directory.mkdir(parents=True, exist_ok=True) (output_directory / "ladder_report.json").write_text( json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" ) return report # ================================================================== CLI def _analyze_existing(path: Path) -> int: analysis = analyze_capture(load_cs8(path)) print(json.dumps(analysis, ensure_ascii=False, indent=2, default=str)) return 0 def _validate_old_ladder(root: Path) -> int: """Отдельная offline-regression внешних raw, не входящих в Git.""" points = [ (8, 8, root / "lna08_vga08" / "rx_cs8.iq"), (16, 8, root / "lna16_vga08" / "rx_cs8.iq"), (16, 16, root / "lna16_vga16" / "rx_cs8.iq"), (24, 16, root / "lna24_vga16" / "rx_cs8.iq"), ] missing = [str(path) for _, _, path in points if not path.is_file()] if missing: raise FileNotFoundError("Отсутствуют regression raw: " + ", ".join(missing)) report = { "root": str(root), "waveform_sha256": EXPECTED_TX_SHA256, "rows": h_ladder_table(points), } print(json.dumps(report, ensure_ascii=False, indent=2, default=str)) return 0 def main() -> int: parser = argparse.ArgumentParser(description="Lab043 BPSK HackRF -> HackRF") parser.add_argument("--analyze-existing", type=Path, help="offline-анализ сохранённого cs8-захвата") parser.add_argument( "--validate-old-ladder", type=Path, help="offline-regression четырёх внешних raw первой gain ladder", ) parser.add_argument("--hardware", action="store_true", help="аппаратный опыт (по умолчанию заблокирован)") parser.add_argument( "--hardware-ladder", action="store_true", help="последовательная gain ladder с анализом и STOP после каждой точки", ) parser.add_argument( "--common-clock-settle-test", action="store_true", help="пять bounded common-clock-only циклов без RF TX", ) parser.add_argument("--lna", type=int, default=24) parser.add_argument("--vga", type=int, default=16) parser.add_argument("--i-understand-hardware", action="store_true", help="явное подтверждение передачи") parser.add_argument("--output-directory", type=Path) arguments = parser.parse_args() if arguments.analyze_existing is not None: return _analyze_existing(arguments.analyze_existing) if arguments.validate_old_ladder is not None: return _validate_old_ladder(arguments.validate_old_ladder) if arguments.common_clock_settle_test: output_directory = arguments.output_directory or default_settle_output_directory() report = run_common_clock_settle_test(output_directory) print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if report["stable"] else 1 if arguments.hardware or arguments.hardware_ladder: if not arguments.i_understand_hardware: print("Аппаратный режим заблокирован. Нужен явный --i-understand-hardware.") return 2 output_directory = arguments.output_directory or default_ladder_output_directory() if arguments.hardware_ladder: report = run_gain_ladder(output_directory) print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if report["classification"] in {"A", "B", "C"} else 1 result = run_one_gain_point(arguments.lna, arguments.vga, output_directory) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 parser.print_help() return 0 if __name__ == "__main__": raise SystemExit(main())