907 lines
45 KiB
Python
907 lines
45 KiB
Python
"""Lab067: офлайн-анализ отказов PHY-заголовка в сохранённой записи Lab066.
|
||
|
||
Строгий результат Lab066 загружается без изменений. Известный TX используется
|
||
только после этого результата для диагностического подсчёта ошибок.
|
||
Аппаратные команды и передача RF в этой лаборатории отсутствуют.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import hashlib
|
||
import json
|
||
import math
|
||
from pathlib import Path
|
||
import struct
|
||
import sys
|
||
from typing import Any
|
||
|
||
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
|
||
|
||
from experiments import lab053_packet_processing as packet_processing
|
||
from experiments import lab058_abab_symbol_quality as symbol_quality
|
||
from experiments import lab066_full_packet_rf_retest as lab066
|
||
from protocol import bpsk_carrier_tracking as carrier_tracking
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
RAW = ROOT / "data/raw/lab066"
|
||
LAB066 = ROOT / "data/processed/lab066"
|
||
OUTPUT = ROOT / "data/processed/lab067"
|
||
TX_PATH = RAW / "tx_stream.cs8"
|
||
RX_PATH = RAW / "rx_stream.cs8"
|
||
TX_SHA256 = "621b0bb8ec72bba4f522c79a5c77669c850740ea481270c4dcc2a7317819f199"
|
||
RX_SHA256 = "72ff7ddd8b30baa4a5c066361ca1f8dfa664f1da37f65c0c373b211ce6cf00f0"
|
||
RX_BYTES = 32_000_000
|
||
RX_SAMPLES = 16_000_000
|
||
SAMPLE_RATE_HZ = 8_000_000
|
||
SYMBOL_RATE = 500_000
|
||
SAMPLES_PER_SYMBOL = 16
|
||
PREAMBLE_SYMBOLS = 1280
|
||
HEADER_SYMBOLS = 144
|
||
HEADER_BITS = 48
|
||
HEADER_REPETITION = 3
|
||
POSTHEADER_SYMBOLS = 128
|
||
FAILED_PACKETS = {1, 2, 3, 9, 15, 16}
|
||
PASSED_PACKETS = set(range(1, 17)) - FAILED_PACKETS
|
||
|
||
|
||
def sha256(path: Path) -> str:
|
||
digest = hashlib.sha256()
|
||
with path.open("rb") as stream:
|
||
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
||
digest.update(block)
|
||
return digest.hexdigest()
|
||
|
||
|
||
def save_json(path: Path, value: Any) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(
|
||
json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def bits_text(bits: np.ndarray) -> str:
|
||
return "".join(str(int(value)) for value in np.asarray(bits).reshape(-1))
|
||
|
||
|
||
def finite(value: float) -> float | None:
|
||
return float(value) if np.isfinite(value) else None
|
||
|
||
|
||
def source_integrity() -> tuple[dict, dict, dict]:
|
||
"""Проверить неизменность Lab066 до любой oracle-диагностики."""
|
||
capture = json.loads((LAB066 / "capture_result.json").read_text(encoding="utf-8"))
|
||
strict = json.loads((LAB066 / "rf_result.json").read_text(encoding="utf-8"))
|
||
metadata = json.loads((LAB066 / "preparation.json").read_text(encoding="utf-8"))
|
||
tx_digest = sha256(TX_PATH)
|
||
rx_digest = sha256(RX_PATH)
|
||
integrity = {
|
||
"tx_path": str(TX_PATH.relative_to(ROOT)),
|
||
"tx_bytes": TX_PATH.stat().st_size,
|
||
"tx_samples": TX_PATH.stat().st_size // 2,
|
||
"tx_sha256": tx_digest,
|
||
"rx_path": str(RX_PATH.relative_to(ROOT)),
|
||
"rx_bytes": RX_PATH.stat().st_size,
|
||
"rx_samples": RX_PATH.stat().st_size // 2,
|
||
"rx_duration_seconds": (RX_PATH.stat().st_size // 2) / SAMPLE_RATE_HZ,
|
||
"rx_sha256": rx_digest,
|
||
"sample_rate_hz": SAMPLE_RATE_HZ,
|
||
}
|
||
if (
|
||
tx_digest != TX_SHA256
|
||
or rx_digest != RX_SHA256
|
||
or integrity["rx_bytes"] != RX_BYTES
|
||
or integrity["rx_samples"] != RX_SAMPLES
|
||
or metadata.get("tx_sha256") != TX_SHA256
|
||
or metadata.get("parameters", {}).get("sample_rate_hz") != SAMPLE_RATE_HZ
|
||
or capture.get("rx_sha256") != RX_SHA256
|
||
or capture.get("rf_tx_count") != 1
|
||
or capture.get("automatic_tx_repeats") != 0
|
||
or capture.get("passed") is not True
|
||
or strict.get("capture_valid") is not True
|
||
):
|
||
raise RuntimeError("Исходники или аппаратная валидность Lab066 не подтверждены")
|
||
for mode in ("run_a_legacy", "run_b_t2"):
|
||
result = strict[mode]
|
||
if (
|
||
result.get("header_pass") != 10
|
||
or result.get("packet_output") != 10
|
||
or result.get("exact_payload_match") != 10
|
||
or result.get("packet_indexes") != sorted(PASSED_PACKETS)
|
||
):
|
||
raise RuntimeError("Исторический строгий результат Lab066 изменился")
|
||
return integrity, metadata, strict
|
||
|
||
|
||
def extract_symbols(
|
||
received: symbol_quality.Cs8File,
|
||
frame_start: int,
|
||
symbol_count: int,
|
||
mode: str,
|
||
) -> dict:
|
||
"""Повторить штатные timing/carrier операции для известного окна пакета."""
|
||
processor = packet_processing.PacketProcessor(carrier_tracking_mode=mode)
|
||
length = (symbol_count + 42) * SAMPLES_PER_SYMBOL
|
||
segment = received.read(frame_start, length).astype(np.complex64)
|
||
matched = processor.filter(processor.shift(segment))
|
||
nominal = 16 * SAMPLES_PER_SYMBOL + processor.taps.size - 1
|
||
timing = int(processor.timing(matched, nominal, processor.expected))
|
||
initial = processor.carrier(
|
||
matched, nominal, timing, processor.bits, len(processor.bits)
|
||
)
|
||
symbols = matched[
|
||
nominal + timing:
|
||
nominal + timing + symbol_count * SAMPLES_PER_SYMBOL:
|
||
SAMPLES_PER_SYMBOL
|
||
].astype(np.complex128)
|
||
if symbols.size != symbol_count or initial["abs_h"] == 0:
|
||
raise RuntimeError("Не удалось извлечь диагностическое окно символов")
|
||
times = (np.arange(symbol_count, dtype=np.float64) + 0.5) / SYMBOL_RATE
|
||
if mode == "legacy":
|
||
corrected = symbols * np.exp(-2j * np.pi * initial["cfo_hz"] * times)
|
||
gain = initial["abs_h"] * np.exp(1j * initial["phase_rad"])
|
||
corrected /= gain
|
||
phase_trace = (
|
||
initial["phase_rad"] + 2 * np.pi * initial["cfo_hz"] * times
|
||
)
|
||
frequency_trace = np.full(symbol_count, initial["cfo_hz"], dtype=float)
|
||
detector_error = np.angle(corrected * np.where(corrected.real >= 0, 1, -1))
|
||
estimate = {
|
||
"frequency_hz": float(initial["cfo_hz"]),
|
||
"gain": complex(gain),
|
||
"phase_fit_rmse_rad": None,
|
||
"squared_coherence": None,
|
||
}
|
||
else:
|
||
estimate_value = carrier_tracking.estimate_p3(
|
||
symbols[:PREAMBLE_SYMBOLS], processor.expected, SYMBOL_RATE
|
||
)
|
||
tracked = carrier_tracking.track_t2(
|
||
symbols, SYMBOL_RATE, estimate_value, start_index=PREAMBLE_SYMBOLS
|
||
)
|
||
corrected = tracked.corrected.copy()
|
||
phase_trace = tracked.phase_rad.copy()
|
||
frequency_trace = tracked.frequency_hz.copy()
|
||
detector_error = tracked.detector_error_rad.copy()
|
||
estimate = {
|
||
"frequency_hz": float(estimate_value.frequency_hz),
|
||
"gain": complex(estimate_value.gain),
|
||
"phase_fit_rmse_rad": float(estimate_value.phase_fit_rmse_rad),
|
||
"squared_coherence": float(estimate_value.squared_coherence),
|
||
}
|
||
return {
|
||
"corrected": corrected,
|
||
"raw_symbols": symbols,
|
||
"matched": matched,
|
||
"hard_bits": (corrected.real >= 0).astype(np.uint8),
|
||
"timing_samples": timing,
|
||
"timing_fraction": timing / SAMPLES_PER_SYMBOL,
|
||
"nominal": nominal,
|
||
"sample_positions": frame_start + nominal + timing + np.arange(symbol_count) * SAMPLES_PER_SYMBOL,
|
||
"initial": estimate,
|
||
"phase_trace": phase_trace,
|
||
"frequency_trace": frequency_trace,
|
||
"detector_error": detector_error,
|
||
}
|
||
|
||
|
||
def reference_sync_diagnostic(extracted: dict, expected_bits: np.ndarray) -> dict:
|
||
"""Oracle-проверка timing/CFO после штатного результата, без влияния на него."""
|
||
expected = 2.0 * np.asarray(expected_bits, dtype=np.float64) - 1.0
|
||
indexes = np.arange(len(expected), dtype=np.int64)
|
||
candidates = []
|
||
for offset in range(-SAMPLES_PER_SYMBOL // 2, SAMPLES_PER_SYMBOL // 2 + 1):
|
||
start = extracted["nominal"] + offset
|
||
symbols = extracted["matched"][
|
||
start:start + len(expected) * SAMPLES_PER_SYMBOL:SAMPLES_PER_SYMBOL
|
||
].astype(np.complex128)
|
||
if len(symbols) != len(expected):
|
||
continue
|
||
frequency, gain, coherence = symbol_quality.fit_cfo_gain(
|
||
symbols, expected, indexes, SYMBOL_RATE, len(expected)
|
||
)
|
||
candidates.append((coherence, offset, frequency, gain))
|
||
coherence, timing, frequency, gain = max(candidates, key=lambda item: item[0])
|
||
return {
|
||
"scope": "KNOWN_BITS_AFTER_STRICT_VERDICT",
|
||
"timing_offset_samples": int(timing),
|
||
"timing_fraction": timing / SAMPLES_PER_SYMBOL,
|
||
"frequency_hz": float(frequency),
|
||
"gain_real": float(gain.real),
|
||
"gain_imag": float(gain.imag),
|
||
"coherence": float(coherence),
|
||
}
|
||
|
||
|
||
def physical_metrics(extracted: dict, expected_bits: np.ndarray, selected: slice) -> dict:
|
||
expected = 2.0 * np.asarray(expected_bits[selected], dtype=np.float64) - 1.0
|
||
raw = extracted["raw_symbols"][selected]
|
||
indexes = np.arange(len(extracted["raw_symbols"]), dtype=np.float64)[selected]
|
||
frequency = extracted["initial"]["frequency_hz"]
|
||
times = (indexes + 0.5) / SYMBOL_RATE
|
||
derotated = raw * np.exp(-2j * np.pi * frequency * times)
|
||
gain = complex(extracted["initial"]["gain"])
|
||
predicted = gain * expected
|
||
residual = derotated - predicted
|
||
matched_power = float(np.mean(abs(predicted) ** 2))
|
||
residual_power = float(np.mean(abs(residual) ** 2))
|
||
normalized = derotated / gain
|
||
errors = int(np.count_nonzero((normalized.real >= 0) != (expected > 0)))
|
||
correlation = float(
|
||
abs(np.vdot(expected, raw))
|
||
/ math.sqrt(float(np.vdot(expected, expected).real * np.vdot(raw, raw).real) + 1e-30)
|
||
)
|
||
return {
|
||
"symbols": int(len(expected)),
|
||
"errors": errors,
|
||
"error_rate": errors / len(expected),
|
||
"correlation": correlation,
|
||
"matched_component_power": matched_power,
|
||
"residual_power": residual_power,
|
||
"matched_residual_db": finite(10 * np.log10(matched_power / residual_power)),
|
||
"rms_symbol_deviation": float(np.sqrt(residual_power / matched_power)),
|
||
}
|
||
|
||
|
||
def header_diagnostics(received_soft: np.ndarray, expected_bits: np.ndarray) -> dict:
|
||
soft = np.asarray(received_soft[:HEADER_SYMBOLS], dtype=np.float64)
|
||
raw = (soft >= 0.0).astype(np.uint8)
|
||
expected = np.asarray(expected_bits[:HEADER_SYMBOLS], dtype=np.uint8)
|
||
raw_groups = raw.reshape(HEADER_BITS, HEADER_REPETITION)
|
||
expected_groups = expected.reshape(HEADER_BITS, HEADER_REPETITION)
|
||
voted = (np.sum(soft.reshape(HEADER_BITS, HEADER_REPETITION), axis=1) >= 0.0).astype(np.uint8)
|
||
expected_voted = expected_groups[:, 0]
|
||
fields = lab066.phy.lab045.lab044.bits_to_bytes(voted)
|
||
magic, version, length, checksum = struct.unpack(
|
||
lab066.phy.lab045.PHY_HEADER_FIELDS_FORMAT, fields
|
||
)
|
||
decoded = lab066.phy.lab045.decode_phy_header_soft(soft)
|
||
field_errors = {
|
||
"magic": int(np.count_nonzero(voted[:16] != expected_voted[:16])),
|
||
"version": int(np.count_nonzero(voted[16:24] != expected_voted[16:24])),
|
||
"length": int(np.count_nonzero(voted[24:40] != expected_voted[24:40])),
|
||
"checksum": int(np.count_nonzero(voted[40:48] != expected_voted[40:48])),
|
||
}
|
||
return {
|
||
"expected_header_bits": bits_text(expected),
|
||
"received_raw_header_bits": bits_text(raw),
|
||
"raw_header_errors": int(np.count_nonzero(raw != expected)),
|
||
"copy_errors": [
|
||
int(np.count_nonzero(raw_groups[:, index] != expected_groups[:, index]))
|
||
for index in range(HEADER_REPETITION)
|
||
],
|
||
"voted_header_bits": bits_text(voted),
|
||
"expected_voted_bits": bits_text(expected_voted),
|
||
"voted_header_errors": int(np.count_nonzero(voted != expected_voted)),
|
||
"magic_errors": field_errors["magic"],
|
||
"version_errors": field_errors["version"],
|
||
"length_errors": field_errors["length"],
|
||
"checksum_errors": field_errors["checksum"],
|
||
"decoded_magic": magic,
|
||
"decoded_version": version,
|
||
"decoded_length": length,
|
||
"decoded_checksum": checksum,
|
||
"accepted": bool(decoded["passed"]),
|
||
"reason": decoded["reason"],
|
||
"raw_error_positions": np.flatnonzero(raw != expected).astype(int).tolist(),
|
||
"voted_error_positions": np.flatnonzero(voted != expected_voted).astype(int).tolist(),
|
||
}
|
||
|
||
|
||
def gap_metrics(received: symbol_quality.Cs8File, start: int, count: int) -> dict:
|
||
values = received.read(start, count).astype(np.complex128)
|
||
mean = np.mean(values)
|
||
centered = values - mean
|
||
edge = min(count // 4, 8000)
|
||
first_power = float(np.mean(abs(centered[:edge]) ** 2))
|
||
middle_start = max(edge, count // 2 - edge // 2)
|
||
middle_power = float(np.mean(abs(centered[middle_start:middle_start + edge]) ** 2))
|
||
tail_ratio_db = float(10 * np.log10(max(first_power, 1e-30) / max(middle_power, 1e-30)))
|
||
return {
|
||
"length_samples": count,
|
||
"duration_seconds": count / SAMPLE_RATE_HZ,
|
||
"background_ac_rms": float(np.sqrt(np.mean(abs(centered) ** 2))),
|
||
"mean_i": float(mean.real),
|
||
"mean_q": float(mean.imag),
|
||
"residual_energy_mean": float(np.mean(abs(centered) ** 2)),
|
||
"leading_edge_to_middle_db": tail_ratio_db,
|
||
"previous_waveform_tail_detected": bool(tail_ratio_db > 3.0),
|
||
}
|
||
|
||
|
||
def strict_candidate_for_packet(rows: list[dict], expected: int) -> dict | None:
|
||
candidates = [row for row in rows if abs(int(row["absolute_refined"]) - expected) <= 64]
|
||
if not candidates:
|
||
return None
|
||
return min(
|
||
candidates,
|
||
key=lambda row: (
|
||
abs(int(row["absolute_refined"]) - expected),
|
||
int(row["absolute_refined"]),
|
||
),
|
||
)
|
||
|
||
|
||
def candidate_analysis(strict: dict, expected_starts: list[int]) -> tuple[dict, list[dict]]:
|
||
rows = strict["run_a_legacy"]["rows"]
|
||
packet_samples = lab066.PACKET_SAMPLES
|
||
gap_intervals = []
|
||
metadata = json.loads((LAB066 / "preparation.json").read_text(encoding="utf-8"))
|
||
stream_start = strict["location_after_strict_runs"]["stream_rx_start_sample"]
|
||
for layout in metadata["layout"][:-1]:
|
||
low = stream_start + layout["following_gap_start_sample"]
|
||
gap_intervals.append((low, low + layout["following_gap_samples"]))
|
||
output_rows = []
|
||
true = repeated = gaps = unrelated = 0
|
||
true_positions = set()
|
||
per_packet = {index: 0 for index in range(1, 17)}
|
||
for number, row in enumerate(rows, start=1):
|
||
position = int(row["absolute_refined"])
|
||
distances = np.abs(np.asarray(expected_starts, dtype=np.int64) - position)
|
||
nearest_index = int(np.argmin(distances)) + 1
|
||
delta = position - expected_starts[nearest_index - 1]
|
||
in_gap = any(low <= position < high for low, high in gap_intervals)
|
||
if abs(delta) <= 64:
|
||
category = "TRUE_PREAMBLE_POSITION"
|
||
true += 1
|
||
true_positions.add((nearest_index, position))
|
||
elif -64 <= delta < packet_samples:
|
||
category = "REPEATED_PRBS_LOCAL_MAXIMUM"
|
||
repeated += 1
|
||
elif in_gap:
|
||
category = "ZERO_GAP"
|
||
gaps += 1
|
||
else:
|
||
category = "UNRELATED"
|
||
unrelated += 1
|
||
if category in {"TRUE_PREAMBLE_POSITION", "REPEATED_PRBS_LOCAL_MAXIMUM"}:
|
||
per_packet[nearest_index] += 1
|
||
output_rows.append({
|
||
"candidate_number": number,
|
||
"position": position,
|
||
"nearest_packet": nearest_index,
|
||
"difference_samples": delta,
|
||
"category": category,
|
||
"strict_reason": row["reason"],
|
||
})
|
||
summary = {
|
||
"total_strict_candidates": len(rows),
|
||
"true_preamble_position_checks": true,
|
||
"unique_true_preamble_positions": len(true_positions),
|
||
"duplicate_true_position_checks": true - len(true_positions),
|
||
"repeated_prbs_local_maxima": repeated,
|
||
"zero_gap_candidates": gaps,
|
||
"unrelated_candidates": unrelated,
|
||
"checks_associated_with_each_real_preamble": per_packet,
|
||
"detector_statistics": strict["run_a_legacy"]["detector_statistics"],
|
||
"candidate_processing_state": "NEW_PACKET_MEASURE_AND_INDEPENDENT_DECODE_PER_CANDIDATE",
|
||
"persistent_state_between_rejected_candidates": False,
|
||
"accepted_until_changes_only_after_crc_pass": True,
|
||
}
|
||
return summary, output_rows
|
||
|
||
|
||
def distribution(rows: list[dict], packets: set[int], key: str) -> dict:
|
||
values = np.asarray([row[key] for row in rows if row["packet"] in packets], dtype=float)
|
||
return {
|
||
"minimum": float(np.min(values)),
|
||
"median": float(np.median(values)),
|
||
"maximum": float(np.max(values)),
|
||
"values": {str(row["packet"]): row[key] for row in rows if row["packet"] in packets},
|
||
}
|
||
|
||
|
||
def assign_classifications(rows: list[dict]) -> None:
|
||
passed = [row for row in rows if row["packet"] in PASSED_PACKETS]
|
||
limits = {
|
||
"preamble_errors_max": max(row["preamble_errors"] for row in passed),
|
||
"preamble_mr_p10": float(np.percentile([row["preamble_matched_residual_db"] for row in passed], 10)),
|
||
"preamble_rms_p90": float(np.percentile([row["preamble_rms_symbol_deviation"] for row in passed], 90)),
|
||
"correlation_p10": float(np.percentile([row["preamble_correlation"] for row in passed], 10)),
|
||
}
|
||
for row in rows:
|
||
if row["packet"] in PASSED_PACKETS:
|
||
row["classification"] = "PASSED"
|
||
continue
|
||
physical_bad = sum((
|
||
row["preamble_errors"] > limits["preamble_errors_max"],
|
||
row["preamble_matched_residual_db"] < limits["preamble_mr_p10"],
|
||
row["preamble_rms_symbol_deviation"] > limits["preamble_rms_p90"],
|
||
row["preamble_correlation"] < limits["correlation_p10"],
|
||
))
|
||
if row["preamble_errors"] > limits["preamble_errors_max"]:
|
||
row["classification"] = "PREAMBLE_ALREADY_BAD"
|
||
elif physical_bad >= 2:
|
||
row["classification"] = "LOCAL_RF_DEGRADATION"
|
||
elif row["header_voted_errors"] > 0:
|
||
row["classification"] = "HEADER_SPECIFIC_FAILURE"
|
||
else:
|
||
row["classification"] = "NOT_LOCALIZED"
|
||
|
||
|
||
def group_summary(rows: list[dict]) -> dict:
|
||
groups = {"START": range(1, 5), "MIDDLE_1": range(5, 9), "MIDDLE_2": range(9, 13), "END": range(13, 17)}
|
||
result = {}
|
||
for name, indexes in groups.items():
|
||
selected = [row for row in rows if row["packet"] in indexes]
|
||
result[name] = {
|
||
"packets": list(indexes),
|
||
"preamble_error_rate": float(np.mean([row["preamble_error_rate"] for row in selected])),
|
||
"header_error_rate": float(np.mean([row["header_raw_errors"] / HEADER_SYMBOLS for row in selected])),
|
||
"matched_power": float(np.mean([row["local_matched_power"] for row in selected])),
|
||
"residual_power": float(np.mean([row["local_residual_power"] for row in selected])),
|
||
"matched_residual_db": float(10 * np.log10(
|
||
np.mean([row["local_matched_power"] for row in selected])
|
||
/ np.mean([row["local_residual_power"] for row in selected])
|
||
)),
|
||
"initial_frequency_offset_hz": float(np.mean([row["legacy_initial_frequency_hz"] for row in selected])),
|
||
"timing_offset_samples": float(np.mean([row["timing_offset_samples"] for row in selected])),
|
||
"background_ac_rms": float(np.mean([row["gap_background_ac_rms"] for row in selected])),
|
||
}
|
||
return result
|
||
|
||
|
||
def make_plots(rows: list[dict], transitions: dict[int, dict], candidates: dict) -> list[str]:
|
||
OUTPUT.mkdir(parents=True, exist_ok=True)
|
||
packets = [row["packet"] for row in rows]
|
||
colors = ["C3" if packet in FAILED_PACKETS else "C0" for packet in packets]
|
||
plots = []
|
||
|
||
def save(name: str) -> None:
|
||
plt.tight_layout()
|
||
plt.savefig(OUTPUT / name, dpi=160)
|
||
plt.close()
|
||
plots.append(name)
|
||
|
||
plt.figure(figsize=(10, 4.5))
|
||
plt.bar(packets, [row["preamble_errors"] for row in rows], color=colors)
|
||
plt.xticks(packets); plt.xlabel("Пакет"); plt.ylabel("Ошибки преамбулы, бит")
|
||
plt.grid(axis="y", alpha=.25); save("preamble_errors_by_packet.png")
|
||
|
||
plt.figure(figsize=(10, 4.5))
|
||
x = np.arange(1, 17)
|
||
plt.bar(x - .18, [row["header_raw_errors"] for row in rows], width=.36, label="raw")
|
||
plt.bar(x + .18, [row["header_voted_errors"] for row in rows], width=.36, label="voted", color=colors)
|
||
plt.xticks(x); plt.xlabel("Пакет"); plt.ylabel("Ошибки PHY header, бит"); plt.legend()
|
||
plt.grid(axis="y", alpha=.25); save("header_errors_by_packet.png")
|
||
|
||
figure, axes = plt.subplots(3, 1, figsize=(10, 8), sharex=True)
|
||
axes[0].bar(packets, [row["local_matched_power"] for row in rows], color=colors)
|
||
axes[0].set_ylabel("Matched power")
|
||
axes[1].bar(packets, [row["local_residual_power"] for row in rows], color=colors)
|
||
axes[1].set_ylabel("Residual power")
|
||
axes[2].bar(packets, [row["local_matched_residual_db"] for row in rows], color=colors)
|
||
axes[2].set_ylabel("Matched/residual, dB"); axes[2].set_xlabel("Пакет"); axes[2].set_xticks(packets)
|
||
for axis in axes: axis.grid(axis="y", alpha=.25)
|
||
save("local_signal_quality_by_packet.png")
|
||
|
||
matrix = np.zeros((16, HEADER_SYMBOLS), dtype=np.uint8)
|
||
for row in rows:
|
||
for position in row["header_raw_error_positions"]:
|
||
matrix[row["packet"] - 1, position] = 1
|
||
plt.figure(figsize=(13, 5))
|
||
plt.imshow(matrix, aspect="auto", interpolation="nearest", cmap="Reds")
|
||
plt.xlabel("Позиция защищённого PHY header, 0...143"); plt.ylabel("Пакет")
|
||
plt.yticks(np.arange(16), np.arange(1, 17)); plt.colorbar(label="Ошибка")
|
||
save("header_error_position_map.png")
|
||
|
||
representative = [4, 1, 14, 16]
|
||
figure, axes = plt.subplots(len(representative), 3, figsize=(13, 10), sharex=True)
|
||
for line, packet in enumerate(representative):
|
||
values = transitions[packet]
|
||
x = np.arange(-128, HEADER_SYMBOLS + 128)
|
||
axes[line, 0].plot(x, values["corrected_real"], lw=.7)
|
||
axes[line, 0].scatter(x[values["errors"]], np.asarray(values["corrected_real"])[values["errors"]], s=12, color="red")
|
||
axes[line, 1].plot(x, values["corrected_imag"], lw=.7)
|
||
axes[line, 2].plot(x, values["phase_error_rad"], lw=.7)
|
||
for axis in axes[line]:
|
||
axis.axvline(0, color="black", lw=.8); axis.axvline(HEADER_SYMBOLS, color="black", lw=.8)
|
||
axis.grid(alpha=.2)
|
||
axes[line, 0].set_ylabel(f"P{packet}")
|
||
axes[0, 0].set_title("Re, ошибки красным"); axes[0, 1].set_title("Im"); axes[0, 2].set_title("Фазовая ошибка, rad")
|
||
for axis in axes[-1]: axis.set_xlabel("Символ относительно начала header")
|
||
save("preamble_header_transition.png")
|
||
|
||
figure, axes = plt.subplots(2, 1, figsize=(10, 6), sharex=True)
|
||
axes[0].bar(packets, [row["timing_offset_samples"] for row in rows], color=colors)
|
||
axes[0].set_ylabel("Timing, samples")
|
||
axes[1].plot(packets, [row["legacy_initial_frequency_hz"] for row in rows], "o-", label="legacy")
|
||
axes[1].plot(packets, [row["t2_initial_frequency_hz"] for row in rows], "s-", label="T2/P3")
|
||
axes[1].set_ylabel("CFO, Hz"); axes[1].set_xlabel("Пакет"); axes[1].legend()
|
||
for axis in axes: axis.grid(alpha=.25)
|
||
save("timing_and_cfo_by_packet.png")
|
||
|
||
plt.figure(figsize=(10, 4.5))
|
||
plt.bar(packets, [row["gap_background_ac_rms"] for row in rows], color=colors)
|
||
plt.xlabel("Пакет"); plt.ylabel("AC RMS перед пакетом"); plt.xticks(packets); plt.grid(axis="y", alpha=.25)
|
||
save("gap_background_by_packet.png")
|
||
|
||
per_packet = candidates["checks_associated_with_each_real_preamble"]
|
||
plt.figure(figsize=(10, 4.5))
|
||
plt.bar(packets, [per_packet[str(packet)] if str(packet) in per_packet else per_packet[packet] for packet in packets], color=colors)
|
||
plt.xlabel("Пакет"); plt.ylabel("Строгие проверки кандидатов"); plt.xticks(packets); plt.grid(axis="y", alpha=.25)
|
||
save("candidate_checks_by_packet.png")
|
||
return plots
|
||
|
||
|
||
def write_csvs(rows: list[dict], candidate_rows: list[dict], bit_rows: list[dict], transition_rows: list[dict]) -> None:
|
||
main_fields = [
|
||
"packet", "strict_result", "candidate_position", "expected_packet_position", "candidate_difference_samples",
|
||
"preamble_errors", "preamble_error_rate", "preamble_correlation", "header_raw_errors", "header_voted_errors",
|
||
"header_copy_1_errors", "header_copy_2_errors", "header_copy_3_errors", "magic_errors", "version_errors",
|
||
"length_errors", "checksum_errors", "expected_header_bits", "received_raw_header_bits", "voted_header_bits",
|
||
"header_accepted", "header_reason", "legacy_header_bits", "t2_header_bits",
|
||
"legacy_initial_frequency_hz", "t2_initial_frequency_hz", "reference_frequency_hz",
|
||
"legacy_frequency_error_vs_reference_hz", "initial_phase_error_rad", "timing_offset_samples", "fractional_timing_offset",
|
||
"reference_timing_offset_samples", "reference_fractional_timing_offset", "local_matched_power", "local_residual_power",
|
||
"local_matched_residual_db", "rms_symbol_deviation", "legacy_t2_header_differences", "classification",
|
||
]
|
||
with (OUTPUT / "header_packet_comparison.csv").open("w", newline="", encoding="utf-8-sig") as stream:
|
||
writer = csv.DictWriter(stream, fieldnames=main_fields, extrasaction="ignore")
|
||
writer.writeheader(); writer.writerows(rows)
|
||
for name, values in (
|
||
("header_bit_errors.csv", bit_rows),
|
||
("candidate_analysis.csv", candidate_rows),
|
||
("transition_metrics.csv", transition_rows),
|
||
):
|
||
with (OUTPUT / name).open("w", newline="", encoding="utf-8-sig") as stream:
|
||
writer = csv.DictWriter(stream, fieldnames=list(values[0]))
|
||
writer.writeheader(); writer.writerows(values)
|
||
|
||
|
||
def analyze() -> dict:
|
||
integrity, metadata, strict = source_integrity()
|
||
received = symbol_quality.Cs8File(RX_PATH)
|
||
stream_start = int(strict["location_after_strict_runs"]["stream_rx_start_sample"])
|
||
expected_starts = [stream_start + int(item["start_sample"]) for item in metadata["layout"]]
|
||
plans = lab066.plans()
|
||
packet_rows: list[dict] = []
|
||
bit_rows: list[dict] = []
|
||
transition_rows: list[dict] = []
|
||
transitions: dict[int, dict] = {}
|
||
tx_headers = []
|
||
|
||
for packet, (layout, frame, rx_start) in enumerate(zip(metadata["layout"], plans, expected_starts), start=1):
|
||
expected_frame = frame.plan.frame_bits.astype(np.uint8)
|
||
expected_header = frame.plan.phy_header_bits.astype(np.uint8)
|
||
tx_headers.append(expected_header)
|
||
legacy = extract_symbols(received, rx_start, PREAMBLE_SYMBOLS + HEADER_SYMBOLS, "legacy")
|
||
t2 = extract_symbols(received, rx_start, PREAMBLE_SYMBOLS + HEADER_SYMBOLS, "t2")
|
||
transition = extract_symbols(
|
||
received, rx_start, PREAMBLE_SYMBOLS + HEADER_SYMBOLS + POSTHEADER_SYMBOLS, "legacy"
|
||
)
|
||
preamble = physical_metrics(legacy, expected_frame, slice(0, PREAMBLE_SYMBOLS))
|
||
local = physical_metrics(legacy, expected_frame, slice(PREAMBLE_SYMBOLS - 128, PREAMBLE_SYMBOLS + HEADER_SYMBOLS))
|
||
reference = reference_sync_diagnostic(
|
||
legacy, expected_frame[:PREAMBLE_SYMBOLS + HEADER_SYMBOLS]
|
||
)
|
||
header_legacy = header_diagnostics(
|
||
legacy["corrected"].real[PREAMBLE_SYMBOLS:], expected_header
|
||
)
|
||
header_t2 = header_diagnostics(
|
||
t2["corrected"].real[PREAMBLE_SYMBOLS:], expected_header
|
||
)
|
||
legacy_t2_differences = int(np.count_nonzero(
|
||
legacy["hard_bits"][PREAMBLE_SYMBOLS:] != t2["hard_bits"][PREAMBLE_SYMBOLS:]
|
||
))
|
||
strict_row = strict_candidate_for_packet(strict["run_a_legacy"]["rows"], rx_start)
|
||
candidate_position = int(strict_row["absolute_refined"]) if strict_row else None
|
||
if packet == 1:
|
||
preceding_start = stream_start
|
||
preceding_count = lab066.GUARD_SAMPLES
|
||
else:
|
||
preceding_count = lab066.INTER_PACKET_GAP_SAMPLES
|
||
preceding_start = rx_start - preceding_count
|
||
gap = gap_metrics(received, preceding_start, preceding_count)
|
||
trailing = gap_metrics(received, rx_start + lab066.PACKET_SAMPLES, lab066.GUARD_SAMPLES) if packet == 16 else None
|
||
row = {
|
||
"packet": packet,
|
||
"strict_result": "PASS" if packet in PASSED_PACKETS else "HEADER_FAIL",
|
||
"candidate_position": candidate_position,
|
||
"expected_packet_position": rx_start,
|
||
"candidate_difference_samples": None if candidate_position is None else candidate_position - rx_start,
|
||
"timing_offset_samples": legacy["timing_samples"],
|
||
"fractional_timing_offset": legacy["timing_fraction"],
|
||
"timing_metric_confidence": None,
|
||
"preamble_symbols": preamble["symbols"],
|
||
"preamble_errors": preamble["errors"],
|
||
"preamble_error_rate": preamble["error_rate"],
|
||
"preamble_correlation": preamble["correlation"],
|
||
"preamble_rms_symbol_deviation": preamble["rms_symbol_deviation"],
|
||
"preamble_matched_power": preamble["matched_component_power"],
|
||
"preamble_residual_power": preamble["residual_power"],
|
||
"preamble_matched_residual_db": preamble["matched_residual_db"],
|
||
"legacy_initial_frequency_hz": legacy["initial"]["frequency_hz"],
|
||
"t2_initial_frequency_hz": t2["initial"]["frequency_hz"],
|
||
"initial_phase_error_rad": float(np.angle(legacy["initial"]["gain"])),
|
||
"frequency_difference_legacy_t2_hz": legacy["initial"]["frequency_hz"] - t2["initial"]["frequency_hz"],
|
||
"reference_frequency_hz": reference["frequency_hz"],
|
||
"legacy_frequency_error_vs_reference_hz": legacy["initial"]["frequency_hz"] - reference["frequency_hz"],
|
||
"reference_timing_offset_samples": reference["timing_offset_samples"],
|
||
"reference_fractional_timing_offset": reference["timing_fraction"],
|
||
"reference_coherence": reference["coherence"],
|
||
"legacy_expected_phase_drift_preamble_rad": float(2 * np.pi * legacy["initial"]["frequency_hz"] * PREAMBLE_SYMBOLS / SYMBOL_RATE),
|
||
"legacy_expected_phase_drift_header_rad": float(2 * np.pi * legacy["initial"]["frequency_hz"] * HEADER_SYMBOLS / SYMBOL_RATE),
|
||
"legacy_expected_phase_drift_full_packet_rad": float(2 * np.pi * legacy["initial"]["frequency_hz"] * len(expected_frame) / SYMBOL_RATE),
|
||
"phase_error_from_cfo_mismatch_over_header_rad": float(
|
||
2 * np.pi * (legacy["initial"]["frequency_hz"] - reference["frequency_hz"])
|
||
* HEADER_SYMBOLS / SYMBOL_RATE
|
||
),
|
||
"expected_header_bits": header_legacy["expected_header_bits"],
|
||
"received_raw_header_bits": header_legacy["received_raw_header_bits"],
|
||
"header_raw_errors": header_legacy["raw_header_errors"],
|
||
"header_copy_1_errors": header_legacy["copy_errors"][0],
|
||
"header_copy_2_errors": header_legacy["copy_errors"][1],
|
||
"header_copy_3_errors": header_legacy["copy_errors"][2],
|
||
"voted_header_bits": header_legacy["voted_header_bits"],
|
||
"header_voted_errors": header_legacy["voted_header_errors"],
|
||
"magic_errors": header_legacy["magic_errors"],
|
||
"version_errors": header_legacy["version_errors"],
|
||
"length_errors": header_legacy["length_errors"],
|
||
"checksum_errors": header_legacy["checksum_errors"],
|
||
"decoded_magic": header_legacy["decoded_magic"],
|
||
"decoded_length": header_legacy["decoded_length"],
|
||
"header_accepted": header_legacy["accepted"],
|
||
"header_reason": header_legacy["reason"],
|
||
"header_raw_error_positions": header_legacy["raw_error_positions"],
|
||
"header_voted_error_positions": header_legacy["voted_error_positions"],
|
||
"legacy_header_bits": header_legacy["received_raw_header_bits"],
|
||
"t2_header_bits": header_t2["received_raw_header_bits"],
|
||
"legacy_t2_header_differences": legacy_t2_differences,
|
||
"local_window": "last 128 preamble symbols plus complete 144-symbol header",
|
||
"local_matched_power": local["matched_component_power"],
|
||
"local_residual_power": local["residual_power"],
|
||
"local_matched_residual_db": local["matched_residual_db"],
|
||
"rms_symbol_deviation": local["rms_symbol_deviation"],
|
||
"header_error_rate": header_legacy["raw_header_errors"] / HEADER_SYMBOLS,
|
||
"preceding_zero_gap_length": gap["length_samples"],
|
||
"gap_background_ac_rms": gap["background_ac_rms"],
|
||
"gap_mean_i": gap["mean_i"], "gap_mean_q": gap["mean_q"],
|
||
"gap_residual_energy": gap["residual_energy_mean"],
|
||
"previous_waveform_tail_detected": gap["previous_waveform_tail_detected"],
|
||
"gap_leading_edge_to_middle_db": gap["leading_edge_to_middle_db"],
|
||
"trailing_guard_packet_16": trailing,
|
||
}
|
||
packet_rows.append(row)
|
||
for position in range(HEADER_SYMBOLS):
|
||
bit_rows.append({
|
||
"packet": packet, "protected_bit_position": position,
|
||
"field_bit_position": position // HEADER_REPETITION,
|
||
"copy": position % HEADER_REPETITION + 1,
|
||
"expected": int(expected_header[position]),
|
||
"legacy_received": int(legacy["hard_bits"][PREAMBLE_SYMBOLS + position]),
|
||
"t2_received": int(t2["hard_bits"][PREAMBLE_SYMBOLS + position]),
|
||
"legacy_error": int(position in header_legacy["raw_error_positions"]),
|
||
"t2_error": int(position in header_t2["raw_error_positions"]),
|
||
})
|
||
start = PREAMBLE_SYMBOLS - 128
|
||
stop = PREAMBLE_SYMBOLS + HEADER_SYMBOLS + 128
|
||
corrected = transition["corrected"][start:stop]
|
||
truth = 2.0 * expected_frame[start:stop] - 1.0
|
||
errors = np.flatnonzero((corrected.real >= 0) != (truth > 0))
|
||
phase_error = np.angle(corrected * truth)
|
||
transitions[packet] = {
|
||
"corrected_real": corrected.real.astype(float).tolist(),
|
||
"corrected_imag": corrected.imag.astype(float).tolist(),
|
||
"magnitude": abs(corrected).astype(float).tolist(),
|
||
"phase_error_rad": phase_error.astype(float).tolist(),
|
||
"hard_decisions": (corrected.real >= 0).astype(int).tolist(),
|
||
"errors": errors.astype(int).tolist(),
|
||
"sample_positions": transition["sample_positions"][start:stop].astype(int).tolist(),
|
||
}
|
||
regions = {"preamble_tail": slice(0, 128), "header": slice(128, 128 + HEADER_SYMBOLS), "postheader": slice(128 + HEADER_SYMBOLS, None)}
|
||
for name, selected in regions.items():
|
||
values = corrected[selected]
|
||
local_truth = truth[selected]
|
||
transition_rows.append({
|
||
"packet": packet, "region": name,
|
||
"symbols": len(values),
|
||
"decision_errors": int(np.count_nonzero((values.real >= 0) != (local_truth > 0))),
|
||
"magnitude_median": float(np.median(abs(values))),
|
||
"magnitude_minimum": float(np.min(abs(values))),
|
||
"phase_error_rms_rad": float(np.sqrt(np.mean(np.angle(values * local_truth) ** 2))),
|
||
"imaginary_rms": float(np.sqrt(np.mean(values.imag ** 2))),
|
||
"timing_offset_samples": transition["timing_samples"],
|
||
})
|
||
|
||
assign_classifications(packet_rows)
|
||
candidate_summary, candidate_rows = candidate_analysis(strict, expected_starts)
|
||
tx_header_identical = all(np.array_equal(tx_headers[0], item) for item in tx_headers[1:])
|
||
tx_header_decodes = [lab066.phy.lab045.decode_phy_header_soft(2.0 * item - 1.0) for item in tx_headers]
|
||
software_current = {
|
||
mode: lab066.decode(mode, TX_PATH, diagnostics=False)
|
||
for mode in ("legacy", "t2")
|
||
}
|
||
software_16_of_16 = all(
|
||
result["packet_output"] == 16 and result["exact_payload_match"] == 16
|
||
for result in software_current.values()
|
||
)
|
||
groups = group_summary(packet_rows)
|
||
distributions = {
|
||
label: {
|
||
metric: distribution(packet_rows, packets, metric)
|
||
for metric in (
|
||
"preamble_errors", "preamble_error_rate", "preamble_correlation",
|
||
"preamble_rms_symbol_deviation", "preamble_matched_power",
|
||
"preamble_residual_power", "preamble_matched_residual_db",
|
||
"header_raw_errors", "header_voted_errors",
|
||
)
|
||
}
|
||
for label, packets in (("passed", PASSED_PACKETS), ("failed", FAILED_PACKETS))
|
||
}
|
||
failed_classes = {str(row["packet"]): row["classification"] for row in packet_rows if row["packet"] in FAILED_PACKETS}
|
||
passed_local_residual = float(np.median([
|
||
row["local_residual_power"] for row in packet_rows if row["packet"] in PASSED_PACKETS
|
||
]))
|
||
failed_local_residual = float(np.median([
|
||
row["local_residual_power"] for row in packet_rows if row["packet"] in FAILED_PACKETS
|
||
]))
|
||
passed_local_ratio = float(np.median([
|
||
row["local_matched_residual_db"] for row in packet_rows if row["packet"] in PASSED_PACKETS
|
||
]))
|
||
failed_local_ratio = float(np.median([
|
||
row["local_matched_residual_db"] for row in packet_rows if row["packet"] in FAILED_PACKETS
|
||
]))
|
||
failed_error_positions = [
|
||
position for row in packet_rows if row["packet"] in FAILED_PACKETS
|
||
for position in row["header_raw_error_positions"]
|
||
]
|
||
repeated_failed_positions = sorted({
|
||
position for position in failed_error_positions
|
||
if failed_error_positions.count(position) > 1
|
||
})
|
||
diagnostic_strict_agreement = all(
|
||
row["header_accepted"] == (row["strict_result"] == "PASS")
|
||
for row in packet_rows
|
||
)
|
||
timing_agreement = all(
|
||
row["timing_offset_samples"] == row["reference_timing_offset_samples"]
|
||
for row in packet_rows
|
||
)
|
||
cfo_scale_small = max(abs(row["phase_error_from_cfo_mismatch_over_header_rad"])
|
||
for row in packet_rows) < 0.1
|
||
candidate_state_independent = (
|
||
candidate_summary["persistent_state_between_rejected_candidates"] is False
|
||
and candidate_summary["accepted_until_changes_only_after_crc_pass"] is True
|
||
)
|
||
local_quality_conditions = {
|
||
"tx_and_software_valid": bool(
|
||
tx_header_identical and all(item["passed"] for item in tx_header_decodes)
|
||
and software_16_of_16
|
||
),
|
||
"strict_header_reproduced": diagnostic_strict_agreement,
|
||
"no_common_failed_error_position": not repeated_failed_positions,
|
||
"failed_local_residual_at_least_20_percent_higher": (
|
||
failed_local_residual > 1.2 * passed_local_residual
|
||
),
|
||
"failed_local_matched_residual_at_least_1_db_lower": (
|
||
failed_local_ratio < passed_local_ratio - 1.0
|
||
),
|
||
"receiver_and_oracle_timing_agree": timing_agreement,
|
||
"cfo_mismatch_phase_scale_below_0p1_rad": cfo_scale_small,
|
||
"candidate_state_independent": candidate_state_independent,
|
||
}
|
||
root_cause = (
|
||
"LOCAL_RF_QUALITY" if all(local_quality_conditions.values())
|
||
else "NOT_LOCALIZED"
|
||
)
|
||
small_fix_evidence = {
|
||
"receiver_header_logic": bool(not diagnostic_strict_agreement),
|
||
"timing_transition": bool(not timing_agreement and not cfo_scale_small),
|
||
"carrier_transition": bool(not cfo_scale_small and timing_agreement),
|
||
"candidate_state_handling": bool(not candidate_state_independent),
|
||
}
|
||
one_small_fix = any(small_fix_evidence.values())
|
||
plots = make_plots(packet_rows, transitions, candidate_summary)
|
||
write_csvs(packet_rows, candidate_rows, bit_rows, transition_rows)
|
||
save_json(OUTPUT / "transition_symbols.json", transitions)
|
||
summary = {
|
||
"lab": "Lab067",
|
||
"mode": "OFFLINE_DIAGNOSTIC_ONLY",
|
||
"rf_tx": 0,
|
||
"hardware_accessed": False,
|
||
"lab066_integrity": integrity,
|
||
"historical_strict_result_unchanged": {
|
||
"legacy": "10/16", "t2": "10/16",
|
||
"passed_packets": sorted(PASSED_PACKETS), "failed_packets": sorted(FAILED_PACKETS),
|
||
},
|
||
"packet_positions_verified": len(expected_starts) == 16 and all(
|
||
row["candidate_difference_samples"] is not None and abs(row["candidate_difference_samples"]) <= 64
|
||
for row in packet_rows
|
||
),
|
||
"packet_rows": packet_rows,
|
||
"passed_failed_distributions": distributions,
|
||
"start_middle_end_groups": groups,
|
||
"failed_packet_classifications": failed_classes,
|
||
"header_error_position_analysis": {
|
||
"failed_raw_error_count": len(failed_error_positions),
|
||
"positions_repeated_between_failed_packets": repeated_failed_positions,
|
||
"systematic_common_position_found": bool(repeated_failed_positions),
|
||
"interpretation": "Позиции ошибок различаются; общей битовой позиции отказа нет.",
|
||
},
|
||
"local_quality_evidence": {
|
||
"passed_local_residual_power_median": passed_local_residual,
|
||
"failed_local_residual_power_median": failed_local_residual,
|
||
"passed_local_matched_residual_db_median": passed_local_ratio,
|
||
"failed_local_matched_residual_db_median": failed_local_ratio,
|
||
"failed_minus_passed_residual_power_median": failed_local_residual - passed_local_residual,
|
||
"failed_minus_passed_matched_residual_db_median": failed_local_ratio - passed_local_ratio,
|
||
"scope": "last 128 preamble symbols plus complete header",
|
||
"decision_conditions": local_quality_conditions,
|
||
},
|
||
"legacy_t2_header_difference": (
|
||
"NONE" if all(row["legacy_t2_header_differences"] == 0 for row in packet_rows)
|
||
else "PRESENT"
|
||
),
|
||
"t2_influence_starts_at_symbol": PREAMBLE_SYMBOLS,
|
||
"t2_explanation": "T2 начинает сопровождение после преамбулы; его решения заголовка сравнены побитно со штатным legacy.",
|
||
"candidate_analysis": candidate_summary,
|
||
"tx_validation": {
|
||
"headers_identical_at_all_packet_positions": tx_header_identical,
|
||
"all_headers_decode_with_correct_magic_version_length_checksum": all(item["passed"] for item in tx_header_decodes),
|
||
"decoded_packet_lengths": [item["packet_bytes"] for item in tx_header_decodes],
|
||
"software_roundtrip_current": software_current,
|
||
"legacy_and_t2_16_of_16": software_16_of_16,
|
||
},
|
||
"header_failure_root_cause": root_cause,
|
||
"one_small_fix_justified": one_small_fix,
|
||
"branch_decision": (
|
||
"ONE_SMALL_FIX_JUSTIFIED" if one_small_fix
|
||
else "NO_SMALL_FIX_JUSTIFIED"
|
||
),
|
||
"small_fix_assessment": {
|
||
"evidence_flags": small_fix_evidence,
|
||
"receiver_header_logic_error_found": small_fix_evidence["receiver_header_logic"],
|
||
"timing_transition_error_found": small_fix_evidence["timing_transition"],
|
||
"carrier_transition_error_explains_six_failures": small_fix_evidence["carrier_transition"],
|
||
"candidate_state_error_found": small_fix_evidence["candidate_state_handling"],
|
||
"reason": "TX и программный тракт дают 16/16; ошибки заголовка не имеют общей позиции, timing совпадает, а локальный остаток в failed header-окнах выше.",
|
||
},
|
||
"achieved_result": "10/16 exact full packets at 500 ksymbol/s, TX VGA 20 dB, hardware-valid OTA capture",
|
||
"recommended_next_project_branch": "CRSF / ELRS CONTROL AND TELEMETRY",
|
||
"qpsk_note": "QPSK с меньшей символьной скоростью остаётся будущим кандидатом для видеоканала; в Lab067 не реализован.",
|
||
"plots": plots,
|
||
"lab067_commit": "NOT PERFORMED",
|
||
"push": "NOT PERFORMED",
|
||
}
|
||
save_json(OUTPUT / "summary.json", summary)
|
||
return summary
|
||
|
||
|
||
def run_functional_tests() -> None:
|
||
expected = np.repeat(np.arange(HEADER_BITS, dtype=np.uint8) % 2, HEADER_REPETITION)
|
||
exact = header_diagnostics(2.0 * expected.astype(float) - 1.0, expected)
|
||
assert exact["raw_header_errors"] == 0
|
||
assert exact["voted_header_errors"] == 0
|
||
changed = expected.copy(); changed[3:5] ^= 1
|
||
result = header_diagnostics(2.0 * changed.astype(float) - 1.0, expected)
|
||
assert result["raw_header_errors"] == 2
|
||
assert result["voted_header_errors"] == 1
|
||
|
||
|
||
def main() -> None:
|
||
run_functional_tests()
|
||
result = analyze()
|
||
print(json.dumps({
|
||
"lab": result["lab"],
|
||
"historical_strict_result": result["historical_strict_result_unchanged"],
|
||
"root_cause": result["header_failure_root_cause"],
|
||
"one_small_fix_justified": result["one_small_fix_justified"],
|
||
"failed_packet_classifications": result["failed_packet_classifications"],
|
||
"plots": result["plots"],
|
||
}, ensure_ascii=False, indent=2))
|
||
print("RF TX = 0")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|