1373 lines
45 KiB
Python
1373 lines
45 KiB
Python
"""Проверенные примитивы BPSK-радиотракта, вынесенные из лабораторных.
|
||
|
||
Модуль не проводит экспериментов и ничего не пишет на диск: его можно
|
||
безопасно импортировать. Это и было причиной его появления — Lab018,
|
||
Lab019 и Lab023 при импорте создают каталоги и сохраняют файлы, поэтому
|
||
переиспользовать их функции напрямую невозможно.
|
||
|
||
Тела функций перенесены дословно из соответствующих лабораторных, чтобы
|
||
результаты остались сопоставимыми. Источник указан над каждой функцией.
|
||
Совпадение с оригиналами проверяется в tests/test_bpsk_radio.py.
|
||
|
||
Универсальные функции формирования и разбора радиокадра добавлены здесь
|
||
заново: в лабораторных они были жёстко привязаны к демонстрационному
|
||
текстовому сообщению и не принимали произвольные данные.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import struct
|
||
from dataclasses import dataclass
|
||
|
||
import numpy as np
|
||
|
||
# Параметры радиокадра. Значения совпадают с Lab018, Lab019 и Lab023.
|
||
RADIO_SYNC_WORD = 0xD391
|
||
PREAMBLE_BIT_COUNT = 64
|
||
RADIO_HEADER_BIT_COUNT = 32
|
||
MARKER_BIT_COUNT = PREAMBLE_BIT_COUNT + 16
|
||
|
||
# Параметры оценки частотного рассогласования. Значения из Lab023.
|
||
SYMBOL_RATE = 20_000
|
||
CFO_REFINEMENT_HALF_WIDTH_HZ = 200.0
|
||
CFO_REFINEMENT_STEP_HZ = 1.0
|
||
|
||
# Пороги разрешения частотной коррекции. Значения из Lab023. Смысл в том,
|
||
# что при малом истинном уходе оценка по короткому маркеру состоит почти
|
||
# целиком из шума, и коррекция таким значением портит длинный кадр сильнее,
|
||
# чем отсутствие коррекции вообще.
|
||
CFO_DEAD_ZONE_HZ = 15.0
|
||
MINIMUM_PHASE_CONSISTENCY = 0.55
|
||
|
||
# Критерии рабочей оценки малой остаточной CFO по отдельному известному
|
||
# пилоту Lab043. Они зафиксированы до нового аппаратного опыта.
|
||
KNOWN_PILOT_BLOCK_SYMBOL_COUNT = 20
|
||
MINIMUM_KNOWN_PILOT_BLOCK_COUNT = 8
|
||
MINIMUM_KNOWN_PILOT_COHERENCE = 0.50
|
||
MAXIMUM_KNOWN_PILOT_PHASE_FIT_RMSE_RAD = 0.25
|
||
|
||
MAXIMUM_PROTOCOL_PACKET_BYTES = 4096
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TwoTonePhaseRefinement:
|
||
"""Явный результат фазового уточнения двух известных тонов."""
|
||
|
||
valid: bool
|
||
invalid_reason: str
|
||
low_frequency_hz: float
|
||
high_frequency_hz: float
|
||
carrier_offset_hz: float
|
||
tone_spacing_hz: float
|
||
residual_cfo_hz: float
|
||
phase_fit_rmse_rad: float
|
||
low_phase_fit_rms_rad: float
|
||
high_phase_fit_rms_rad: float
|
||
paired_phase_fit_rms_rad: float
|
||
block_count: int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class KnownPilotCarrierEstimate:
|
||
"""Оценка малой остаточной CFO по отдельному известному BPSK-пилоту."""
|
||
|
||
valid: bool
|
||
invalid_reason: str
|
||
frequency_hz: float
|
||
phase_increment_rad_per_symbol: float
|
||
initial_phase_rad: float
|
||
phase_fit_rmse_rad: float
|
||
mean_block_coherence: float
|
||
block_count: int
|
||
known_symbol_count: int
|
||
|
||
# Перенесено дословно из Lab023 (tests/lab023_guarded_cfo_correction.py).
|
||
def bytes_to_bits(
|
||
data: bytes,
|
||
) -> np.ndarray:
|
||
"""
|
||
Преобразовать bytes в одномерный массив битов.
|
||
"""
|
||
|
||
if not isinstance(data, bytes):
|
||
raise TypeError(
|
||
"data должен иметь тип bytes"
|
||
)
|
||
|
||
return np.unpackbits(
|
||
np.frombuffer(
|
||
data,
|
||
dtype=np.uint8,
|
||
)
|
||
)
|
||
|
||
|
||
# Перенесено дословно из Lab023 (tests/lab023_guarded_cfo_correction.py).
|
||
def bits_to_bytes(
|
||
bits: np.ndarray,
|
||
) -> bytes:
|
||
"""
|
||
Упаковать массив битов обратно в bytes.
|
||
"""
|
||
|
||
bits = np.asarray(
|
||
bits,
|
||
dtype=np.uint8,
|
||
)
|
||
|
||
if bits.ndim != 1:
|
||
raise ValueError(
|
||
"bits должен быть одномерным массивом"
|
||
)
|
||
|
||
if len(bits) % 8 != 0:
|
||
raise ValueError(
|
||
"Количество битов должно быть кратно восьми"
|
||
)
|
||
|
||
if not np.all(
|
||
(bits == 0) | (bits == 1)
|
||
):
|
||
raise ValueError(
|
||
"bits должен содержать только 0 и 1"
|
||
)
|
||
|
||
return np.packbits(
|
||
bits
|
||
).tobytes()
|
||
|
||
|
||
# Перенесено дословно из Lab023 (tests/lab023_guarded_cfo_correction.py).
|
||
def bpsk_modulate(
|
||
bits: np.ndarray,
|
||
) -> np.ndarray:
|
||
"""
|
||
Преобразовать биты в BPSK-символы.
|
||
|
||
0 → -1
|
||
1 → +1
|
||
"""
|
||
|
||
bits = np.asarray(
|
||
bits,
|
||
dtype=np.uint8,
|
||
)
|
||
|
||
symbols = (
|
||
2.0
|
||
* bits.astype(np.float64)
|
||
- 1.0
|
||
)
|
||
|
||
return symbols.astype(
|
||
np.complex128
|
||
)
|
||
|
||
|
||
# Перенесено дословно из Lab023 (tests/lab023_guarded_cfo_correction.py).
|
||
def bpsk_demodulate(
|
||
symbols: np.ndarray,
|
||
) -> np.ndarray:
|
||
"""
|
||
Демодулировать BPSK по знаку компоненты I.
|
||
"""
|
||
|
||
return (
|
||
symbols.real >= 0.0
|
||
).astype(np.uint8)
|
||
|
||
|
||
def find_known_tone_peaks(
|
||
frequency_axis_hz: np.ndarray,
|
||
power_spectrum: np.ndarray,
|
||
expected_offset_hz: float,
|
||
search_half_width_hz: float,
|
||
) -> dict:
|
||
"""Найти максимумы мощности около двух известных симметричных тонов.
|
||
|
||
Функция не задаёт способ оценки шумового фона и порог достоверности:
|
||
эти решения принадлежат конкретному эксперименту. Если в одном из
|
||
поисковых окон нет конечных значений мощности, соответствующие частота
|
||
и мощность возвращаются как ``NaN``.
|
||
"""
|
||
|
||
frequencies = np.asarray(frequency_axis_hz, dtype=np.float64)
|
||
powers = np.asarray(power_spectrum, dtype=np.float64)
|
||
|
||
if frequencies.ndim != 1 or powers.ndim != 1:
|
||
raise ValueError("Ось частот и спектр мощности должны быть одномерными")
|
||
if len(frequencies) != len(powers):
|
||
raise ValueError("Ось частот и спектр мощности должны иметь одинаковую длину")
|
||
if not np.isfinite(expected_offset_hz) or expected_offset_hz <= 0.0:
|
||
raise ValueError("Ожидаемый отступ тона должен быть положительным")
|
||
if not np.isfinite(search_half_width_hz) or search_half_width_hz <= 0.0:
|
||
raise ValueError("Полуширина окна поиска должна быть положительной")
|
||
|
||
def peak_in_window(center_hz: float) -> tuple[float, float]:
|
||
in_window = (
|
||
(frequencies >= center_hz - search_half_width_hz)
|
||
& (frequencies <= center_hz + search_half_width_hz)
|
||
& np.isfinite(frequencies)
|
||
& np.isfinite(powers)
|
||
& (powers >= 0.0)
|
||
)
|
||
indexes = np.flatnonzero(in_window)
|
||
if indexes.size == 0:
|
||
return float("nan"), float("nan")
|
||
peak_index = int(indexes[np.argmax(powers[indexes])])
|
||
return float(frequencies[peak_index]), float(powers[peak_index])
|
||
|
||
low_frequency_hz, low_power = peak_in_window(-expected_offset_hz)
|
||
high_frequency_hz, high_power = peak_in_window(expected_offset_hz)
|
||
return {
|
||
"low_frequency_hz": low_frequency_hz,
|
||
"high_frequency_hz": high_frequency_hz,
|
||
"low_power": low_power,
|
||
"high_power": high_power,
|
||
}
|
||
|
||
|
||
def estimate_two_tone_offsets(
|
||
low_frequency_hz: float,
|
||
high_frequency_hz: float,
|
||
known_tone_offset_hz: float,
|
||
) -> dict:
|
||
"""Оценить грубую CFO и отношение тактов по двум известным тонам.
|
||
|
||
``clock_scale`` определён как отношение масштаба TX к масштабу RX.
|
||
Поэтому для перехода принятой последовательности на временную сетку
|
||
передатчика её ожидаемая длина равна ``round(N * clock_scale)``.
|
||
Невычислимая оценка представляется только значениями ``NaN``.
|
||
"""
|
||
|
||
if not np.isfinite(known_tone_offset_hz) or known_tone_offset_hz <= 0.0:
|
||
raise ValueError("Известный отступ тона должен быть положительным")
|
||
|
||
invalid = {
|
||
"carrier_offset_hz": float("nan"),
|
||
"clock_scale": float("nan"),
|
||
"sample_clock_error_ppm": float("nan"),
|
||
}
|
||
if not np.isfinite(low_frequency_hz) or not np.isfinite(high_frequency_hz):
|
||
return invalid
|
||
if high_frequency_hz <= low_frequency_hz:
|
||
return invalid
|
||
|
||
carrier_offset_hz = (low_frequency_hz + high_frequency_hz) / 2.0
|
||
clock_scale = (
|
||
(high_frequency_hz - low_frequency_hz)
|
||
/ (2.0 * known_tone_offset_hz)
|
||
)
|
||
if not np.isfinite(clock_scale) or clock_scale <= 0.0:
|
||
return invalid
|
||
|
||
return {
|
||
"carrier_offset_hz": float(carrier_offset_hz),
|
||
"clock_scale": float(clock_scale),
|
||
"sample_clock_error_ppm": float((clock_scale - 1.0) * 1e6),
|
||
}
|
||
|
||
|
||
def _tone_block_projections(
|
||
samples: np.ndarray,
|
||
sample_rate_hz: float,
|
||
coarse_frequency_hz: float,
|
||
block_samples: int,
|
||
hop_samples: int,
|
||
) -> tuple[np.ndarray, np.ndarray]:
|
||
"""Спроецировать фазово-непрерывный сигнал на грубую частоту тона."""
|
||
|
||
received = np.asarray(samples, dtype=np.complex128)
|
||
if received.ndim != 1:
|
||
raise ValueError("Комплексные отсчёты должны быть одномерными")
|
||
if not np.isfinite(sample_rate_hz) or sample_rate_hz <= 0.0:
|
||
raise ValueError("Частота дискретизации должна быть положительной")
|
||
if not np.isfinite(coarse_frequency_hz):
|
||
raise ValueError("Грубая частота тона должна быть конечной")
|
||
if not isinstance(block_samples, (int, np.integer)) or block_samples < 3:
|
||
raise ValueError("Размер блока должен быть целым числом не меньше трёх")
|
||
if not isinstance(hop_samples, (int, np.integer)) or hop_samples <= 0:
|
||
raise ValueError("Шаг блоков должен быть положительным целым числом")
|
||
if len(received) < block_samples + 2 * hop_samples:
|
||
raise ValueError("Для фазовой оценки нужны не менее трёх блоков")
|
||
|
||
starts = np.arange(
|
||
0,
|
||
len(received) - block_samples + 1,
|
||
hop_samples,
|
||
dtype=np.int64,
|
||
)
|
||
sample_indexes = np.arange(len(received), dtype=np.float64)
|
||
mixed = received * np.exp(
|
||
-1j * 2.0 * np.pi * coarse_frequency_hz * sample_indexes / sample_rate_hz
|
||
)
|
||
window = np.hanning(block_samples)
|
||
projections = np.asarray(
|
||
[
|
||
np.sum(mixed[start : start + block_samples] * window)
|
||
for start in starts
|
||
],
|
||
dtype=np.complex128,
|
||
)
|
||
center_times_seconds = (
|
||
starts.astype(np.float64) + (block_samples - 1) / 2.0
|
||
) / sample_rate_hz
|
||
return center_times_seconds, projections
|
||
|
||
|
||
def _weighted_phase_frequency(
|
||
center_times_seconds: np.ndarray,
|
||
projections: np.ndarray,
|
||
) -> tuple[float, float]:
|
||
"""Оценить наклон развёрнутой фазы и СКО остатка линейной модели."""
|
||
|
||
times = np.asarray(center_times_seconds, dtype=np.float64)
|
||
values = np.asarray(projections, dtype=np.complex128)
|
||
weights = np.abs(values) ** 2
|
||
weight_sum = float(np.sum(weights))
|
||
if (
|
||
times.ndim != 1
|
||
or values.ndim != 1
|
||
or len(times) != len(values)
|
||
or len(times) < 3
|
||
or not np.all(np.isfinite(times))
|
||
or not np.all(np.isfinite(values))
|
||
or not np.isfinite(weight_sum)
|
||
or weight_sum <= 0.0
|
||
):
|
||
raise ValueError("Фазовые проекции не позволяют оценить частоту")
|
||
|
||
phases = np.unwrap(np.angle(values))
|
||
mean_time = float(np.sum(weights * times) / weight_sum)
|
||
mean_phase = float(np.sum(weights * phases) / weight_sum)
|
||
centered_times = times - mean_time
|
||
denominator = float(np.sum(weights * centered_times**2))
|
||
if not np.isfinite(denominator) or denominator <= 0.0:
|
||
raise ValueError("Временные точки не позволяют оценить наклон фазы")
|
||
|
||
slope_rad_per_second = float(
|
||
np.sum(weights * centered_times * (phases - mean_phase)) / denominator
|
||
)
|
||
intercept = mean_phase - slope_rad_per_second * mean_time
|
||
residuals = phases - (intercept + slope_rad_per_second * times)
|
||
phase_fit_rms_rad = float(
|
||
np.sqrt(np.sum(weights * residuals**2) / weight_sum)
|
||
)
|
||
return slope_rad_per_second / (2.0 * np.pi), phase_fit_rms_rad
|
||
|
||
|
||
def refine_tone_frequency_from_phase(
|
||
samples: np.ndarray,
|
||
sample_rate_hz: float,
|
||
coarse_frequency_hz: float,
|
||
block_samples: int,
|
||
hop_samples: int | None = None,
|
||
) -> dict:
|
||
"""Уточнить частоту тона по наклону фазы когерентных проекций.
|
||
|
||
Входной участок должен быть фазово-непрерывным. Функцию нельзя применять
|
||
через границы отдельных аппаратных чтений, если непрерывность их фазы не
|
||
доказана. Грубая частота должна быть достаточно точной, чтобы остаточная
|
||
фаза между соседними блоками разворачивалась без неоднозначности.
|
||
"""
|
||
|
||
hop = block_samples // 2 if hop_samples is None else hop_samples
|
||
times, projections = _tone_block_projections(
|
||
samples,
|
||
sample_rate_hz,
|
||
coarse_frequency_hz,
|
||
block_samples,
|
||
hop,
|
||
)
|
||
residual_frequency_hz, phase_fit_rms_rad = _weighted_phase_frequency(
|
||
times,
|
||
projections,
|
||
)
|
||
return {
|
||
"frequency_hz": float(coarse_frequency_hz + residual_frequency_hz),
|
||
"residual_frequency_hz": float(residual_frequency_hz),
|
||
"phase_fit_rms_rad": float(phase_fit_rms_rad),
|
||
"block_count": int(len(projections)),
|
||
}
|
||
|
||
|
||
def refine_two_tone_frequencies_from_phase(
|
||
samples: np.ndarray,
|
||
sample_rate_hz: float,
|
||
low_coarse_frequency_hz: float,
|
||
high_coarse_frequency_hz: float,
|
||
block_samples: int,
|
||
hop_samples: int | None = None,
|
||
) -> TwoTonePhaseRefinement:
|
||
"""Уточнить несущую и расстояние двух тонов на непрерывном участке.
|
||
|
||
Несущая получается из двух индивидуальных наклонов фазы. Расстояние тонов
|
||
оценивается по фазе произведения верхней проекции на сопряжённую нижнюю:
|
||
общая фазовая ошибка приёмника при этом сокращается. Итоговые частоты
|
||
строятся из общей оценки центра и парной оценки расстояния.
|
||
"""
|
||
|
||
if high_coarse_frequency_hz <= low_coarse_frequency_hz:
|
||
raise ValueError("Частота верхнего тона должна быть выше частоты нижнего")
|
||
hop = block_samples // 2 if hop_samples is None else hop_samples
|
||
times, low_projections = _tone_block_projections(
|
||
samples,
|
||
sample_rate_hz,
|
||
low_coarse_frequency_hz,
|
||
block_samples,
|
||
hop,
|
||
)
|
||
high_times, high_projections = _tone_block_projections(
|
||
samples,
|
||
sample_rate_hz,
|
||
high_coarse_frequency_hz,
|
||
block_samples,
|
||
hop,
|
||
)
|
||
if not np.array_equal(times, high_times):
|
||
raise RuntimeError("Временные сетки двух тонов не совпали")
|
||
|
||
low_residual_hz, low_rms_rad = _weighted_phase_frequency(times, low_projections)
|
||
high_residual_hz, high_rms_rad = _weighted_phase_frequency(times, high_projections)
|
||
spacing_residual_hz, paired_rms_rad = _weighted_phase_frequency(
|
||
times,
|
||
high_projections * np.conj(low_projections),
|
||
)
|
||
|
||
individual_low_hz = low_coarse_frequency_hz + low_residual_hz
|
||
individual_high_hz = high_coarse_frequency_hz + high_residual_hz
|
||
center_hz = (individual_low_hz + individual_high_hz) / 2.0
|
||
spacing_hz = (
|
||
high_coarse_frequency_hz
|
||
- low_coarse_frequency_hz
|
||
+ spacing_residual_hz
|
||
)
|
||
low_frequency_hz = float(center_hz - spacing_hz / 2.0)
|
||
high_frequency_hz = float(center_hz + spacing_hz / 2.0)
|
||
residual_cfo_hz = float(
|
||
center_hz
|
||
- (low_coarse_frequency_hz + high_coarse_frequency_hz) / 2.0
|
||
)
|
||
phase_fit_rmse_rad = float(
|
||
np.sqrt(np.mean(np.square((low_rms_rad, high_rms_rad, paired_rms_rad))))
|
||
)
|
||
numeric = (
|
||
low_frequency_hz,
|
||
high_frequency_hz,
|
||
center_hz,
|
||
spacing_hz,
|
||
residual_cfo_hz,
|
||
phase_fit_rmse_rad,
|
||
low_rms_rad,
|
||
high_rms_rad,
|
||
paired_rms_rad,
|
||
)
|
||
valid = bool(all(np.isfinite(value) for value in numeric))
|
||
if not valid:
|
||
nan = float("nan")
|
||
return TwoTonePhaseRefinement(
|
||
valid=False,
|
||
invalid_reason="фазовое уточнение вернуло невычислимое значение",
|
||
low_frequency_hz=nan,
|
||
high_frequency_hz=nan,
|
||
carrier_offset_hz=nan,
|
||
tone_spacing_hz=nan,
|
||
residual_cfo_hz=nan,
|
||
phase_fit_rmse_rad=nan,
|
||
low_phase_fit_rms_rad=nan,
|
||
high_phase_fit_rms_rad=nan,
|
||
paired_phase_fit_rms_rad=nan,
|
||
block_count=int(len(low_projections)),
|
||
)
|
||
return TwoTonePhaseRefinement(
|
||
valid=True,
|
||
invalid_reason="",
|
||
low_frequency_hz=low_frequency_hz,
|
||
high_frequency_hz=high_frequency_hz,
|
||
carrier_offset_hz=float(center_hz),
|
||
tone_spacing_hz=float(spacing_hz),
|
||
residual_cfo_hz=residual_cfo_hz,
|
||
phase_fit_rmse_rad=phase_fit_rmse_rad,
|
||
low_phase_fit_rms_rad=float(low_rms_rad),
|
||
high_phase_fit_rms_rad=float(high_rms_rad),
|
||
paired_phase_fit_rms_rad=float(paired_rms_rad),
|
||
block_count=int(len(low_projections)),
|
||
)
|
||
|
||
|
||
def apply_coarse_frequency_correction(
|
||
samples: np.ndarray,
|
||
carrier_offset_hz: float,
|
||
sample_rate_hz: float,
|
||
) -> np.ndarray:
|
||
"""Убрать измеренный сдвиг несущей из комплексных отсчётов."""
|
||
|
||
received = np.asarray(samples, dtype=np.complex128)
|
||
if received.ndim != 1:
|
||
raise ValueError("Комплексные отсчёты должны быть одномерными")
|
||
if not np.isfinite(carrier_offset_hz):
|
||
raise ValueError("Сдвиг несущей должен быть конечным")
|
||
if not np.isfinite(sample_rate_hz) or sample_rate_hz <= 0.0:
|
||
raise ValueError("Частота дискретизации должна быть положительной")
|
||
|
||
sample_indexes = np.arange(len(received), dtype=np.float64)
|
||
correction = np.exp(
|
||
-1j * 2.0 * np.pi * carrier_offset_hz * sample_indexes / sample_rate_hz
|
||
)
|
||
return received * correction
|
||
|
||
|
||
def resample_for_clock_scale(
|
||
samples: np.ndarray,
|
||
clock_scale: float,
|
||
) -> np.ndarray:
|
||
"""Передискретизировать отсчёты на временную сетку передатчика.
|
||
|
||
Используется комплексная линейная интерполяция. Она детерминирована,
|
||
не требует рационального приближения ppm-отношения и подходит для
|
||
сильно передискретизированного BPSK-тракта. Направление преобразования
|
||
закрепляется синтетическими тестами для ошибок обоих знаков.
|
||
"""
|
||
|
||
received = np.asarray(samples, dtype=np.complex128)
|
||
if received.ndim != 1:
|
||
raise ValueError("Комплексные отсчёты должны быть одномерными")
|
||
if not np.isfinite(clock_scale) or clock_scale <= 0.0:
|
||
raise ValueError("Масштаб такта должен быть положительным")
|
||
if received.size == 0:
|
||
return received.copy()
|
||
if received.size == 1:
|
||
return np.repeat(received, max(1, round(clock_scale)))
|
||
|
||
output_length = max(1, round(len(received) * clock_scale))
|
||
output_indexes = np.arange(output_length, dtype=np.float64)
|
||
source_positions = np.minimum(output_indexes / clock_scale, len(received) - 1.0)
|
||
source_indexes = np.arange(len(received), dtype=np.float64)
|
||
real = np.interp(source_positions, source_indexes, received.real)
|
||
imaginary = np.interp(source_positions, source_indexes, received.imag)
|
||
return real + 1j * imaginary
|
||
|
||
|
||
def estimate_known_pilot_carrier(
|
||
received_symbols: np.ndarray,
|
||
known_symbols: np.ndarray,
|
||
symbol_rate: float = SYMBOL_RATE,
|
||
block_symbol_count: int = KNOWN_PILOT_BLOCK_SYMBOL_COUNT,
|
||
minimum_block_coherence: float = MINIMUM_KNOWN_PILOT_COHERENCE,
|
||
maximum_phase_fit_rmse_rad: float = MAXIMUM_KNOWN_PILOT_PHASE_FIT_RMSE_RAD,
|
||
) -> KnownPilotCarrierEstimate:
|
||
"""Оценить малую остаточную CFO по отдельному известному BPSK-пилоту.
|
||
|
||
Известные знаки удаляются, затем символы объединяются в короткие
|
||
когерентные блоки. Развёрнутая фаза блоков аппроксимируется устойчивой
|
||
линейной моделью Huber. Полезная нагрузка для оценки не используется.
|
||
"""
|
||
|
||
received = np.asarray(received_symbols, dtype=np.complex128)
|
||
known = np.asarray(known_symbols, dtype=np.complex128)
|
||
if received.ndim != 1 or known.ndim != 1:
|
||
raise ValueError("Принятый и известный пилоты должны быть одномерными")
|
||
if len(received) != len(known):
|
||
raise ValueError("Принятый и известный пилоты должны иметь одинаковую длину")
|
||
if not np.isfinite(symbol_rate) or symbol_rate <= 0.0:
|
||
raise ValueError("Символьная скорость должна быть положительной")
|
||
if not isinstance(block_symbol_count, (int, np.integer)) or block_symbol_count < 2:
|
||
raise ValueError("Размер когерентного блока должен быть целым и не меньше двух")
|
||
if not 0.0 < minimum_block_coherence <= 1.0:
|
||
raise ValueError("Порог когерентности должен находиться в интервале (0, 1]")
|
||
if not np.isfinite(maximum_phase_fit_rmse_rad) or maximum_phase_fit_rmse_rad <= 0.0:
|
||
raise ValueError("Порог ошибки фазовой модели должен быть положительным")
|
||
if not np.all(np.isfinite(received)) or not np.all(np.isfinite(known)):
|
||
raise ValueError("Пилот не должен содержать нечисловые значения")
|
||
if np.any(np.abs(known) <= 0.0):
|
||
raise ValueError("Все известные символы пилота должны быть ненулевыми")
|
||
|
||
block_count = len(received) // block_symbol_count
|
||
if block_count < MINIMUM_KNOWN_PILOT_BLOCK_COUNT:
|
||
raise ValueError(
|
||
f"Для оценки нужны не менее {MINIMUM_KNOWN_PILOT_BLOCK_COUNT} когерентных блоков"
|
||
)
|
||
usable_count = block_count * block_symbol_count
|
||
despread = received[:usable_count] * np.conj(known[:usable_count])
|
||
blocks = despread.reshape(block_count, block_symbol_count)
|
||
block_sums = np.sum(blocks, axis=1)
|
||
block_magnitude_sums = np.sum(np.abs(blocks), axis=1) + 1e-12
|
||
block_coherences = np.abs(block_sums) / block_magnitude_sums
|
||
mean_block_coherence = float(np.mean(block_coherences))
|
||
|
||
centers_symbols = (
|
||
np.arange(block_count, dtype=np.float64) * block_symbol_count
|
||
+ (block_symbol_count - 1) / 2.0
|
||
)
|
||
center_times_seconds = centers_symbols / symbol_rate
|
||
phases = np.unwrap(np.angle(block_sums))
|
||
base_weights = np.maximum(np.abs(block_sums), 1e-12)
|
||
design = np.column_stack((np.ones(block_count), center_times_seconds))
|
||
weights = base_weights.copy()
|
||
coefficients = np.zeros(2, dtype=np.float64)
|
||
|
||
for _iteration in range(12):
|
||
root_weights = np.sqrt(weights)
|
||
new_coefficients, *_unused = np.linalg.lstsq(
|
||
design * root_weights[:, np.newaxis],
|
||
phases * root_weights,
|
||
rcond=None,
|
||
)
|
||
residuals = phases - design @ new_coefficients
|
||
residual_median = float(np.median(residuals))
|
||
robust_scale = (
|
||
1.4826 * float(np.median(np.abs(residuals - residual_median)))
|
||
+ 1e-9
|
||
)
|
||
huber_limit = 1.345 * robust_scale
|
||
robust_weights = np.ones_like(residuals)
|
||
outliers = np.abs(residuals) > huber_limit
|
||
robust_weights[outliers] = huber_limit / np.abs(residuals[outliers])
|
||
weights = base_weights * robust_weights
|
||
if np.allclose(coefficients, new_coefficients, rtol=0.0, atol=1e-12):
|
||
coefficients = new_coefficients
|
||
break
|
||
coefficients = new_coefficients
|
||
|
||
residuals = phases - design @ coefficients
|
||
phase_fit_rmse_rad = float(
|
||
np.sqrt(np.sum(base_weights * residuals**2) / np.sum(base_weights))
|
||
)
|
||
frequency_hz = float(coefficients[1] / (2.0 * np.pi))
|
||
initial_phase_rad = float(coefficients[0])
|
||
phase_increment = float(2.0 * np.pi * frequency_hz / symbol_rate)
|
||
|
||
invalid_reasons: list[str] = []
|
||
if not all(
|
||
np.isfinite(value)
|
||
for value in (frequency_hz, initial_phase_rad, phase_fit_rmse_rad, mean_block_coherence)
|
||
):
|
||
invalid_reasons.append("оценка содержит нечисловое значение")
|
||
if mean_block_coherence < minimum_block_coherence:
|
||
invalid_reasons.append(
|
||
f"когерентность {mean_block_coherence:.4f} ниже порога {minimum_block_coherence:.2f}"
|
||
)
|
||
if phase_fit_rmse_rad > maximum_phase_fit_rmse_rad:
|
||
invalid_reasons.append(
|
||
f"RMSE фазы {phase_fit_rmse_rad:.4f} выше порога {maximum_phase_fit_rmse_rad:.2f} рад"
|
||
)
|
||
|
||
valid = not invalid_reasons
|
||
nan = float("nan")
|
||
return KnownPilotCarrierEstimate(
|
||
valid=valid,
|
||
invalid_reason="; ".join(invalid_reasons),
|
||
frequency_hz=frequency_hz if valid else nan,
|
||
phase_increment_rad_per_symbol=phase_increment if valid else nan,
|
||
initial_phase_rad=initial_phase_rad if valid else nan,
|
||
phase_fit_rmse_rad=phase_fit_rmse_rad,
|
||
mean_block_coherence=mean_block_coherence,
|
||
block_count=block_count,
|
||
known_symbol_count=len(known),
|
||
)
|
||
|
||
|
||
# Перенесено дословно из Lab023 (tests/lab023_guarded_cfo_correction.py).
|
||
def estimate_carrier_parameters(
|
||
received_symbols: np.ndarray,
|
||
marker_symbols: np.ndarray,
|
||
) -> dict:
|
||
"""
|
||
Оценить фазу и частотное рассогласование по маркеру.
|
||
|
||
Алгоритм:
|
||
|
||
1. Удалить известные BPSK-знаки маркера.
|
||
2. Получить грубую оценку CFO по соседним символам.
|
||
3. Выполнить уточняющий частотный поиск.
|
||
4. Оценить начальную фазу после компенсации CFO.
|
||
5. Рассчитать достоверность оценки.
|
||
"""
|
||
|
||
marker_length = len(
|
||
marker_symbols
|
||
)
|
||
|
||
received_marker = received_symbols[
|
||
:marker_length
|
||
]
|
||
|
||
if len(received_marker) != marker_length:
|
||
raise ValueError(
|
||
"Недостаточно символов маркера"
|
||
)
|
||
|
||
# Известные BPSK-знаки равны -1 или +1.
|
||
# Умножение удаляет переданную манипуляцию,
|
||
# оставляя фазу канала и шум.
|
||
despread_marker = (
|
||
received_marker
|
||
* marker_symbols
|
||
)
|
||
|
||
marker_indexes = np.arange(
|
||
marker_length,
|
||
dtype=np.float64,
|
||
)
|
||
|
||
marker_magnitude_sum = float(
|
||
np.sum(
|
||
np.abs(
|
||
despread_marker
|
||
)
|
||
)
|
||
)
|
||
|
||
# ========================================================
|
||
# Постоянная фаза без CFO-компенсации
|
||
# ========================================================
|
||
|
||
constant_coherent_sum = np.sum(
|
||
despread_marker
|
||
)
|
||
|
||
constant_phase = float(
|
||
np.angle(
|
||
constant_coherent_sum
|
||
)
|
||
)
|
||
|
||
constant_coherence = float(
|
||
np.abs(
|
||
constant_coherent_sum
|
||
)
|
||
/ (
|
||
marker_magnitude_sum
|
||
+ 1e-12
|
||
)
|
||
)
|
||
|
||
# ========================================================
|
||
# Грубая оценка CFO
|
||
# ========================================================
|
||
|
||
adjacent_products = (
|
||
despread_marker[1:]
|
||
* np.conj(
|
||
despread_marker[:-1]
|
||
)
|
||
)
|
||
|
||
adjacent_sum = np.sum(
|
||
adjacent_products
|
||
)
|
||
|
||
coarse_phase_increment = float(
|
||
np.angle(
|
||
adjacent_sum
|
||
)
|
||
)
|
||
|
||
coarse_frequency_hz = (
|
||
coarse_phase_increment
|
||
* SYMBOL_RATE
|
||
/ (2.0 * np.pi)
|
||
)
|
||
|
||
phase_consistency = float(
|
||
np.abs(
|
||
adjacent_sum
|
||
)
|
||
/ (
|
||
np.sum(
|
||
np.abs(
|
||
adjacent_products
|
||
)
|
||
)
|
||
+ 1e-12
|
||
)
|
||
)
|
||
|
||
# ========================================================
|
||
# Уточняющий поиск CFO
|
||
# ========================================================
|
||
|
||
frequency_candidates_hz = np.arange(
|
||
(
|
||
coarse_frequency_hz
|
||
- CFO_REFINEMENT_HALF_WIDTH_HZ
|
||
),
|
||
(
|
||
coarse_frequency_hz
|
||
+ CFO_REFINEMENT_HALF_WIDTH_HZ
|
||
+ CFO_REFINEMENT_STEP_HZ / 2.0
|
||
),
|
||
CFO_REFINEMENT_STEP_HZ,
|
||
dtype=np.float64,
|
||
)
|
||
|
||
phase_increment_candidates = (
|
||
2.0
|
||
* np.pi
|
||
* frequency_candidates_hz
|
||
/ SYMBOL_RATE
|
||
)
|
||
|
||
candidate_compensation = np.exp(
|
||
-1j
|
||
* phase_increment_candidates[
|
||
:, np.newaxis
|
||
]
|
||
* marker_indexes[
|
||
np.newaxis, :
|
||
]
|
||
)
|
||
|
||
coherent_sums = np.sum(
|
||
despread_marker[
|
||
np.newaxis, :
|
||
]
|
||
* candidate_compensation,
|
||
axis=1,
|
||
)
|
||
|
||
best_candidate_index = int(
|
||
np.argmax(
|
||
np.abs(
|
||
coherent_sums
|
||
)
|
||
)
|
||
)
|
||
|
||
estimated_frequency_hz = float(
|
||
frequency_candidates_hz[
|
||
best_candidate_index
|
||
]
|
||
)
|
||
|
||
phase_increment = float(
|
||
phase_increment_candidates[
|
||
best_candidate_index
|
||
]
|
||
)
|
||
|
||
best_coherent_sum = (
|
||
coherent_sums[
|
||
best_candidate_index
|
||
]
|
||
)
|
||
|
||
initial_phase_after_cfo = float(
|
||
np.angle(
|
||
best_coherent_sum
|
||
)
|
||
)
|
||
|
||
cfo_coherence = float(
|
||
np.abs(
|
||
best_coherent_sum
|
||
)
|
||
/ (
|
||
marker_magnitude_sum
|
||
+ 1e-12
|
||
)
|
||
)
|
||
|
||
coherence_gain = (
|
||
cfo_coherence
|
||
- constant_coherence
|
||
)
|
||
|
||
return {
|
||
"constant_phase": constant_phase,
|
||
"constant_coherence": constant_coherence,
|
||
"phase_increment": phase_increment,
|
||
"estimated_frequency_hz": (
|
||
estimated_frequency_hz
|
||
),
|
||
"initial_phase_after_cfo": (
|
||
initial_phase_after_cfo
|
||
),
|
||
"phase_consistency": phase_consistency,
|
||
"cfo_coherence": cfo_coherence,
|
||
"coherence_gain": coherence_gain,
|
||
}
|
||
|
||
|
||
# Перенесено дословно из Lab023 (tests/lab023_guarded_cfo_correction.py).
|
||
def correct_phase_and_frequency(
|
||
received_symbols: np.ndarray,
|
||
initial_phase_radians: float,
|
||
phase_increment: float,
|
||
) -> np.ndarray:
|
||
"""
|
||
Компенсировать постоянную фазу и CFO.
|
||
"""
|
||
|
||
symbol_indexes = np.arange(
|
||
len(received_symbols),
|
||
dtype=np.float64,
|
||
)
|
||
|
||
phase_model = (
|
||
initial_phase_radians
|
||
+ phase_increment
|
||
* symbol_indexes
|
||
)
|
||
|
||
return (
|
||
received_symbols
|
||
* np.exp(
|
||
-1j * phase_model
|
||
)
|
||
)
|
||
|
||
|
||
# Перенесено дословно из Lab019 (tests/lab019_bpsk_receiver.py).
|
||
def build_frame_marker(
|
||
) -> tuple[np.ndarray, np.ndarray]:
|
||
"""
|
||
Сформировать:
|
||
|
||
PREAMBLE + RADIO SYNC
|
||
|
||
Возвращает биты и BPSK-символы маркера.
|
||
"""
|
||
|
||
preamble_bits = np.tile(
|
||
np.array(
|
||
[1, 0],
|
||
dtype=np.uint8,
|
||
),
|
||
PREAMBLE_BIT_COUNT // 2,
|
||
)
|
||
|
||
sync_bytes = struct.pack(
|
||
">H",
|
||
RADIO_SYNC_WORD,
|
||
)
|
||
|
||
sync_bits = bytes_to_bits(
|
||
sync_bytes
|
||
)
|
||
|
||
marker_bits = np.concatenate(
|
||
[
|
||
preamble_bits,
|
||
sync_bits,
|
||
]
|
||
)
|
||
|
||
marker_symbols = bpsk_modulate(
|
||
marker_bits
|
||
)
|
||
|
||
return marker_bits, marker_symbols
|
||
|
||
|
||
# Перенесено дословно из Lab019 (tests/lab019_bpsk_receiver.py).
|
||
def find_radio_frame(
|
||
matched_iq: np.ndarray,
|
||
marker_symbols: np.ndarray,
|
||
samples_per_symbol: int,
|
||
) -> dict:
|
||
"""
|
||
Найти фазу дискретизации и начало радиокадра.
|
||
|
||
Для каждой возможной фазы:
|
||
|
||
0, 1, 2, ... SPS - 1
|
||
|
||
берём по одному сэмплу на символ и вычисляем
|
||
нормированную корреляцию с известным маркером.
|
||
|
||
Использование комплексной корреляции позволяет
|
||
одновременно оценить постоянный фазовый поворот.
|
||
"""
|
||
|
||
marker_energy = float(
|
||
np.sum(
|
||
np.abs(marker_symbols) ** 2
|
||
)
|
||
)
|
||
|
||
best_result = None
|
||
|
||
for sample_phase in range(
|
||
samples_per_symbol
|
||
):
|
||
|
||
symbol_samples = matched_iq[
|
||
sample_phase::samples_per_symbol
|
||
]
|
||
|
||
if len(symbol_samples) < len(
|
||
marker_symbols
|
||
):
|
||
continue
|
||
|
||
correlation = np.correlate(
|
||
symbol_samples,
|
||
marker_symbols,
|
||
mode="valid",
|
||
)
|
||
|
||
window_energy = np.convolve(
|
||
np.abs(symbol_samples) ** 2,
|
||
np.ones(
|
||
len(marker_symbols)
|
||
),
|
||
mode="valid",
|
||
)
|
||
|
||
denominator = (
|
||
np.sqrt(
|
||
window_energy
|
||
* marker_energy
|
||
)
|
||
+ 1e-12
|
||
)
|
||
|
||
normalized_correlation = (
|
||
np.abs(correlation)
|
||
/ denominator
|
||
)
|
||
|
||
start_symbol_index = int(
|
||
np.argmax(
|
||
normalized_correlation
|
||
)
|
||
)
|
||
|
||
correlation_score = float(
|
||
normalized_correlation[
|
||
start_symbol_index
|
||
]
|
||
)
|
||
|
||
complex_correlation = correlation[
|
||
start_symbol_index
|
||
]
|
||
|
||
if (
|
||
best_result is None
|
||
or correlation_score
|
||
> best_result["score"]
|
||
):
|
||
best_result = {
|
||
"score": correlation_score,
|
||
"sample_phase": sample_phase,
|
||
"start_symbol_index": (
|
||
start_symbol_index
|
||
),
|
||
"symbol_samples": (
|
||
symbol_samples
|
||
),
|
||
"correlation": correlation,
|
||
"normalized_correlation": (
|
||
normalized_correlation
|
||
),
|
||
"complex_correlation": (
|
||
complex_correlation
|
||
),
|
||
}
|
||
|
||
if best_result is None:
|
||
raise RuntimeError(
|
||
"Не удалось выполнить поиск радиокадра"
|
||
)
|
||
|
||
return best_result
|
||
|
||
|
||
# Перенесено дословно из Lab018 (tests/lab018_bpsk_radio_frame.py).
|
||
def root_raised_cosine_taps(
|
||
rolloff: float,
|
||
samples_per_symbol: int,
|
||
span_symbols: int,
|
||
) -> np.ndarray:
|
||
"""
|
||
Рассчитать коэффициенты Root Raised Cosine-фильтра.
|
||
|
||
Параметры
|
||
----------
|
||
rolloff:
|
||
Коэффициент скругления beta.
|
||
|
||
samples_per_symbol:
|
||
Количество сэмплов на символ.
|
||
|
||
span_symbols:
|
||
Полная длина фильтра в символах.
|
||
|
||
Возвращает
|
||
----------
|
||
Одномерный массив коэффициентов фильтра.
|
||
"""
|
||
|
||
if not 0.0 < rolloff <= 1.0:
|
||
raise ValueError(
|
||
"rolloff должен находиться в диапазоне 0...1"
|
||
)
|
||
|
||
if samples_per_symbol <= 0:
|
||
raise ValueError(
|
||
"samples_per_symbol должен быть положительным"
|
||
)
|
||
|
||
if span_symbols <= 0:
|
||
raise ValueError(
|
||
"span_symbols должен быть положительным"
|
||
)
|
||
|
||
if span_symbols % 2 != 0:
|
||
raise ValueError(
|
||
"span_symbols должен быть чётным"
|
||
)
|
||
|
||
half_sample_count = (
|
||
span_symbols
|
||
* samples_per_symbol
|
||
// 2
|
||
)
|
||
|
||
sample_indexes = np.arange(
|
||
-half_sample_count,
|
||
half_sample_count + 1,
|
||
dtype=np.float64,
|
||
)
|
||
|
||
# Время нормировано к длительности одного символа.
|
||
time_values = (
|
||
sample_indexes
|
||
/ samples_per_symbol
|
||
)
|
||
|
||
taps = np.zeros_like(
|
||
time_values
|
||
)
|
||
|
||
beta = rolloff
|
||
|
||
for index, time_value in enumerate(
|
||
time_values
|
||
):
|
||
|
||
# Особая точка t = 0.
|
||
if np.isclose(
|
||
time_value,
|
||
0.0,
|
||
):
|
||
taps[index] = (
|
||
1.0
|
||
- beta
|
||
+ (
|
||
4.0
|
||
* beta
|
||
/ np.pi
|
||
)
|
||
)
|
||
|
||
continue
|
||
|
||
# Особые точки t = ±1/(4 beta).
|
||
if np.isclose(
|
||
abs(time_value),
|
||
1.0 / (4.0 * beta),
|
||
):
|
||
taps[index] = (
|
||
beta
|
||
/ np.sqrt(2.0)
|
||
* (
|
||
(
|
||
1.0
|
||
+ 2.0 / np.pi
|
||
)
|
||
* np.sin(
|
||
np.pi
|
||
/ (4.0 * beta)
|
||
)
|
||
+ (
|
||
1.0
|
||
- 2.0 / np.pi
|
||
)
|
||
* np.cos(
|
||
np.pi
|
||
/ (4.0 * beta)
|
||
)
|
||
)
|
||
)
|
||
|
||
continue
|
||
|
||
numerator = (
|
||
np.sin(
|
||
np.pi
|
||
* time_value
|
||
* (1.0 - beta)
|
||
)
|
||
+ (
|
||
4.0
|
||
* beta
|
||
* time_value
|
||
* np.cos(
|
||
np.pi
|
||
* time_value
|
||
* (1.0 + beta)
|
||
)
|
||
)
|
||
)
|
||
|
||
denominator = (
|
||
np.pi
|
||
* time_value
|
||
* (
|
||
1.0
|
||
- (
|
||
4.0
|
||
* beta
|
||
* time_value
|
||
) ** 2
|
||
)
|
||
)
|
||
|
||
taps[index] = (
|
||
numerator
|
||
/ denominator
|
||
)
|
||
|
||
# Нормируем энергию фильтра.
|
||
taps /= np.sqrt(
|
||
np.sum(
|
||
taps ** 2
|
||
)
|
||
)
|
||
|
||
return taps
|
||
|
||
|
||
# Ниже — функции, добавленные при выделении модуля. В Lab018, Lab019 и
|
||
# Lab023 формирование кадра не принимало аргументов и собирало жёстко
|
||
# заданное текстовое сообщение, а разбор возвращал строку состояния.
|
||
# Структура кадра сохранена без изменений:
|
||
#
|
||
# [преамбула 64 бита] [синхрослово 16 | длина 16] [пакет протокола]
|
||
|
||
|
||
def build_radio_frame(
|
||
protocol_packet: bytes,
|
||
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
|
||
"""Сформировать радиокадр вокруг готового пакета протокола.
|
||
|
||
Возвращает биты кадра, символы кадра и известные символы маркера,
|
||
состоящего из преамбулы и синхрослова.
|
||
"""
|
||
|
||
if not isinstance(protocol_packet, (bytes, bytearray)):
|
||
raise TypeError("protocol_packet должен иметь тип bytes")
|
||
|
||
protocol_packet = bytes(protocol_packet)
|
||
|
||
if not 1 <= len(protocol_packet) <= MAXIMUM_PROTOCOL_PACKET_BYTES:
|
||
raise ValueError(
|
||
"Длина пакета протокола должна быть от 1 до "
|
||
f"{MAXIMUM_PROTOCOL_PACKET_BYTES} байт"
|
||
)
|
||
|
||
radio_header = struct.pack(">HH", RADIO_SYNC_WORD, len(protocol_packet))
|
||
|
||
preamble_bits = np.tile(
|
||
np.array([1, 0], dtype=np.uint8),
|
||
PREAMBLE_BIT_COUNT // 2,
|
||
)
|
||
|
||
frame_bits = np.concatenate(
|
||
[
|
||
preamble_bits,
|
||
bytes_to_bits(radio_header),
|
||
bytes_to_bits(protocol_packet),
|
||
]
|
||
)
|
||
|
||
frame_symbols = bpsk_modulate(frame_bits)
|
||
marker_symbols = frame_symbols[:MARKER_BIT_COUNT]
|
||
|
||
return frame_bits, frame_symbols, marker_symbols
|
||
|
||
|
||
def parse_radio_frame(received_bits: np.ndarray) -> bytes | None:
|
||
"""Извлечь пакет протокола из битов принятого кадра.
|
||
|
||
Возвращает пакет либо None, если заголовок кадра не разобран.
|
||
Проверка контрольной суммы самого пакета в задачу не входит и
|
||
выполняется отдельно через protocol.packet.parse_packet.
|
||
"""
|
||
|
||
received_bits = np.asarray(received_bits, dtype=np.uint8)
|
||
|
||
header_start = PREAMBLE_BIT_COUNT
|
||
header_end = header_start + RADIO_HEADER_BIT_COUNT
|
||
|
||
if len(received_bits) < header_end:
|
||
return None
|
||
|
||
try:
|
||
radio_header = bits_to_bytes(received_bits[header_start:header_end])
|
||
received_sync, packet_length = struct.unpack(">HH", radio_header)
|
||
except (ValueError, struct.error):
|
||
return None
|
||
|
||
if received_sync != RADIO_SYNC_WORD:
|
||
return None
|
||
|
||
if not 1 <= packet_length <= MAXIMUM_PROTOCOL_PACKET_BYTES:
|
||
return None
|
||
|
||
packet_start = header_end
|
||
packet_end = packet_start + packet_length * 8
|
||
|
||
if len(received_bits) < packet_end:
|
||
return None
|
||
|
||
try:
|
||
return bits_to_bytes(received_bits[packet_start:packet_end])
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def radio_frame_bit_count(protocol_packet_bytes: int) -> int:
|
||
"""Вернуть длину радиокадра в битах для пакета заданного размера."""
|
||
|
||
return (
|
||
PREAMBLE_BIT_COUNT
|
||
+ RADIO_HEADER_BIT_COUNT
|
||
+ protocol_packet_bytes * 8
|
||
)
|
||
|
||
|
||
# Перенесено дословно из Lab023 (experiments/lab023_guarded_cfo_correction.py).
|
||
def should_apply_cfo_correction(
|
||
estimated_frequency_hz: float,
|
||
phase_consistency: float,
|
||
coherence_gain: float,
|
||
) -> bool:
|
||
"""
|
||
Разрешить CFO-коррекцию только при наличии
|
||
достаточных оснований.
|
||
|
||
Требования:
|
||
|
||
1. Оценка находится вне мёртвой зоны.
|
||
2. Межсимвольное вращение достаточно согласованно.
|
||
3. Компенсация CFO действительно повышает
|
||
когерентность известного маркера.
|
||
"""
|
||
|
||
MINIMUM_COHERENCE_GAIN = 0.02
|
||
|
||
return (
|
||
abs(
|
||
estimated_frequency_hz
|
||
)
|
||
>= CFO_DEAD_ZONE_HZ
|
||
and phase_consistency
|
||
>= MINIMUM_PHASE_CONSISTENCY
|
||
and coherence_gain
|
||
>= MINIMUM_COHERENCE_GAIN
|
||
)
|