diff --git a/PROJECT_LOG.md b/PROJECT_LOG.md index e6c9b2f..bc1571c 100644 --- a/PROJECT_LOG.md +++ b/PROJECT_LOG.md @@ -1641,3 +1641,116 @@ Git add/commit/push/pull/fetch не выполнялись. и проверкой выполнимости 5 кадров/с. Сопряжение CRSF и его сценарии отказов требуют отдельной постановки и уточнения аппаратуры. Нового RF, QPSK, CRSF/ELRS-протокола, перекодирования JPEG или переноса на 180 МГц здесь нет. + +# Запись 039 + +## Дата + +8 сентября 2026 года. + +## Тема + +Lab050: потоковая архитектура JPEG, ограничение возраста и блочный IQ-приём. + +## Выполнено + +По явному заданию проверены HEAD, status, diff/stat/check и staged scope Lab049. +Создан коммит cdf5396e807b83f82a264632dc46f71428fcd616 с сообщением +Lab049: define ELRS control and SDR video architecture. Вошли 12 файлов: +код, тесты, документы, журнал, индекс и указатель README. Сгенерированные данные, +настройки окружения и посторонние локальные файлы не добавлялись. Push не выполнялся. +Lab050 начата от нового HEAD и оставлена без staging/commit. + +Аудит Lab034–037 подтвердил возможность переиспользовать существующий +protocol/video_frame_scheduler.py: активное изображение завершается, +ожидающее заменяется самым свежим, перед началом прогнозируется завершение. +Добавлены готовность JPEG отдельно от времени съёмки, произвольный MAX_VIDEO_AGE, +прогноз задержки после RF и счётчик максимальной ожидающей очереди. +Пять прежних политик сохраняют поведение на проверенном сценарии видео+CONTROL. +В самой Lab050 SDR-очередь содержит только VIDEO; управление ELRS отдельно. + +Verifier/adversary воспроизвели ошибку первой реализации: позднее завершение +кодирования старого снимка вытесняло более свежий, либо возвращало старый снимок +после опустошения очереди. Исправлено сравнением времени съёмки и ID с сохранением +самого свежего виденного значения. Две регрессии и повторная независимая проверка PASS. + +Из data/raw/lab011_scene.jpg созданы четыре фиксированных профиля: +A 320×240 RGB q35 — 8724 байта; B 160×120 RGB q35 — 2771; +C 128×96 RGB q25 — 1740; D 160×120 grayscale q30 — 2028. +Первый A совпал с SHA Lab047. По 30 детерминированных сдвигов фото, не съёмка камеры; +размеры соответственно 8724–9250, 2771–3031, 1740–1893, 2028–2233 байта. +PSNR относительно несжатого изображения того же профиля: +28,12 / 28,11 / 27,17 / 30,15 dB. Это не оценка пригодности для управления. + +Нынешний пакет несёт 244 байта JPEG и занимает 5766 символьных интервалов. +Точный предел Jmax=244*floor(Rs/(5766*fps)) включает фрагментацию, добивку, +FEC, PHY header, преамбулу, защитные интервалы и фильтр. +На 500/800/1000 ксимв/с для 5 кадров/с предел 4148/6588/8296 байт; +для 10 — 1952/3172/4148 байт. BPSK, FEC R=1/2 K=7, CRC и форматы сохранены. +QPSK, новый кодек и 12+3 стирающий FEC не вводились. + +Выполнены 240 событийных сценариев: четыре профиля, вход 3/5/10 кадров/с, +Rs=200/400/500/800/1000 ксимв/с при Fs=8 Мвыб/с, предел возраста +200/333/500/1000 мс. Ожидающая очередь не превышает один снимок. +Сохраняются capture/encode/queue/TX/RX/complete/decode/display timestamps; +не выполненные стадии имеют null. FPS внутри окна и после опустошения разделены. + +Идеальная модель C/500k/5fps/200ms передаёт и показывает 30/30 за 6 с, +без удалений, FRESH DISPLAY FPS=5, возраст min/median/max +92,563 / 92,590 / 92,629 мс. Это модель нулевой стоимости RX и экрана. +При C/200k, камере 10 кадров/с и возрасте 333 мс передано 14, заменено 16; +внутри 3 с показано 13, FRESH DISPLAY FPS=4,333, ещё один после окна. +При возрасте 200 мс все 30 отклонены: радио само требует 230,64 мс. + +Добавлен StreamingReceiver.push_iq_block с выдачей пакетов до finish. +Корреляция и строгий PHY/FEC/CRC взяты из Lab048; в старый декодер добавлен +необязательный IQ-reader без изменения поведения по умолчанию. +Буфер удерживает блоки, левый запас и максимальный допустимый PHY-кадр; +TX-границы, payload, ожидаемые ID и число кадров RX не получает. +Децимация поиска 20/10/8/5/4 сохраняет две выборки/символ, порог Lab047 +0.030964473262429237 не меняется. Адаптер последовательный, live-драйвер не подключён. + +На каждой скорости выполнен шумный опыт трёх JPEG C, 24 RF-пакета: +единый coded-bit Eb/N0=7 dB, seed=50050, CFO +10 Гц, фаза 0,73 рад, смена полярности. +Все пять опытов дали 24/24 пакета и 3/3 точных JPEG до finish. +Большой IQ-файл не создавался; SHA проверялся после независимого CRC/сборки. +Обработка заняла 35,12417 / 59,47086 / 55,15755 / 60,95497 / 55,92195 с +при IQ 0,792483 / 0,401868 / 0,323745 / 0,206561 / 0,167499 с. +Отношения 44,32166 / 147,98602 / 170,37365 / 295,09484 / 333,86514: +приёмник не успевает в реальном времени. На 500k строгая обработка 54,586 с +против 0,492 с грубого поиска, 2374 кандидата и 24 принятых пакета. +После исправления вызовов API packet в модели итоговая матрица пересчитана +через --reuse-benchmarks; пять тяжёлых замеров сохранены без повторения. + +Пик удерживаемых IQ 2,13–6,51 МиБ; расчёт массивов с запасом 47,5–143,4 МиБ, +не RSS процесса. Предел IQ-буфера не зависит от длительности сессии. +История событий симулятора хранится для отчёта и не выдаётся за постоянную +память бесконечной камеры. Потери блоков/переполнение live-драйвера не реализованы. + +При подстановке измеренной стоимости RX все 30 кадров C/5fps отклоняются до TX, +FRESH DISPLAY FPS=0 для всех проверенных пределов возраста. +Условный вариант: JPEG ≤1952 байт, 500 ксимв/с, MAX_VIDEO_AGE=200 мс. +При стоимости RX-пакета ≤16 мс модель даёт 5 кадров/с и возраст около 192,5 мс; +это требует примерно 144-кратного ускорения данного замера. +Для 10 кадров/с нужны ≤12 мс и примерно 192-кратное ускорение; возраст около 160,5 мс, +29 показов в первые 3 с (9,667 кадра/с), ещё один после окна, интервалы около 10 кадров/с. +Ускорения не достигнуты и не обещаются. Обе цели без QPSK — CONDITIONAL; +нынешний RX — NO по реальному времени. Камера, экран, jitter и запас ещё не измерены. + +Добавлены experiments/lab050_video_stream.py, lab050_streaming_receiver.py, +tests/test_lab050_video_stream.py, docs/lab050_current_state.md и локальные +результаты data/processed/lab050. Обновлён LABS_INDEX. Модули аварии, +управления и сеансов не менялись; при потере ELRS видео не продлевает движение. + +44 целевых теста PASS. Verifier независимо проверил границы JPEG, 60 сценариев +очереди, 32 прогноза RX, исходные IQ-числа и все итоговые временные отметки JSON. +Adversary подтвердил устранение найденной регрессии свежести. +quick_gate PASS: 650 тестов и 14 функциональных проверок. +pytest: 650 passed за 78,73 с, одно предупреждение о доступе к .pytest_cache. +git diff --check PASS. RF TX=0, оборудование не запускалось. +Lab050 git add/commit и push/pull/fetch не выполнялись. + +Следующий этап: STOP. Рекомендуемый Lab051 — программное профилирование +и ускорение строгой проверки ложных кандидатов на C/500k с сохранением +независимости RX и FEC/CRC. До доказанной скорости и свежести не выполнять +RF, QPSK, новый видеокодек или перенос SDR на 180 МГц. diff --git a/data/processed/lab050/receiver_benchmarks.json b/data/processed/lab050/receiver_benchmarks.json new file mode 100644 index 0000000..c767e3a --- /dev/null +++ b/data/processed/lab050/receiver_benchmarks.json @@ -0,0 +1,385 @@ +[ + { + "symbol_rate": 200000, + "samples_per_symbol": 40, + "detection_decimation": 20, + "input_images": 3, + "tx_rf_packets": 24, + "accepted_rf_packets": 24, + "completed_images": 3, + "all_sha_match": true, + "packets_before_finish": 24, + "images_before_finish": 3, + "capture_seconds": 0.792483125, + "payload_airtime_seconds": 0.69192, + "processing_seconds": 35.12416689947713, + "processing_per_rf_second": 44.32165908829558, + "processing_per_payload_airtime_second": 50.76333521140758, + "coarse_seconds": 2.0254860994173214, + "strict_seconds": 32.75070630130358, + "assembly_jpeg_decode_seconds": 0.0013477999018505216, + "processing_seconds_per_rf_packet": 1.4635069541448804, + "peak_iq_buffer_samples": 853027, + "iq_buffer_limit_samples": 853556, + "working_arrays_estimate_bytes": 150378080, + "memory_is_process_rss_measurement": false, + "peak_object_payload_bytes": 1821, + "candidate_count": 2191, + "peak_candidates_in_block": 2145, + "rejected_reasons": { + "PHY_HEADER_MAGIC_MISMATCH": 2146, + "TRUNCATED_IQ_WINDOW": 21 + }, + "buffer_release_delay_upper_seconds": 0.106195, + "noisy": true, + "coded_bit_ebn0_db": 7, + "cfo_hz": 10, + "candidate_threshold": 0.030964473262429237, + "completed": [ + { + "image_id": 1, + "sha256": "15723773a8788c9d610c874707ad2d513582322ee1d42009b0b39d3de336e10d", + "bytes": 1740, + "dimensions": [ + 128, + 96 + ], + "first_sample": 4321, + "last_sample": 1854754, + "emitted_after_sample": 2382902 + }, + { + "image_id": 2, + "sha256": "66404b14ca9e3a6627871ad9d3b5dd4311915d7820a99dfa182fc3a20fbe9d85", + "bytes": 1790, + "dimensions": [ + 128, + 96 + ], + "first_sample": 1855061, + "last_sample": 3705166, + "emitted_after_sample": 4299138 + }, + { + "image_id": 3, + "sha256": "166d33d72820bf3385e0d30823d991b453147c84ca75f788ab8819ad5e3c3ced", + "bytes": 1821, + "dimensions": [ + 128, + 96 + ], + "first_sample": 3705569, + "last_sample": 5555346, + "emitted_after_sample": 6145669 + } + ], + "rf_tx": 0 + }, + { + "symbol_rate": 400000, + "samples_per_symbol": 20, + "detection_decimation": 10, + "input_images": 3, + "tx_rf_packets": 24, + "accepted_rf_packets": 24, + "completed_images": 3, + "all_sha_match": true, + "packets_before_finish": 24, + "images_before_finish": 3, + "capture_seconds": 0.401868125, + "payload_airtime_seconds": 0.34596, + "processing_seconds": 59.47086290002335, + "processing_per_rf_second": 147.98601630826866, + "processing_per_payload_airtime_second": 171.90097959308403, + "coarse_seconds": 0.6804420000407845, + "strict_seconds": 58.67772239865735, + "assembly_jpeg_decode_seconds": 0.0011727997334674, + "processing_seconds_per_rf_packet": 2.4779526208343063, + "peak_iq_buffer_samples": 493587, + "iq_buffer_limit_samples": 496316, + "working_arrays_estimate_bytes": 87503840, + "memory_is_process_rss_measurement": false, + "peak_object_payload_bytes": 1821, + "candidate_count": 2575, + "peak_candidates_in_block": 2364, + "rejected_reasons": { + "PHY_HEADER_MAGIC_MISMATCH": 2516, + "TRUNCATED_SYMBOLS": 17, + "TRUNCATED_IQ_WINDOW": 18 + }, + "buffer_release_delay_upper_seconds": 0.06154, + "noisy": true, + "coded_bit_ebn0_db": 7, + "cfo_hz": 10, + "candidate_threshold": 0.030964473262429237, + "completed": [ + { + "image_id": 1, + "sha256": "15723773a8788c9d610c874707ad2d513582322ee1d42009b0b39d3de336e10d", + "bytes": 1740, + "dimensions": [ + 128, + 96 + ], + "first_sample": 4321, + "last_sample": 932194, + "emitted_after_sample": 1229702 + }, + { + "image_id": 2, + "sha256": "66404b14ca9e3a6627871ad9d3b5dd4311915d7820a99dfa182fc3a20fbe9d85", + "bytes": 1790, + "dimensions": [ + 128, + 96 + ], + "first_sample": 932501, + "last_sample": 1860046, + "emitted_after_sample": 2157842 + }, + { + "image_id": 3, + "sha256": "166d33d72820bf3385e0d30823d991b453147c84ca75f788ab8819ad5e3c3ced", + "bytes": 1821, + "dimensions": [ + 128, + 96 + ], + "first_sample": 1860449, + "last_sample": 2787666, + "emitted_after_sample": 3050309 + } + ], + "rf_tx": 0 + }, + { + "symbol_rate": 500000, + "samples_per_symbol": 16, + "detection_decimation": 8, + "input_images": 3, + "tx_rf_packets": 24, + "accepted_rf_packets": 24, + "completed_images": 3, + "all_sha_match": true, + "packets_before_finish": 24, + "images_before_finish": 3, + "capture_seconds": 0.323744625, + "payload_airtime_seconds": 0.276768, + "processing_seconds": 55.15755260013975, + "processing_per_rf_second": 170.37364743936598, + "processing_per_payload_airtime_second": 199.29165438251442, + "coarse_seconds": 0.49211360013578087, + "strict_seconds": 54.585659996955656, + "assembly_jpeg_decode_seconds": 0.001369199831970036, + "processing_seconds_per_rf_packet": 2.2982313583391565, + "peak_iq_buffer_samples": 424309, + "iq_buffer_limit_samples": 424864, + "working_arrays_estimate_bytes": 74928288, + "memory_is_process_rss_measurement": false, + "peak_object_payload_bytes": 1821, + "candidate_count": 2374, + "peak_candidates_in_block": 2114, + "rejected_reasons": { + "PHY_HEADER_MAGIC_MISMATCH": 2298, + "TRUNCATED_SYMBOLS": 38, + "TRUNCATED_IQ_WINDOW": 14 + }, + "buffer_release_delay_upper_seconds": 0.052608, + "noisy": true, + "coded_bit_ebn0_db": 7, + "cfo_hz": 10, + "candidate_threshold": 0.030964473262429237, + "completed": [ + { + "image_id": 1, + "sha256": "15723773a8788c9d610c874707ad2d513582322ee1d42009b0b39d3de336e10d", + "bytes": 1740, + "dimensions": [ + 128, + 96 + ], + "first_sample": 4321, + "last_sample": 747682, + "emitted_after_sample": 1025782 + }, + { + "image_id": 2, + "sha256": "66404b14ca9e3a6627871ad9d3b5dd4311915d7820a99dfa182fc3a20fbe9d85", + "bytes": 1790, + "dimensions": [ + 128, + 96 + ], + "first_sample": 747989, + "last_sample": 1491022, + "emitted_after_sample": 1742690 + }, + { + "image_id": 3, + "sha256": "166d33d72820bf3385e0d30823d991b453147c84ca75f788ab8819ad5e3c3ced", + "bytes": 1821, + "dimensions": [ + 128, + 96 + ], + "first_sample": 1491425, + "last_sample": 2234130, + "emitted_after_sample": 2496773 + } + ], + "rf_tx": 0 + }, + { + "symbol_rate": 800000, + "samples_per_symbol": 10, + "detection_decimation": 5, + "input_images": 3, + "tx_rf_packets": 24, + "accepted_rf_packets": 24, + "completed_images": 3, + "all_sha_match": true, + "packets_before_finish": 24, + "images_before_finish": 3, + "capture_seconds": 0.206560625, + "payload_airtime_seconds": 0.17298, + "processing_seconds": 60.9549745998811, + "processing_per_rf_second": 295.09484007361567, + "processing_per_payload_airtime_second": 352.38163140178693, + "coarse_seconds": 0.2952118997927755, + "strict_seconds": 60.60142269870266, + "assembly_jpeg_decode_seconds": 0.0011926002334803343, + "processing_seconds_per_rf_packet": 2.539790608328379, + "peak_iq_buffer_samples": 308819, + "iq_buffer_limit_samples": 317696, + "working_arrays_estimate_bytes": 56066720, + "memory_is_process_rss_measurement": false, + "peak_object_payload_bytes": 1821, + "candidate_count": 3113, + "peak_candidates_in_block": 2707, + "rejected_reasons": { + "PHY_HEADER_MAGIC_MISMATCH": 3002, + "TRUNCATED_SYMBOLS": 87 + }, + "buffer_release_delay_upper_seconds": 0.0392125, + "noisy": true, + "coded_bit_ebn0_db": 7, + "cfo_hz": 10, + "candidate_threshold": 0.030964473262429237, + "completed": [ + { + "image_id": 1, + "sha256": "15723773a8788c9d610c874707ad2d513582322ee1d42009b0b39d3de336e10d", + "bytes": 1740, + "dimensions": [ + 128, + 96 + ], + "first_sample": 4321, + "last_sample": 470914, + "emitted_after_sample": 645226 + }, + { + "image_id": 2, + "sha256": "66404b14ca9e3a6627871ad9d3b5dd4311915d7820a99dfa182fc3a20fbe9d85", + "bytes": 1790, + "dimensions": [ + 128, + 96 + ], + "first_sample": 471221, + "last_sample": 937486, + "emitted_after_sample": 1112086 + }, + { + "image_id": 3, + "sha256": "166d33d72820bf3385e0d30823d991b453147c84ca75f788ab8819ad5e3c3ced", + "bytes": 1821, + "dimensions": [ + 128, + 96 + ], + "first_sample": 937889, + "last_sample": 1403826, + "emitted_after_sample": 1600933 + } + ], + "rf_tx": 0 + }, + { + "symbol_rate": 1000000, + "samples_per_symbol": 8, + "detection_decimation": 4, + "input_images": 3, + "tx_rf_packets": 24, + "accepted_rf_packets": 24, + "completed_images": 3, + "all_sha_match": true, + "packets_before_finish": 24, + "images_before_finish": 3, + "capture_seconds": 0.167498625, + "payload_airtime_seconds": 0.138384, + "processing_seconds": 55.921951099997386, + "processing_per_rf_second": 333.86513531079663, + "processing_per_payload_airtime_second": 404.1070578968478, + "coarse_seconds": 0.20818859990686178, + "strict_seconds": 55.66615750128403, + "assembly_jpeg_decode_seconds": 0.0011203999165445566, + "processing_seconds_per_rf_packet": 2.3300812958332244, + "peak_iq_buffer_samples": 279589, + "iq_buffer_limit_samples": 281968, + "working_arrays_estimate_bytes": 49778592, + "memory_is_process_rss_measurement": false, + "peak_object_payload_bytes": 1821, + "candidate_count": 2995, + "peak_candidates_in_block": 2361, + "rejected_reasons": { + "PHY_HEADER_MAGIC_MISMATCH": 2832, + "TRUNCATED_SYMBOLS": 96, + "TRUNCATED_IQ_WINDOW": 43 + }, + "buffer_release_delay_upper_seconds": 0.034746, + "noisy": true, + "coded_bit_ebn0_db": 7, + "cfo_hz": 10, + "candidate_threshold": 0.030964473262429237, + "completed": [ + { + "image_id": 1, + "sha256": "15723773a8788c9d610c874707ad2d513582322ee1d42009b0b39d3de336e10d", + "bytes": 1740, + "dimensions": [ + 128, + 96 + ], + "first_sample": 4321, + "last_sample": 378658, + "emitted_after_sample": 565220 + }, + { + "image_id": 2, + "sha256": "66404b14ca9e3a6627871ad9d3b5dd4311915d7820a99dfa182fc3a20fbe9d85", + "bytes": 1790, + "dimensions": [ + 128, + 96 + ], + "first_sample": 378965, + "last_sample": 752974, + "emitted_after_sample": 892978 + }, + { + "image_id": 3, + "sha256": "166d33d72820bf3385e0d30823d991b453147c84ca75f788ab8819ad5e3c3ced", + "bytes": 1821, + "dimensions": [ + 128, + 96 + ], + "first_sample": 753377, + "last_sample": 1127058, + "emitted_after_sample": 1324165 + } + ], + "rf_tx": 0 + } +] diff --git a/docs/LABS_INDEX.md b/docs/LABS_INDEX.md index 926b93d..9830da1 100644 --- a/docs/LABS_INDEX.md +++ b/docs/LABS_INDEX.md @@ -240,6 +240,41 @@ - **Далее:** программный Lab050 — профиль RX, поток видео и ограничение возраста очереди. - **RF:** RF TX = 0, STOP. +## Lab050 + +- **Цель:** последовательность JPEG через отдельный SDR-транспорт, ограничение + возраста изображения и блочный независимый IQ-приём. Управление ELRS отдельно. +- **База:** Lab049 закреплена коммитом + `cdf5396e807b83f82a264632dc46f71428fcd616`; Lab050 не добавлена в индекс и не коммитилась. +- **Переиспользовано:** планировщик Lab035 с LATEST_ONLY и завершением активного + изображения. Добавлены готовность JPEG отдельно от съёмки, произвольный предел + возраста и прогноз RX. Поздно готовый старый снимок не вытесняет свежий. +- **Профили:** A 320×240 RGB q35 — 8724 байта; B 160×120 RGB q35 — 2771; + C 128×96 RGB q25 — 1740; D 160×120 grayscale q30 — 2028. По 30 + детерминированных сдвигов исходного фото; это не видео камеры. +- **Матрица:** 240 сценариев, вход 3/5/10 кадров/с, Rs=200/400/500/800/1000 ксимв/с, + MAX_VIDEO_AGE=200/333/500/1000 мс. Ожидающая очередь ≤1. +- **Идеальная модель:** C/500k — 5 или 10 свежих кадров/с, возраст около 92,6 мс. + При камере 10 и Rs=200k, возрасте 333 мс: 14 передач, 16 замен, 4,333 свежих + кадра/с внутри окна и один дополнительный после него. +- **Потоковый RX:** 24/24 RF-пакета и 3/3 точных JPEG до `finish` на каждой скорости; + IQ-память ограничена. В шумном опыте processing/RF = 44,322 / 147,986 / + 170,374 / 295,095 / 333,865. Реального времени нет. +- **Память:** удерживаемые IQ 2,13–6,51 МиБ; расчёт массивов 47,5–143,4 МиБ, + не измеренный RSS. Состояние RX не растёт вместе с длиной сессии. +- **Достижимость:** 5/10 кадров/с без QPSK — CONDITIONAL. C≤1952 байт, + 500 ксимв/с, стоимость RX-пакета ≤16/12 мс для модели возраста 200 мс; + требуется ещё примерно 144/192-кратное ускорение относительно этого замера. +- **Код:** [видеомодель](../experiments/lab050_video_stream.py), + [потоковый RX](../experiments/lab050_streaming_receiver.py). +- **Проверки:** [44 теста Lab050](../tests/test_lab050_video_stream.py), + quick_gate — 650 тестов и 14 функциональных проверок; независимые verifier/adversary. +- **Отчёт:** [состояние](lab050_current_state.md), + [JSON](../data/processed/lab050/result.json), + [замеры RX](../data/processed/lab050/receiver_benchmarks.json). Журнал: запись 039. +- **Далее:** Lab051 — профиль и ускорение строгой обработки ложных кандидатов. +- **RF:** RF TX = 0. Lab050 commit/push не выполнялись. STOP. + ## Правило хранения данных Большие записи IQ, потоки `cs8`, бинарные объекты и другие аппаратные исходные diff --git a/docs/lab050_current_state.md b/docs/lab050_current_state.md new file mode 100644 index 0000000..416706a --- /dev/null +++ b/docs/lab050_current_state.md @@ -0,0 +1,379 @@ +# Lab050: потоковая архитектура видео и возраст изображения + +8 сентября 2026 года. SOFTWARE ONLY. RF TX = 0. Оборудование не запускалось +и не инвентаризировалось. Управление ELRS/CRSF и SDR-видео остаются раздельными. + +**Получен блочный независимый RX с ограниченным IQ-буфером и модель свежей +видеопередачи. Текущий RX не обеспечивает 5 или 10 свежих изображений/с.** +По времени радио оба уровня возможны на уменьшенном JPEG без QPSK; итог +полной системы — CONDITIONAL, с конкретными ещё не выполненными условиями ниже. + +## Фиксация Lab049 и исходный аудит + +Исходный HEAD `ac6cc792b896e34f7dd6d291955a0387fdafcf41`, ветка main. +Проверены status, diff/stat, diff/check и точный staged scope. В отдельный +коммит вошли 12 файлов Lab049: три скрипта, три файла тестов, три отчёта, +README, PROJECT_LOG и LABS_INDEX. Сгенерированные данные, `.codex`, локальные +AGENTS/CLAUDE и посторонние файлы не добавлены. + +Создан разрешённый коммит: +`cdf5396e807b83f82a264632dc46f71428fcd616` +— `Lab049: define ELRS control and SDR video architecture`. +От него выполнена Lab050; её изменения не добавлялись в индекс и не коммитились. +Push/pull/fetch не выполнялись. + +Аудит Lab034–037 и связанных модулей показал: + +| Уже существовало | Как используется в Lab050 | +|---|---| +| `image_fragments.py`, Lab046: image ID, индекс/число фрагментов, размер объекта | Неизменная фрагментация JPEG, отдельный ID каждого снимка | +| Lab034: группировка изображения и изоляция стирающего FEC | Переиспользована семантика границ изображения; старый 12+3 FEC не включён | +| Lab035, `video_frame_scheduler.py`: активный кадр, очередь ожидающих, LATEST_ONLY, прогноз завершения | Расширен существующий планировщик; новая независимая очередь не создана | +| Lab036–037: один самый свежий ожидающий кадр, завершение активного | Сохранена политика; очередь ожидающих не больше одного изображения | +| Lab048: корреляционный поиск и строгий PHY/FEC/CRC | Использованы в блочном API; чтение IQ отделено от memmap | + +Изменения кода: `protocol/video_frame_scheduler.py` получает готовность JPEG +отдельно от времени съёмки, произвольный MAX_VIDEO_AGE и прогноз задержки после +радио. Старые вызовы и все пять прежних политик сохраняют поведение. +`lab046c_offline_recovery.decode_candidate_from_memmap` принимает необязательный +читатель IQ; прежнее чтение CS8 остаётся поведением по умолчанию. +Новые файлы: [модель и эксперимент](../experiments/lab050_video_stream.py), +[блочный RX](../experiments/lab050_streaming_receiver.py), +[тесты](../tests/test_lab050_video_stream.py). + +## Политика свежести + +Активное изображение является выбранной атомарной единицей обслуживания: +после допуска оно завершается целиком. Отдельный физический пакет тем более +не прерывается. Во время передачи каждого пакета принимаются события готовности +новых JPEG и заменяется ожидающий снимок. При постоянной скорости, отсутствии +CONTROL в SDR и известном числе пакетов точный прогноз допуска позволяет +не начинать изображение, уже не укладывающееся в возраст. + +Свежесть определяется парой `(capture_timestamp, image_id)`, а не порядком +завершения кодирования. Поздно готовый старый снимок не вытесняет более свежий +и не возвращается на экран после опустошения очереди. Обнаруженный независимыми +проверяющими отказ этого правила исправлен и покрыт двумя регрессиями. + +Причины удаления различаются: `replaced_by_newest`, `older_capture_arrived_late`, +`prediction_reject`; при оценке загруженного RX возможен также отказ допуска +к обработке. Если уже нижняя оценка задержки RX делает кадр просроченным, +его отклоняет TX-допуск, до занятия эфира. Удаление снимков — нормальная +политика видео, а не потеря обязательной управляющей команды. + +Планировщик здесь остаётся событийной моделью с известными входами и историей +для отчёта. Его массив входов/результатов зависит от числа исследуемых кадров; +**логическая ожидающая очередь** ограничена одним снимком. Готового адаптера +живой камеры с ограниченным журналом пока нет. Независимость памяти от длины +сессии реализована и проверена непосредственно для IQ-приёмника. + +## Профили JPEG и источник + +Источник — реальное фото `data/raw/lab011_scene.jpg`. Для каждого профиля +созданы 30 исходных кадров: первый повторяет исходную подготовку, остальные +получены детерминированным циклическим сдвигом `(7i mod 61, 3i mod 31)` пикселей +до уменьшения. Это искусственная последовательность одного фото, не съёмка камеры. +Разрешение, цвет и quality фиксированы; размеры файлов не подбирались вручную. +Во всех профилях subsampling=2, optimize=False, progressive=False. + +| Профиль | Разрешение | Цвет | Quality | Первый JPEG, байт | Диапазон 30 JPEG, байт | PSNR первого, dB | +|---|---|---|---:|---:|---:|---:| +| [A](../data/processed/lab050/profile_A.jpg) | 320×240 | RGB | 35 | 8724 | 8724–9250 | 28,12 | +| [B](../data/processed/lab050/profile_B.jpg) | 160×120 | RGB | 35 | 2771 | 2771–3031 | 28,11 | +| [C](../data/processed/lab050/profile_C.jpg) | 128×96 | RGB | 25 | 1740 | 1740–1893 | 27,17 | +| [D](../data/processed/lab050/profile_D.jpg) | 160×120 | grayscale | 30 | 2028 | 2028–2233 | 30,15 | + +PSNR = 10 log10(255²/MSE); сравнение с несжатым изображением **того же +разрешения и цветового режима** после преобразования. Оно характеризует +JPEG-искажение внутри профиля, не потерю деталей при уменьшении и не +пригодность картинки для управления. Между RGB и grayscale эти числа не +доказывают превосходство качества. Четыре выходных JPEG визуально проверены; +субъективный вердикт пригодности для оператора не присваивался. + +Первый A побайтно совпал с Lab047, SHA-256: +`e913dab9ac32795250880a4be67fbbbf89a4f96536842476ab591d1acaafc02a`. +Исходный контрольный JPEG не изменён. Перекодирование новых профилей разрешено +постановкой Lab050; новый видеокодек не разрабатывался. + +## Точная ёмкость нынешнего PHY + +Один RF-пакет переносит 244 байта JPEG плюс 12 байт метаданных фрагмента, +дополняется до 256 байт нагрузки packet и до 268 байт вместе с packet header/CRC. +FEC R=1/2 K=7, шесть хвостовых бит, перемежитель, PHY header, PRBS7 и BPSK +сохранены. RF-пакет занимает 5724 передаваемых бита и 42 дополнительных +символьных интервала, итого **5766**. Проверено по настоящему плану Lab045. + +Для JPEG размера J: N=ceil(J/244), T_RF=N×5766/Rs. +Точный максимальный размер для f кадров/с: +`Jmax = 244 × floor(Rs/(5766×f))`. +Это ступенчатая граница с добивкой, не деление полезного битрейта на восемь. + +| Rs, ксимв/с | Выборок/символ при 8 Мвыб/с | Jmax для 3 кадров/с | Jmax для 5 кадров/с | Jmax для 10 кадров/с | +|---:|---:|---:|---:|---:| +| 200 | 40 | 2684 | 1464 | 732 | +| 400 | 20 | 5612 | 3172 | 1464 | +| 500 | 16 | 6832 | 4148 | 1952 | +| 800 | 10 | 11224 | 6588 | 3172 | +| 1000 | 8 | 13908 | 8296 | 4148 | + +Все размеры в байтах. На пределе нет запаса для потерь, пауз, объёмных данных +или роста JPEG. Никакие ELRS CONTROL/TELEMETRY из этого бюджета не вычитаются. +Физическая стоимость старого LinkPacket используется лишь как эквивалентная +шкала времени существующего симулятора: его длительность принудительно равна +5766/Rs. LinkPacket не добавлен в реальную PHY-упаковку Lab046. + +Максимум изображений/с по времени радио для **первого** JPEG каждого профиля: + +| Rs, ксимв/с | A, 36 пакетов | B, 12 пакетов | C, 8 пакетов | D, 9 пакетов | +|---:|---:|---:|---:|---:| +| 200 | 0,964 | 2,891 | 4,336 | 3,854 | +| 400 | 1,927 | 5,781 | 8,672 | 7,708 | +| 500 | 2,409 | 7,226 | 10,839 | 9,635 | +| 800 | 3,854 | 11,562 | 17,343 | 15,416 | +| 1000 | 4,818 | 14,453 | 21,679 | 19,270 | + +По всей последовательности A использует до 38 пакетов, B до 13, D до 10; +C остаётся в восьми пакетах. Именно C удобен для следующего воспроизводимого +опыта. Профиль B также укладывается по радио в 5 кадров/с на 400 ксимв/с. +JPEG A не достигает 5 кадров/с даже на 1 Мсимв/с: 8724 > 8296 байт. + +## Матрица очередей и фактический смысл FPS + +Выполнено 240 сценариев: четыре профиля × входные 3/5/10 кадров/с × пять Rs +× пределы возраста 200/333/500/1000 мс. По 30 входных изображений в каждом. +Идеальный пакетный путь сериализует packet, проверяет CRC и собирает точные +JPEG. Стоимость FEC/поиска IQ в этой матрице сначала равна нулю **по модели**; +её реальный замер выполнен отдельно, ниже. Ошибок радио в матрице нет. + +Окно наблюдения H=30/input_fps: 10, 6 или 3 с. Основные RAW RECEIVED FPS и +FRESH DISPLAY FPS считают только события внутри H. Полученные при дальнейшем +опустошении кадры показываются отдельно. Свежим считается только показанный +снимок с возрастом ≤MAX_VIDEO_AGE. Просроченное изображение на экран не выпускается, +поэтому raw display FPS совпадает с fresh display FPS; повторный показ старого +изображения не засчитывается новым обновлением. + +| Модель | Возраст | Вход | Передано | Удалено до TX | Получено / показано всего | Fresh FPS в H | После H | +|---|---:|---:|---:|---:|---:|---:|---:| +| C, камера 5, Rs=500k | 200 мс | 30 | 30 | 0 | 30 / 30 | **5,000** | 0 | +| C, камера 10, Rs=500k | 200 мс | 30 | 30 | 0 | 30 / 30 | **10,000** | 0 | +| B, камера 5, Rs=400k | 200 мс | 30 | 30 | 0 | 30 / 30 | **5,000** | 0 | +| C, камера 10, Rs=200k | 200 мс | 30 | 0 | 30, прогноз возраста | 0 / 0 | **0** | 0 | +| C, камера 10, Rs=200k | 333 мс | 30 | 14 | 16, замена свежим | 14 / 14 | **4,333** | 1 | +| A, камера 5, Rs=1000k | 200 мс | 30 | 0 | 30, прогноз возраста | 0 / 0 | **0** | 0 | + +Для перегруженного C/10/200k при 333 мс возраст показанных снимков примерно +231–329 мс, ожидающая очередь ≤1. Увеличение предела до 500/1000 мс в данном +сценарии не меняет 14 передач и 16 замен; оно не создаёт дополнительную ёмкость. +При 200 мс уже само радио требует 230,64 мс и все снимки обоснованно отклонены. + +Результат C/5/500k: IMAGE AGE min/median/max = **92,563 / 92,590 / 92,629 мс**. +Это возраст **в модели идеального RX и мгновенного экрана**, не камера→монитор. +Вход 30, transmitted=received=displayed=30, dropped=0, H=6 с, +RAW RECEIVED FPS=RAW DISPLAY FPS=FRESH DISPLAY FPS=5. + +## Блочный приёмник + +```mermaid +flowchart LR + I[IQ-блок до 65536 выборок] --> B[Ограниченный буфер и хвост] + B --> C[Грубый поиск в принадлежащем блоку интервале] + C --> S[Штатная строгая проверка PHY/FEC/CRC] + S --> P[Выдача принятого пакета из push_iq_block] + P --> A[Сборщик текущего image ID] + A --> J[JPEG decode; далее проверка возраста и экран] +``` + +`StreamingReceiver.push_iq_block()` принимает непрерывные complex64-блоки +без границ TX, payload, ожидаемых ID или числа кадров. `finish()` проверяет +остаток, не дополняет недостающую запись нулями. Выдача пакетов не ждёт конца +сессии. Глобальные индексы сохраняются; принятая область исключается из +повторной строгой проверки. Список кандидатов относится лишь к одному блоку. +Внутри RX нет истории принятых пакетов; вызывающий код должен сразу их забирать. + +Размер удерживаемого хвоста определяется максимально допустимым PHY packet +1024 байта, а не известными передатчику границами. Сохраняются левый запас +уточнения, опорная последовательность, счётчики и текущие IQ. Порог Lab047 +`0.030964473262429237` не меняется и не подстраивается по успеху JPEG. +Скалярная статистика корреляций не объявляется новой калибровкой фона. + +Децимация 20 старого поиска непригодна как универсальная константа для +высоких скоростей. В адаптере она равна min(20, samples_per_symbol//2): +**20/10/8/5/4** для 200/400/500/800/1000 ксимв/с, оставляя две выборки/символ. +Меняются параметры поиска, но не FEC, CRC, модуляция или порог. +Константы старых лабораторий восстанавливаются после вызова; это +последовательный программный адаптер, не потокобезопасная многопоточная библиотека. + +IQ-блоки должны быть непрерывными. При будущих потерях блоков драйвера нужно +явно сбросить незавершённый поиск/объект и обозначить разрыв; live-подключение, +обработка таких разрывов и управление переполнением драйвера здесь не реализованы. +Синхронный `push` создаёт обратное давление программному источнику; при текущей +скорости RX настоящий источник 8 Мвыб/с потребовал бы потерь/переполнения. + +## Интеграционный опыт IQ и производительность + +На **каждой** из пяти скоростей поданы три JPEG C, 24 физических пакета, +с неизвестным началом, межкадровыми интервалами и границами входных блоков. +CS8-квантование передатчика сохранено. Шум один для всех: coded-bit Eb/N0=7 dB, +seed=50050, остаточная частота +10 Гц, фаза 0,73 рад и чередование полярности. +После добавления шума поток complex64 не переквантован. Фон также генерируется +детерминированно. TX создаёт лишь очередной RF-пакет, целая IQ-запись не хранится. + +**Все пять проходов: 24/24 пакета, 3/3 JPEG, точный SHA; все JPEG выданы до finish.** +Сборщик принимает ID/индексы только из проверенных CRC пакетов, сопоставление +с TX-хешами происходит после сборки. В этом опыте ID монотонны в одной сессии; +wrap/restart видеосеанса не реализован. Разрешён только текущий объект, +ограничены его размер (64 КиБ) и число фрагментов; журнал результатов опыта +отделён от памяти RX. Дополнительно пройдены интеграционные тесты идеального +канала на всех скоростях, границ двух блоков, усечённого потока и длинного фона. + +| Rs, ксимв/с | IQ, с | Обработка, с | Обработка / RF-секунда | Кандидатов строгой проверки | Оценка массивов RX, МиБ | +|---:|---:|---:|---:|---:|---:| +| 200 | 0,792483 | 35,124167 | **44,322** | 2191 | 143,412 | +| 400 | 0,401868 | 59,470863 | **147,986** | 2575 | 83,450 | +| 500 | 0,323745 | 55,157553 | **170,374** | 2374 | 71,457 | +| 800 | 0,206561 | 60,954975 | **295,095** | 3113 | 53,469 | +| 1000 | 0,167499 | 55,921951 | **333,865** | 2995 | 47,473 | + +Знаменатель — все поданные IQ, включая фон/хвост. В JSON отдельно есть время +чистой передачи полезных RF-пакетов и соответствующее ещё худшее отношение. +CPU-замер включает push/finish, сборку и JPEG decode; генерация TX/IQ, +преобразование исходной фотографии и сравнение TX-хешей исключены. +Это однократные настольные wall-clock замеры, не норматив CPU и не оценка +предельной дальности. Они не сравниваются напрямую с Lab048 на другой записи. + +На 500 ксимв/с грубый поиск занял 0,492 с, строгая обработка — 54,586 с. +Из 2374 кандидатов приняты 24; большинство остальных отклонено по magic +PHY-заголовка. Новый ограниченный буфер **не устранил** стоимость ложных +кандидатов. Увеличение Rs не делает этот приёмник быстрее реального времени. +Факт Lab048 «15 с → 13,966 с» остаётся характеристикой прежней записи. + +Пять тяжёлых проходов выполнены один раз. Первый запуск после сохранения +замеров остановился на несовпадении API `build_packet/parse_packet` в модели +очереди; вызовы исправлены и проверены тестами. Итоговая матрица построена +с `--reuse-benchmarks`, без повторения RF/IQ-замеров. Режим проверяет скорости, +параметры шума/порог и хеши восстановленных контрольных объектов. + +## Память + +Предел блока 65536 выборок. Для каждого Rs буфер ограничен выражением: +`core + maximum_PHY_waveform + refinement_margin + left_margin + input_block`. +Число кадров в сессии в формулу не входит. В тесте 100 последовательных +блоков сохраняется тот же предел и не появляется история IQ. + +| Rs, ксимв/с | Измеренный пик удерживаемых IQ, выборок | Проверяемый предел, выборок | IQ в пике, МиБ | +|---:|---:|---:|---:| +| 200 | 853027 | 853556 | 6,508 | +| 400 | 493587 | 496316 | 3,766 | +| 500 | 424309 | 424864 | 3,237 | +| 800 | 308819 | 317696 | 2,356 | +| 1000 | 279589 | 281968 | 2,133 | + +С запасом на копирование, грубый и строгий поиск расчёт массивов даёт +47,5–143,4 МиБ против исторической оценки Lab048 около 1,44 ГиБ. +Это **расчёт массивов, не измеренный RSS процесса**: Python/SciPy, аллокатор, +кэши и вызывающая программа добавляют память. История отчёта и заранее +подготовленные 30 JPEG относятся к эксперименту, не к обязательному IQ-состоянию. + +## Задержка и условие достижимости + +Для каждого снимка JSON хранит capture, encode_complete, queue_enter, +first_tx, last_tx, first_rx, object_complete, decode_complete и display. +У не переданных/не допущенных стадий стоит null, причина удаления явна. +QUEUE AGE=first_tx−queue_enter; RF AGE=last_tx−first_tx; +END-TO-END IMAGE AGE=display−capture. + +| Слагаемое | Статус | +|---|---| +| T_CAPTURE | Реальная экспозиция/считывание неизвестны; источник задан событиями, длительность моделируется нулём | +| T_ENCODE | Измерено CPU-кодирование подготовленных изображений; C медиана около 0,143 мс; уменьшение/сдвиг вне этого замера | +| T_QUEUE | Рассчитано существующим планировщиком; не измерение живой очереди | +| T_RF | Рассчитано по настоящему формату; C на 500k — 92,256 мс | +| T_RX_PROCESSING | Измерено в IQ-опыте, включая взаимодействие с буфером; в модели подставляется отдельно как стоимость пакета | +| T_DECODE | Измерено CPU JPEG decode; C медиана около 0,187 мс; в общем IQ-замере уже включено, повторно туда не прибавляется | +| T_DISPLAY | Экран не подключён; длительность моделируется нулём | + +В конвейере RF и обработка ранних пакетов перекрываются. Нельзя прибавлять +полный CPU service time к полному RF-времени и называть это измеренной +сквозной задержкой. Прогноз без предыдущей очереди RX после last_tx: +`lookahead + p + (N−1)×max(0,p−T_packet) + JPEG_decode`, где p — стоимость +пакета. Формула независимо проверена последовательным расчётом 32 наборов +событий и допуском по обе стороны точной границы возраста. + +На 500k консервативный предел выдачи из буфера без стоимости CPU — 52,608 мс. +Средняя измеренная стоимость полного RX-опыта — 2,298 с на переданный RF-пакет +(включает фон и отвергнутых кандидатов, не только принятый FEC). При подстановке +этой стоимости все 30 кадров C/5fps отклоняются ещё до TX при любом проверенном +MAX_VIDEO_AGE; **FRESH DISPLAY FPS=0**. Это прогноз с измеренной стоимостью, +не эксперимент с живым экраном. + +Проверены также два явно условных варианта ускоренного RX: + +| Условие | 5 кадров/с | 10 кадров/с | +|---|---:|---:| +| Выбранный JPEG | ≤1952 байт, 8 пакетов; C фактически ≤1893 | То же | +| Выбранный Rs | 500 ксимв/с, 16 выборок/символ | То же | +| Нижняя граница только по радио для 8 пакетов | 230,640 ксимв/с | 461,280 ксимв/с | +| Предположенная стоимость RX-пакета | ≤16 мс | ≤12 мс | +| Ускорение относительно данного замера | **≥143,64 раза** | **≥191,52 раза** | +| MAX_VIDEO_AGE | 200 мс | 200 мс | +| Расчётный возраст min/median/max | 192,447 / 192,474 / 192,513 мс | 160,447 / 160,474 / 160,513 мс | +| Свежих в исходном окне | 30/6 с = **5** | 29/3 с = **9,667** | +| С учётом опустошения | 30; без дополнительных | 30; один после 3 с | +| Частота по интервалам последовательных показов | около 5 | около 10 | +| FEASIBLE | **CONDITIONAL** | **CONDITIONAL** | + +Для 10 кадров/с стартовая задержка превышает период камеры, поэтому первый +трёхсекундный отчёт честно даёт 9,667, а не 10 с включением позднего кадра. +Частота около 10 относится к установившимся интервалам. Условия не означают, +что такое ускорение достигнуто или достижимо только небольшой правкой. +Живые capture/display, входной jitter, потери и запас CPU ещё должны уложиться +в оставшийся бюджет. Для 5 кадров/с запас возраста этого примера лишь около 7,5 мс. + +Итак, **по пропускной способности BPSK достаточно**, без QPSK и нового FEC; +**с нынешним измеренным приёмником — нет**. Аппаратная работоспособность +HackRF на этих Rs в данном задании не проверялась. 3 кадра/с остаются +промежуточным уровнем, цель проекта >3, ориентиры 5 и 10. + +## ELRS и следующий опыт + +CONTROL LINK OK/LOST — внешнее состояние для контроллера ровера. Видеопланировщик +вообще не принимает его как радиопакет. Проверка на существующем TwoStageFailsafe: +при отсутствии новых команд 100 мс → замедление, 200 мс → торможение, +а результат видеомодели остаётся тем же. Это проверка разделения моделей, +не готовое сопряжение CRSF. Safety/session/emergency-модули не менялись. + +Рекомендуемый Lab051 — программное профилирование и ускорение строгого +обработчика ложных кандидатов на сохранённом детерминированном сценарии +C/500k, с отдельной проверкой фоновых ложных принятий и неизменными FEC/CRC. +Измерять стоимость уточнения, оценки частоты, заголовка и FEC; сохранить +независимость от TX и корректность на границах блоков. Не объявлять 5 кадров/с +до полного непрерывного опыта со свежими JPEG и устойчивым временем обработки. +RF, QPSK, новый кодек, 12+3 FEC и перенос на 180 МГц этим этапом не разрешаются. + +## Воспроизведение + +```powershell +python experiments/lab050_video_stream.py +# Только пересчёт матрицы с уже сохранёнными локальными замерами IQ: +python experiments/lab050_video_stream.py --reuse-benchmarks +python -m pytest tests/test_lab050_video_stream.py -q -p no:cacheprovider +python tools/quick_gate.py +python -m pytest tests/ -q +git diff --check +``` + +Полный первый запуск занимает минуты; новые RF-сеансы в нём отсутствуют. +[result.json](../data/processed/lab050/result.json) содержит 240 сценариев, +все временные отметки, профили, расчёты ёмкости и отдельные условные модели. +[receiver_benchmarks.json](../data/processed/lab050/receiver_benchmarks.json) +содержит исходные пять IQ-замеров. Большие IQ-файлы не создавались. + +44 целевых теста проходят. Verifier/adversary подтвердили исправление +поздно завершённого старого кадра; verifier независимо проверил границы +ёмкости, 60 сценариев очереди, 32 прогноза RX и арифметику IQ-замеров. +Итог: quick_gate PASS — 650 тестов и 14 функциональных проверок; +полный pytest — 650 passed за 78,73 с, одно предупреждение о доступе +к `.pytest_cache`; git diff --check PASS. Подробности — в записи 039 журнала. + +LAB050 GIT ADD = NOT PERFORMED. LAB050 COMMIT = NOT PERFORMED. +PUSH = NOT PERFORMED. RF TX = 0. STOP. diff --git a/experiments/lab046c_offline_recovery.py b/experiments/lab046c_offline_recovery.py index fee9a70..05cf2df 100644 --- a/experiments/lab046c_offline_recovery.py +++ b/experiments/lab046c_offline_recovery.py @@ -639,10 +639,13 @@ def decode_candidate_from_memmap( sample_count: int, candidate: CoarseCandidate, detection_number: int, + *, + read_iq_window: Callable | None = None, ) -> CandidateOutcome: """Строго проверить один candidate локальными IQ-окнами.""" started = time.perf_counter() + window_reader = lab046c._cs8_window if read_iq_window is None else read_iq_window reference = lab046c.lab044._preamble_reference_samples() probe_symbols = lab046c.lab045.PREAMBLE_BIT_COUNT + lab046c.lab045.PHY_HEADER_BIT_COUNT probe_waveform = ( @@ -655,7 +658,7 @@ def decode_candidate_from_memmap( sample_count, candidate.sample_offset + probe_waveform + REFINEMENT_MARGIN_SAMPLES, ) - probe_iq = lab046c._cs8_window(raw, probe_start, probe_end) + probe_iq = window_reader(raw, probe_start, probe_end) maximum_window = int(probe_iq.size) local_coarse = candidate.sample_offset - probe_start try: @@ -740,7 +743,7 @@ def decode_candidate_from_memmap( maximum_window=maximum_window, started=started, ) - full_iq = lab046c._cs8_window(raw, refined_global, full_end) + full_iq = window_reader(raw, refined_global, full_end) maximum_window = max(maximum_window, int(full_iq.size)) full = lab046b._extract_soft_window(full_iq, 0, frame_symbols) if not full.passed: diff --git a/experiments/lab050_streaming_receiver.py b/experiments/lab050_streaming_receiver.py new file mode 100644 index 0000000..d8a6bbd --- /dev/null +++ b/experiments/lab050_streaming_receiver.py @@ -0,0 +1,163 @@ +"""Lab050: независимый приём IQ-блоков с ограниченной памятью, без RF. + +Использует корреляцию и строгий PHY/FEC/CRC Lab048. Принятые пакеты сразу +возвращаются вызывающему коду; история IQ/пакетов внутри не накапливается. +Адаптер констант старых лабораторий рассчитан на последовательный вызов. +""" + +from __future__ import annotations + +from collections import Counter +from contextlib import contextmanager +from dataclasses import dataclass +import math +import time + +import numpy as np + +from experiments import lab046c_offline_recovery as offline +from experiments import lab049_bpsk_symbol_rate as phy +from experiments.lab049_independent_iq import receiver_mode + + +@contextmanager +def streaming_mode(mode: phy.PhyMode): + """Оставить не менее двух выборок/символ в грубом поиске.""" + module = phy.lab045.lab044.lab043 + original = module.DETECTION_DECIMATION + with receiver_mode(mode): + module.DETECTION_DECIMATION = min(20, mode.samples_per_symbol // 2) + try: + yield + finally: + module.DETECTION_DECIMATION = original + + +@dataclass(frozen=True) +class ReceivedPacket: + start_sample: int + end_sample: int + emitted_after_sample: int + payload: bytes + + +class StreamingReceiver: + """push_iq_block выдаёт пакеты до конца потока; finish только опустошает хвост. + + Размер кадра для удержания IQ определяется максимальной допустимой PHY + length (1024 байта), а не TX-границами/числом ожидаемых пакетов. + Порог Lab047 остаётся фиксированным. Источник и размеры объекта неизвестны. + """ + + def __init__(self, symbol_rate: int, *, block_samples: int = 65_536): + self.mode = phy.PhyMode(symbol_rate) + if self.mode.samples_per_symbol < 4: + raise ValueError("Для этого адаптера нужно не менее четырёх выборок/символ") + if type(block_samples) is not int or block_samples < 1024: + raise ValueError("Размер блока должен быть целым и не меньше 1024") + self.block_samples = block_samples + self.decimation = min(20, self.mode.samples_per_symbol // 2) + self.core_samples = math.ceil(block_samples / self.decimation) * self.decimation + margin = offline.REFINEMENT_MARGIN_SAMPLES + self.margin = math.ceil(margin / self.decimation) * self.decimation + maximum_symbols = (phy.lab045.PREAMBLE_BIT_COUNT + phy.lab045.PHY_HEADER_BIT_COUNT + + phy.lab045.conv.encoded_bit_count(phy.lab045.MAX_PACKET_BYTES * 8) + + 2 * phy.lab045.GUARD_SYMBOL_COUNT + phy.lab045.RRC_SPAN_SYMBOLS) + self.lookahead = maximum_symbols * self.mode.samples_per_symbol + self.margin + self.buffer_limit_samples = self.core_samples + self.lookahead + self.margin + block_samples + self.buffer = np.empty(0, np.complex64) + self.base_sample = self.scan_sample = self.total_samples = self.accepted_until = 0 + self.peak_buffer_samples = self.peak_candidate_count = 0 + self.packet_count = self.candidate_count = self.blocks_processed = 0 + self.processing_seconds = self.coarse_seconds = self.strict_seconds = 0.0 + self.score_count = 0 + self.score_sum = 0.0 + self.rejected_reasons: Counter = Counter() + self.closed = False + with streaming_mode(self.mode): + self.reference = offline.lab046b._decimate_then_downconvert( + phy.lab045.lab044._preamble_reference_samples()).copy() + + @property + def memory_estimate_bytes(self) -> int: + """Консервативная оценка массивов, не измеренный RSS Python/SciPy.""" + window_bytes = self.buffer_limit_samples * np.dtype(np.complex64).itemsize + # Два буфера при копировании + прежний коэффициент временных массивов + # Lab048 + запас для строгого matched filter и индексов кандидатов. + return 2 * window_bytes + 12 * window_bytes + 8 * window_bytes + self.reference.nbytes + + def _scan(self, owned_end: int) -> list[ReceivedPacket]: + started = time.perf_counter() + scores = offline._correlation_scores(self.buffer, self.reference) + self.coarse_seconds += time.perf_counter() - started + indexes = offline._all_local_peak_indexes(scores) + offsets = self.base_sample + indexes * self.decimation + indexes = indexes[(offsets >= self.scan_sample) & (offsets < owned_end)] + self.score_count += len(indexes) + self.score_sum += float(np.sum(scores[indexes], dtype=np.float64)) + indexes = indexes[scores[indexes] >= np.float32(phy.lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD)] + self.peak_candidate_count = max(self.peak_candidate_count, len(indexes)) + emitted = [] + for index in indexes: + absolute = self.base_sample + int(index) * self.decimation + if absolute < self.accepted_until: + continue + self.candidate_count += 1 + started = time.perf_counter() + outcome = offline.decode_candidate_from_memmap( + self.buffer.view(np.float32), self.buffer.size, + offline.CoarseCandidate(int(index) * self.decimation, float(scores[index])), + self.candidate_count, + read_iq_window=lambda raw, start, end: raw[2*start:2*end].view(np.complex64), + ) + self.strict_seconds += time.perf_counter() - started + if outcome.accepted and outcome.crc == phy.lab045.CRC_PASS and outcome.frame is not None: + end = self.base_sample + outcome.next_sample + self.accepted_until = end + self.packet_count += 1 + emitted.append(ReceivedPacket( + self.base_sample + outcome.refined_start_sample, end, self.total_samples, + bytes(outcome.frame.reception.payload_bytes), + )) + else: + self.rejected_reasons[outcome.reason] += 1 + self.blocks_processed += 1 + self.scan_sample = owned_end + retain_start = max(0, self.scan_sample - self.margin) + trim = retain_start - self.base_sample + if trim > 0: + self.buffer = self.buffer[trim:].copy() + self.base_sample = retain_start + return emitted + + def push_iq_block(self, samples: np.ndarray) -> list[ReceivedPacket]: + if self.closed: + raise RuntimeError("Поток уже закрыт") + samples = np.asarray(samples, dtype=np.complex64) + if samples.ndim != 1 or samples.size > self.block_samples or not np.all(np.isfinite(samples)): + raise ValueError("Нужен конечный одномерный IQ-блок не больше block_samples") + started = time.perf_counter() + self.total_samples += samples.size + self.buffer = np.concatenate((self.buffer, samples)) + self.peak_buffer_samples = max(self.peak_buffer_samples, self.buffer.size) + if self.buffer.size > self.buffer_limit_samples: + raise RuntimeError("Превышен предел буфера IQ") + emitted = [] + with streaming_mode(self.mode): + while self.total_samples >= self.scan_sample + self.core_samples + self.lookahead: + emitted.extend(self._scan(self.scan_sample + self.core_samples)) + self.processing_seconds += time.perf_counter() - started + return emitted + + def finish(self) -> list[ReceivedPacket]: + if self.closed: + raise RuntimeError("Поток уже закрыт") + started = time.perf_counter() + emitted = [] + with streaming_mode(self.mode): + if self.scan_sample < self.total_samples: + emitted = self._scan(self.total_samples) + self.buffer = np.empty(0, np.complex64) + self.closed = True + self.processing_seconds += time.perf_counter() - started + return emitted diff --git a/experiments/lab050_video_stream.py b/experiments/lab050_video_stream.py new file mode 100644 index 0000000..15f6192 --- /dev/null +++ b/experiments/lab050_video_stream.py @@ -0,0 +1,470 @@ +"""Lab050: JPEG-профили, свежесть видеопотока и блочный RX без оборудования.""" + +from __future__ import annotations + +import argparse +from collections import Counter +from dataclasses import asdict, dataclass +import hashlib +from io import BytesIO +import json +import math +from pathlib import Path +import statistics +import sys +import time + +if __package__ in (None, ""): + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import numpy as np +from PIL import Image, ImageChops + +from experiments import lab046_bpsk_object_transport as transport +from experiments import lab049_bpsk_symbol_rate as phy +from experiments.lab050_streaming_receiver import StreamingReceiver +from protocol import packet as packet_protocol +from protocol.link_packet import Direction, LinkPacket, TrafficClass, HEADER_SIZE +from protocol.video_age_policy import AgePolicyPacket +from protocol.video_frame_scheduler import FramePolicy, VideoFrameGroup, schedule_video_frames + + +ROOT = Path(__file__).resolve().parents[1] +OUTPUT = ROOT / "data/processed/lab050" +SYMBOL_RATES = (200_000, 400_000, 500_000, 800_000, 1_000_000) +INPUT_RATES = (3, 5, 10) +AGE_LIMITS = (0.200, 0.333, 0.500, 1.000) +FRAME_INTERVALS = 5766 +FRAGMENT_BYTES = transport.FRAGMENT_DATA_SIZE + + +@dataclass(frozen=True) +class JpegProfile: + name: str + width: int + height: int + color_mode: str + quality: int + + +PROFILES = ( + JpegProfile("A", 320, 240, "RGB", 35), + JpegProfile("B", 160, 120, "RGB", 35), + JpegProfile("C", 128, 96, "RGB", 25), + JpegProfile("D", 160, 120, "L", 30), +) + + +@dataclass(frozen=True) +class EncodedImage: + image_id: int + jpeg: bytes + encode_seconds: float + decode_seconds: float + psnr_db: float + + +def fragment_count(jpeg_bytes: int) -> int: + if type(jpeg_bytes) is not int or jpeg_bytes <= 0: + raise ValueError("Размер JPEG должен быть положительным целым") + return (jpeg_bytes + FRAGMENT_BYTES - 1) // FRAGMENT_BYTES + + +def maximum_jpeg_bytes(symbol_rate: int, fps: int) -> int: + """Точный ступенчатый предел с добивкой последнего физического кадра.""" + phy.PhyMode(symbol_rate) + if type(fps) is not int or fps <= 0: + raise ValueError("Частота изображений должна быть положительным целым") + return (symbol_rate // (fps * FRAME_INTERVALS)) * FRAGMENT_BYTES + + +def encode_sequence(source: Image.Image, profile: JpegProfile, count: int = 30) -> list[EncodedImage]: + """Детерминированные сдвиги фотографии; это не видео реальной камеры.""" + if count < 1: + raise ValueError("Нужна непустая последовательность") + sequence = [] + for index in range(count): + # Первый кадр совпадает с подготовкой Lab047. Остальные — циклический + # сдвиг исходного фото до уменьшения; амплитуда задана, не подобрана по байтам. + shifted = source if index == 0 else ImageChops.offset(source, (index * 7) % 61, (index * 3) % 31) + reference = shifted.convert(profile.color_mode).resize( + (profile.width, profile.height), Image.Resampling.LANCZOS) + started = time.perf_counter() + stream = BytesIO() + reference.save(stream, format="JPEG", quality=profile.quality, subsampling=2, + optimize=False, progressive=False) + jpeg = stream.getvalue() + encode_seconds = time.perf_counter() - started + started = time.perf_counter() + with Image.open(BytesIO(jpeg)) as image: + image.load() + decoded = np.asarray(image).copy() + decode_seconds = time.perf_counter() - started + difference = np.asarray(reference, dtype=np.float64) - decoded.astype(np.float64) + mean_square = float(np.mean(difference * difference)) + psnr = 10 * math.log10(255**2 / mean_square) if mean_square else 999.0 + sequence.append(EncodedImage(index + 1, jpeg, encode_seconds, decode_seconds, psnr)) + return sequence + + +def make_frame_groups(sequence: list[EncodedImage], input_fps: int) -> list[VideoFrameGroup]: + """Использовать существующий планировщик; LinkPacket здесь только модель времени. + + В эфире остаются packet/fragment Lab046, без LinkPacket и без стирающего FEC. + Все элементы планировщика имеют 256 байт, его эквивалентная скорость ниже + выбирается так, чтобы каждый занимал ровно 5766/Rs секунд. + """ + if input_fps <= 0: + raise ValueError("Частота камеры должна быть положительной") + frames = [] + order = 0 + for index, image in enumerate(sequence): + capture_us = round(index / input_fps * 1_000_000) + ready_us = capture_us + math.ceil(image.encode_seconds * 1_000_000) + fragments = transport.fragments.split_image_bytes(image.jpeg, image.image_id, FRAGMENT_BYTES) + packets = [] + for fragment in fragments: + payload = transport.fragments.encode_image_fragment(fragment) + payload += bytes(256 - len(payload)) + model_packet = LinkPacket(TrafficClass.VIDEO, Direction.ROVER_TO_GROUND, + 50, order, capture_us, 0, payload) + packets.append(AgePolicyPacket(model_packet, order, ready_us, image.image_id)) + order += 1 + frames.append(VideoFrameGroup(image.image_id, capture_us, tuple(packets), ready_us)) + return frames + + +def simulate_sequence(sequence: list[EncodedImage], input_fps: int, symbol_rate: int, + max_video_age: float, *, rx_seconds_per_packet: float = 0.0, + receiver_lookahead_seconds: float = 0.0) -> dict: + """Очередь и идеальная доставка пакетов, не измерение эфира/камеры/экрана. + + Измеренная стоимость RX может быть подставлена как отдельная оценка: + последовательный обработчик пакетов, устаревший объект не допускается + к обработке. Эта оценка не заменяет живой IQ-прогон. + """ + phy.PhyMode(symbol_rate) + if not math.isfinite(rx_seconds_per_packet) or rx_seconds_per_packet < 0: + raise ValueError("Нужна конечная неотрицательная стоимость RX") + if not math.isfinite(receiver_lookahead_seconds) or receiver_lookahead_seconds < 0: + raise ValueError("Нужна конечная неотрицательная задержка буфера RX") + frames = make_frame_groups(sequence, input_fps) + image_lookup = {image.image_id: image for image in sequence} + equivalent_bitrate = (256 + HEADER_SIZE) * 8 * symbol_rate / FRAME_INTERVALS + def minimum_post_rf_latency(frame: VideoFrameGroup) -> float: + # Идеальный конвейер RF -> последовательный RX без предыдущего backlog. + # Если уже эта нижняя оценка не укладывается в возраст, TX бесполезен. + packet_time = FRAME_INTERVALS / symbol_rate + return (receiver_lookahead_seconds + rx_seconds_per_packet + + (len(frame.packets)-1) * max(0.0, rx_seconds_per_packet-packet_time) + + image_lookup[frame.composite_frame_id].decode_seconds) + schedule = schedule_video_frames( + frames, (), FramePolicy.LATEST_ONLY, equivalent_bitrate, + max_video_age_seconds=max_video_age, + completion_latency_seconds=minimum_post_rf_latency, + ) + arrivals: dict[int, list] = {} + for sent in schedule.transmitted: + arrivals.setdefault(sent.item.composite_frame_id, []).append(sent) + frame_lookup = {frame.composite_frame_id: frame for frame in frames} + horizon = len(sequence) / input_fps + rx_cursor = 0.0 + displayed = [] + traces = [] + received = 0 + discarded_rx = 0 + for frame_id, sent in arrivals.items(): + frame = frame_lookup[frame_id] + image = image_lookup[frame_id] + first_tx, last_tx = sent[0].start_seconds, sent[-1].end_seconds + # Идеальный packet-level путь действительно проверяет сериализованный + # CRC и собирает JPEG; FEC/IQ проверяется отдельным интеграционным опытом. + assembler = transport.ObjectAssembler() + for item in sent: + wire = packet_protocol.build_packet(item.item.packet.payload, + phy.lab045.MESSAGE_TYPE, phy.lab045.SEQUENCE_NUMBER) + parsed = packet_protocol.parse_packet(wire) + assembler.offer(transport.recover_fragment_from_payload(parsed.payload)) + if assembler.recovered_object() != image.jpeg: + raise RuntimeError("Идеальная сборка изменила JPEG") + first_rx = sent[0].end_seconds + predicted_end = rx_cursor + for item in sent: + predicted_end = max(predicted_end, item.end_seconds + receiver_lookahead_seconds) + rx_seconds_per_packet + predicted_end += image.decode_seconds + dropped = predicted_end - frame.generation_time_seconds > max_video_age + 1e-12 + if dropped: + discarded_rx += 1 + object_complete = decode_complete = display_time = None + else: + received += 1 + rx_cursor = predicted_end + decode_complete = predicted_end + object_complete = decode_complete - image.decode_seconds + display_time = decode_complete # Только модель мгновенного экрана. + displayed.append((frame_id, display_time, display_time - frame.generation_time_seconds)) + traces.append({ + "image_id": frame_id, "capture_timestamp": frame.generation_time_seconds, + "encode_complete_timestamp": frame.available_time_seconds, + "queue_enter_timestamp": frame.available_time_seconds, + "first_tx_timestamp": first_tx, "last_tx_timestamp": last_tx, + "first_rx_timestamp": first_rx, "object_complete_timestamp": object_complete, + "decode_complete_timestamp": decode_complete, "display_timestamp": display_time, + "queue_age_seconds": first_tx - frame.available_time_seconds, + "rf_age_seconds": last_tx - first_tx, + "end_to_end_image_age_seconds": None if display_time is None else display_time - frame.generation_time_seconds, + "dropped_at_rx_admission": dropped, + }) + dropped_before = {drop.frame.composite_frame_id: drop for drop in schedule.dropped_frames} + for frame in frames: + if frame.composite_frame_id not in dropped_before: + continue + drop = dropped_before[frame.composite_frame_id] + traces.append({"image_id": frame.composite_frame_id, + "capture_timestamp": frame.generation_time_seconds, + "encode_complete_timestamp": frame.available_time_seconds, + "queue_enter_timestamp": frame.available_time_seconds, + **{key: None for key in ("first_tx_timestamp", "last_tx_timestamp", "first_rx_timestamp", + "object_complete_timestamp", "decode_complete_timestamp", "display_timestamp", + "queue_age_seconds", "rf_age_seconds", "end_to_end_image_age_seconds")}, + "drop_reason": drop.reason, "drop_timestamp": drop.drop_time_seconds}) + active_received = sum(row["object_complete_timestamp"] is not None + and row["object_complete_timestamp"] <= horizon + 1e-12 for row in traces) + active_displayed = [item for item in displayed if item[1] <= horizon + 1e-12] + ages = [item[2] for item in displayed] + drops = Counter(drop.reason for drop in schedule.dropped_frames) + return { + "scope": "packet-level ideal RF model, measured JPEG CPU durations; display/camera not measured", + "input_fps": input_fps, "symbol_rate": symbol_rate, "max_video_age_seconds": max_video_age, + "input_frames": len(sequence), "transmitted_frames": len(arrivals), + "dropped_before_tx": len(schedule.dropped_frames), "drop_reasons": dict(drops), + "rf_delivered_objects": len(arrivals), "received_frames": received, + "dropped_at_rx_admission": discarded_rx, + "displayed_frames": len(displayed), "maximum_waiting_frames": schedule.maximum_waiting_frames, + "observation_seconds": horizon, "active_received_frames": active_received, + "active_displayed_frames": len(active_displayed), + "raw_received_fps": active_received / horizon, + "raw_display_fps": len(active_displayed) / horizon, + "fresh_display_fps": len(active_displayed) / horizon, + "steady_fresh_display_fps": ((len(displayed)-1)/(displayed[-1][1]-displayed[0][1]) + if len(displayed) > 1 else 0.0), + "drain_displayed_frames": len(displayed) - len(active_displayed), + "image_age_min_seconds": min(ages) if ages else None, + "image_age_median_seconds": statistics.median(ages) if ages else None, + "image_age_max_seconds": max(ages) if ages else None, + "rx_cost_seconds_per_packet_modeled": rx_seconds_per_packet, + "receiver_lookahead_seconds_modeled": receiver_lookahead_seconds, + "elrs_control_in_video_queue": False, + "frames": sorted(traces, key=lambda row: row["image_id"]), + } + + +def benchmark_receiver(sequence: list[EncodedImage], symbol_rate: int, *, noisy: bool = True) -> dict: + """Подать несколько JPEG частями IQ без TX-метаданных в RX, замерить CPU. + + TX хранит лишь очередной физический кадр, RX — несколько блоков. + Проверка SHA производится после независимой выдачи и сборки объекта. + """ + mode = phy.PhyMode(symbol_rate) + receiver = StreamingReceiver(symbol_rate) + random_generator = np.random.default_rng(50_050) + completed = [] + current_assembler = None + current_id = None + last_completed_id = -1 + object_first_sample = None + assembly_decode_seconds = 0.0 + retained_object_bytes_peak = 0 + packets_before_finish = 0 + emitted_jpegs_before_finish = 0 + + def consume(packets): + nonlocal current_assembler, current_id, last_completed_id, object_first_sample + nonlocal assembly_decode_seconds, retained_object_bytes_peak + started = time.perf_counter() + for packet in packets: + fragment = transport.recover_fragment_from_payload(packet.payload) + if fragment is None or fragment.original_size > 65536 or fragment.fragment_count > 269: + continue + # Для этой конечной сессии ID возрастают, без wrap/restart. Строгий + # transport assembler защищает согласованность индексов/дубликатов. + if fragment.image_id <= last_completed_id or (current_id is not None and fragment.image_id < current_id): + continue + if current_id != fragment.image_id: + current_id = fragment.image_id + current_assembler = transport.ObjectAssembler() + object_first_sample = packet.start_sample + current_assembler.offer(fragment) + retained_object_bytes_peak = max(retained_object_bytes_peak, fragment.original_size) + recovered = current_assembler.recovered_object() + if recovered is not None: + with Image.open(BytesIO(recovered)) as image: + image.load() + dimensions = image.size + completed.append({"image_id": current_id, "sha256": hashlib.sha256(recovered).hexdigest(), + "bytes": len(recovered), "dimensions": dimensions, + "first_sample": object_first_sample, "last_sample": packet.end_sample, + "emitted_after_sample": packet.emitted_after_sample}) + last_completed_id = current_id + current_id = current_assembler = None + assembly_decode_seconds += time.perf_counter() - started + + def push_array(values): + for offset in range(0, values.size, receiver.block_samples): + consume(receiver.push_iq_block(values[offset:offset + receiver.block_samples])) + + # Ненулевые несогласованные с символом интервалы, неизвестные RX. + push_array((random_generator.normal(0, 0.2, 4321) + 1j * random_generator.normal(0, 0.2, 4321)).astype(np.complex64)) + transmitted_packets = 0 + air_samples = 0 + for image in sequence: + for frame in transport.build_object_frames(image.jpeg, image.image_id): + cs8 = phy.build_parameterized_tx_cs8(frame.plan, mode) + components = np.frombuffer(cs8, dtype=np.int8).astype(np.float32) + values = components[0::2] + 1j * components[1::2] + air_samples += values.size + if noisy: + indexes = np.arange(values.size, dtype=np.float64) + values = values * np.exp(1j * (0.73 + 2*np.pi*10*indexes/mode.sample_rate_hz)) * (-1 if transmitted_packets % 2 else 1) + energy = float(np.sum(abs(values)**2)) / frame.plan.frame_symbol_count + sigma = math.sqrt(energy / (2 * 10**(7/10))) + values += sigma * (random_generator.normal(size=values.size) + 1j * random_generator.normal(size=values.size)) + push_array(np.asarray(values, np.complex64)) + transmitted_packets += 1 + gap = 211 + (transmitted_packets * 137) % 1000 + push_array((random_generator.normal(0, 0.2, gap) + 1j * random_generator.normal(0, 0.2, gap)).astype(np.complex64)) + # Хвост — ограниченный поток фона, не вся запись. Позволяет проверить + # выдачу последнего JPEG именно из push, до finish. + left = receiver.lookahead + receiver.core_samples + while left: + count = min(left, receiver.block_samples) + push_array((random_generator.normal(0, 0.2, count) + 1j * random_generator.normal(0, 0.2, count)).astype(np.complex64)) + left -= count + packets_before_finish = receiver.packet_count + emitted_jpegs_before_finish = len(completed) + consume(receiver.finish()) + expected = {image.image_id: hashlib.sha256(image.jpeg).hexdigest() for image in sequence} + exact = {item["image_id"]: item["sha256"] for item in completed} == expected + processing = receiver.processing_seconds + assembly_decode_seconds + capture_seconds = receiver.total_samples / mode.sample_rate_hz + return { + "symbol_rate": symbol_rate, "samples_per_symbol": mode.samples_per_symbol, + "detection_decimation": receiver.decimation, "input_images": len(sequence), + "tx_rf_packets": transmitted_packets, "accepted_rf_packets": receiver.packet_count, + "completed_images": len(completed), "all_sha_match": exact, + "packets_before_finish": packets_before_finish, "images_before_finish": emitted_jpegs_before_finish, + "capture_seconds": capture_seconds, "payload_airtime_seconds": air_samples / mode.sample_rate_hz, + "processing_seconds": processing, "processing_per_rf_second": processing / capture_seconds, + "processing_per_payload_airtime_second": processing / (air_samples / mode.sample_rate_hz), + "coarse_seconds": receiver.coarse_seconds, "strict_seconds": receiver.strict_seconds, + "assembly_jpeg_decode_seconds": assembly_decode_seconds, + "processing_seconds_per_rf_packet": processing / transmitted_packets, + "peak_iq_buffer_samples": receiver.peak_buffer_samples, + "iq_buffer_limit_samples": receiver.buffer_limit_samples, + "working_arrays_estimate_bytes": receiver.memory_estimate_bytes + 2 * 65536, + "memory_is_process_rss_measurement": False, + "peak_object_payload_bytes": retained_object_bytes_peak, + "candidate_count": receiver.candidate_count, "peak_candidates_in_block": receiver.peak_candidate_count, + "rejected_reasons": dict(receiver.rejected_reasons), + "buffer_release_delay_upper_seconds": (receiver.lookahead + 2*receiver.core_samples) / mode.sample_rate_hz, + "noisy": noisy, "coded_bit_ebn0_db": 7 if noisy else None, "cfo_hz": 10 if noisy else 0, + "candidate_threshold": phy.lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD, + "completed": completed, "rf_tx": 0, + } + + +def run_functional_tests() -> None: + assert maximum_jpeg_bytes(500_000, 5) == 4148 + assert maximum_jpeg_bytes(1_000_000, 10) == 4148 + assert fragment_count(244) == 1 and fragment_count(245) == 2 + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--reuse-benchmarks", action="store_true", help="Переиспользовать текущие локальные замеры IQ") + args = parser.parse_args() + run_functional_tests() + OUTPUT.mkdir(parents=True, exist_ok=True) + source_path = ROOT / "data/raw/lab011_scene.jpg" + with Image.open(source_path) as source: + source.load() + sequences = {profile.name: encode_sequence(source, profile) for profile in PROFILES} + control = sequences["A"][0].jpeg + if hashlib.sha256(control).hexdigest() != phy.EXPECTED_JPEG_SHA256: + raise RuntimeError("Профиль A не воспроизвёл контрольный JPEG Lab047") + summaries = [] + for profile in PROFILES: + sequence = sequences[profile.name] + (OUTPUT / f"profile_{profile.name}.jpg").write_bytes(sequence[0].jpeg) + summaries.append({**asdict(profile), "first_jpeg_bytes": len(sequence[0].jpeg), + "min_jpeg_bytes": min(len(item.jpeg) for item in sequence), + "max_jpeg_bytes": max(len(item.jpeg) for item in sequence), + "first_psnr_db": sequence[0].psnr_db, + "median_encode_seconds": statistics.median(item.encode_seconds for item in sequence), + "median_decode_seconds": statistics.median(item.decode_seconds for item in sequence), + "sha256_first": hashlib.sha256(sequence[0].jpeg).hexdigest(), + "airtime_capacity": {str(rate): rate/(fragment_count(len(sequence[0].jpeg))*FRAME_INTERVALS) for rate in SYMBOL_RATES}}) + print(json.dumps(summaries, ensure_ascii=False), flush=True) + benchmark_path = OUTPUT / "receiver_benchmarks.json" + if args.reuse_benchmarks: + benchmarks = json.loads(benchmark_path.read_text(encoding="utf-8")) + expected_hashes = {image.image_id: hashlib.sha256(image.jpeg).hexdigest() for image in sequences["C"][:3]} + if [row["symbol_rate"] for row in benchmarks] != list(SYMBOL_RATES): + raise ValueError("Сохранены замеры других скоростей") + for row in benchmarks: + if (row["input_images"] != 3 or row["coded_bit_ebn0_db"] != 7 + or row["candidate_threshold"] != phy.lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD + or any(expected_hashes.get(item["image_id"]) != item["sha256"] for item in row["completed"])): + raise ValueError("Сохранённые замеры относятся к другому сценарию") + else: + benchmarks = [] + for rate in SYMBOL_RATES: + print(f"IQ benchmark {rate}: 3 JPEG profile C", flush=True) + result = benchmark_receiver(sequences["C"][:3], rate) + benchmarks.append(result) + print(json.dumps({key: result[key] for key in ("symbol_rate", "completed_images", "all_sha_match", "processing_per_rf_second", "working_arrays_estimate_bytes")}), flush=True) + benchmark_path.write_text(json.dumps(benchmarks, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + matrix = [] + receiver_estimates = [] + for profile in PROFILES: + for fps in INPUT_RATES: + for rate in SYMBOL_RATES: + for age in AGE_LIMITS: + row = simulate_sequence(sequences[profile.name], fps, rate, age) + matrix.append({"profile": profile.name, **row}) + for measured in benchmarks: + if not measured["all_sha_match"]: + continue + for age in AGE_LIMITS: + receiver_estimates.append({"profile": "C", **simulate_sequence(sequences["C"], 5, + measured["symbol_rate"], age, rx_seconds_per_packet=measured["processing_seconds_per_rf_packet"], + receiver_lookahead_seconds=measured["buffer_release_delay_upper_seconds"])}) + result = { + "stage": "Lab050 software video stream", "lab049_commit": "cdf5396e807b83f82a264632dc46f71428fcd616", + "source_sha256": hashlib.sha256(source_path.read_bytes()).hexdigest(), + "sequence": "30 deterministic cyclic shifts of one project photo, not camera video", + "profiles": summaries, + "maximum_jpeg_bytes": {str(rate): {str(fps): maximum_jpeg_bytes(rate, fps) for fps in INPUT_RATES} for rate in SYMBOL_RATES}, + "receiver_benchmarks": benchmarks, "packet_model_matrix": matrix, + "measured_rx_cost_model": receiver_estimates, + "conditional_accelerated_models": [ + {"profile": "C", "is_current_receiver_performance": False, + "required_speedup_vs_measured_cost": next(row["processing_seconds_per_rf_packet"] + for row in benchmarks if row["symbol_rate"] == 500000) / cost, + **simulate_sequence(sequences["C"], fps, 500000, .2, rx_seconds_per_packet=cost, + receiver_lookahead_seconds=next(row["buffer_release_delay_upper_seconds"] + for row in benchmarks if row["symbol_rate"] == 500000))} + for fps, cost in ((5, .016), (10, .012)) + ], + "modeled_unknown_latencies": {"capture": "unknown, modeled zero", "display": "unknown, modeled zero", + "encode": "JPEG encode CPU measured; source preparation outside timing", + "decode": "JPEG decode CPU measured", "rf": "calculated waveform airtime"}, + "elrs_control_path": "SEPARATE", "rf_tx": 0, + } + (OUTPUT / "result.json").write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + selected = next(row for row in matrix if row["profile"] == "C" and row["input_fps"] == 5 and row["symbol_rate"] == 500000 and row["max_video_age_seconds"] == .2) + print(json.dumps({key: value for key, value in selected.items() if key != "frames"}, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/protocol/video_frame_scheduler.py b/protocol/video_frame_scheduler.py index cbd0288..64f94b7 100644 --- a/protocol/video_frame_scheduler.py +++ b/protocol/video_frame_scheduler.py @@ -4,7 +4,8 @@ from __future__ import annotations from dataclasses import dataclass from enum import Enum -from typing import Iterable +import math +from typing import Callable, Iterable from protocol.link_packet import TrafficClass, encode_link_packet from protocol.priority_scheduler import transmission_duration_seconds @@ -35,6 +36,12 @@ class VideoFrameGroup: composite_frame_id: int generation_time_us: int packets: tuple[AgePolicyPacket, ...] + available_time_us: int | None = None + + @property + def available_time_seconds(self) -> float: + """Готовность закодированного изображения отдельно от момента съёмки.""" + return (self.generation_time_us if self.available_time_us is None else self.available_time_us) / 1_000_000.0 @property def generation_time_seconds(self) -> float: @@ -90,6 +97,7 @@ class FrameScheduleResult: started_frame_ids: tuple[int, ...] completed_frame_ids: tuple[int, ...] admissions: tuple[FrameAdmission, ...] + maximum_waiting_frames: int = 0 def _replace_state( @@ -195,13 +203,23 @@ def schedule_video_frames( high_priority_packets: Iterable[AgePolicyPacket], policy: FramePolicy, channel_bitrate_bps: float, + *, + max_video_age_seconds: float | None = None, + completion_latency_seconds: float | Callable[[VideoFrameGroup], float] = 0.0, ) -> FrameScheduleResult: """Run strict-priority service while keeping video frames contiguous.""" policy = FramePolicy(policy) if channel_bitrate_bps <= 0.0: raise ValueError("channel bitrate must be positive") - frame_arrivals = tuple(sorted(frames, key=lambda frame: (frame.generation_time_seconds, frame.composite_frame_id))) + if max_video_age_seconds is not None and (not math.isfinite(max_video_age_seconds) or max_video_age_seconds <= 0): + raise ValueError("Возраст видео должен быть конечным положительным числом") + if not callable(completion_latency_seconds) and (not math.isfinite(completion_latency_seconds) or completion_latency_seconds < 0): + raise ValueError("Задержка после RF должна быть конечной и неотрицательной") + frame_arrivals = tuple(sorted(frames, key=lambda frame: (frame.available_time_seconds, frame.composite_frame_id))) + for frame in frame_arrivals: + if not frame.packets or frame.available_time_seconds < frame.generation_time_seconds: + raise ValueError("Нужны непустые изображения, готовые не раньше съёмки") high_arrivals = tuple(sorted(high_priority_packets, key=lambda item: (item.available_time_seconds, item.arrival_order))) ready_high: list[AgePolicyPacket] = [] pending_frames: list[VideoFrameGroup] = [] @@ -214,15 +232,17 @@ def schedule_video_frames( actual_by_frame: dict[int, float] = {} active: VideoFrameGroup | None = None active_index = 0 + maximum_waiting = 0 + newest_capture_key: tuple[int, int] | None = None cursor = 0.0 frame_index = high_index = 0 blocker_by_order: dict[int, tuple[AgePolicyPacket, float]] = {} - def drop_frame(frame: VideoFrameGroup, reason: str) -> None: - dropped.append(FrameDrop(frame, cursor, reason)) + def drop_frame(frame: VideoFrameGroup, reason: str, when: float | None = None) -> None: + dropped.append(FrameDrop(frame, cursor if when is None else when, reason)) def admit(now: float, active_packet: AgePolicyPacket | None = None, active_end: float = 0.0) -> None: - nonlocal frame_index, high_index + nonlocal frame_index, high_index, maximum_waiting, newest_capture_key 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 @@ -232,12 +252,19 @@ def schedule_video_frames( max(0.0, active_end - item.available_time_seconds), ) _replace_state(ready_high, item, replacements) - while frame_index < len(frame_arrivals) and frame_arrivals[frame_index].generation_time_seconds <= now + TIME_EPSILON_SECONDS: + while frame_index < len(frame_arrivals) and frame_arrivals[frame_index].available_time_seconds <= now + TIME_EPSILON_SECONDS: frame = frame_arrivals[frame_index] frame_index += 1 if policy is FramePolicy.LATEST_ONLY: + capture_key = (frame.generation_time_us, frame.composite_frame_id) + # Позднее завершение кодирования не делает старый снимок новым. + # Помним свежесть и когда очередь уже опустела/новый кадр активен. + if newest_capture_key is not None and capture_key < newest_capture_key: + drop_frame(frame, "older_capture_arrived_late", frame.available_time_seconds) + continue + newest_capture_key = capture_key for old in pending_frames: - drop_frame(old, "replaced_by_newest") + drop_frame(old, "replaced_by_newest", frame.available_time_seconds) pending_frames[:] = [frame] elif policy is FramePolicy.TWO_WAITING: pending_frames.append(frame) @@ -245,6 +272,7 @@ def schedule_video_frames( drop_frame(pending_frames.pop(0), "waiting_limit") else: pending_frames.append(frame) + maximum_waiting = max(maximum_waiting, len(pending_frames)) while ( frame_index < len(frame_arrivals) @@ -256,7 +284,7 @@ def schedule_video_frames( 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) + next_times.append(frame_arrivals[frame_index].available_time_seconds) if high_index < len(high_arrivals): next_times.append(high_arrivals[high_index].available_time_seconds) cursor = max(cursor, min(next_times)) @@ -273,7 +301,7 @@ def schedule_video_frames( if active is None: while pending_frames and active is None: candidate = pending_frames.pop(0) - deadline = policy.deadline_seconds + deadline = max_video_age_seconds if max_video_age_seconds is not None else policy.deadline_seconds if deadline is not None: future_high = high_arrivals[high_index:] predicted = predict_frame_completion( @@ -283,7 +311,11 @@ def schedule_video_frames( tuple(ready_high), future_high, ) - if predicted - candidate.generation_time_seconds > deadline + TIME_EPSILON_SECONDS: + post_latency = (completion_latency_seconds(candidate) if callable(completion_latency_seconds) + else completion_latency_seconds) + if not math.isfinite(post_latency) or post_latency < 0: + raise ValueError("Прогноз задержки должен быть конечным и неотрицательным") + if predicted + post_latency - candidate.generation_time_seconds > deadline + TIME_EPSILON_SECONDS: drop_frame(candidate, "prediction_reject") continue predicted_by_frame[candidate.composite_frame_id] = predicted @@ -327,4 +359,5 @@ def schedule_video_frames( started_frame_ids=tuple(started), completed_frame_ids=tuple(completed), admissions=admissions, + maximum_waiting_frames=maximum_waiting, ) diff --git a/tests/test_lab050_video_stream.py b/tests/test_lab050_video_stream.py new file mode 100644 index 0000000..f304b2c --- /dev/null +++ b/tests/test_lab050_video_stream.py @@ -0,0 +1,176 @@ +"""Свежесть изображений и непрерывный IQ, без файлов эксперимента и RF.""" + +from __future__ import annotations + +import numpy as np +import pytest + +from experiments import lab049_bpsk_symbol_rate as phy +from experiments import lab050_video_stream as lab +from experiments.lab050_streaming_receiver import StreamingReceiver +from protocol.two_stage_failsafe import TwoStageFailsafe, ThresholdPolicy, DecelerationProfile, SafetyState + + +def sequence(size: int = 1740, count: int = 30) -> list[lab.EncodedImage]: + return [lab.EncodedImage(index+1, bytes([index % 256])*size, .001, .001, 0.0) for index in range(count)] + + +@pytest.mark.parametrize("rate,fps,maximum", [(500000,3,6832),(500000,5,4148),(500000,10,1952), + (800000,3,11224),(800000,5,6588),(800000,10,3172), + (1000000,3,13908),(1000000,5,8296),(1000000,10,4148)]) +def test_maximum_jpeg_is_exact_at_fragment_boundary(rate, fps, maximum): + assert lab.maximum_jpeg_bytes(rate, fps) == maximum + assert lab.fragment_count(maximum) * 5766 * fps <= rate + assert lab.fragment_count(maximum+1) * 5766 * fps > rate + + +@pytest.mark.parametrize("fps", [3, 5, 10]) +@pytest.mark.parametrize("size", [244, 245, 1740, 8724]) +def test_newest_queue_is_bounded_and_accounting_conserves_inputs(fps, size): + result = lab.simulate_sequence(sequence(size), fps, 200000, 1.0) + assert result["maximum_waiting_frames"] <= 1 + assert result["input_frames"] == result["transmitted_frames"] + result["dropped_before_tx"] + assert result["received_frames"] == result["transmitted_frames"] + assert result["displayed_frames"] == result["received_frames"] + assert result["image_age_max_seconds"] is None or result["image_age_max_seconds"] <= 1.0 + + +def test_newer_unstarted_frame_replaces_old_without_damaging_active_image(): + result = lab.simulate_sequence(sequence(), 10, 200000, 1.0) + assert result["drop_reasons"]["replaced_by_newest"] > 0 + sent = [frame for frame in result["frames"] if frame["first_tx_timestamp"] is not None] + assert all(frame["rf_age_seconds"] == pytest.approx(8*5766/200000) for frame in sent) + assert all(frame["last_tx_timestamp"] <= next_frame["first_tx_timestamp"] for frame,next_frame in zip(sent,sent[1:])) + + +@pytest.mark.parametrize("limit", [.2,.333,.5,1.0]) +def test_age_limit_predictively_drops_impossible_jpeg(limit): + result = lab.simulate_sequence(sequence(8724), 5, 200000, limit) + assert result["transmitted_frames"] == 0 + assert result["dropped_before_tx"] == 30 + assert result["fresh_display_fps"] == 0 + + +def test_source_time_not_encoding_completion_defines_age(): + images=[lab.EncodedImage(1,bytes(244),.3,.001,0)] + result = lab.simulate_sequence(images,5,1000000,.2) + assert result["transmitted_frames"] == 0 + + +def test_late_old_encoding_does_not_replace_fresher_waiting_image(): + images = [lab.EncodedImage(index+1,bytes(1740),delay,.001,0) + for index,delay in enumerate((.001,.12,.001))] + result=lab.simulate_sequence(images,10,200000,1.0) + shown=[row["image_id"] for row in result["frames"] if row["display_timestamp"] is not None] + assert shown==[1,3] + assert result["drop_reasons"]["older_capture_arrived_late"]==1 + + +def test_late_old_encoding_does_not_regress_display_after_queue_drains(): + images = [lab.EncodedImage(index+1,bytes(244),delay,.001,0) + for index,delay in enumerate((.5,.001,.001))] + result=lab.simulate_sequence(images,5,1000000,1.0) + shown=[row["image_id"] for row in result["frames"] if row["display_timestamp"] is not None] + assert shown==[2,3] + assert result["drop_reasons"]["older_capture_arrived_late"]==1 + + +def test_five_fresh_images_per_second_and_all_timestamps(): + result = lab.simulate_sequence(sequence(),5,500000,.2) + assert result["fresh_display_fps"] == 5 + assert result["raw_display_fps"] == 5 + assert result["image_age_max_seconds"] == pytest.approx(.001+8*5766/500000+.001) + for row in result["frames"]: + names = ["capture_timestamp","encode_complete_timestamp","queue_enter_timestamp","first_tx_timestamp", + "first_rx_timestamp","last_tx_timestamp","object_complete_timestamp","decode_complete_timestamp","display_timestamp"] + values = [row[name] for name in names] + assert values == sorted(values) + + +def test_slow_rx_drops_before_cpu_and_does_not_claim_reception(): + result = lab.simulate_sequence(sequence(),5,500000,.2,rx_seconds_per_packet=.1) + assert result["rf_delivered_objects"] == 0 + assert result["received_frames"] == result["displayed_frames"] == 0 + assert result["dropped_before_tx"] == 30 + + +@pytest.mark.parametrize("fps,cost", [(5,.016),(10,.012)]) +def test_conditional_receiver_speed_meets_freshness_and_rate(fps,cost): + result=lab.simulate_sequence(sequence(),fps,500000,.2,rx_seconds_per_packet=cost, + receiver_lookahead_seconds=.052608) + assert result["fresh_display_fps"] == (5 if fps==5 else 29/3) + assert result["steady_fresh_display_fps"] == pytest.approx(fps) + assert result["displayed_frames"] == 30 + assert result["drain_displayed_frames"] == (0 if fps==5 else 1) + assert result["image_age_max_seconds"] < .2 + + +def test_sufficient_cpu_throughput_alone_does_not_guarantee_age(): + # 8*.02=.16 с/изображение укладывается в период .2 с, но сквозной возраст + # с накоплением нужного хвоста IQ уже превышает .2 с. + result=lab.simulate_sequence(sequence(),5,500000,.2,rx_seconds_per_packet=.02, + receiver_lookahead_seconds=.052608) + assert result["transmitted_frames"] == 0 + + +def test_elrs_loss_is_external_to_video_queue(): + before=lab.simulate_sequence(sequence(count=3),5,500000,.2) + safety=TwoStageFailsafe(ThresholdPolicy("test",.1,.2),DecelerationProfile("test",1,2)) + assert safety.apply_watchdog(.2) is SafetyState.STAGE2_BRAKING + after=lab.simulate_sequence(sequence(count=3),5,500000,.2) + assert before==after + assert not after["elrs_control_in_video_queue"] + + +@pytest.mark.parametrize("rate", [200000,400000,500000,800000,1000000]) +def test_stream_crosses_blocks_and_emits_before_finish(rate): + payload=bytes(range(256)) + plan=phy.lab045.build_transmission_plan(payload) + receiver=StreamingReceiver(rate) + waveform=phy.build_parameterized_tx_complex(plan,phy.PhyMode(rate)).astype(np.complex64) + signal=np.concatenate((np.zeros(63217,np.complex64),waveform, + np.zeros(receiver.lookahead+2*receiver.core_samples,np.complex64))) + outputs=[] + chunk=32749 + for start in range(0,len(signal),chunk): + outputs.extend(receiver.push_iq_block(signal[start:start+chunk])) + assert receiver.buffer.size <= receiver.buffer_limit_samples + assert [item.payload for item in outputs] == [payload] + assert outputs[0].start_sample == pytest.approx(63217,abs=100) + assert not receiver.finish() + assert receiver.buffer.size == 0 + + +def test_memory_bound_for_long_session_without_retained_history(): + receiver=StreamingReceiver(1000000) + block=np.zeros(receiver.block_samples,np.complex64) + limits=[] + for _ in range(100): + assert receiver.push_iq_block(block)==[] + limits.append(receiver.buffer.size) + assert max(limits) <= receiver.buffer_limit_samples + assert receiver.total_samples > 20*receiver.buffer_limit_samples + assert receiver.packet_count == 0 + assert not receiver.finish() + + +def test_truncated_stream_does_not_accept_partial_frame(): + plan=phy.lab045.build_transmission_plan(bytes(256)) + receiver=StreamingReceiver(1000000) + waveform=phy.build_parameterized_tx_complex(plan,phy.PhyMode(1000000)).astype(np.complex64) + receiver.push_iq_block(waveform[:len(waveform)//2]) + assert receiver.finish()==[] + with pytest.raises(RuntimeError): receiver.push_iq_block(np.zeros(1,np.complex64)) + + +def test_input_limit_and_nonfinite_rejected_without_advancing_stream(): + receiver=StreamingReceiver(1000000) + for block in [np.zeros(receiver.block_samples+1,np.complex64),np.array([complex(float("nan"),0)])]: + with pytest.raises(ValueError): receiver.push_iq_block(block) + assert receiver.total_samples==0 + + +def test_fixed_formats_match_actual_plan(): + frame=lab.transport.build_object_frames(bytes(1740))[0] + assert frame.plan.frame_symbol_count+42==lab.FRAME_INTERVALS + assert lab.transport.FRAGMENT_DATA_SIZE==244