"""Lab060: офлайн-сопровождение BPSK-несущей по решениям приёмника.""" from __future__ import annotations import csv from concurrent.futures import ThreadPoolExecutor from dataclasses import asdict, dataclass from functools import lru_cache 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 matplotlib.use("Agg") import matplotlib.pyplot as plt import numpy as np try: from numba import njit except ImportError: # pragma: no cover - резерв для расчётной среды без Numba def njit(*_args, **_kwargs): def decorate(function): return function return decorate from experiments import lab054_hardware_control as lab054 from experiments import lab056_abab_energy as lab056 from experiments import lab057_symbol_diagnosis as lab057 from experiments import lab058_abab_symbol_quality as lab058 from experiments import lab059_preamble_carrier_limits as lab059 ROOT = Path(__file__).resolve().parents[1] OUTPUT = ROOT / "data/processed/lab060" TRACKERS = ("T0", "T1", "T2") PREAMBLE_SYMBOLS = 1280 PROBE_SYMBOLS = 1424 BLOCKS = 16 LOST_BLOCK_SER = 0.25 RECOVERED_BLOCK_SER = 0.10 LOSS_CONSECUTIVE_BLOCKS = 2 DETECTOR_CLIP_RAD = math.pi / 3.0 LOCK_SCORE_THRESHOLD = 0.15 DAMPING_FACTOR = 1.0 / math.sqrt(2.0) NORMALIZED_BANDWIDTHS = (0.0002, 0.0005, 0.001, 0.002, 0.004) INITIAL_FREQUENCY_ERRORS_HZ = (-10.0, -5.0, -2.0, -1.0, -0.5, -0.25, 0.0, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0) INITIAL_PHASE_ERRORS_RAD = (-0.35, 0.0, 0.35) QUALITY_LEVELS = (("LAB058_A_EVM", 0.317390), ("INTERMEDIATE_EVM", 0.55), ("LAB058_B_EVM", 0.845350)) TUNING_TRIALS = 1 VALIDATION_TRIALS = 2 TUNING_SEED = 60_100_000 VALIDATION_SEED = 60_200_000 SYNTHETIC_WORKERS = 4 @dataclass(frozen=True) class LoopParameters: """Физически документированные коэффициенты петли на один символ.""" tracker: str normalized_bandwidth: float damping_factor: float phase_coefficient: float frequency_coefficient: float @dataclass class TrackResult: """Полная причинная траектория одного burst.""" tracker: str corrected: np.ndarray hard_decisions: np.ndarray phase_rad: np.ndarray frequency_hz: np.ndarray detector_error_rad: np.ndarray parameters: LoopParameters initial_frequency_hz: float initial_phase_rad: float def serializable(value): """Преобразовать numpy, dataclass и complex в строгий JSON.""" if hasattr(value, "__dataclass_fields__"): return serializable(asdict(value)) if isinstance(value, dict): return {str(key): serializable(item) for key, item in value.items()} if isinstance(value, (list, tuple)): return [serializable(item) for item in value] if isinstance(value, np.ndarray): return serializable(value.tolist()) if isinstance(value, complex): return {"real": float(value.real), "imag": float(value.imag)} if isinstance(value, np.generic): return serializable(value.item()) return value def write_json(path: Path, value) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(serializable(value), ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8", ) def write_csv(path: Path, rows: list[dict]) -> None: fields: list[str] = [] for row in rows: fields.extend(key for key in row if key not in fields) with path.open("w", newline="", encoding="utf-8-sig") as stream: writer = csv.DictWriter(stream, fieldnames=fields) writer.writeheader() for row in rows: writer.writerow({key: _csv_value(row.get(key)) for key in fields}) def _csv_value(value): if isinstance(value, (dict, list, tuple)) or hasattr(value, "__dataclass_fields__"): return json.dumps(serializable(value), ensure_ascii=False, separators=(",", ":")) if isinstance(value, complex): return json.dumps(serializable(value), ensure_ascii=False) return value def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: while block := stream.read(1024 * 1024): digest.update(block) return digest.hexdigest() def normalized_phase_detector(symbol: complex, decision: float | None = None) -> float: """Фазовая ошибка BPSK, не зависящая от амплитуды входного символа.""" hard = (1.0 if symbol.real >= 0.0 else -1.0) if decision is None else float(decision) error = math.atan2(symbol.imag * hard, symbol.real * hard) return float(np.clip(error, -DETECTOR_CLIP_RAD, DETECTOR_CLIP_RAD)) def loop_parameters(tracker: str, normalized_bandwidth: float) -> LoopParameters: """Преобразовать нормированную полосу в коэффициенты на один символ.""" if tracker not in ("T1", "T2"): if tracker == "T0": return LoopParameters("T0", 0.0, DAMPING_FACTOR, 0.0, 0.0) raise ValueError(f"Неизвестный tracker {tracker}") if normalized_bandwidth <= 0.0: raise ValueError("Нормированная полоса должна быть положительной") if tracker == "T1": phase = 1.0 - math.exp(-2.0 * math.pi * normalized_bandwidth) frequency = 0.0 else: theta = normalized_bandwidth / ( DAMPING_FACTOR + 0.25 / DAMPING_FACTOR ) denominator = 1.0 + 2.0 * DAMPING_FACTOR * theta + theta * theta phase = 4.0 * DAMPING_FACTOR * theta / denominator frequency = 4.0 * theta * theta / denominator return LoopParameters( tracker, float(normalized_bandwidth), DAMPING_FACTOR, float(phase), float(frequency), ) @njit(cache=False, nogil=True) def _track_core( values: np.ndarray, symbol_rate: float, initial_frequency_hz: float, initial_phase_rad: float, amplitude: float, start_index: int, phase_coefficient: float, frequency_coefficient: float, detector_clip_rad: float, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]: count = len(values) corrected = np.empty(count, dtype=np.complex128) decisions = np.empty(count, dtype=np.int8) phases = np.empty(count, dtype=np.float64) frequencies = np.empty(count, dtype=np.float64) detector = np.empty(count, dtype=np.float64) omega = 2.0 * math.pi * initial_frequency_hz / symbol_rate phase = initial_phase_rad + omega * (start_index + 0.5) safe_amplitude = max(amplitude, 1e-30) for index in range(count): if index < start_index: current_phase = initial_phase_rad + omega * (index + 0.5) else: current_phase = phase cosine = math.cos(current_phase) sine = math.sin(current_phase) real = (values[index].real * cosine + values[index].imag * sine) / safe_amplitude imag = (values[index].imag * cosine - values[index].real * sine) / safe_amplitude hard = 1 if real >= 0.0 else -1 error = math.atan2(imag * hard, real * hard) if error > detector_clip_rad: error = detector_clip_rad elif error < -detector_clip_rad: error = -detector_clip_rad corrected[index] = real + 1j * imag decisions[index] = hard phases[index] = current_phase frequencies[index] = omega * symbol_rate / (2.0 * math.pi) detector[index] = error if index >= start_index: phase = phase + omega + phase_coefficient * error omega = omega + frequency_coefficient * error return corrected, decisions, phases, frequencies, detector @njit(cache=False, nogil=True) def _track_summary_core( values: np.ndarray, expected: np.ndarray, symbol_rate: float, initial_frequency_hz: float, initial_phase_rad: float, start_index: int, phase_coefficient: float, frequency_coefficient: float, detector_clip_rad: float, ) -> tuple[int, np.ndarray, float, float]: block_errors = np.zeros(BLOCKS, dtype=np.int64) block_sizes = np.zeros(BLOCKS, dtype=np.int64) omega = 2.0 * math.pi * initial_frequency_hz / symbol_rate phase = initial_phase_rad + omega * (start_index + 0.5) errors = 0 detector_score = 0.0 payload_count = len(values) - start_index for index in range(start_index, len(values)): cosine = math.cos(phase) sine = math.sin(phase) real = values[index].real * cosine + values[index].imag * sine imag = values[index].imag * cosine - values[index].real * sine hard = 1 if real >= 0.0 else -1 error = math.atan2(imag * hard, real * hard) if error > detector_clip_rad: error = detector_clip_rad elif error < -detector_clip_rad: error = -detector_clip_rad wrong = hard != (1 if expected[index] >= 0.0 else -1) block = min(BLOCKS - 1, (index - start_index) * BLOCKS // payload_count) if wrong: errors += 1 block_errors[block] += 1 block_sizes[block] += 1 detector_score += math.cos(2.0 * error) phase = phase + omega + phase_coefficient * error omega = omega + frequency_coefficient * error block_ser = block_errors.astype(np.float64) / block_sizes final_frequency = omega * symbol_rate / (2.0 * math.pi) return errors, block_ser, final_frequency, detector_score / payload_count def track_symbols( symbols: np.ndarray, symbol_rate: int, initial_frequency_hz: float, initial_gain: complex, tracker: str, parameters: LoopParameters, ) -> TrackResult: """Причинно обработать burst; после преамбулы используются только решения RX.""" values = np.asarray(symbols, dtype=np.complex128) if tracker == "T0": times = (np.arange(len(values), dtype=np.float64) + 0.5) / symbol_rate phases = np.angle(initial_gain) + 2.0 * np.pi * initial_frequency_hz * times corrected = values * np.exp(-1j * phases) / max(abs(initial_gain), 1e-30) decisions = np.where(corrected.real >= 0.0, 1, -1).astype(np.int8) detector = np.angle(corrected * decisions) frequencies = np.full(len(values), initial_frequency_hz, dtype=np.float64) else: corrected, decisions, phases, frequencies, detector = _track_core( values, float(symbol_rate), float(initial_frequency_hz), float(np.angle(initial_gain)), float(abs(initial_gain)), PREAMBLE_SYMBOLS, parameters.phase_coefficient, parameters.frequency_coefficient, DETECTOR_CLIP_RAD, ) return TrackResult( tracker, corrected, decisions, phases, frequencies, detector, parameters, float(initial_frequency_hz), float(np.angle(initial_gain)), ) def _consecutive_true(values: np.ndarray, count: int) -> bool: run = 0 for value in values: run = run + 1 if value else 0 if run >= count: return True return False def loss_metrics(block_ser: np.ndarray, block_seconds: float) -> dict: """CRC-независимый критерий: два соседних блока с SER не менее 25%.""" values = np.asarray(block_ser, dtype=np.float64) bad = values >= LOST_BLOCK_SER loss = _consecutive_true(bad, LOSS_CONSECUTIVE_BLOCKS) longest = current = 0 for value in bad: current = current + 1 if value else 0 longest = max(longest, current) recovered = False if loss: first = next(index for index in range(len(bad) - 1) if bad[index] and bad[index + 1]) recovered = _consecutive_true(values[first + 2:] <= RECOVERED_BLOCK_SER, 2) return { "loss_of_lock": bool(loss), "bad_blocks": int(np.count_nonzero(bad)), "longest_bad_state_seconds": float(longest * block_seconds), "self_recovered": bool(recovered), } def receiver_lock_metrics(detector_errors: np.ndarray) -> dict: """Оракул-независимый индикатор осевой BPSK-когерентности по 16 блокам.""" selected = np.asarray(detector_errors)[PREAMBLE_SYMBOLS:] scores = np.asarray([ float(np.mean(np.cos(2.0 * block))) for block in np.array_split(selected, BLOCKS) ]) locked = scores >= LOCK_SCORE_THRESHOLD return { "block_scores": scores, "locked_blocks": int(np.count_nonzero(locked)), "stable_lock": bool(np.count_nonzero(locked) >= 12 and not _consecutive_true(~locked, 3)), "mean_score": float(np.mean(scores)), } def evaluate_tracking(result: TrackResult, expected: np.ndarray, symbol_rate: int) -> tuple[dict, list[dict]]: """Подключить TX truth только после завершения причинной траектории.""" truth = np.asarray(expected, dtype=np.float64) selected = np.arange(len(truth)) >= PREAMBLE_SYMBOLS corrected = result.corrected[selected] expected_payload = truth[selected] decisions = result.hard_decisions[selected] symbol_errors = decisions != np.where(expected_payload >= 0.0, 1, -1) gain = complex(np.mean(corrected * expected_payload)) residual = corrected - gain * expected_payload matched_power = abs(gain) ** 2 residual_power = float(np.mean(abs(residual) ** 2)) evm = float(np.sqrt(residual_power / max(matched_power, 1e-30))) ratio = float(10.0 * np.log10(max(matched_power, 1e-30) / max(residual_power, 1e-30))) blocks = [] block_indexes = np.array_split(np.arange(PREAMBLE_SYMBOLS, len(truth)), BLOCKS) for number, indexes in enumerate(block_indexes, 1): block_error = result.hard_decisions[indexes] != np.where(truth[indexes] >= 0.0, 1, -1) reference_phase = np.angle(result.corrected[indexes] * truth[indexes]) blocks.append({ "block": number, "start_seconds": float(indexes[0] / symbol_rate), "end_seconds": float((indexes[-1] + 1) / symbol_rate), "symbols": len(indexes), "symbol_errors": int(np.count_nonzero(block_error)), "symbol_error_rate": float(np.mean(block_error)), "mean_phase_error_rad": float(np.angle(np.mean(np.exp(1j * reference_phase)))), "mean_frequency_hz": float(np.mean(result.frequency_hz[indexes])), "mean_detector_error_rad": float(np.mean(result.detector_error_rad[indexes])), }) block_ser = np.asarray([row["symbol_error_rate"] for row in blocks]) loss = loss_metrics(block_ser, (len(truth) - PREAMBLE_SYMBOLS) / symbol_rate / BLOCKS) lock = receiver_lock_metrics(result.detector_error_rad) phase_centers = np.unwrap(np.asarray([row["mean_phase_error_rad"] for row in blocks])) slips = int(np.count_nonzero(abs(np.diff(phase_centers)) > math.pi / 2.0)) metrics = { "symbol_errors": int(np.count_nonzero(symbol_errors)), "symbols": int(np.count_nonzero(selected)), "symbol_error_rate": float(np.mean(symbol_errors)), "evm_rms": evm, "matched_component_power": float(matched_power), "residual_power": residual_power, "matched_residual_db": ratio, "final_frequency_hz": float(result.frequency_hz[-1]), "large_phase_slips": slips, **loss, "receiver_lock": lock, } return metrics, blocks @lru_cache(maxsize=2) def clean_rrc_symbols(symbol_rate: int) -> tuple[np.ndarray, np.ndarray, dict]: """Полный чистый burst через действующие генератор, Fs, quantization и RX RRC.""" interleaved = lab056.build_burst(symbol_rate) raw = interleaved[::2].astype(np.float32) + 1j * interleaved[1::2].astype(np.float32) expected = lab058.expected_symbols(symbol_rate) frontend = lab057.Frontend(symbol_rate) stock = frontend.extract(raw, len(expected)) if not stock.passed: raise RuntimeError(f"Synthetic clean {symbol_rate}: {stock.reason}") matched = frontend.filter(raw) symbols = lab059._extract_symbols( matched, frontend.nominal, stock.timing_offset_samples, len(expected), frontend.samples_per_symbol, ) gain = complex(np.mean(symbols * expected)) normalized = symbols / gain metadata = { "sample_rate_hz": lab059.SAMPLE_RATE, "symbol_rate": symbol_rate, "samples_per_symbol": frontend.samples_per_symbol, "rrc_taps": len(frontend.taps), "symbols": len(expected), "clean_evm_rms": float(np.sqrt(np.mean(abs(normalized - expected) ** 2))), } return normalized.astype(np.complex128), expected.astype(np.float64), metadata def synthetic_received( symbol_rate: int, evm_rms: float, seed: int, true_frequency_hz: float, true_phase_rad: float, ) -> tuple[np.ndarray, np.ndarray]: """Добавить фазу/CFO и AWGN к чистым RRC-точкам с фиксированным seed.""" clean, expected, _metadata = clean_rrc_symbols(symbol_rate) times = (np.arange(len(clean), dtype=np.float64) + 0.5) / symbol_rate random_generator = np.random.default_rng(seed) noise = evm_rms / math.sqrt(2.0) * ( random_generator.normal(size=len(clean)) + 1j * random_generator.normal(size=len(clean)) ) values = clean * np.exp(1j * (true_phase_rad + 2.0 * np.pi * true_frequency_hz * times)) + noise return values.astype(np.complex128), expected def synthetic_case( values: np.ndarray, expected: np.ndarray, symbol_rate: int, true_frequency_hz: float, true_phase_rad: float, initial_frequency_error_hz: float, initial_phase_error_rad: float, tracker: str, parameters: LoopParameters, ) -> dict: initial_frequency = true_frequency_hz + initial_frequency_error_hz initial_phase = true_phase_rad + initial_phase_error_rad errors, block_ser, final_frequency, detector_score = _track_summary_core( values, expected, float(symbol_rate), initial_frequency, initial_phase, PREAMBLE_SYMBOLS, parameters.phase_coefficient, parameters.frequency_coefficient, DETECTOR_CLIP_RAD, ) block_seconds = (len(values) - PREAMBLE_SYMBOLS) / symbol_rate / BLOCKS loss = loss_metrics(block_ser, block_seconds) return { "symbol_errors": int(errors), "symbols": len(values) - PREAMBLE_SYMBOLS, "symbol_error_rate": float(errors / (len(values) - PREAMBLE_SYMBOLS)), "block_symbol_error_rates": block_ser, "final_frequency_hz": float(final_frequency), "final_frequency_error_hz": float(final_frequency - true_frequency_hz), "mean_detector_lock_score": float(detector_score), "catastrophic_failure": bool(loss["loss_of_lock"]), **loss, } def tune_loop_parameters() -> tuple[dict[str, LoopParameters], list[dict]]: """Выбрать T1/T2 только по отдельной synthetic 500K/B выборке.""" symbol_rate = 500_000 true_frequency = 2.75 candidate_results = { (tracker, bandwidth): [] for tracker in ("T1", "T2") for bandwidth in NORMALIZED_BANDWIDTHS } jobs = [ (frequency_index, frequency_error, phase_index, phase_error, trial) for frequency_index, frequency_error in enumerate(INITIAL_FREQUENCY_ERRORS_HZ) for phase_index, phase_error in enumerate(INITIAL_PHASE_ERRORS_RAD) for trial in range(TUNING_TRIALS) ] clean_rrc_symbols(symbol_rate) warm_values, warm_expected = synthetic_received( symbol_rate, 0.845350, TUNING_SEED, true_frequency, -0.8 ) synthetic_case( warm_values[:PREAMBLE_SYMBOLS + BLOCKS], warm_expected[:PREAMBLE_SYMBOLS + BLOCKS], symbol_rate, true_frequency, -0.8, 0.0, 0.0, "T1", loop_parameters("T1", NORMALIZED_BANDWIDTHS[0]), ) def evaluate_job(job): frequency_index, frequency_error, phase_index, phase_error, trial = job seed = TUNING_SEED + frequency_index * 1_000 + phase_index * 100 + trial true_phase = -0.8 + 0.31 * trial values, expected = synthetic_received( symbol_rate, 0.845350, seed, true_frequency, true_phase ) output = [] for tracker in ("T1", "T2"): for bandwidth in NORMALIZED_BANDWIDTHS: output.append((tracker, bandwidth, synthetic_case( values, expected, symbol_rate, true_frequency, true_phase, frequency_error, phase_error, tracker, loop_parameters(tracker, bandwidth), ))) return output with ThreadPoolExecutor(max_workers=SYNTHETIC_WORKERS) as executor: for output in executor.map(evaluate_job, jobs): for tracker, bandwidth, result in output: candidate_results[(tracker, bandwidth)].append(result) rows: list[dict] = [] for tracker in ("T1", "T2"): for bandwidth in NORMALIZED_BANDWIDTHS: parameters = loop_parameters(tracker, bandwidth) results = candidate_results[(tracker, bandwidth)] row = { "tracker": tracker, **asdict(parameters), "cases": len(results), "catastrophic_failure_rate": float(np.mean([ item["catastrophic_failure"] for item in results ])), "loss_of_lock_rate": float(np.mean([item["loss_of_lock"] for item in results])), "mean_symbol_error_rate": float(np.mean([ item["symbol_error_rate"] for item in results ])), "p95_symbol_error_rate": float(np.percentile([ item["symbol_error_rate"] for item in results ], 95)), "mean_abs_final_frequency_error_hz": float(np.mean([ abs(item["final_frequency_error_hz"]) for item in results ])), "selection_data": "synthetic_500k_lab058_b_evm_only", } rows.append(row) selected: dict[str, LoopParameters] = {"T0": loop_parameters("T0", 0.0)} for tracker in ("T1", "T2"): candidates = [row for row in rows if row["tracker"] == tracker] winner = min(candidates, key=lambda row: ( row["catastrophic_failure_rate"], row["mean_symbol_error_rate"], row["normalized_bandwidth"], )) selected[tracker] = loop_parameters(tracker, winner["normalized_bandwidth"]) for row in candidates: row["selected"] = row is winner return selected, rows def synthetic_validation(parameters: dict[str, LoopParameters]) -> tuple[list[dict], list[dict]]: """Независимая fixed-seed проверка после заморозки параметров.""" cases: list[dict] = [] true_frequency = 2.75 jobs = [ (symbol_rate, quality_index, quality_name, evm_rms, frequency_index, frequency_error, phase_index, phase_error, trial) for symbol_rate in (20_000, 500_000) for quality_index, (quality_name, evm_rms) in enumerate(QUALITY_LEVELS) for frequency_index, frequency_error in enumerate(INITIAL_FREQUENCY_ERRORS_HZ) for phase_index, phase_error in enumerate(INITIAL_PHASE_ERRORS_RAD) for trial in range(VALIDATION_TRIALS) ] clean_rrc_symbols(20_000) clean_rrc_symbols(500_000) def evaluate_job(job): (symbol_rate, quality_index, quality_name, evm_rms, frequency_index, frequency_error, phase_index, phase_error, trial) = job seed = (VALIDATION_SEED + symbol_rate + quality_index * 1_000_000 + frequency_index * 10_000 + phase_index * 100 + trial) true_phase = 0.63 - 0.27 * trial values, expected = synthetic_received( symbol_rate, evm_rms, seed, true_frequency, true_phase ) output = [] for tracker in TRACKERS: result = synthetic_case( values, expected, symbol_rate, true_frequency, true_phase, frequency_error, phase_error, tracker, parameters[tracker], ) output.append({ "symbol_rate": symbol_rate, "quality": quality_name, "target_evm_rms": evm_rms, "initial_frequency_error_hz": frequency_error, "initial_phase_error_rad": phase_error, "trial": trial, "seed": seed, "tracker": tracker, **{key: value for key, value in result.items() if key != "block_symbol_error_rates"}, }) return output with ThreadPoolExecutor(max_workers=SYNTHETIC_WORKERS) as executor: for output in executor.map(evaluate_job, jobs): cases.extend(output) aggregate: list[dict] = [] for symbol_rate in (20_000, 500_000): for quality_name, evm_rms in QUALITY_LEVELS: for tracker in TRACKERS: selected = [row for row in cases if row["symbol_rate"] == symbol_rate and row["quality"] == quality_name and row["tracker"] == tracker] aggregate.append({ "symbol_rate": symbol_rate, "quality": quality_name, "target_evm_rms": evm_rms, "tracker": tracker, "cases": len(selected), "catastrophic_failure_rate": float(np.mean([ row["catastrophic_failure"] for row in selected ])), "loss_of_lock_rate": float(np.mean([row["loss_of_lock"] for row in selected])), "self_recovery_rate_after_loss": ( float(np.mean([row["self_recovered"] for row in selected if row["loss_of_lock"]])) if any(row["loss_of_lock"] for row in selected) else 0.0 ), "mean_symbol_error_rate": float(np.mean([ row["symbol_error_rate"] for row in selected ])), "p95_symbol_error_rate": float(np.percentile([ row["symbol_error_rate"] for row in selected ], 95)), "mean_abs_final_frequency_error_hz": float(np.mean([ abs(row["final_frequency_error_hz"]) for row in selected ])), }) by_frequency: list[dict] = [] for frequency_error in INITIAL_FREQUENCY_ERRORS_HZ: for tracker in TRACKERS: selected = [row for row in cases if row["symbol_rate"] == 500_000 and row["quality"] == "LAB058_B_EVM" and row["initial_frequency_error_hz"] == frequency_error and row["tracker"] == tracker] by_frequency.append({ "symbol_rate": 500_000, "quality": "LAB058_B_EVM", "initial_frequency_error_hz": frequency_error, "tracker": tracker, "cases": len(selected), "catastrophic_failure_rate": float(np.mean([ row["catastrophic_failure"] for row in selected ])), "loss_of_lock_rate": float(np.mean([row["loss_of_lock"] for row in selected])), "mean_symbol_error_rate": float(np.mean([ row["symbol_error_rate"] for row in selected ])), }) return aggregate, by_frequency def _lab056_data() -> list[dict]: inputs, oracle_frequency = lab059._lab056_inputs() oracle_summary = json.loads( (ROOT / "data/processed/lab058/summary.json").read_text(encoding="utf-8") ) oracle_by_label = {row["label"]: row["oracle"] for row in oracle_summary["segments"]} reader = lab058.Cs8File(ROOT / "data/raw/lab056/rx_stream.cs8") rows = [] for item in inputs: symbol_rate = int(item["symbol_rate"]) expected = lab058.expected_symbols(symbol_rate) raw = reader.read(int(item["rx_start"]), int(item["samples"])) frontend = lab057.Frontend(symbol_rate) stock = frontend.extract(raw, len(expected)) if not stock.passed: raise RuntimeError(f"Lab056 {item['label']}: {stock.reason}") matched = frontend.filter(raw) symbols = lab059._extract_symbols( matched, frontend.nominal, stock.timing_offset_samples, len(expected), frontend.samples_per_symbol, ) estimate = lab059.real_carrier_estimates( raw, frontend, stock.timing_offset_samples, symbols, symbol_rate, ("P3",), )["P3"] rows.append({ **item, "expected": expected, "symbols": symbols, "estimate": estimate, "oracle_frequency_hz": oracle_frequency[item["label"]], "oracle": oracle_by_label[item["label"]], }) return rows def analyze_lab056( parameters: dict[str, LoopParameters], ) -> tuple[list[dict], list[dict], dict[str, np.ndarray]]: """Применить замороженные synthetic-параметры к A1/B1/A2/B2.""" rows: list[dict] = [] blocks: list[dict] = [] trajectories: dict[str, np.ndarray] = {} for item in _lab056_data(): for tracker in TRACKERS: result = track_symbols( item["symbols"], item["symbol_rate"], item["estimate"].frequency_hz, item["estimate"].gain, tracker, parameters[tracker], ) metrics, block_rows = evaluate_tracking(result, item["expected"], item["symbol_rate"]) key = f"{item['label']}_{tracker}" trajectories[f"{key}_phase_rad"] = result.phase_rad trajectories[f"{key}_frequency_hz"] = result.frequency_hz trajectories[f"{key}_detector_error_rad"] = result.detector_error_rad trajectories[f"{key}_hard_decisions"] = result.hard_decisions row = { "dataset": "Lab056", "label": item["label"], "symbol_rate": item["symbol_rate"], "tracker": tracker, "initial_frequency_hz": item["estimate"].frequency_hz, "initial_phase_rad": float(np.angle(item["estimate"].gain)), "diagnostic_reference_frequency_hz": item["oracle_frequency_hz"], "diagnostic_reference_phase_rad": float(math.atan2( item["oracle"]["h"]["imag"], item["oracle"]["h"]["real"] )), "parameters": parameters[tracker], **metrics, } rows.append(row) blocks.extend({"dataset": "Lab056", "label": item["label"], "tracker": tracker, **block} for block in block_rows) t0_b = [row["symbol_error_rate"] for row in rows if row["tracker"] == "T0" and row["label"].startswith("B")] if not np.isclose(np.mean(t0_b), 0.036169897151898736, atol=1e-12): raise AssertionError("T0 не воспроизвёл зафиксированный P3 baseline Lab059") return rows, blocks, trajectories def _decode_tracker_probe( symbols: np.ndarray, symbol_rate: int, estimate: lab059.CarrierEstimate, tracker: str, parameters: LoopParameters, ) -> tuple[TrackResult, dict]: result = track_symbols( symbols, symbol_rate, estimate.frequency_hz, estimate.gain, tracker, parameters, ) header = lab057.physical.decode_phy_header_soft( result.corrected.real[PREAMBLE_SYMBOLS:PROBE_SYMBOLS] ) return result, header def analyze_packet_frame( reader: lab058.Cs8File, start: int, symbol_rate: int, dataset: str, frame_number: int, expected_header: np.ndarray, parameters: dict[str, LoopParameters], ) -> list[dict]: """Декодировать кадр без знания его длины до принятия PHY header.""" frontend = lab057.Frontend(symbol_rate) probe_raw = reader.read(start, (PROBE_SYMBOLS + 42) * frontend.samples_per_symbol) stock = frontend.extract(probe_raw, PROBE_SYMBOLS) if not stock.passed: raise RuntimeError(f"{dataset} frame {frame_number}: {stock.reason}") probe_symbols = lab059._extract_symbols( frontend.filter(probe_raw), frontend.nominal, stock.timing_offset_samples, PROBE_SYMBOLS, frontend.samples_per_symbol, ) estimate = lab059.real_carrier_estimates( probe_raw, frontend, stock.timing_offset_samples, probe_symbols, symbol_rate, ("P3",), )["P3"] probes = { tracker: _decode_tracker_probe( probe_symbols, symbol_rate, estimate, tracker, parameters[tracker] ) for tracker in TRACKERS } full_cache: dict[int, np.ndarray] = {} rows = [] for tracker in TRACKERS: probe_result, header = probes[tracker] result = probe_result if header["passed"]: packet_bytes = int(header["packet_bytes"]) count = (PREAMBLE_SYMBOLS + 144 + lab057.physical.conv.encoded_bit_count(packet_bytes * 8)) if count not in full_cache: raw = reader.read(start, (count + 42) * frontend.samples_per_symbol) full_cache[count] = lab059._extract_symbols( frontend.filter(raw), frontend.nominal, stock.timing_offset_samples, count, frontend.samples_per_symbol, ) result = track_symbols( full_cache[count], symbol_rate, estimate.frequency_hz, estimate.gain, tracker, parameters[tracker], ) decoded, _packet = lab057.decode_body(result.corrected.real[PREAMBLE_SYMBOLS:]) decisions = probe_result.hard_decisions voted = probe_result.corrected.real[PREAMBLE_SYMBOLS:PROBE_SYMBOLS].reshape(48, 3).sum(axis=1) >= 0 expected_bool = expected_header.astype(bool) lock = receiver_lock_metrics(result.detector_error_rad) rows.append({ "dataset": dataset, "frame": frame_number, "tracker": tracker, "symbol_rate": symbol_rate, "initial_frequency_hz": estimate.frequency_hz, "final_frequency_hz": float(result.frequency_hz[-1]), "preamble_errors": int(np.count_nonzero( decisions[:PREAMBLE_SYMBOLS] != np.where(lab057.PREAMBLE, 1, -1) )), "header_raw_errors": int(np.count_nonzero( decisions[PREAMBLE_SYMBOLS:PROBE_SYMBOLS] != np.where(expected_bool, 1, -1) )), "header_voted_errors": int(np.count_nonzero(voted != expected_bool[::3])), "HEADER_ACCEPTED": bool(decoded["HEADER_ACCEPTED"]), "FEC_STARTED": bool(decoded["FEC_STARTED"]), "FEC_PASS": decoded["FEC_RESULT"] == "PASS", "CRC_STARTED": bool(decoded["CRC_STARTED"]), "CRC_PASS": decoded["CRC_RESULT"] == lab057.physical.CRC_PASS, "PACKET_OUTPUT": bool(decoded["PACKET_OUTPUT"]), "phase_detector_mean": float(np.mean(result.detector_error_rad[PREAMBLE_SYMBOLS:])), "phase_detector_rms": float(np.sqrt(np.mean( result.detector_error_rad[PREAMBLE_SYMBOLS:] ** 2 ))), "receiver_lock": lock, }) return rows def analyze_packets(parameters: dict[str, LoopParameters]) -> list[dict]: lab054_items, lab047_items, lab054_reader, lab047_reader = lab059._packet_sources() rows: list[dict] = [] plans = lab054.plans() for item in lab054_items: frame = int(item["true_start_frame"]) rows.extend(analyze_packet_frame( lab054_reader, int(item["refined_start"]), 500_000, "Lab054", frame, plans[frame - 1].plan.phy_header_bits, parameters, )) expected_header = lab057.physical.encode_phy_header(268) for frame, item in enumerate(lab047_items, 1): rows.extend(analyze_packet_frame( lab047_reader, int(item["refined_start_sample"]), 20_000, "Lab047", frame, expected_header, parameters, )) return rows def summarize_packets(rows: list[dict]) -> dict: summary = {} for dataset, frames in (("Lab054", 16), ("Lab047", 36)): summary[dataset] = {} for tracker in TRACKERS: selected = [row for row in rows if row["dataset"] == dataset and row["tracker"] == tracker] summary[dataset][tracker] = { "frames": len(selected), "preamble_errors": sum(row["preamble_errors"] for row in selected), "header_raw_errors": sum(row["header_raw_errors"] for row in selected), "header_voted_errors": sum(row["header_voted_errors"] for row in selected), "header_accepted": sum(row["HEADER_ACCEPTED"] for row in selected), "fec_started": sum(row["FEC_STARTED"] for row in selected), "fec_pass": sum(row["FEC_PASS"] for row in selected), "crc_started": sum(row["CRC_STARTED"] for row in selected), "crc_pass": sum(row["CRC_PASS"] for row in selected), "packet_output": sum(row["PACKET_OUTPUT"] for row in selected), "expected_frames": frames, } return summary def analyze_background(parameters: dict[str, LoopParameters]) -> list[dict]: """Проверить, не превращает ли DD-петля фон Lab056 в ложный BPSK lock.""" metadata = json.loads((ROOT / "data/raw/lab056/tx_metadata.json").read_text(encoding="utf-8")) analysis = json.loads((ROOT / "data/processed/lab056/rf_analysis.json").read_text(encoding="utf-8")) count = int(metadata["segments"][1]["samples"]) start = int(analysis["alignment"]["start"] - count - 100_000) reader = lab058.Cs8File(ROOT / "data/raw/lab056/rx_stream.cs8") raw = reader.read(start, count) symbol_rate = 500_000 expected_count = len(lab058.expected_symbols(symbol_rate)) frontend = lab057.Frontend(symbol_rate) stock = frontend.extract(raw, expected_count) if not stock.passed: raise RuntimeError(f"Background extract: {stock.reason}") symbols = lab059._extract_symbols( frontend.filter(raw), frontend.nominal, stock.timing_offset_samples, expected_count, frontend.samples_per_symbol, ) estimate = lab059.real_carrier_estimates( raw, frontend, stock.timing_offset_samples, symbols, symbol_rate, ("P3",), )["P3"] rows = [] for tracker in ("T1", "T2"): result, header = _decode_tracker_probe( symbols, symbol_rate, estimate, tracker, parameters[tracker] ) decoded, _packet = lab057.decode_body(result.corrected.real[PREAMBLE_SYMBOLS:PROBE_SYMBOLS]) lock = receiver_lock_metrics(result.detector_error_rad) rows.append({ "tracker": tracker, "rx_start": start, "samples": count, "initial_frequency_hz": estimate.frequency_hz, "final_frequency_hz": float(result.frequency_hz[-1]), "phase_detector_mean": float(np.mean(result.detector_error_rad[PREAMBLE_SYMBOLS:])), "phase_detector_rms": float(np.sqrt(np.mean( result.detector_error_rad[PREAMBLE_SYMBOLS:] ** 2 ))), "receiver_lock": lock, "false_headers": int(bool(header["passed"])), "false_packets": int(bool(decoded["PACKET_OUTPUT"])), }) return rows def payload_truth_leak_control(parameters: dict[str, LoopParameters]) -> dict: """Изменить только диагностический TX reference и сравнить траектории.""" values, expected = synthetic_received(20_000, 0.55, 60_060, 2.75, 0.4) changed_reference = expected.copy() changed_reference[PREAMBLE_SYMBOLS:] *= -1.0 rows = [] for tracker in ("T1", "T2"): first = track_symbols( values, 20_000, 4.75, np.exp(0.4j), tracker, parameters[tracker] ) first_metrics, _ = evaluate_tracking(first, expected, 20_000) second = track_symbols( values, 20_000, 4.75, np.exp(0.4j), tracker, parameters[tracker] ) second_metrics, _ = evaluate_tracking(second, changed_reference, 20_000) trajectory_equal = ( np.array_equal(first.phase_rad, second.phase_rad) and np.array_equal(first.frequency_hz, second.frequency_hz) and np.array_equal(first.hard_decisions, second.hard_decisions) ) rows.append({ "tracker": tracker, "phase_trajectory_equal": np.array_equal(first.phase_rad, second.phase_rad), "frequency_trajectory_equal": np.array_equal(first.frequency_hz, second.frequency_hz), "hard_decisions_equal": np.array_equal(first.hard_decisions, second.hard_decisions), "trajectory_equal": trajectory_equal, "original_diagnostic_symbol_error_rate": first_metrics["symbol_error_rate"], "changed_reference_diagnostic_symbol_error_rate": second_metrics["symbol_error_rate"], "diagnostic_metric_changed": ( first_metrics["symbol_error_rate"] != second_metrics["symbol_error_rate"] ), }) return { "receiver_inputs": [ "current complex RX symbol", "own hard decision", "previous loop state" ], "changed_only": "diagnostic TX payload reference after preamble", "rows": rows, "passed": all(row["trajectory_equal"] and row["diagnostic_metric_changed"] for row in rows), } def integrity_snapshot() -> dict: paths = [ ROOT / "data/raw/lab056/rx_stream.cs8", ROOT / "data/raw/lab056/tx_abab.cs8", ROOT / "data/raw/lab054/rx_stream.cs8", ROOT / "data/raw/lab054/tx_stream.cs8", ] _lab054_items, _lab047_items, _reader054, reader047 = lab059._packet_sources() paths.append(reader047.path) return {str(path): {"bytes": path.stat().st_size, "sha256": sha256(path)} for path in paths} def choose_verdict( real_rows: list[dict], synthetic_rows: list[dict], packet_summary: dict, background_rows: list[dict], ) -> tuple[str, str, str, str]: b_quality = {row["tracker"]: row for row in synthetic_rows if row["symbol_rate"] == 500_000 and row["quality"] == "LAB058_B_EVM"} real_b = {tracker: float(np.mean([ row["symbol_error_rate"] for row in real_rows if row["tracker"] == tracker and row["label"].startswith("B") ])) for tracker in TRACKERS} candidates = ("T1", "T2") best = min(candidates, key=lambda tracker: ( b_quality[tracker]["catastrophic_failure_rate"], real_b[tracker] )) passes = ( b_quality[best]["catastrophic_failure_rate"] < 0.5 * b_quality["T0"]["catastrophic_failure_rate"] and packet_summary["Lab047"][best]["packet_output"] == 36 and all(row["false_packets"] == 0 for row in background_rows) and real_b[best] <= 0.06 ) if passes: return "SUCCESS", best, ( "существенно снижены synthetic-срывы, Lab047 и background сохранены, " "реальный B остаётся в пределах инженерной цели" ), "подготовить отдельное решение о внедрении tracker в штатный receiver" helped = b_quality[best]["catastrophic_failure_rate"] < b_quality["T0"]["catastrophic_failure_rate"] if helped: return "PARTIAL", best, "сопровождение помогает, но критерии устойчивости выполнены не все", ( "рассмотреть BPSK Costas loop отдельной офлайн-лабораторией" ) return "FAIL", best, "decision-directed сопровождение не снизило число срывов", ( "рассмотреть BPSK Costas loop отдельной офлайн-лабораторией" ) def make_plots( real_rows: list[dict], block_rows: list[dict], trajectories: dict[str, np.ndarray], synthetic_by_frequency: list[dict], packet_summary: dict, ) -> None: colors = {"T0": "#6c757d", "T1": "#2878b5", "T2": "#e17c05"} for label in ("B1", "B2"): figure, axis = plt.subplots(figsize=(10, 5), constrained_layout=True) count = len(trajectories[f"{label}_T0_phase_rad"]) indexes = np.arange(0, count, max(1, count // 2500)) time_ms = (indexes + 0.5) / 500_000 * 1000.0 reference_row = next(row for row in real_rows if row["label"] == label) reference = (reference_row["diagnostic_reference_phase_rad"] + 2.0 * np.pi * reference_row["diagnostic_reference_frequency_hz"] * (indexes + 0.5) / 500_000) axis.plot(time_ms, np.unwrap(reference), color="black", linestyle="--", linewidth=1.2, label="diagnostic reference") for tracker in TRACKERS: axis.plot(time_ms, np.unwrap(trajectories[f"{label}_{tracker}_phase_rad"][indexes]), color=colors[tracker], linewidth=1.0, label=tracker) axis.axvline(PREAMBLE_SYMBOLS / 500_000 * 1000.0, color="black", alpha=0.35) axis.set(xlabel="Время, мс", ylabel="Развёрнутая фаза, рад", title=f"Lab056 {label}: траектория фазы") axis.grid(alpha=0.25); axis.legend(ncol=4) figure.savefig(OUTPUT / f"phase_trajectory_{label.lower()}.png", dpi=160) plt.close(figure) figure, axes = plt.subplots(2, 1, figsize=(10, 7), sharex=True, constrained_layout=True) for axis, label in zip(axes, ("B1", "B2")): count = len(trajectories[f"{label}_T0_frequency_hz"]) indexes = np.arange(0, count, max(1, count // 2500)) for tracker in TRACKERS: axis.plot((indexes + 0.5) / 500_000 * 1000.0, trajectories[f"{label}_{tracker}_frequency_hz"][indexes], color=colors[tracker], label=tracker) reference = next(row for row in real_rows if row["label"] == label) axis.axhline(reference["diagnostic_reference_frequency_hz"], color="black", linestyle="--", label="diagnostic reference") axis.set(ylabel="Частота, Гц", title=label); axis.grid(alpha=0.25) axes[-1].set_xlabel("Время, мс"); axes[0].legend(ncol=4) figure.savefig(OUTPUT / "frequency_state_trajectory.png", dpi=160) plt.close(figure) figure, axes = plt.subplots(1, 2, figsize=(11, 4.8), sharey=True, constrained_layout=True) for axis, label in zip(axes, ("B1", "B2")): for tracker in TRACKERS: selected = [row for row in block_rows if row["label"] == label and row["tracker"] == tracker] axis.plot([row["end_seconds"] * 1000 for row in selected], [100 * row["symbol_error_rate"] for row in selected], marker="o", color=colors[tracker], label=tracker) axis.axhline(25, color="black", linestyle="--", linewidth=0.8, label="LOSS block") axis.set(title=label, xlabel="Конец блока, мс"); axis.grid(alpha=0.25) axes[0].set_ylabel("Ошибочные символы, %"); axes[0].legend() figure.savefig(OUTPUT / "lab056_b_block_symbol_error_rate.png", dpi=160) plt.close(figure) figure, axis = plt.subplots(figsize=(10, 5), constrained_layout=True) for tracker in TRACKERS: selected = [row for row in synthetic_by_frequency if row["tracker"] == tracker] axis.plot([row["initial_frequency_error_hz"] for row in selected], [100 * row["catastrophic_failure_rate"] for row in selected], marker="o", color=colors[tracker], label=tracker) axis.set(xlabel="Начальная ошибка частоты, Гц", ylabel="Катастрофические срывы, %", title="Synthetic 500K, качество Lab056 B") axis.grid(alpha=0.25); axis.legend() figure.savefig(OUTPUT / "synthetic_failure_by_initial_cfo.png", dpi=160) plt.close(figure) b_means = [np.mean([row["symbol_error_rate"] for row in real_rows if row["tracker"] == tracker and row["label"].startswith("B")]) for tracker in TRACKERS] figure, axis = plt.subplots(figsize=(8, 5), constrained_layout=True) axis.bar(TRACKERS, 100 * np.asarray(b_means), color=[colors[item] for item in TRACKERS]) axis.axhline(3.495896, color="black", linestyle="--", label="diagnostic lower reference") axis.axhline(6.0, color="black", linestyle=":", label="engineering target") axis.set(ylabel="Средняя SER B, %", title="Lab056 B: T0/T1/T2") axis.grid(axis="y", alpha=0.25); axis.legend() figure.savefig(OUTPUT / "tracker_comparison.png", dpi=160) plt.close(figure) figure, axis = plt.subplots(figsize=(9, 5), constrained_layout=True) headers = [packet_summary["Lab054"][tracker]["header_accepted"] for tracker in TRACKERS] packets = [packet_summary["Lab054"][tracker]["packet_output"] for tracker in TRACKERS] positions = np.arange(len(TRACKERS)) axis.bar(positions - 0.18, headers, 0.36, label="HEADER_ACCEPTED") axis.bar(positions + 0.18, packets, 0.36, label="PACKET_OUTPUT") axis.set(xticks=positions, xticklabels=TRACKERS, xlim=(-0.6, 2.6), ylim=(0, 16.5), ylabel="Кадры из 16", title="Lab054 после carrier tracking") if max(headers + packets) == 0: axis.text(1.0, 8.0, "Все методы: 0/16", ha="center") axis.grid(axis="y", alpha=0.25); axis.legend() figure.savefig(OUTPUT / "lab054_result_by_tracker.png", dpi=160) plt.close(figure) def run_functional_tests() -> None: assert normalized_phase_detector(2.0 * np.exp(0.2j)) == normalized_phase_detector( 20.0 * np.exp(0.2j) ) first = loop_parameters("T1", 0.001) second = loop_parameters("T2", 0.001) assert first.phase_coefficient > 0 and first.frequency_coefficient == 0 assert second.phase_coefficient > 0 and second.frequency_coefficient > 0 assert len(INITIAL_FREQUENCY_ERRORS_HZ) == 13 def main() -> None: OUTPUT.mkdir(parents=True, exist_ok=True) before = integrity_snapshot() parameters, tuning_rows = tune_loop_parameters() write_csv(OUTPUT / "tuning_results.csv", tuning_rows) synthetic_rows, synthetic_by_frequency = synthetic_validation(parameters) write_csv(OUTPUT / "synthetic_results.csv", synthetic_rows) write_csv(OUTPUT / "synthetic_by_initial_cfo.csv", synthetic_by_frequency) real_rows, block_rows, trajectories = analyze_lab056(parameters) write_csv(OUTPUT / "real_results.csv", real_rows) write_csv(OUTPUT / "real_blocks.csv", block_rows) np.savez_compressed(OUTPUT / "real_trajectories.npz", **trajectories) packet_rows = analyze_packets(parameters) write_csv(OUTPUT / "packet_results.csv", packet_rows) packet_summary = summarize_packets(packet_rows) background_rows = analyze_background(parameters) write_csv(OUTPUT / "background_results.csv", background_rows) leak_control = payload_truth_leak_control(parameters) if not leak_control["passed"]: raise RuntimeError("TX payload truth влияет на траекторию tracker") verdict, best, reason, next_step = choose_verdict( real_rows, synthetic_rows, packet_summary, background_rows ) after = integrity_snapshot() if before != after: raise RuntimeError("Исходные IQ изменились во время Lab060") real_summary = { tracker: { "A_mean_symbol_error_rate": float(np.mean([ row["symbol_error_rate"] for row in real_rows if row["tracker"] == tracker and row["label"].startswith("A") ])), "B_mean_symbol_error_rate": float(np.mean([ row["symbol_error_rate"] for row in real_rows if row["tracker"] == tracker and row["label"].startswith("B") ])), "B_loss_of_lock": sum(row["loss_of_lock"] for row in real_rows if row["tracker"] == tracker and row["label"].startswith("B")), } for tracker in TRACKERS } summary = { "lab": "Lab060", "status": "COMPLETED", "scope": "OFFLINE_SAVED_FILES_AND_SYNTHETIC_ONLY", "rf_tx": 0, "hardware_access": 0, "lab059_commit": "e16ec23fdb57e95ade46e9b6abc215502fca0686", "trackers": list(TRACKERS), "fixed_design": { "preamble_symbols": PREAMBLE_SYMBOLS, "detector": "normalized decision-directed BPSK phase error", "detector_clip_rad": DETECTOR_CLIP_RAD, "loss_criterion": ( "two consecutive of 16 post-preamble blocks with symbol error rate >= 0.25; " "diagnostic truth only, independent of CRC" ), "receiver_lock_score": "mean cos(2*phase_detector_error), threshold 0.15", "candidate_normalized_bandwidths": list(NORMALIZED_BANDWIDTHS), "damping_factor": DAMPING_FACTOR, }, "selected_parameters": parameters, "selection_source": "synthetic tuning only; frozen before real IQ", "synthetic_clean_rrc": { str(rate): clean_rrc_symbols(rate)[2] for rate in (20_000, 500_000) }, "real": real_summary, "synthetic": synthetic_rows, "packet_controls": packet_summary, "background": background_rows, "payload_truth_leak_test": leak_control, "decision_directed_verdict": verdict, "best_tracker": best, "why": reason, "remaining_physical_limitation": { "diagnostic_symbol_error_rate": 0.03495896, "evm_rms": 0.845350, "matched_residual_db": 1.464846, "band_gain_nonuniformity_db": 5.471382, }, "recommended_next_single_step": next_step, "lab054_historical_result": "NOT_A, 0/16", "lab060_commit": "NOT_PERFORMED", "push": "NOT_PERFORMED", "integrity": {"before": before, "after": after, "passed": before == after}, } write_json(OUTPUT / "summary.json", summary) make_plots(real_rows, block_rows, trajectories, synthetic_by_frequency, packet_summary) print(json.dumps(serializable(summary), ensure_ascii=False, indent=2)) if __name__ == "__main__": run_functional_tests() main()