"""Точный расчёт допустимых пакетов и радиобюджета VIDEO-003, без SDR.""" from __future__ import annotations import argparse from dataclasses import asdict, dataclass import json import math from pathlib import Path from protocol import punctured_fec from protocol.packet import CRC_SIZE, HEADER_SIZE as PACKET_HEADER_SIZE from protocol.video_stream_transport import HEADER_SIZE as VIDEO_HEADER_SIZE from protocol.video003_modem import MAX_FULL_PACKET_BYTES, PREAMBLE_SYMBOLS, PHY_HEADER_SYMBOLS ROOT = Path(__file__).resolve().parents[1] DEFAULT_OUTPUT = ROOT / "data/processed/video003/design_space.json" PHY_PAYLOADS = (256, 512, 768, MAX_FULL_PACKET_BYTES - PACKET_HEADER_SIZE - CRC_SIZE) GAPS_MS = (5.0, 2.0, 1.0, 0.5, 0.0) 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")) TRANSIENT_SYMBOLS = 42 RRC_ROLLOFF = 0.35 @dataclass(frozen=True) class DesignRow: modulation: str symbol_rate: int fec_rate: str phy_payload_bytes: int video_fragment_bytes: int full_packet_bytes: int gap_ms: float raw_bit_rate_bps: int coded_useful_rate_bps: float effective_fec_rate: float coded_body_bits: int packet_symbols: int packet_airtime_ms: float gap_airtime_ms: float duty_cycle: float useful_video_rate_bps: float overhead_fraction: float occupied_bandwidth_estimate_hz: float samples_per_symbol_at_8msps: float current_lab_generator_implemented: bool format_valid: bool def as_dict(self) -> dict: return asdict(self) def estimate(modulation: str, symbol_rate: int, fec_rate: str, phy_payload_bytes: int, gap_ms: float, video_header_bytes: int = VIDEO_HEADER_SIZE) -> DesignRow: """PHY payload содержит видеозаголовок; полный пакет содержит ещё 12 байт.""" if modulation not in ("BPSK", "QPSK") or symbol_rate <= 0: raise ValueError("неверная модуляция или символьная скорость") if fec_rate not in punctured_fec.PUNCTURE_PATTERNS: raise ValueError("неверная скорость FEC") if gap_ms < 0 or phy_payload_bytes <= video_header_bytes: raise ValueError("неверная нагрузка или пауза") packet_bytes = phy_payload_bytes + PACKET_HEADER_SIZE + CRC_SIZE if packet_bytes > MAX_FULL_PACKET_BYTES: raise ValueError("полный пакет превышает 1024 байта") video_bytes = phy_payload_bytes - video_header_bytes information_bits = packet_bytes * 8 coded_bits = punctured_fec.coded_bit_count(information_bits, fec_rate) bits_per_symbol = 1 if modulation == "BPSK" else 2 symbols = (PREAMBLE_SYMBOLS + PHY_HEADER_SYMBOLS + TRANSIENT_SYMBOLS + math.ceil(coded_bits / bits_per_symbol)) airtime_ms = 1000 * symbols / symbol_rate cycle_ms = airtime_ms + gap_ms raw = symbol_rate * bits_per_symbol useful = video_bytes * 8000 / cycle_ms return DesignRow( modulation, symbol_rate, fec_rate, phy_payload_bytes, video_bytes, packet_bytes, gap_ms, raw, raw * punctured_fec.effective_rate(information_bits, fec_rate), punctured_fec.effective_rate(information_bits, fec_rate), coded_bits, symbols, airtime_ms, gap_ms, airtime_ms / cycle_ms, useful, 1 - useful / raw, symbol_rate * (1 + RRC_ROLLOFF), 8_000_000 / symbol_rate, modulation == "BPSK" and symbol_rate == 500_000 and fec_rate == "1/2" and phy_payload_bytes == 256 and gap_ms == 5, True, ) def measured_video_profiles() -> dict[str, dict]: """Взять максимум двух окон VIDEO-001 для каждого требуемого режима.""" output: dict[str, dict] = {} sources = { "ROBUST": ("comparison_traffic_160k.json", 320, 240, 5), "NORMAL": ("comparison_traffic_320k.json", 640, 360, 10), "HIGH_QUALITY": ("comparison_traffic_320k.json", 854, 480, 10), "720P5": ("comparison_traffic_320k.json", 1280, 720, 5), } for name, (filename, width, height, fps) in sources.items(): source = ROOT / "data/processed/video001" / filename document = json.loads(source.read_text(encoding="utf-8")) cases = [case for case in document["cases"] if case["resolution"] == [width, height] and case["fps"] == fps and case["codec"] == "H264_LIBX264"] if len(cases) != 2: raise ValueError(f"ожидались два измеренных окна для {name}") average = max(case["average_bitrate_bps"] for case in cases) peak = max(case["peak_one_second_bitrate_bps"] for case in cases) output[name] = { "resolution": [width, height], "fps": fps, "measured_average_bps": average, "measured_peak_1s_bps": peak, "required_20pct_bps": 1.2 * peak, "required_50pct_bps": 1.5 * peak, "source_files": [str(source.relative_to(ROOT))], } return output def run() -> dict: profiles = measured_video_profiles() rows = [] for modulation, symbol_rate, fec_rate in MODES: for payload in PHY_PAYLOADS: for gap_ms in GAPS_MS: item = estimate(modulation, symbol_rate, fec_rate, payload, gap_ms).as_dict() item["headroom_by_video_mode"] = { name: item["useful_video_rate_bps"] / profile["measured_peak_1s_bps"] for name, profile in profiles.items() } rows.append(item) best = max(rows, key=lambda item: item["useful_video_rate_bps"]) minimum_rate = math.ceil(600_000 * best["packet_symbols"] / (best["video_fragment_bytes"] * 8)) return { "max_full_phy_packet_bytes": MAX_FULL_PACKET_BYTES, "current_fixed_lab_generator_phy_payload_bytes": 256, "packet_header_bytes": PACKET_HEADER_SIZE, "packet_crc_bytes": CRC_SIZE, "video_header_bytes": VIDEO_HEADER_SIZE, "max_format_phy_payload_bytes": PHY_PAYLOADS[-1], "max_format_video_fragment_bytes": PHY_PAYLOADS[-1] - VIDEO_HEADER_SIZE, "gap_ms": list(GAPS_MS), "measured_video_profiles": profiles, "rows": rows, "best_calculated": best, "minimum_symbol_rate_hz_for_600k_at_best_packet": minimum_rate, "minimum_integer_sps_at_8msps_symbol_rate_hz": 8_000_000 / 15, "rf_tx": 0, "method": "exact puncturing length, ideal continuous packet schedule, zero loss", } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) args = parser.parse_args() result = run() args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") best = result["best_calculated"] print(f"VIDEO-003: {len(result['rows'])} строк, максимум " f"{best['useful_video_rate_bps'] / 1000:.3f} кбит/с; {args.output}") if __name__ == "__main__": main()