""" Lab027E. Операторское preview синхронного BASE + ROI. Лабораторная имитирует изображение на операторской стороне для предварительно выбранного профиля BASE 240x135 Q23 и ROI 320x180 Q33 при трёх атомарных обновлениях в секунду. Содержимое исходной сцены воспроизводится со скоростью 0.5x. JPEG-кодирование выполняется только при обновлении составного изображения. Отдельные JPEG-файлы не создаются. Первый проход измеряет payload, второй повторяет расписание и записывает preview. """ 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/lab027e") CSV_PATH = OUTPUT_DIRECTORY / "lab027e_profile.csv" REPORT_PATH = OUTPUT_DIRECTORY / "lab027e_report.txt" PREVIEW_DIRECTORY = Path("data/raw/lab027e_previews") MP4_PREVIEW_PATH = ( PREVIEW_DIRECTORY / "lab027e_operator_view_0_5x.mp4" ) AVI_PREVIEW_PATH = ( PREVIEW_DIRECTORY / "lab027e_operator_view_0_5x.avi" ) PROFILE_NAME = "sync_3fps_base240_q23_roi320_q33_speed_0_5x" PLAYBACK_SPEED_FACTOR = 0.5 COMPOSITE_FPS = 3.0 BASE_WIDTH = 240 BASE_HEIGHT = 135 BASE_QUALITY = 23 ROI_WIDTH = 320 ROI_HEIGHT = 180 ROI_QUALITY = 33 ROI_X_MIN = 0.20 ROI_X_MAX = 0.80 ROI_Y_MIN = 0.42 ROI_Y_MAX = 1.00 COMPOSITE_WIDTH = 640 COMPOSITE_HEIGHT = 360 OUTPUT_WIDTH = 1280 OUTPUT_HEIGHT = 720 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 CSV_FIELD_NAMES = [ "profile_name", "playback_speed_factor", "source_width", "source_height", "source_fps", "source_frames", "source_duration_s", "output_width", "output_height", "output_fps", "output_frames", "output_duration_s", "composite_fps", "base_width", "base_height", "base_quality", "roi_width", "roi_height", "roi_quality", "selected_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 OperatorProfile: """ Описывает выбранный операторский видеорежим. """ profile_name: str playback_speed_factor: float composite_fps: float base_width: int base_height: int base_quality: int roi_width: int roi_height: int roi_quality: int @property def composite_period_seconds(self) -> float: """ Возвращает период обновления составного изображения. """ return 1.0 / self.composite_fps @dataclass class CompositeState: """ Хранит последнее атомарно опубликованное изображение. """ next_composite_time: float = 0.0 last_composite_update_time: float = float("-inf") composite_frame_id: int = -1 source_frame_index: int = -1 source_timestamp: float = 0.0 latest_composite: np.ndarray | None = None selected_composite_frames: int = 0 timestamp_mismatch_count: int = 0 frame_id_mismatch_count: int = 0 @dataclass(frozen=True) class ProfileStatistics: """ Содержит итоговую строку измерений Lab027E. """ profile_name: str playback_speed_factor: float source_width: int source_height: int source_fps: float source_frames: int source_duration_s: float output_width: int output_height: int output_fps: float output_frames: int output_duration_s: float composite_fps: float base_width: int base_height: int base_quality: int roi_width: int roi_height: int roi_quality: int selected_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, list[int]] def build_profile() -> OperatorProfile: """ Создаёт единственный профиль Lab027E. """ return OperatorProfile( profile_name=PROFILE_NAME, playback_speed_factor=PLAYBACK_SPEED_FACTOR, composite_fps=COMPOSITE_FPS, base_width=BASE_WIDTH, base_height=BASE_HEIGHT, base_quality=BASE_QUALITY, roi_width=ROI_WIDTH, roi_height=ROI_HEIGHT, roi_quality=ROI_QUALITY, ) def read_video_metadata( source_path: Path, ) -> tuple[int, int, float, int, float, int]: """ Читает метаданные исходного MP4 без изменения файла. """ 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 calculate_output_frame_count( source_frame_count: int, source_fps: float, output_fps: float, playback_speed_factor: float, ) -> int: """ Вычисляет число выходных кадров без выхода за исходный диапазон. Последний допустимый output_frame_index должен давать source_frame_index не больше source_frame_count - 1. """ if ( source_frame_count <= 0 or source_fps <= 0.0 or output_fps <= 0.0 or playback_speed_factor <= 0.0 ): raise ValueError( "Параметры временной модели должны быть положительными." ) source_frames_per_output_frame = ( playback_speed_factor * source_fps / output_fps ) output_frame_count = int( np.ceil( source_frame_count / source_frames_per_output_frame ) ) if output_frame_count <= 0: raise RuntimeError( "Рассчитано некорректное число выходных кадров." ) last_source_index = source_frame_index_for_output( output_frame_count - 1, output_fps, playback_speed_factor, source_fps, source_frame_count, ) next_unclamped_source_index = int( np.floor( output_frame_count / output_fps * playback_speed_factor * source_fps + FRAME_TIME_EPSILON_SECONDS ) ) if last_source_index >= source_frame_count: raise RuntimeError( "Последний выходной кадр вышел за исходный диапазон." ) if next_unclamped_source_index < source_frame_count: raise RuntimeError( "Рассчитано недостаточное число выходных кадров." ) return output_frame_count def source_frame_index_for_output( output_frame_index: int, output_fps: float, playback_speed_factor: float, source_fps: float, source_frame_count: int, ) -> int: """ Детерминированно сопоставляет выходной кадр исходному. """ output_time = output_frame_index / output_fps source_time = output_time * playback_speed_factor source_frame_index = int( np.floor( source_time * source_fps + FRAME_TIME_EPSILON_SECONDS ) ) return min( max(source_frame_index, 0), source_frame_count - 1, ) 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( output_time: float, next_composite_time: float, ) -> bool: """ Проверяет наступление времени composite-обновления. """ return ( output_time + FRAME_TIME_EPSILON_SECONDS >= next_composite_time ) def read_source_frame_sequentially( capture: cv2.VideoCapture, current_source_frame_index: int, current_source_frame: np.ndarray | None, target_source_frame_index: int, ) -> tuple[int, np.ndarray]: """ Последовательно читает исходник до заданного индекса. Случайный seek не используется. Если запрошен тот же индекс, возвращается уже прочитанный кадр. """ if target_source_frame_index < current_source_frame_index: raise RuntimeError( "Последовательное чтение не допускает возврат назад." ) while current_source_frame_index < target_source_frame_index: frame_read, source_frame = capture.read() if not frame_read or source_frame is None: raise RuntimeError( "Исходное видео закончилось до целевого кадра " f"{target_source_frame_index}." ) current_source_frame_index += 1 current_source_frame = source_frame if current_source_frame is None: raise RuntimeError("Не удалось получить исходный кадр.") return current_source_frame_index, current_source_frame def encode_decode_jpeg( grayscale_image: np.ndarray, quality: int, ) -> tuple[int, np.ndarray]: """ Кодирует grayscale-изображение в JPEG в памяти и декодирует. """ encoded, jpeg_buffer = cv2.imencode( ".jpg", grayscale_image, [ int(cv2.IMWRITE_JPEG_QUALITY), int(quality), ], ) 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: OperatorProfile, ) -> tuple[int, np.ndarray]: """ Формирует BASE заданного размера и JPEG Quality. """ 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: OperatorProfile, ) -> tuple[int, np.ndarray]: """ Формирует ROI заданного размера и JPEG Quality. """ 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 build_composite( decoded_base: np.ndarray, decoded_roi: np.ndarray, ) -> np.ndarray: """ Собирает фактическое составное изображение 640x360. ROI заменяет соответствующую область напрямую, без рамки, смешивания и сглаживания границы. """ base_large = cv2.resize( decoded_base, (COMPOSITE_WIDTH, COMPOSITE_HEIGHT), interpolation=cv2.INTER_LINEAR, ) composite = cv2.cvtColor( base_large, cv2.COLOR_GRAY2BGR, ) x_min, y_min, x_max, y_max = normalized_roi_to_pixels( COMPOSITE_WIDTH, COMPOSITE_HEIGHT, ) roi_large = cv2.resize( decoded_roi, (x_max - x_min, y_max - y_min), interpolation=cv2.INTER_LINEAR, ) composite[y_min:y_max, x_min:x_max] = cv2.cvtColor( roi_large, cv2.COLOR_GRAY2BGR, ) return composite def create_measurements() -> Measurements: """ Создаёт пустые измерения одного прохода. """ return { "base": [], "roi": [], "composite": [], "source_indices": [], } def update_composite_state( source_frame: np.ndarray, source_roi: tuple[int, int, int, int], source_frame_index: int, source_fps: float, output_time: float, profile: OperatorProfile, state: CompositeState, measurements: Measurements, ) -> None: """ Кодирует обе части и атомарно публикует составной кадр. """ next_composite_frame_id = state.composite_frame_id + 1 source_timestamp = source_frame_index / source_fps base_source_frame_index = source_frame_index roi_source_frame_index = source_frame_index base_source_timestamp = source_timestamp roi_source_timestamp = source_timestamp 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_source_timestamp != roi_source_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 composite = build_composite(decoded_base, decoded_roi) # Все отображаемые поля заменяются только после готовности # BASE, ROI и реконструированного изображения. state.latest_composite = composite state.composite_frame_id = next_composite_frame_id state.source_frame_index = source_frame_index state.source_timestamp = source_timestamp state.last_composite_update_time = output_time state.next_composite_time += ( profile.composite_period_seconds ) state.selected_composite_frames += 1 measurements["base"].append(base_size) measurements["roi"].append(roi_size) measurements["composite"].append(base_size + roi_size) measurements["source_indices"].append(source_frame_index) def process_first_pass( source_path: Path, profile: OperatorProfile, source_width: int, source_height: int, source_fps: float, source_frame_count: int, output_frame_count: int, ) -> tuple[Measurements, CompositeState]: """ Измеряет JPEG payload без создания видео. """ measurements = create_measurements() state = CompositeState() 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}" ) current_source_frame_index = -1 current_source_frame: np.ndarray | None = None try: for output_frame_index in range(output_frame_count): output_time = output_frame_index / OUTPUT_FPS if not should_update( output_time, state.next_composite_time, ): continue target_source_frame_index = ( source_frame_index_for_output( output_frame_index, OUTPUT_FPS, profile.playback_speed_factor, source_fps, source_frame_count, ) ) ( current_source_frame_index, current_source_frame, ) = read_source_frame_sequentially( capture, current_source_frame_index, current_source_frame, target_source_frame_index, ) update_composite_state( current_source_frame, source_roi, current_source_frame_index, source_fps, output_time, profile, state, measurements, ) if state.selected_composite_frames % 25 == 0: print( " First pass composite frames: " f"{state.selected_composite_frames}" ) finally: capture.release() if state.latest_composite is None: raise RuntimeError( "Первый проход не сформировал составных кадров." ) return measurements, state def calculate_statistics( profile: OperatorProfile, measurements: Measurements, state: CompositeState, source_width: int, source_height: int, source_fps: float, source_frame_count: int, source_duration_seconds: float, output_frame_count: int, ) -> ProfileStatistics: """ Рассчитывает bitrate по длительности замедленного preview. """ base_sizes = measurements["base"] roi_sizes = measurements["roi"] composite_sizes = measurements["composite"] if not base_sizes or not roi_sizes or not composite_sizes: raise RuntimeError("Получены пустые измерения JPEG.") if not ( len(base_sizes) == len(roi_sizes) == len(composite_sizes) == len(measurements["source_indices"]) == state.selected_composite_frames ): raise RuntimeError( "Число измерений и составных кадров не совпало." ) output_duration_seconds = output_frame_count / OUTPUT_FPS 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 / output_duration_seconds / 1000.0 ) roi_bitrate_kbps = ( total_roi_bytes * 8.0 / output_duration_seconds / 1000.0 ) total_payload_bitrate_kbps = ( total_payload_bytes * 8.0 / output_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 ) return ProfileStatistics( profile_name=profile.profile_name, playback_speed_factor=profile.playback_speed_factor, source_width=source_width, source_height=source_height, source_fps=source_fps, source_frames=source_frame_count, source_duration_s=source_duration_seconds, output_width=OUTPUT_WIDTH, output_height=OUTPUT_HEIGHT, output_fps=OUTPUT_FPS, output_frames=output_frame_count, output_duration_s=output_duration_seconds, composite_fps=profile.composite_fps, base_width=profile.base_width, base_height=profile.base_height, base_quality=profile.base_quality, roi_width=profile.roi_width, roi_height=profile.roi_height, roi_quality=profile.roi_quality, selected_composite_frames=( state.selected_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 ), ) def draw_minimal_status( preview_frame: np.ndarray, profile: OperatorProfile, statistics: ProfileStatistics, composite_age_seconds: float, ) -> None: """ Рисует небольшой непрозрачный статусный блок слева сверху. """ block_width = 410 block_height = 142 cv2.rectangle( preview_frame, (0, 0), (block_width, block_height), (0, 0, 0), cv2.FILLED, ) lines = [ f"{profile.composite_fps:.0f} fps", 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}", f"playback {profile.playback_speed_factor:.1f}x", "payload " f"{statistics.total_payload_bitrate_kbps:.3f} kbit/s", f"age {composite_age_seconds:.3f} s", ] for line_index, text in enumerate(lines): cv2.putText( preview_frame, text, (10, 22 + line_index * 22), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), 1, cv2.LINE_AA, ) 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_operator_preview( source_path: Path, profile: OperatorProfile, statistics: ProfileStatistics, source_width: int, source_height: int, source_fps: float, source_frame_count: int, output_frame_count: int, ) -> tuple[ Path, bool, Measurements, CompositeState, ]: """ Повторяет расписание и записывает одно полноэкранное preview. """ measurements = create_measurements() state = CompositeState() 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() current_source_frame_index = -1 current_source_frame: np.ndarray | None = None try: for output_frame_index in range(output_frame_count): output_time = output_frame_index / OUTPUT_FPS if should_update( output_time, state.next_composite_time, ): target_source_frame_index = ( source_frame_index_for_output( output_frame_index, OUTPUT_FPS, profile.playback_speed_factor, source_fps, source_frame_count, ) ) ( current_source_frame_index, current_source_frame, ) = read_source_frame_sequentially( capture, current_source_frame_index, current_source_frame, target_source_frame_index, ) update_composite_state( current_source_frame, source_roi, current_source_frame_index, source_fps, output_time, profile, state, measurements, ) if state.latest_composite is None: raise RuntimeError( "Отсутствует составное изображение для preview." ) preview_frame = cv2.resize( state.latest_composite, (OUTPUT_WIDTH, OUTPUT_HEIGHT), interpolation=cv2.INTER_LINEAR, ) composite_age_seconds = max( 0.0, output_time - state.last_composite_update_time, ) draw_minimal_status( preview_frame, profile, statistics, composite_age_seconds, ) writer.write(preview_frame) written_frames = output_frame_index + 1 if ( written_frames % 200 == 0 or written_frames == output_frame_count ): print( " Preview frames: " f"{written_frames}/{output_frame_count}" ) finally: capture.release() writer.release() return ( preview_path, fallback_used, measurements, state, ) def compare_passes( first_measurements: Measurements, second_measurements: Measurements, first_state: CompositeState, second_state: CompositeState, ) -> None: """ Проверяет точное совпадение обоих проходов. """ for measurement_name in [ "base", "roi", "composite", "source_indices", ]: if ( first_measurements[measurement_name] != second_measurements[measurement_name] ): raise RuntimeError( "Проходы различаются по измерению: " f"{measurement_name}." ) if ( first_state.selected_composite_frames != second_state.selected_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( "Итоговые состояния двух проходов не совпали." ) def validate_statistics( statistics: ProfileStatistics, ) -> None: """ Проверяет синхронизацию и арифметику итоговой строки. """ if statistics.selected_composite_frames <= 0: raise RuntimeError("Число составных обновлений равно нулю.") if ( statistics.timestamp_mismatch_count != 0 or statistics.frame_id_mismatch_count != 0 ): raise RuntimeError("Обнаружен mismatch синхронного профиля.") if ( statistics.total_payload_bytes != statistics.total_base_bytes + statistics.total_roi_bytes ): raise RuntimeError("Некорректная сумма payload bytes.") def save_csv(statistics: 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() writer.writerow( { "profile_name": statistics.profile_name, "playback_speed_factor": ( f"{statistics.playback_speed_factor:.6f}" ), "source_width": statistics.source_width, "source_height": statistics.source_height, "source_fps": f"{statistics.source_fps:.6f}", "source_frames": statistics.source_frames, "source_duration_s": ( f"{statistics.source_duration_s:.6f}" ), "output_width": statistics.output_width, "output_height": statistics.output_height, "output_fps": f"{statistics.output_fps:.6f}", "output_frames": statistics.output_frames, "output_duration_s": ( f"{statistics.output_duration_s:.6f}" ), "composite_fps": ( f"{statistics.composite_fps:.6f}" ), "base_width": statistics.base_width, "base_height": statistics.base_height, "base_quality": statistics.base_quality, "roi_width": statistics.roi_width, "roi_height": statistics.roi_height, "roi_quality": statistics.roi_quality, "selected_composite_frames": ( statistics.selected_composite_frames ), "total_base_bytes": statistics.total_base_bytes, "total_roi_bytes": statistics.total_roi_bytes, "total_payload_bytes": ( statistics.total_payload_bytes ), "mean_base_frame_bytes": ( f"{statistics.mean_base_frame_bytes:.3f}" ), "mean_roi_frame_bytes": ( f"{statistics.mean_roi_frame_bytes:.3f}" ), "mean_composite_frame_bytes": ( f"{statistics.mean_composite_frame_bytes:.3f}" ), "p95_composite_frame_bytes": ( f"{statistics.p95_composite_frame_bytes:.3f}" ), "max_composite_frame_bytes": ( statistics.max_composite_frame_bytes ), "base_bitrate_kbps": ( f"{statistics.base_bitrate_kbps:.6f}" ), "roi_bitrate_kbps": ( f"{statistics.roi_bitrate_kbps:.6f}" ), "total_payload_bitrate_kbps": ( f"{statistics.total_payload_bitrate_kbps:.6f}" ), "channel_rate_fec_2_3_kbps": ( f"{statistics.channel_rate_fec_2_3_kbps:.6f}" ), "channel_rate_fec_1_2_kbps": ( f"{statistics.channel_rate_fec_1_2_kbps:.6f}" ), "timestamp_mismatch_count": ( statistics.timestamp_mismatch_count ), "frame_id_mismatch_count": ( statistics.frame_id_mismatch_count ), } ) def write_report( statistics: ProfileStatistics, preview_path: Path, fallback_used: bool, ) -> None: """ Сохраняет учебный отчёт без окончательного утверждения режима. """ lines = [ "Lab027E. Операторское preview при playback 0.5x", "", "Результаты ручной оценки Lab027D:", "- 3 fps достаточно на исходном быстром видео;", "- Q20/Q30 приемлем;", "- Q23/Q33 приемлем;", "- разница Q23/Q33 и Q25/Q35 практически отсутствует.", "", "Предварительно выбран профиль Q23/Q33.", "От Q25/Q35 предварительно отказались, поскольку " "визуальный выигрыш почти отсутствует.", "Профиль не считается окончательно утверждённым до " "ручного просмотра этого preview.", "", "Исходная сцена замедлена для имитации более медленного " "движения ровера и оценки удобства управления.", f"Коэффициент playback: " f"{statistics.playback_speed_factor:.3f}x.", "Предупреждение: реальная скорость исходного автомобиля " "неизвестна.", "", "Исходное видео:", f"- путь: {SOURCE_VIDEO_PATH};", f"- разрешение: " f"{statistics.source_width}x{statistics.source_height};", f"- FPS: {statistics.source_fps:.6f};", f"- кадров: {statistics.source_frames};", f"- длительность: " f"{statistics.source_duration_s:.6f} с.", "", "Выходное preview:", f"- разрешение: " f"{statistics.output_width}x{statistics.output_height};", f"- FPS: {statistics.output_fps:.6f};", f"- кадров: {statistics.output_frames};", f"- длительность: " f"{statistics.output_duration_s:.6f} с;", f"- composite FPS: {statistics.composite_fps:.6f}.", "", "Профиль:", f"- BASE {statistics.base_width}x" f"{statistics.base_height} Q{statistics.base_quality};", f"- ROI {statistics.roi_width}x" f"{statistics.roi_height} Q{statistics.roi_quality};", f"- составных обновлений: " f"{statistics.selected_composite_frames}.", "", "Фактические скорости:", f"- BASE: {statistics.base_bitrate_kbps:.6f} kbit/s;", f"- ROI: {statistics.roi_bitrate_kbps:.6f} kbit/s;", "- total payload: " f"{statistics.total_payload_bitrate_kbps:.6f} kbit/s;", "- channel rate FEC 2/3: " f"{statistics.channel_rate_fec_2_3_kbps:.6f} kbit/s;", "- channel rate FEC 1/2: " f"{statistics.channel_rate_fec_1_2_kbps:.6f} kbit/s.", "", "Оценки канальной скорости иллюстративны и не являются " "окончательной архитектурой радиоканала.", "Mismatch-проверка:", "- timestamp mismatch: " f"{statistics.timestamp_mismatch_count};", "- frame ID mismatch: " f"{statistics.frame_id_mismatch_count}.", "", f"Preview: {preview_path}", "Формат: " + ("AVI/MJPG fallback." if fallback_used else "MP4/mp4v."), "Следующий шаг: ручная оценка удобства управления " "пользователем.", "", ] 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, expected_frame_count: int, expected_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 != expected_frame_count: raise RuntimeError( "Число кадров preview не совпало с расчётным." ) if ( abs(duration_seconds - expected_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: """ Выполняет оба прохода Lab027E и проверяет результат. """ 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) profile = build_profile() output_frame_count = calculate_output_frame_count( source_frame_count, source_fps, OUTPUT_FPS, profile.playback_speed_factor, ) output_duration_seconds = ( output_frame_count / OUTPUT_FPS ) 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" Source duration: {source_duration_seconds:.6f} s") print( f" Playback speed: " f"{profile.playback_speed_factor:.3f}x" ) print(f" Output frames: {output_frame_count}") print(f" Output duration: {output_duration_seconds:.6f} s") print("First pass: measuring JPEG payload...") first_measurements, first_state = process_first_pass( SOURCE_VIDEO_PATH, profile, source_width, source_height, source_fps, source_frame_count, output_frame_count, ) statistics = calculate_statistics( profile, first_measurements, first_state, source_width, source_height, source_fps, source_frame_count, source_duration_seconds, output_frame_count, ) validate_statistics(statistics) print("Second pass: writing operator preview...") ( preview_path, fallback_used, second_measurements, second_state, ) = write_operator_preview( SOURCE_VIDEO_PATH, profile, statistics, source_width, source_height, source_fps, source_frame_count, output_frame_count, ) compare_passes( first_measurements, second_measurements, first_state, second_state, ) second_statistics = calculate_statistics( profile, second_measurements, second_state, source_width, source_height, source_fps, source_frame_count, source_duration_seconds, output_frame_count, ) 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, output_frame_count, output_duration_seconds, ) OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) save_csv(statistics) write_report( statistics, preview_path, fallback_used, ) print("") print("Measured profile:") print(f" {statistics.profile_name}") print( " Composite updates: " f"{statistics.selected_composite_frames}" ) print( f" BASE bitrate: " f"{statistics.base_bitrate_kbps:.6f} kbit/s" ) print( f" ROI bitrate: " f"{statistics.roi_bitrate_kbps:.6f} kbit/s" ) print( f" Total payload bitrate: " f"{statistics.total_payload_bitrate_kbps:.6f} kbit/s" ) print( f" Channel rate FEC 2/3: " f"{statistics.channel_rate_fec_2_3_kbps:.6f} kbit/s" ) print( f" Channel rate FEC 1/2: " f"{statistics.channel_rate_fec_1_2_kbps:.6f} kbit/s" ) print( f" Timestamp mismatch: " f"{statistics.timestamp_mismatch_count}" ) print( f" Frame ID mismatch: " f"{statistics.frame_id_mismatch_count}" ) print("") print( "Two-pass JPEG sizes, source indices 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"Preview 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("Lab027E completed successfully.") if __name__ == "__main__": main()