158 lines
9.0 KiB
Python
158 lines
9.0 KiB
Python
"""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()
|