261 lines
14 KiB
Python
261 lines
14 KiB
Python
"""Lab061: офлайн-интеграция T2 в существующий строгий приёмник."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import hashlib
|
||
import json
|
||
import math
|
||
from pathlib import Path
|
||
import sys
|
||
|
||
if __package__ in (None, ""):
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
||
import matplotlib.pyplot as plt
|
||
import numpy as np
|
||
|
||
from experiments import lab045_hackrf_bpsk_fec as lab045
|
||
from experiments import lab049_bpsk_symbol_rate as lab049
|
||
from experiments import lab052_candidate_detection as detection
|
||
from experiments import lab053_packet_processing as packet
|
||
from experiments import lab053_streaming_receiver as streaming
|
||
from experiments import lab058_abab_symbol_quality as lab058
|
||
from experiments import lab059_preamble_carrier_limits as lab059
|
||
from experiments import lab060_decision_directed_carrier_tracking as lab060
|
||
from protocol import bpsk_carrier_tracking as tracking
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
OUTPUT = ROOT / "data/processed/lab061"
|
||
FREQUENCIES = (-10, -5, -2, -1, -0.5, -0.25, 0, 0.25, 0.5, 1, 2, 5, 10)
|
||
QUALITIES = {"GOOD": 2.0, "MEDIUM": 24.0, "LAB056_B_LIKE": 105.0}
|
||
|
||
|
||
def write_csv(path: Path, rows: list[dict]) -> None:
|
||
if not rows:
|
||
return
|
||
columns = list(rows[0])
|
||
with path.open("w", newline="", encoding="utf-8") as stream:
|
||
writer = csv.DictWriter(stream, columns)
|
||
writer.writeheader()
|
||
writer.writerows(rows)
|
||
|
||
|
||
def full_packet() -> tuple[np.ndarray, bytes, np.ndarray]:
|
||
payload = bytes(range(256))
|
||
plan = lab045.build_transmission_plan(payload)
|
||
raw = np.frombuffer(
|
||
lab049.build_parameterized_tx_cs8(plan, lab049.PhyMode(500_000)), np.int8
|
||
)
|
||
values = raw[::2].astype(np.float32) + 1j * raw[1::2].astype(np.float32)
|
||
expected = lab049.radio.bpsk_modulate(plan.frame_bits)
|
||
return values.astype(np.complex64), payload, expected
|
||
|
||
|
||
def stream_decode(values: np.ndarray, mode: str) -> tuple[list, streaming.StreamingReceiver, list[dict]]:
|
||
observed: list[dict] = []
|
||
receiver = streaming.StreamingReceiver(
|
||
processor=packet.PacketProcessor(carrier_tracking_mode=mode),
|
||
packet_observer=lambda row, outcome: observed.append({
|
||
"reason": outcome.reason,
|
||
"accepted": outcome.accepted,
|
||
"header": bool(outcome.phy_header_pass),
|
||
"fec": bool(outcome.fec_pass),
|
||
"crc": outcome.crc == "PASS",
|
||
"traces": row["carrier_tracking_traces"],
|
||
}),
|
||
)
|
||
output = []
|
||
for start in range(0, len(values), receiver.block_samples):
|
||
output.extend(receiver.push_iq_block(values[start:start + receiver.block_samples]))
|
||
output.extend(receiver.finish())
|
||
return output, receiver, observed
|
||
|
||
|
||
def synthetic_study() -> tuple[list[dict], dict[str, np.ndarray]]:
|
||
clean, payload, expected = full_packet()
|
||
indexes = np.arange(len(clean))
|
||
rows: list[dict] = []
|
||
representative: dict[str, np.ndarray] = {}
|
||
for quality_number, (quality, noise_sigma) in enumerate(QUALITIES.items()):
|
||
for frequency in FREQUENCIES:
|
||
seed = 61_000 + quality_number * 100 + FREQUENCIES.index(frequency)
|
||
random_generator = np.random.default_rng(seed)
|
||
noise = random_generator.normal(0, noise_sigma, len(clean)) + 1j * random_generator.normal(0, noise_sigma, len(clean))
|
||
impaired = clean * np.exp(1j * (0.37 + 2 * np.pi * frequency * indexes / 8_000_000)) + noise
|
||
values = np.concatenate((np.zeros(20_000), impaired, np.zeros(20_000))).astype(np.complex64)
|
||
for mode in ("legacy", "t2"):
|
||
output, receiver, observed = stream_decode(values, mode)
|
||
header = any(row["header"] for row in observed)
|
||
fec = any(row["fec"] for row in observed)
|
||
crc = any(row["crc"] for row in observed)
|
||
traces = [trace for item in observed for trace in item["traces"]]
|
||
full_trace = max(traces, key=lambda item: len(item.corrected)) if traces else None
|
||
measured_evm = None
|
||
if full_trace is not None and len(full_trace.corrected) == len(expected):
|
||
selected = slice(1280, None)
|
||
gain = np.mean(full_trace.corrected[selected] * expected[selected])
|
||
residual = full_trace.corrected[selected] - gain * expected[selected]
|
||
measured_evm = float(np.sqrt(np.mean(abs(residual)**2) / abs(gain)**2))
|
||
rows.append({
|
||
"quality": quality, "noise_sigma": noise_sigma,
|
||
"initial_frequency_error_hz": frequency, "initial_phase_rad": 0.37,
|
||
"seed": seed, "mode": mode, "measured_evm_rms": measured_evm,
|
||
"candidates": receiver.candidate_count,
|
||
"HEADER_ACCEPTED": header, "FEC_STARTED": header,
|
||
"FEC_PASS": fec, "CRC_STARTED": fec, "CRC_PASS": crc,
|
||
"PACKET_OUTPUT": len(output) == 1 and output[0].payload == payload,
|
||
})
|
||
if mode == "t2" and quality == "LAB056_B_LIKE" and frequency == 10 and observed:
|
||
if traces:
|
||
trace = max(traces, key=lambda item: len(item.phase_rad))
|
||
representative = {
|
||
"phase_rad": trace.phase_rad,
|
||
"frequency_hz": trace.frequency_hz,
|
||
"detector_error_rad": trace.detector_error_rad,
|
||
}
|
||
return rows, representative
|
||
|
||
|
||
def decode_known_candidates() -> list[dict]:
|
||
lab054_items, lab047_items, lab054_reader, lab047_reader = lab059._packet_sources()
|
||
rows: list[dict] = []
|
||
for dataset, symbol_rate, items, reader, key in (
|
||
("Lab047", 20_000, lab047_items, lab047_reader, "candidate_offset"),
|
||
("Lab054", 500_000, lab054_items, lab054_reader, "coarse_start"),
|
||
):
|
||
for mode in ("legacy", "t2"):
|
||
processor = packet.PacketProcessor(carrier_tracking_mode=mode, symbol_rate=symbol_rate)
|
||
for number, item in enumerate(items, 1):
|
||
start = int(item[key])
|
||
low = max(0, start - 4_000)
|
||
length = 2_400_000 if symbol_rate == 20_000 else 120_000
|
||
values = reader.read(low, min(length, reader.size - low))
|
||
candidate = detection.offline.CoarseCandidate(start - low, 1.0)
|
||
outcome, detail = processor.decode(values, candidate, number)
|
||
rows.append({
|
||
"dataset": dataset, "frame": number, "mode": mode,
|
||
"HEADER_ACCEPTED": outcome.phy_header_pass,
|
||
"FEC_STARTED": outcome.phy_header_pass,
|
||
"FEC_PASS": outcome.fec_pass,
|
||
"CRC_STARTED": outcome.fec_pass,
|
||
"CRC_PASS": outcome.crc == "PASS", "PACKET_OUTPUT": outcome.accepted,
|
||
"reason": outcome.reason, "initial_frequency_hz": detail.get("cfo_hz"),
|
||
})
|
||
return rows
|
||
|
||
|
||
def lab056_study() -> tuple[list[dict], dict[str, np.ndarray]]:
|
||
rows, trajectories = [], {}
|
||
for item in lab060._lab056_data():
|
||
source = item["estimate"]
|
||
estimate = tracking.P3CarrierEstimate(
|
||
source.frequency_hz, source.gain, source.phase_fit_rmse_rad, source.mean_block_coherence
|
||
)
|
||
result = tracking.track_t2(item["symbols"], item["symbol_rate"], estimate)
|
||
compatible = lab060.TrackResult(
|
||
tracker="T2", corrected=result.corrected,
|
||
hard_decisions=result.hard_decisions, phase_rad=result.phase_rad,
|
||
frequency_hz=result.frequency_hz,
|
||
detector_error_rad=result.detector_error_rad,
|
||
parameters=lab060.loop_parameters("T2", 0.0005),
|
||
initial_frequency_hz=estimate.frequency_hz,
|
||
initial_phase_rad=float(np.angle(estimate.gain)),
|
||
)
|
||
metrics, blocks = lab060.evaluate_tracking(compatible, item["expected"], item["symbol_rate"])
|
||
rows.append({"label": item["label"], "symbol_rate": item["symbol_rate"], **{k: v for k, v in metrics.items() if k != "receiver_lock"}, "stable_lock": metrics["receiver_lock"]["stable_lock"]})
|
||
trajectories[f"{item['label']}_phase_rad"] = result.phase_rad
|
||
trajectories[f"{item['label']}_frequency_hz"] = result.frequency_hz
|
||
return rows, trajectories
|
||
|
||
|
||
def background_study() -> dict:
|
||
source = json.loads((ROOT / "data/processed/lab060/summary.json").read_text(encoding="utf-8"))
|
||
background = source["background"][0]
|
||
start, count = int(background["rx_start"]), int(background["samples"])
|
||
reader = lab058.Cs8File(ROOT / "data/raw/lab054/rx_stream.cs8")
|
||
values = reader.read(start, count)
|
||
results = {}
|
||
for mode in ("legacy", "t2"):
|
||
output, receiver, observed = stream_decode(values, mode)
|
||
results[mode] = {"headers": sum(row["header"] for row in observed), "packets": len(output), "candidates": receiver.candidate_count}
|
||
return results
|
||
|
||
|
||
def saved_tx_control() -> dict:
|
||
reader = lab058.Cs8File(ROOT / "data/raw/lab054/tx_stream.cs8")
|
||
values = reader.read(0, reader.size)
|
||
results = {}
|
||
for mode in ("legacy", "t2"):
|
||
output, receiver, observed = stream_decode(values, mode)
|
||
results[mode] = {"headers": sum(row["header"] for row in observed), "packets": len(output)}
|
||
return results
|
||
|
||
|
||
def robustness() -> list[dict]:
|
||
symbols, expected, unused = lab060.clean_rrc_symbols(500_000)
|
||
estimate = tracking.estimate_p3(symbols[:1280], expected[:1280], 500_000)
|
||
cases = {}
|
||
for name, selected in (("single", slice(2000, 2001)), ("short", slice(2500, 2510)), ("long", slice(3000, 3080))):
|
||
damaged = symbols.astype(np.complex128)
|
||
damaged[selected] *= -1
|
||
cases[name] = damaged
|
||
fade = symbols.astype(np.complex128); fade[3500:3700] *= 0.02; cases["amplitude_drop"] = fade
|
||
noise = symbols.astype(np.complex128); random_generator = np.random.default_rng(61999); noise[4000:4300] = random_generator.normal(size=300) + 1j*random_generator.normal(size=300); cases["noise"] = noise
|
||
rows = []
|
||
for name, values in cases.items():
|
||
result = tracking.track_t2(values, 500_000, estimate)
|
||
rows.append({"case": name, "finite": bool(np.all(np.isfinite(result.corrected)) and np.all(np.isfinite(result.frequency_hz))), "max_abs_frequency_hz": float(np.max(abs(result.frequency_hz)))})
|
||
return rows
|
||
|
||
|
||
def plots(synthetic: list[dict], representative: dict, real: list[dict], background: dict) -> None:
|
||
plt.figure(); plt.plot(np.arange(len(representative["phase_rad"]))/500_000, representative["phase_rad"]); plt.xlabel("Время, с"); plt.ylabel("Фаза T2, рад"); plt.tight_layout(); plt.savefig(OUTPUT/"synthetic_t2_phase.png", dpi=160); plt.close()
|
||
plt.figure(); plt.plot(np.arange(len(representative["frequency_hz"]))/500_000, representative["frequency_hz"]); plt.xlabel("Время, с"); plt.ylabel("Частотное состояние, Гц"); plt.tight_layout(); plt.savefig(OUTPUT/"synthetic_t2_frequency.png", dpi=160); plt.close()
|
||
for mode in ("legacy", "t2"):
|
||
values = [np.mean([r["PACKET_OUTPUT"] for r in synthetic if r["mode"] == mode and r["initial_frequency_error_hz"] == f]) for f in FREQUENCIES]
|
||
plt.plot(FREQUENCIES, values, marker="o", label=mode)
|
||
plt.xlabel("Начальная ошибка частоты, Гц"); plt.ylabel("Доля пакетов"); plt.legend(); plt.tight_layout(); plt.savefig(OUTPUT/"packet_success_by_initial_cfo.png", dpi=160); plt.close()
|
||
for dataset, filename in (("Lab047", "lab047_regression.png"), ("Lab054", "lab054_strict_result.png")):
|
||
values = [sum(r["PACKET_OUTPUT"] for r in real if r["dataset"] == dataset and r["mode"] == mode) for mode in ("legacy", "t2")]
|
||
plt.figure(); plt.bar(("legacy", "t2"), values); plt.ylabel("CRC PASS пакеты"); plt.tight_layout(); plt.savefig(OUTPUT/filename, dpi=160); plt.close()
|
||
plt.figure(); plt.bar(("legacy headers", "legacy packets", "T2 headers", "T2 packets"), (background["legacy"]["headers"], background["legacy"]["packets"], background["t2"]["headers"], background["t2"]["packets"])); plt.ylabel("Количество"); plt.tight_layout(); plt.savefig(OUTPUT/"background_result.png", dpi=160); plt.close()
|
||
|
||
|
||
def main() -> None:
|
||
OUTPUT.mkdir(parents=True, exist_ok=True)
|
||
synthetic, representative = synthetic_study()
|
||
real = decode_known_candidates()
|
||
lab056, lab056_trajectories = lab056_study()
|
||
background = background_study()
|
||
tx_control = saved_tx_control()
|
||
robust = robustness()
|
||
write_csv(OUTPUT/"synthetic_packet_results.csv", synthetic)
|
||
write_csv(OUTPUT/"real_results.csv", real)
|
||
write_csv(OUTPUT/"lab056_results.csv", lab056)
|
||
write_csv(OUTPUT/"robustness_results.csv", robust)
|
||
np.savez_compressed(OUTPUT/"trajectories.npz", **representative, **lab056_trajectories)
|
||
plots(synthetic, representative, real, background)
|
||
by_mode = {mode: {"header_pass_rate": float(np.mean([r["HEADER_ACCEPTED"] for r in synthetic if r["mode"] == mode])), "crc_pass_rate": float(np.mean([r["CRC_PASS"] for r in synthetic if r["mode"] == mode]))} for mode in ("legacy", "t2")}
|
||
by_frequency = {str(f): float(np.mean([r["PACKET_OUTPUT"] for r in synthetic if r["mode"] == "t2" and r["initial_frequency_error_hz"] == f])) for f in FREQUENCIES}
|
||
summary = {
|
||
"lab": "Lab061", "status": "COMPLETED", "scope": "OFFLINE", "rf_tx": 0,
|
||
"hardware_queries": 0, "legacy_default": True, "automatic_fallback": False,
|
||
"t2_parameters": vars(tracking.t2_parameters()),
|
||
"synthetic": by_mode, "t2_packet_success_by_initial_frequency_error": by_frequency,
|
||
"real": {dataset: {mode: sum(r["PACKET_OUTPUT"] for r in real if r["dataset"] == dataset and r["mode"] == mode) for mode in ("legacy", "t2")} for dataset in ("Lab047", "Lab054")},
|
||
"lab056": lab056, "background": background, "saved_tx_control": tx_control,
|
||
"robustness": robust,
|
||
"payload_truth_leak_test": "PASS: strict API and T2 have no reference payload input; causal prefix test PASS",
|
||
"production_decision": "READY_BEHIND_EXPLICIT_MODE",
|
||
"physical_packet": {"bytes": 268, "symbols": 5724, "duration_seconds": len(full_packet()[0])/8_000_000},
|
||
"source_sha256": {str(path.relative_to(ROOT)): hashlib.sha256(path.read_bytes()).hexdigest() for path in (ROOT/"data/raw/lab054/rx_stream.cs8",)},
|
||
}
|
||
(OUTPUT/"summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2, default=lambda value: value.item()), encoding="utf-8")
|
||
print(json.dumps(summary, ensure_ascii=False, indent=2, default=lambda value: value.item()))
|
||
print("RF TX = 0")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|