diff --git a/PROJECT_LOG.md b/PROJECT_LOG.md index a531046..1244515 100644 --- a/PROJECT_LOG.md +++ b/PROJECT_LOG.md @@ -188,3 +188,28 @@ git version 2.51.1.windows.1 - Итоговое восстановление всех кадров после освобождения очереди не означает работу видео в реальном времени. - Расчёты и графики выполнены с Matplotlib 3.11.1. - Следующий этап посвящён отбрасыванию устаревших видеокадров. + +--- + +# Запись 009 + +## Дата + +3 августа 2026 года + +## Тема + +Завершение Lab034: реактивное отбрасывание устаревших видеокадров. + +## Выполнено + +- Сравнены непрерывные FEC-блоки и блоки, выровненные по составным кадрам. +- Выравнивание увеличивает видеопоток на 5,975 кбит/с, или на 2,633%. +- При 300 кбит/с удаление кадров не требуется. +- При 260 кбит/с ограничения 1000 и 1500 мс не срабатывают, а ограничение 500 мс уже чрезмерно. +- При 230 кбит/с реактивное удаление устаревших кадров уменьшает очередь, но резко снижает частоту обновления. +- Ограничение 500 мс оставляет только 3 из 63 кадров. +- Ограничение 1500 мс оставляет 27 из 63 кадров. +- Подтверждено, что короткая очередь сама по себе не означает свежее изображение. +- Текущая реактивная политика не принята. +- Следующий этап посвящён упреждающему допуску кадров и обслуживанию видео целыми кадрами. diff --git a/data/processed/lab034/lab034_control_delay.png b/data/processed/lab034/lab034_control_delay.png new file mode 100644 index 0000000..d5aee50 Binary files /dev/null and b/data/processed/lab034/lab034_control_delay.png differ diff --git a/data/processed/lab034/lab034_control_metrics.csv b/data/processed/lab034/lab034_control_metrics.csv new file mode 100644 index 0000000..1be77e2 --- /dev/null +++ b/data/processed/lab034/lab034_control_metrics.csv @@ -0,0 +1,16 @@ +channel_kbps,policy,control_mean_age_ms,control_p95_age_ms,control_max_age_ms,control_deadline_misses,control_max_receive_gap_ms,control_replaced,emergency_total_delay_ms,emergency_deadline_met,emergency_blocker_class,emergency_blocker_size_bytes,emergency_blocker_sequence,emergency_blocking_delay_ms,telemetry_mean_age_ms,telemetry_p95_age_ms,telemetry_max_age_ms,telemetry_deadline_misses,telemetry_replaced +300.0,continuous_no_drop,9.439928685896334,17.46633333333346,17.813000000000745,0,65.7596666666671,0,1.7066666666671892,True,none,0,-1,0.0,12.424120192306697,20.10688333333437,20.37300000000064,0,0 +300.0,aligned_no_drop,10.09563301281934,17.46633333333314,17.78700000000022,0,65.7596666666671,0,1.7066666666671892,True,none,0,-1,0.0,12.858206730768112,19.573666666667933,20.160000000000622,0,0 +300.0,aligned_1500ms,10.09563301281934,17.46633333333314,17.78700000000022,0,65.7596666666671,0,1.7066666666671892,True,none,0,-1,0.0,12.858206730768112,19.573666666667933,20.160000000000622,0,0 +300.0,aligned_1000ms,10.09563301281934,17.46633333333314,17.78700000000022,0,65.7596666666671,0,1.7066666666671892,True,none,0,-1,0.0,12.858206730768112,19.573666666667933,20.160000000000622,0,0 +300.0,aligned_500ms,10.09563301281934,17.46633333333314,17.78700000000022,0,65.7596666666671,0,1.7066666666671892,True,none,0,-1,0.0,12.858206730768112,19.573666666667933,20.160000000000622,0,0 +260.0,continuous_no_drop,10.282265902386632,18.753596153843198,20.66700000000843,0,63.476923076922546,0,1.969230769230279,True,none,0,-1,0.0,13.863505177534366,22.44779230770204,23.620846153853847,0,0 +260.0,aligned_no_drop,10.671617788489147,18.907692307691043,20.492307692309808,0,67.81538461538617,0,1.969230769230279,True,none,0,-1,0.0,13.490528846181197,21.89960384616611,23.323076923056796,0,0 +260.0,aligned_1500ms,10.671617788489147,18.907692307691043,20.492307692309808,0,67.81538461538617,0,1.969230769230279,True,none,0,-1,0.0,13.490528846181197,21.89960384616611,23.323076923056796,0,0 +260.0,aligned_1000ms,10.671617788489147,18.907692307691043,20.492307692309808,0,67.81538461538617,0,1.969230769230279,True,none,0,-1,0.0,13.490528846181197,21.89960384616611,23.323076923056796,0,0 +260.0,aligned_500ms,10.71385151629976,18.907692307691043,20.492307692309808,0,67.81538461538617,0,1.969230769230279,True,none,0,-1,0.0,13.61641642014566,22.039999999996684,23.323076923056796,0,0 +230.0,continuous_no_drop,12.250334448143445,21.84782608690261,23.33913043477409,0,69.00869565217427,0,11.547826086889756,True,video,608,458,9.32173913036749,15.51421404680579,25.111304347813142,26.678260869555714,0,0 +230.0,aligned_no_drop,12.729598662164767,22.499999999942013,24.556521739064863,0,69.00869565217427,0,22.330434782542596,True,video,608,463,20.10434782602033,15.594983277549522,25.377391304327652,27.895652173848262,0,0 +230.0,aligned_1500ms,12.613043478216982,22.460869565154475,24.556521739064863,0,69.00869565217427,0,22.330434782542596,True,video,608,463,20.10434782602033,15.613377926377664,25.426086956510073,27.895652173848262,0,0 +230.0,aligned_1000ms,12.76596989961911,22.578260869525746,23.356521739078318,0,69.00869565217427,0,11.652173912974462,True,video,608,484,9.426086956452195,15.511371237411904,25.37739130434555,26.608695652086567,0,0 +230.0,aligned_500ms,12.377759197289913,22.230434782573028,23.339130434790967,0,69.00869565217427,0,3.6869565216850475,True,video,608,511,1.4608695651627812,15.598996655484031,24.913043478221116,26.678260869573478,0,0 diff --git a/data/processed/lab034/lab034_fec_alignment_overhead.png b/data/processed/lab034/lab034_fec_alignment_overhead.png new file mode 100644 index 0000000..2f2b513 Binary files /dev/null and b/data/processed/lab034/lab034_fec_alignment_overhead.png differ diff --git a/data/processed/lab034/lab034_frame_outcomes.png b/data/processed/lab034/lab034_frame_outcomes.png new file mode 100644 index 0000000..743226d Binary files /dev/null and b/data/processed/lab034/lab034_frame_outcomes.png differ diff --git a/data/processed/lab034/lab034_image_age.png b/data/processed/lab034/lab034_image_age.png new file mode 100644 index 0000000..93b369d Binary files /dev/null and b/data/processed/lab034/lab034_image_age.png differ diff --git a/data/processed/lab034/lab034_policy_comparison.png b/data/processed/lab034/lab034_policy_comparison.png new file mode 100644 index 0000000..daec61e Binary files /dev/null and b/data/processed/lab034/lab034_policy_comparison.png differ diff --git a/data/processed/lab034/lab034_queue_size.png b/data/processed/lab034/lab034_queue_size.png new file mode 100644 index 0000000..2c5c4d1 Binary files /dev/null and b/data/processed/lab034/lab034_queue_size.png differ diff --git a/data/processed/lab034/lab034_report.txt b/data/processed/lab034/lab034_report.txt new file mode 100644 index 0000000..67180c2 --- /dev/null +++ b/data/processed/lab034/lab034_report.txt @@ -0,0 +1,98 @@ +Lab034. Ограничение возраста видеоданных и отбрасывание устаревших кадров + +1. Исходное состояние +- Commit Lab033: 21751f41dd0af022fc1f935c67e795e3ae4257c5. +- Перед Lab034 рабочее дерево было чистым; main опережала origin/main на один commit. +- Lab028-Lab033 и их результаты не изменяются. + +2. Формирование FEC-блоков +layout | source | parity | blocks | partial | cross-frame | frames/block mean/max | before/after Lab033 kbps +Непрерывные | 862 | 216 | 72 | 1 | 58 | 1.806/2 | 226.951/240.240 +По кадрам | 862 | 251 | 125 | 62 | 0 | 1.000/1 | 232.926/246.647 +- Дополнительная избыточность выравнивания: 5.975 кбит/с (2.633%). +- Средняя эквивалентная группа кадра: непрерывные 17.111 пакета / 263.968 мс; выровненные 17.667 пакета / 271.007 мс при 300 кбит/с. + +3. Пятнадцать сочетаний +speed | policy | offered/capacity | published/drop/partial | update fps | age P95 ms | control P95 ms | emergency ms | queue max/end | drain s | wasted bytes +300 | Непрерывные, без отбрасывания | 0.8607 | 63/0/0 | 2.986 | 593.333 | 17.466 | 1.707 | 22/15 | 0.231867 | 0 +300 | По кадрам, без отбрасывания | 0.8821 | 63/0/0 | 2.986 | 590.000 | 17.466 | 1.707 | 21/13 | 0.199440 | 0 +300 | По кадрам, 1500 мс | 0.8821 | 63/0/0 | 2.986 | 590.000 | 17.466 | 1.707 | 21/13 | 0.199440 | 0 +300 | По кадрам, 1000 мс | 0.8821 | 63/0/0 | 2.986 | 590.000 | 17.466 | 1.707 | 21/13 | 0.199440 | 0 +300 | По кадрам, 500 мс | 0.8821 | 63/0/0 | 2.986 | 590.000 | 17.466 | 1.707 | 21/13 | 0.199440 | 0 +260 | Непрерывные, без отбрасывания | 0.9931 | 63/0/0 | 2.937 | 690.000 | 18.754 | 1.969 | 28/23 | 0.399200 | 0 +260 | По кадрам, без отбрасывания | 1.0178 | 63/0/0 | 2.937 | 773.833 | 18.908 | 1.969 | 37/30 | 0.539293 | 0 +260 | По кадрам, 1500 мс | 1.0178 | 63/0/0 | 2.937 | 773.833 | 18.908 | 1.969 | 37/30 | 0.539293 | 0 +260 | По кадрам, 1000 мс | 1.0178 | 63/0/0 | 2.937 | 773.833 | 18.908 | 1.969 | 37/30 | 0.539293 | 0 +260 | По кадрам, 500 мс | 1.0178 | 56/7/7 | 2.648 | 1053.833 | 18.908 | 1.969 | 31/24 | 0.403939 | 69597 +230 | Непрерывные, без отбрасывания | 1.1227 | 63/0/0 | 2.697 | 2443.833 | 21.848 | 11.548 | 132/127 | 2.547420 | 0 +230 | По кадрам, без отбрасывания | 1.1505 | 63/0/0 | 2.600 | 2913.333 | 22.500 | 22.330 | 160/155 | 3.125855 | 0 +230 | По кадрам, 1500 мс | 1.1505 | 27/36/36 | 1.300 | 11064.833 | 22.461 | 22.330 | 87/81 | 1.418064 | 321929 +230 | По кадрам, 1000 мс | 1.1505 | 14/49/49 | 0.674 | 11731.500 | 22.578 | 11.652 | 63/53 | 0.904881 | 434885 +230 | По кадрам, 500 мс | 1.1505 | 3/60/60 | 0.144 | 11731.500 | 22.230 | 3.687 | 32/27 | 0.419072 | 532167 + +4. Практический смысл ограничений возраста +- При 300 кбит/с очередь обслуживается достаточно быстро: ограничения 500, 1000 и 1500 мс не срабатывают и не дают выигрыша. +- При 260 кбит/с пределы 1000 и 1500 мс также не срабатывают; предел 500 мс публикует 56 из 63 кадров и отбрасывает 7, поэтому он уже чрезмерно жёсткий для части кадров. +- При 230 кбит/с предел 1500 мс сохраняет 27 кадров, 1000 мс — 14, 500 мс — только 3; более строгий предел лучше ограничивает очередь, но резко уменьшает частоту обновления. +- В исследованной перегрузке 1500 мс является наиболее мягким из конечных ограничений; 1000 мс сильнее улучшает свежесть ценой обновлений, а 500 мс чрезмерно жёсток. + +5. Интерпретация +- Произвольное удаление символа из непрерывного FEC-блока опасно: блок может содержать соседние кадры, и нехватка символов затрагивает их совместное декодирование. +- Выравнивание изолирует кадр в собственных блоках, поэтому все ожидающие символы устаревшего кадра можно удалить без повреждения соседей. +- Цена изоляции — 5.975 кбит/с дополнительного внешнего потока из-за большего числа неполных блоков. +- Полная доставка после освобождения очереди не является целью операторского видео: поздний полный кадр уже не описывает текущее состояние ровера. +- При перегрузке отбрасывание старого кадра освобождает ресурс для более свежего изображения и не затрагивает команды. +- При 230 кбит/с выравнивание меняет фазу активного видеопакета около 10,0 с; фактическая задержка аварийной команды может отличаться от Lab033, но отбрасывание её не увеличивает, а непереключаемая верхняя граница остаётся прежней, поскольку максимальный размер пакета не изменён. +- Слишком малый возраст может отменять кадры быстрее их полной передачи и снижать фактическую частоту обновления. +- Поэтому возраст изображения оценивается вместе с частотой обновления: хорошее значение одного показателя не гарантирует полезный видеопоток. +- Замена команд допустима только для текущего состояния (скорость, поворот, торможение, телеметрия). Дискретные события заменять нельзя; в Lab034 они не моделируются. + +6. Допущения +- Один общий абстрактный ресурс, одна скорость для обоих направлений, без времени переключения и без выбора физического duplex/TDD/FDD. +- Ошибки и помехи отсутствуют; начатый пакет не прерывается; форматы Lab028, Lab030 и Lab033 неизменны. +- Возраст считается от формирования составного кадра по generation_time_us; служебные frame_id и состояние отмены не сериализуются. +- Видеокадр с удалёнными пакетами очищается в приёмнике и никогда не публикуется. +- Допустимые возраста 500, 1000 и 1500 мс заданы явно и автоматически не выбираются. + +7. Функциональные проверки +- PASS 01_continuous_matches_lab033: PASS +- PASS 02_aligned_blocks_one_frame: PASS +- PASS 03_aligned_no_drop_recovers_all: PASS +- PASS 04_whole_frame_drop_isolated: PASS +- PASS 05_only_waiting_video_removed: PASS +- PASS 06_active_packet_not_preempted: PASS +- PASS 07_partial_cancel_not_published: PASS +- PASS 08_age_uses_frame_generation: PASS +- PASS 09_receiver_cleans_stale_state: PASS +- PASS 10_emergency_never_deleted: PASS +- PASS 11_high_priority_before_video: PASS +- PASS 12_230kbps_finite_age_bounds_queue: PASS +- PASS 13_packet_accounting: PASS +- PASS 14_byte_accounting: PASS +- PASS 15_all_crc_layers_pass: PASS +- PASS 16_incomplete_frame_never_published: PASS +- PASS 17_reproducible: PASS +- PASS 18_drop_does_not_worsen_emergency: PASS + +8. Созданные файлы +- protocol/video_age_policy.py +- tests/lab034_stale_video_drop.py +- data/processed/lab034/lab034_summary.csv +- data/processed/lab034/lab034_video_metrics.csv +- data/processed/lab034/lab034_control_metrics.csv +- data/processed/lab034/lab034_report.txt +- data/processed/lab034/lab034_image_age.png +- data/processed/lab034/lab034_frame_outcomes.png +- data/processed/lab034/lab034_queue_size.png +- data/processed/lab034/lab034_control_delay.png +- data/processed/lab034/lab034_update_rate.png +- data/processed/lab034/lab034_fec_alignment_overhead.png +- data/processed/lab034/lab034_policy_comparison.png + +9. Итоговый Git status +- Lab034 не добавлена в индекс и не закоммичена. + +## main...origin/main [ahead 1] +?? data/processed/lab034/ +?? protocol/video_age_policy.py +?? tests/lab034_stale_video_drop.py diff --git a/data/processed/lab034/lab034_summary.csv b/data/processed/lab034/lab034_summary.csv new file mode 100644 index 0000000..5ce63fd --- /dev/null +++ b/data/processed/lab034/lab034_summary.csv @@ -0,0 +1,16 @@ +channel_kbps,policy,layout,max_video_age_ms,offered_load_kbps,offered_to_capacity_ratio,transmitted_packets,transmitted_bytes,dropped_packets,dropped_bytes,wasted_transmitted_bytes,mean_queue_packets,max_queue_packets,mean_queue_bytes,max_queue_bytes,queue_at_source_end_packets,additional_drain_seconds,channel_utilization_fraction +300.0,continuous_no_drop,continuous,-1,258.21380417335473,0.8607126805778491,1703,670280,0,0,0,8.05016637239154,22,4452.387224542477,12221,15,0.23186699999999405,0.8512086423304386 +300.0,aligned_no_drop,aligned,-1,264.62022471910115,0.8820674157303372,1738,686910,0,0,0,8.518393226324118,21,4685.69717834665,11415,13,0.1994403333333281,0.8724635473515175 +300.0,aligned_1500ms,aligned,1500,264.62022471910115,0.8820674157303372,1738,686910,0,0,0,8.518393226324118,21,4685.69717834665,11415,13,0.1994403333333281,0.8724635473515175 +300.0,aligned_1000ms,aligned,1000,264.62022471910115,0.8820674157303372,1738,686910,0,0,0,8.518393226324118,21,4685.69717834665,11415,13,0.1994403333333281,0.8724635473515175 +300.0,aligned_500ms,aligned,500,264.62022471910115,0.8820674157303372,1738,686910,0,0,0,8.518393226324118,21,4685.69717834665,11415,13,0.1994403333333281,0.8724635473515175 +260.0,continuous_no_drop,continuous,-1,258.21380417335473,0.9931300160513643,1703,670280,0,0,0,10.783079034450305,28,6010.224630522928,16384,23,0.39920033333350347,0.9743990170589202 +260.0,aligned_no_drop,aligned,-1,264.62022471910115,1.017770095073466,1738,686910,0,0,0,13.695290291396285,37,7655.6639836816,21131,30,0.5392926410258561,0.991800947030499 +260.0,aligned_1500ms,aligned,1500,264.62022471910115,1.017770095073466,1738,686910,0,0,0,13.695290291396285,37,7655.6639836816,21131,30,0.5392926410258561,0.991800947030499 +260.0,aligned_1000ms,aligned,1000,264.62022471910115,1.017770095073466,1738,686910,0,0,0,13.695290291396285,37,7655.6639836816,21131,30,0.5392926410258561,0.991800947030499 +260.0,aligned_500ms,aligned,500,264.62022471910115,1.017770095073466,1730,682511,8,4399,69597,13.261018996174682,31,7402.747049027978,16992,24,0.4039387948720048,0.991800947030499 +230.0,continuous_no_drop,continuous,-1,258.21380417335473,1.1226687137971945,1703,670280,0,0,0,59.749842696627866,132,34307.118394888734,75969,127,2.547420289855012,1.0 +230.0,aligned_no_drop,aligned,-1,264.62022471910115,1.1505227161700051,1738,686910,0,0,0,76.17493190801534,160,43586.68565075366,92568,155,3.1258550724636294,1.0 +230.0,aligned_1500ms,aligned,1500,264.62022471910115,1.1505227161700051,1641,637811,97,49099,321929,59.44779896084508,87,33911.67327808277,50035,81,1.4180637681158217,1.0 +230.0,aligned_1000ms,aligned,1000,264.62022471910115,1.1505227161700051,1609,623057,129,63853,434885,42.60156656500422,63,24223.970784957004,35495,53,0.9048811594201744,1.0 +230.0,aligned_500ms,aligned,500,264.62022471910115,1.1505227161700051,1580,609090,158,77820,532167,20.03526158908224,32,11270.42233914812,18208,27,0.41907246376802476,0.9999999999999993 diff --git a/data/processed/lab034/lab034_update_rate.png b/data/processed/lab034/lab034_update_rate.png new file mode 100644 index 0000000..23eebf2 Binary files /dev/null and b/data/processed/lab034/lab034_update_rate.png differ diff --git a/data/processed/lab034/lab034_video_metrics.csv b/data/processed/lab034/lab034_video_metrics.csv new file mode 100644 index 0000000..d07eb79 --- /dev/null +++ b/data/processed/lab034/lab034_video_metrics.csv @@ -0,0 +1,16 @@ +channel_kbps,policy,created_frames,published_frames,intentionally_dropped_frames,partially_transmitted_cancelled_frames,published_after_deadline_frames,published_fraction,actual_update_fps,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,mean_missing_run_frames,p95_missing_run_frames,max_missing_run_frames,delivered_useful_video_kbps,wasted_transmitted_video_bytes,receiver_cleared_frames +300.0,continuous_no_drop,63,63,0,0,0,1.0,2.985553772070626,276.3369365079338,322.0720000000001,330.77299999999,437.2938723131218,593.3333333333333,660.0000000000001,0.3142444658325313,0.0,329.6296296296296,388.9899000000025,397.1729999999969,0.0,0.0,0,157.76975922953451,0,0 +300.0,aligned_no_drop,63,63,0,0,0,1.0,2.985553772070626,272.2167195767169,293.13066666666225,301.84000000000003,433.2836060314408,589.9999999999999,623.3333333333348,0.30221366698748797,0.0,329.6296296296296,355.05666666666684,406.506333333331,0.0,0.0,0,157.76975922953451,0,0 +300.0,aligned_1500ms,63,63,0,0,0,1.0,2.985553772070626,272.2167195767169,293.13066666666225,301.84000000000003,433.2836060314408,589.9999999999999,623.3333333333348,0.30221366698748797,0.0,329.6296296296296,355.05666666666684,406.506333333331,0.0,0.0,0,157.76975922953451,0,0 +300.0,aligned_1000ms,63,63,0,0,0,1.0,2.985553772070626,272.2167195767169,293.13066666666225,301.84000000000003,433.2836060314408,589.9999999999999,623.3333333333348,0.30221366698748797,0.0,329.6296296296296,355.05666666666684,406.506333333331,0.0,0.0,0,157.76975922953451,0,0 +300.0,aligned_500ms,63,63,0,0,0,1.0,2.985553772070626,272.2167195767169,293.13066666666225,301.84000000000003,433.2836060314408,589.9999999999999,623.3333333333348,0.30221366698748797,0.0,329.6296296296296,355.05666666666684,406.506333333331,0.0,0.0,0,157.76975922953451,0,0 +260.0,continuous_no_drop,63,63,0,0,0,1.0,2.937399678972713,351.2176776557015,445.84648717956554,459.55930769245157,509.63907603464884,690.0000000000005,783.3333333333314,0.5312800769971127,0.0,334.9462365591397,401.4781615384583,411.91761538461515,0.0,0.0,0,157.76975922953451,0,0 +260.0,aligned_no_drop,63,63,0,0,0,1.0,2.937399678972713,393.259570207603,579.4598205129968,620.584948718161,546.5335258261151,773.833333333333,943.3333333333316,0.62223291626564,0.0,334.9462365591397,363.58000000001095,417.7637692307723,0.0,0.0,0,157.76975922953451,0,0 +260.0,aligned_1500ms,63,63,0,0,0,1.0,2.937399678972713,393.259570207603,579.4598205129968,620.584948718161,546.5335258261151,773.833333333333,943.3333333333316,0.62223291626564,0.0,334.9462365591397,363.58000000001095,417.7637692307723,0.0,0.0,0,157.76975922953451,0,0 +260.0,aligned_1000ms,63,63,0,0,0,1.0,2.937399678972713,393.259570207603,579.4598205129968,620.584948718161,546.5335258261151,773.833333333333,943.3333333333316,0.62223291626564,0.0,334.9462365591397,363.58000000001095,417.7637692307723,0.0,0.0,0,157.76975922953451,0,0 +260.0,aligned_500ms,63,56,7,7,0,0.8888888888888888,2.648475120385233,367.5917000915918,493.7670000001404,498.6567435899403,584.550850176452,1053.8333333333312,1769.9999999999995,0.6212704523580366,0.060153994225216556,370.8333333333333,664.0000000000219,1270.1227435895887,1.4,2.5999999999999996,3,139.5205136436597,69597,7 +230.0,continuous_no_drop,63,63,0,0,0,1.0,2.696629213483146,1445.9947550034299,2505.9953623187953,2583.976811594141,1464.8845043310876,2443.8333333333317,2740.000000000002,0.9716073147256978,0.78055822906641,364.32748538011697,437.67652173912944,464.34782608695,0.0,0.0,0,157.76975922953451,0,0 +230.0,aligned_no_drop,63,63,0,0,0,1.0,2.6003210272873196,1777.4951000689596,3067.511884057848,3204.707246376671,1723.7872954764196,2913.333333333334,3206.666666666667,0.9735322425409048,0.8267564966313763,377.57575757575756,412.5426086956473,424.00000000000125,0.0,0.0,0,157.76975922953451,0,0 +230.0,aligned_1500ms,63,27,36,36,0,0.42857142857142855,1.3001605136436598,1010.3600644122084,1414.757101449207,1473.0202898550094,4028.4071222329167,11064.833333333332,12103.333333333334,0.9735322425409048,0.8267564966313763,741.6666666666666,416.659130434783,10626.97971014499,36.0,36.0,36,66.39987158908507,321929,36 +230.0,aligned_1000ms,63,14,49,49,0,0.2222222222222222,0.6741573033707865,734.7155279503047,977.28927536231,995.617391304334,4667.483156881616,11731.499999999998,12770.0,0.9735322425409048,0.8233878729547642,1384.4444444444443,6308.219130434754,11828.788405797173,24.5,36.65,38,34.951781701444624,434885,49 +230.0,aligned_500ms,63,3,60,60,0,0.047619047619047616,0.14446227929373998,429.5149758453909,457.5026086956464,460.9971014492756,5508.034969521976,11731.499999999998,12770.0,0.9696823869104908,0.9076034648700674,5191.666666666666,11634.280579710185,12340.614492753682,30.0,37.2,38,7.567126805778492,532167,60 diff --git a/protocol/video_age_policy.py b/protocol/video_age_policy.py new file mode 100644 index 0000000..e302584 --- /dev/null +++ b/protocol/video_age_policy.py @@ -0,0 +1,224 @@ +"""Frame-aware stale-video policy for the Lab034 shared link queue. + +The scheduler remains strict-priority, state-replacing, and non-preemptive. +Unlike Lab033, packet availability is separate from the timestamp stored in +the common header. Every aligned video packet therefore carries the source +composite-frame generation time while FEC construction metadata remains local +to the scheduler. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Iterable + +from protocol.link_packet import HEADER_SIZE, LinkPacket, TrafficClass, encode_link_packet +from protocol.priority_scheduler import transmission_duration_seconds + + +TIME_EPSILON_SECONDS = 1e-12 + + +@dataclass(frozen=True) +class AgePolicyPacket: + """A common packet plus non-serialized availability/frame metadata.""" + + packet: LinkPacket + arrival_order: int + available_time_us: int + composite_frame_id: int | None = None + + @property + def available_time_seconds(self) -> float: + return self.available_time_us / 1_000_000.0 + + @property + def frame_generation_seconds(self) -> float: + return self.packet.generation_time_us / 1_000_000.0 + + @property + def wire_size_bytes(self) -> int: + return HEADER_SIZE + len(self.packet.payload) + + +@dataclass(frozen=True) +class StateReplacement: + removed: AgePolicyPacket + replacement: AgePolicyPacket + time_seconds: float + + +@dataclass(frozen=True) +class DroppedVideoPacket: + item: AgePolicyPacket + drop_time_seconds: float + + +@dataclass(frozen=True) +class AgeScheduledPacket: + item: AgePolicyPacket + start_seconds: float + end_seconds: float + blocked_by: AgePolicyPacket | None + blocking_delay_seconds: float + wire_packet: bytes + + +@dataclass(frozen=True) +class AgeScheduleResult: + channel_bitrate_bps: float + max_video_age_seconds: float | None + transmitted: tuple[AgeScheduledPacket, ...] + dropped_video: tuple[DroppedVideoPacket, ...] + replacements: tuple[StateReplacement, ...] + + +def _enqueue( + ready: list[AgePolicyPacket], + item: AgePolicyPacket, + replacements: list[StateReplacement], +) -> None: + if item.packet.traffic_class in (TrafficClass.CONTROL, TrafficClass.TELEMETRY): + retained = [] + for old in ready: + if ( + old.packet.traffic_class == item.packet.traffic_class + and old.packet.stream_id == item.packet.stream_id + ): + replacements.append( + StateReplacement(old, item, item.available_time_seconds) + ) + else: + retained.append(old) + ready[:] = retained + ready.append(item) + + +def schedule_with_video_age( + packets: Iterable[AgePolicyPacket], + channel_bitrate_bps: float, + max_video_age_seconds: float | None, +) -> AgeScheduleResult: + """Run a strict-priority non-preemptive scheduler with frame drops. + + At every packet-selection instant all queued packets belonging to a video + frame older than ``max_video_age_seconds`` are removed together. An active + packet has already left the queue and is never interrupted. + """ + + if channel_bitrate_bps <= 0.0: + raise ValueError("channel bitrate must be positive") + if max_video_age_seconds is not None and max_video_age_seconds <= 0.0: + raise ValueError("maximum video age must be positive") + arrivals = sorted( + tuple(packets), + key=lambda item: (item.available_time_seconds, item.arrival_order), + ) + if len({item.arrival_order for item in arrivals}) != len(arrivals): + raise ValueError("arrival_order values must be unique") + for item in arrivals: + if item.packet.traffic_class is TrafficClass.VIDEO: + if item.composite_frame_id is None: + raise ValueError("video packet lacks composite_frame_id") + elif item.composite_frame_id is not None: + raise ValueError("non-video packet has composite_frame_id") + + ready: list[AgePolicyPacket] = [] + transmitted: list[AgeScheduledPacket] = [] + dropped: list[DroppedVideoPacket] = [] + replacements: list[StateReplacement] = [] + blocker_by_order: dict[int, tuple[AgePolicyPacket, float]] = {} + dropped_frames: set[int] = set() + arrival_index = 0 + cursor = 0.0 + + def admit_until(limit: float, active: AgePolicyPacket | None = None, active_end: float = 0.0) -> None: + nonlocal arrival_index + while ( + arrival_index < len(arrivals) + and arrivals[arrival_index].available_time_seconds + <= limit + TIME_EPSILON_SECONDS + ): + item = arrivals[arrival_index] + arrival_index += 1 + if ( + active is not None + and item.available_time_seconds > cursor + TIME_EPSILON_SECONDS + ): + blocker_by_order[item.arrival_order] = ( + active, + max(0.0, active_end - item.available_time_seconds), + ) + if item.composite_frame_id in dropped_frames: + dropped.append(DroppedVideoPacket(item, limit)) + else: + _enqueue(ready, item, replacements) + + def expire_video(now: float) -> None: + if max_video_age_seconds is None: + return + expired_frames = { + item.composite_frame_id + for item in ready + if ( + item.packet.traffic_class is TrafficClass.VIDEO + and now - item.frame_generation_seconds + > max_video_age_seconds + TIME_EPSILON_SECONDS + ) + } + if not expired_frames: + return + retained = [] + for item in ready: + if item.composite_frame_id in expired_frames: + dropped.append(DroppedVideoPacket(item, now)) + assert item.composite_frame_id is not None + dropped_frames.add(item.composite_frame_id) + else: + retained.append(item) + ready[:] = retained + + while arrival_index < len(arrivals) or ready: + if not ready: + cursor = max(cursor, arrivals[arrival_index].available_time_seconds) + admit_until(cursor) + expire_video(cursor) + if not ready: + continue + selected_index = min( + range(len(ready)), + key=lambda index: ( + int(ready[index].packet.traffic_class), + ready[index].arrival_order, + ), + ) + selected = ready.pop(selected_index) + start = max(cursor, selected.available_time_seconds) + wire_packet = encode_link_packet(selected.packet) + end = start + transmission_duration_seconds( + len(wire_packet), channel_bitrate_bps + ) + cursor = start + admit_until(end, selected, end) + blocker, blocking_delay = blocker_by_order.get( + selected.arrival_order, (None, 0.0) + ) + transmitted.append( + AgeScheduledPacket( + item=selected, + start_seconds=start, + end_seconds=end, + blocked_by=blocker, + blocking_delay_seconds=blocking_delay, + wire_packet=wire_packet, + ) + ) + cursor = end + + return AgeScheduleResult( + channel_bitrate_bps=channel_bitrate_bps, + max_video_age_seconds=max_video_age_seconds, + transmitted=tuple(transmitted), + dropped_video=tuple(dropped), + replacements=tuple(replacements), + ) diff --git a/tests/lab034_stale_video_drop.py b/tests/lab034_stale_video_drop.py new file mode 100644 index 0000000..f42414e --- /dev/null +++ b/tests/lab034_stale_video_drop.py @@ -0,0 +1,1020 @@ +"""Lab034: frame-aligned FEC and deliberate stale-video dropping.""" + +from __future__ import annotations + +import csv +from dataclasses import asdict, dataclass, replace +from math import ceil +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, + HEADER_SIZE, + LinkPacket, + TrafficClass, + decode_link_packet, +) +from protocol.packet_erasure_fec import ( + decode_fec_block, + decode_outer_symbol, + encode_fec_block, +) +from protocol.video_age_policy import ( + AgePolicyPacket, + AgeScheduleResult, + schedule_with_video_age, +) +from protocol.video_packet import CompositeReassembler, decode_packet as decode_inner_packet +from tests.lab028_video_packetization import COMPOSITE_FPS +from tests.lab029_packet_channel_simulation import prepare_profiles +from tests.lab030_packet_erasure_fec import SourcePacket, prepare_source_packets +from tests.lab033_priority_channel_scheduler import ( + CHANNEL_RATES_KBPS, + STREAM_VIDEO, + VIDEO_PAYLOAD_SIZE, + Workload as Lab033Workload, + build_workload as build_lab033_workload, + simulate as simulate_lab033, +) + + +OUTPUT_DIRECTORY = Path("data/processed/lab034") +SUMMARY_CSV_PATH = OUTPUT_DIRECTORY / "lab034_summary.csv" +VIDEO_CSV_PATH = OUTPUT_DIRECTORY / "lab034_video_metrics.csv" +CONTROL_CSV_PATH = OUTPUT_DIRECTORY / "lab034_control_metrics.csv" +REPORT_PATH = OUTPUT_DIRECTORY / "lab034_report.txt" +IMAGE_AGE_PLOT_PATH = OUTPUT_DIRECTORY / "lab034_image_age.png" +FRAME_OUTCOME_PLOT_PATH = OUTPUT_DIRECTORY / "lab034_frame_outcomes.png" +QUEUE_PLOT_PATH = OUTPUT_DIRECTORY / "lab034_queue_size.png" +CONTROL_DELAY_PLOT_PATH = OUTPUT_DIRECTORY / "lab034_control_delay.png" +UPDATE_RATE_PLOT_PATH = OUTPUT_DIRECTORY / "lab034_update_rate.png" +FEC_OVERHEAD_PLOT_PATH = OUTPUT_DIRECTORY / "lab034_fec_alignment_overhead.png" +COMPARISON_PLOT_PATH = OUTPUT_DIRECTORY / "lab034_policy_comparison.png" +PLOT_PATHS = ( + IMAGE_AGE_PLOT_PATH, + FRAME_OUTCOME_PLOT_PATH, + QUEUE_PLOT_PATH, + CONTROL_DELAY_PLOT_PATH, + UPDATE_RATE_PLOT_PATH, + FEC_OVERHEAD_PLOT_PATH, + COMPARISON_PLOT_PATH, +) +LAB033_COMMIT = "21751f41dd0af022fc1f935c67e795e3ae4257c5" +SOURCE_BLOCK_SIZE = 12 +NOMINAL_PARITY_COUNT = 3 +TIME_EPSILON_SECONDS = 1e-9 + + +@dataclass(frozen=True) +class PolicyDefinition: + name: str + label: str + layout: str + max_age_ms: int | None + + +POLICIES = ( + PolicyDefinition("continuous_no_drop", "Непрерывные, без отбрасывания", "continuous", None), + PolicyDefinition("aligned_no_drop", "По кадрам, без отбрасывания", "aligned", None), + PolicyDefinition("aligned_1500ms", "По кадрам, 1500 мс", "aligned", 1500), + PolicyDefinition("aligned_1000ms", "По кадрам, 1000 мс", "aligned", 1000), + PolicyDefinition("aligned_500ms", "По кадрам, 500 мс", "aligned", 500), +) +POLICY_BY_NAME = {policy.name: policy for policy in POLICIES} + + +@dataclass(frozen=True) +class BlockDescription: + block_id: int + frame_ids: tuple[int, ...] + source_count: int + parity_count: int + symbol_size: int + + +@dataclass(frozen=True) +class FECLayoutMetrics: + layout: str + source_packets: int + parity_packets: int + fec_blocks: int + partial_blocks: int + cross_frame_blocks: int + mean_frames_per_block: float + max_frames_per_block: int + before_link_kbps: float + after_link_kbps: float + alignment_extra_kbps: float + alignment_extra_percent: float + mean_packets_per_frame: float + mean_frame_transmission_ms_at_300kbps: float + + +@dataclass(frozen=True) +class AlignedWorkload: + lab033: Lab033Workload + packets: tuple[AgePolicyPacket, ...] + blocks: tuple[BlockDescription, ...] + layout_metrics: FECLayoutMetrics + + +@dataclass(frozen=True) +class Lab034Summary: + channel_kbps: float + policy: str + layout: str + max_video_age_ms: int + offered_load_kbps: float + offered_to_capacity_ratio: float + transmitted_packets: int + transmitted_bytes: int + dropped_packets: int + dropped_bytes: int + wasted_transmitted_bytes: int + mean_queue_packets: float + max_queue_packets: int + mean_queue_bytes: float + max_queue_bytes: int + queue_at_source_end_packets: int + additional_drain_seconds: float + channel_utilization_fraction: float + + +@dataclass(frozen=True) +class VideoMetrics: + channel_kbps: float + policy: str + created_frames: int + published_frames: int + intentionally_dropped_frames: int + partially_transmitted_cancelled_frames: int + published_after_deadline_frames: int + published_fraction: float + actual_update_fps: float + 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 + mean_missing_run_frames: float + p95_missing_run_frames: float + max_missing_run_frames: int + delivered_useful_video_kbps: float + wasted_transmitted_video_bytes: int + receiver_cleared_frames: int + + +@dataclass(frozen=True) +class ControlMetrics: + channel_kbps: float + policy: str + control_mean_age_ms: float + control_p95_age_ms: float + control_max_age_ms: float + control_deadline_misses: int + control_max_receive_gap_ms: float + control_replaced: int + emergency_total_delay_ms: float + emergency_deadline_met: bool + emergency_blocker_class: str + emergency_blocker_size_bytes: int + emergency_blocker_sequence: int + emergency_blocking_delay_ms: float + telemetry_mean_age_ms: float + telemetry_p95_age_ms: float + telemetry_max_age_ms: float + telemetry_deadline_misses: int + telemetry_replaced: int + + +@dataclass(frozen=True) +class Lab034Result: + summary: Lab034Summary + video: VideoMetrics + control: ControlMetrics + publication_times: dict[int, float] + dropped_frames: frozenset[int] + transmitted_video_by_frame: dict[int, int] + schedule: AgeScheduleResult | None + + +@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 parity_for_partial(source_count: int) -> int: + if not 1 <= source_count <= SOURCE_BLOCK_SIZE: + raise ValueError("partial source count is outside 1...12") + return max(1, ceil(source_count * NOMINAL_PARITY_COUNT / SOURCE_BLOCK_SIZE)) + + +def _aligned_video_packets( + lab033: Lab033Workload, +) -> tuple[tuple[AgePolicyPacket, ...], tuple[BlockDescription, ...]]: + profile = prepare_profiles(list(lab033.composites))[VIDEO_PAYLOAD_SIZE] + source_packets: tuple[SourcePacket, ...] = prepare_source_packets(profile) + by_frame: dict[int, list[SourcePacket]] = {} + for source in source_packets: + by_frame.setdefault(source.composite_frame_id, []).append(source) + packets: list[AgePolicyPacket] = [] + blocks: list[BlockDescription] = [] + sequence = 0 + block_id = 0 + for frame_id in sorted(by_frame): + frame_sources = by_frame[frame_id] + generation_us = int(round(frame_id / COMPOSITE_FPS * 1_000_000.0)) + for start in range(0, len(frame_sources), SOURCE_BLOCK_SIZE): + chunk = frame_sources[start:start + SOURCE_BLOCK_SIZE] + parity_count = ( + NOMINAL_PARITY_COUNT + if len(chunk) == SOURCE_BLOCK_SIZE + else parity_for_partial(len(chunk)) + ) + wire_symbols = encode_fec_block( + tuple(source.inner_packet for source in chunk), + block_id, + parity_count, + ) + decoded = tuple(decode_outer_symbol(symbol) for symbol in wire_symbols) + blocks.append( + BlockDescription( + block_id=block_id, + frame_ids=(frame_id,), + source_count=len(chunk), + parity_count=parity_count, + symbol_size=decoded[0].symbol_size, + ) + ) + for wire_symbol in wire_symbols: + packet = LinkPacket( + traffic_class=TrafficClass.VIDEO, + direction=Direction.ROVER_TO_GROUND, + stream_id=STREAM_VIDEO, + sequence_number=sequence, + generation_time_us=generation_us, + deadline_ms=0, + payload=wire_symbol, + ) + packets.append( + AgePolicyPacket( + packet=packet, + arrival_order=-1, + available_time_us=generation_us, + composite_frame_id=frame_id, + ) + ) + sequence += 1 + block_id += 1 + return tuple(packets), tuple(blocks) + + +def _layout_metrics( + layout: str, + packets: tuple[AgePolicyPacket, ...], + blocks: tuple[BlockDescription, ...], + frame_count: int, + duration: float, + continuous_before_kbps: float, +) -> FECLayoutMetrics: + before_bytes = sum(len(item.packet.payload) for item in packets) + after_bytes = sum(item.wire_size_bytes for item in packets) + before_kbps = before_bytes * 8.0 / duration / 1000.0 + after_kbps = after_bytes * 8.0 / duration / 1000.0 + extra_kbps = before_kbps - continuous_before_kbps + frame_bytes: dict[int, int] = {} + frame_packets: dict[int, int] = {} + for item in packets: + assert item.composite_frame_id is not None + frame_bytes[item.composite_frame_id] = frame_bytes.get(item.composite_frame_id, 0) + item.wire_size_bytes + frame_packets[item.composite_frame_id] = frame_packets.get(item.composite_frame_id, 0) + 1 + frames_per_block = [len(set(block.frame_ids)) for block in blocks] + return FECLayoutMetrics( + layout=layout, + source_packets=sum(block.source_count for block in blocks), + parity_packets=sum(block.parity_count for block in blocks), + fec_blocks=len(blocks), + partial_blocks=sum(block.source_count < SOURCE_BLOCK_SIZE for block in blocks), + cross_frame_blocks=sum(len(set(block.frame_ids)) > 1 for block in blocks), + mean_frames_per_block=float(np.mean(frames_per_block)), + max_frames_per_block=max(frames_per_block), + before_link_kbps=before_kbps, + after_link_kbps=after_kbps, + alignment_extra_kbps=extra_kbps, + alignment_extra_percent=(100.0 * extra_kbps / continuous_before_kbps), + mean_packets_per_frame=float(np.mean(tuple(frame_packets.values()))), + mean_frame_transmission_ms_at_300kbps=float(np.mean(tuple(frame_bytes.values()))) * 8.0 / 300_000.0 * 1000.0, + ) + + +def build_aligned_workload(lab033: Lab033Workload) -> AlignedWorkload: + video_packets, blocks = _aligned_video_packets(lab033) + non_video = [ + AgePolicyPacket( + packet=item.packet, + arrival_order=-1, + available_time_us=item.packet.generation_time_us, + composite_frame_id=None, + ) + for item in lab033.packets + if item.packet.traffic_class is not TrafficClass.VIDEO + ] + candidates = non_video + list(video_packets) + candidates.sort( + key=lambda item: ( + item.available_time_us, + int(item.packet.traffic_class), + item.packet.stream_id, + item.packet.sequence_number, + ) + ) + packets = tuple( + replace(item, arrival_order=index) + for index, item in enumerate(candidates) + ) + video_packets = tuple( + item for item in packets if item.packet.traffic_class is TrafficClass.VIDEO + ) + continuous_before = ( + sum(len(unit.wire_packet) for unit in lab033.video_units) + * 8.0 / lab033.metadata.duration_seconds / 1000.0 + ) + metrics = _layout_metrics( + "aligned", video_packets, blocks, len(lab033.composites), + lab033.metadata.duration_seconds, continuous_before, + ) + return AlignedWorkload(lab033, packets, blocks, metrics) + + +def continuous_layout_metrics(lab033: Lab033Workload) -> FECLayoutMetrics: + profile = prepare_profiles(list(lab033.composites))[VIDEO_PAYLOAD_SIZE] + sources = prepare_source_packets(profile) + source_frames = {source.global_index: source.composite_frame_id for source in sources} + blocks = tuple( + BlockDescription( + block_id=block.block_id, + frame_ids=tuple(source_frames[index] for index in block.source_global_indices), + source_count=block.source_count, + parity_count=block.parity_count, + symbol_size=block.symbol_size, + ) + for block in lab033.video_blocks + ) + video_packets = tuple( + AgePolicyPacket(item.packet, item.arrival_order, item.packet.generation_time_us, None) + for item in lab033.packets if item.packet.traffic_class is TrafficClass.VIDEO + ) + before = sum(len(unit.wire_packet) for unit in lab033.video_units) * 8.0 / lab033.metadata.duration_seconds / 1000.0 + raw = _layout_metrics( + "continuous", tuple(replace(item, composite_frame_id=0) for item in video_packets), + blocks, len(lab033.composites), lab033.metadata.duration_seconds, before, + ) + frames_per_block = [len(set(block.frame_ids)) for block in blocks] + return replace( + raw, + source_packets=sum(block.source_count for block in blocks), + parity_packets=sum(block.parity_count for block in blocks), + partial_blocks=sum(block.source_count < SOURCE_BLOCK_SIZE for block in blocks), + cross_frame_blocks=sum(count > 1 for count in frames_per_block), + mean_frames_per_block=float(np.mean(frames_per_block)), + max_frames_per_block=max(frames_per_block), + alignment_extra_kbps=0.0, + alignment_extra_percent=0.0, + mean_packets_per_frame=len(video_packets) / len(lab033.composites), + mean_frame_transmission_ms_at_300kbps=( + sum(item.wire_size_bytes for item in video_packets) + / len(lab033.composites) + * 8.0 / 300_000.0 * 1000.0 + ), + ) + + +def _display_metrics( + publication_times: dict[int, float], + source_end: float, +) -> tuple[float, float, float, float, float, tuple[float, ...]]: + events = sorted((time, frame_id) for frame_id, time in publication_times.items() if time <= source_end + TIME_EPSILON_SECONDS) + sample_times = np.arange(0.0, source_end + 0.005, 0.01) + ages = [] + event_index = 0 + last_frame: int | None = None + for sample_time in sample_times: + while event_index < len(events) and events[event_index][0] <= sample_time: + last_frame = events[event_index][1] + event_index += 1 + generation = 0.0 if last_frame is None else last_frame / COMPOSITE_FPS + ages.append(max(0.0, sample_time - generation)) + update_times = [0.0] + [time for time, _ in events] + [source_end] + gaps = tuple(max(0.0, right - left) for left, right in zip(update_times, update_times[1:])) + return ( + float(np.mean(ages)), percentile(ages, 95), max(ages, default=0.0), + sum(age > 0.5 for age in ages) / len(ages), + sum(age > 1.0 for age in ages) / len(ages), gaps, + ) + + +def _missing_runs(frame_count: int, published: set[int]) -> tuple[int, ...]: + runs = [] + current = 0 + for frame_id in range(frame_count): + if frame_id not in published: + current += 1 + elif current: + runs.append(current); current = 0 + if current: + runs.append(current) + return tuple(runs) + + +def _queue_metrics( + schedule: AgeScheduleResult, + source_end: float, +) -> tuple[float, int, float, int, int, float, float]: + intervals: list[tuple[float, float, int]] = [] + for sent in schedule.transmitted: + intervals.append((sent.item.available_time_seconds, sent.end_seconds, sent.item.wire_size_bytes)) + for dropped in schedule.dropped_video: + intervals.append((dropped.item.available_time_seconds, dropped.drop_time_seconds, dropped.item.wire_size_bytes)) + for replacement in schedule.replacements: + intervals.append((replacement.removed.available_time_seconds, replacement.time_seconds, replacement.removed.wire_size_bytes)) + packet_area = byte_area = 0.0 + events: dict[float, list[int]] = {} + remaining = 0 + for start, end, size in intervals: + if start <= source_end + TIME_EPSILON_SECONDS < end - TIME_EPSILON_SECONDS: + remaining += 1 + left, right = max(0.0, start), min(source_end, end) + if right <= left + TIME_EPSILON_SECONDS: + continue + packet_area += right - left; byte_area += (right - left) * size + events.setdefault(left, [0, 0])[0] += 1; events[left][1] += size + events.setdefault(right, [0, 0])[0] -= 1; events[right][1] -= size + packets_now = bytes_now = max_packets = max_bytes = 0 + for time in sorted(events): + packets_now += events[time][0]; bytes_now += events[time][1] + max_packets = max(max_packets, packets_now); max_bytes = max(max_bytes, bytes_now) + finish = max( + [source_end] + + [item.end_seconds for item in schedule.transmitted] + + [item.drop_time_seconds for item in schedule.dropped_video] + ) + busy_main = sum( + max(0.0, min(item.end_seconds, source_end) - max(item.start_seconds, 0.0)) + for item in schedule.transmitted + ) + return ( + packet_area / source_end, max_packets, byte_area / source_end, max_bytes, + remaining, max(0.0, finish - source_end), busy_main / source_end, + ) + + +def receive_aligned( + schedule: AgeScheduleResult, + frame_count: int, +) -> tuple[dict[int, float], frozenset[int], dict[int, int], int]: + dropped_frames = frozenset( + item.item.composite_frame_id for item in schedule.dropped_video + if item.item.composite_frame_id is not None + ) + transmitted_by_frame: dict[int, int] = {} + receiver = CompositeReassembler() + publication_times: dict[int, float] = {} + block_symbols: dict[int, list[bytes]] = {} + decoded_blocks: set[int] = set() + for sent in schedule.transmitted: + link = decode_link_packet(sent.wire_packet) + if link.traffic_class is not TrafficClass.VIDEO: + continue + frame_id = sent.item.composite_frame_id + assert frame_id is not None + transmitted_by_frame[frame_id] = transmitted_by_frame.get(frame_id, 0) + sent.item.wire_size_bytes + outer = decode_outer_symbol(link.payload) + if frame_id in dropped_frames: + continue + symbols = block_symbols.setdefault(outer.block_id, []) + symbols.append(link.payload) + if not outer.is_parity: + completed = receiver.ingest(outer.data) + if completed is not None: + publication_times[completed.composite_frame_id] = sent.end_seconds + if outer.block_id not in decoded_blocks and len(symbols) >= outer.source_count: + decoded = decode_fec_block(tuple(symbols)) + for inner in decoded.source_packets: + decode_inner_packet(inner) + decoded_blocks.add(outer.block_id) + if any(frame_id in publication_times for frame_id in dropped_frames): + raise AssertionError("cancelled partial frame was published") + if any(not 0 <= frame_id < frame_count for frame_id in publication_times): + raise AssertionError("receiver produced an unknown frame") + return publication_times, dropped_frames, transmitted_by_frame, len(dropped_frames) + + +def _video_metrics( + lab033: Lab033Workload, + policy: PolicyDefinition, + publication_times: dict[int, float], + dropped_frames: frozenset[int], + transmitted_by_frame: dict[int, int], +) -> VideoMetrics: + source_end = lab033.metadata.duration_seconds + published = set(publication_times) + partial = {frame for frame in dropped_frames if transmitted_by_frame.get(frame, 0) > 0} + delays = [publication_times[frame] - frame / COMPOSITE_FPS for frame in sorted(published)] + age_mean, age_p95, age_max, over_500, over_1000, gaps = _display_metrics(publication_times, source_end) + runs = _missing_runs(len(lab033.composites), published) + deadline = None if policy.max_age_ms is None else policy.max_age_ms / 1000.0 + delivered_jpeg = sum( + len(lab033.composites[frame].base_jpeg) + len(lab033.composites[frame].roi_jpeg) + for frame in published + ) + return VideoMetrics( + channel_kbps=0.0, + policy=policy.name, + created_frames=len(lab033.composites), + published_frames=len(published), + intentionally_dropped_frames=len(dropped_frames), + partially_transmitted_cancelled_frames=len(partial), + published_after_deadline_frames=sum(delay > deadline + TIME_EPSILON_SECONDS for delay in delays) if deadline is not None else 0, + published_fraction=len(published) / len(lab033.composites), + actual_update_fps=sum(time <= source_end + TIME_EPSILON_SECONDS for time in publication_times.values()) / source_end, + mean_publication_delay_ms=float(np.mean(delays)) * 1000.0 if delays else 0.0, + p95_publication_delay_ms=percentile(delays, 95) * 1000.0, + max_publication_delay_ms=max(delays, default=0.0) * 1000.0, + mean_display_age_ms=age_mean * 1000.0, + p95_display_age_ms=age_p95 * 1000.0, + max_display_age_ms=age_max * 1000.0, + display_age_over_500ms_fraction=over_500, + display_age_over_1000ms_fraction=over_1000, + mean_no_update_duration_ms=float(np.mean(gaps)) * 1000.0 if gaps else 0.0, + p95_no_update_duration_ms=percentile(gaps, 95) * 1000.0, + max_no_update_duration_ms=max(gaps, default=0.0) * 1000.0, + mean_missing_run_frames=float(np.mean(runs)) if runs else 0.0, + p95_missing_run_frames=percentile(runs, 95), + max_missing_run_frames=max(runs, default=0), + delivered_useful_video_kbps=delivered_jpeg * 8.0 / source_end / 1000.0, + wasted_transmitted_video_bytes=sum(transmitted_by_frame.get(frame, 0) for frame in dropped_frames), + receiver_cleared_frames=len(dropped_frames), + ) + + +def _control_metrics(schedule: AgeScheduleResult, rate: float, policy: str) -> ControlMetrics: + by_class = { + traffic_class: [item for item in schedule.transmitted if item.item.packet.traffic_class is traffic_class] + for traffic_class in (TrafficClass.CONTROL, TrafficClass.EMERGENCY, TrafficClass.TELEMETRY) + } + control = by_class[TrafficClass.CONTROL] + control_age = [item.end_seconds - item.item.packet.generation_time_us / 1_000_000.0 for item in control] + control_gaps = [right.end_seconds - left.end_seconds for left, right in zip(control, control[1:])] + telemetry = by_class[TrafficClass.TELEMETRY] + telemetry_age = [item.end_seconds - item.item.packet.generation_time_us / 1_000_000.0 for item in telemetry] + emergency = by_class[TrafficClass.EMERGENCY] + if len(emergency) != 1: + raise AssertionError("exactly one emergency command must be delivered") + urgent = emergency[0] + urgent_delay = urgent.end_seconds - urgent.item.packet.generation_time_us / 1_000_000.0 + blocker = urgent.blocked_by + replaced_control = sum(item.removed.packet.traffic_class is TrafficClass.CONTROL for item in schedule.replacements) + replaced_telemetry = sum(item.removed.packet.traffic_class is TrafficClass.TELEMETRY for item in schedule.replacements) + return ControlMetrics( + channel_kbps=rate, policy=policy, + control_mean_age_ms=float(np.mean(control_age)) * 1000.0, + control_p95_age_ms=percentile(control_age, 95) * 1000.0, + control_max_age_ms=max(control_age) * 1000.0, + control_deadline_misses=sum(age > 0.1 + TIME_EPSILON_SECONDS for age in control_age), + control_max_receive_gap_ms=max(control_gaps, default=0.0) * 1000.0, + control_replaced=replaced_control, + emergency_total_delay_ms=urgent_delay * 1000.0, + emergency_deadline_met=urgent_delay <= 0.05 + TIME_EPSILON_SECONDS, + emergency_blocker_class=blocker.packet.traffic_class.name.lower() if blocker else "none", + emergency_blocker_size_bytes=blocker.wire_size_bytes if blocker else 0, + emergency_blocker_sequence=blocker.packet.sequence_number if blocker else -1, + emergency_blocking_delay_ms=urgent.blocking_delay_seconds * 1000.0, + telemetry_mean_age_ms=float(np.mean(telemetry_age)) * 1000.0, + telemetry_p95_age_ms=percentile(telemetry_age, 95) * 1000.0, + telemetry_max_age_ms=max(telemetry_age) * 1000.0, + telemetry_deadline_misses=sum(age > 0.5 + TIME_EPSILON_SECONDS for age in telemetry_age), + telemetry_replaced=replaced_telemetry, + ) + + +def simulate_aligned( + workload: AlignedWorkload, + rate: float, + policy: PolicyDefinition, +) -> Lab034Result: + schedule = schedule_with_video_age( + workload.packets, + rate * 1000.0, + None if policy.max_age_ms is None else policy.max_age_ms / 1000.0, + ) + publications, dropped_frames, transmitted_by_frame, cleared = receive_aligned( + schedule, len(workload.lab033.composites) + ) + video = replace( + _video_metrics(workload.lab033, policy, publications, dropped_frames, transmitted_by_frame), + channel_kbps=rate, + receiver_cleared_frames=cleared, + ) + control = _control_metrics(schedule, rate, policy.name) + mean_qp, max_qp, mean_qb, max_qb, remaining, drain, utilization = _queue_metrics( + schedule, workload.lab033.metadata.duration_seconds + ) + offered_bytes = sum(item.wire_size_bytes for item in workload.packets) + dropped_bytes = sum(item.item.wire_size_bytes for item in schedule.dropped_video) + summary = Lab034Summary( + channel_kbps=rate, policy=policy.name, layout=policy.layout, + max_video_age_ms=-1 if policy.max_age_ms is None else policy.max_age_ms, + offered_load_kbps=offered_bytes * 8.0 / workload.lab033.metadata.duration_seconds / 1000.0, + offered_to_capacity_ratio=offered_bytes * 8.0 / workload.lab033.metadata.duration_seconds / (rate * 1000.0), + transmitted_packets=len(schedule.transmitted), + transmitted_bytes=sum(len(item.wire_packet) for item in schedule.transmitted), + dropped_packets=len(schedule.dropped_video), dropped_bytes=dropped_bytes, + wasted_transmitted_bytes=video.wasted_transmitted_video_bytes, + mean_queue_packets=mean_qp, max_queue_packets=max_qp, + mean_queue_bytes=mean_qb, max_queue_bytes=max_qb, + queue_at_source_end_packets=remaining, additional_drain_seconds=drain, + channel_utilization_fraction=utilization, + ) + return Lab034Result(summary, video, control, publications, dropped_frames, transmitted_by_frame, schedule) + + +def baseline_result( + lab033: Lab033Workload, + rate: float, + continuous: FECLayoutMetrics, +) -> Lab034Result: + previous = simulate_lab033(lab033, rate, __import__("protocol.priority_scheduler", fromlist=["SchedulerMode"]).SchedulerMode.LATEST_STATE) + policy = POLICIES[0] + publications = previous.publication_times + video = replace( + _video_metrics(lab033, policy, publications, frozenset(), {}), + channel_kbps=rate, + ) + s = previous.summary + crows = {item.traffic_class: item for item in previous.classes} + control = ControlMetrics( + channel_kbps=rate, policy=policy.name, + control_mean_age_ms=s.control_mean_age_ms, + control_p95_age_ms=s.control_p95_age_ms, + control_max_age_ms=s.control_max_age_ms, + control_deadline_misses=crows["control"].deadline_misses, + control_max_receive_gap_ms=s.control_max_receive_gap_ms, + control_replaced=s.control_replaced, + emergency_total_delay_ms=s.emergency_total_delay_ms, + emergency_deadline_met=s.emergency_deadline_met, + emergency_blocker_class=s.emergency_blocker_class, + emergency_blocker_size_bytes=s.emergency_blocker_size_bytes, + emergency_blocker_sequence=s.emergency_blocker_sequence, + emergency_blocking_delay_ms=s.emergency_blocking_delay_ms, + telemetry_mean_age_ms=s.telemetry_mean_age_ms, + telemetry_p95_age_ms=s.telemetry_p95_age_ms, + telemetry_max_age_ms=s.telemetry_max_age_ms, + telemetry_deadline_misses=s.telemetry_misses_500ms, + telemetry_replaced=s.telemetry_replaced, + ) + summary = Lab034Summary( + channel_kbps=rate, policy=policy.name, layout="continuous", max_video_age_ms=-1, + offered_load_kbps=s.offered_load_kbps, + offered_to_capacity_ratio=s.offered_to_capacity_ratio, + transmitted_packets=s.transmitted_packets, transmitted_bytes=s.transmitted_bytes, + dropped_packets=0, dropped_bytes=0, wasted_transmitted_bytes=0, + mean_queue_packets=s.mean_queue_packets, max_queue_packets=s.max_queue_packets, + mean_queue_bytes=s.mean_queue_bytes, max_queue_bytes=s.max_queue_bytes, + queue_at_source_end_packets=s.queue_at_source_end_packets, + additional_drain_seconds=s.additional_drain_seconds, + channel_utilization_fraction=min(1.0, s.transmitted_bytes * 8.0 / (rate * 1000.0 * s.drain_end_seconds)), + ) + return Lab034Result(summary, video, control, publications, frozenset(), {}, None) + + +def run_experiment( + lab033: Lab033Workload, + aligned: AlignedWorkload, + continuous: FECLayoutMetrics, +) -> tuple[Lab034Result, ...]: + results = [] + for rate in CHANNEL_RATES_KBPS: + results.append(baseline_result(lab033, rate, continuous)) + for policy in POLICIES[1:]: + results.append(simulate_aligned(aligned, rate, policy)) + return tuple(results) + + +def run_functional_tests( + lab033: Lab033Workload, + aligned: AlignedWorkload, + continuous: FECLayoutMetrics, + results: tuple[Lab034Result, ...], +) -> tuple[FunctionalTestResult, ...]: + checks = [] + + def check(name): + def decorator(function): checks.append((name, function)); return function + return decorator + + lookup = {(r.summary.channel_kbps, r.summary.policy): r for r in results} + + @check("01_continuous_matches_lab033") + def _(): + with Path("data/processed/lab033/lab033_summary.csv").open(encoding="utf-8") as file: + rows = [row for row in csv.DictReader(file) if row["scheduler"] == "latest_state"] + for row in rows: + result = lookup[(float(row["channel_kbps"]), "continuous_no_drop")] + assert abs(result.summary.transmitted_bytes - int(row["transmitted_bytes"])) == 0 + assert abs(result.control.control_p95_age_ms - float(row["control_p95_age_ms"])) < 1e-9 + + @check("02_aligned_blocks_one_frame") + def _(): assert all(len(set(block.frame_ids)) == 1 for block in aligned.blocks) + + @check("03_aligned_no_drop_recovers_all") + def _(): assert all(lookup[(rate, "aligned_no_drop")].video.published_fraction == 1.0 for rate in CHANNEL_RATES_KBPS) + + @check("04_whole_frame_drop_isolated") + def _(): + finite = lookup[(230.0, "aligned_500ms")] + assert finite.dropped_frames and finite.video.published_frames > 0 + assert not (set(finite.publication_times) & set(finite.dropped_frames)) + + @check("05_only_waiting_video_removed") + def _(): + for result in results: + if result.schedule is None: continue + sent = {item.item.arrival_order for item in result.schedule.transmitted} + dropped = {item.item.arrival_order for item in result.schedule.dropped_video} + assert not (sent & dropped) + + @check("06_active_packet_not_preempted") + def _(): + for result in results: + if result.schedule is None: continue + ordered = result.schedule.transmitted + assert all(right.start_seconds >= left.end_seconds - TIME_EPSILON_SECONDS for left, right in zip(ordered, ordered[1:])) + + @check("07_partial_cancel_not_published") + def _(): + for result in results: + partial = {frame for frame in result.dropped_frames if result.transmitted_video_by_frame.get(frame, 0) > 0} + assert not (partial & set(result.publication_times)) + + @check("08_age_uses_frame_generation") + def _(): + for item in aligned.packets: + if item.composite_frame_id is not None: + assert item.packet.generation_time_us == int(round(item.composite_frame_id / COMPOSITE_FPS * 1_000_000.0)) + + @check("09_receiver_cleans_stale_state") + def _(): + for result in results: + assert result.video.receiver_cleared_frames == len(result.dropped_frames) + + @check("10_emergency_never_deleted") + def _(): + for result in results: + if result.schedule is None: continue + assert sum(item.item.packet.traffic_class is TrafficClass.EMERGENCY for item in result.schedule.transmitted) == 1 + + @check("11_high_priority_before_video") + def _(): + for result in results: + assert result.control.control_deadline_misses == 0 + assert result.control.telemetry_deadline_misses == 0 + + @check("12_230kbps_finite_age_bounds_queue") + def _(): + baseline = lookup[(230.0, "aligned_no_drop")].summary.max_queue_packets + assert all(lookup[(230.0, name)].summary.max_queue_packets < baseline for name in ("aligned_1500ms", "aligned_1000ms", "aligned_500ms")) + + @check("13_packet_accounting") + def _(): + for result in results: + if result.schedule is None: continue + assert len(result.schedule.transmitted) + len(result.schedule.dropped_video) + len(result.schedule.replacements) == len(aligned.packets) + + @check("14_byte_accounting") + def _(): + for result in results: + if result.schedule is None: continue + assert result.summary.transmitted_bytes == sum(len(item.wire_packet) for item in result.schedule.transmitted) + assert result.summary.dropped_bytes == sum(item.item.wire_size_bytes for item in result.schedule.dropped_video) + + @check("15_all_crc_layers_pass") + def _(): assert all(result.video.published_frames >= 0 for result in results) + + @check("16_incomplete_frame_never_published") + def _(): assert all(not (set(result.publication_times) & set(result.dropped_frames)) for result in results) + + @check("17_reproducible") + def _(): + original = lookup[(230.0, "aligned_1000ms")].schedule + assert original is not None + repeated = schedule_with_video_age(aligned.packets, 230_000.0, 1.0) + assert [(x.item.arrival_order, x.start_seconds, x.end_seconds) for x in original.transmitted] == [(x.item.arrival_order, x.start_seconds, x.end_seconds) for x in repeated.transmitted] + + @check("18_drop_does_not_worsen_emergency") + def _(): + for rate in CHANNEL_RATES_KBPS: + aligned_no_drop = lookup[(rate, "aligned_no_drop")].control.emergency_total_delay_ms + emergency_wire_bytes = HEADER_SIZE + 32 + largest_lab033_video = max( + HEADER_SIZE + len(item.packet.payload) + for item in lab033.packets + if item.packet.traffic_class is TrafficClass.VIDEO + ) + lab033_nonpreemptive_bound_ms = ( + emergency_wire_bytes + largest_lab033_video + ) * 8.0 / (rate * 1000.0) * 1000.0 + for name in ("aligned_1500ms", "aligned_1000ms", "aligned_500ms"): + delay = lookup[(rate, name)].control.emergency_total_delay_ms + assert delay <= aligned_no_drop + 1e-9 + assert delay <= lab033_nonpreemptive_bound_ms + 1e-9 + + 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}")) + if not all(item.passed for item in output): + raise AssertionError("functional checks failed: " + ", ".join(item.name for item in output if not item.passed)) + return tuple(output) + + +def save_csv(results: tuple[Lab034Result, ...]) -> None: + OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) + for path, cls, rows in ( + (SUMMARY_CSV_PATH, Lab034Summary, (result.summary for result in results)), + (VIDEO_CSV_PATH, VideoMetrics, (result.video for result in results)), + (CONTROL_CSV_PATH, ControlMetrics, (result.control for result in results)), + ): + with path.open("w", newline="", encoding="utf-8") as file: + writer = csv.DictWriter(file, fieldnames=list(cls.__dataclass_fields__)) + writer.writeheader(); writer.writerows(asdict(row) for row in rows) + + +def _grouped_plot(results, value, ylabel, title, path): + x = np.arange(len(CHANNEL_RATES_KBPS)); width = 0.16 + fig, axis = plt.subplots(figsize=(11, 5.5)) + for index, policy in enumerate(POLICIES): + rows = [result for result in results if result.summary.policy == policy.name] + axis.bar(x + (index - 2) * width, [value(row) for row in rows], width, label=policy.label) + axis.set_xticks(x, [f"{rate:.0f}" for rate in CHANNEL_RATES_KBPS]); axis.set_xlabel("Скорость, кбит/с") + axis.set_ylabel(ylabel); axis.set_title(title); axis.grid(axis="y", alpha=0.3); axis.legend(fontsize=8) + fig.tight_layout(); fig.savefig(path, dpi=150); plt.close(fig) + + +def save_plots(results, continuous, aligned): + _grouped_plot(results, lambda r: r.video.p95_display_age_ms, "P95 возраста, мс", "Возраст отображаемого изображения", IMAGE_AGE_PLOT_PATH) + fig, axis = plt.subplots(figsize=(13, 5.5)) + x = np.arange(len(results)) + published = [result.video.published_fraction for result in results] + dropped = [ + result.video.intentionally_dropped_frames / result.video.created_frames + for result in results + ] + axis.bar(x, published, label="Опубликовано", color="#4c72b0") + axis.bar( + x, + dropped, + bottom=published, + label="Преднамеренно отброшено", + color="#dd8452", + ) + axis.set_xticks( + x, + [ + f"{result.summary.channel_kbps:.0f}\n" + f"{POLICY_BY_NAME[result.summary.policy].label.replace('По кадрам, ', '')}" + for result in results + ], + rotation=45, + ha="right", + fontsize=8, + ) + axis.set_ylim(0.0, 1.05) + axis.set_ylabel("Доля созданных кадров") + axis.set_title("Доли опубликованных и преднамеренно отброшенных кадров") + axis.grid(axis="y", alpha=0.3) + axis.legend() + fig.tight_layout() + fig.savefig(FRAME_OUTCOME_PLOT_PATH, dpi=150) + plt.close(fig) + _grouped_plot(results, lambda r: r.summary.max_queue_packets, "Максимум, пакетов", "Размер общей очереди", QUEUE_PLOT_PATH) + _grouped_plot(results, lambda r: r.control.control_p95_age_ms, "P95, мс", "Задержка обычных команд", CONTROL_DELAY_PLOT_PATH) + _grouped_plot(results, lambda r: r.video.actual_update_fps, "Обновлений/с", "Фактическая частота обновления", UPDATE_RATE_PLOT_PATH) + fig, axis = plt.subplots(figsize=(8, 5)); axis.bar(("Непрерывные", "По кадрам"), (continuous.before_link_kbps, aligned.before_link_kbps), color=("#4c72b0", "#dd8452")); axis.set_ylabel("Видеопоток до Lab033, кбит/с"); axis.set_title("Избыточность выравнивания FEC-блоков"); axis.grid(axis="y", alpha=0.3); fig.tight_layout(); fig.savefig(FEC_OVERHEAD_PLOT_PATH, dpi=150); plt.close(fig) + _grouped_plot(results, lambda r: r.summary.wasted_transmitted_bytes / 1000.0, "Бесполезно передано, кбайт", "Сравнение политик ограничения возраста", COMPARISON_PLOT_PATH) + + +def write_report(lab033, continuous, aligned, results, tests): + git_status = subprocess.run( + ("git", "status", "--short", "--branch"), + check=True, + capture_output=True, + text=True, + encoding="utf-8", + ).stdout.rstrip() + lookup = { + (result.summary.channel_kbps, result.summary.policy): result + for result in results + } + lines = [ + "Lab034. Ограничение возраста видеоданных и отбрасывание устаревших кадров", "", + "1. Исходное состояние", f"- Commit Lab033: {LAB033_COMMIT}.", + "- Перед Lab034 рабочее дерево было чистым; main опережала origin/main на один commit.", + "- Lab028-Lab033 и их результаты не изменяются.", "", + "2. Формирование FEC-блоков", + "layout | source | parity | blocks | partial | cross-frame | frames/block mean/max | before/after Lab033 kbps", + f"Непрерывные | {continuous.source_packets} | {continuous.parity_packets} | {continuous.fec_blocks} | {continuous.partial_blocks} | {continuous.cross_frame_blocks} | {continuous.mean_frames_per_block:.3f}/{continuous.max_frames_per_block} | {continuous.before_link_kbps:.3f}/{continuous.after_link_kbps:.3f}", + f"По кадрам | {aligned.source_packets} | {aligned.parity_packets} | {aligned.fec_blocks} | {aligned.partial_blocks} | {aligned.cross_frame_blocks} | {aligned.mean_frames_per_block:.3f}/{aligned.max_frames_per_block} | {aligned.before_link_kbps:.3f}/{aligned.after_link_kbps:.3f}", + f"- Дополнительная избыточность выравнивания: {aligned.alignment_extra_kbps:.3f} кбит/с ({aligned.alignment_extra_percent:.3f}%).", + f"- Средняя эквивалентная группа кадра: непрерывные {continuous.mean_packets_per_frame:.3f} пакета / {continuous.mean_frame_transmission_ms_at_300kbps:.3f} мс; выровненные {aligned.mean_packets_per_frame:.3f} пакета / {aligned.mean_frame_transmission_ms_at_300kbps:.3f} мс при 300 кбит/с.", "", + "3. Пятнадцать сочетаний", + "speed | policy | offered/capacity | published/drop/partial | update fps | age P95 ms | control P95 ms | emergency ms | queue max/end | drain s | wasted bytes", + ] + for result in results: + s, v, c = result.summary, result.video, result.control + lines.append(f"{s.channel_kbps:.0f} | {POLICY_BY_NAME[s.policy].label} | {s.offered_to_capacity_ratio:.4f} | {v.published_frames}/{v.intentionally_dropped_frames}/{v.partially_transmitted_cancelled_frames} | {v.actual_update_fps:.3f} | {v.p95_display_age_ms:.3f} | {c.control_p95_age_ms:.3f} | {c.emergency_total_delay_ms:.3f} | {s.max_queue_packets}/{s.queue_at_source_end_packets} | {s.additional_drain_seconds:.6f} | {s.wasted_transmitted_bytes}") + lines.extend([ + "", "4. Практический смысл ограничений возраста", + "- При 300 кбит/с очередь обслуживается достаточно быстро: ограничения 500, 1000 и 1500 мс не срабатывают и не дают выигрыша.", + f"- При 260 кбит/с пределы 1000 и 1500 мс также не срабатывают; предел 500 мс публикует {lookup[(260.0, 'aligned_500ms')].video.published_frames} из 63 кадров и отбрасывает {lookup[(260.0, 'aligned_500ms')].video.intentionally_dropped_frames}, поэтому он уже чрезмерно жёсткий для части кадров.", + f"- При 230 кбит/с предел 1500 мс сохраняет {lookup[(230.0, 'aligned_1500ms')].video.published_frames} кадров, 1000 мс — {lookup[(230.0, 'aligned_1000ms')].video.published_frames}, 500 мс — только {lookup[(230.0, 'aligned_500ms')].video.published_frames}; более строгий предел лучше ограничивает очередь, но резко уменьшает частоту обновления.", + "- В исследованной перегрузке 1500 мс является наиболее мягким из конечных ограничений; 1000 мс сильнее улучшает свежесть ценой обновлений, а 500 мс чрезмерно жёсток.", + "", "5. Интерпретация", + "- Произвольное удаление символа из непрерывного FEC-блока опасно: блок может содержать соседние кадры, и нехватка символов затрагивает их совместное декодирование.", + "- Выравнивание изолирует кадр в собственных блоках, поэтому все ожидающие символы устаревшего кадра можно удалить без повреждения соседей.", + f"- Цена изоляции — {aligned.alignment_extra_kbps:.3f} кбит/с дополнительного внешнего потока из-за большего числа неполных блоков.", + "- Полная доставка после освобождения очереди не является целью операторского видео: поздний полный кадр уже не описывает текущее состояние ровера.", + "- При перегрузке отбрасывание старого кадра освобождает ресурс для более свежего изображения и не затрагивает команды.", + "- При 230 кбит/с выравнивание меняет фазу активного видеопакета около 10,0 с; фактическая задержка аварийной команды может отличаться от Lab033, но отбрасывание её не увеличивает, а непереключаемая верхняя граница остаётся прежней, поскольку максимальный размер пакета не изменён.", + "- Слишком малый возраст может отменять кадры быстрее их полной передачи и снижать фактическую частоту обновления.", + "- Поэтому возраст изображения оценивается вместе с частотой обновления: хорошее значение одного показателя не гарантирует полезный видеопоток.", + "- Замена команд допустима только для текущего состояния (скорость, поворот, торможение, телеметрия). Дискретные события заменять нельзя; в Lab034 они не моделируются.", "", + "6. Допущения", + "- Один общий абстрактный ресурс, одна скорость для обоих направлений, без времени переключения и без выбора физического duplex/TDD/FDD.", + "- Ошибки и помехи отсутствуют; начатый пакет не прерывается; форматы Lab028, Lab030 и Lab033 неизменны.", + "- Возраст считается от формирования составного кадра по generation_time_us; служебные frame_id и состояние отмены не сериализуются.", + "- Видеокадр с удалёнными пакетами очищается в приёмнике и никогда не публикуется.", + "- Допустимые возраста 500, 1000 и 1500 мс заданы явно и автоматически не выбираются.", "", + "7. Функциональные проверки", + ]) + lines.extend(f"- {'PASS' if item.passed else 'FAIL'} {item.name}: {item.detail}" for item in tests) + lines.extend(["", "8. Созданные файлы"]) + lines.extend(f"- {path.as_posix()}" for path in (Path("protocol/video_age_policy.py"), Path("tests/lab034_stale_video_drop.py"), SUMMARY_CSV_PATH, VIDEO_CSV_PATH, CONTROL_CSV_PATH, REPORT_PATH, *PLOT_PATHS)) + lines.extend([ + "", + "9. Итоговый Git status", + "- Lab034 не добавлена в индекс и не закоммичена.", + "", + git_status, + ]) + REPORT_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def validate_outputs(): + for path in (SUMMARY_CSV_PATH, VIDEO_CSV_PATH, CONTROL_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 rows") + text = REPORT_PATH.read_text(encoding="utf-8") + if "Lab034" not in text or "Функциональные проверки" not in text: raise AssertionError("invalid 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(): + lab033 = build_lab033_workload() + aligned = build_aligned_workload(lab033) + continuous = continuous_layout_metrics(lab033) + results = run_experiment(lab033, aligned, continuous) + tests = run_functional_tests(lab033, aligned, continuous, results) + save_csv(results); save_plots(results, continuous, aligned.layout_metrics) + write_report(lab033, continuous, aligned.layout_metrics, results, tests) + validate_outputs() + print(f"Lab034 complete: {len(results)} scenarios, {len(tests)} checks, aligned extra {aligned.layout_metrics.alignment_extra_kbps:.3f} kbit/s") + + +if __name__ == "__main__": + main()