360 lines
16 KiB
Python
360 lines
16 KiB
Python
"""RF-001: файловый IQ → существующий VIDEO-005 receiver → отчёт после CRC."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
import statistics
|
||
import sys
|
||
from typing import Any
|
||
|
||
import numpy as np
|
||
from scipy.signal import oaconvolve
|
||
|
||
if __package__ in (None, ""):
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
||
from protocol.packet import build_packet, parse_packet
|
||
from protocol import punctured_fec
|
||
from protocol.video003_modem import (
|
||
PHY_HEADER_SYMBOLS, PREAMBLE_SYMBOLS, SAMPLE_RATE_HZ, receive_frame,
|
||
)
|
||
from protocol.video005_carrier import (
|
||
correct_pilot_estimate, estimate_from_preamble, track_qpsk_body,
|
||
)
|
||
from protocol.video005_iq_waveform import NOMINAL_CARRIER_HZ, SYMBOL_RATE, rrc_taps
|
||
from protocol.video005_receiver import IqStreamResult, receive_iq_stream
|
||
from protocol.video005_timing import sample_symbols
|
||
from tools.qpsk_rf_common import (
|
||
GAP_MS, PACKET_COUNT, PAYLOAD_BYTES, SEED, TEST_ID,
|
||
payload_for_sequence, read_iq_file, sha256_file,
|
||
)
|
||
from tools.rf001a_streaming import (
|
||
DEFAULT_CHUNK_SAMPLES,
|
||
MemoryPeakSampler,
|
||
cs8_sample_count,
|
||
decode_cs8_streaming,
|
||
)
|
||
|
||
|
||
def _bit_errors(observed: bytes, expected: bytes) -> tuple[int, int]:
|
||
"""Считает ошибки только для сопоставленных пакетов равной длины."""
|
||
if len(observed) != len(expected):
|
||
return 0, 0
|
||
return sum((left ^ right).bit_count() for left, right in zip(
|
||
observed, expected, strict=True
|
||
)), len(expected) * 8
|
||
|
||
|
||
def _raw_ber_after_decode(
|
||
iq: np.ndarray,
|
||
reception: IqStreamResult,
|
||
seed: int,
|
||
decoder_backend: str,
|
||
) -> tuple[int, int, int]:
|
||
"""Повторяет штатные PHY-функции для BER только после строгого первого приёма.
|
||
|
||
Метрика относится к CRC-valid кандидатам. Границы отвергнутых кандидатов
|
||
VIDEO-005 не возвращает; их биты здесь не выдумываются.
|
||
"""
|
||
period = round(SAMPLE_RATE_HZ / NOMINAL_CARRIER_HZ)
|
||
cycle = np.exp(-2j * np.pi * np.arange(period) / period).astype(np.complex64)
|
||
baseband = np.asarray(iq, dtype=np.complex64) * np.resize(cycle, iq.size)
|
||
matched = oaconvolve(baseband, rrc_taps().astype(np.float32), mode="full")
|
||
if decoder_backend == "numba":
|
||
from protocol.video004_viterbi import decode_soft as fec_decoder
|
||
else:
|
||
fec_decoder = punctured_fec.decode_soft
|
||
bit_errors = 0
|
||
bits_compared = 0
|
||
replay_failures = 0
|
||
for packet in reception.packets:
|
||
sequence_number = parse_packet(packet.packet).sequence_number
|
||
coded_count = punctured_fec.coded_bit_count(len(packet.packet) * 8, "2/3")
|
||
symbol_count = PREAMBLE_SYMBOLS + PHY_HEADER_SYMBOLS + (coded_count + 1) // 2
|
||
symbols = sample_symbols(matched, packet.start_sample, 16, symbol_count)
|
||
if len(symbols) != symbol_count:
|
||
replay_failures += 1
|
||
continue
|
||
estimate = estimate_from_preamble(symbols, SYMBOL_RATE)
|
||
corrected = correct_pilot_estimate(symbols, estimate)
|
||
tracked, _ = track_qpsk_body(corrected)
|
||
replay = receive_frame(tracked, "QPSK", "2/3", SYMBOL_RATE,
|
||
maximum_start=0, fec_decoder=fec_decoder)
|
||
if replay.packet != packet.packet or replay.coded_soft is None:
|
||
replay_failures += 1
|
||
continue
|
||
from protocol.video003_modem import build_frame
|
||
expected_bits = build_frame(payload_for_sequence(seed, sequence_number),
|
||
"QPSK", "2/3", sequence_number).coded_bits
|
||
observed_bits = (replay.coded_soft >= 0).astype(np.uint8)
|
||
if len(observed_bits) != len(expected_bits):
|
||
replay_failures += 1
|
||
continue
|
||
bit_errors += int(np.count_nonzero(observed_bits != expected_bits))
|
||
bits_compared += len(expected_bits)
|
||
return bit_errors, bits_compared, replay_failures
|
||
|
||
|
||
def summarize_result(
|
||
reception: IqStreamResult,
|
||
*,
|
||
iq_sample_count: int,
|
||
iq_sha256: str,
|
||
iq_format: str,
|
||
clipping_codes: int | None,
|
||
truth: dict[str, Any] | None,
|
||
raw_bits: tuple[int, int, int] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Выполняет сравнение с эталоном только после завершения IQ-приёма."""
|
||
rows = []
|
||
observed_by_sequence: dict[int, bytes] = {}
|
||
for packet in reception.packets:
|
||
parsed = parse_packet(packet.packet)
|
||
observed_by_sequence.setdefault(parsed.sequence_number, packet.packet)
|
||
rows.append({
|
||
"sequence_number": parsed.sequence_number,
|
||
"packet_sha256": hashlib.sha256(packet.packet).hexdigest(),
|
||
"payload_sha256": hashlib.sha256(parsed.payload).hexdigest(),
|
||
"start_sample": packet.start_sample,
|
||
"preamble_score": packet.preamble_score,
|
||
"cfo_hz": packet.carrier_offset_hz,
|
||
"phase_rad": packet.initial_phase_rad,
|
||
"timing_phase_symbols": packet.timing_phase_symbols,
|
||
"qpsk_quadrant": packet.quadrant,
|
||
"tracking_error_rms_rad": packet.tracking_error_rms_rad,
|
||
})
|
||
sequences = [row["sequence_number"] for row in rows]
|
||
unique_sequences = set(sequences)
|
||
cfo = [row["cfo_hz"] for row in rows]
|
||
report: dict[str, Any] = {
|
||
"rf001_status": "OFFLINE_ANALYZED",
|
||
"test_id": truth.get("test_id") if truth is not None else None,
|
||
"modulation": "QPSK",
|
||
"symbol_rate": 500_000,
|
||
"inner_fec": "2/3",
|
||
"payload_bytes": 256,
|
||
"inter_packet_gap_ms": 5,
|
||
"tx_packets": truth.get("packet_count") if truth is not None else None,
|
||
"iq_format": iq_format,
|
||
"iq_sample_count": iq_sample_count,
|
||
"iq_sha256": iq_sha256,
|
||
"detected_preambles": reception.candidate_count,
|
||
"rejected_candidates": reception.rejected_candidates,
|
||
"decoded_packets": len(rows),
|
||
"crc_valid_packets": len(rows),
|
||
"packet_sequences": sequences,
|
||
"duplicates": len(sequences) - len(unique_sequences),
|
||
"missing_packets": None,
|
||
"exact_packets": None,
|
||
"packet_success_rate": None,
|
||
"raw_ber": None,
|
||
"raw_ber_scope": "UNAVAILABLE_FOR_REJECTED_CANDIDATES_FROM_VIDEO005_API",
|
||
"raw_bit_errors_on_crc_valid_packets": None,
|
||
"raw_bits_compared_on_crc_valid_packets": None,
|
||
"raw_ber_on_crc_valid_packets": None,
|
||
"raw_ber_replay_failures": None,
|
||
"post_fec_bit_errors": None,
|
||
"post_fec_bits_compared": None,
|
||
"post_fec_error_rate": None,
|
||
"post_fec_error_rate_on_compared_packets": None,
|
||
"post_fec_uncomparable_packets": None,
|
||
"cfo_median_hz": statistics.median(cfo) if cfo else None,
|
||
"cfo_p95_absolute_hz": float(np.percentile(np.abs(cfo), 95)) if cfo else None,
|
||
"phase_error_rad": None,
|
||
"timing_offset_symbols": None,
|
||
"phase_and_timing_estimates": rows,
|
||
"tx_underruns": "NOT_RUN",
|
||
"rx_overruns": "NOT_RUN",
|
||
"rx_clipping_rail_codes": clipping_codes if clipping_codes is not None else "NOT_APPLICABLE",
|
||
"common_10mhz_clock": "NOT_RUN",
|
||
"sdr_hardware": "NOT_ACCESSED",
|
||
"rf_tx": 0,
|
||
}
|
||
if truth is None:
|
||
return report
|
||
expected_rows = truth["expected_packet_hashes"]
|
||
expected = {int(item["sequence_number"]): item["packet_sha256"]
|
||
for item in expected_rows}
|
||
if (len(expected) != len(expected_rows) or len(expected) != truth["packet_count"]
|
||
or set(expected) != set(range(PACKET_COUNT))):
|
||
raise ValueError("ведомость содержит повторные или недостающие номера")
|
||
exact = {
|
||
row["sequence_number"] for row in rows
|
||
if expected.get(row["sequence_number"]) == row["packet_sha256"]
|
||
}
|
||
missing = sorted(expected.keys() - unique_sequences)
|
||
seed = int(truth["seed"])
|
||
if (truth["packet_count"] != PACKET_COUNT
|
||
or truth["application_payload_bytes"] != PAYLOAD_BYTES
|
||
or truth["seed"] != SEED):
|
||
raise ValueError("ведомость не соответствует каноническому RF-001")
|
||
for sequence_number, digest in expected.items():
|
||
reference = build_packet(payload_for_sequence(seed, sequence_number),
|
||
message_type=3, sequence_number=sequence_number)
|
||
if hashlib.sha256(reference).hexdigest() != digest:
|
||
raise ValueError("эталонный SHA пакета не совпадает с seed")
|
||
errors = 0
|
||
compared = 0
|
||
uncomparable = 0
|
||
for sequence_number, observed in observed_by_sequence.items():
|
||
if sequence_number not in expected:
|
||
continue
|
||
reference = build_packet(payload_for_sequence(seed, sequence_number),
|
||
message_type=3, sequence_number=sequence_number)
|
||
packet_errors, packet_bits = _bit_errors(observed, reference)
|
||
if packet_bits == 0:
|
||
uncomparable += 1
|
||
errors += packet_errors
|
||
compared += packet_bits
|
||
complete_post_fec = (unique_sequences == expected.keys()
|
||
and len(sequences) == truth["packet_count"]
|
||
and uncomparable == 0)
|
||
report.update({
|
||
"missing_packets": missing,
|
||
"unexpected_sequences": sorted(unique_sequences - expected.keys()),
|
||
"exact_packets": len(exact),
|
||
"packet_success_rate": len(exact) / truth["packet_count"],
|
||
"post_fec_bit_errors": errors,
|
||
"post_fec_bits_compared": compared,
|
||
"post_fec_error_rate": errors / compared if compared and complete_post_fec else None,
|
||
"post_fec_error_rate_on_compared_packets": errors / compared if compared else None,
|
||
"post_fec_uncomparable_packets": uncomparable,
|
||
"same_bytes_as_tx_waveform": iq_sha256 == truth.get("iq_sha256"),
|
||
})
|
||
if raw_bits is not None:
|
||
raw_errors, raw_count, replay_failures = raw_bits
|
||
conditional_ber = raw_errors / raw_count if raw_count else None
|
||
complete = (unique_sequences == expected.keys()
|
||
and len(sequences) == truth["packet_count"]
|
||
and replay_failures == 0)
|
||
report.update({
|
||
"raw_bit_errors_on_crc_valid_packets": raw_errors,
|
||
"raw_bits_compared_on_crc_valid_packets": raw_count,
|
||
"raw_ber_on_crc_valid_packets": conditional_ber,
|
||
"raw_ber_replay_failures": replay_failures,
|
||
"raw_ber": conditional_ber if complete else None,
|
||
"raw_ber_scope": ("ALL_EXPECTED_PACKETS" if complete else
|
||
"CRC_VALID_PACKETS_ONLY; FULL_RAW_BER_UNAVAILABLE"),
|
||
})
|
||
return report
|
||
|
||
|
||
def decode_file(
|
||
iq_path: Path,
|
||
*,
|
||
iq_format: str = "c64le",
|
||
truth_metadata_path: Path | None = None,
|
||
decoder_backend: str = "numba",
|
||
capture_started_at: str | None = None,
|
||
tx_started_at: str | None = None,
|
||
tx_finished_at: str | None = None,
|
||
chunk_samples: int = DEFAULT_CHUNK_SAMPLES,
|
||
) -> dict[str, Any]:
|
||
"""Декодирует c64le целиком, а большие cs8 — ограниченными окнами."""
|
||
|
||
streaming = None
|
||
with MemoryPeakSampler() as memory:
|
||
if iq_format == "cs8":
|
||
streaming = decode_cs8_streaming(
|
||
iq_path,
|
||
capture_started_at=capture_started_at,
|
||
tx_started_at=tx_started_at,
|
||
tx_finished_at=tx_finished_at,
|
||
decoder_backend=decoder_backend,
|
||
chunk_samples=chunk_samples,
|
||
)
|
||
reception = streaming.reception
|
||
iq_sample_count = cs8_sample_count(iq_path)
|
||
clipping_codes = streaming.clipping_codes
|
||
iq = None
|
||
else:
|
||
iq, clipping_codes = read_iq_file(iq_path, iq_format)
|
||
reception = receive_iq_stream(iq, decoder_backend=decoder_backend)
|
||
iq_sample_count = int(iq.size)
|
||
truth = (json.loads(truth_metadata_path.read_text(encoding="utf-8"))
|
||
if truth_metadata_path is not None else None)
|
||
if truth is not None and (
|
||
truth.get("schema_version") != 1
|
||
or truth.get("test_id") != TEST_ID
|
||
or truth.get("modulation") != "QPSK"
|
||
or truth.get("symbol_rate") != SYMBOL_RATE
|
||
or truth.get("sample_rate") != SAMPLE_RATE_HZ
|
||
or truth.get("fec_rate") != "2/3"
|
||
or truth.get("digital_carrier_offset_hz") != NOMINAL_CARRIER_HZ
|
||
or truth.get("packet_count") != PACKET_COUNT
|
||
or truth.get("application_payload_bytes") != PAYLOAD_BYTES
|
||
or truth.get("inter_packet_gap_ms") != GAP_MS
|
||
or truth.get("seed") != SEED
|
||
):
|
||
raise ValueError("TX-ведомость не соответствует режиму RF-001")
|
||
raw_bits = (
|
||
_raw_ber_after_decode(iq, reception, int(truth["seed"]), decoder_backend)
|
||
if truth is not None and iq is not None
|
||
else None
|
||
)
|
||
report = summarize_result(
|
||
reception,
|
||
iq_sample_count=iq_sample_count,
|
||
iq_sha256=sha256_file(iq_path),
|
||
iq_format=iq_format,
|
||
clipping_codes=clipping_codes,
|
||
truth=truth,
|
||
raw_bits=raw_bits,
|
||
)
|
||
report["peak_analysis_memory_bytes"] = memory.peak_bytes
|
||
report["peak_additional_ram_bytes"] = memory.additional_peak_bytes
|
||
if streaming is not None:
|
||
report.update({
|
||
"streaming_decoder": True,
|
||
"window_source": streaming.window_source,
|
||
"analysis_windows": [
|
||
{"start_sample": window.start, "stop_sample": window.stop}
|
||
for window in streaming.windows
|
||
],
|
||
"overlap_samples": streaming.overlap_samples,
|
||
"maximum_loaded_complex_samples": streaming.maximum_loaded_samples,
|
||
"maximum_loaded_iq_bytes": streaming.maximum_loaded_samples * 8,
|
||
"raw_ber_scope": "NOT_COMPUTED_FOR_STREAMING_WINDOWS",
|
||
})
|
||
else:
|
||
report["streaming_decoder"] = False
|
||
return report
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--iq-file", required=True, type=Path)
|
||
parser.add_argument("--iq-format", choices=("c64le", "cs8"), default="c64le")
|
||
parser.add_argument("--truth-metadata", type=Path)
|
||
parser.add_argument("--report", required=True, type=Path)
|
||
parser.add_argument("--numba-packages", type=Path)
|
||
parser.add_argument("--capture-started-at")
|
||
parser.add_argument("--tx-started-at")
|
||
parser.add_argument("--tx-finished-at")
|
||
parser.add_argument("--chunk-samples", type=int, default=DEFAULT_CHUNK_SAMPLES)
|
||
arguments = parser.parse_args()
|
||
if arguments.report.exists():
|
||
parser.error("отчёт уже существует; перезапись запрещена")
|
||
if arguments.numba_packages is not None:
|
||
if not arguments.numba_packages.is_dir():
|
||
parser.error("каталог пакетов Numba отсутствует")
|
||
sys.path.insert(0, str(arguments.numba_packages.resolve()))
|
||
result = decode_file(arguments.iq_file, iq_format=arguments.iq_format,
|
||
truth_metadata_path=arguments.truth_metadata,
|
||
capture_started_at=arguments.capture_started_at,
|
||
tx_started_at=arguments.tx_started_at,
|
||
tx_finished_at=arguments.tx_finished_at,
|
||
chunk_samples=arguments.chunk_samples)
|
||
arguments.report.parent.mkdir(parents=True, exist_ok=True)
|
||
arguments.report.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8")
|
||
print(f"RF-001 FILE RX: CRC={result['crc_valid_packets']}; "
|
||
f"exact={result['exact_packets']}; RF TX=0")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|