1541 lines
62 KiB
Python
1541 lines
62 KiB
Python
"""Lab043: воспроизводимый BPSK-опыт HackRF → HackRF при общей опоре.
|
||
|
||
Инструмент превращает уже выполненный вручную аппаратный опыт в
|
||
повторяемую процедуру внутри проекта. Он НЕ вводит новую архитектуру:
|
||
форма сигнала и примитивы взяты из уже проверенного кода проекта.
|
||
|
||
BPSK (Binary Phase Shift Keying) — двоичная фазовая манипуляция, два
|
||
состояния фазы, один бит на символ.
|
||
|
||
Два режима:
|
||
|
||
--analyze-existing <raw.iq> offline-анализ сохранённого захвата,
|
||
оборудование не требуется;
|
||
|
||
--hardware --lna L --vga V один аппаратный опыт (передача + приём);
|
||
--hardware-ladder последовательная разрешённая gain ladder.
|
||
|
||
Оба аппаратных режима по умолчанию заблокированы и требуют явного
|
||
--i-understand-hardware.
|
||
|
||
Восстановление формы (см. build_tx_cs8) проверяется по SHA-256; при
|
||
несовпадении аппаратный режим останавливается до передачи.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
from functools import lru_cache
|
||
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, oaconvolve, 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)))
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def _matched_filter_taps() -> np.ndarray:
|
||
"""Вернуть неизменяемые RRC-коэффициенты доказанной рабочей точки."""
|
||
|
||
taps = radio.root_raised_cosine_taps(
|
||
0.35,
|
||
SAMPLES_PER_SYMBOL,
|
||
10,
|
||
).astype(np.float32)
|
||
taps.flags.writeable = False
|
||
return taps
|
||
|
||
|
||
@lru_cache(maxsize=4)
|
||
def _carrier_downconversion_vector(start: int, length: int) -> np.ndarray:
|
||
"""Один раз вычислить неизменяемый перенос несущей для размера окна."""
|
||
|
||
indexes = start + np.arange(length)
|
||
rotation = np.exp(
|
||
-2j * np.pi * CARRIER_OFFSET_HZ * indexes / SAMPLE_RATE_HZ
|
||
).astype(np.complex64)
|
||
rotation.flags.writeable = False
|
||
return rotation
|
||
|
||
|
||
def _matched_waveform(samples: np.ndarray, start: int, tx_length: int) -> tuple[np.ndarray, int]:
|
||
segment = samples[start : start + tx_length] - np.mean(samples)
|
||
baseband = segment * _carrier_downconversion_vector(start, len(segment))
|
||
taps = _matched_filter_taps()
|
||
matched = oaconvolve(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 _preamble_timing_offset(
|
||
matched: np.ndarray,
|
||
nominal: int,
|
||
expected_symbols: np.ndarray,
|
||
) -> int:
|
||
"""Найти то же преамбульное смещение без второй одинаковой оценки."""
|
||
|
||
expected = np.asarray(expected_symbols)
|
||
offsets = np.arange(-800, 801)
|
||
scores = np.zeros(offsets.size, dtype=np.float64)
|
||
first_sample = nominal + int(offsets[0])
|
||
for phase in range(SAMPLES_PER_SYMBOL):
|
||
score_indexes = np.arange(phase, offsets.size, SAMPLES_PER_SYMBOL)
|
||
sequence_length = expected.size + score_indexes.size - 1
|
||
start = first_sample + phase
|
||
stop = start + sequence_length * SAMPLES_PER_SYMBOL
|
||
sequence = matched[start:stop:SAMPLES_PER_SYMBOL]
|
||
if start < 0 or sequence.size != sequence_length:
|
||
return _timing_offsets(matched, nominal, expected)[0]
|
||
windows = np.lib.stride_tricks.sliding_window_view(sequence, expected.size)
|
||
correlation = windows @ np.conj(expected)
|
||
energy = np.sum(np.abs(windows) ** 2, axis=1)
|
||
scores[score_indexes] = np.abs(correlation) / np.sqrt(
|
||
expected.size * energy + 1e-30
|
||
)
|
||
return int(offsets[int(np.argmax(scores))])
|
||
|
||
|
||
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())
|