diff --git a/PROJECT_LOG.md b/PROJECT_LOG.md index 197d12a..6c20762 100644 --- a/PROJECT_LOG.md +++ b/PROJECT_LOG.md @@ -239,3 +239,28 @@ git version 2.51.1.windows.1 - Прогноз 1000 мс не дал преимущества над более простой политикой. - Прогноз 500 мс чрезмерно снижает частоту обновления. - Следующая лабораторная проверяет долговременную устойчивость и изменение пропускной способности канала. + +--- + +# Запись 011 + +## Дата + +3 августа 2026 года + +## Тема + +Завершение Lab036: долговременная устойчивость видеопланировщика при изменении пропускной способности. + +## Выполнено + +- Lab036 проверяет видеопланировщик на протяжении 600 секунд. +- Расчётный рост очереди без удаления совпал с моделью. +- При 230 кбит/с очередь без удаления выросла до 4139 пакетов. +- 95-й процентиль возраста изображения достиг 74,6 секунды. +- Плавное поступление сильно устаревших кадров непригодно для управления. +- Политика хранения только самого свежего не начатого кадра ограничивает очередь. +- При 230 кбит/с она обеспечила около 2,615 кадра/с. +- После восстановления скорости возраст изображения снизился ниже 500 мс за 0,077 секунды. +- Политика двух ожидающих кадров оставлена резервной. +- Следующая лабораторная объединяет планировщик с моделью ошибок радиоканала и защитой команд управления. diff --git a/data/processed/lab036/lab036_control_delay.png b/data/processed/lab036/lab036_control_delay.png new file mode 100644 index 0000000..7fc905b Binary files /dev/null and b/data/processed/lab036/lab036_control_delay.png differ diff --git a/data/processed/lab036/lab036_control_metrics.csv b/data/processed/lab036/lab036_control_metrics.csv new file mode 100644 index 0000000..eb1fa9d --- /dev/null +++ b/data/processed/lab036/lab036_control_metrics.csv @@ -0,0 +1,16 @@ +scenario,policy,control_mean_delay_ms,control_p95_delay_ms,control_max_delay_ms,control_over_100ms,control_max_receive_gap_ms,emergency_delays_ms,emergency_max_delay_ms,all_emergency_under_50ms,telemetry_mean_delay_ms,telemetry_p95_delay_ms,telemetry_max_delay_ms,telemetry_over_500ms +constant_300,no_drop,10.082484055552035,17.466333333331363,17.787000000083708,0,65.75966666673594,1.706667;1.706667;1.706667;1.706667,1.7066666666778474,True,12.830822611113824,19.573666666701683,20.16000000007523,0 +constant_300,latest_only,10.082484055552035,17.466333333331363,17.787000000083708,0,65.75966666673594,1.706667;1.706667;1.706667;1.706667,1.7066666666778474,True,12.830822611113824,19.573666666701683,20.16000000007523,0 +constant_300,two_waiting,10.082484055552035,17.466333333331363,17.787000000083708,0,65.75966666673594,1.706667;1.706667;1.706667;1.706667,1.7066666666778474,True,12.830822611113824,19.573666666701683,20.16000000007523,0 +constant_260,no_drop,11.072842185081562,19.651615385654253,20.66700023078738,0,68.4307692308721,18.420846;16.882385;4.974692;3.251616,18.420846156274706,True,14.047702896620706,22.63623076796577,23.62084638127726,0 +constant_260,latest_only,10.616307327810727,19.04615384603403,20.667000017851933,0,67.99999999995521,7.528538;1.969231;10.513154;7.528538,10.513153850581602,True,13.756108911150676,22.143923060525594,23.590076939512983,0 +constant_260,two_waiting,11.020678082480492,19.651615408761813,20.667000205151,0,68.4000000001106,18.420846;3.713154;11.620846;15.897769,18.420846156274706,True,13.98955930684015,22.63623076432637,23.62084635899464,0 +constant_230,no_drop,12.447544944822914,22.156521740192048,23.3565218202898,0,70.12173913045672,14.295652;5.391304;15.895652;10.817391,15.895652187168707,True,15.792707263668593,25.49565216446439,26.678260933408637,0 +constant_230,latest_only,12.496684077687183,22.156521750258662,23.356521818243436,0,69.60000000003674,6.852174;4.660870;9.808696;10.817391,10.817391378679986,True,15.7147652371126,25.426086964586148,26.678260937728737,0 +constant_230,two_waiting,12.432011596181523,22.191304351743923,23.35652178805958,0,70.64347826093353,2.295652;4.556522;10.608696;12.765217,12.765217432615827,True,15.712892755606843,25.495652190528517,26.678260914195562,0 +step_300_230_300,no_drop,10.645963064766441,19.408695629502404,23.6869565184179,0,69.00869565220091,1.706667;21.460870;5.680000;1.706667,21.460869561906293,True,13.517029441587948,22.678260879320078,27.026086953213735,0 +step_300_230_300,latest_only,10.812854489366694,19.44434781658656,23.356521740737435,0,69.00869565220091,1.706667;3.791304;1.706667;1.706667,3.7913043232720156,True,13.695971329953036,22.54086954938544,26.678260865082848,0 +step_300_230_300,two_waiting,10.76303822392693,19.33913042324491,23.35652172334335,0,70.64347826081985,1.706667;3.826087;1.706667;1.706667,3.826086949771934,True,13.677599522246727,22.749565219797326,26.67826085848901,0 +variable_300_260_230,no_drop,11.053582892926142,20.403454847835434,23.369899646013437,0,69.80869565218484,7.528538;19.076923;8.359308;9.090724,19.076923066961626,True,14.018366272815697,23.467000011780215,26.679040185285885,0 +variable_300_260_230,latest_only,11.040001685623347,20.19402340485995,23.373354529439894,0,69.00869565220091,7.528538;1.969231;10.513154;14.593065,14.593064683936063,True,14.14441419306391,23.375628775394834,26.704458198324232,0 +variable_300_260_230,two_waiting,11.097355222802879,20.30926972252018,23.36989965402836,0,69.00869565220091,7.528538;19.076923;8.359308;10.482028,19.076923066961626,True,14.058443590060339,23.415384582420984,26.696989889956058,0 diff --git a/data/processed/lab036/lab036_frame_outcomes.png b/data/processed/lab036/lab036_frame_outcomes.png new file mode 100644 index 0000000..dc132a4 Binary files /dev/null and b/data/processed/lab036/lab036_frame_outcomes.png differ diff --git a/data/processed/lab036/lab036_image_age_timeline.png b/data/processed/lab036/lab036_image_age_timeline.png new file mode 100644 index 0000000..ef34ef9 Binary files /dev/null and b/data/processed/lab036/lab036_image_age_timeline.png differ diff --git a/data/processed/lab036/lab036_minute_update_rate.png b/data/processed/lab036/lab036_minute_update_rate.png new file mode 100644 index 0000000..6526561 Binary files /dev/null and b/data/processed/lab036/lab036_minute_update_rate.png differ diff --git a/data/processed/lab036/lab036_policy_comparison.png b/data/processed/lab036/lab036_policy_comparison.png new file mode 100644 index 0000000..fac57ff Binary files /dev/null and b/data/processed/lab036/lab036_policy_comparison.png differ diff --git a/data/processed/lab036/lab036_queue_growth_theory.png b/data/processed/lab036/lab036_queue_growth_theory.png new file mode 100644 index 0000000..e00ab2a Binary files /dev/null and b/data/processed/lab036/lab036_queue_growth_theory.png differ diff --git a/data/processed/lab036/lab036_queue_timeline.png b/data/processed/lab036/lab036_queue_timeline.png new file mode 100644 index 0000000..6d34c38 Binary files /dev/null and b/data/processed/lab036/lab036_queue_timeline.png differ diff --git a/data/processed/lab036/lab036_recovery.png b/data/processed/lab036/lab036_recovery.png new file mode 100644 index 0000000..6883d8a Binary files /dev/null and b/data/processed/lab036/lab036_recovery.png differ diff --git a/data/processed/lab036/lab036_recovery_metrics.csv b/data/processed/lab036/lab036_recovery_metrics.csv new file mode 100644 index 0000000..fb9a4a8 --- /dev/null +++ b/data/processed/lab036/lab036_recovery_metrics.csv @@ -0,0 +1,16 @@ +scenario,policy,applicable,image_age_at_300s_ms,time_to_age_below_1000ms_seconds,time_to_age_below_500ms_seconds,excess_queue_clear_seconds,frames_dropped_after_recovery,old_frames_published_after_recovery +constant_300,no_drop,False,0.0,0.0,0.0,0.0,0,0 +constant_300,latest_only,False,0.0,0.0,0.0,0.0,0,0 +constant_300,two_waiting,False,0.0,0.0,0.0,0.0,0,0 +constant_260,no_drop,False,0.0,0.0,0.0,0.0,0,0 +constant_260,latest_only,False,0.0,0.0,0.0,0.0,0,0 +constant_260,two_waiting,False,0.0,0.0,0.0,0.0,0,0 +constant_230,no_drop,False,0.0,0.0,0.0,0.0,0,0 +constant_230,latest_only,False,0.0,0.0,0.0,0.0,0,0 +constant_230,two_waiting,False,0.0,0.0,0.0,0.0,0,0 +step_300_230_300,no_drop,True,24333.333333333314,147.64077333334558,149.80178666667894,150.02250666667896,0,72 +step_300_230_300,latest_only,True,666.6666666666856,0.0,0.07729391299602639,0.00844057966270384,0,1 +step_300_230_300,two_waiting,True,1000.0,0.05866202900034523,1.1352486956670305,1.3550620290003508,0,2 +variable_300_260_230,no_drop,False,0.0,0.0,0.0,0.0,0,0 +variable_300_260_230,latest_only,False,0.0,0.0,0.0,0.0,0,0 +variable_300_260_230,two_waiting,False,0.0,0.0,0.0,0.0,0,0 diff --git a/data/processed/lab036/lab036_report.txt b/data/processed/lab036/lab036_report.txt new file mode 100644 index 0000000..1d87880 --- /dev/null +++ b/data/processed/lab036/lab036_report.txt @@ -0,0 +1,104 @@ +Lab036. Долговременная устойчивость видеопланировщика при изменении пропускной способности + +1. Исходное состояние и конфигурация +- Commit Lab035: 22c2eeab8cc794e67378485f49c236a622f814a7. +- Рабочее дерево перед Lab036 было чистым; ветка main. +- Модель использует 600 с, 1800 кадров, повтор реальной последовательности размеров 63 кадров, BASE 240×135 Q23, ROI 320×180 Q33 и 3 составных кадра/с. +- Lab036 использует ровно 3 кадра/с в течение 600 с; короткий ролик Lab034 и предшествующих лабораторных давал немного отличающуюся фактическую частоту формирования кадров. +- Длинный опыт уменьшает влияние начального и конечного участков, поэтому значения общей предложенной нагрузки Lab034 и Lab036 не обязаны побайтно совпадать. +- Каждый повтор получает новые composite_frame_id, времена, заголовки и CRC; JPEG и пакеты на диск не записываются. +- Используются внутренний пакет Lab028 с payload 512 байт, FEC 12+3 с выравниванием по кадрам и общий пакет Lab033. +- Команды 20 Гц, телеметрия 10 Гц; аварийные команды создаются в 60, 180, 310 и 480 с. +- Ошибки радиоканала отсутствуют. +- Смена скорости не прерывает начатый пакет: новая скорость применяется только к следующему пакету после завершения текущего. + +2. Пятнадцать сочетаний +scenario | policy | published/drop | fps | age P95 ms | queue max packets | queue at 600 packets | drain s | control P95 ms | emergency max ms +constant_300 | no_drop | 1800/0 | 3.000 | 600.000 | 22 | 0 | 0.000 | 17.466 | 1.707 +constant_300 | latest_only | 1800/0 | 3.000 | 600.000 | 22 | 0 | 0.000 | 17.466 | 1.707 +constant_300 | two_waiting | 1800/0 | 3.000 | 600.000 | 22 | 0 | 0.000 | 17.466 | 1.707 +constant_260 | no_drop | 1800/0 | 2.977 | 4866.667 | 268 | 238 | 4.146 | 19.652 | 18.421 +constant_260 | latest_only | 1772/28 | 2.953 | 833.333 | 40 | 0 | 0.000 | 19.046 | 10.513 +constant_260 | two_waiting | 1787/13 | 2.977 | 1166.667 | 58 | 6 | 0.084 | 19.652 | 18.421 +constant_230 | no_drop | 1800/0 | 2.608 | 74633.333 | 4155 | 4139 | 82.753 | 22.157 | 15.896 +constant_230 | latest_only | 1570/230 | 2.615 | 1000.000 | 40 | 17 | 0.317 | 22.157 | 10.817 +constant_230 | two_waiting | 1571/229 | 2.615 | 1333.333 | 57 | 26 | 0.486 | 22.191 | 12.765 +step_300_230_300 | no_drop | 1800/0 | 3.000 | 21966.667 | 1279 | 0 | 0.000 | 19.409 | 21.461 +step_300_230_300 | latest_only | 1730/70 | 2.883 | 900.000 | 40 | 0 | 0.000 | 19.444 | 3.791 +step_300_230_300 | two_waiting | 1731/69 | 2.885 | 1233.333 | 58 | 0 | 0.000 | 19.339 | 3.826 +variable_300_260_230 | no_drop | 1800/0 | 2.970 | 5666.667 | 326 | 295 | 5.804 | 20.403 | 19.077 +variable_300_260_230 | latest_only | 1716/84 | 2.858 | 900.000 | 40 | 14 | 0.261 | 20.194 | 14.593 +variable_300_260_230 | two_waiting | 1725/75 | 2.872 | 1233.333 | 56 | 29 | 0.562 | 20.309 | 19.077 + +3. Постоянные скорости: теория и модель +rate | policy | reserve kbps | theory growth B/s | model growth B/s | minimum skip | stable fps | minute queue packets +300 | no_drop | 38.278 | 0.000 | -11.601 | 0.000000 | 3.000 | 21;21;21;20;19;21;21;21;21;0 +300 | latest_only | 38.278 | 0.000 | -11.601 | 0.000000 | 3.000 | 21;21;21;20;19;21;21;21;21;0 +300 | two_waiting | 38.278 | 0.000 | -11.601 | 0.000000 | 3.000 | 21;21;21;20;19;21;21;21;21;0 +260 | no_drop | -1.722 | 215.258 | 218.649 | 0.007063 | 2.979 | 48;65;90;117;149;176;197;214;233;238 +260 | latest_only | -1.722 | 215.258 | -5.668 | 0.007063 | 2.979 | 32;24;21;20;21;24;39;32;24;0 +260 | two_waiting | -1.722 | 215.258 | -12.199 | 0.007063 | 2.979 | 48;29;35;43;39;49;52;37;37;6 +230 | no_drop | -31.722 | 3965.258 | 3967.819 | 0.130116 | 2.610 | 429;841;1255;1672;2093;2511;2924;3335;3745;4139 +230 | latest_only | -31.722 | 3965.258 | -6.550 | 0.130116 | 2.610 | 37;33;38;26;34;26;28;37;38;17 +230 | two_waiting | -31.722 | 3965.258 | -22.350 | 0.130116 | 2.610 | 42;56;52;53;44;55;55;47;43;26 + +4. Восстановление после ухудшения +policy | age at 300 ms | below 1000 s | below 500 s | excess queue clear s | dropped old | published old +no_drop | 24333.333 | 147.641 | 149.802 | 150.023 | 0 | 72 +latest_only | 666.667 | 0.000 | 0.077 | 0.008 | 0 | 1 +two_waiting | 1000.000 | 0.059 | 1.135 | 1.355 | 0 | 2 + +5. Видео, очередь, команды и телеметрия +- constant_300: fps=3.000, age P95=600.000 мс, queue max=22, waiting max=1, control P95/max=17.466/17.787 мс, emergency=1.706667;1.706667;1.706667;1.706667 мс, telemetry P95=19.574 мс. +- constant_260: fps=2.953, age P95=833.333 мс, queue max=40, waiting max=1, control P95/max=19.046/20.667 мс, emergency=7.528538;1.969231;10.513154;7.528538 мс, telemetry P95=22.144 мс. +- constant_230: fps=2.615, age P95=1000.000 мс, queue max=40, waiting max=1, control P95/max=22.157/23.357 мс, emergency=6.852174;4.660870;9.808696;10.817391 мс, telemetry P95=25.426 мс. +- step_300_230_300: fps=2.883, age P95=900.000 мс, queue max=40, waiting max=1, control P95/max=19.444/23.357 мс, emergency=1.706667;3.791304;1.706667;1.706667 мс, telemetry P95=22.541 мс. +- variable_300_260_230: fps=2.858, age P95=900.000 мс, queue max=40, waiting max=1, control P95/max=20.194/23.373 мс, emergency=7.528538;1.969231;10.513154;14.593065 мс, telemetry P95=23.376 мс. +- Политика самого свежего кадра ограничивает ожидающую видеоочередь одним кадром и не передаёт накопленную историю после восстановления. +- Политика двух ожидающих кадров ограничивает очередь двумя кадрами, но увеличивает возраст изображения относительно основной политики. +- Без удаления при дефиците скорости очередь растёт; при 300 кбит/с запас достаточен. +- У упреждающих политик нет бесполезно переданных видеобайтов: начатый кадр всегда завершается. + +6. Функциональные проверки +- PASS 01_unique_repeated_frames_and_crc: PASS +- PASS 02_300kbps_latest_has_no_unnecessary_drop: PASS +- PASS 03_260kbps_no_drop_matches_small_deficit: PASS +- PASS 04_230kbps_no_drop_grows_faster: PASS +- PASS 05_latest_waiting_limit: PASS +- PASS 06_two_waiting_limit: PASS +- PASS 07_started_frame_never_dropped: PASS +- PASS 08_active_video_frames_do_not_interleave: PASS +- PASS 09_priority_between_video_packets: PASS +- PASS 10_emergency_never_deleted: PASS +- PASS 11_speed_change_bit_accounting: PASS +- PASS 12_latest_does_not_replay_full_history: PASS +- PASS 13_proactive_policies_have_no_waste: PASS +- PASS 14_incomplete_frame_not_published: PASS +- PASS 15_all_three_crc_layers_pass: PASS +- PASS 16_frame_accounting: PASS +- PASS 17_packet_and_byte_accounting: PASS +- PASS 18_reproducible: PASS +- CRC-проверено общих пакетов: 49794; FEC-блоков: 3571. + +7. Созданные файлы +- tests/lab036_long_duration_scheduler.py +- data/processed/lab036/lab036_summary.csv +- data/processed/lab036/lab036_video_metrics.csv +- data/processed/lab036/lab036_control_metrics.csv +- data/processed/lab036/lab036_recovery_metrics.csv +- data/processed/lab036/lab036_report.txt +- data/processed/lab036/lab036_image_age_timeline.png +- data/processed/lab036/lab036_queue_timeline.png +- data/processed/lab036/lab036_minute_update_rate.png +- data/processed/lab036/lab036_recovery.png +- data/processed/lab036/lab036_control_delay.png +- data/processed/lab036/lab036_frame_outcomes.png +- data/processed/lab036/lab036_queue_growth_theory.png +- data/processed/lab036/lab036_policy_comparison.png + +8. Итоговый Git status +- Lab036 не добавлена в индекс и не закоммичена. + +## main...origin/main +?? data/processed/lab036/ +?? tests/lab036_long_duration_scheduler.py diff --git a/data/processed/lab036/lab036_summary.csv b/data/processed/lab036/lab036_summary.csv new file mode 100644 index 0000000..d5625d5 --- /dev/null +++ b/data/processed/lab036/lab036_summary.csv @@ -0,0 +1,16 @@ +scenario,policy,duration_seconds,mean_channel_kbps,offered_load_kbps,speed_reserve_kbps,theoretical_queue_growth_bytes_per_second,theoretical_minimum_skip_fraction,theoretical_sustainable_fps,actual_queue_growth_bytes_per_second,actual_queue_growth_bytes_per_minute,unbounded_queue_growth,mean_queue_packets,max_queue_packets,mean_queue_bytes,max_queue_bytes,mean_waiting_video_frames,max_waiting_video_frames,queue_packets_each_minute,queue_bytes_each_minute,queue_at_600s_packets,queue_at_600s_bytes,frames_remaining_at_600s,additional_drain_seconds,transmitted_packets,transmitted_bytes,replaced_state_packets +constant_300,no_drop,600.0,300.0,261.7220666666667,38.277933333333294,0.0,0.0,3.0,-11.600833333333322,-696.0499999999993,False,8.492667996111212,22,4669.600783430745,11415,0.004278044444460347,1,21;21;21;20;19;21;21;21;21;0,10503;11309;10462;10676;10137;11083;11415;10503;11309;0,0,0,0,0.0,49794,19629155,0 +constant_300,latest_only,600.0,300.0,261.7220666666667,38.277933333333294,0.0,0.0,3.0,-11.600833333333322,-696.0499999999993,False,8.492667996111212,22,4669.600783430745,11415,0.004278044444460347,1,21;21;21;20;19;21;21;21;21;0,10503;11309;10462;10676;10137;11083;11415;10503;11309;0,0,0,0,0.0,49794,19629155,0 +constant_300,two_waiting,600.0,300.0,261.7220666666667,38.277933333333294,0.0,0.0,3.0,-11.600833333333322,-696.0499999999993,False,8.492667996111212,22,4669.600783430745,11415,0.004278044444460347,1,21;21;21;20;19;21;21;21;21;0,10503;11309;10462;10676;10137;11083;11415;10503;11309;0,0,0,0,0.0,49794,19629155,0 +constant_260,no_drop,600.0,260.0,261.7220666666667,-1.7220666666667057,215.25833333333821,0.007063478994333017,2.978809563017001,218.64944444444447,13118.966666666669,True,133.31816348352757,268,76437.33087055078,153293,6.995598665791334,15,48;65;90;117;149;176;197;214;233;238,26336;36685;49266;66106;84626;100392;113622;122901;132742;135216,238,135216,14,4.146236230854015,49794,19629155,0 +constant_260,latest_only,600.0,260.0,261.7220666666667,-1.7220666666667057,215.25833333333821,0.007063478994333017,2.978809563017001,-5.668055555555536,-340.0833333333321,False,13.60260795417588,40,7605.446317388057,22420,0.2193330062846088,1,32;24;21;20;21;24;39;32;24;0,17102;12852;10462;10676;10831;12075;21812;17102;12852;0,0,0,0,0.0,49262,19314015,0 +constant_260,two_waiting,600.0,260.0,261.7220666666667,-1.7220666666667057,215.25833333333821,0.007063478994333017,2.978809563017001,-12.19888888888884,-731.9333333333304,False,28.79937767032035,58,16327.143562298448,33090,1.0761970065603674,2,48;29;35;43;39;49;52;37;37;6,26336;15892;18207;24279;21038;26530;29623;19832;20554;3230,6,3230,1,0.0844823847068028,49565,19497148,0 +constant_230,no_drop,600.0,230.0,261.7220666666667,-31.722066666666706,3965.258333333338,0.13011584039922786,2.6096524788023165,3967.8194444444443,238069.16666666666,True,2079.4202965547893,4155,1195926.0317054717,2388166,117.14617014497261,235,429;841;1255;1672;2093;2511;2924;3335;3745;4139,245926;481391;719062;960442;1204374;1444453;1683007;1917098;2152265;2379540,4139,2379540,235,82.75321739112997,49794,19629155,0 +constant_230,latest_only,600.0,230.0,261.7220666666667,-31.722066666666706,3965.258333333338,0.13011584039922786,2.6096524788023165,-6.549722222222197,-392.9833333333318,False,19.634379733025998,40,11027.21948478135,22420,0.5571525326601725,1,37;33;38;26;34;26;28;37;38;17,19832;18324;20031;14224;18420;13291;15252;19832;21162;9569,17,9569,1,0.3167652173551687,45694,17259107,0 +constant_230,two_waiting,600.0,230.0,261.7220666666667,-31.722066666666706,3965.258333333338,0.13011584039922786,2.6096524788023165,-22.350277777777745,-1341.0166666666646,False,37.376577605014,57,21232.67400510985,32658,1.5629541073964863,2,42;56;52;53;44;55;55;47;43;26,22495;31370;27643;30124;24078;29921;31266;25728;23820;14523,26,14523,2,0.4859478260450487,45709,17263971,0 +step_300_230_300,no_drop,600.0,279.0,261.7220666666667,17.277933333333294,0.0,0.0,3.0,-636.2352777777767,-38174.1166666666,False,356.1464225450075,1279,204638.53781412638,734558,19.64118583244118,72,21;21;440;854;1279;781;283;21;21;0,10503;11309;248843;491202;734558;449558;163347;10503;11309;0,0,0,0,0.0,49794,19629155,0 +step_300_230_300,latest_only,600.0,279.0,261.7220666666667,17.277933333333294,0.0,0.0,3.0,-16.11861111111109,-967.1166666666654,False,11.838952278707378,40,6580.107300552987,22420,0.17015513173553526,1,21;21;25;26;25;21;21;21;21;0,10503;11309;12476;14224;13263;11083;11415;10503;11309;0,0,0,0,0.0,48547,18907911,0 +step_300_230_300,two_waiting,600.0,279.0,261.7220666666667,17.277933333333294,0.0,0.0,3.0,-36.86027777777775,-2211.616666666665,False,17.155767414441645,58,9639.972938771809,33090,0.47114729028487184,2,21;21;46;52;40;21;21;21;21;0,10503;11309;24377;29516;21646;11083;11415;10503;11309;0,0,0,0,0.0,48563,18916663,0 +variable_300_260_230,no_drop,600.0,263.3333333333333,261.7220666666667,1.6112666666666087,0.0,0.0,3.0,-11.902777777777905,-714.1666666666742,False,103.59885527746508,326,59342.94770940146,187430,5.320544485386984,18,32;311;30;321;41;326;39;315;28;295,17102;177237;15516;182330;22254;187430;21812;181426;15284;167261,295,167261,18,5.803705685504838,49794,19629155,0 +variable_300_260_230,latest_only,600.0,263.3333333333333,261.7220666666667,1.6112666666666087,0.0,0.0,3.0,-9.321944444444432,-559.3166666666659,False,14.017274428919372,40,7830.931831177453,22420,0.2651953377439788,1,32;38;21;35;21;24;39;38;24;14,17102;21162;10462;19461;10831;12463;21812;20440;12852;7745,14,7745,1,0.2608542909119933,48300,18769207,0 +variable_300_260_230,two_waiting,600.0,263.3333333333333,261.7220666666667,1.6112666666666087,0.0,0.0,3.0,-18.437222222222193,-1106.2333333333315,False,23.386572906900973,56,13214.022895715687,32658,0.7951346653714808,2,32;56;30;50;41;42;39;47;28;29,17102;31370;15516;28300;22254;22274;21812;25728;15284;16347,29,16347,2,0.5623143811660611,48458,18857145,0 diff --git a/data/processed/lab036/lab036_video_metrics.csv b/data/processed/lab036/lab036_video_metrics.csv new file mode 100644 index 0000000..715f734 --- /dev/null +++ b/data/processed/lab036/lab036_video_metrics.csv @@ -0,0 +1,16 @@ +scenario,policy,created_frames,started_frames,published_frames,published_by_600s_frames,dropped_before_start_frames,published_fraction,actual_update_fps,update_fps_each_minute,mean_publication_delay_ms,p95_publication_delay_ms,max_publication_delay_ms,mean_display_age_ms,p95_display_age_ms,max_display_age_ms,display_age_over_500ms_fraction,display_age_over_1000ms_fraction,mean_no_update_duration_ms,p95_no_update_duration_ms,max_no_update_duration_ms,max_consecutive_missing_frames,mean_publication_gap_ms,max_publication_gap_ms,transmitted_video_bytes,wasted_transmitted_video_bytes,delivered_useful_video_kbps +constant_300,no_drop,1800,1800,1800,1800,0,1.0,3.0,3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000,287.5610518518373,311.3870000000247,322.3200000001043,459.1456979392323,600.0000000000227,633.3333333333542,0.3277787035494084,0.0,333.1482509716824,359.040666666667,380.9063333333427,0,333.30502149342215,380.9063333333427,18284899,0,155.94405333333336 +constant_300,latest_only,1800,1800,1800,1800,0,1.0,3.0,3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000,287.5610518518373,311.3870000000247,322.3200000001043,459.1456979392323,600.0000000000227,633.3333333333542,0.3277787035494084,0.0,333.1482509716824,359.040666666667,380.9063333333427,0,333.30502149342215,380.9063333333427,18284899,0,155.94405333333336 +constant_300,two_waiting,1800,1800,1800,1800,0,1.0,3.0,3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000;3.000000,287.5610518518373,311.3870000000247,322.3200000001043,459.1456979392323,600.0000000000227,633.3333333333542,0.3277787035494084,0.0,333.1482509716824,359.040666666667,380.9063333333427,0,333.30502149342215,380.9063333333427,18284899,0,155.94405333333336 +constant_260,no_drop,1800,1800,1800,1786,0,1.0,2.9766666666666666,2.966667;2.983333;2.983333;2.966667;2.966667;2.983333;2.983333;2.983333;2.966667;2.983333,2680.896184452496,4704.89366684435,5049.795205226246,2829.13958784647,4866.666666666674,5366.666666666674,0.9866688885185803,0.9030161639726713,335.75825405707894,364.2153846158749,377.67146153846375,0,335.90542861457425,377.67146153846375,18284899,0,154.79058666666668 +constant_260,latest_only,1800,1772,1772,1772,28,0.9844444444444445,2.953333333333333,2.950000;2.950000;2.966667;2.950000;2.933333;2.950000;2.966667;2.950000;2.950000;2.966667,402.5770375508439,599.6618718049831,680.6464871910407,571.0159417874798,833.3333333333428,966.666666666697,0.6213964339276787,0.0,338.40947546531305,365.6923076928251,377.6714615384691,1,338.57024154106733,377.6714615384691,17969759,0,153.18773333333334 +constant_260,two_waiting,1800,1787,1787,1786,13,0.9927777777777778,2.9766666666666666,2.966667;2.983333;2.983333;2.966667;2.966667;2.983333;2.983333;2.983333;2.983333;2.966667,690.9434287448678,934.1911025966964,1036.8823846180817,858.9124034883074,1166.6666666666288,1366.6666666666458,0.9716713881019831,0.19513414430928178,335.75825405707894,364.2153846153668,377.67146153846375,1,335.799658306451,377.67146153846375,18152892,0,154.73507999999998 +constant_230,no_drop,1800,1800,1800,1565,0,1.0,2.6083333333333334,2.616667;2.600000;2.616667;2.600000;2.616667;2.600000;2.616667;2.600000;2.600000;2.616667,44808.52197101605,81595.11768103977,83092.17391287029,39545.96456146199,74633.33333333327,78799.99999999996,0.9990001666388936,0.9938343609398433,383.1417624521073,416.486956521112,428.2086956523017,0,383.23811853661965,428.2086956523017,18284899,0,135.61342666666667 +constant_230,latest_only,1800,1570,1570,1569,230,0.8722222222222222,2.615,2.616667;2.616667;2.616667;2.600000;2.616667;2.600000;2.616667;2.633333;2.616667;2.616667,544.8765660678318,697.5368116767044,748.5217391683818,736.8216408376383,1000.0,1100.0000000000227,0.9208465255790702,0.03166138976837194,382.1656050955414,416.486956521112,428.2086956523017,1,382.3697204968717,428.2086956523017,15914851,0,135.62884 +constant_230,two_waiting,1800,1571,1571,1569,229,0.8727777777777778,2.615,2.600000;2.633333;2.616667;2.616667;2.600000;2.616667;2.616667;2.616667;2.616667;2.616667,880.3126375723886,1028.5507246275927,1075.6637681217853,1076.042881741932,1333.3333333333144,1466.666666666697,0.9990001666388936,0.648891851358107,382.1656050955414,416.4869565213465,428.2086956523017,1,382.2583629103555,428.2086956523017,15919715,0,135.59438666666665 +step_300_230_300,no_drop,1800,1800,1800,1800,0,1.0,3.0,3.000000;3.000000;2.583333;2.616667;2.600000;3.483333;3.466667;3.250000;3.000000;3.000000,6858.4316214417395,21791.709275375164,24108.666666679823,7025.662389601733,21966.666666666697,24433.333333333336,0.6963839360106648,0.5420763206132311,333.1482509716824,399.7217391305412,457.1161884057773,0,333.30502149342215,457.1161884057773,18284899,0,155.94405333333336 +step_300_230_300,latest_only,1800,1730,1730,1730,70,0.9611111111111111,2.8833333333333333,3.000000;3.000000;2.600000;2.616667;2.600000;3.016667;3.000000;3.000000;3.000000;3.000000,357.42493844221974,650.9101448891982,741.3565216782843,542.1318669110703,900.0000000000057,1099.9999999999943,0.5044159306782203,0.009498416930511582,346.6204506065858,399.82608695632393,457.1161884057773,1,346.7991519182571,457.1161884057773,17563655,0,149.78249333333335 +step_300_230_300,two_waiting,1800,1731,1731,1731,69,0.9616666666666667,2.885,3.000000;3.000000;2.600000;2.616667;2.600000;3.033333;3.000000;3.000000;3.000000;3.000000,448.4849129664068,983.2637681128205,1075.8724637457817,644.6147864244849,1233.3333333333485,1466.6666666666686,0.5299116813864356,0.1948008665222463,346.4203233256351,400.1234782608647,457.1161884057773,1,346.5986899807321,457.1161884057773,17572407,0,149.85798666666668 +variable_300_260_230,no_drop,1800,1800,1800,1782,0,1.0,2.97,2.983333;2.733333;3.266667;2.716667;3.266667;2.750000;3.266667;2.733333;3.266667;2.716667,2130.1320615592713,5594.812172849652,6163.147290942874,2266.28895184136,5666.6666666666715,6433.333333333338,0.8961839693384436,0.5707382102982836,336.51149747616375,400.4869565215188,428.2086956523017,0,336.53952764120913,428.2086956523017,18284899,0,154.45788000000002 +variable_300_260_230,latest_only,1800,1716,1716,1715,84,0.9533333333333334,2.8583333333333334,2.983333;2.733333;3.000000;2.716667;2.983333;2.716667;3.000000;2.733333;2.983333;2.733333,408.07981144593083,663.8076912153963,740.7304347776176,589.1795811809143,900.0000000000057,1100.0000000000227,0.6287285452424596,0.010998166972171305,349.65034965034965,400.48695652183494,428.2086956523017,1,349.8269973286741,428.2086956523017,17424951,0,148.50604 +variable_300_260_230,two_waiting,1800,1725,1725,1723,75,0.9583333333333334,2.8716666666666666,2.983333;2.750000;3.016667;2.733333;3.000000;2.733333;3.016667;2.733333;3.016667;2.733333,579.2497507600725,989.023923120439,1070.3080256992052,767.0943731600289,1233.3333333333344,1433.3333333333371,0.7512081319780036,0.2422929511748042,348.0278422273782,400.48695652188826,424.8695652167953,1,348.18301951135805,424.8695652167953,17512889,0,149.19149333333334 diff --git a/tests/lab036_long_duration_scheduler.py b/tests/lab036_long_duration_scheduler.py new file mode 100644 index 0000000..0f95b46 --- /dev/null +++ b/tests/lab036_long_duration_scheduler.py @@ -0,0 +1,1361 @@ +"""Lab036: long-duration whole-frame scheduling with changing capacity.""" + +from __future__ import annotations + +import csv +from dataclasses import asdict, dataclass, replace +import hashlib +from pathlib import Path +import subprocess +from typing import Iterable + +import cv2 +import matplotlib +import numpy as np + +matplotlib.use("Agg") +import matplotlib.pyplot as plt + +from protocol.link_packet import ( + Direction, + LinkPacket, + TrafficClass, + decode_link_packet, + encode_link_packet, +) +from protocol.packet_erasure_fec import ( + decode_fec_block, + decode_outer_symbol, + encode_fec_block, +) +from protocol.video_age_policy import AgePolicyPacket +from protocol.video_frame_scheduler import FramePolicy, VideoFrameGroup +from protocol.video_packet import decode_packet as decode_inner_packet +from tests.lab028_video_packetization import ( + COMPOSITE_FPS, + EncodedComposite, + packets_for_composite, +) +from tests.lab033_priority_channel_scheduler import ( + STREAM_CONTROL, + STREAM_EMERGENCY, + STREAM_TELEMETRY, + STREAM_VIDEO, + build_workload as build_lab033_workload, + deterministic_payload, +) +from tests.lab034_stale_video_drop import parity_for_partial + + +OUTPUT_DIRECTORY = Path("data/processed/lab036") +SUMMARY_CSV_PATH = OUTPUT_DIRECTORY / "lab036_summary.csv" +VIDEO_CSV_PATH = OUTPUT_DIRECTORY / "lab036_video_metrics.csv" +CONTROL_CSV_PATH = OUTPUT_DIRECTORY / "lab036_control_metrics.csv" +RECOVERY_CSV_PATH = OUTPUT_DIRECTORY / "lab036_recovery_metrics.csv" +REPORT_PATH = OUTPUT_DIRECTORY / "lab036_report.txt" +AGE_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_image_age_timeline.png" +QUEUE_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_queue_timeline.png" +MINUTE_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_minute_update_rate.png" +RECOVERY_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_recovery.png" +CONTROL_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_control_delay.png" +OUTCOME_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_frame_outcomes.png" +GROWTH_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_queue_growth_theory.png" +COMPARISON_PLOT_PATH = OUTPUT_DIRECTORY / "lab036_policy_comparison.png" +PLOT_PATHS = ( + AGE_PLOT_PATH, + QUEUE_PLOT_PATH, + MINUTE_PLOT_PATH, + RECOVERY_PLOT_PATH, + CONTROL_PLOT_PATH, + OUTCOME_PLOT_PATH, + GROWTH_PLOT_PATH, + COMPARISON_PLOT_PATH, +) + +LAB035_COMMIT = "22c2eeab8cc794e67378485f49c236a622f814a7" +DURATION_SECONDS = 600.0 +FRAME_COUNT = int(DURATION_SECONDS * COMPOSITE_FPS) +VIDEO_PAYLOAD_BYTES = 512 +SOURCE_BLOCK_SIZE = 12 +NOMINAL_PARITY_COUNT = 3 +CONTROL_PERIOD_US = 50_000 +TELEMETRY_PERIOD_US = 100_000 +EMERGENCY_TIMES_US = (60_000_000, 180_000_000, 310_000_000, 480_000_000) +TIME_EPSILON_SECONDS = 1e-9 + + +@dataclass(frozen=True) +class SpeedProfile: + name: str + label: str + kind: str + constant_kbps: float | None = None + + def rate_kbps(self, time_seconds: float) -> float: + if self.kind == "constant": + assert self.constant_kbps is not None + return self.constant_kbps + if self.kind == "step": + if time_seconds < 120.0: + return 300.0 + if time_seconds < 300.0: + return 230.0 + return 300.0 + if self.kind == "variable": + if time_seconds >= DURATION_SECONDS: + return 230.0 + phase = time_seconds % 120.0 + return 300.0 if phase < 40.0 else 260.0 if phase < 80.0 else 230.0 + raise ValueError(f"unknown speed profile: {self.kind}") + + @property + def mean_rate_kbps(self) -> float: + if self.kind == "constant": + assert self.constant_kbps is not None + return self.constant_kbps + if self.kind == "step": + return (120.0 * 300.0 + 180.0 * 230.0 + 300.0 * 300.0) / DURATION_SECONDS + return (300.0 + 260.0 + 230.0) / 3.0 + + +@dataclass(frozen=True) +class PolicyDefinition: + name: str + label: str + policy: FramePolicy + + +SPEED_PROFILES = ( + SpeedProfile("constant_300", "Постоянно 300", "constant", 300.0), + SpeedProfile("constant_260", "Постоянно 260", "constant", 260.0), + SpeedProfile("constant_230", "Постоянно 230", "constant", 230.0), + SpeedProfile("step_300_230_300", "300→230→300", "step"), + SpeedProfile("variable_300_260_230", "Цикл 300/260/230", "variable"), +) +POLICIES = ( + PolicyDefinition("no_drop", "Без удаления", FramePolicy.NO_DROP), + PolicyDefinition("latest_only", "Самый свежий", FramePolicy.LATEST_ONLY), + PolicyDefinition("two_waiting", "Два ожидающих", FramePolicy.TWO_WAITING), +) +PROFILE_BY_NAME = {item.name: item for item in SPEED_PROFILES} +POLICY_BY_NAME = {item.name: item for item in POLICIES} + + +@dataclass(frozen=True) +class LongWorkload: + frames: tuple[VideoFrameGroup, ...] + high_priority: tuple[AgePolicyPacket, ...] + jpeg_bytes_by_frame: tuple[int, ...] + original_frame_count: int + video_wire_bytes: int + nonvideo_wire_bytes: int + crc_packets_checked: int + crc_blocks_checked: int + repeated_headers_are_fresh: bool + + @property + def offered_wire_bytes(self) -> int: + return self.video_wire_bytes + self.nonvideo_wire_bytes + + +@dataclass(frozen=True) +class DroppedFrame: + frame: VideoFrameGroup + time_seconds: float + reason: str + + +@dataclass(frozen=True) +class Replacement: + removed: AgePolicyPacket + replacement: AgePolicyPacket + time_seconds: float + + +@dataclass(frozen=True) +class ScheduledPacket: + item: AgePolicyPacket + start_seconds: float + end_seconds: float + rate_kbps: float + blocked_by: AgePolicyPacket | None + blocking_delay_seconds: float + + +@dataclass(frozen=True) +class LongSchedule: + profile: SpeedProfile + policy: PolicyDefinition + transmitted: tuple[ScheduledPacket, ...] + dropped: tuple[DroppedFrame, ...] + replacements: tuple[Replacement, ...] + started_frame_ids: tuple[int, ...] + completed_frame_ids: tuple[int, ...] + publication_times: dict[int, float] + digest: str + bit_accounting_error: float + rate_rule_ok: bool + frames_do_not_interleave: bool + high_priority_between_video_packets: bool + + +@dataclass(frozen=True) +class QueueStatistics: + mean_packets: float + max_packets: int + mean_bytes: float + max_bytes: int + mean_waiting_frames: float + max_waiting_frames: int + minute_packets: tuple[int, ...] + minute_bytes: tuple[int, ...] + timeline_seconds: tuple[float, ...] + timeline_packets: tuple[int, ...] + timeline_bytes: tuple[int, ...] + remaining_packets: int + remaining_bytes: int + remaining_frames: int + drain_seconds: float + growth_bytes_per_second: float + growth_bytes_per_minute: float + excess_queue_clear_time_seconds: float + + +@dataclass(frozen=True) +class SummaryMetrics: + scenario: str + policy: str + duration_seconds: float + mean_channel_kbps: float + offered_load_kbps: float + speed_reserve_kbps: float + theoretical_queue_growth_bytes_per_second: float + theoretical_minimum_skip_fraction: float + theoretical_sustainable_fps: float + actual_queue_growth_bytes_per_second: float + actual_queue_growth_bytes_per_minute: float + unbounded_queue_growth: bool + mean_queue_packets: float + max_queue_packets: int + mean_queue_bytes: float + max_queue_bytes: int + mean_waiting_video_frames: float + max_waiting_video_frames: int + queue_packets_each_minute: str + queue_bytes_each_minute: str + queue_at_600s_packets: int + queue_at_600s_bytes: int + frames_remaining_at_600s: int + additional_drain_seconds: float + transmitted_packets: int + transmitted_bytes: int + replaced_state_packets: int + + +@dataclass(frozen=True) +class VideoMetrics: + scenario: str + policy: str + created_frames: int + started_frames: int + published_frames: int + published_by_600s_frames: int + dropped_before_start_frames: int + published_fraction: float + actual_update_fps: float + update_fps_each_minute: str + mean_publication_delay_ms: float + p95_publication_delay_ms: float + max_publication_delay_ms: float + mean_display_age_ms: float + p95_display_age_ms: float + max_display_age_ms: float + display_age_over_500ms_fraction: float + display_age_over_1000ms_fraction: float + mean_no_update_duration_ms: float + p95_no_update_duration_ms: float + max_no_update_duration_ms: float + max_consecutive_missing_frames: int + mean_publication_gap_ms: float + max_publication_gap_ms: float + transmitted_video_bytes: int + wasted_transmitted_video_bytes: int + delivered_useful_video_kbps: float + + +@dataclass(frozen=True) +class ControlMetrics: + scenario: str + policy: str + control_mean_delay_ms: float + control_p95_delay_ms: float + control_max_delay_ms: float + control_over_100ms: int + control_max_receive_gap_ms: float + emergency_delays_ms: str + emergency_max_delay_ms: float + all_emergency_under_50ms: bool + telemetry_mean_delay_ms: float + telemetry_p95_delay_ms: float + telemetry_max_delay_ms: float + telemetry_over_500ms: int + + +@dataclass(frozen=True) +class RecoveryMetrics: + scenario: str + policy: str + applicable: bool + image_age_at_300s_ms: float + time_to_age_below_1000ms_seconds: float + time_to_age_below_500ms_seconds: float + excess_queue_clear_seconds: float + frames_dropped_after_recovery: int + old_frames_published_after_recovery: int + + +@dataclass(frozen=True) +class ScenarioResult: + summary: SummaryMetrics + video: VideoMetrics + control: ControlMetrics + recovery: RecoveryMetrics + minute_update_fps: tuple[float, ...] + age_timeline_seconds: tuple[float, ...] + age_timeline_ms: tuple[float, ...] + queue_timeline_seconds: tuple[float, ...] + queue_timeline_bytes: tuple[int, ...] + digest: str + accounting_ok: bool + started_frames_complete: bool + no_interleave: bool + priority_between_video: bool + rate_rule_ok: bool + bit_accounting_error: float + + +@dataclass(frozen=True) +class FunctionalTestResult: + name: str + passed: bool + detail: str + + +def percentile(values: Iterable[float], q: float) -> float: + values = tuple(values) + return float(np.percentile(values, q)) if values else 0.0 + + +def _link_packet( + traffic_class: TrafficClass, + direction: Direction, + stream_id: int, + sequence: int, + generation_time_us: int, + deadline_ms: int, + payload: bytes, +) -> LinkPacket: + return LinkPacket( + traffic_class=traffic_class, + direction=direction, + stream_id=stream_id, + sequence_number=sequence, + generation_time_us=generation_time_us, + deadline_ms=deadline_ms, + payload=payload, + ) + + +def _validated_link(packet: LinkPacket) -> None: + decoded = decode_link_packet(encode_link_packet(packet)) + assert decoded.traffic_class is packet.traffic_class + assert decoded.sequence_number == packet.sequence_number + assert decoded.generation_time_us == packet.generation_time_us + assert decoded.payload == packet.payload + + +def build_long_workload() -> LongWorkload: + """Repeat the 63 real JPEG-size pairs while rebuilding every header/CRC.""" + + lab033 = build_lab033_workload() + originals = tuple(lab033.composites) + if len(originals) != 63: + raise AssertionError("Lab036 requires the 63-frame source profile") + frames: list[VideoFrameGroup] = [] + jpeg_sizes: list[int] = [] + video_sequence = 0 + block_id = 0 + crc_packets = 0 + crc_blocks = 0 + repeated_inner_headers: dict[int, bytes] = {} + for frame_id in range(FRAME_COUNT): + original = originals[frame_id % len(originals)] + composite = EncodedComposite( + composite_frame_id=frame_id, + source_frame_index=original.source_frame_index, + base_jpeg=original.base_jpeg, + roi_jpeg=original.roi_jpeg, + ) + base_packets, roi_packets = packets_for_composite(composite, VIDEO_PAYLOAD_BYTES) + inner_packets = tuple(base_packets + roi_packets) + if frame_id in (0, len(originals)): + repeated_inner_headers[frame_id] = inner_packets[0] + generation_us = int(round(frame_id / COMPOSITE_FPS * 1_000_000.0)) + frame_packets: list[AgePolicyPacket] = [] + for start in range(0, len(inner_packets), SOURCE_BLOCK_SIZE): + chunk = inner_packets[start:start + SOURCE_BLOCK_SIZE] + parity_count = ( + NOMINAL_PARITY_COUNT + if len(chunk) == SOURCE_BLOCK_SIZE + else parity_for_partial(len(chunk)) + ) + outer_packets = encode_fec_block(chunk, block_id, parity_count) + decoded_block = decode_fec_block(tuple(outer_packets)) + assert decoded_block.source_packets == chunk + for inner in decoded_block.source_packets: + assert decode_inner_packet(inner).composite_frame_id == frame_id + crc_blocks += 1 + for outer_wire in outer_packets: + decode_outer_symbol(outer_wire) + packet = _link_packet( + TrafficClass.VIDEO, + Direction.ROVER_TO_GROUND, + STREAM_VIDEO, + video_sequence, + generation_us, + 0, + outer_wire, + ) + _validated_link(packet) + frame_packets.append( + AgePolicyPacket(packet, video_sequence, generation_us, frame_id) + ) + video_sequence += 1 + crc_packets += 1 + block_id += 1 + frames.append(VideoFrameGroup(frame_id, generation_us, tuple(frame_packets))) + jpeg_sizes.append(len(original.base_jpeg) + len(original.roi_jpeg)) + + high_packets: list[AgePolicyPacket] = [] + for sequence, generation_us in enumerate( + range(0, int(DURATION_SECONDS * 1_000_000), CONTROL_PERIOD_US) + ): + packet = _link_packet( + TrafficClass.CONTROL, Direction.GROUND_TO_ROVER, STREAM_CONTROL, + sequence, generation_us, 100, + deterministic_payload(b"CONTROL", sequence, 32), + ) + high_packets.append(AgePolicyPacket(packet, -1, generation_us, None)) + for sequence, generation_us in enumerate( + range(0, int(DURATION_SECONDS * 1_000_000), TELEMETRY_PERIOD_US) + ): + packet = _link_packet( + TrafficClass.TELEMETRY, Direction.ROVER_TO_GROUND, STREAM_TELEMETRY, + sequence, generation_us, 500, + deterministic_payload(b"TELEM", sequence, 64), + ) + high_packets.append(AgePolicyPacket(packet, -1, generation_us, None)) + for sequence, generation_us in enumerate(EMERGENCY_TIMES_US): + packet = _link_packet( + TrafficClass.EMERGENCY, Direction.GROUND_TO_ROVER, STREAM_EMERGENCY, + sequence, generation_us, 50, + deterministic_payload(b"E-STOP", sequence, 32), + ) + high_packets.append(AgePolicyPacket(packet, -1, generation_us, None)) + high_packets.sort( + key=lambda item: ( + item.available_time_us, + int(item.packet.traffic_class), + item.packet.stream_id, + item.packet.sequence_number, + ) + ) + high_packets = [ + replace(item, arrival_order=video_sequence + index) + for index, item in enumerate(high_packets) + ] + for item in high_packets: + _validated_link(item.packet) + crc_packets += 1 + return LongWorkload( + frames=tuple(frames), + high_priority=tuple(high_packets), + jpeg_bytes_by_frame=tuple(jpeg_sizes), + original_frame_count=len(originals), + video_wire_bytes=sum(frame.wire_size_bytes for frame in frames), + nonvideo_wire_bytes=sum(item.wire_size_bytes for item in high_packets), + crc_packets_checked=crc_packets, + crc_blocks_checked=crc_blocks, + repeated_headers_are_fresh=( + repeated_inner_headers[0] != repeated_inner_headers[len(originals)] + ), + ) + + +def schedule_long( + workload: LongWorkload, + profile: SpeedProfile, + policy: PolicyDefinition, +) -> LongSchedule: + frame_arrivals = workload.frames + high_arrivals = workload.high_priority + ready_high: list[AgePolicyPacket] = [] + pending_frames: list[VideoFrameGroup] = [] + transmitted: list[ScheduledPacket] = [] + dropped: list[DroppedFrame] = [] + replacements: list[Replacement] = [] + started: list[int] = [] + completed: list[int] = [] + publications: dict[int, float] = {} + blocker_by_order: dict[int, tuple[AgePolicyPacket, float]] = {} + active: VideoFrameGroup | None = None + active_index = 0 + cursor = 0.0 + frame_index = 0 + high_index = 0 + + def enqueue_high(item: AgePolicyPacket) -> None: + if item.packet.traffic_class in (TrafficClass.CONTROL, TrafficClass.TELEMETRY): + retained = [] + for old in ready_high: + if ( + old.packet.traffic_class is item.packet.traffic_class + and old.packet.stream_id == item.packet.stream_id + ): + replacements.append(Replacement(old, item, item.available_time_seconds)) + else: + retained.append(old) + ready_high[:] = retained + ready_high.append(item) + + def admit(now: float, current: AgePolicyPacket | None = None, start: float = 0.0) -> None: + nonlocal frame_index, high_index + while ( + high_index < len(high_arrivals) + and high_arrivals[high_index].available_time_seconds <= now + TIME_EPSILON_SECONDS + ): + item = high_arrivals[high_index] + high_index += 1 + if current is not None and item.available_time_seconds > start + TIME_EPSILON_SECONDS: + blocker_by_order[item.arrival_order] = ( + current, + max(0.0, now - item.available_time_seconds), + ) + enqueue_high(item) + while ( + frame_index < len(frame_arrivals) + and frame_arrivals[frame_index].generation_time_seconds <= now + TIME_EPSILON_SECONDS + ): + frame = frame_arrivals[frame_index] + frame_index += 1 + arrival = frame.generation_time_seconds + if policy.policy is FramePolicy.LATEST_ONLY: + dropped.extend( + DroppedFrame(old, arrival, "replaced_by_newest") + for old in pending_frames + ) + pending_frames[:] = [frame] + elif policy.policy is FramePolicy.TWO_WAITING: + pending_frames.append(frame) + while len(pending_frames) > 2: + dropped.append(DroppedFrame(pending_frames.pop(0), arrival, "waiting_limit")) + else: + pending_frames.append(frame) + + while ( + frame_index < len(frame_arrivals) + or high_index < len(high_arrivals) + or ready_high + or pending_frames + or active is not None + ): + if not ready_high and not pending_frames and active is None: + next_times = [] + if frame_index < len(frame_arrivals): + next_times.append(frame_arrivals[frame_index].generation_time_seconds) + if high_index < len(high_arrivals): + next_times.append(high_arrivals[high_index].available_time_seconds) + cursor = max(cursor, min(next_times)) + admit(cursor) + selected: AgePolicyPacket | None = None + if ready_high: + selected = min( + ready_high, + key=lambda item: (int(item.packet.traffic_class), item.arrival_order), + ) + ready_high.remove(selected) + else: + if active is None and pending_frames: + active = pending_frames.pop(0) + active_index = 0 + started.append(active.composite_frame_id) + if active is not None: + selected = active.packets[active_index] + if selected is None: + continue + start = max(cursor, selected.available_time_seconds) + rate_kbps = profile.rate_kbps(start) + end = start + selected.wire_size_bytes * 8.0 / (rate_kbps * 1000.0) + admit(end, selected, start) + blocker, blocking_delay = blocker_by_order.get(selected.arrival_order, (None, 0.0)) + transmitted.append( + ScheduledPacket(selected, start, end, rate_kbps, blocker, blocking_delay) + ) + cursor = end + if selected.packet.traffic_class is TrafficClass.VIDEO: + assert active is not None + active_index += 1 + if active_index == len(active.packets): + frame_id = active.composite_frame_id + completed.append(frame_id) + publications[frame_id] = end + active = None + active_index = 0 + + digest = hashlib.sha256() + bit_error = 0.0 + rate_rule_ok = True + compressed_video: list[int] = [] + high_since_video = False + priority_between = False + last_video: int | None = None + for sent in transmitted: + digest.update( + f"{sent.item.arrival_order}:{sent.start_seconds:.12f}:{sent.end_seconds:.12f}:{sent.rate_kbps:.3f};".encode() + ) + bit_error += abs( + (sent.end_seconds - sent.start_seconds) * sent.rate_kbps * 1000.0 + - sent.item.wire_size_bytes * 8.0 + ) + rate_rule_ok &= sent.rate_kbps == profile.rate_kbps(sent.start_seconds) + frame_id = sent.item.composite_frame_id + if frame_id is None: + if last_video is not None: + high_since_video = True + else: + if frame_id == last_video and high_since_video: + priority_between = True + if not compressed_video or compressed_video[-1] != frame_id: + compressed_video.append(frame_id) + last_video = frame_id + high_since_video = False + for item in dropped: + digest.update(f"D{item.frame.composite_frame_id}:{item.time_seconds:.12f};".encode()) + return LongSchedule( + profile, + policy, + tuple(transmitted), + tuple(dropped), + tuple(replacements), + tuple(started), + tuple(completed), + publications, + digest.hexdigest(), + bit_error, + rate_rule_ok, + len(compressed_video) == len(set(compressed_video)), + priority_between, + ) + + +def _sample_events( + events: dict[float, tuple[int, int]], + sample_times: tuple[float, ...], +) -> tuple[tuple[int, ...], tuple[int, ...]]: + ordered = sorted(events.items()) + index = 0 + packets = 0 + byte_count = 0 + packet_values = [] + byte_values = [] + for sample in sample_times: + while index < len(ordered) and ordered[index][0] <= sample + TIME_EPSILON_SECONDS: + packets += ordered[index][1][0] + byte_count += ordered[index][1][1] + index += 1 + packet_values.append(packets) + byte_values.append(byte_count) + return tuple(packet_values), tuple(byte_values) + + +def queue_statistics( + workload: LongWorkload, + schedule: LongSchedule, +) -> QueueStatistics: + intervals: list[tuple[float, float, int]] = [] + first_start: dict[int, float] = {} + drop_time = {item.frame.composite_frame_id: item.time_seconds for item in schedule.dropped} + for sent in schedule.transmitted: + intervals.append((sent.item.available_time_seconds, sent.end_seconds, sent.item.wire_size_bytes)) + if sent.item.composite_frame_id is not None: + first_start.setdefault(sent.item.composite_frame_id, sent.start_seconds) + for item in schedule.dropped: + intervals.extend( + (packet.available_time_seconds, item.time_seconds, packet.wire_size_bytes) + for packet in item.frame.packets + ) + intervals.extend( + (item.removed.available_time_seconds, item.time_seconds, item.removed.wire_size_bytes) + for item in schedule.replacements + ) + events: dict[float, tuple[int, int]] = {} + packet_area = 0.0 + byte_area = 0.0 + for start, end, size in intervals: + left = max(0.0, start) + right = min(DURATION_SECONDS, end) + if right <= left + TIME_EPSILON_SECONDS: + continue + packet_area += right - left + byte_area += (right - left) * size + old = events.get(start, (0, 0)) + events[start] = (old[0] + 1, old[1] + size) + old = events.get(end, (0, 0)) + events[end] = (old[0] - 1, old[1] - size) + count = 0 + byte_count = 0 + max_count = 0 + max_bytes = 0 + for time, delta in sorted(events.items()): + if time > DURATION_SECONDS + TIME_EPSILON_SECONDS: + break + count += delta[0] + byte_count += delta[1] + max_count = max(max_count, count) + max_bytes = max(max_bytes, byte_count) + # One largest frame plus coincident control, telemetry, and emergency is + # the normal 300-kbit/s envelope, not residual video backlog. + baseline_limit = max(len(frame.packets) for frame in workload.frames) + 3 + count = 0 + excess_active = False + excess_clear = 300.0 + for time, delta in sorted(events.items()): + count += delta[0] + if time < 300.0 - TIME_EPSILON_SECONDS: + continue + if count > baseline_limit: + excess_active = True + elif excess_active: + excess_clear = time + excess_active = False + minute_times = tuple(float(value) for value in range(60, 601, 60)) + minute_packets, minute_bytes = _sample_events(events, minute_times) + timeline_times = tuple(float(value) for value in range(0, 601, 5)) + timeline_packets, timeline_bytes = _sample_events(events, timeline_times) + fit_times = np.asarray(minute_times[1:], dtype=float) + fit_bytes = np.asarray(minute_bytes[1:], dtype=float) + growth = float(np.polyfit(fit_times, fit_bytes, 1)[0]) if len(fit_times) > 1 else 0.0 + + frame_events: dict[float, int] = {} + frame_area = 0.0 + for frame in workload.frames: + start = frame.generation_time_seconds + end = first_start.get(frame.composite_frame_id, drop_time.get(frame.composite_frame_id, start)) + left = max(0.0, start) + right = min(DURATION_SECONDS, end) + if right <= left + TIME_EPSILON_SECONDS: + continue + frame_area += right - left + frame_events[left] = frame_events.get(left, 0) + 1 + frame_events[right] = frame_events.get(right, 0) - 1 + waiting = 0 + max_waiting = 0 + for _, delta in sorted(frame_events.items()): + waiting += delta + max_waiting = max(max_waiting, waiting) + published_by_end = sum(time <= DURATION_SECONDS + TIME_EPSILON_SECONDS for time in schedule.publication_times.values()) + dropped_by_end = sum(item.time_seconds <= DURATION_SECONDS + TIME_EPSILON_SECONDS for item in schedule.dropped) + remaining_frames = FRAME_COUNT - published_by_end - dropped_by_end + finish = max( + [DURATION_SECONDS] + + [item.end_seconds for item in schedule.transmitted] + + [item.time_seconds for item in schedule.dropped] + + [item.time_seconds for item in schedule.replacements] + ) + return QueueStatistics( + packet_area / DURATION_SECONDS, + max_count, + byte_area / DURATION_SECONDS, + max_bytes, + frame_area / DURATION_SECONDS, + max_waiting, + minute_packets, + minute_bytes, + timeline_times, + timeline_packets, + timeline_bytes, + minute_packets[-1], + minute_bytes[-1], + remaining_frames, + max(0.0, finish - DURATION_SECONDS), + growth, + growth * 60.0, + max(0.0, excess_clear - 300.0), + ) + + +def display_ages( + publications: dict[int, float], + sample_times: tuple[float, ...], +) -> tuple[float, ...]: + events = sorted((time, frame_id) for frame_id, time in publications.items()) + ages = [] + index = 0 + last_frame: int | None = None + for sample in sample_times: + while index < len(events) and events[index][0] <= sample + TIME_EPSILON_SECONDS: + last_frame = events[index][1] + index += 1 + generation = 0.0 if last_frame is None else last_frame / COMPOSITE_FPS + ages.append(max(0.0, sample - generation)) + return tuple(ages) + + +def _missing_run(published: set[int]) -> int: + current = 0 + maximum = 0 + for frame_id in range(FRAME_COUNT): + if frame_id in published: + current = 0 + else: + current += 1 + maximum = max(maximum, current) + return maximum + + +def video_metrics( + workload: LongWorkload, + schedule: LongSchedule, +) -> tuple[VideoMetrics, tuple[float, ...], tuple[float, ...], tuple[float, ...]]: + publications = schedule.publication_times + published = set(publications) + within = sorted((time, frame_id) for frame_id, time in publications.items() if time <= DURATION_SECONDS + TIME_EPSILON_SECONDS) + minute_fps = tuple( + sum(left <= time < right for time, _ in within) / 60.0 + for left, right in zip(range(0, 600, 60), range(60, 601, 60)) + ) + sample_times = tuple(float(value) / 10.0 for value in range(0, 6001)) + ages = display_ages(publications, sample_times) + timeline_times = tuple(float(value) for value in range(0, 601)) + timeline_ages = display_ages(publications, timeline_times) + publication_delays = [ + time - frame_id / COMPOSITE_FPS for frame_id, time in publications.items() + ] + update_times = [0.0] + [time for time, _ in within] + [DURATION_SECONDS] + no_update = [max(0.0, right - left) for left, right in zip(update_times, update_times[1:])] + gaps = [right[0] - left[0] for left, right in zip(within, within[1:])] + transmitted_video_bytes = sum( + item.item.wire_size_bytes + for item in schedule.transmitted + if item.item.packet.traffic_class is TrafficClass.VIDEO + ) + useful_bytes = sum(workload.jpeg_bytes_by_frame[frame_id] for _, frame_id in within) + metrics = VideoMetrics( + schedule.profile.name, + schedule.policy.name, + FRAME_COUNT, + len(schedule.started_frame_ids), + len(published), + len(within), + len(schedule.dropped), + len(published) / FRAME_COUNT, + len(within) / DURATION_SECONDS, + ";".join(f"{value:.6f}" for value in minute_fps), + float(np.mean(publication_delays)) * 1000.0, + percentile(publication_delays, 95) * 1000.0, + max(publication_delays, default=0.0) * 1000.0, + float(np.mean(ages)) * 1000.0, + percentile(ages, 95) * 1000.0, + max(ages, default=0.0) * 1000.0, + sum(value > 0.5 for value in ages) / len(ages), + sum(value > 1.0 for value in ages) / len(ages), + float(np.mean(no_update)) * 1000.0, + percentile(no_update, 95) * 1000.0, + max(no_update, default=0.0) * 1000.0, + _missing_run(published), + float(np.mean(gaps)) * 1000.0 if gaps else 0.0, + max(gaps, default=0.0) * 1000.0, + transmitted_video_bytes, + 0, + useful_bytes * 8.0 / DURATION_SECONDS / 1000.0, + ) + return metrics, minute_fps, timeline_times, tuple(value * 1000.0 for value in timeline_ages) + + +def control_metrics(schedule: LongSchedule) -> ControlMetrics: + by_class = { + traffic: [item for item in schedule.transmitted if item.item.packet.traffic_class is traffic] + for traffic in (TrafficClass.CONTROL, TrafficClass.EMERGENCY, TrafficClass.TELEMETRY) + } + control = by_class[TrafficClass.CONTROL] + control_delays = [item.end_seconds - item.item.frame_generation_seconds for item in control] + control_gaps = [right.end_seconds - left.end_seconds for left, right in zip(control, control[1:])] + telemetry_delays = [ + item.end_seconds - item.item.frame_generation_seconds + for item in by_class[TrafficClass.TELEMETRY] + ] + emergency = by_class[TrafficClass.EMERGENCY] + if len(emergency) != len(EMERGENCY_TIMES_US): + raise AssertionError("all four emergency commands must be transmitted") + emergency_delays = [item.end_seconds - item.item.frame_generation_seconds for item in emergency] + return ControlMetrics( + schedule.profile.name, + schedule.policy.name, + float(np.mean(control_delays)) * 1000.0, + percentile(control_delays, 95) * 1000.0, + max(control_delays) * 1000.0, + sum(value > 0.1 + TIME_EPSILON_SECONDS for value in control_delays), + max(control_gaps, default=0.0) * 1000.0, + ";".join(f"{value * 1000.0:.6f}" for value in emergency_delays), + max(emergency_delays) * 1000.0, + all(value <= 0.05 + TIME_EPSILON_SECONDS for value in emergency_delays), + float(np.mean(telemetry_delays)) * 1000.0, + percentile(telemetry_delays, 95) * 1000.0, + max(telemetry_delays) * 1000.0, + sum(value > 0.5 + TIME_EPSILON_SECONDS for value in telemetry_delays), + ) + + +def _age_at(publications: dict[int, float], time_seconds: float) -> float: + available = [(time, frame_id) for frame_id, time in publications.items() if time <= time_seconds + TIME_EPSILON_SECONDS] + if not available: + return time_seconds + _, frame_id = max(available) + return max(0.0, time_seconds - frame_id / COMPOSITE_FPS) + + +def _time_to_age( + publications: dict[int, float], + start_seconds: float, + threshold_seconds: float, +) -> float: + if _age_at(publications, start_seconds) < threshold_seconds: + return 0.0 + for frame_id, time in sorted(publications.items(), key=lambda item: item[1]): + if time < start_seconds - TIME_EPSILON_SECONDS or time > DURATION_SECONDS + TIME_EPSILON_SECONDS: + continue + if time - frame_id / COMPOSITE_FPS < threshold_seconds: + return time - start_seconds + return -1.0 + + +def recovery_metrics(schedule: LongSchedule, queue: QueueStatistics) -> RecoveryMetrics: + applicable = schedule.profile.kind == "step" + if not applicable: + return RecoveryMetrics(schedule.profile.name, schedule.policy.name, False, 0.0, 0.0, 0.0, 0.0, 0, 0) + dropped_after = sum( + item.frame.generation_time_seconds < 300.0 + and item.time_seconds >= 300.0 - TIME_EPSILON_SECONDS + for item in schedule.dropped + ) + old_published = sum( + frame_id / COMPOSITE_FPS < 300.0 + and time >= 300.0 - TIME_EPSILON_SECONDS + for frame_id, time in schedule.publication_times.items() + ) + return RecoveryMetrics( + schedule.profile.name, + schedule.policy.name, + True, + _age_at(schedule.publication_times, 300.0) * 1000.0, + _time_to_age(schedule.publication_times, 300.0, 1.0), + _time_to_age(schedule.publication_times, 300.0, 0.5), + queue.excess_queue_clear_time_seconds, + dropped_after, + old_published, + ) + + +def analyze_schedule(workload: LongWorkload, schedule: LongSchedule) -> ScenarioResult: + queue = queue_statistics(workload, schedule) + video, minute_fps, age_times, age_values = video_metrics(workload, schedule) + control = control_metrics(schedule) + recovery = recovery_metrics(schedule, queue) + offered_kbps = workload.offered_wire_bytes * 8.0 / DURATION_SECONDS / 1000.0 + video_kbps = workload.video_wire_bytes * 8.0 / DURATION_SECONDS / 1000.0 + nonvideo_kbps = workload.nonvideo_wire_bytes * 8.0 / DURATION_SECONDS / 1000.0 + mean_rate = schedule.profile.mean_rate_kbps + reserve = mean_rate - offered_kbps + remaining_video = max(0.0, mean_rate - nonvideo_kbps) + skip = max(0.0, 1.0 - remaining_video / video_kbps) + unbounded = ( + schedule.policy.policy is FramePolicy.NO_DROP + and reserve < 0.0 + and queue.growth_bytes_per_second > 0.0 + ) + transmitted_bytes = sum(item.item.wire_size_bytes for item in schedule.transmitted) + summary = SummaryMetrics( + schedule.profile.name, + schedule.policy.name, + DURATION_SECONDS, + mean_rate, + offered_kbps, + reserve, + max(0.0, -reserve) * 1000.0 / 8.0, + skip, + COMPOSITE_FPS * min(1.0, remaining_video / video_kbps), + queue.growth_bytes_per_second, + queue.growth_bytes_per_minute, + unbounded, + queue.mean_packets, + queue.max_packets, + queue.mean_bytes, + queue.max_bytes, + queue.mean_waiting_frames, + queue.max_waiting_frames, + ";".join(str(value) for value in queue.minute_packets), + ";".join(str(value) for value in queue.minute_bytes), + queue.remaining_packets, + queue.remaining_bytes, + queue.remaining_frames, + queue.drain_seconds, + len(schedule.transmitted), + transmitted_bytes, + len(schedule.replacements), + ) + input_packets = sum(len(frame.packets) for frame in workload.frames) + len(workload.high_priority) + input_bytes = workload.offered_wire_bytes + dropped_packets = sum(len(item.frame.packets) for item in schedule.dropped) + dropped_bytes = sum(item.frame.wire_size_bytes for item in schedule.dropped) + replaced_packets = len(schedule.replacements) + replaced_bytes = sum(item.removed.wire_size_bytes for item in schedule.replacements) + accounting_ok = ( + input_packets == len(schedule.transmitted) + dropped_packets + replaced_packets + and input_bytes == transmitted_bytes + dropped_bytes + replaced_bytes + and FRAME_COUNT == len(schedule.completed_frame_ids) + len(schedule.dropped) + ) + started_complete = ( + set(schedule.started_frame_ids) == set(schedule.completed_frame_ids) + and not set(schedule.started_frame_ids).intersection( + item.frame.composite_frame_id for item in schedule.dropped + ) + ) + return ScenarioResult( + summary, + video, + control, + recovery, + minute_fps, + age_times, + age_values, + queue.timeline_seconds, + queue.timeline_bytes, + schedule.digest, + accounting_ok, + started_complete, + schedule.frames_do_not_interleave, + schedule.high_priority_between_video_packets, + schedule.rate_rule_ok, + schedule.bit_accounting_error, + ) + + +def run_experiment(workload: LongWorkload) -> tuple[ScenarioResult, ...]: + results = [] + for profile in SPEED_PROFILES: + for policy in POLICIES: + results.append(analyze_schedule(workload, schedule_long(workload, profile, policy))) + print(f" {profile.name} / {policy.name}") + return tuple(results) + + +def run_functional_tests( + workload: LongWorkload, + results: tuple[ScenarioResult, ...], +) -> tuple[FunctionalTestResult, ...]: + lookup = {(item.summary.scenario, item.summary.policy): item for item in results} + checks: list[tuple[str, callable]] = [] + + def check(name): + def register(function): + checks.append((name, function)) + return function + return register + + @check("01_unique_repeated_frames_and_crc") + def _(): + assert len({frame.composite_frame_id for frame in workload.frames}) == FRAME_COUNT + assert workload.repeated_headers_are_fresh + assert workload.crc_packets_checked > 0 and workload.crc_blocks_checked > 0 + + @check("02_300kbps_latest_has_no_unnecessary_drop") + def _(): + assert lookup[("constant_300", "latest_only")].video.dropped_before_start_frames == 0 + + @check("03_260kbps_no_drop_matches_small_deficit") + def _(): + row = lookup[("constant_260", "no_drop")].summary + assert row.theoretical_queue_growth_bytes_per_second > 0.0 + assert abs(row.actual_queue_growth_bytes_per_second - row.theoretical_queue_growth_bytes_per_second) / row.theoretical_queue_growth_bytes_per_second < 0.45 + + @check("04_230kbps_no_drop_grows_faster") + def _(): + slow = lookup[("constant_230", "no_drop")].summary.actual_queue_growth_bytes_per_second + marginal = lookup[("constant_260", "no_drop")].summary.actual_queue_growth_bytes_per_second + assert slow > marginal * 4.0 + + @check("05_latest_waiting_limit") + def _(): + assert all(item.summary.max_waiting_video_frames <= 1 for item in results if item.summary.policy == "latest_only") + + @check("06_two_waiting_limit") + def _(): + assert all(item.summary.max_waiting_video_frames <= 2 for item in results if item.summary.policy == "two_waiting") + + @check("07_started_frame_never_dropped") + def _(): + assert all(item.started_frames_complete for item in results) + + @check("08_active_video_frames_do_not_interleave") + def _(): + assert all(item.no_interleave for item in results) + + @check("09_priority_between_video_packets") + def _(): + assert all(item.priority_between_video for item in results) + + @check("10_emergency_never_deleted") + def _(): + assert all(len(item.control.emergency_delays_ms.split(";")) == 4 for item in results) + + @check("11_speed_change_bit_accounting") + def _(): + assert all(item.rate_rule_ok and item.bit_accounting_error < 1e-3 for item in results) + + @check("12_latest_does_not_replay_full_history") + def _(): + assert lookup[("step_300_230_300", "latest_only")].recovery.old_frames_published_after_recovery <= 2 + + @check("13_proactive_policies_have_no_waste") + def _(): + assert all(item.video.wasted_transmitted_video_bytes == 0 for item in results if item.summary.policy != "no_drop") + + @check("14_incomplete_frame_not_published") + def _(): + assert all(item.video.published_frames == item.video.started_frames for item in results) + + @check("15_all_three_crc_layers_pass") + def _(): + assert workload.crc_packets_checked == sum(len(frame.packets) for frame in workload.frames) + len(workload.high_priority) + + @check("16_frame_accounting") + def _(): + assert all(item.video.created_frames == item.video.published_frames + item.video.dropped_before_start_frames for item in results) + + @check("17_packet_and_byte_accounting") + def _(): + assert all(item.accounting_ok for item in results) + + @check("18_reproducible") + def _(): + repeated = schedule_long( + workload, + PROFILE_BY_NAME["constant_230"], + POLICY_BY_NAME["latest_only"], + ) + assert repeated.digest == lookup[("constant_230", "latest_only")].digest + + output = [] + for name, function in checks: + try: + function() + output.append(FunctionalTestResult(name, True, "PASS")) + except Exception as error: + output.append(FunctionalTestResult(name, False, f"{type(error).__name__}: {error}")) + failed = [item.name for item in output if not item.passed] + if failed: + raise AssertionError("functional checks failed: " + ", ".join(failed)) + return tuple(output) + + +def save_csv(results: tuple[ScenarioResult, ...]) -> None: + OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) + for path, cls, rows in ( + (SUMMARY_CSV_PATH, SummaryMetrics, (item.summary for item in results)), + (VIDEO_CSV_PATH, VideoMetrics, (item.video for item in results)), + (CONTROL_CSV_PATH, ControlMetrics, (item.control for item in results)), + (RECOVERY_CSV_PATH, RecoveryMetrics, (item.recovery for item in results)), + ): + with path.open("w", encoding="utf-8", newline="") as file: + writer = csv.DictWriter(file, fieldnames=list(cls.__dataclass_fields__)) + writer.writeheader() + writer.writerows(asdict(row) for row in rows) + + +def _step_results(results: tuple[ScenarioResult, ...]) -> list[ScenarioResult]: + return [item for item in results if item.summary.scenario == "step_300_230_300"] + + +def save_plots(results: tuple[ScenarioResult, ...]) -> None: + step = _step_results(results) + fig, axis = plt.subplots(figsize=(11, 5.5)) + for item in step: + axis.plot(item.age_timeline_seconds, item.age_timeline_ms, label=POLICY_BY_NAME[item.summary.policy].label) + axis.axvline(120, color="black", ls="--", alpha=.4); axis.axvline(300, color="black", ls="--", alpha=.4) + axis.set(xlabel="Время, с", ylabel="Возраст, мс", title="Возраст изображения: ступенчатый сценарий") + axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(AGE_PLOT_PATH, dpi=150); plt.close(fig) + + fig, axis = plt.subplots(figsize=(11, 5.5)) + for item in step: + axis.plot(item.queue_timeline_seconds, np.asarray(item.queue_timeline_bytes) / 1_000_000.0, label=POLICY_BY_NAME[item.summary.policy].label) + axis.axvline(120, color="black", ls="--", alpha=.4); axis.axvline(300, color="black", ls="--", alpha=.4) + axis.set(xlabel="Время, с", ylabel="Очередь, Мбайт", title="Очередь во времени") + axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(QUEUE_PLOT_PATH, dpi=150); plt.close(fig) + + variable = [item for item in results if item.summary.scenario == "variable_300_260_230"] + fig, axis = plt.subplots(figsize=(10, 5.2)) + minutes = np.arange(1, 11) + for item in variable: + axis.plot(minutes, item.minute_update_fps, marker="o", label=POLICY_BY_NAME[item.summary.policy].label) + axis.set(xlabel="Минута", ylabel="Обновлений/с", title="Частота обновления по минутам") + axis.set_xticks(minutes); axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(MINUTE_PLOT_PATH, dpi=150); plt.close(fig) + + fig, axis = plt.subplots(figsize=(10, 5.2)) + for item in step: + mask = np.asarray(item.age_timeline_seconds) >= 280.0 + axis.plot(np.asarray(item.age_timeline_seconds)[mask], np.asarray(item.age_timeline_ms)[mask], label=POLICY_BY_NAME[item.summary.policy].label) + axis.axhline(1000, color="orange", ls="--"); axis.axhline(500, color="green", ls=":"); axis.axvline(300, color="black", ls="--") + axis.set(xlabel="Время, с", ylabel="Возраст, мс", title="Восстановление после возврата к 300 кбит/с") + axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(RECOVERY_PLOT_PATH, dpi=150); plt.close(fig) + + labels = [f"{PROFILE_BY_NAME[item.summary.scenario].label}\n{POLICY_BY_NAME[item.summary.policy].label}" for item in results] + x = np.arange(len(results)) + fig, axis = plt.subplots(figsize=(16, 6)) + axis.bar(x, [item.control.control_p95_delay_ms for item in results]) + axis.set_xticks(x, labels, rotation=55, ha="right", fontsize=8) + axis.set(ylabel="P95, мс", title="Задержка обычных команд"); axis.grid(axis="y", alpha=.3) + fig.tight_layout(); fig.savefig(CONTROL_PLOT_PATH, dpi=150); plt.close(fig) + + fig, axis = plt.subplots(figsize=(16, 6)) + width = .38 + axis.bar(x - width / 2, [item.video.published_frames for item in results], width, label="Опубликовано") + axis.bar(x + width / 2, [item.video.dropped_before_start_frames for item in results], width, label="Удалено") + axis.set_xticks(x, labels, rotation=55, ha="right", fontsize=8) + axis.set(ylabel="Кадров", title="Опубликованные и удалённые кадры"); axis.legend(); axis.grid(axis="y", alpha=.3) + fig.tight_layout(); fig.savefig(OUTCOME_PLOT_PATH, dpi=150); plt.close(fig) + + constants = [item for item in results if item.summary.scenario.startswith("constant_") and item.summary.policy == "no_drop"] + rates = [item.summary.mean_channel_kbps for item in constants] + fig, axis = plt.subplots(figsize=(8.5, 5.2)) + axis.plot(rates, [item.summary.theoretical_queue_growth_bytes_per_second for item in constants], marker="o", label="Теория") + axis.plot(rates, [max(0.0, item.summary.actual_queue_growth_bytes_per_second) for item in constants], marker="s", label="Модель") + axis.set(xlabel="Скорость, кбит/с", ylabel="Рост, байт/с", title="Теоретический и фактический рост очереди") + axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(GROWTH_PLOT_PATH, dpi=150); plt.close(fig) + + fig, axis = plt.subplots(figsize=(9, 6)) + for policy in POLICIES: + rows = [item for item in results if item.summary.policy == policy.name] + axis.scatter([item.video.actual_update_fps for item in rows], [item.video.p95_display_age_ms for item in rows], s=70, label=policy.label) + axis.set(xlabel="Обновлений/с", ylabel="P95 возраста, мс", title="Сравнение политик") + axis.grid(alpha=.3); axis.legend(); fig.tight_layout(); fig.savefig(COMPARISON_PLOT_PATH, dpi=150); plt.close(fig) + + +def write_report( + workload: LongWorkload, + results: tuple[ScenarioResult, ...], + tests: tuple[FunctionalTestResult, ...], +) -> None: + git_status = subprocess.run( + ("git", "status", "--short", "--branch"), + check=True, + capture_output=True, + text=True, + encoding="utf-8", + ).stdout.rstrip() + lines = [ + "Lab036. Долговременная устойчивость видеопланировщика при изменении пропускной способности", + "", + "1. Исходное состояние и конфигурация", + f"- Commit Lab035: {LAB035_COMMIT}.", + "- Рабочее дерево перед Lab036 было чистым; ветка main.", + "- Модель использует 600 с, 1800 кадров, повтор реальной последовательности размеров 63 кадров, BASE 240×135 Q23, ROI 320×180 Q33 и 3 составных кадра/с.", + "- Lab036 использует ровно 3 кадра/с в течение 600 с; короткий ролик Lab034 и предшествующих лабораторных давал немного отличающуюся фактическую частоту формирования кадров.", + "- Длинный опыт уменьшает влияние начального и конечного участков, поэтому значения общей предложенной нагрузки Lab034 и Lab036 не обязаны побайтно совпадать.", + "- Каждый повтор получает новые composite_frame_id, времена, заголовки и CRC; JPEG и пакеты на диск не записываются.", + "- Используются внутренний пакет Lab028 с payload 512 байт, FEC 12+3 с выравниванием по кадрам и общий пакет Lab033.", + "- Команды 20 Гц, телеметрия 10 Гц; аварийные команды создаются в 60, 180, 310 и 480 с.", + "- Ошибки радиоканала отсутствуют.", + "- Смена скорости не прерывает начатый пакет: новая скорость применяется только к следующему пакету после завершения текущего.", + "", + "2. Пятнадцать сочетаний", + "scenario | policy | published/drop | fps | age P95 ms | queue max packets | queue at 600 packets | drain s | control P95 ms | emergency max ms", + ] + for item in results: + s, v, c = item.summary, item.video, item.control + lines.append( + f"{s.scenario} | {s.policy} | {v.published_frames}/{v.dropped_before_start_frames} | {v.actual_update_fps:.3f} | {v.p95_display_age_ms:.3f} | {s.max_queue_packets} | {s.queue_at_600s_packets} | {s.additional_drain_seconds:.3f} | {c.control_p95_delay_ms:.3f} | {c.emergency_max_delay_ms:.3f}" + ) + lines.extend(["", "3. Постоянные скорости: теория и модель", "rate | policy | reserve kbps | theory growth B/s | model growth B/s | minimum skip | stable fps | minute queue packets"]) + for item in results: + if not item.summary.scenario.startswith("constant_"): + continue + s = item.summary + lines.append( + f"{s.mean_channel_kbps:.0f} | {s.policy} | {s.speed_reserve_kbps:.3f} | {s.theoretical_queue_growth_bytes_per_second:.3f} | {s.actual_queue_growth_bytes_per_second:.3f} | {s.theoretical_minimum_skip_fraction:.6f} | {s.theoretical_sustainable_fps:.3f} | {s.queue_packets_each_minute}" + ) + lines.extend(["", "4. Восстановление после ухудшения", "policy | age at 300 ms | below 1000 s | below 500 s | excess queue clear s | dropped old | published old"]) + for item in _step_results(results): + r = item.recovery + lines.append( + f"{r.policy} | {r.image_age_at_300s_ms:.3f} | {r.time_to_age_below_1000ms_seconds:.3f} | {r.time_to_age_below_500ms_seconds:.3f} | {r.excess_queue_clear_seconds:.3f} | {r.frames_dropped_after_recovery} | {r.old_frames_published_after_recovery}" + ) + lines.extend(["", "5. Видео, очередь, команды и телеметрия"]) + latest_rows = [item for item in results if item.summary.policy == "latest_only"] + lines.extend( + f"- {item.summary.scenario}: fps={item.video.actual_update_fps:.3f}, age P95={item.video.p95_display_age_ms:.3f} мс, queue max={item.summary.max_queue_packets}, waiting max={item.summary.max_waiting_video_frames}, control P95/max={item.control.control_p95_delay_ms:.3f}/{item.control.control_max_delay_ms:.3f} мс, emergency={item.control.emergency_delays_ms} мс, telemetry P95={item.control.telemetry_p95_delay_ms:.3f} мс." + for item in latest_rows + ) + lines.extend([ + "- Политика самого свежего кадра ограничивает ожидающую видеоочередь одним кадром и не передаёт накопленную историю после восстановления.", + "- Политика двух ожидающих кадров ограничивает очередь двумя кадрами, но увеличивает возраст изображения относительно основной политики.", + "- Без удаления при дефиците скорости очередь растёт; при 300 кбит/с запас достаточен.", + "- У упреждающих политик нет бесполезно переданных видеобайтов: начатый кадр всегда завершается.", + "", + "6. Функциональные проверки", + ]) + lines.extend(f"- {'PASS' if item.passed else 'FAIL'} {item.name}: {item.detail}" for item in tests) + lines.extend([ + f"- CRC-проверено общих пакетов: {workload.crc_packets_checked}; FEC-блоков: {workload.crc_blocks_checked}.", + "", + "7. Созданные файлы", + "- tests/lab036_long_duration_scheduler.py", + "- data/processed/lab036/lab036_summary.csv", + "- data/processed/lab036/lab036_video_metrics.csv", + "- data/processed/lab036/lab036_control_metrics.csv", + "- data/processed/lab036/lab036_recovery_metrics.csv", + "- data/processed/lab036/lab036_report.txt", + ]) + lines.extend(f"- {path.as_posix()}" for path in PLOT_PATHS) + lines.extend(["", "8. Итоговый Git status", "- Lab036 не добавлена в индекс и не закоммичена.", "", git_status]) + REPORT_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def validate_outputs() -> None: + for path in (SUMMARY_CSV_PATH, VIDEO_CSV_PATH, CONTROL_CSV_PATH, RECOVERY_CSV_PATH): + with path.open(encoding="utf-8", newline="") as file: + rows = list(csv.DictReader(file)) + if len(rows) != 15: + raise AssertionError(f"{path} must contain 15 result rows") + if "Lab036" not in REPORT_PATH.read_text(encoding="utf-8"): + raise AssertionError("invalid Lab036 report") + for path in PLOT_PATHS: + image = cv2.imread(str(path), cv2.IMREAD_UNCHANGED) + if image is None or image.size == 0: + raise AssertionError(f"OpenCV could not read {path}") + + +def main() -> None: + print("Lab036: building 600-second workload") + workload = build_long_workload() + print("Lab036: running 15 combinations") + results = run_experiment(workload) + tests = run_functional_tests(workload, results) + save_csv(results) + save_plots(results) + write_report(workload, results, tests) + validate_outputs() + print(f"Lab036 complete: {len(results)} combinations, {len(tests)} checks") + + +if __name__ == "__main__": + main()