diff --git a/PROJECT_LOG.md b/PROJECT_LOG.md index 3f8a36d..aa6d4f6 100644 --- a/PROJECT_LOG.md +++ b/PROJECT_LOG.md @@ -96,3 +96,24 @@ git version 2.51.1.windows.1 - Payload 512 байт выбран для дальнейшей разработки. - Payload 1024 байта оставлен резервным вариантом. - Подтверждено, что без исправления ошибок возможны длительные прерывания видеопотока. + +--- + +# Запись 005 + +## Дата + +29 июля 2026 года + +## Тема + +Завершение Lab030: блочное исправление стираний. + +## Выполнено + +- Реализовано блочное исправление стираний над GF(256). +- Проверены режимы без исправления, 8+1, 8+2 и 8+4. +- Режим 8+2 выбран основным рабочим режимом. +- Режим 8+4 оставлен контрольным. +- Подтверждено, что без перемежения продолжительные серии потерь не исправляются. +- Следующая лабораторная посвящена перемежению пакетов. diff --git a/data/processed/lab030/lab030_composite_success.png b/data/processed/lab030/lab030_composite_success.png new file mode 100644 index 0000000..ca1f27b Binary files /dev/null and b/data/processed/lab030/lab030_composite_success.png differ diff --git a/data/processed/lab030/lab030_delay_queue.png b/data/processed/lab030/lab030_delay_queue.png new file mode 100644 index 0000000..21eb80e Binary files /dev/null and b/data/processed/lab030/lab030_delay_queue.png differ diff --git a/data/processed/lab030/lab030_mode_comparison.png b/data/processed/lab030/lab030_mode_comparison.png new file mode 100644 index 0000000..b0ac51c Binary files /dev/null and b/data/processed/lab030/lab030_mode_comparison.png differ diff --git a/data/processed/lab030/lab030_no_image_duration.png b/data/processed/lab030/lab030_no_image_duration.png new file mode 100644 index 0000000..9c399c5 Binary files /dev/null and b/data/processed/lab030/lab030_no_image_duration.png differ diff --git a/data/processed/lab030/lab030_report.txt b/data/processed/lab030/lab030_report.txt new file mode 100644 index 0000000..3b82593 --- /dev/null +++ b/data/processed/lab030/lab030_report.txt @@ -0,0 +1,87 @@ +Lab030. Пакетное избыточное кодирование стираний + +Исходный профиль и неизменный внутренний транспорт +- Видео: data\raw\lab026_rover_source.mp4 +- 63 реальных пар BASE/ROI, 3 fps; BASE 240x135 grayscale JPEG Q23, ROI 320x180 grayscale JPEG Q33. +- Внутренний Lab028 packet не изменён: payload 512 байт, 32-byte header, packet CRC32 и object CRC32. + +GF(256) и блочный код +- Примитивный полином: 0x11D (x^8+x^4+x^3+x^2+1). +- Матрица Вандермонда n×k умножается на обратную верхнюю k×k матрицу и становится систематической. +- Любые k доступных строк систематической генераторной матрицы восстанавливают k исходных символов. +- Нулевое дополнение используется только в математике; восстановленный внутренний пакет обрезается по payload_length Lab028 и проходит собственный CRC. + +Внешний пакет FEC +- struct: !4sBBHIHHHHHHII +- Размер: 32 байта, network byte order, padding отсутствует. +- Layout: magic[4]@0, version:u8@4, flags:u8@5, header_size:u16@6, block_id:u32@8, symbol_index:u16@12, k:u16@14, r:u16@16, symbol_size:u16@18, data_length:u16@20, reserved16:u16@22, reserved32:u32@24, outer_crc32:u32@28. +- flags bit0: parity. CRC32 вычисляется по заголовку с нулевым outer_crc32 и данным внешнего символа. + +Формирование блоков и передача +- Последовательный поток внутренних пакетов BASE→ROI делится на k=8; блок может пересекать границу кадра. +- Сначала передаются systematic symbols, затем parity symbols этого блока; перемежение отсутствует. +- Последний блок содержит 6 исходных пакетов. Для него r_last=max(1, ceil(k_last*r/8)): режимы 8+1, 8+2, 8+4 получают соответственно 1, 2 и 3 parity. +- Режим Без FEC передаёт исходные Lab028 packets без внешней обёртки; остальные режимы используют внешний заголовок. +- FIFO учитывает время генерации кадра, все внешние пакеты, фактический размер и скорость 300 кбит/с. Очередь не сбрасывается на границах кадров. + +Скорость, избыточность и очередь +mode | source packets | parity packets | last block k+r | JPEG kbit/s | inner kbit/s | outer kbit/s | overhead | queue mean/max | timeline s +----:|---------------:|---------------:|---------------:|------------:|-------------:|-------------:|---------:|---------------:|----------: +Без FEC | 862 | 0 | none | 157.770 | 168.396 | 168.396 | 6.310% | 13.717/15 | 20.869680 +8+1 | 862 | 108 | 6+1 | 157.770 | 168.396 | 202.987 | 22.276% | 15.452/18 | 20.928560 +8+2 | 862 | 216 | 6+2 | 157.770 | 168.396 | 226.951 | 30.483% | 17.210/21 | 20.974640 +8+4 | 862 | 431 | 6+3 | 157.770 | 168.396 | 274.659 | 42.558% | 20.859/26 | 21.051440 + +Временная модель +- Экспоненциальные Good/Bad интервалы Lab029B, Bad≈2%, mean Bad 10/50/200/1000 мс, 200 повторов, fixed seeds 300300...300303. +- Пакет теряется при любом пересечении его передачи с Bad. +- FEC исправляет стирания; CRC обнаруживает повреждения, но битовые ошибки отдельно не добавляются. + +Функциональные проверки +- PASS gf256_arithmetic: inverse, division, and distributivity passed +- PASS fec_without_loss: systematic 8+4 block is byte-exact without loss +- PASS recover_any_r_erasures: all 549 exact-r erasure combinations recovered +- PASS r_plus_one_not_guaranteed: 8+2 correctly rejected seven available symbols +- PASS restored_inner_packet_and_crc: restored Lab028 packets are byte-exact and CRC-valid +- PASS outer_crc_detection: outer payload bit flip rejected by outer CRC32 +- PASS zero_bad_100_percent: all modes restored all 63 frames without Bad +- PASS fixed_seed_reproducibility: identical seed produced identical result fields +- PASS atomic_incomplete_composite: missing ROI fragment prevented atomic publication +- PASS queue_counts_parity: 431 parity packets included in FIFO timing and queue metrics + +Результаты Monte Carlo +Bad ms | mode | Bad actual | lost src/parity | recovered pkt | blocks recovered/failed | block recovery | composite | BASE-only/ROI-only | no image mean/p95/max s | delay mean/p95/max s | video kbit/s +------:|-----:|-----------:|----------------:|--------------:|------------------------:|---------------:|----------:|------------------:|-------------------------:|----------------------:|------------: +10 | Без FEC | 2.024% | 8130/0 | 0 | 0/0 | 0.000% | 66.960% | 2341/1207 | 0.814/1.346/3.008 | 0.185/0.203/0.210 | 105.091 +10 | 8+1 | 2.038% | 8457/1121 | 2390 | 2390/2725 | 46.725% | 78.873% | 1471/776 | 0.767/1.317/2.360 | 0.231/0.364/0.451 | 123.422 +10 | 8+2 | 2.055% | 8451/2229 | 5560 | 4070/1028 | 79.835% | 91.286% | 600/295 | 0.731/1.017/1.798 | 0.260/0.395/0.466 | 142.607 +10 | 8+4 | 1.990% | 8364/4311 | 7797 | 4860/144 | 97.122% | 98.698% | 84/36 | 0.710/1.017/1.069 | 0.308/0.410/0.496 | 153.620 +50 | Без FEC | 1.914% | 4062/0 | 0 | 0/0 | 0.000% | 91.452% | 529/296 | 0.693/0.998/1.687 | 0.185/0.203/0.210 | 143.553 +50 | 8+1 | 1.929% | 4255/562 | 325 | 325/1216 | 21.090% | 91.762% | 542/263 | 0.709/1.004/1.994 | 0.223/0.245/0.451 | 143.615 +50 | 8+2 | 1.995% | 4326/1104 | 971 | 658/905 | 42.099% | 93.333% | 423/214 | 0.713/1.010/1.632 | 0.248/0.275/0.465 | 145.767 +50 | 8+4 | 2.035% | 4506/2275 | 2062 | 979/520 | 65.310% | 95.532% | 261/141 | 0.749/1.057/1.406 | 0.298/0.339/0.495 | 148.638 +200 | Без FEC | 2.117% | 3877/0 | 0 | 0/0 | 0.000% | 96.071% | 146/83 | 0.786/1.325/2.333 | 0.185/0.203/0.210 | 150.881 +200 | 8+1 | 2.169% | 3983/513 | 74 | 74/734 | 9.158% | 95.865% | 162/91 | 0.810/1.332/2.334 | 0.221/0.244/0.448 | 150.129 +200 | 8+2 | 1.945% | 3573/908 | 144 | 101/602 | 14.367% | 96.302% | 149/80 | 0.837/1.339/2.334 | 0.246/0.275/0.467 | 150.483 +200 | 8+4 | 1.940% | 3550/1777 | 375 | 160/522 | 23.460% | 96.310% | 153/90 | 0.900/1.348/2.334 | 0.296/0.337/0.466 | 149.944 +1000 | Без FEC | 1.722% | 3014/0 | 0 | 0/0 | 0.000% | 97.976% | 27/10 | 1.403/3.050/4.327 | 0.185/0.203/0.210 | 153.809 +1000 | 8+1 | 1.720% | 3003/374 | 13 | 13/423 | 2.982% | 97.921% | 33/13 | 1.437/3.040/4.311 | 0.221/0.244/0.250 | 153.287 +1000 | 8+2 | 1.763% | 3079/781 | 25 | 16/427 | 3.612% | 97.833% | 39/15 | 1.466/3.048/4.295 | 0.246/0.275/0.450 | 152.807 +1000 | 8+4 | 1.719% | 3004/1509 | 64 | 27/400 | 6.323% | 97.817% | 47/21 | 1.533/3.095/4.275 | 0.296/0.337/0.465 | 152.233 + +Допущения и ограничения +- Не реализованы перемежение, ARQ, повторные передачи, команды управления, телеметрия и реальный SDR. +- Любое частичное пересечение Bad уничтожает весь внешний или базовый внутренний пакет. +- Принятые systematic packets немедленно поступают reassembler; восстановленные стирания становятся доступны при получении k символов блока. +- Publication delay измеряется от frame_id/3 до атомарной выдачи BASE+ROI. +- Queue length измеряется в моменты генерации пакетов и включает ожидающие и обслуживаемый пакет. +- Окончательный режим FEC автоматически не выбирается. + +Артефакты +- CSV: data\processed\lab030\lab030_results.csv +- Composite success: data\processed\lab030\lab030_composite_success.png +- No-image duration: data\processed\lab030\lab030_no_image_duration.png +- Stream rate/overhead: data\processed\lab030\lab030_stream_rate_overhead.png +- Delay/queue: data\processed\lab030\lab030_delay_queue.png +- Mode comparison: data\processed\lab030\lab030_mode_comparison.png +- JPEG, внутренние/внешние packets и бинарные дампы не сохранялись. diff --git a/data/processed/lab030/lab030_results.csv b/data/processed/lab030/lab030_results.csv new file mode 100644 index 0000000..d2d7490 --- /dev/null +++ b/data/processed/lab030/lab030_results.csv @@ -0,0 +1,17 @@ +mode,source_block_size,nominal_parity_count,mean_bad_duration_ms,mean_good_duration_ms,target_bad_time_fraction,actual_bad_time_fraction,monte_carlo_repetitions,seed,source_jpeg_bitrate_kbps,inner_packet_stream_bitrate_kbps,outer_fec_stream_bitrate_kbps,service_and_parity_percent,control_stream_bitrate_kbps,schedule_duration_seconds,mean_queue_length_packets,max_queue_length_packets,mean_publication_delay_seconds,p95_publication_delay_seconds,max_publication_delay_seconds,transmitted_source_packets,transmitted_parity_packets,lost_source_packets,lost_parity_packets,recovered_source_packets,fec_recovered_blocks,fec_unrecoverable_blocks,fec_affected_block_recovery_rate,base_objects_completed,roi_objects_completed,atomic_composite_frames_completed,composite_success_rate,base_only_frames,roi_only_frames,incomplete_frames,mean_no_new_image_duration_seconds,p95_no_new_image_duration_seconds,max_no_new_image_duration_seconds,mean_consecutive_incomplete_frames,p95_consecutive_incomplete_frames,max_consecutive_incomplete_frames,effective_delivered_video_bitrate_kbps +none,8,0,10,490,0.02,0.0202372337783,200,300300,157.76975923,168.396019262,168.396019262,6.3102798265,300,20.86968,13.716937355,15,0.184972089605,0.203013333333,0.20968,172400,0,8130,0,0,0,0,0,10778,9644,8437,0.669603174603,2341,1207,4163,0.813657881797,1.34557333333,3.0084,1.47624113475,3,8,105.090548585 +8+1,8,1,10,490,0.02,0.0203834086527,200,300300,157.76975923,168.396019262,202.986837881,22.2758673043,300,20.92856,15.4515463918,18,0.2311155162,0.364053333333,0.451386666667,172400,21600,8457,1121,2390,2390,2725,0.467253176931,11409,10714,9938,0.78873015873,1471,776,2662,0.766680350008,1.31661333333,2.35997333333,1.31847449232,3,6,123.421642005 +8+2,8,2,10,490,0.02,0.0205493425986,200,300300,157.76975923,168.396019262,226.951396469,30.4830189704,300,20.97464,17.2096474954,21,0.260180715238,0.394773333333,0.4656,172400,43200,8451,2229,5560,4070,1028,0.798352295018,12102,11797,11502,0.912857142857,600,295,1098,0.730752592593,1.01705733333,1.7976,1.19607843137,2,4,142.607264773 +8+4,8,4,10,490,0.02,0.0199022658407,200,300300,157.76975923,168.396019262,274.658619583,42.55787076,300,21.05144,20.8592420727,26,0.308434444087,0.410133333333,0.49632,172400,86200,8364,4311,7797,4860,144,0.971223021583,12520,12472,12436,0.986984126984,84,36,164,0.709739906103,1.01696933333,1.06861333333,1.15492957746,2,2,153.619731477 +none,8,0,50,2450,0.02,0.0191374257321,200,300301,157.76975923,168.396019262,168.396019262,6.3102798265,300,20.86968,13.716937355,15,0.185003224855,0.203013333333,0.20968,172400,0,4062,0,0,0,0,0,12052,11819,11523,0.914523809524,529,296,1077,0.69345654043,0.998314666667,1.68717333333,1.10235414534,2,4,143.552817293 +8+1,8,1,50,2450,0.02,0.0192899572498,200,300301,157.76975923,168.396019262,202.986837881,22.2758673043,300,20.92856,15.4515463918,18,0.222813291818,0.244506666667,0.451386666667,172400,21600,4255,562,325,325,1216,0.210902011681,12104,11825,11562,0.917619047619,542,263,1038,0.709203642987,1.00418666667,1.99370666667,1.13442622951,2,5,143.614717878 +8+2,8,2,50,2450,0.02,0.0199477213011,200,300301,157.76975923,168.396019262,226.951396469,30.4830189704,300,20.97464,17.2096474954,21,0.248295249433,0.275226666667,0.465306666667,172400,43200,4326,1104,971,658,905,0.420985284709,12183,11974,11760,0.933333333333,423,214,840,0.712576776557,1.009596,1.63154666667,1.15384615385,2,4,145.767170259 +8+4,8,4,50,2450,0.02,0.0203533945389,200,300301,157.76975923,168.396019262,274.658619583,42.55787076,300,21.05144,20.8592420727,26,0.298285623771,0.338693333333,0.49544,172400,86200,4506,2275,2062,979,520,0.653102068045,12298,12178,12037,0.955317460317,261,141,563,0.749229398664,1.05712,1.40586666667,1.25389755011,2,3,148.638096016 +none,8,0,200,9800,0.02,0.0211716551168,200,300302,157.76975923,168.396019262,168.396019262,6.3102798265,300,20.86968,13.716937355,15,0.18509811593,0.203013333333,0.20968,172400,0,3877,0,0,0,0,0,12251,12188,12105,0.960714285714,146,83,495,0.786258898226,1.325184,2.33330666667,1.38655462185,3,6,150.881316819 +8+1,8,1,200,9800,0.02,0.0216871677039,200,300302,157.76975923,168.396019262,202.986837881,22.2758673043,300,20.92856,15.4515463918,18,0.221480990148,0.244346666667,0.448133333333,172400,21600,3983,513,74,74,734,0.0915841584158,12241,12170,12079,0.958650793651,162,91,521,0.809597474466,1.332496,2.33416,1.45125348189,3,6,150.12864908 +8+2,8,2,200,9800,0.02,0.0194531290427,200,300302,157.76975923,168.396019262,226.951396469,30.4830189704,300,20.97464,17.2096474954,21,0.246329619252,0.275066666667,0.466746666667,172400,43200,3573,908,144,101,602,0.143669985775,12283,12214,12134,0.963015873016,149,80,466,0.837464218579,1.33874666667,2.33416,1.52786885246,3,6,150.482540821 +8+4,8,4,200,9800,0.02,0.0194004010436,200,300302,157.76975923,168.396019262,274.658619583,42.55787076,300,21.05144,20.8592420727,26,0.295920402143,0.336666666667,0.465946666667,172400,86200,3550,1777,375,160,522,0.234604105572,12288,12225,12135,0.963095238095,153,90,465,0.899563467643,1.34797333333,2.33416,1.7032967033,3,6,149.943889824 +none,8,0,1000,49000,0.02,0.0172219900037,200,300303,157.76975923,168.396019262,168.396019262,6.3102798265,300,20.86968,13.716937355,15,0.185021764277,0.203013333333,0.20968,172400,0,3014,0,0,0,0,0,12372,12355,12345,0.979761904762,27,10,255,1.40302564103,3.049944,4.32696,3.26923076923,8.15,12,153.808757968 +8+1,8,1,1000,49000,0.02,0.0172016717987,200,300303,157.76975923,168.396019262,202.986837881,22.2758673043,300,20.92856,15.4515463918,18,0.221316092289,0.244346666667,0.250373333333,172400,21600,3003,374,13,13,423,0.0298165137615,12371,12351,12338,0.979206349206,33,13,262,1.43732205128,3.039692,4.31074666667,3.35897435897,8.15,12,153.287364252 +8+2,8,2,1000,49000,0.02,0.0176312789901,200,300303,157.76975923,168.396019262,226.951396469,30.4830189704,300,20.97464,17.2096474954,21,0.24594955031,0.275066666667,0.449946666667,172400,43200,3079,781,25,16,427,0.0361173814898,12366,12342,12327,0.978333333333,39,15,273,1.46630616034,3.04793866667,4.29538666667,3.45569620253,8.1,12,152.806923027 +8+4,8,4,1000,49000,0.02,0.0171897343344,200,300303,157.76975923,168.396019262,274.658619583,42.55787076,300,21.05144,20.8592420727,26,0.295510529817,0.336506666667,0.464693333333,172400,86200,3004,1509,64,27,400,0.0632318501171,12372,12346,12325,0.978174603175,47,21,275,1.53289614035,3.09500666667,4.27541333333,3.61842105263,8.25,12,152.233086193 diff --git a/data/processed/lab030/lab030_stream_rate_overhead.png b/data/processed/lab030/lab030_stream_rate_overhead.png new file mode 100644 index 0000000..95a409f Binary files /dev/null and b/data/processed/lab030/lab030_stream_rate_overhead.png differ diff --git a/protocol/packet_erasure_fec.py b/protocol/packet_erasure_fec.py new file mode 100644 index 0000000..5737fcf --- /dev/null +++ b/protocol/packet_erasure_fec.py @@ -0,0 +1,613 @@ +""" +Systematic packet-erasure FEC over GF(256) for Lab028 wire packets. + +The code uses a Vandermonde generator matrix over GF(256), transformed to +systematic form. The primitive polynomial is x^8+x^4+x^3+x^2+1 (0x11D). +Any k available symbols from an n=k+r block reconstruct all k source +symbols. + +Each source or parity symbol is carried in a fixed 32-byte outer header:: + + !4sBBHIHHHHHHII + + Offset Size Field + 0 4 magic (b"SFE1") + 4 1 version + 5 1 flags (bit 0: parity) + 6 2 header_size + 8 4 block_id + 12 2 symbol_index + 14 2 source_count (k) + 16 2 parity_count (r) + 18 2 symbol_size + 20 2 data_length + 22 2 reserved16 (zero) + 24 4 reserved32 (zero) + 28 4 outer_crc32 + +Source symbols carry the original serialized Lab028 packet without padding. +Parity symbols carry symbol_size bytes. Zero padding is applied only during +GF(256) mathematics. A recovered source symbol is trimmed from the Lab028 +payload_length field and must pass the original Lab028 packet CRC32. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from functools import lru_cache +import struct +import zlib + +from protocol.video_packet import ( + HEADER_FORMAT as INNER_HEADER_FORMAT, + HEADER_SIZE as INNER_HEADER_SIZE, + MAGIC as INNER_MAGIC, + VERSION as INNER_VERSION, + decode_packet as decode_inner_packet, +) + + +GF_PRIMITIVE_POLYNOMIAL = 0x11D +GF_FIELD_SIZE = 256 +GF_ORDER = 255 + +OUTER_MAGIC = b"SFE1" +OUTER_VERSION = 1 +OUTER_FLAG_PARITY = 0x01 +OUTER_HEADER_FORMAT = "!4sBBHIHHHHHHII" +OUTER_HEADER_SIZE = struct.calcsize(OUTER_HEADER_FORMAT) +MAX_SYMBOL_SIZE = 0xFFFF +MAX_BLOCK_SYMBOLS = 0xFF + + +class FECError(ValueError): + """Base class for outer packets and erasure-code failures.""" + + +class OuterPacketCRCError(FECError): + """The outer packet failed its CRC32 check.""" + + +class InsufficientSymbolsError(FECError): + """Fewer than k unique symbols are available.""" + + +class MatrixSingularError(FECError): + """A GF(256) matrix has no inverse.""" + + +@dataclass(frozen=True) +class OuterSymbol: + """Decoded source or parity symbol after outer CRC validation.""" + + block_id: int + symbol_index: int + source_count: int + parity_count: int + symbol_size: int + data: bytes + is_parity: bool + outer_crc32: int = 0 + + +@dataclass(frozen=True) +class DecodedFECBlock: + """All reconstructed Lab028 packets from one FEC block.""" + + block_id: int + source_packets: tuple[bytes, ...] + recovered_indices: tuple[int, ...] + + +def _build_gf_tables() -> tuple[ + tuple[int, ...], + tuple[int, ...], +]: + exponent = [0] * (GF_ORDER * 2) + logarithm = [0] * GF_FIELD_SIZE + value = 1 + for index in range(GF_ORDER): + exponent[index] = value + logarithm[value] = index + value <<= 1 + if value & GF_FIELD_SIZE: + value ^= GF_PRIMITIVE_POLYNOMIAL + for index in range(GF_ORDER, GF_ORDER * 2): + exponent[index] = exponent[index - GF_ORDER] + return tuple(exponent), tuple(logarithm) + + +GF_EXP, GF_LOG = _build_gf_tables() + + +def gf_add(left: int, right: int) -> int: + """Addition/subtraction in GF(256).""" + + return left ^ right + + +def gf_mul(left: int, right: int) -> int: + """Multiplication in GF(256).""" + + if left == 0 or right == 0: + return 0 + return GF_EXP[GF_LOG[left] + GF_LOG[right]] + + +def gf_div(dividend: int, divisor: int) -> int: + """Division in GF(256).""" + + if divisor == 0: + raise ZeroDivisionError("GF(256) division by zero") + if dividend == 0: + return 0 + return GF_EXP[ + (GF_LOG[dividend] - GF_LOG[divisor]) % GF_ORDER + ] + + +def gf_inverse(value: int) -> int: + """Multiplicative inverse in GF(256).""" + + if value == 0: + raise ZeroDivisionError("zero has no GF(256) inverse") + return GF_EXP[GF_ORDER - GF_LOG[value]] + + +def gf_pow(value: int, exponent: int) -> int: + """Non-negative integer power in GF(256).""" + + if exponent < 0: + raise ValueError("negative GF exponent is unsupported") + if exponent == 0: + return 1 + if value == 0: + return 0 + return GF_EXP[(GF_LOG[value] * exponent) % GF_ORDER] + + +GF_MUL_TRANSLATIONS = tuple( + bytes(gf_mul(coefficient, value) for value in range(256)) + for coefficient in range(256) +) + + +def matrix_multiply( + left: tuple[tuple[int, ...], ...], + right: tuple[tuple[int, ...], ...], +) -> tuple[tuple[int, ...], ...]: + """Multiply two small matrices over GF(256).""" + + if not left or not right: + raise ValueError("matrices must not be empty") + inner_size = len(left[0]) + if inner_size != len(right): + raise ValueError("matrix dimensions do not match") + column_count = len(right[0]) + result = [] + for row in left: + if len(row) != inner_size: + raise ValueError("left matrix is ragged") + result_row = [] + for column_index in range(column_count): + value = 0 + for inner_index in range(inner_size): + value ^= gf_mul( + row[inner_index], + right[inner_index][column_index], + ) + result_row.append(value) + result.append(tuple(result_row)) + return tuple(result) + + +def matrix_inverse( + matrix: tuple[tuple[int, ...], ...], +) -> tuple[tuple[int, ...], ...]: + """Invert a square matrix using GF(256) Gauss-Jordan elimination.""" + + size = len(matrix) + if size == 0 or any(len(row) != size for row in matrix): + raise ValueError("matrix must be non-empty and square") + augmented = [ + list(row) + + [1 if row_index == column_index else 0 + for column_index in range(size)] + for row_index, row in enumerate(matrix) + ] + + for pivot_column in range(size): + pivot_row = next( + ( + row_index + for row_index in range(pivot_column, size) + if augmented[row_index][pivot_column] != 0 + ), + None, + ) + if pivot_row is None: + raise MatrixSingularError("GF(256) matrix is singular") + if pivot_row != pivot_column: + augmented[pivot_column], augmented[pivot_row] = ( + augmented[pivot_row], + augmented[pivot_column], + ) + + pivot_inverse = gf_inverse( + augmented[pivot_column][pivot_column] + ) + augmented[pivot_column] = [ + gf_mul(value, pivot_inverse) + for value in augmented[pivot_column] + ] + + for row_index in range(size): + if row_index == pivot_column: + continue + factor = augmented[row_index][pivot_column] + if factor == 0: + continue + augmented[row_index] = [ + value ^ gf_mul(factor, pivot_value) + for value, pivot_value in zip( + augmented[row_index], + augmented[pivot_column], + ) + ] + + return tuple( + tuple(row[size:]) for row in augmented + ) + + +@lru_cache(maxsize=None) +def systematic_generator_matrix( + source_count: int, + parity_count: int, +) -> tuple[tuple[int, ...], ...]: + """Return an n×k systematic Vandermonde generator matrix.""" + + if not 1 <= source_count <= MAX_BLOCK_SYMBOLS: + raise ValueError("source_count is outside 1...255") + if not 0 <= parity_count <= ( + MAX_BLOCK_SYMBOLS - source_count + ): + raise ValueError("source_count + parity_count exceeds 255") + total_count = source_count + parity_count + vandermonde = tuple( + tuple( + gf_pow(row_index + 1, column_index) + for column_index in range(source_count) + ) + for row_index in range(total_count) + ) + top_inverse = matrix_inverse( + vandermonde[:source_count] + ) + generator = matrix_multiply(vandermonde, top_inverse) + identity = tuple( + tuple( + 1 if row_index == column_index else 0 + for column_index in range(source_count) + ) + for row_index in range(source_count) + ) + if generator[:source_count] != identity: + raise RuntimeError("generator matrix is not systematic") + return generator + + +def _linear_combine( + coefficients: tuple[int, ...], + symbols: tuple[bytes, ...], + symbol_size: int, +) -> bytes: + """Combine equal-size byte strings over GF(256), using C-level helpers.""" + + accumulator = 0 + for coefficient, symbol in zip(coefficients, symbols): + if coefficient == 0: + continue + translated = symbol.translate( + GF_MUL_TRANSLATIONS[coefficient] + ) + accumulator ^= int.from_bytes(translated, "little") + return accumulator.to_bytes(symbol_size, "little") + + +def encode_parity_symbols( + source_symbols: tuple[bytes, ...], + parity_count: int, +) -> tuple[bytes, ...]: + """Encode zero-padded source symbols into parity symbols.""" + + if not source_symbols: + raise ValueError("source symbol list is empty") + symbol_size = len(source_symbols[0]) + if symbol_size == 0: + raise ValueError("symbol size is zero") + if any(len(symbol) != symbol_size for symbol in source_symbols): + raise ValueError("source symbols must have equal size") + generator = systematic_generator_matrix( + len(source_symbols), parity_count + ) + return tuple( + _linear_combine(row, source_symbols, symbol_size) + for row in generator[len(source_symbols):] + ) + + +def outer_crc32(data: bytes) -> int: + return zlib.crc32(data) & 0xFFFFFFFF + + +def _validate_outer_symbol(symbol: OuterSymbol) -> None: + total_count = symbol.source_count + symbol.parity_count + if not 0 <= symbol.block_id <= 0xFFFFFFFF: + raise FECError("block_id is outside uint32") + if not 1 <= symbol.source_count <= MAX_BLOCK_SYMBOLS: + raise FECError("source_count is outside 1...255") + if not 0 <= symbol.parity_count <= ( + MAX_BLOCK_SYMBOLS - symbol.source_count + ): + raise FECError("invalid parity_count") + if not 0 <= symbol.symbol_index < total_count: + raise FECError("symbol_index is outside the block") + if not 1 <= symbol.symbol_size <= MAX_SYMBOL_SIZE: + raise FECError("symbol_size is outside uint16") + if not 1 <= len(symbol.data) <= symbol.symbol_size: + raise FECError("data length is outside symbol_size") + expected_parity = symbol.symbol_index >= symbol.source_count + if symbol.is_parity != expected_parity: + raise FECError("parity flag conflicts with symbol_index") + if symbol.is_parity and len(symbol.data) != symbol.symbol_size: + raise FECError("parity data must equal symbol_size") + + +def _pack_outer_header( + symbol: OuterSymbol, + crc_value: int, +) -> bytes: + return struct.pack( + OUTER_HEADER_FORMAT, + OUTER_MAGIC, + OUTER_VERSION, + OUTER_FLAG_PARITY if symbol.is_parity else 0, + OUTER_HEADER_SIZE, + symbol.block_id, + symbol.symbol_index, + symbol.source_count, + symbol.parity_count, + symbol.symbol_size, + len(symbol.data), + 0, + 0, + crc_value, + ) + + +def encode_outer_symbol(symbol: OuterSymbol) -> bytes: + """Serialize one source/parity symbol and calculate outer CRC32.""" + + if not isinstance(symbol, OuterSymbol): + raise TypeError("symbol must be OuterSymbol") + _validate_outer_symbol(symbol) + header_without_crc = _pack_outer_header(symbol, 0) + crc_value = outer_crc32(header_without_crc + symbol.data) + return _pack_outer_header(symbol, crc_value) + symbol.data + + +def decode_outer_symbol(wire_packet: bytes) -> OuterSymbol: + """Deserialize and validate one outer FEC packet.""" + + if not isinstance(wire_packet, (bytes, bytearray)): + raise TypeError("wire_packet must be bytes or bytearray") + wire_packet = bytes(wire_packet) + if len(wire_packet) < OUTER_HEADER_SIZE + 1: + raise FECError("outer packet is too short") + ( + magic, + version, + flags, + header_size, + block_id, + symbol_index, + source_count, + parity_count, + symbol_size, + data_length, + reserved16, + reserved32, + received_crc, + ) = struct.unpack( + OUTER_HEADER_FORMAT, + wire_packet[:OUTER_HEADER_SIZE], + ) + if magic != OUTER_MAGIC: + raise FECError("invalid outer magic") + if version != OUTER_VERSION: + raise FECError("unsupported outer version") + if flags & ~OUTER_FLAG_PARITY: + raise FECError("unsupported outer flags") + if header_size != OUTER_HEADER_SIZE: + raise FECError("invalid outer header size") + if reserved16 != 0 or reserved32 != 0: + raise FECError("reserved outer fields must be zero") + if len(wire_packet) != OUTER_HEADER_SIZE + data_length: + raise FECError("outer data_length does not match packet length") + + symbol = OuterSymbol( + block_id=block_id, + symbol_index=symbol_index, + source_count=source_count, + parity_count=parity_count, + symbol_size=symbol_size, + data=wire_packet[OUTER_HEADER_SIZE:], + is_parity=bool(flags & OUTER_FLAG_PARITY), + outer_crc32=received_crc, + ) + _validate_outer_symbol(symbol) + calculated_crc = outer_crc32( + _pack_outer_header(symbol, 0) + symbol.data + ) + if calculated_crc != received_crc: + raise OuterPacketCRCError( + "outer CRC mismatch: " + f"received 0x{received_crc:08X}, " + f"calculated 0x{calculated_crc:08X}" + ) + return symbol + + +def encode_fec_block( + inner_packets: tuple[bytes, ...], + block_id: int, + parity_count: int, +) -> tuple[bytes, ...]: + """Wrap systematic Lab028 packets and append parity packets.""" + + if not inner_packets: + raise ValueError("inner packet block is empty") + if len(inner_packets) + parity_count > MAX_BLOCK_SYMBOLS: + raise ValueError("FEC block has more than 255 symbols") + validated_packets = [] + for packet in inner_packets: + packet_bytes = bytes(packet) + decode_inner_packet(packet_bytes) + validated_packets.append(packet_bytes) + symbol_size = max(len(packet) for packet in validated_packets) + if symbol_size > MAX_SYMBOL_SIZE: + raise ValueError("inner packet is too large for outer symbol_size") + padded_sources = tuple( + packet + bytes(symbol_size - len(packet)) + for packet in validated_packets + ) + parity_symbols = encode_parity_symbols( + padded_sources, parity_count + ) + source_count = len(validated_packets) + outer_packets = [ + encode_outer_symbol( + OuterSymbol( + block_id=block_id, + symbol_index=index, + source_count=source_count, + parity_count=parity_count, + symbol_size=symbol_size, + data=packet, + is_parity=False, + ) + ) + for index, packet in enumerate(validated_packets) + ] + outer_packets.extend( + encode_outer_symbol( + OuterSymbol( + block_id=block_id, + symbol_index=source_count + parity_index, + source_count=source_count, + parity_count=parity_count, + symbol_size=symbol_size, + data=parity, + is_parity=True, + ) + ) + for parity_index, parity in enumerate(parity_symbols) + ) + return tuple(outer_packets) + + +def _padded_symbol(symbol: OuterSymbol) -> bytes: + return symbol.data + bytes(symbol.symbol_size - len(symbol.data)) + + +def trim_recovered_inner_packet(padded_packet: bytes) -> bytes: + """Trim GF padding using the recovered Lab028 header and validate CRC.""" + + if len(padded_packet) < INNER_HEADER_SIZE: + raise FECError("recovered symbol is shorter than Lab028 header") + unpacked = struct.unpack( + INNER_HEADER_FORMAT, + padded_packet[:INNER_HEADER_SIZE], + ) + magic = unpacked[0] + version = unpacked[1] + payload_length = unpacked[7] + if magic != INNER_MAGIC or version != INNER_VERSION: + raise FECError("recovered Lab028 magic/version is invalid") + packet_length = INNER_HEADER_SIZE + payload_length + if packet_length > len(padded_packet): + raise FECError("recovered Lab028 length exceeds symbol_size") + inner_packet = padded_packet[:packet_length] + decode_inner_packet(inner_packet) + return inner_packet + + +def decode_fec_block( + wire_symbols: tuple[bytes, ...], +) -> DecodedFECBlock: + """Recover all source Lab028 packets from any k valid outer symbols.""" + + if not wire_symbols: + raise InsufficientSymbolsError("no outer symbols received") + symbols_by_index: dict[int, OuterSymbol] = {} + metadata = None + for wire_symbol in wire_symbols: + symbol = decode_outer_symbol(wire_symbol) + current_metadata = ( + symbol.block_id, + symbol.source_count, + symbol.parity_count, + symbol.symbol_size, + ) + if metadata is None: + metadata = current_metadata + elif current_metadata != metadata: + raise FECError("outer symbols belong to different blocks") + existing = symbols_by_index.get(symbol.symbol_index) + if existing is not None: + if existing != symbol: + raise FECError("conflicting duplicate outer symbol") + continue + symbols_by_index[symbol.symbol_index] = symbol + + assert metadata is not None + block_id, source_count, parity_count, symbol_size = metadata + if len(symbols_by_index) < source_count: + raise InsufficientSymbolsError( + f"need {source_count} symbols, got {len(symbols_by_index)}" + ) + + selected_indices = tuple(sorted(symbols_by_index)[:source_count]) + generator = systematic_generator_matrix( + source_count, parity_count + ) + decode_matrix = matrix_inverse( + tuple(generator[index] for index in selected_indices) + ) + selected_symbols = tuple( + _padded_symbol(symbols_by_index[index]) + for index in selected_indices + ) + + source_packets = [] + recovered_indices = [] + for source_index in range(source_count): + received_source = symbols_by_index.get(source_index) + if received_source is not None: + inner_packet = received_source.data + decode_inner_packet(inner_packet) + else: + padded_packet = _linear_combine( + decode_matrix[source_index], + selected_symbols, + symbol_size, + ) + inner_packet = trim_recovered_inner_packet(padded_packet) + recovered_indices.append(source_index) + source_packets.append(inner_packet) + return DecodedFECBlock( + block_id=block_id, + source_packets=tuple(source_packets), + recovered_indices=tuple(recovered_indices), + ) diff --git a/tests/lab030_packet_erasure_fec.py b/tests/lab030_packet_erasure_fec.py new file mode 100644 index 0000000..e57ff8e --- /dev/null +++ b/tests/lab030_packet_erasure_fec.py @@ -0,0 +1,1711 @@ +""" +Lab030. Systematic packet-erasure FEC for the synchronous video stream. + +Real Lab028 packets with 512-byte payload are grouped consecutively in +blocks of k=8. Modes without FEC and with r=1,2,4 parity packets are compared +over the time-based Lab029B channel at 300 kbit/s. Blocks may cross composite +frame boundaries. Systematic packets precede parity packets; no interleaving, +ARQ, or retransmission is used. +""" + +from __future__ import annotations + +from bisect import bisect_right +import csv +from dataclasses import asdict, dataclass +from itertools import combinations +from pathlib import Path +from typing import Callable + +import cv2 +import matplotlib +import numpy as np + +matplotlib.use("Agg") +import matplotlib.pyplot as plt + +from protocol.packet_erasure_fec import ( + GF_PRIMITIVE_POLYNOMIAL, + OUTER_HEADER_FORMAT, + OUTER_HEADER_SIZE, + DecodedFECBlock, + InsufficientSymbolsError, + OuterPacketCRCError, + decode_fec_block, + decode_outer_symbol, + encode_fec_block, + gf_add, + gf_div, + gf_inverse, + gf_mul, +) +from protocol.video_packet import ( + CompositeReassembler, + ObjectType, + decode_packet as decode_inner_packet, +) +from tests.lab028_video_packetization import ( + COMPOSITE_FPS, + SOURCE_VIDEO_PATH, + EncodedComposite, + VideoMetadata, + load_video_profile, +) +from tests.lab029_packet_channel_simulation import ( + PreparedProfile, + prepare_profiles, +) +from tests.lab029b_time_based_burst_simulation import ( + BAD_TIME_FRACTION, + CONTROL_STREAM_BITRATE_BPS, + CONTROL_STREAM_BITRATE_KBPS, + MEAN_BAD_DURATIONS_SECONDS, + TimeInterval, + generate_bad_intervals, + percentile, + positive_runs, +) + + +OUTPUT_DIRECTORY = Path("data/processed/lab030") +CSV_PATH = OUTPUT_DIRECTORY / "lab030_results.csv" +REPORT_PATH = OUTPUT_DIRECTORY / "lab030_report.txt" +COMPOSITE_SUCCESS_PLOT_PATH = ( + OUTPUT_DIRECTORY / "lab030_composite_success.png" +) +NO_IMAGE_PLOT_PATH = OUTPUT_DIRECTORY / "lab030_no_image_duration.png" +STREAM_RATE_PLOT_PATH = ( + OUTPUT_DIRECTORY / "lab030_stream_rate_overhead.png" +) +DELAY_QUEUE_PLOT_PATH = ( + OUTPUT_DIRECTORY / "lab030_delay_queue.png" +) +MODE_COMPARISON_PLOT_PATH = ( + OUTPUT_DIRECTORY / "lab030_mode_comparison.png" +) + +INNER_PAYLOAD_SIZE = 512 +SOURCE_BLOCK_SIZE = 8 +MONTE_CARLO_REPETITIONS = 200 +MASTER_SEED = 300_300 +SEED_BASE = MASTER_SEED +TIME_EPSILON_SECONDS = 1e-12 + + +@dataclass(frozen=True) +class FECMode: + name: str + parity_count: int + label: str + + +FEC_MODES = ( + FECMode("none", 0, "Без FEC"), + FECMode("8+1", 1, "8+1"), + FECMode("8+2", 2, "8+2"), + FECMode("8+4", 4, "8+4"), +) + +CSV_FIELDS = [ + "mode", + "source_block_size", + "nominal_parity_count", + "mean_bad_duration_ms", + "mean_good_duration_ms", + "target_bad_time_fraction", + "actual_bad_time_fraction", + "monte_carlo_repetitions", + "seed", + "source_jpeg_bitrate_kbps", + "inner_packet_stream_bitrate_kbps", + "outer_fec_stream_bitrate_kbps", + "service_and_parity_percent", + "control_stream_bitrate_kbps", + "schedule_duration_seconds", + "mean_queue_length_packets", + "max_queue_length_packets", + "mean_publication_delay_seconds", + "p95_publication_delay_seconds", + "max_publication_delay_seconds", + "transmitted_source_packets", + "transmitted_parity_packets", + "lost_source_packets", + "lost_parity_packets", + "recovered_source_packets", + "fec_recovered_blocks", + "fec_unrecoverable_blocks", + "fec_affected_block_recovery_rate", + "base_objects_completed", + "roi_objects_completed", + "atomic_composite_frames_completed", + "composite_success_rate", + "base_only_frames", + "roi_only_frames", + "incomplete_frames", + "mean_no_new_image_duration_seconds", + "p95_no_new_image_duration_seconds", + "max_no_new_image_duration_seconds", + "mean_consecutive_incomplete_frames", + "p95_consecutive_incomplete_frames", + "max_consecutive_incomplete_frames", + "effective_delivered_video_bitrate_kbps", +] + + +@dataclass(frozen=True) +class SourcePacket: + global_index: int + composite_frame_id: int + generation_time_seconds: float + inner_packet: bytes + + +@dataclass(frozen=True) +class FECBlockPlan: + block_id: int + source_global_indices: tuple[int, ...] + source_count: int + parity_count: int + symbol_size: int + + +@dataclass(frozen=True) +class TransmissionUnit: + sequence_index: int + generation_time_seconds: float + wire_packet: bytes + is_parity: bool + block_id: int + symbol_index: int + source_global_index: int | None + + +@dataclass(frozen=True) +class ScheduledUnit: + unit: TransmissionUnit + start_seconds: float + end_seconds: float + + +@dataclass(frozen=True) +class ModeSchedule: + mode: FECMode + source_packets: tuple[SourcePacket, ...] + blocks: tuple[FECBlockPlan, ...] + units: tuple[ScheduledUnit, ...] + source_duration_seconds: float + duration_seconds: float + total_jpeg_bytes: int + total_inner_bytes: int + total_transmitted_bytes: int + mean_queue_length_packets: float + max_queue_length_packets: int + + +@dataclass +class BlockReceiveState: + plan: FECBlockPlan + received_outer_packets: list[bytes] + delivered_source_indices: set[int] + lost_source_indices: set[int] + decoded: bool = False + recovery_failed: bool = False + recovered_source_packets: int = 0 + + +@dataclass(frozen=True) +class RepetitionResult: + lost_source_packets: int + lost_parity_packets: int + recovered_source_packets: int + fec_recovered_blocks: int + fec_unrecoverable_blocks: int + base_objects_completed: int + roi_objects_completed: int + atomic_composite_frames_completed: int + base_only_frames: int + roi_only_frames: int + incomplete_frames: int + delivered_jpeg_bytes: int + publication_delays: tuple[float, ...] + no_new_image_durations: tuple[float, ...] + incomplete_frame_runs: tuple[int, ...] + bad_time_seconds: float + + +@dataclass(frozen=True) +class SimulationResult: + mode: str + source_block_size: int + nominal_parity_count: int + mean_bad_duration_ms: float + mean_good_duration_ms: float + target_bad_time_fraction: float + actual_bad_time_fraction: float + monte_carlo_repetitions: int + seed: int + source_jpeg_bitrate_kbps: float + inner_packet_stream_bitrate_kbps: float + outer_fec_stream_bitrate_kbps: float + service_and_parity_percent: float + control_stream_bitrate_kbps: float + schedule_duration_seconds: float + mean_queue_length_packets: float + max_queue_length_packets: int + mean_publication_delay_seconds: float + p95_publication_delay_seconds: float + max_publication_delay_seconds: float + transmitted_source_packets: int + transmitted_parity_packets: int + lost_source_packets: int + lost_parity_packets: int + recovered_source_packets: int + fec_recovered_blocks: int + fec_unrecoverable_blocks: int + fec_affected_block_recovery_rate: float + base_objects_completed: int + roi_objects_completed: int + atomic_composite_frames_completed: int + composite_success_rate: float + base_only_frames: int + roi_only_frames: int + incomplete_frames: int + mean_no_new_image_duration_seconds: float + p95_no_new_image_duration_seconds: float + max_no_new_image_duration_seconds: float + mean_consecutive_incomplete_frames: float + p95_consecutive_incomplete_frames: float + max_consecutive_incomplete_frames: int + effective_delivered_video_bitrate_kbps: float + + +@dataclass(frozen=True) +class FunctionalTestResult: + name: str + passed: bool + detail: str + + +def parity_for_last_block( + source_count: int, + nominal_parity_count: int, +) -> int: + """Preserve the nominal parity ratio in a final partial block.""" + + if nominal_parity_count == 0: + return 0 + return max( + 1, + int( + np.ceil( + source_count + * nominal_parity_count + / SOURCE_BLOCK_SIZE + ) + ), + ) + + +def prepare_source_packets( + profile: PreparedProfile, +) -> tuple[SourcePacket, ...]: + return tuple( + SourcePacket( + global_index=index, + composite_frame_id=prepared.composite_frame_id, + generation_time_seconds=( + prepared.composite_frame_id / COMPOSITE_FPS + ), + inner_packet=prepared.wire_packet, + ) + for index, prepared in enumerate(profile.packets) + ) + + +def build_mode_units( + source_packets: tuple[SourcePacket, ...], + mode: FECMode, +) -> tuple[ + tuple[TransmissionUnit, ...], + tuple[FECBlockPlan, ...], +]: + """Build sequential systematic/parity transmission units.""" + + if mode.parity_count == 0: + units = tuple( + TransmissionUnit( + sequence_index=index, + generation_time_seconds=packet.generation_time_seconds, + wire_packet=packet.inner_packet, + is_parity=False, + block_id=index // SOURCE_BLOCK_SIZE, + symbol_index=index % SOURCE_BLOCK_SIZE, + source_global_index=index, + ) + for index, packet in enumerate(source_packets) + ) + return units, () + + units = [] + blocks = [] + sequence_index = 0 + for block_id, start in enumerate( + range(0, len(source_packets), SOURCE_BLOCK_SIZE) + ): + block_sources = source_packets[ + start:start + SOURCE_BLOCK_SIZE + ] + source_count = len(block_sources) + parity_count = ( + mode.parity_count + if source_count == SOURCE_BLOCK_SIZE + else parity_for_last_block( + source_count, mode.parity_count + ) + ) + inner_packets = tuple( + packet.inner_packet for packet in block_sources + ) + outer_packets = encode_fec_block( + inner_packets, block_id, parity_count + ) + parsed_outer = [ + decode_outer_symbol(packet) for packet in outer_packets + ] + generation_complete = max( + packet.generation_time_seconds + for packet in block_sources + ) + blocks.append( + FECBlockPlan( + block_id=block_id, + source_global_indices=tuple( + packet.global_index for packet in block_sources + ), + source_count=source_count, + parity_count=parity_count, + symbol_size=parsed_outer[0].symbol_size, + ) + ) + for local_index, (wire_packet, parsed) in enumerate( + zip(outer_packets, parsed_outer) + ): + is_parity = parsed.is_parity + units.append( + TransmissionUnit( + sequence_index=sequence_index, + generation_time_seconds=( + generation_complete + if is_parity + else block_sources[ + local_index + ].generation_time_seconds + ), + wire_packet=wire_packet, + is_parity=is_parity, + block_id=block_id, + symbol_index=parsed.symbol_index, + source_global_index=( + None + if is_parity + else block_sources[ + local_index + ].global_index + ), + ) + ) + sequence_index += 1 + return tuple(units), tuple(blocks) + + +def schedule_units( + units: tuple[TransmissionUnit, ...], +) -> tuple[ + tuple[ScheduledUnit, ...], + float, + float, + int, +]: + """Run the transmitter as one FIFO queue at 300 kbit/s.""" + + scheduled = [] + cursor = 0.0 + for unit in units: + start = max(cursor, unit.generation_time_seconds) + end = ( + start + + len(unit.wire_packet) + * 8.0 + / CONTROL_STREAM_BITRATE_BPS + ) + scheduled.append(ScheduledUnit(unit, start, end)) + cursor = end + + queue_depth_samples = [] + for arriving in units: + generation_time = arriving.generation_time_seconds + queue_depth_samples.append( + sum( + 1 + for scheduled_unit in scheduled + if ( + scheduled_unit.unit.generation_time_seconds + <= generation_time + TIME_EPSILON_SECONDS + and scheduled_unit.end_seconds + > generation_time + TIME_EPSILON_SECONDS + ) + ) + ) + return ( + tuple(scheduled), + scheduled[-1].end_seconds, + float(np.mean(queue_depth_samples)), + max(queue_depth_samples), + ) + + +def build_mode_schedules( + metadata: VideoMetadata, + composites: list[EncodedComposite], + profile: PreparedProfile, +) -> dict[str, ModeSchedule]: + source_packets = prepare_source_packets(profile) + total_jpeg_bytes = sum( + len(composite.base_jpeg) + len(composite.roi_jpeg) + for composite in composites + ) + total_inner_bytes = sum( + len(packet.inner_packet) for packet in source_packets + ) + schedules = {} + for mode in FEC_MODES: + units, blocks = build_mode_units(source_packets, mode) + scheduled, duration, mean_queue, max_queue = schedule_units( + units + ) + schedules[mode.name] = ModeSchedule( + mode=mode, + source_packets=source_packets, + blocks=blocks, + units=scheduled, + source_duration_seconds=metadata.duration_seconds, + duration_seconds=duration, + total_jpeg_bytes=total_jpeg_bytes, + total_inner_bytes=total_inner_bytes, + total_transmitted_bytes=sum( + len(unit.unit.wire_packet) for unit in scheduled + ), + mean_queue_length_packets=mean_queue, + max_queue_length_packets=max_queue, + ) + return schedules + + +def overlap_loss_flags( + schedule: ModeSchedule, + intervals: tuple[TimeInterval, ...], +) -> np.ndarray: + flags = np.zeros(len(schedule.units), dtype=np.bool_) + interval_index = 0 + for unit_index, scheduled in enumerate(schedule.units): + while ( + interval_index < len(intervals) + and intervals[interval_index].end_seconds + <= scheduled.start_seconds + TIME_EPSILON_SECONDS + ): + interval_index += 1 + if interval_index >= len(intervals): + break + interval = intervals[interval_index] + if ( + interval.start_seconds + < scheduled.end_seconds - TIME_EPSILON_SECONDS + and interval.end_seconds + > scheduled.start_seconds + TIME_EPSILON_SECONDS + ): + flags[unit_index] = True + return flags + + +def frame_outage_metrics( + frame_count: int, + completion_times: dict[int, float], + schedule_end: float, +) -> tuple[ + tuple[float, ...], + tuple[int, ...], + tuple[float, ...], +]: + complete = [ + frame_id in completion_times for frame_id in range(frame_count) + ] + incomplete_runs = positive_runs([not flag for flag in complete]) + no_image = [] + index = 0 + while index < frame_count: + if complete[index]: + index += 1 + continue + run_start = index + while index < frame_count and not complete[index]: + index += 1 + start_time = ( + completion_times[run_start - 1] + if run_start > 0 + else 0.0 + ) + end_time = ( + completion_times[index] + if index < frame_count + else schedule_end + ) + no_image.append(max(0.0, end_time - start_time)) + publication_delays = tuple( + completion_time - frame_id / COMPOSITE_FPS + for frame_id, completion_time in completion_times.items() + ) + return tuple(no_image), incomplete_runs, publication_delays + + +def simulate_baseline( + schedule: ModeSchedule, + loss_flags: np.ndarray, + frame_count: int, +) -> RepetitionResult: + receiver = CompositeReassembler() + completion_times = {} + lost_source = 0 + delivered_jpeg_bytes = 0 + for lost, scheduled in zip(loss_flags, schedule.units): + if bool(lost): + lost_source += 1 + continue + completed = receiver.ingest(scheduled.unit.wire_packet) + if completed is not None: + completion_times[completed.composite_frame_id] = ( + scheduled.end_seconds + ) + delivered_jpeg_bytes += ( + len(completed.base_jpeg) + len(completed.roi_jpeg) + ) + return finish_repetition( + schedule, + receiver, + completion_times, + frame_count, + lost_source, + 0, + 0, + 0, + 0, + delivered_jpeg_bytes, + ) + + +def deliver_inner_packet( + receiver: CompositeReassembler, + inner_packet: bytes, + delivery_time: float, + completion_times: dict[int, float], +) -> int: + completed = receiver.ingest(inner_packet) + if completed is None: + return 0 + completion_times[completed.composite_frame_id] = delivery_time + return len(completed.base_jpeg) + len(completed.roi_jpeg) + + +def simulate_fec( + schedule: ModeSchedule, + loss_flags: np.ndarray, + frame_count: int, +) -> RepetitionResult: + receiver = CompositeReassembler() + completion_times = {} + plans = {plan.block_id: plan for plan in schedule.blocks} + states = { + block_id: BlockReceiveState( + plan=plan, + received_outer_packets=[], + delivered_source_indices=set(), + lost_source_indices=set(), + ) + for block_id, plan in plans.items() + } + lost_source = 0 + lost_parity = 0 + delivered_jpeg_bytes = 0 + + for lost, scheduled in zip(loss_flags, schedule.units): + unit = scheduled.unit + state = states[unit.block_id] + if bool(lost): + if unit.is_parity: + lost_parity += 1 + else: + lost_source += 1 + assert unit.source_global_index is not None + state.lost_source_indices.add( + unit.source_global_index + ) + continue + + state.received_outer_packets.append(unit.wire_packet) + parsed = decode_outer_symbol(unit.wire_packet) + if not parsed.is_parity: + global_index = state.plan.source_global_indices[ + parsed.symbol_index + ] + if global_index not in state.delivered_source_indices: + state.delivered_source_indices.add(global_index) + delivered_jpeg_bytes += deliver_inner_packet( + receiver, + parsed.data, + scheduled.end_seconds, + completion_times, + ) + + if ( + not state.decoded + and len(state.received_outer_packets) + >= state.plan.source_count + ): + decoded = decode_fec_block( + tuple(state.received_outer_packets) + ) + state.decoded = True + for local_index in decoded.recovered_indices: + global_index = state.plan.source_global_indices[ + local_index + ] + if global_index in state.delivered_source_indices: + continue + state.delivered_source_indices.add(global_index) + state.recovered_source_packets += 1 + delivered_jpeg_bytes += deliver_inner_packet( + receiver, + decoded.source_packets[local_index], + scheduled.end_seconds, + completion_times, + ) + + recovered_packets = sum( + state.recovered_source_packets for state in states.values() + ) + recovered_blocks = sum( + 1 + for state in states.values() + if ( + state.lost_source_indices + and state.lost_source_indices + <= state.delivered_source_indices + ) + ) + failed_blocks = sum( + 1 + for state in states.values() + if ( + state.lost_source_indices + and not ( + state.lost_source_indices + <= state.delivered_source_indices + ) + ) + ) + return finish_repetition( + schedule, + receiver, + completion_times, + frame_count, + lost_source, + lost_parity, + recovered_packets, + recovered_blocks, + failed_blocks, + delivered_jpeg_bytes, + ) + + +def finish_repetition( + schedule: ModeSchedule, + receiver: CompositeReassembler, + completion_times: dict[int, float], + frame_count: int, + lost_source: int, + lost_parity: int, + recovered_packets: int, + recovered_blocks: int, + failed_blocks: int, + delivered_jpeg_bytes: int, +) -> RepetitionResult: + base_completed = 0 + roi_completed = 0 + base_only = 0 + roi_only = 0 + for frame_id in range(frame_count): + atomic = frame_id in completion_times + base = ( + atomic + or receiver.object_is_complete(frame_id, ObjectType.BASE) + ) + roi = ( + atomic + or receiver.object_is_complete(frame_id, ObjectType.ROI) + ) + base_completed += int(base) + roi_completed += int(roi) + base_only += int(base and not roi) + roi_only += int(roi and not base) + no_image, incomplete_runs, delays = frame_outage_metrics( + frame_count, completion_times, schedule.duration_seconds + ) + return RepetitionResult( + lost_source_packets=lost_source, + lost_parity_packets=lost_parity, + recovered_source_packets=recovered_packets, + fec_recovered_blocks=recovered_blocks, + fec_unrecoverable_blocks=failed_blocks, + base_objects_completed=base_completed, + roi_objects_completed=roi_completed, + atomic_composite_frames_completed=len(completion_times), + base_only_frames=base_only, + roi_only_frames=roi_only, + incomplete_frames=frame_count - len(completion_times), + delivered_jpeg_bytes=delivered_jpeg_bytes, + publication_delays=delays, + no_new_image_durations=no_image, + incomplete_frame_runs=incomplete_runs, + bad_time_seconds=0.0, + ) + + +def simulate_condition( + schedule: ModeSchedule, + frame_count: int, + mean_bad_duration_seconds: float, + seed: int, + repetitions: int, +) -> SimulationResult: + rng = np.random.default_rng(seed) + repetition_results = [] + bad_time_seconds = 0.0 + for _ in range(repetitions): + intervals = generate_bad_intervals( + schedule.duration_seconds, + mean_bad_duration_seconds, + rng, + ) + bad_time_seconds += sum( + interval.duration_seconds for interval in intervals + ) + loss_flags = overlap_loss_flags(schedule, intervals) + repetition_results.append( + simulate_baseline(schedule, loss_flags, frame_count) + if schedule.mode.parity_count == 0 + else simulate_fec(schedule, loss_flags, frame_count) + ) + + def total(field: str) -> int: + return sum( + int(getattr(result, field)) + for result in repetition_results + ) + + def flattened(field: str) -> list[float]: + return [ + float(value) + for result in repetition_results + for value in getattr(result, field) + ] + + publication_delays = flattened("publication_delays") + no_image = flattened("no_new_image_durations") + incomplete_runs = flattened("incomplete_frame_runs") + recovered_blocks = total("fec_recovered_blocks") + failed_blocks = total("fec_unrecoverable_blocks") + affected_blocks = recovered_blocks + failed_blocks + total_frames = frame_count * repetitions + source_packets_per_pass = len(schedule.source_packets) + parity_packets_per_pass = sum( + 1 for unit in schedule.units if unit.unit.is_parity + ) + source_jpeg_rate = ( + schedule.total_jpeg_bytes + * 8.0 + / schedule.source_duration_seconds + / 1000.0 + ) + inner_rate = ( + schedule.total_inner_bytes + * 8.0 + / schedule.source_duration_seconds + / 1000.0 + ) + outer_rate = ( + schedule.total_transmitted_bytes + * 8.0 + / schedule.source_duration_seconds + / 1000.0 + ) + delivered_bytes = total("delivered_jpeg_bytes") + return SimulationResult( + mode=schedule.mode.name, + source_block_size=SOURCE_BLOCK_SIZE, + nominal_parity_count=schedule.mode.parity_count, + mean_bad_duration_ms=mean_bad_duration_seconds * 1000.0, + mean_good_duration_ms=( + mean_bad_duration_seconds + * (1.0 - BAD_TIME_FRACTION) + / BAD_TIME_FRACTION + * 1000.0 + ), + target_bad_time_fraction=BAD_TIME_FRACTION, + actual_bad_time_fraction=( + bad_time_seconds + / (schedule.duration_seconds * repetitions) + ), + monte_carlo_repetitions=repetitions, + seed=seed, + source_jpeg_bitrate_kbps=source_jpeg_rate, + inner_packet_stream_bitrate_kbps=inner_rate, + outer_fec_stream_bitrate_kbps=outer_rate, + service_and_parity_percent=( + ( + schedule.total_transmitted_bytes + - schedule.total_jpeg_bytes + ) + / schedule.total_transmitted_bytes + * 100.0 + ), + control_stream_bitrate_kbps=CONTROL_STREAM_BITRATE_KBPS, + schedule_duration_seconds=schedule.duration_seconds, + mean_queue_length_packets=( + schedule.mean_queue_length_packets + ), + max_queue_length_packets=schedule.max_queue_length_packets, + mean_publication_delay_seconds=( + float(np.mean(publication_delays)) + if publication_delays + else 0.0 + ), + p95_publication_delay_seconds=percentile( + publication_delays, 95 + ), + max_publication_delay_seconds=( + max(publication_delays) if publication_delays else 0.0 + ), + transmitted_source_packets=( + source_packets_per_pass * repetitions + ), + transmitted_parity_packets=( + parity_packets_per_pass * repetitions + ), + lost_source_packets=total("lost_source_packets"), + lost_parity_packets=total("lost_parity_packets"), + recovered_source_packets=total("recovered_source_packets"), + fec_recovered_blocks=recovered_blocks, + fec_unrecoverable_blocks=failed_blocks, + fec_affected_block_recovery_rate=( + recovered_blocks / affected_blocks + if affected_blocks + else 0.0 + ), + base_objects_completed=total("base_objects_completed"), + roi_objects_completed=total("roi_objects_completed"), + atomic_composite_frames_completed=total( + "atomic_composite_frames_completed" + ), + composite_success_rate=( + total("atomic_composite_frames_completed") / total_frames + ), + base_only_frames=total("base_only_frames"), + roi_only_frames=total("roi_only_frames"), + incomplete_frames=total("incomplete_frames"), + mean_no_new_image_duration_seconds=( + float(np.mean(no_image)) if no_image else 0.0 + ), + p95_no_new_image_duration_seconds=percentile(no_image, 95), + max_no_new_image_duration_seconds=( + max(no_image) if no_image else 0.0 + ), + mean_consecutive_incomplete_frames=( + float(np.mean(incomplete_runs)) + if incomplete_runs + else 0.0 + ), + p95_consecutive_incomplete_frames=percentile( + incomplete_runs, 95 + ), + max_consecutive_incomplete_frames=( + int(max(incomplete_runs)) if incomplete_runs else 0 + ), + effective_delivered_video_bitrate_kbps=( + delivered_bytes + * 8.0 + / (schedule.duration_seconds * repetitions) + / 1000.0 + ), + ) + + +def run_monte_carlo( + schedules: dict[str, ModeSchedule], + frame_count: int, +) -> list[SimulationResult]: + results = [] + for duration_index, duration in enumerate( + MEAN_BAD_DURATIONS_SECONDS + ): + seed = SEED_BASE + duration_index + for mode in FEC_MODES: + results.append( + simulate_condition( + schedules[mode.name], + frame_count, + duration, + seed, + MONTE_CARLO_REPETITIONS, + ) + ) + return results + + +def result_lookup( + results: list[SimulationResult], + mean_bad_duration_ms: float, +) -> dict[str, SimulationResult]: + return { + result.mode: result + for result in results + if result.mean_bad_duration_ms == mean_bad_duration_ms + } + + +def run_functional_tests( + composites: list[EncodedComposite], + schedules: dict[str, ModeSchedule], + results: list[SimulationResult], +) -> list[FunctionalTestResult]: + tests: list[tuple[str, Callable[[], str]]] = [] + sample_inner = tuple( + packet.inner_packet + for packet in schedules["none"].source_packets[:8] + ) + + def gf_arithmetic() -> str: + for value in range(1, 256): + inverse = gf_inverse(value) + if gf_mul(value, inverse) != 1: + raise AssertionError(f"inverse failed for {value}") + if gf_div(value, value) != 1: + raise AssertionError(f"division failed for {value}") + for left in range(0, 256, 17): + for right in range(0, 256, 19): + for third in range(0, 256, 31): + if gf_mul(left, gf_add(right, third)) != gf_add( + gf_mul(left, right), gf_mul(left, third) + ): + raise AssertionError("distributivity failed") + return "inverse, division, and distributivity passed" + + def encode_decode_without_loss() -> str: + outer = encode_fec_block(sample_inner, 1, 4) + decoded = decode_fec_block(outer) + if decoded.source_packets != sample_inner: + raise AssertionError("lossless FEC round trip changed bytes") + return "systematic 8+4 block is byte-exact without loss" + + def recover_any_r_losses() -> str: + checked = 0 + for parity_count in (1, 2, 4): + outer = encode_fec_block( + sample_inner, parity_count, parity_count + ) + for missing in combinations( + range(len(outer)), parity_count + ): + available = tuple( + packet + for index, packet in enumerate(outer) + if index not in missing + ) + decoded = decode_fec_block(available) + if decoded.source_packets != sample_inner: + raise AssertionError( + f"failed r={parity_count}, missing={missing}" + ) + checked += 1 + return f"all {checked} exact-r erasure combinations recovered" + + def r_plus_one_is_not_guaranteed() -> str: + outer = encode_fec_block(sample_inner, 7, 2) + available = outer[3:] + try: + decode_fec_block(available) + except InsufficientSymbolsError: + return "8+2 correctly rejected seven available symbols" + raise AssertionError("r+1 erasures unexpectedly guaranteed") + + def recovered_inner_is_exact_and_crc_valid() -> str: + outer = encode_fec_block(sample_inner, 9, 4) + available = tuple( + packet + for index, packet in enumerate(outer) + if index not in {0, 2, 5, 9} + ) + decoded = decode_fec_block(available) + for original, restored in zip( + sample_inner, decoded.source_packets + ): + if restored != original: + raise AssertionError("restored inner bytes differ") + decode_inner_packet(restored) + return "restored Lab028 packets are byte-exact and CRC-valid" + + def outer_crc_detects_corruption() -> str: + outer = bytearray( + encode_fec_block(sample_inner, 10, 2)[0] + ) + outer[-1] ^= 0x01 + try: + decode_outer_symbol(bytes(outer)) + except OuterPacketCRCError: + return "outer payload bit flip rejected by outer CRC32" + raise AssertionError("outer CRC did not reject corruption") + + def zero_bad_restores_all_frames() -> str: + frame_count = len(composites) + for mode in FEC_MODES: + schedule = schedules[mode.name] + loss_flags = np.zeros( + len(schedule.units), dtype=np.bool_ + ) + repetition = ( + simulate_baseline(schedule, loss_flags, frame_count) + if mode.parity_count == 0 + else simulate_fec(schedule, loss_flags, frame_count) + ) + if ( + repetition.atomic_composite_frames_completed + != frame_count + ): + raise AssertionError( + f"zero-Bad failed for {mode.name}" + ) + return "all modes restored all 63 frames without Bad" + + def fixed_seed_is_reproducible() -> str: + first = simulate_condition( + schedules["8+2"], len(composites), 0.05, 399_399, 3 + ) + second = simulate_condition( + schedules["8+2"], len(composites), 0.05, 399_399, 3 + ) + if first != second: + raise AssertionError("same seed changed aggregate results") + return "identical seed produced identical result fields" + + def incomplete_composite_is_not_published() -> str: + receiver = CompositeReassembler() + first_frame = [ + packet.inner_packet + for packet in schedules["none"].source_packets + if packet.composite_frame_id == 0 + ] + published = 0 + for packet in first_frame[:-1]: + published += int(receiver.ingest(packet) is not None) + if published: + raise AssertionError("incomplete composite was published") + return "missing ROI fragment prevented atomic publication" + + def queue_includes_parity() -> str: + baseline = schedules["none"] + protected = schedules["8+4"] + parity_units = sum( + unit.unit.is_parity for unit in protected.units + ) + if parity_units <= 0: + raise AssertionError("8+4 schedule has no parity") + if len(protected.units) != ( + len(baseline.source_packets) + parity_units + ): + raise AssertionError("parity is absent from FIFO units") + if ( + protected.total_transmitted_bytes + <= baseline.total_transmitted_bytes + ): + raise AssertionError("parity did not increase wire bytes") + if protected.duration_seconds < baseline.duration_seconds: + raise AssertionError("parity shortened transmitter schedule") + return ( + f"{parity_units} parity packets included in FIFO timing " + "and queue metrics" + ) + + tests.extend( + [ + ("gf256_arithmetic", gf_arithmetic), + ("fec_without_loss", encode_decode_without_loss), + ("recover_any_r_erasures", recover_any_r_losses), + ("r_plus_one_not_guaranteed", r_plus_one_is_not_guaranteed), + ( + "restored_inner_packet_and_crc", + recovered_inner_is_exact_and_crc_valid, + ), + ("outer_crc_detection", outer_crc_detects_corruption), + ("zero_bad_100_percent", zero_bad_restores_all_frames), + ("fixed_seed_reproducibility", fixed_seed_is_reproducible), + ( + "atomic_incomplete_composite", + incomplete_composite_is_not_published, + ), + ("queue_counts_parity", queue_includes_parity), + ] + ) + test_results = [] + for name, test in tests: + try: + detail = test() + except Exception as error: + test_results.append( + FunctionalTestResult(name, False, str(error)) + ) + else: + test_results.append( + FunctionalTestResult(name, True, detail) + ) + failed = [test for test in test_results if not test.passed] + if failed: + raise RuntimeError( + "Lab030 functional checks failed: " + + "; ".join( + f"{test.name}: {test.detail}" for test in failed + ) + ) + return test_results + + +def validate_results(results: list[SimulationResult]) -> None: + if len(results) != 16: + raise RuntimeError(f"expected 16 rows, got {len(results)}") + keys = { + (result.mode, result.mean_bad_duration_ms) + for result in results + } + if len(keys) != 16: + raise RuntimeError("Lab030 result rows are not unique") + if any( + not 0.0 <= result.composite_success_rate <= 1.0 + for result in results + ): + raise RuntimeError("composite success is outside 0...1") + + +def save_csv(results: list[SimulationResult]) -> None: + OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) + with CSV_PATH.open("w", encoding="utf-8", newline="") as csv_file: + writer = csv.DictWriter(csv_file, fieldnames=CSV_FIELDS) + writer.writeheader() + for result in results: + raw = asdict(result) + writer.writerow( + { + field: ( + f"{raw[field]:.12g}" + if isinstance(raw[field], float) + else raw[field] + ) + for field in CSV_FIELDS + } + ) + + +def save_line_plot( + results: list[SimulationResult], + getter: Callable[[SimulationResult], float], + ylabel: str, + title: str, + path: Path, +) -> None: + x_values = [ + duration * 1000.0 + for duration in MEAN_BAD_DURATIONS_SECONDS + ] + figure, axis = plt.subplots(figsize=(9, 5.5)) + for mode in FEC_MODES: + values = [ + getter(result_lookup(results, duration_ms)[mode.name]) + for duration_ms in x_values + ] + axis.plot( + x_values, + values, + marker="o", + linewidth=2, + label=mode.label, + ) + axis.set_xscale("log") + axis.set_xticks(x_values) + axis.set_xticklabels([f"{value:g}" for value in x_values]) + axis.set_xlabel("Средняя длительность Bad, мс") + axis.set_ylabel(ylabel) + axis.set_title(title) + axis.grid(True, which="both", alpha=0.3) + axis.legend() + figure.tight_layout() + figure.savefig(path, dpi=160) + plt.close(figure) + + +def save_plots(results: list[SimulationResult]) -> None: + save_line_plot( + results, + lambda result: result.composite_success_rate * 100.0, + "Полностью восстановленные составные кадры, %", + "Lab030. Эффективность пакетного FEC", + COMPOSITE_SUCCESS_PLOT_PATH, + ) + save_line_plot( + results, + lambda result: result.p95_no_new_image_duration_seconds, + "P95 отсутствия нового изображения, с", + "Lab030. Прерывания видеопотока", + NO_IMAGE_PLOT_PATH, + ) + + representative = result_lookup(results, 200.0) + x = np.arange(len(FEC_MODES)) + rates = [ + representative[mode.name].outer_fec_stream_bitrate_kbps + for mode in FEC_MODES + ] + overhead = [ + representative[mode.name].service_and_parity_percent + for mode in FEC_MODES + ] + figure, rate_axis = plt.subplots(figsize=(9, 5.5)) + overhead_axis = rate_axis.twinx() + rate_bars = rate_axis.bar( + x - 0.18, rates, 0.36, + color="tab:blue", label="Wire rate", + ) + overhead_bars = overhead_axis.bar( + x + 0.18, overhead, 0.36, + color="tab:orange", label="Service + parity", + ) + rate_axis.axhline( + CONTROL_STREAM_BITRATE_KBPS, + color="tab:red", + linestyle="--", + label="300 kbit/s", + ) + rate_axis.set_xticks(x) + rate_axis.set_xticklabels([mode.label for mode in FEC_MODES]) + rate_axis.set_ylabel("Предлагаемая скорость, кбит/с") + overhead_axis.set_ylabel("Служебные и parity-данные, %") + rate_axis.set_title("Lab030. Скорость и избыточность") + rate_axis.grid(True, axis="y", alpha=0.3) + rate_axis.legend( + [rate_bars, overhead_bars, rate_axis.lines[0]], + ["Wire rate", "Service + parity", "300 kbit/s"], + loc="best", + ) + figure.tight_layout() + figure.savefig(STREAM_RATE_PLOT_PATH, dpi=160) + plt.close(figure) + + delays = [ + representative[mode.name].p95_publication_delay_seconds + for mode in FEC_MODES + ] + queues = [ + representative[mode.name].max_queue_length_packets + for mode in FEC_MODES + ] + figure, delay_axis = plt.subplots(figsize=(9, 5.5)) + queue_axis = delay_axis.twinx() + delay_bars = delay_axis.bar( + x - 0.18, delays, 0.36, + color="tab:blue", label="P95 delay", + ) + queue_bars = queue_axis.bar( + x + 0.18, queues, 0.36, + color="tab:orange", label="Max queue", + ) + delay_axis.set_xticks(x) + delay_axis.set_xticklabels([mode.label for mode in FEC_MODES]) + delay_axis.set_ylabel("P95 задержки публикации, с") + queue_axis.set_ylabel("Максимальная очередь, пакетов") + delay_axis.set_title( + "Lab030. Задержка и очередь при Bad 200 мс" + ) + delay_axis.grid(True, axis="y", alpha=0.3) + delay_axis.legend( + [delay_bars, queue_bars], + ["P95 publication delay", "Max queue"], + loc="best", + ) + figure.tight_layout() + figure.savefig(DELAY_QUEUE_PLOT_PATH, dpi=160) + plt.close(figure) + + success = [ + representative[mode.name].composite_success_rate * 100.0 + for mode in FEC_MODES + ] + recovery = [ + representative[ + mode.name + ].fec_affected_block_recovery_rate + * 100.0 + for mode in FEC_MODES + ] + figure, axis = plt.subplots(figsize=(9, 5.5)) + axis.bar( + x - 0.18, success, 0.36, label="Composite success" + ) + axis.bar( + x + 0.18, recovery, 0.36, label="Affected FEC blocks recovered" + ) + axis.set_xticks(x) + axis.set_xticklabels([mode.label for mode in FEC_MODES]) + axis.set_ylabel("Доля, %") + axis.set_title("Lab030. Сравнение режимов при Bad 200 мс") + axis.grid(True, axis="y", alpha=0.3) + axis.legend() + figure.tight_layout() + figure.savefig(MODE_COMPARISON_PLOT_PATH, dpi=160) + plt.close(figure) + + +def mode_rate_table( + schedules: dict[str, ModeSchedule], +) -> list[str]: + lines = [ + ( + "mode | source packets | parity packets | last block k+r | " + "JPEG kbit/s | inner kbit/s | outer kbit/s | overhead | " + "queue mean/max | timeline s" + ), + ( + "----:|---------------:|---------------:|---------------:|" + "------------:|-------------:|-------------:|---------:|" + "---------------:|----------:" + ), + ] + for mode in FEC_MODES: + schedule = schedules[mode.name] + parity_packets = sum( + unit.unit.is_parity for unit in schedule.units + ) + last_block = ( + f"{schedule.blocks[-1].source_count}+" + f"{schedule.blocks[-1].parity_count}" + if schedule.blocks + else "none" + ) + jpeg_rate = ( + schedule.total_jpeg_bytes + * 8 + / schedule.source_duration_seconds + / 1000 + ) + inner_rate = ( + schedule.total_inner_bytes + * 8 + / schedule.source_duration_seconds + / 1000 + ) + outer_rate = ( + schedule.total_transmitted_bytes + * 8 + / schedule.source_duration_seconds + / 1000 + ) + overhead = ( + ( + schedule.total_transmitted_bytes + - schedule.total_jpeg_bytes + ) + / schedule.total_transmitted_bytes + * 100 + ) + lines.append( + f"{mode.label} | {len(schedule.source_packets)} | " + f"{parity_packets} | {last_block} | " + f"{jpeg_rate:.3f} | {inner_rate:.3f} | " + f"{outer_rate:.3f} | {overhead:.3f}% | " + f"{schedule.mean_queue_length_packets:.3f}/" + f"{schedule.max_queue_length_packets} | " + f"{schedule.duration_seconds:.6f}" + ) + return lines + + +def result_table(results: list[SimulationResult]) -> list[str]: + lines = [ + ( + "Bad ms | mode | Bad actual | lost src/parity | recovered pkt | " + "blocks recovered/failed | block recovery | composite | " + "BASE-only/ROI-only | no image mean/p95/max s | " + "delay mean/p95/max s | video kbit/s" + ), + ( + "------:|-----:|-----------:|----------------:|--------------:|" + "------------------------:|---------------:|----------:|" + "------------------:|-------------------------:|" + "----------------------:|------------:" + ), + ] + for duration in MEAN_BAD_DURATIONS_SECONDS: + by_mode = result_lookup(results, duration * 1000.0) + for mode in FEC_MODES: + result = by_mode[mode.name] + lines.append( + f"{duration * 1000.0:.0f} | {mode.label} | " + f"{result.actual_bad_time_fraction * 100:.3f}% | " + f"{result.lost_source_packets}/" + f"{result.lost_parity_packets} | " + f"{result.recovered_source_packets} | " + f"{result.fec_recovered_blocks}/" + f"{result.fec_unrecoverable_blocks} | " + f"{result.fec_affected_block_recovery_rate * 100:.3f}% | " + f"{result.composite_success_rate * 100:.3f}% | " + f"{result.base_only_frames}/{result.roi_only_frames} | " + f"{result.mean_no_new_image_duration_seconds:.3f}/" + f"{result.p95_no_new_image_duration_seconds:.3f}/" + f"{result.max_no_new_image_duration_seconds:.3f} | " + f"{result.mean_publication_delay_seconds:.3f}/" + f"{result.p95_publication_delay_seconds:.3f}/" + f"{result.max_publication_delay_seconds:.3f} | " + f"{result.effective_delivered_video_bitrate_kbps:.3f}" + ) + return lines + + +def write_report( + metadata: VideoMetadata, + composites: list[EncodedComposite], + schedules: dict[str, ModeSchedule], + results: list[SimulationResult], + tests: list[FunctionalTestResult], +) -> None: + lines = [ + "Lab030. Пакетное избыточное кодирование стираний", + "", + "Исходный профиль и неизменный внутренний транспорт", + f"- Видео: {SOURCE_VIDEO_PATH}", + ( + f"- {len(composites)} реальных пар BASE/ROI, 3 fps; " + "BASE 240x135 grayscale JPEG Q23, " + "ROI 320x180 grayscale JPEG Q33." + ), + ( + "- Внутренний Lab028 packet не изменён: payload 512 байт, " + "32-byte header, packet CRC32 и object CRC32." + ), + "", + "GF(256) и блочный код", + ( + f"- Примитивный полином: 0x{GF_PRIMITIVE_POLYNOMIAL:X} " + "(x^8+x^4+x^3+x^2+1)." + ), + ( + "- Матрица Вандермонда n×k умножается на обратную верхнюю " + "k×k матрицу и становится систематической." + ), + ( + "- Любые k доступных строк систематической генераторной " + "матрицы восстанавливают k исходных символов." + ), + ( + "- Нулевое дополнение используется только в математике; " + "восстановленный внутренний пакет обрезается по payload_length " + "Lab028 и проходит собственный CRC." + ), + "", + "Внешний пакет FEC", + f"- struct: {OUTER_HEADER_FORMAT}", + ( + f"- Размер: {OUTER_HEADER_SIZE} байта, network byte order, " + "padding отсутствует." + ), + ( + "- Layout: magic[4]@0, version:u8@4, flags:u8@5, " + "header_size:u16@6, block_id:u32@8, symbol_index:u16@12, " + "k:u16@14, r:u16@16, symbol_size:u16@18, " + "data_length:u16@20, reserved16:u16@22, " + "reserved32:u32@24, outer_crc32:u32@28." + ), + ( + "- flags bit0: parity. CRC32 вычисляется по заголовку с " + "нулевым outer_crc32 и данным внешнего символа." + ), + "", + "Формирование блоков и передача", + ( + "- Последовательный поток внутренних пакетов BASE→ROI делится " + "на k=8; блок может пересекать границу кадра." + ), + ( + "- Сначала передаются systematic symbols, затем parity symbols " + "этого блока; перемежение отсутствует." + ), + ( + "- Последний блок содержит 6 исходных пакетов. Для него " + "r_last=max(1, ceil(k_last*r/8)): режимы 8+1, 8+2, 8+4 " + "получают соответственно 1, 2 и 3 parity." + ), + ( + "- Режим Без FEC передаёт исходные Lab028 packets без внешней " + "обёртки; остальные режимы используют внешний заголовок." + ), + ( + "- FIFO учитывает время генерации кадра, все внешние пакеты, " + "фактический размер и скорость 300 кбит/с. Очередь не " + "сбрасывается на границах кадров." + ), + "", + "Скорость, избыточность и очередь", + *mode_rate_table(schedules), + "", + "Временная модель", + ( + "- Экспоненциальные Good/Bad интервалы Lab029B, Bad≈2%, " + "mean Bad 10/50/200/1000 мс, 200 повторов, fixed seeds " + f"{SEED_BASE}...{SEED_BASE + 3}." + ), + ( + "- Пакет теряется при любом пересечении его передачи с Bad." + ), + ( + "- FEC исправляет стирания; CRC обнаруживает повреждения, но " + "битовые ошибки отдельно не добавляются." + ), + "", + "Функциональные проверки", + ] + lines.extend( + f"- {'PASS' if test.passed else 'FAIL'} {test.name}: {test.detail}" + for test in tests + ) + lines.extend( + [ + "", + "Результаты Monte Carlo", + *result_table(results), + "", + "Допущения и ограничения", + ( + "- Не реализованы перемежение, ARQ, повторные передачи, " + "команды управления, телеметрия и реальный SDR." + ), + ( + "- Любое частичное пересечение Bad уничтожает весь внешний " + "или базовый внутренний пакет." + ), + ( + "- Принятые systematic packets немедленно поступают " + "reassembler; восстановленные стирания становятся доступны " + "при получении k символов блока." + ), + ( + "- Publication delay измеряется от frame_id/3 до атомарной " + "выдачи BASE+ROI." + ), + ( + "- Queue length измеряется в моменты генерации пакетов и " + "включает ожидающие и обслуживаемый пакет." + ), + ( + "- Окончательный режим FEC автоматически не выбирается." + ), + "", + "Артефакты", + f"- CSV: {CSV_PATH}", + f"- Composite success: {COMPOSITE_SUCCESS_PLOT_PATH}", + f"- No-image duration: {NO_IMAGE_PLOT_PATH}", + f"- Stream rate/overhead: {STREAM_RATE_PLOT_PATH}", + f"- Delay/queue: {DELAY_QUEUE_PLOT_PATH}", + f"- Mode comparison: {MODE_COMPARISON_PLOT_PATH}", + ( + "- JPEG, внутренние/внешние packets и бинарные дампы " + "не сохранялись." + ), + "", + ] + ) + REPORT_PATH.write_text("\n".join(lines), encoding="utf-8") + + +def validate_outputs() -> None: + for path in ( + CSV_PATH, + REPORT_PATH, + COMPOSITE_SUCCESS_PLOT_PATH, + NO_IMAGE_PLOT_PATH, + STREAM_RATE_PLOT_PATH, + DELAY_QUEUE_PLOT_PATH, + MODE_COMPARISON_PLOT_PATH, + ): + if not path.exists() or path.stat().st_size <= 0: + raise RuntimeError(f"missing or empty output: {path}") + + +def main() -> None: + print("Lab030: loading real 512-byte-payload Lab028 stream...") + metadata, composites = load_video_profile(SOURCE_VIDEO_PATH) + profile = prepare_profiles(composites)[INNER_PAYLOAD_SIZE] + schedules = build_mode_schedules( + metadata, composites, profile + ) + for mode in FEC_MODES: + schedule = schedules[mode.name] + print( + f" {mode.label}: units={len(schedule.units)}, " + f"offered=" + f"{schedule.total_transmitted_bytes * 8 / metadata.duration_seconds / 1000:.3f} " + f"kbit/s, timeline={schedule.duration_seconds:.6f} s" + ) + + print( + f"Running 16 conditions, " + f"{MONTE_CARLO_REPETITIONS} repetitions each..." + ) + results = run_monte_carlo(schedules, len(composites)) + validate_results(results) + + print("Running Lab030 functional checks...") + tests = run_functional_tests( + composites, schedules, results + ) + for test in tests: + print(f" PASS {test.name}: {test.detail}") + + OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) + save_csv(results) + save_plots(results) + write_report( + metadata, composites, schedules, results, tests + ) + validate_outputs() + + print("Representative Bad=200 ms results:") + representative = result_lookup(results, 200.0) + for mode in FEC_MODES: + result = representative[mode.name] + print( + f" {mode.label}: composite=" + f"{result.composite_success_rate:.6f}, " + f"blocks={result.fec_recovered_blocks}/" + f"{result.fec_unrecoverable_blocks}, " + f"delay_p95={result.p95_publication_delay_seconds:.6f} s" + ) + print(f"CSV: {CSV_PATH}") + print(f"Report: {REPORT_PATH}") + print("Lab030 completed successfully.") + + +if __name__ == "__main__": + main()