"""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()