"""Воспроизводимый AWGN-контроль BPSK/QPSK и проколотого FEC VIDEO-003.""" from __future__ import annotations import argparse from collections import defaultdict import json import math from pathlib import Path import numpy as np from protocol import punctured_fec from protocol.video003_modem import build_frame, receive_frame, rrc_roundtrip ROOT = Path(__file__).resolve().parents[1] DEFAULT_OUTPUT = ROOT / "data/processed/video003/synthetic_phy.json" MODES = (("BPSK", 500_000, "1/2"), ("QPSK", 250_000, "1/2"), ("QPSK", 250_000, "2/3"), ("QPSK", 250_000, "3/4"), ("QPSK", 500_000, "1/2"), ("QPSK", 500_000, "2/3"), ("QPSK", 500_000, "3/4")) EBN0_DB = (0.0, 3.0, 6.0, 9.0) TRIALS = 32 SEED = 20260916 PHY_PAYLOAD_BYTES = 256 CARRIER_OFFSET_HZ = 73.0 def _channel(symbols: np.ndarray, modulation: str, rate: str, symbol_rate: int, ebn0_db: float, random_generator: np.random.Generator ) -> tuple[np.ndarray, float]: """Единичная энергия символа; Eb относится к информационным битам тела. Фиксированные PHY overhead и пауза не входят в это определение Eb/N0. Поэтому сравнение помехоустойчивости не подменяет расчёт goodput. """ bits_per_symbol = 1 if modulation == "BPSK" else 2 info_bits = (PHY_PAYLOAD_BYTES + 12) * 8 coded_bits = punctured_fec.coded_bit_count(info_bits, rate) body_symbols = math.ceil(coded_bits / bits_per_symbol) n0 = (body_symbols / info_bits) / 10 ** (ebn0_db / 10) sigma = math.sqrt(n0 / 2) phase = random_generator.uniform(-math.pi, math.pi) if modulation == "QPSK": phase += math.pi / 2 indexes = np.arange(len(symbols), dtype=np.float64) rotated = symbols * np.exp(1j * (phase + 2 * math.pi * CARRIER_OFFSET_HZ * indexes / symbol_rate)) noise = sigma * (random_generator.standard_normal(len(symbols)) + 1j * random_generator.standard_normal(len(symbols))) leading = sigma * (random_generator.standard_normal(11) + 1j * random_generator.standard_normal(11)) return np.concatenate((leading, rotated + noise)), n0 def run(trials: int = TRIALS, seed: int = SEED) -> dict: if trials < 1: raise ValueError("требуется хотя бы одна реализация") random_generator = np.random.default_rng(seed) records = [] for modulation, symbol_rate, rate in MODES: for ebn0_db in EBN0_DB: for trial in range(trials): payload = random_generator.integers( 0, 256, PHY_PAYLOAD_BYTES, dtype=np.uint8).tobytes() frame = build_frame(payload, modulation, rate) received, n0 = _channel(frame.symbols, modulation, rate, symbol_rate, ebn0_db, random_generator) outcome = receive_frame(received, modulation, rate, symbol_rate) pre_errors = pre_bits = post_errors = post_bits = None if outcome.coded_soft is not None: pre_errors = int(np.count_nonzero( (outcome.coded_soft >= 0).astype(np.uint8) != frame.coded_bits)) pre_bits = len(frame.coded_bits) if outcome.packet is not None: recovered = np.unpackbits(np.frombuffer(outcome.packet, dtype=np.uint8)) original = np.unpackbits(np.frombuffer(frame.packet, dtype=np.uint8)) post_errors = int(np.count_nonzero(recovered != original)) post_bits = len(original) records.append({ "modulation": modulation, "symbol_rate": symbol_rate, "fec_rate": rate, "ebn0_db": ebn0_db, "trial": trial, "n0_complex": n0, "preamble_found": outcome.preamble_found, "phy_header_pass": outcome.phy_header_pass, "fec_exact_packet": outcome.packet == frame.packet, "crc_pass": outcome.crc_pass, "pre_fec_bit_errors": pre_errors, "pre_fec_bits_compared": pre_bits, "post_fec_bit_errors": post_errors, "post_fec_bits_compared": post_bits, "estimated_cfo_hz": outcome.frequency_offset_hz, }) groups: dict[tuple, list[dict]] = defaultdict(list) for record in records: groups[(record["modulation"], record["symbol_rate"], record["fec_rate"], record["ebn0_db"])].append(record) summaries = [] for key, items in groups.items(): pre_errors = sum(item["pre_fec_bit_errors"] or 0 for item in items) pre_bits = sum(item["pre_fec_bits_compared"] or 0 for item in items) post_errors = sum(item["post_fec_bit_errors"] or 0 for item in items) post_bits = sum(item["post_fec_bits_compared"] or 0 for item in items) summaries.append({ "modulation": key[0], "symbol_rate": key[1], "fec_rate": key[2], "ebn0_db": key[3], "trials": len(items), "preamble_found_count": sum(item["preamble_found"] for item in items), "phy_header_pass_count": sum(item["phy_header_pass"] for item in items), "fec_exact_packet_count": sum(item["fec_exact_packet"] for item in items), "crc_pass_count": sum(item["crc_pass"] for item in items), "pre_fec_ber_on_reached_bodies": pre_errors / pre_bits if pre_bits else None, "pre_fec_bits_compared": pre_bits, "post_fec_ber_on_decoded_packets": post_errors / post_bits if post_bits else None, "post_fec_bits_compared": post_bits, }) rrc = [] for symbol_rate in (250_000, 500_000): frame = build_frame(bytes(range(PHY_PAYLOAD_BYTES)), "QPSK", "3/4") shaped = rrc_roundtrip(frame.symbols, symbol_rate) outcome = receive_frame(shaped, "QPSK", "3/4", symbol_rate) rrc.append({"symbol_rate": symbol_rate, "samples_per_symbol": 8_000_000 // symbol_rate, "packet_crc_pass": outcome.crc_pass, "packet_exact": outcome.packet == frame.packet}) return { "seed": seed, "trials_per_point": trials, "ebn0_db": list(EBN0_DB), "packet_phy_payload_bytes": PHY_PAYLOAD_BYTES, "carrier_offset_hz": CARRIER_OFFSET_HZ, "energy_condition": "Es=1; Eb=Es*body_symbols/information_bits, includes QPSK padding but excludes fixed frame overhead", "channel": "symbol-spaced complex AWGN plus fixed CFO and random phase; 11 leading noise symbols", "pre_ber_scope": "only reached bodies, with denominator recorded", "post_ber_scope": "only decoded packets, with denominator recorded", "records": records, "summary": summaries, "rrc_controls": rrc, "rf_tx": 0, } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--trials", type=int, default=TRIALS) parser.add_argument("--seed", type=int, default=SEED) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) args = parser.parse_args() result = run(args.trials, args.seed) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(f"VIDEO-003: {len(result['records'])} AWGN-реализаций; {args.output}") if __name__ == "__main__": main()