turov_av/SDR-Rover
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Lab052: suppress false preamble candidates

Browse Source
This commit is contained in:
LittleSam129
2026-09-08 17:29:52 +03:00
parent 9f33607d3a
commit cc38900cae
27 changed files with 77397 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
experiments/lab052_detection_statistics.py Normal file
View File

@@ -0,0 +1,157 @@
"""Lab052: распределения и численные проверки после независимых решений RX."""
from __future__ import annotations
import json
from pathlib import Path
import sys
if __package__ in (None, ""):
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import numpy as np
from experiments.lab052_detector_study import OUTPUT, quantiles, write_result
from experiments import lab052_candidate_detection as detection
from experiments.lab051_receiver_profile import array_observation
def measure_arrays() -> dict:
"""Отдельный короткий RX replay, без смешивания tracing с замером скорости.
ndarray в Python-locals даёт нижнюю границу наблюдаемых массивов; C/SciPy
могут иметь невидимые временные выделения. tracemalloc peak — отдельная
измеренная величина, не RSS и не универсальная верхняя граница памяти.
"""
physical = detection.phy.lab045
payload = bytes(range(256))
plan = physical.build_transmission_plan(payload)
raw = detection.phy.build_parameterized_tx_cs8(plan, detection.phy.PhyMode(500000))
components = np.frombuffer(raw, np.int8).astype(np.float32)
values = components[0::2] + 1j * components[1::2]
random_generator = np.random.default_rng(52040)
sigma = np.sqrt(float(np.sum(abs(values)**2)) / plan.frame_symbol_count / (2 * 10**.7))
values += sigma * (random_generator.normal(size=len(values)) + 1j*random_generator.normal(size=len(values)))
samples = .2 * (random_generator.normal(size=63217 + len(values) + 350000)
+ 1j * random_generator.normal(size=63217 + len(values) + 350000))
samples[63217:63217 + len(values)] = values
samples = samples.astype(np.complex64)
def replay():
receiver = detection.StreamingReceiver()
packets = []
for start in range(0, len(samples), receiver.block_samples):
packets.extend(receiver.push_iq_block(samples[start:start + receiver.block_samples]))
packets.extend(receiver.finish())
assert len(packets) == 1 and packets[0].payload == payload
result = {**array_observation(replay), "iq_seconds": len(samples) / 8000000,
"packets": 1, "interpretation": "observed Python ndarray lower bound; tracked RX allocation peak; input IQ prepared before tracing",
"timing_not_used_as_benchmark": True, "rf_tx": 0}
write_result("allocations.json", result)
return result
def analyze() -> dict:
baseline = json.loads((OUTPUT / "baseline_profile.json").read_text(encoding="utf-8"))
optimized = json.loads((OUTPUT / "parts2.json").read_text(encoding="utf-8"))
before = np.load(OUTPUT / "baseline_observations.npz")
after = np.load(OUTPUT / "parts2_observations.npz")
peaks, outcomes = before["peaks"], before["outcomes"]
true = outcomes[outcomes[:, 3] == 1]
false = outcomes[outcomes[:, 3] == 0]
starts = np.sort(true[:, 2].astype(np.int64))
new_starts = np.sort(after["outcomes"][after["outcomes"][:, 3] == 1, 2].astype(np.int64))
missing = np.setdiff1d(starts, new_starts)
unexpected = np.setdiff1d(new_starts, starts)
main_scores = []
for start in starts:
low, high = np.searchsorted(peaks[:, 0], [start - detection.COARSE_ERROR_SAMPLES,
start + detection.COARSE_ERROR_SAMPLES], side="left")
main_scores.append(float(peaks[low:high, 1].max()))
# Окна чистого фона выделяются ТОЛЬКО после независимых принятых кадров.
# В этом опыте decoded PHY packet 268B => waveform 92256 samples.
frame_samples = (detection.phy.lab045.PREAMBLE_BIT_COUNT + detection.phy.lab045.PHY_HEADER_BIT_COUNT
+ detection.phy.lab045.conv.encoded_bit_count((256 + detection.phy.lab045.MIN_PACKET_BYTES) * 8)
+ 42) * 16
blocks = baseline["detection_profile"]["blocks"]
noise_blocks = []
for block in blocks:
low, high = block["start_sample"], block["end_sample"] + (1280 + 42) * 16
if not np.any((starts < high) & (starts + frame_samples > low)):
noise_blocks.append(block)
noise_statistics = {name: quantiles([block["score_statistics"][name] for block in noise_blocks])
for name in ("median", "mad", "p90", "p99", "p999")}
cheap = np.load(OUTPUT / "parts2_cheap.npy")
accepted_coarse = after["outcomes"][after["outcomes"][:, 3] == 1, 0]
is_true = np.isin(cheap[:, 0], accepted_coarse)
part_min = cheap[:, 3:].min(axis=1)
false_count = len(false)
new_false = optimized["detector"]["strict_false_candidates"]
result = {
"classification": "strict PHY/FEC/CRC verdict first; no diagnostic feedback into RX",
"baseline_true_unique_starts": len(np.unique(starts)),
"optimized_true_unique_starts": len(np.unique(new_starts)),
"missing_true_starts": missing.tolist(), "unexpected_true_starts": unexpected.tolist(),
"true_main_peak_scores": quantiles(main_scores), "false_strict_scores": quantiles(false[:, 1]),
"baseline_false_refined_positions": len(np.unique(false[:, 2])),
"baseline_false_repeated_refinements": len(false) - len(np.unique(false[:, 2])),
"noise_only_blocks": len(noise_blocks), "noise_block_statistics": noise_statistics,
"split_minimum_true_strict": quantiles(part_min[is_true]),
"split_minimum_other_candidates": quantiles(part_min[~is_true]),
"strict_false_reduction": false_count / new_false,
"strict_false_reduction_fraction": 1 - new_false / false_count,
"first_stage_threshold_unchanged": detection.phy.lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD,
"part_threshold": detection.part_threshold(2), "rf_tx": 0,
}
write_result("statistics.json", result)
# Научные статические рисунки сохраняются самостоятельными артефактами.
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
figure, axes = plt.subplots(2, 2, figsize=(12, 8), constrained_layout=True)
axis = axes[0, 0]
bins = np.linspace(0, 1, 101)
axis.hist(false[:, 1], bins, histtype="step", label="Ложные строгие кандидаты", log=True)
axis.hist(main_scores, bins, histtype="step", label="Главный пик настоящей преамбулы", log=True)
axis.axvline(result["first_stage_threshold_unchanged"], color="gray", linestyle="--", label="Первый порог")
axis.set(xlabel="Нормированная корреляция", ylabel="Количество", title="Классификация после PHY/FEC/CRC")
axis.legend(fontsize=8)
axis = axes[0, 1]
for name, label in (("median", "Медиана"), ("mad", "MAD"), ("p99", "Квантиль 99%")):
axis.plot([block["start_sample"] / 8000000 for block in noise_blocks],
[block["score_statistics"][name] for block in noise_blocks], ".", markersize=2, label=label)
axis.set(xlabel="Время IQ, с", ylabel="Корреляционная статистика", title="Блоки чистого фона, определённые после RX")
axis.legend(fontsize=8)
axis = axes[1, 0]
axis.hist(part_min[~is_true], bins, histtype="step", log=True, label="Остальные локальные кандидаты")
axis.hist(part_min[is_true], bins, histtype="step", log=True, label="Кандидаты с последующим CRC PASS")
axis.axvline(detection.part_threshold(2), color="gray", linestyle="--", label="Порог каждой половины")
axis.set(xlabel="Меньшая корреляция двух половин", ylabel="Количество", title="Дешёвая проверка обеих частей")
axis.legend(fontsize=8)
axis = axes[1, 1]
rule_data = json.loads((OUTPUT / "rule_comparison.json").read_text(encoding="utf-8"))["rules"]
counts = [row["detector"]["strict_false_candidates"] for row in rule_data]
bars = axis.bar(["Группы", "1 часть", "2 части", "3 части"], counts)
axis.set_yscale("log")
for bar, count in zip(bars, counts):
axis.text(bar.get_x() + bar.get_width()/2, count*1.15, str(count), ha="center", fontsize=9)
axis.set(ylabel="Ложные строгие проверки", title="Все четыре варианта сохранили 240 пакетов")
figure.savefig(OUTPUT / "detector_distributions.png", dpi=160)
figure.savefig(OUTPUT / "detector_distributions.svg")
plt.close(figure)
return result
def run_functional_tests() -> None:
assert detection.part_threshold(2) < detection.part_threshold(3)
def main() -> None:
result = analyze()
print(json.dumps(result, ensure_ascii=False, indent=2))
print(json.dumps(measure_arrays(), ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()