"""Безопасный однотоновый поиск рабочего RX-усиления HackRF для Lab043. Разрешён один непрерывный циклический тон Pluto+ и заранее фиксированная лестница усилений HackRF. RF amp и antenna bias всегда выключены. Здесь нет BPSK, PRBS, пакетов, счётчика или JPEG. """ from __future__ import annotations import argparse import json import math import threading import time from datetime import datetime, timezone from pathlib import Path import numpy as np from experiments import lab043_hackrf_continuity as continuity from experiments import lab043_hackrf_hardware as hackrf from experiments import lab043_pluto_to_rtlsdr as lab043 from experiments import lab043_prbs11_hardware as pluto_hardware GAIN_LADDER = ((0, 0), (8, 8), (16, 8), (16, 16), (24, 16), (24, 24)) SETTLE_SECONDS = 0.25 MEASUREMENT_SECONDS = 0.75 FIRST_CAPTURE_SECONDS = SETTLE_SECONDS + MEASUREMENT_SECONDS FFT_SAMPLES = 262_144 SEARCH_LOW_HZ = 35_000.0 SEARCH_HIGH_HZ = 65_000.0 NOISE_EXCLUSION_HZ = 2_000.0 DESIRABLE_TONE_OVER_NOISE_DB = 35.0 NEAR_LIMIT_COMPONENT = 0.90 PREFERRED_MAX_COMPONENT = 0.75 CLIPPED_COMPONENT = 127.0 / 128.0 NEAR_LIMIT_FRACTION_STOP = 1e-4 NONLINEAR_HARMONIC_LIMIT_DBC = -30.0 HARMONIC_DETECTION_MARGIN_DB = 10.0 HARMONIC_SEARCH_HALF_WIDTH_HZ = 2_000.0 HARMONIC_NOISE_OUTER_HALF_WIDTH_HZ = 10_000.0 OUTPUT_DIRECTORY = Path("data/processed/lab043_hackrf_gain_ladder") def _db(power: float) -> float: return 10.0 * math.log10(max(float(power), 1e-300)) def alias_frequency_to_nyquist(frequency_hz: float, sample_rate_hz: float) -> float: """Свернуть частоту в полуоткрытую полосу [-Fs/2, Fs/2).""" if not np.isfinite(sample_rate_hz) or sample_rate_hz <= 0.0: raise ValueError("Частота дискретизации должна быть положительной") return float( (float(frequency_hz) + sample_rate_hz / 2.0) % sample_rate_hz - sample_rate_hz / 2.0 ) def _cyclic_frequency_distance( frequencies_hz: np.ndarray, target_hz: float, sample_rate_hz: float, ) -> np.ndarray: return np.abs( (frequencies_hz - target_hz + sample_rate_hz / 2.0) % sample_rate_hz - sample_rate_hz / 2.0 ) def analyze_harmonic( powers: np.ndarray, frequencies: np.ndarray, fundamental_frequency_hz: float, fundamental_power: float, order: int, sample_rate_hz: float, ) -> dict: """Оценить гармонику отдельно относительно тона и локального фона.""" if order not in (2, 3): raise ValueError("Поддерживаются только вторая и третья гармоники") expected_hz = alias_frequency_to_nyquist( order * fundamental_frequency_hz, sample_rate_hz, ) distances = _cyclic_frequency_distance(frequencies, expected_hz, sample_rate_hz) peak_mask = distances <= HARMONIC_SEARCH_HALF_WIDTH_HZ noise_mask = ( (distances > HARMONIC_SEARCH_HALF_WIDTH_HZ) & (distances <= HARMONIC_NOISE_OUTER_HALF_WIDTH_HZ) ) if not np.any(peak_mask) or not np.any(noise_mask): raise ValueError("Недостаточно спектральных отсчётов около гармоники") local_indexes = np.flatnonzero(peak_mask) peak_index = int(local_indexes[np.argmax(powers[local_indexes])]) peak_power = float(powers[peak_index]) local_noise_power = float(np.median(powers[noise_mask])) above_local_noise_db = _db(peak_power) - _db(local_noise_power) relative_to_fundamental_db = _db(peak_power) - _db(fundamental_power) credible = bool(above_local_noise_db >= HARMONIC_DETECTION_MARGIN_DB) nonlinear = bool( credible and relative_to_fundamental_db >= NONLINEAR_HARMONIC_LIMIT_DBC ) return { "order": order, "expected_frequency_hz": expected_hz, "peak_frequency_hz": float(frequencies[peak_index]), "peak_power_dbfs": _db(peak_power), "local_noise_floor_dbfs_per_fft_bin": _db(local_noise_power), "above_local_noise_db": float(above_local_noise_db), "relative_to_fundamental_db": float(relative_to_fundamental_db), "credible": credible, "nonlinear": nonlinear, } def measure_tone_segment( samples: np.ndarray, sample_rate_hz: float, fft_samples: int = FFT_SAMPLES, ) -> dict: """Измерить тон и перегрузку по неизменяемой спектральной методике.""" received = np.asarray(samples, dtype=np.complex64) if received.ndim != 1 or len(received) < fft_samples: raise ValueError("Участок короче одного фиксированного БПФ-окна") if not np.all(np.isfinite(received)): raise ValueError("Участок содержит нечисловые отсчёты") segment_count = len(received) // fft_samples window = np.hanning(fft_samples).astype(np.float64) coherent_scale = float(np.sum(window)) powers = np.zeros(fft_samples, dtype=np.float64) for index in range(segment_count): segment = received[index * fft_samples : (index + 1) * fft_samples] spectrum = np.fft.fft(segment * window) powers += np.abs(spectrum / coherent_scale) ** 2 powers /= segment_count frequencies = np.fft.fftfreq(fft_samples, d=1.0 / sample_rate_hz) search = (frequencies >= SEARCH_LOW_HZ) & (frequencies <= SEARCH_HIGH_HZ) candidates = np.flatnonzero(search) peak_index = int(candidates[np.argmax(powers[candidates])]) peak_bin_hz = float(frequencies[peak_index]) resolution_hz = float(sample_rate_hz / fft_samples) left = (peak_index - 1) % fft_samples right = (peak_index + 1) % fft_samples y0, y1, y2 = np.log(np.maximum(powers[[left, peak_index, right]], 1e-300)) denominator = y0 - 2.0 * y1 + y2 bin_offset = 0.0 if abs(denominator) < 1e-18 else 0.5 * (y0 - y2) / denominator tone_frequency_hz = peak_bin_hz + float(np.clip(bin_offset, -0.5, 0.5)) * resolution_hz noise_mask = search & (np.abs(frequencies - tone_frequency_hz) >= NOISE_EXCLUSION_HZ) noise_power = float(np.median(powers[noise_mask])) tone_power = float(powers[peak_index]) tone_over_noise_db = _db(tone_power) - _db(noise_power) second = analyze_harmonic( powers, frequencies, tone_frequency_hz, tone_power, 2, sample_rate_hz, ) third = analyze_harmonic( powers, frequencies, tone_frequency_hz, tone_power, 3, sample_rate_hz, ) component_peak = np.maximum(np.abs(received.real), np.abs(received.imag)) near_fraction = float(np.mean(component_peak >= NEAR_LIMIT_COMPONENT)) clipped_fraction = float(np.mean(component_peak >= CLIPPED_COMPONENT)) max_component = float(np.max(component_peak)) harmonic_nonlinearity = bool(second["nonlinear"] or third["nonlinear"]) stop = bool( clipped_fraction > 0.0 or near_fraction >= NEAR_LIMIT_FRACTION_STOP or max_component >= NEAR_LIMIT_COMPONENT or harmonic_nonlinearity ) return { "sample_count": int(len(received)), "fft_samples": int(fft_samples), "fft_segment_count": int(segment_count), "fft_resolution_hz": resolution_hz, "tone_frequency_hz": float(tone_frequency_hz), "tone_peak_bin_hz": peak_bin_hz, "tone_power_dbfs": _db(tone_power), "noise_floor_dbfs_per_fft_bin": _db(noise_power), "tone_over_noise_db": float(tone_over_noise_db), "mean_amplitude": float(np.mean(np.abs(received))), "rms_amplitude": float(np.sqrt(np.mean(np.abs(received) ** 2))), "maximum_amplitude": float(np.max(np.abs(received))), "maximum_component": max_component, "near_limit_component_fraction": near_fraction, "clipped_component_fraction": clipped_fraction, "dc_mean_i": float(np.mean(received.real)), "dc_mean_q": float(np.mean(received.imag)), "second_harmonic": second, "third_harmonic": third, "second_harmonic_dbc": second["relative_to_fundamental_db"], "third_harmonic_dbc": third["relative_to_fundamental_db"], "harmonic_nonlinearity_detected": harmonic_nonlinearity, "stop_gain_escalation": stop, "stop_reason": ( "перегрузка/близость к пределу/гармоническая нелинейность" if stop else "" ), } def choose_preferred_gain(rows: list[dict]) -> dict | None: """Выбрать минимальное безопасное усиление без погони за лишними децибелами.""" safe = [ row for row in rows if not row["metrics"]["stop_gain_escalation"] and row["metrics"]["maximum_component"] <= PREFERRED_MAX_COMPONENT ] if not safe: return None desirable = [ row for row in safe if row["metrics"]["tone_over_noise_db"] >= DESIRABLE_TONE_OVER_NOISE_DB ] if desirable: candidates = desirable else: best_margin = max(row["metrics"]["tone_over_noise_db"] for row in safe) candidates = [ row for row in safe if row["metrics"]["tone_over_noise_db"] >= best_margin - 2.0 ] return min( candidates, key=lambda row: (row["lna_gain_db"] + row["vga_gain_db"], row["step_index"]), ) def _capture(receiver: hackrf.HackrfRxDevice, seconds: float) -> hackrf.HackrfCapture: count = math.ceil(seconds * hackrf.HACKRF_SAMPLE_RATE_HZ) return receiver.capture(count) def run_gain_ladder(output_directory: Path = OUTPUT_DIRECTORY) -> dict: """Выполнить один аппаратный сеанс с одним непрерывным тоном.""" for lna, vga in GAIN_LADDER: hackrf.validate_hackrf_rx_gains(lna, vga) pluto_preflight = pluto_hardware.preflight_pluto() hackrf_preflight = hackrf.preflight_hackrf() pluto, tx_parameters = pluto_hardware.configure_pluto() actual_tx_rate = float(tx_parameters["actual_sample_rate_hz"]) waveform, actual_tone_hz = continuity.build_periodic_tone(actual_tx_rate) device_waveform = lab043.to_pluto_tx_samples(waveform) rows: list[dict] = [] tx_started = False stop_reason = "лестница завершена" started = datetime.now(timezone.utc) try: with hackrf.HackrfRxDevice() as receiver: rx_parameters = receiver.configure_rx() first_holder: list[hackrf.HackrfCapture] = [] first_errors: list[str] = [] def first_receive() -> None: try: first_holder.append(_capture(receiver, FIRST_CAPTURE_SECONDS)) except Exception as error: # pragma: no cover - аппаратный путь first_errors.append(f"{type(error).__name__}: {error}") first_gain = receiver.set_rx_gains(*GAIN_LADDER[0]) rx_thread = threading.Thread(target=first_receive, daemon=True) rx_thread.start() deadline = time.perf_counter() + 5.0 while time.perf_counter() < deadline: if receiver.library.hackrf_is_streaming(receiver.device) == 1: break if first_errors: break time.sleep(0.01) else: raise RuntimeError("HackRF RX не подтвердил поток; TX не включён") time.sleep(SETTLE_SECONDS) pluto.tx_cyclic_buffer = True pluto.tx_destroy_buffer() pluto.tx(device_waveform) tx_started = True rx_thread.join(timeout=hackrf.HACKRF_STREAM_TIMEOUT_SECONDS + 5.0) if rx_thread.is_alive(): raise RuntimeError("Первый короткий RX-сеанс не завершился") if first_errors: raise RuntimeError("; ".join(first_errors)) if len(first_holder) != 1: raise RuntimeError("Первый RX-сеанс не вернул единственный массив") first_capture = first_holder[0] measurement_count = math.floor(MEASUREMENT_SECONDS * hackrf.HACKRF_SAMPLE_RATE_HZ) metrics = measure_tone_segment(first_capture.samples[-measurement_count:], hackrf.HACKRF_SAMPLE_RATE_HZ) rows.append({ "step_index": 0, "lna_gain_db": GAIN_LADDER[0][0], "vga_gain_db": GAIN_LADDER[0][1], "gain_configuration": first_gain, "capture_diagnostics": first_capture.diagnostics, "metrics": metrics, }) if metrics["stop_gain_escalation"]: stop_reason = metrics["stop_reason"] else: for step_index, (lna_gain, vga_gain) in enumerate(GAIN_LADDER[1:], start=1): gain_configuration = receiver.set_rx_gains(lna_gain, vga_gain) time.sleep(SETTLE_SECONDS) capture = _capture(receiver, MEASUREMENT_SECONDS) metrics = measure_tone_segment(capture.samples, hackrf.HACKRF_SAMPLE_RATE_HZ) rows.append({ "step_index": step_index, "lna_gain_db": lna_gain, "vga_gain_db": vga_gain, "gain_configuration": gain_configuration, "capture_diagnostics": capture.diagnostics, "metrics": metrics, }) if metrics["stop_gain_escalation"]: stop_reason = metrics["stop_reason"] break finally: if tx_started: pluto.tx_destroy_buffer() preferred = choose_preferred_gain(rows) preferred_summary = None if preferred is None else { "step_index": preferred["step_index"], "lna_gain_db": preferred["lna_gain_db"], "vga_gain_db": preferred["vga_gain_db"], "tone_over_noise_db": preferred["metrics"]["tone_over_noise_db"], "desirable_margin_reached": bool( preferred["metrics"]["tone_over_noise_db"] >= DESIRABLE_TONE_OVER_NOISE_DB ), } result = { "experiment": "Lab043 HackRF RX gain ladder with one continuous tone", "started_utc": started.isoformat().replace("+00:00", "Z"), "physical_scheme": "Pluto+ TX -> SMA -> AT30S 30 dB -> HackRF RX; antennas removed", "method": "one continuous cyclic Pluto tone; separate short HackRF RX sessions per gain point", "fixed_gain_ladder": [list(point) for point in GAIN_LADDER], "fixed_thresholds": { "desirable_tone_over_noise_db": DESIRABLE_TONE_OVER_NOISE_DB, "near_limit_component": NEAR_LIMIT_COMPONENT, "preferred_max_component": PREFERRED_MAX_COMPONENT, "clipped_component": CLIPPED_COMPONENT, "near_limit_fraction_stop": NEAR_LIMIT_FRACTION_STOP, "nonlinear_harmonic_limit_dbc": NONLINEAR_HARMONIC_LIMIT_DBC, "harmonic_detection_margin_db": HARMONIC_DETECTION_MARGIN_DB, "harmonic_search_half_width_hz": HARMONIC_SEARCH_HALF_WIDTH_HZ, "harmonic_noise_outer_half_width_hz": HARMONIC_NOISE_OUTER_HALF_WIDTH_HZ, }, "pluto_preflight": pluto_preflight, "hackrf_preflight": hackrf_preflight, "tx_parameters": tx_parameters, "rx_parameters": rx_parameters, "actual_transmitted_tone_offset_hz": actual_tone_hz, "rf_amp_enabled": False, "antenna_bias_enabled": False, "completed_step_count": len(rows), "stop_reason": stop_reason, "rows": rows, "preferred_gain": preferred_summary, "digital_experiment_authorized": False, } output_directory.mkdir(parents=True, exist_ok=True) stamp = started.strftime("%Y%m%d_%H%M%S") output_path = output_directory / f"lab043_hackrf_gain_ladder_{stamp}.json" output_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") result["result_path"] = str(output_path) return result def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--output-directory", type=Path, default=OUTPUT_DIRECTORY) arguments = parser.parse_args() result = run_gain_ladder(arguments.output_directory) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())