403 lines
17 KiB
Python
403 lines
17 KiB
Python
"""Безопасный однотоновый поиск рабочего 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())
|