""" Lab027D. Сравнение синхронного BASE + ROI при 2 и 3 fps. Лабораторная формирует четыре синхронных профиля с различными частотой обновления и JPEG Quality. Все JPEG существуют только в памяти. Первый проход измеряет фактический payload, второй проход повторяет то же расписание и записывает сравнительное preview-видео 2x2. Лучший профиль программой не выбирается: итоговый выбор должен сделать пользователь после просмотра динамического сравнения. """ from __future__ import annotations import csv from dataclasses import dataclass from pathlib import Path import cv2 import numpy as np SOURCE_VIDEO_PATH = Path("data/raw/lab026_rover_source.mp4") OUTPUT_DIRECTORY = Path("data/processed/lab027d") CSV_PATH = OUTPUT_DIRECTORY / "lab027d_profiles.csv" REPORT_PATH = OUTPUT_DIRECTORY / "lab027d_report.txt" PREVIEW_DIRECTORY = Path("data/raw/lab027d_previews") MP4_PREVIEW_PATH = ( PREVIEW_DIRECTORY / "lab027d_fps_quality_preview.mp4" ) AVI_PREVIEW_PATH = ( PREVIEW_DIRECTORY / "lab027d_fps_quality_preview.avi" ) ROI_X_MIN = 0.20 ROI_X_MAX = 0.80 ROI_Y_MIN = 0.42 ROI_Y_MAX = 1.00 PANEL_WIDTH = 640 PANEL_HEIGHT = 360 OUTPUT_WIDTH = PANEL_WIDTH * 2 OUTPUT_HEIGHT = PANEL_HEIGHT * 2 OUTPUT_FPS = 30.0 SERVICE_OVERHEAD_FACTOR = 1.10 FEC_RATE_TWO_THIRDS = 2.0 / 3.0 FEC_RATE_ONE_HALF = 0.5 FRAME_TIME_EPSILON_SECONDS = 1e-9 NEW_LABEL_DURATION_SECONDS = 0.15 CSV_FIELD_NAMES = [ "profile_name", "base_width", "base_height", "fps", "base_quality", "roi_width", "roi_height", "roi_quality", "source_duration_s", "selected_base_frames", "selected_roi_frames", "synchronized_composite_frames", "total_base_bytes", "total_roi_bytes", "total_payload_bytes", "mean_base_frame_bytes", "mean_roi_frame_bytes", "mean_composite_frame_bytes", "p95_composite_frame_bytes", "max_composite_frame_bytes", "base_bitrate_kbps", "roi_bitrate_kbps", "total_payload_bitrate_kbps", "channel_rate_fec_2_3_kbps", "channel_rate_fec_1_2_kbps", "timestamp_mismatch_count", "frame_id_mismatch_count", ] @dataclass(frozen=True) class PreviewProfile: """ Описывает один синхронный профиль BASE + ROI. """ profile_name: str fps: float base_width: int base_height: int base_quality: int roi_width: int roi_height: int roi_quality: int @property def period_seconds(self) -> float: """ Возвращает период атомарного обновления профиля. """ return 1.0 / self.fps @dataclass class SyncState: """ Хранит последнее атомарно опубликованное состояние профиля. """ next_composite_time: float = 0.0 last_composite_update_time: float = float("-inf") composite_frame_id: int = -1 source_frame_index: int = -1 timestamp: float = 0.0 latest_base: np.ndarray | None = None latest_roi: np.ndarray | None = None selected_base_frames: int = 0 selected_roi_frames: int = 0 synchronized_composite_frames: int = 0 timestamp_mismatch_count: int = 0 frame_id_mismatch_count: int = 0 @dataclass(frozen=True) class ProfileStatistics: """ Содержит все измерения одного профиля для CSV и отчёта. """ profile_name: str base_width: int base_height: int fps: float base_quality: int roi_width: int roi_height: int roi_quality: int source_duration_s: float selected_base_frames: int selected_roi_frames: int synchronized_composite_frames: int total_base_bytes: int total_roi_bytes: int total_payload_bytes: int mean_base_frame_bytes: float mean_roi_frame_bytes: float mean_composite_frame_bytes: float p95_composite_frame_bytes: float max_composite_frame_bytes: int base_bitrate_kbps: float roi_bitrate_kbps: float total_payload_bitrate_kbps: float channel_rate_fec_2_3_kbps: float channel_rate_fec_1_2_kbps: float timestamp_mismatch_count: int frame_id_mismatch_count: int Measurements = dict[str, dict[str, list[int]]] def read_video_metadata( source_path: Path, ) -> tuple[int, int, float, int, float, int]: """ Читает метаданные исходного видео без изменения файла. """ if not source_path.exists(): raise FileNotFoundError( f"Исходное видео отсутствует: {source_path}" ) capture = cv2.VideoCapture(str(source_path)) if not capture.isOpened(): raise RuntimeError( f"OpenCV не смог открыть видео: {source_path}" ) try: width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = float(capture.get(cv2.CAP_PROP_FPS)) frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT)) finally: capture.release() if width <= 0 or height <= 0: raise RuntimeError("Некорректное разрешение исходного видео.") if fps <= 0.0 or frame_count <= 0: raise RuntimeError( "Некорректные FPS или число кадров исходного видео." ) duration_seconds = frame_count / fps file_size_bytes = source_path.stat().st_size if duration_seconds <= 0.0 or file_size_bytes <= 0: raise RuntimeError( "Некорректная длительность или размер исходного видео." ) return ( width, height, fps, frame_count, duration_seconds, file_size_bytes, ) def build_profiles() -> list[PreviewProfile]: """ Создаёт ровно четыре профиля, заданных для Lab027D. """ profiles = [ PreviewProfile( profile_name="sync_2fps_base_q20_roi_q30", fps=2.0, base_width=240, base_height=135, base_quality=20, roi_width=320, roi_height=180, roi_quality=30, ), PreviewProfile( profile_name="sync_3fps_base_q20_roi_q30", fps=3.0, base_width=240, base_height=135, base_quality=20, roi_width=320, roi_height=180, roi_quality=30, ), PreviewProfile( profile_name="sync_3fps_base_q23_roi_q33", fps=3.0, base_width=240, base_height=135, base_quality=23, roi_width=320, roi_height=180, roi_quality=33, ), PreviewProfile( profile_name="sync_3fps_base_q25_roi_q35", fps=3.0, base_width=240, base_height=135, base_quality=25, roi_width=320, roi_height=180, roi_quality=35, ), ] if len(profiles) != 4: raise RuntimeError("Lab027D должна содержать четыре профиля.") if len({profile.profile_name for profile in profiles}) != 4: raise RuntimeError("Имена профилей Lab027D не уникальны.") return profiles def normalized_roi_to_pixels( width: int, height: int, ) -> tuple[int, int, int, int]: """ Переводит нормализованные координаты ROI в пиксели. """ x_min = int(round(width * ROI_X_MIN)) x_max = int(round(width * ROI_X_MAX)) y_min = int(round(height * ROI_Y_MIN)) y_max = int(round(height * ROI_Y_MAX)) if not ( 0 <= x_min < x_max <= width and 0 <= y_min < y_max <= height ): raise RuntimeError("Расчётная ROI выходит за границы кадра.") return x_min, y_min, x_max, y_max def should_update( current_time: float, next_update_time: float, ) -> bool: """ Проверяет наступление времени следующего обновления. """ return ( current_time + FRAME_TIME_EPSILON_SECONDS >= next_update_time ) def encode_decode_jpeg( grayscale_image: np.ndarray, quality: int, ) -> tuple[int, np.ndarray]: """ Кодирует grayscale-кадр в JPEG в памяти и сразу декодирует. """ encode_parameters = [ int(cv2.IMWRITE_JPEG_QUALITY), int(quality), ] encoded, jpeg_buffer = cv2.imencode( ".jpg", grayscale_image, encode_parameters, ) if not encoded: raise RuntimeError("OpenCV не смог закодировать JPEG.") decoded = cv2.imdecode( jpeg_buffer, cv2.IMREAD_GRAYSCALE, ) if decoded is None: raise RuntimeError("OpenCV не смог декодировать JPEG.") return int(jpeg_buffer.nbytes), decoded def encode_base( source_frame: np.ndarray, profile: PreviewProfile, ) -> tuple[int, np.ndarray]: """ Формирует, кодирует и декодирует BASE одного профиля. """ grayscale = cv2.cvtColor( source_frame, cv2.COLOR_BGR2GRAY, ) resized = cv2.resize( grayscale, (profile.base_width, profile.base_height), interpolation=cv2.INTER_AREA, ) return encode_decode_jpeg(resized, profile.base_quality) def encode_roi( source_frame: np.ndarray, source_roi: tuple[int, int, int, int], profile: PreviewProfile, ) -> tuple[int, np.ndarray]: """ Вырезает из исходного кадра ROI и обрабатывает JPEG в памяти. """ x_min, y_min, x_max, y_max = source_roi roi = source_frame[y_min:y_max, x_min:x_max] if roi.size == 0: raise RuntimeError("Получена пустая ROI.") grayscale = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY) resized = cv2.resize( grayscale, (profile.roi_width, profile.roi_height), interpolation=cv2.INTER_AREA, ) return encode_decode_jpeg(resized, profile.roi_quality) def create_measurements( profiles: list[PreviewProfile], ) -> Measurements: """ Создаёт пустые списки размеров JPEG для одного прохода. """ return { profile.profile_name: { "base": [], "roi": [], "composite": [], } for profile in profiles } def create_states( profiles: list[PreviewProfile], ) -> list[SyncState]: """ Создаёт независимое синхронное состояние каждого профиля. """ return [SyncState() for _ in profiles] def update_sync_state( source_frame: np.ndarray, source_roi: tuple[int, int, int, int], source_frame_index: int, current_time: float, profile: PreviewProfile, state: SyncState, measurements: dict[str, list[int]], ) -> bool: """ Атомарно обновляет BASE и ROI из одного исходного кадра. """ if not should_update(current_time, state.next_composite_time): return False next_composite_frame_id = state.composite_frame_id + 1 # Метаданные обеих частей намеренно формируются отдельно, # чтобы проверка синхронизации была явной. base_source_frame_index = source_frame_index roi_source_frame_index = source_frame_index base_timestamp = current_time roi_timestamp = current_time base_composite_frame_id = next_composite_frame_id roi_composite_frame_id = next_composite_frame_id base_size, decoded_base = encode_base(source_frame, profile) roi_size, decoded_roi = encode_roi( source_frame, source_roi, profile, ) if base_timestamp != roi_timestamp: state.timestamp_mismatch_count += 1 if ( base_source_frame_index != roi_source_frame_index or base_composite_frame_id != roi_composite_frame_id ): state.frame_id_mismatch_count += 1 # Публикация выполняется только после готовности обеих частей. state.latest_base = decoded_base state.latest_roi = decoded_roi state.composite_frame_id = next_composite_frame_id state.source_frame_index = source_frame_index state.timestamp = current_time state.last_composite_update_time = current_time state.next_composite_time += profile.period_seconds state.selected_base_frames += 1 state.selected_roi_frames += 1 state.synchronized_composite_frames += 1 measurements["base"].append(base_size) measurements["roi"].append(roi_size) measurements["composite"].append(base_size + roi_size) return True def process_first_pass( source_path: Path, profiles: list[PreviewProfile], source_width: int, source_height: int, source_fps: float, expected_frame_count: int, ) -> tuple[Measurements, list[SyncState]]: """ Выполняет первый проход и измеряет JPEG payload без видео. """ measurements = create_measurements(profiles) states = create_states(profiles) source_roi = normalized_roi_to_pixels( source_width, source_height, ) capture = cv2.VideoCapture(str(source_path)) if not capture.isOpened(): raise RuntimeError( f"OpenCV не смог открыть видео: {source_path}" ) frame_index = 0 try: while True: frame_read, source_frame = capture.read() if not frame_read: break if source_frame is None: raise RuntimeError( f"Получен пустой исходный кадр {frame_index}." ) if ( source_frame.shape[1] != source_width or source_frame.shape[0] != source_height ): raise RuntimeError( "Размер исходного кадра отличается от метаданных." ) current_time = frame_index / source_fps for profile, state in zip(profiles, states): update_sync_state( source_frame, source_roi, frame_index, current_time, profile, state, measurements[profile.profile_name], ) frame_index += 1 if ( frame_index % 100 == 0 or frame_index == expected_frame_count ): print( " First pass frames: " f"{frame_index}/{expected_frame_count}" ) finally: capture.release() if frame_index != expected_frame_count: raise RuntimeError( "Первый проход прочитал некорректное число кадров." ) return measurements, states def calculate_statistics( profiles: list[PreviewProfile], measurements: Measurements, states: list[SyncState], source_duration_seconds: float, ) -> list[ProfileStatistics]: """ Рассчитывает payload и иллюстративные канальные скорости. """ statistics: list[ProfileStatistics] = [] for profile, state in zip(profiles, states): profile_measurements = measurements[ profile.profile_name ] base_sizes = profile_measurements["base"] roi_sizes = profile_measurements["roi"] composite_sizes = profile_measurements["composite"] if not base_sizes or not roi_sizes or not composite_sizes: raise RuntimeError( f"Пустые измерения: {profile.profile_name}" ) if not ( len(base_sizes) == len(roi_sizes) == len(composite_sizes) == state.synchronized_composite_frames ): raise RuntimeError( f"Число обновлений не совпало: " f"{profile.profile_name}" ) total_base_bytes = int(sum(base_sizes)) total_roi_bytes = int(sum(roi_sizes)) total_payload_bytes = total_base_bytes + total_roi_bytes base_bitrate_kbps = ( total_base_bytes * 8.0 / source_duration_seconds / 1000.0 ) roi_bitrate_kbps = ( total_roi_bytes * 8.0 / source_duration_seconds / 1000.0 ) total_payload_bitrate_kbps = ( total_payload_bytes * 8.0 / source_duration_seconds / 1000.0 ) if not np.isclose( total_payload_bitrate_kbps, base_bitrate_kbps + roi_bitrate_kbps, rtol=0.0, atol=1e-12, ): raise RuntimeError( "Суммарный bitrate не равен BASE + ROI." ) channel_rate_fec_2_3_kbps = ( total_payload_bitrate_kbps * SERVICE_OVERHEAD_FACTOR / FEC_RATE_TWO_THIRDS ) channel_rate_fec_1_2_kbps = ( total_payload_bitrate_kbps * SERVICE_OVERHEAD_FACTOR / FEC_RATE_ONE_HALF ) statistics.append( ProfileStatistics( profile_name=profile.profile_name, base_width=profile.base_width, base_height=profile.base_height, fps=profile.fps, base_quality=profile.base_quality, roi_width=profile.roi_width, roi_height=profile.roi_height, roi_quality=profile.roi_quality, source_duration_s=source_duration_seconds, selected_base_frames=state.selected_base_frames, selected_roi_frames=state.selected_roi_frames, synchronized_composite_frames=( state.synchronized_composite_frames ), total_base_bytes=total_base_bytes, total_roi_bytes=total_roi_bytes, total_payload_bytes=total_payload_bytes, mean_base_frame_bytes=float(np.mean(base_sizes)), mean_roi_frame_bytes=float(np.mean(roi_sizes)), mean_composite_frame_bytes=float( np.mean(composite_sizes) ), p95_composite_frame_bytes=float( np.percentile(composite_sizes, 95) ), max_composite_frame_bytes=int(max(composite_sizes)), base_bitrate_kbps=base_bitrate_kbps, roi_bitrate_kbps=roi_bitrate_kbps, total_payload_bitrate_kbps=( total_payload_bitrate_kbps ), channel_rate_fec_2_3_kbps=( channel_rate_fec_2_3_kbps ), channel_rate_fec_1_2_kbps=( channel_rate_fec_1_2_kbps ), timestamp_mismatch_count=( state.timestamp_mismatch_count ), frame_id_mismatch_count=( state.frame_id_mismatch_count ), ) ) return statistics def reconstruct_panel( profile: PreviewProfile, state: SyncState, ) -> np.ndarray: """ Реконструирует панель 640x360 из последнего BASE и ROI. """ if state.latest_base is None or state.latest_roi is None: return np.zeros( (PANEL_HEIGHT, PANEL_WIDTH, 3), dtype=np.uint8, ) base_large = cv2.resize( state.latest_base, (PANEL_WIDTH, PANEL_HEIGHT), interpolation=cv2.INTER_LINEAR, ) panel = cv2.cvtColor(base_large, cv2.COLOR_GRAY2BGR) panel_roi = normalized_roi_to_pixels( PANEL_WIDTH, PANEL_HEIGHT, ) x_min, y_min, x_max, y_max = panel_roi roi_large = cv2.resize( state.latest_roi, (x_max - x_min, y_max - y_min), interpolation=cv2.INTER_LINEAR, ) roi_bgr = cv2.cvtColor(roi_large, cv2.COLOR_GRAY2BGR) panel[y_min:y_max, x_min:x_max] = roi_bgr cv2.rectangle( panel, (x_min, y_min), (x_max - 1, y_max - 1), (0, 255, 255), 2, ) return panel def draw_text_line( panel: np.ndarray, text: str, line_index: int, color: tuple[int, int, int] = (255, 255, 255), ) -> None: """ Рисует одну строку читаемой подписи на панели. """ y_position = 22 + line_index * 21 cv2.putText( panel, text, (9, y_position), cv2.FONT_HERSHEY_SIMPLEX, 0.48, (0, 0, 0), 3, cv2.LINE_AA, ) cv2.putText( panel, text, (9, y_position), cv2.FONT_HERSHEY_SIMPLEX, 0.48, color, 1, cv2.LINE_AA, ) def draw_sync_information( panel: np.ndarray, profile: PreviewProfile, state: SyncState, statistics: ProfileStatistics, current_time: float, ) -> None: """ Добавляет к панели параметры и динамическое состояние. """ age_seconds = max( 0.0, current_time - state.last_composite_update_time, ) lines = [ f"{profile.profile_name} | SYNCHRONOUS", f"FPS {profile.fps:.0f} | " f"BASE {profile.base_width}x{profile.base_height} " f"Q{profile.base_quality}", f"ROI {profile.roi_width}x{profile.roi_height} " f"Q{profile.roi_quality}", "payload " f"{statistics.total_payload_bitrate_kbps:.3f} kbit/s", "channel FEC 2/3 " f"{statistics.channel_rate_fec_2_3_kbps:.3f} kbit/s", f"time {current_time:.3f} s | " f"composite ID {state.composite_frame_id}", f"source frame {state.source_frame_index} | " f"age {age_seconds:.3f} s", ] for line_index, text in enumerate(lines): draw_text_line(panel, text, line_index) if age_seconds <= NEW_LABEL_DURATION_SECONDS: draw_text_line( panel, "NEW COMPOSITE", len(lines), color=(0, 255, 0), ) def compose_grid(panels: list[np.ndarray]) -> np.ndarray: """ Объединяет четыре панели в сетку 2x2 размером 1280x720. """ if len(panels) != 4: raise RuntimeError("Для preview требуется четыре панели.") top_row = np.hstack((panels[0], panels[1])) bottom_row = np.hstack((panels[2], panels[3])) grid = np.vstack((top_row, bottom_row)) if grid.shape != (OUTPUT_HEIGHT, OUTPUT_WIDTH, 3): raise RuntimeError( f"Некорректный размер сетки: {grid.shape}" ) return grid def open_preview_writer() -> tuple[cv2.VideoWriter, Path, bool]: """ Открывает MP4/mp4v или разрешённый AVI/MJPG fallback. """ PREVIEW_DIRECTORY.mkdir(parents=True, exist_ok=True) mp4_writer = cv2.VideoWriter( str(MP4_PREVIEW_PATH), cv2.VideoWriter_fourcc(*"mp4v"), OUTPUT_FPS, (OUTPUT_WIDTH, OUTPUT_HEIGHT), ) if mp4_writer.isOpened(): return mp4_writer, MP4_PREVIEW_PATH, False mp4_writer.release() if MP4_PREVIEW_PATH.exists(): MP4_PREVIEW_PATH.unlink() avi_writer = cv2.VideoWriter( str(AVI_PREVIEW_PATH), cv2.VideoWriter_fourcc(*"MJPG"), OUTPUT_FPS, (OUTPUT_WIDTH, OUTPUT_HEIGHT), ) if not avi_writer.isOpened(): avi_writer.release() raise RuntimeError( "Не удалось открыть MP4/mp4v и AVI/MJPG writer." ) return avi_writer, AVI_PREVIEW_PATH, True def write_preview( source_path: Path, profiles: list[PreviewProfile], statistics: list[ProfileStatistics], source_width: int, source_height: int, source_fps: float, expected_frame_count: int, ) -> tuple[ Path, bool, int, Measurements, list[SyncState], ]: """ Выполняет второй проход и записывает сравнительное preview. """ measurements = create_measurements(profiles) states = create_states(profiles) source_roi = normalized_roi_to_pixels( source_width, source_height, ) capture = cv2.VideoCapture(str(source_path)) if not capture.isOpened(): raise RuntimeError( f"OpenCV не смог открыть видео: {source_path}" ) writer, preview_path, fallback_used = open_preview_writer() frame_index = 0 try: while True: frame_read, source_frame = capture.read() if not frame_read: break if source_frame is None: raise RuntimeError( f"Получен пустой исходный кадр {frame_index}." ) current_time = frame_index / source_fps panels: list[np.ndarray] = [] for profile, state, profile_statistics in zip( profiles, states, statistics, ): update_sync_state( source_frame, source_roi, frame_index, current_time, profile, state, measurements[profile.profile_name], ) panel = reconstruct_panel(profile, state) draw_sync_information( panel, profile, state, profile_statistics, current_time, ) panels.append(panel) writer.write(compose_grid(panels)) frame_index += 1 if ( frame_index % 100 == 0 or frame_index == expected_frame_count ): print( " Second pass frames: " f"{frame_index}/{expected_frame_count}" ) finally: capture.release() writer.release() if frame_index != expected_frame_count: raise RuntimeError( "Второй проход записал некорректное число кадров." ) return ( preview_path, fallback_used, frame_index, measurements, states, ) def compare_passes( profiles: list[PreviewProfile], first_measurements: Measurements, second_measurements: Measurements, first_states: list[SyncState], second_states: list[SyncState], ) -> None: """ Проверяет полное совпадение измерений двух проходов. """ for profile, first_state, second_state in zip( profiles, first_states, second_states, ): profile_name = profile.profile_name for stream_name in ["base", "roi", "composite"]: if ( first_measurements[profile_name][stream_name] != second_measurements[profile_name][stream_name] ): raise RuntimeError( "Размеры JPEG двух проходов не совпали: " f"{profile_name}, {stream_name}." ) if ( first_state.selected_base_frames != second_state.selected_base_frames or first_state.selected_roi_frames != second_state.selected_roi_frames or first_state.synchronized_composite_frames != second_state.synchronized_composite_frames or first_state.timestamp_mismatch_count != second_state.timestamp_mismatch_count or first_state.frame_id_mismatch_count != second_state.frame_id_mismatch_count ): raise RuntimeError( "Состояния двух проходов не совпали: " f"{profile_name}." ) def validate_statistics( statistics: list[ProfileStatistics], ) -> None: """ Проверяет синхронность и арифметику всех профилей. """ if len(statistics) != 4: raise RuntimeError("Ожидалось четыре строки статистики.") for item in statistics: if not ( item.selected_base_frames == item.selected_roi_frames == item.synchronized_composite_frames ): raise RuntimeError( f"Обновления не синхронны: {item.profile_name}" ) if ( item.timestamp_mismatch_count != 0 or item.frame_id_mismatch_count != 0 ): raise RuntimeError( f"Обнаружен mismatch: {item.profile_name}" ) if ( item.total_payload_bytes != item.total_base_bytes + item.total_roi_bytes ): raise RuntimeError( f"Ошибка суммы payload: {item.profile_name}" ) def save_csv(statistics: list[ProfileStatistics]) -> None: """ Сохраняет четыре полные строки результатов в UTF-8 CSV. """ with CSV_PATH.open( "w", encoding="utf-8", newline="", ) as csv_file: writer = csv.DictWriter( csv_file, fieldnames=CSV_FIELD_NAMES, ) writer.writeheader() for item in statistics: writer.writerow( { "profile_name": item.profile_name, "base_width": item.base_width, "base_height": item.base_height, "fps": f"{item.fps:.6f}", "base_quality": item.base_quality, "roi_width": item.roi_width, "roi_height": item.roi_height, "roi_quality": item.roi_quality, "source_duration_s": ( f"{item.source_duration_s:.6f}" ), "selected_base_frames": ( item.selected_base_frames ), "selected_roi_frames": ( item.selected_roi_frames ), "synchronized_composite_frames": ( item.synchronized_composite_frames ), "total_base_bytes": item.total_base_bytes, "total_roi_bytes": item.total_roi_bytes, "total_payload_bytes": ( item.total_payload_bytes ), "mean_base_frame_bytes": ( f"{item.mean_base_frame_bytes:.3f}" ), "mean_roi_frame_bytes": ( f"{item.mean_roi_frame_bytes:.3f}" ), "mean_composite_frame_bytes": ( f"{item.mean_composite_frame_bytes:.3f}" ), "p95_composite_frame_bytes": ( f"{item.p95_composite_frame_bytes:.3f}" ), "max_composite_frame_bytes": ( item.max_composite_frame_bytes ), "base_bitrate_kbps": ( f"{item.base_bitrate_kbps:.6f}" ), "roi_bitrate_kbps": ( f"{item.roi_bitrate_kbps:.6f}" ), "total_payload_bitrate_kbps": ( f"{item.total_payload_bitrate_kbps:.6f}" ), "channel_rate_fec_2_3_kbps": ( f"{item.channel_rate_fec_2_3_kbps:.6f}" ), "channel_rate_fec_1_2_kbps": ( f"{item.channel_rate_fec_1_2_kbps:.6f}" ), "timestamp_mismatch_count": ( item.timestamp_mismatch_count ), "frame_id_mismatch_count": ( item.frame_id_mismatch_count ), } ) def format_statistics_line(item: ProfileStatistics) -> str: """ Формирует одну подробную строку текстового отчёта. """ return ( f"{item.profile_name}: " f"{item.fps:.3f} fps; " f"BASE={item.base_width}x{item.base_height}, " f"Q{item.base_quality}, " f"frames={item.selected_base_frames}, " f"bytes={item.total_base_bytes}, " f"bitrate={item.base_bitrate_kbps:.6f} kbit/s; " f"ROI={item.roi_width}x{item.roi_height}, " f"Q{item.roi_quality}, " f"frames={item.selected_roi_frames}, " f"bytes={item.total_roi_bytes}, " f"bitrate={item.roi_bitrate_kbps:.6f} kbit/s; " f"total_bytes={item.total_payload_bytes}, " f"payload={item.total_payload_bitrate_kbps:.6f} kbit/s; " f"channel FEC 2/3=" f"{item.channel_rate_fec_2_3_kbps:.6f} kbit/s; " f"channel FEC 1/2=" f"{item.channel_rate_fec_1_2_kbps:.6f} kbit/s; " f"sync_frames={item.synchronized_composite_frames}; " f"timestamp_mismatch=" f"{item.timestamp_mismatch_count}; " f"frame_id_mismatch={item.frame_id_mismatch_count}" ) def write_report( statistics: list[ProfileStatistics], source_width: int, source_height: int, source_fps: float, source_frame_count: int, source_duration_seconds: float, preview_path: Path, fallback_used: bool, ) -> None: """ Сохраняет UTF-8 отчёт без автоматического выбора профиля. """ source_roi = normalized_roi_to_pixels( source_width, source_height, ) lines = [ "Lab027D. Синхронный BASE + ROI при 2 и 3 fps", "", "Цель: сравнить текущий синхронный профиль 2 fps " "с тремя профилями 3 fps при разных JPEG Quality.", "", "Результат ручной оценки Lab027C:", "- лучший из показанных вариантов — нижний левый;", "- синхронизация BASE и ROI устранила рассогласование;", "- 2 fps недостаточно;", "- BASE 160x90 слишком грубый;", "- BASE 240x135 Q20 немного недостаточен по качеству.", "", "Причина повышения FPS: ручная оценка показала, что " "синхронное обновление устраняет рассогласование, но " "частота 2 fps недостаточна для динамической сцены.", "", f"Исходное видео: {SOURCE_VIDEO_PATH}", f"Разрешение: {source_width}x{source_height}", f"FPS: {source_fps:.6f}", f"Кадров: {source_frame_count}", f"Длительность: {source_duration_seconds:.6f} с", "ROI: " f"x={ROI_X_MIN:.2f}...{ROI_X_MAX:.2f}, " f"y={ROI_Y_MIN:.2f}...{ROI_Y_MAX:.2f}; " f"пиксели x={source_roi[0]}...{source_roi[2]}, " f"y={source_roi[1]}...{source_roi[3]}.", "", "Все профили синхронные: BASE и ROI формируются из " "одного source frame, получают единый timestamp и " "composite frame ID и публикуются атомарно.", "", "Фактические результаты:", ] lines.extend( format_statistics_line(item) for item in statistics ) lines.extend( [ "", "Оценки channel rate иллюстративны: к payload " "добавлено 10% служебных данных, затем применена " "оценка FEC 2/3 или FEC 1/2.", "Эти значения не являются окончательной архитектурой " "радиоканала.", "", f"Preview: {preview_path}", "Формат: " + ("AVI/MJPG fallback." if fallback_used else "MP4/mp4v."), "", "Предупреждение: оценки пока не включают окончательную " "модуляцию, полосу сигнала, интерливинг, повторы и " "команды управления.", "Программа не выбирает лучший профиль автоматически.", "Следующий шаг: ручной выбор пользователя после " "просмотра preview-видео.", "", ] ) REPORT_PATH.write_text( "\n".join(lines), encoding="utf-8", ) def read_frame_at( capture: cv2.VideoCapture, frame_index: int, ) -> tuple[bool, tuple[int, ...] | None]: """ Читает один контрольный кадр preview. """ capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index) frame_read, frame = capture.read() if not frame_read or frame is None: return False, None return True, frame.shape def verify_preview( preview_path: Path, source_frame_count: int, source_duration_seconds: float, ) -> tuple[ int, int, float, int, float, tuple[bool, tuple[int, ...] | None], tuple[bool, tuple[int, ...] | None], tuple[bool, tuple[int, ...] | None], ]: """ Проверяет метаданные и три контрольных кадра preview. """ if not preview_path.exists() or preview_path.stat().st_size <= 0: raise RuntimeError("Preview отсутствует или имеет нулевой размер.") capture = cv2.VideoCapture(str(preview_path)) if not capture.isOpened(): raise RuntimeError( f"OpenCV не смог открыть preview: {preview_path}" ) try: width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = float(capture.get(cv2.CAP_PROP_FPS)) frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT)) if fps <= 0.0: raise RuntimeError("FPS preview равен нулю.") duration_seconds = frame_count / fps first_frame = read_frame_at(capture, 0) middle_frame = read_frame_at( capture, frame_count // 2, ) last_frame = read_frame_at( capture, frame_count - 1, ) finally: capture.release() if (width, height) != (OUTPUT_WIDTH, OUTPUT_HEIGHT): raise RuntimeError( f"Некорректное разрешение preview: {width}x{height}." ) if frame_count != source_frame_count: raise RuntimeError( "Число кадров preview не совпало с исходным." ) if ( abs(duration_seconds - source_duration_seconds) > 1.0 / fps + FRAME_TIME_EPSILON_SECONDS ): raise RuntimeError( "Длительность preview отличается более чем на кадр." ) if not all( frame_result[0] for frame_result in [ first_frame, middle_frame, last_frame, ] ): raise RuntimeError( "Не удалось прочитать контрольный кадр preview." ) return ( width, height, fps, frame_count, duration_seconds, first_frame, middle_frame, last_frame, ) def main() -> None: """ Выполняет два прохода Lab027D и проверяет результаты. """ print("Reading source video metadata...") ( source_width, source_height, source_fps, source_frame_count, source_duration_seconds, source_file_size_bytes, ) = read_video_metadata(SOURCE_VIDEO_PATH) print(f" Source: {SOURCE_VIDEO_PATH}") print(f" File size: {source_file_size_bytes} bytes") print(f" Resolution: {source_width}x{source_height}") print(f" FPS: {source_fps:.6f}") print(f" Frames: {source_frame_count}") print(f" Duration: {source_duration_seconds:.6f} s") profiles = build_profiles() print("First pass: measuring synchronized JPEG payload...") first_measurements, first_states = process_first_pass( SOURCE_VIDEO_PATH, profiles, source_width, source_height, source_fps, source_frame_count, ) statistics = calculate_statistics( profiles, first_measurements, first_states, source_duration_seconds, ) validate_statistics(statistics) print("Second pass: writing preview...") ( preview_path, fallback_used, written_frame_count, second_measurements, second_states, ) = write_preview( SOURCE_VIDEO_PATH, profiles, statistics, source_width, source_height, source_fps, source_frame_count, ) compare_passes( profiles, first_measurements, second_measurements, first_states, second_states, ) second_statistics = calculate_statistics( profiles, second_measurements, second_states, source_duration_seconds, ) validate_statistics(second_statistics) if statistics != second_statistics: raise RuntimeError( "Итоговая статистика двух проходов не совпала." ) ( preview_width, preview_height, preview_fps, verified_frame_count, preview_duration_seconds, first_frame, middle_frame, last_frame, ) = verify_preview( preview_path, source_frame_count, source_duration_seconds, ) OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) save_csv(statistics) write_report( statistics, source_width, source_height, source_fps, source_frame_count, source_duration_seconds, preview_path, fallback_used, ) print("") print("Measured profiles:") for item in statistics: print( f" {item.profile_name}: " f"updates={item.synchronized_composite_frames}, " f"BASE={item.base_bitrate_kbps:.6f}, " f"ROI={item.roi_bitrate_kbps:.6f}, " f"payload={item.total_payload_bitrate_kbps:.6f}, " f"FEC 2/3={item.channel_rate_fec_2_3_kbps:.6f}, " f"FEC 1/2={item.channel_rate_fec_1_2_kbps:.6f} " "kbit/s, " f"timestamp mismatch={item.timestamp_mismatch_count}, " f"frame ID mismatch={item.frame_id_mismatch_count}" ) print("") print("Two-pass JPEG sizes and statistics: identical") print(f"Preview: {preview_path}") print(f"Fallback used: {fallback_used}") print(f"Preview size: {preview_path.stat().st_size} bytes") print( f"Preview resolution: " f"{preview_width}x{preview_height}" ) print(f"Preview FPS: {preview_fps:.6f}") print(f"Written frames: {written_frame_count}") print(f"Verified frames: {verified_frame_count}") print( f"Preview duration: " f"{preview_duration_seconds:.6f} s" ) print(f"First frame: {first_frame}") print(f"Middle frame: {middle_frame}") print(f"Last frame: {last_frame}") print("") print("Lab027D completed successfully.") if __name__ == "__main__": main()