""" Lab027. Многомасштабная передача изображения: общий кадр и фиксированная область интереса. Лабораторная сравнивает контрольный низкоскоростной JPEG-поток с восемью профилями, в которых независимо передаются: - BASE — уменьшенный общий grayscale-кадр сцены; - ROI — центральная нижняя область исходного кадра с отдельными разрешением, частотой обновления и JPEG quality. Все JPEG-кадры кодируются и декодируются только в памяти. Радиопротокол, CRC, FEC, фрагментация и повторные передачи в расчёт не входят. """ from __future__ import annotations import csv import math from dataclasses import dataclass, field from pathlib import Path import cv2 import matplotlib.pyplot as plt import numpy as np SOURCE_VIDEO_PATH = Path("data/raw/lab026_rover_source.mp4") OUTPUT_DIRECTORY = Path("data/processed/lab027") CSV_PATH = OUTPUT_DIRECTORY / "lab027_profiles.csv" REPORT_PATH = OUTPUT_DIRECTORY / "lab027_report.txt" BITRATE_PLOT_PATH = ( OUTPUT_DIRECTORY / "lab027_bitrate_comparison.png" ) QUALITY_PLOT_PATH = ( OUTPUT_DIRECTORY / "lab027_quality_comparison.png" ) CANDIDATE_IMAGE_PATH = ( OUTPUT_DIRECTORY / "lab027_candidate_profiles.png" ) EXPECTED_OUTPUT_PATHS = [ CSV_PATH, REPORT_PATH, BITRATE_PLOT_PATH, QUALITY_PLOT_PATH, CANDIDATE_IMAGE_PATH, ] BASELINE_BITRATE_KBPS = 80.283660 FRAME_TIME_EPSILON_SECONDS = 1e-9 MAXIMUM_PSNR_DB = 100.0 ROI_X_MIN = 0.20 ROI_X_MAX = 0.80 ROI_Y_MIN = 0.42 ROI_Y_MAX = 1.00 RECONSTRUCTION_WIDTH = 640 RECONSTRUCTION_HEIGHT = 360 CSV_FIELD_NAMES = [ "profile_name", "base_width", "base_height", "base_fps", "base_quality", "roi_enabled", "roi_x_min", "roi_y_min", "roi_x_max", "roi_y_max", "roi_width", "roi_height", "roi_fps", "roi_quality", "source_duration_s", "selected_base_frames", "selected_roi_frames", "actual_base_fps", "actual_roi_fps", "total_base_bytes", "total_roi_bytes", "total_payload_bytes", "mean_base_frame_bytes", "mean_roi_frame_bytes", "p95_base_frame_bytes", "p95_roi_frame_bytes", "max_base_frame_bytes", "max_roi_frame_bytes", "base_bitrate_kbps", "roi_bitrate_kbps", "total_payload_bitrate_kbps", "baseline_bitrate_kbps", "difference_from_baseline_kbps", "percent_of_baseline", "mean_base_jpeg_psnr_db", "mean_roi_jpeg_psnr_db", "mean_reconstructed_full_psnr_db", "mean_reconstructed_roi_psnr_db", ] @dataclass(frozen=True) class RoiProfile: """ Описывает один фиксированный профиль передачи BASE и ROI. """ profile_name: str base_width: int base_height: int base_fps: float base_quality: int roi_enabled: bool roi_width: int = 0 roi_height: int = 0 roi_fps: float = 0.0 roi_quality: int = 0 @dataclass(frozen=True) class RoiResult: """ Хранит итоговые метрики одного профиля Lab027. """ profile_name: str base_width: int base_height: int base_fps: float base_quality: int roi_enabled: bool roi_x_min: float roi_y_min: float roi_x_max: float roi_y_max: float roi_width: int roi_height: int roi_fps: float roi_quality: int source_duration_s: float selected_base_frames: int selected_roi_frames: int actual_base_fps: float actual_roi_fps: float total_base_bytes: int total_roi_bytes: int total_payload_bytes: int mean_base_frame_bytes: float mean_roi_frame_bytes: float p95_base_frame_bytes: float p95_roi_frame_bytes: float max_base_frame_bytes: int max_roi_frame_bytes: int base_bitrate_kbps: float roi_bitrate_kbps: float total_payload_bitrate_kbps: float baseline_bitrate_kbps: float difference_from_baseline_kbps: float percent_of_baseline: float mean_base_jpeg_psnr_db: float mean_roi_jpeg_psnr_db: float | None mean_reconstructed_full_psnr_db: float mean_reconstructed_roi_psnr_db: float | None @dataclass class ProfileAccumulator: """ Накапливает размеры, PSNR и последнее состояние потоков. """ next_base_time: float = 0.0 next_roi_time: float = 0.0 base_sizes_bytes: list[int] = field(default_factory=list) roi_sizes_bytes: list[int] = field(default_factory=list) base_jpeg_psnr_values_db: list[float] = field( default_factory=list ) roi_jpeg_psnr_values_db: list[float] = field( default_factory=list ) reconstructed_full_psnr_values_db: list[float] = field( default_factory=list ) reconstructed_roi_psnr_values_db: list[float] = field( default_factory=list ) latest_base_frame: np.ndarray | None = None latest_roi_frame: np.ndarray | None = None def read_video_metadata( source_path: Path, ) -> tuple[int, int, float, int, float, int]: """ Читает и проверяет основные параметры исходного видео. """ if not source_path.exists(): raise RuntimeError( 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)) source_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( "OpenCV вернул некорректное разрешение видео." ) if source_fps <= 0.0: raise RuntimeError("FPS исходного видео равен нулю.") if frame_count <= 0: raise RuntimeError("Число кадров исходного видео равно нулю.") duration_seconds = frame_count / source_fps file_size_bytes = source_path.stat().st_size if duration_seconds <= 0.0 or file_size_bytes <= 0: raise RuntimeError( "Длительность или размер исходного видео некорректны." ) return ( width, height, source_fps, frame_count, duration_seconds, file_size_bytes, ) def build_profiles() -> list[RoiProfile]: """ Создаёт ровно девять заданных в Lab027 профилей. """ profiles = [ RoiProfile( "baseline_320x180_gray_2fps_q40", 320, 180, 2.0, 40, False, ), RoiProfile( "base240_1fps_q25_roi320_2fps_q35", 240, 135, 1.0, 25, True, 320, 180, 2.0, 35, ), RoiProfile( "base240_0_5fps_q25_roi320_2fps_q35", 240, 135, 0.5, 25, True, 320, 180, 2.0, 35, ), RoiProfile( "base160_1fps_q25_roi320_2fps_q35", 160, 90, 1.0, 25, True, 320, 180, 2.0, 35, ), RoiProfile( "base240_1fps_q25_roi448_1fps_q30", 240, 135, 1.0, 25, True, 448, 252, 1.0, 30, ), RoiProfile( "base240_0_5fps_q25_roi448_1fps_q30", 240, 135, 0.5, 25, True, 448, 252, 1.0, 30, ), RoiProfile( "base320_1fps_q30_roi320_1fps_q35", 320, 180, 1.0, 30, True, 320, 180, 1.0, 35, ), RoiProfile( "base240_1fps_q20_roi320_1fps_q30", 240, 135, 1.0, 20, True, 320, 180, 1.0, 30, ), RoiProfile( "base160_0_5fps_q20_roi320_2fps_q30", 160, 90, 0.5, 20, True, 320, 180, 2.0, 30, ), ] if len(profiles) != 9: raise RuntimeError("Lab027 должна содержать ровно 9 профилей.") if len({profile.profile_name for profile in profiles}) != 9: raise RuntimeError("Имена профилей Lab027 не уникальны.") return profiles def normalized_roi_to_pixels( width: int, height: int, ) -> tuple[int, int, int, int]: """ Переводит фиксированные нормализованные границы ROI в пиксели. Возвращается полуоткрытый прямоугольник: x_min, y_min, x_max, y_max. """ if width <= 0 or height <= 0: raise ValueError("Размер кадра должен быть положительным.") 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)) x_min = min(max(x_min, 0), width - 1) x_max = min(max(x_max, x_min + 1), width) y_min = min(max(y_min, 0), height - 1) y_max = min(max(y_max, y_min + 1), height) return x_min, y_min, x_max, y_max def should_sample_frame( frame_time_seconds: float, next_sample_time_seconds: float, epsilon_seconds: float = FRAME_TIME_EPSILON_SECONDS, ) -> bool: """ Проверяет наступление времени очередного независимого обновления. """ return ( frame_time_seconds + epsilon_seconds >= next_sample_time_seconds ) def encode_decode_jpeg( gray_frame: np.ndarray, jpeg_quality: int, ) -> tuple[int, np.ndarray]: """ Кодирует grayscale JPEG в памяти и декодирует его обратно. """ if gray_frame.ndim != 2: raise RuntimeError( "JPEG Lab027 должен получать только grayscale-кадр." ) encoded_ok, encoded = cv2.imencode( ".jpg", gray_frame, [cv2.IMWRITE_JPEG_QUALITY, jpeg_quality], ) if not encoded_ok or encoded is None or encoded.size == 0: raise RuntimeError("OpenCV не смог закодировать JPEG.") decoded = cv2.imdecode( encoded, cv2.IMREAD_GRAYSCALE, ) if decoded is None or decoded.shape != gray_frame.shape: raise RuntimeError( "Декодированный JPEG имеет некорректный размер." ) return int(encoded.size), decoded def calculate_psnr( reference_frame: np.ndarray, reconstructed_frame: np.ndarray, ) -> float: """ Рассчитывает PSNR в float64, используя 100 дБ при MSE=0. """ if reference_frame.shape != reconstructed_frame.shape: raise RuntimeError( "Нельзя рассчитать PSNR для разных размеров кадров." ) difference = ( reference_frame.astype(np.float64) - reconstructed_frame.astype(np.float64) ) mean_squared_error = float(np.mean(difference ** 2)) if mean_squared_error == 0.0: return MAXIMUM_PSNR_DB return min( 10.0 * math.log10( (255.0 ** 2) / mean_squared_error ), MAXIMUM_PSNR_DB, ) def reconstruct_frame( decoded_base: np.ndarray, decoded_roi: np.ndarray | None, roi_enabled: bool, reconstruction_roi: tuple[int, int, int, int], ) -> np.ndarray: """ Восстанавливает итоговый grayscale-кадр размером 640x360. """ reconstructed = cv2.resize( decoded_base, (RECONSTRUCTION_WIDTH, RECONSTRUCTION_HEIGHT), interpolation=cv2.INTER_LINEAR, ) if not roi_enabled: return reconstructed if decoded_roi is None: raise RuntimeError( "Для ROI-профиля отсутствует декодированное состояние ROI." ) x_min, y_min, x_max, y_max = reconstruction_roi roi_width = x_max - x_min roi_height = y_max - y_min resized_roi = cv2.resize( decoded_roi, (roi_width, roi_height), interpolation=cv2.INTER_LINEAR, ) reconstructed[y_min:y_max, x_min:x_max] = resized_roi return reconstructed def safe_mean(values: list[int]) -> float: """ Возвращает среднее списка либо ноль для отсутствующего ROI. """ if not values: return 0.0 return float(np.mean(np.asarray(values, dtype=np.float64))) def safe_percentile(values: list[int], percentile: float) -> float: """ Возвращает процентиль списка либо ноль для отсутствующего ROI. """ if not values: return 0.0 return float( np.percentile( np.asarray(values, dtype=np.float64), percentile, ) ) def process_profiles( source_path: Path, profiles: list[RoiProfile], source_width: int, source_height: int, source_fps: float, source_frame_count: int, source_duration_seconds: float, ) -> list[RoiResult]: """ Обрабатывает все профили за один последовательный проход по видео. BASE и ROI обновляются независимо. Итоговая реконструкция оценивается на каждом исходном кадре с удержанием последнего декодированного состояния обоих потоков. """ if len(profiles) != 9: raise RuntimeError("Ожидалось ровно 9 профилей Lab027.") accumulators = { profile.profile_name: ProfileAccumulator() for profile in profiles } source_roi = normalized_roi_to_pixels( source_width, source_height, ) reconstruction_roi = normalized_roi_to_pixels( RECONSTRUCTION_WIDTH, RECONSTRUCTION_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( "Размер прочитанного кадра отличается от метаданных." ) frame_time_seconds = frame_index / source_fps source_gray = cv2.cvtColor( source_frame, cv2.COLOR_BGR2GRAY, ) source_roi_gray = source_gray[ source_roi[1]:source_roi[3], source_roi[0]:source_roi[2], ] reference_full = cv2.resize( source_gray, (RECONSTRUCTION_WIDTH, RECONSTRUCTION_HEIGHT), interpolation=cv2.INTER_AREA, ) reference_roi = reference_full[ reconstruction_roi[1]:reconstruction_roi[3], reconstruction_roi[0]:reconstruction_roi[2], ] if source_roi_gray.size == 0 or reference_roi.size == 0: raise RuntimeError("Рассчитана пустая область ROI.") for profile in profiles: accumulator = accumulators[profile.profile_name] if should_sample_frame( frame_time_seconds, accumulator.next_base_time, ): prepared_base = cv2.resize( source_gray, (profile.base_width, profile.base_height), interpolation=cv2.INTER_AREA, ) base_size, decoded_base = encode_decode_jpeg( prepared_base, profile.base_quality, ) accumulator.base_sizes_bytes.append(base_size) accumulator.base_jpeg_psnr_values_db.append( calculate_psnr( prepared_base, decoded_base, ) ) accumulator.latest_base_frame = decoded_base accumulator.next_base_time += ( 1.0 / profile.base_fps ) if ( profile.roi_enabled and should_sample_frame( frame_time_seconds, accumulator.next_roi_time, ) ): prepared_roi = cv2.resize( source_roi_gray, (profile.roi_width, profile.roi_height), interpolation=cv2.INTER_AREA, ) roi_size, decoded_roi = encode_decode_jpeg( prepared_roi, profile.roi_quality, ) accumulator.roi_sizes_bytes.append(roi_size) accumulator.roi_jpeg_psnr_values_db.append( calculate_psnr( prepared_roi, decoded_roi, ) ) accumulator.latest_roi_frame = decoded_roi accumulator.next_roi_time += ( 1.0 / profile.roi_fps ) if accumulator.latest_base_frame is None: raise RuntimeError( "BASE не был инициализирован при времени 0." ) reconstructed = reconstruct_frame( accumulator.latest_base_frame, accumulator.latest_roi_frame, profile.roi_enabled, reconstruction_roi, ) accumulator.reconstructed_full_psnr_values_db.append( calculate_psnr( reference_full, reconstructed, ) ) if profile.roi_enabled: reconstructed_roi = reconstructed[ reconstruction_roi[1]:reconstruction_roi[3], reconstruction_roi[0]:reconstruction_roi[2], ] accumulator.reconstructed_roi_psnr_values_db.append( calculate_psnr( reference_roi, reconstructed_roi, ) ) frame_index += 1 if ( frame_index % 100 == 0 or frame_index == source_frame_count ): print( f" Обработано кадров: " f"{frame_index}/{source_frame_count}" ) finally: capture.release() if frame_index == 0: raise RuntimeError("Не прочитан ни один кадр исходного видео.") if frame_index != source_frame_count: print( "Предупреждение: фактическое число кадров " f"{frame_index}, по метаданным {source_frame_count}." ) results: list[RoiResult] = [] for profile in profiles: accumulator = accumulators[profile.profile_name] if not accumulator.base_sizes_bytes: raise RuntimeError( f"BASE не содержит кадров: {profile.profile_name}" ) if ( profile.roi_enabled and not accumulator.roi_sizes_bytes ): raise RuntimeError( f"ROI не содержит кадров: {profile.profile_name}" ) selected_base_frames = len( accumulator.base_sizes_bytes ) selected_roi_frames = len( accumulator.roi_sizes_bytes ) total_base_bytes = int( sum(accumulator.base_sizes_bytes) ) total_roi_bytes = int( sum(accumulator.roi_sizes_bytes) ) total_payload_bytes = ( total_base_bytes + total_roi_bytes ) actual_base_fps = ( selected_base_frames / source_duration_seconds ) actual_roi_fps = ( selected_roi_frames / source_duration_seconds ) 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 total_payload_bitrate_kbps <= 0.0: raise RuntimeError( f"Неположительный битрейт: {profile.profile_name}" ) if not math.isclose( total_payload_bitrate_kbps, base_bitrate_kbps + roi_bitrate_kbps, rel_tol=1e-12, abs_tol=1e-12, ): raise RuntimeError( "Суммарный битрейт не равен сумме BASE и ROI." ) mean_roi_jpeg_psnr = ( float( np.mean( accumulator.roi_jpeg_psnr_values_db ) ) if profile.roi_enabled else None ) mean_reconstructed_roi_psnr = ( float( np.mean( accumulator.reconstructed_roi_psnr_values_db ) ) if profile.roi_enabled else None ) results.append( RoiResult( profile_name=profile.profile_name, base_width=profile.base_width, base_height=profile.base_height, base_fps=profile.base_fps, base_quality=profile.base_quality, roi_enabled=profile.roi_enabled, roi_x_min=( ROI_X_MIN if profile.roi_enabled else 0.0 ), roi_y_min=( ROI_Y_MIN if profile.roi_enabled else 0.0 ), roi_x_max=( ROI_X_MAX if profile.roi_enabled else 0.0 ), roi_y_max=( ROI_Y_MAX if profile.roi_enabled else 0.0 ), roi_width=( profile.roi_width if profile.roi_enabled else 0 ), roi_height=( profile.roi_height if profile.roi_enabled else 0 ), roi_fps=( profile.roi_fps if profile.roi_enabled else 0.0 ), roi_quality=( profile.roi_quality if profile.roi_enabled else 0 ), source_duration_s=source_duration_seconds, selected_base_frames=selected_base_frames, selected_roi_frames=selected_roi_frames, actual_base_fps=actual_base_fps, actual_roi_fps=actual_roi_fps, total_base_bytes=total_base_bytes, total_roi_bytes=total_roi_bytes, total_payload_bytes=total_payload_bytes, mean_base_frame_bytes=safe_mean( accumulator.base_sizes_bytes ), mean_roi_frame_bytes=safe_mean( accumulator.roi_sizes_bytes ), p95_base_frame_bytes=safe_percentile( accumulator.base_sizes_bytes, 95, ), p95_roi_frame_bytes=safe_percentile( accumulator.roi_sizes_bytes, 95, ), max_base_frame_bytes=max( accumulator.base_sizes_bytes ), max_roi_frame_bytes=( max(accumulator.roi_sizes_bytes) if accumulator.roi_sizes_bytes else 0 ), base_bitrate_kbps=base_bitrate_kbps, roi_bitrate_kbps=roi_bitrate_kbps, total_payload_bitrate_kbps=( total_payload_bitrate_kbps ), baseline_bitrate_kbps=BASELINE_BITRATE_KBPS, difference_from_baseline_kbps=( total_payload_bitrate_kbps - BASELINE_BITRATE_KBPS ), percent_of_baseline=( total_payload_bitrate_kbps / BASELINE_BITRATE_KBPS * 100.0 ), mean_base_jpeg_psnr_db=float( np.mean( accumulator.base_jpeg_psnr_values_db ) ), mean_roi_jpeg_psnr_db=mean_roi_jpeg_psnr, mean_reconstructed_full_psnr_db=float( np.mean( accumulator .reconstructed_full_psnr_values_db ) ), mean_reconstructed_roi_psnr_db=( mean_reconstructed_roi_psnr ), ) ) if len(results) != 9: raise RuntimeError("Создано не 9 результатов Lab027.") return results def optional_float_text(value: float | None) -> str: """ Представляет необязательное значение PSNR для CSV и отчёта. """ if value is None: return "" return f"{value:.6f}" def save_csv(results: list[RoiResult]) -> None: """ Сохраняет девять результатов Lab027 в 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 result in results: writer.writerow( { "profile_name": result.profile_name, "base_width": result.base_width, "base_height": result.base_height, "base_fps": f"{result.base_fps:.6f}", "base_quality": result.base_quality, "roi_enabled": int(result.roi_enabled), "roi_x_min": f"{result.roi_x_min:.6f}", "roi_y_min": f"{result.roi_y_min:.6f}", "roi_x_max": f"{result.roi_x_max:.6f}", "roi_y_max": f"{result.roi_y_max:.6f}", "roi_width": result.roi_width, "roi_height": result.roi_height, "roi_fps": f"{result.roi_fps:.6f}", "roi_quality": result.roi_quality, "source_duration_s": ( f"{result.source_duration_s:.6f}" ), "selected_base_frames": ( result.selected_base_frames ), "selected_roi_frames": ( result.selected_roi_frames ), "actual_base_fps": ( f"{result.actual_base_fps:.6f}" ), "actual_roi_fps": ( f"{result.actual_roi_fps:.6f}" ), "total_base_bytes": result.total_base_bytes, "total_roi_bytes": result.total_roi_bytes, "total_payload_bytes": ( result.total_payload_bytes ), "mean_base_frame_bytes": ( f"{result.mean_base_frame_bytes:.3f}" ), "mean_roi_frame_bytes": ( f"{result.mean_roi_frame_bytes:.3f}" ), "p95_base_frame_bytes": ( f"{result.p95_base_frame_bytes:.3f}" ), "p95_roi_frame_bytes": ( f"{result.p95_roi_frame_bytes:.3f}" ), "max_base_frame_bytes": ( result.max_base_frame_bytes ), "max_roi_frame_bytes": ( result.max_roi_frame_bytes ), "base_bitrate_kbps": ( f"{result.base_bitrate_kbps:.6f}" ), "roi_bitrate_kbps": ( f"{result.roi_bitrate_kbps:.6f}" ), "total_payload_bitrate_kbps": ( f"{result.total_payload_bitrate_kbps:.6f}" ), "baseline_bitrate_kbps": ( f"{result.baseline_bitrate_kbps:.6f}" ), "difference_from_baseline_kbps": ( f"{result.difference_from_baseline_kbps:.6f}" ), "percent_of_baseline": ( f"{result.percent_of_baseline:.6f}" ), "mean_base_jpeg_psnr_db": ( f"{result.mean_base_jpeg_psnr_db:.6f}" ), "mean_roi_jpeg_psnr_db": optional_float_text( result.mean_roi_jpeg_psnr_db ), "mean_reconstructed_full_psnr_db": ( f"{result.mean_reconstructed_full_psnr_db:.6f}" ), "mean_reconstructed_roi_psnr_db": ( optional_float_text( result .mean_reconstructed_roi_psnr_db ) ), } ) def short_profile_label(profile_name: str) -> str: """ Создаёт компактную многострочную подпись профиля для графиков. """ return profile_name.replace("_roi", "\nroi", 1) def save_bitrate_plot(results: list[RoiResult]) -> None: """ Строит составные столбцы BASE/ROI и контрольные линии битрейта. """ x_positions = np.arange(len(results)) base_values = [ result.base_bitrate_kbps for result in results ] roi_values = [ result.roi_bitrate_kbps for result in results ] figure, axes = plt.subplots(figsize=(15, 8)) axes.bar( x_positions, base_values, label="BASE bitrate", ) axes.bar( x_positions, roi_values, bottom=base_values, label="ROI bitrate", ) axes.axhline( BASELINE_BITRATE_KBPS, linestyle="--", label=f"Baseline {BASELINE_BITRATE_KBPS:.6f} kbps", ) axes.axhline( 100.0, linestyle=":", label="100 kbps", ) for position, result in zip(x_positions, results): axes.annotate( f"{result.total_payload_bitrate_kbps:.1f}", ( position, result.total_payload_bitrate_kbps, ), xytext=(0, 5), textcoords="offset points", ha="center", fontsize=8, ) axes.set_xticks( x_positions, [ short_profile_label(result.profile_name) for result in results ], rotation=35, ha="right", ) axes.set_ylabel("Полезный битрейт, кбит/с") axes.set_title( "Lab027. Распределение payload bitrate между BASE и ROI" ) axes.legend() axes.grid(True, axis="y") figure.tight_layout() figure.savefig(BITRATE_PLOT_PATH, dpi=160) plt.close(figure) def save_quality_plot(results: list[RoiResult]) -> None: """ Сравнивает средний PSNR полной реконструкции и области ROI. """ x_positions = np.arange(len(results)) bar_width = 0.38 full_values = [ result.mean_reconstructed_full_psnr_db for result in results ] roi_values = [ ( result.mean_reconstructed_roi_psnr_db if result.mean_reconstructed_roi_psnr_db is not None else np.nan ) for result in results ] figure, axes = plt.subplots(figsize=(15, 8)) axes.bar( x_positions - bar_width / 2, full_values, width=bar_width, label="Full-frame PSNR", ) axes.bar( x_positions + bar_width / 2, roi_values, width=bar_width, label="ROI PSNR", ) axes.set_xticks( x_positions, [ short_profile_label(result.profile_name) for result in results ], rotation=35, ha="right", ) axes.set_ylabel("Средний PSNR реконструкции, дБ") axes.set_title( "Lab027. Качество полной реконструкции и области ROI" ) axes.legend() axes.grid(True, axis="y") figure.tight_layout() figure.savefig(QUALITY_PLOT_PATH, dpi=160) plt.close(figure) def read_representative_frame( source_path: Path, frame_count: int, ) -> np.ndarray: """ Читает кадр около середины ролика для сравнительного изображения. """ capture = cv2.VideoCapture(str(source_path)) if not capture.isOpened(): raise RuntimeError( f"OpenCV не смог открыть видео: {source_path}" ) middle_index = frame_count // 2 try: capture.set( cv2.CAP_PROP_POS_FRAMES, middle_index, ) frame_read, frame = capture.read() finally: capture.release() if not frame_read or frame is None: raise RuntimeError( "Не удалось прочитать репрезентативный кадр." ) return frame def representative_reconstruction( source_frame: np.ndarray, profile: RoiProfile, source_roi: tuple[int, int, int, int], reconstruction_roi: tuple[int, int, int, int], ) -> np.ndarray: """ Кодирует BASE/ROI репрезентативного кадра и реконструирует его. """ source_gray = cv2.cvtColor( source_frame, cv2.COLOR_BGR2GRAY, ) prepared_base = cv2.resize( source_gray, (profile.base_width, profile.base_height), interpolation=cv2.INTER_AREA, ) _, decoded_base = encode_decode_jpeg( prepared_base, profile.base_quality, ) decoded_roi: np.ndarray | None = None if profile.roi_enabled: source_roi_gray = source_gray[ source_roi[1]:source_roi[3], source_roi[0]:source_roi[2], ] prepared_roi = cv2.resize( source_roi_gray, (profile.roi_width, profile.roi_height), interpolation=cv2.INTER_AREA, ) _, decoded_roi = encode_decode_jpeg( prepared_roi, profile.roi_quality, ) return reconstruct_frame( decoded_base, decoded_roi, profile.roi_enabled, reconstruction_roi, ) def save_candidate_image( source_path: Path, profiles: list[RoiProfile], results: list[RoiResult], source_width: int, source_height: int, source_frame_count: int, ) -> None: """ Создаёт девять панелей реконструкции репрезентативного кадра. """ source_frame = read_representative_frame( source_path, source_frame_count, ) source_roi = normalized_roi_to_pixels( source_width, source_height, ) reconstruction_roi = normalized_roi_to_pixels( RECONSTRUCTION_WIDTH, RECONSTRUCTION_HEIGHT, ) result_by_name = { result.profile_name: result for result in results } figure, axes = plt.subplots( 3, 3, figsize=(18, 12), ) for axes_item, profile in zip(axes.flat, profiles): result = result_by_name[profile.profile_name] reconstructed = representative_reconstruction( source_frame, profile, source_roi, reconstruction_roi, ) displayed = reconstructed.copy() cv2.rectangle( displayed, ( reconstruction_roi[0], reconstruction_roi[1], ), ( reconstruction_roi[2] - 1, reconstruction_roi[3] - 1, ), 255, 2, ) if result.mean_reconstructed_roi_psnr_db is None: roi_psnr_text = "ROI отсутствует" else: roi_psnr_text = ( "ROI PSNR=" f"{result.mean_reconstructed_roi_psnr_db:.2f} dB" ) axes_item.imshow( displayed, cmap="gray", vmin=0, vmax=255, interpolation="nearest", ) axes_item.set_title( f"{profile.profile_name}\n" f"total={result.total_payload_bitrate_kbps:.2f} kbps, " f"BASE={result.base_bitrate_kbps:.2f}, " f"ROI={result.roi_bitrate_kbps:.2f}\n" f"full PSNR=" f"{result.mean_reconstructed_full_psnr_db:.2f} dB, " f"{roi_psnr_text}", fontsize=8, ) axes_item.axis("off") figure.suptitle( "Lab027. Общий кадр и фиксированная область интереса", fontsize=14, ) figure.tight_layout() figure.savefig(CANDIDATE_IMAGE_PATH, dpi=160) plt.close(figure) def format_result_line(result: RoiResult) -> str: """ Формирует подробную строку одного результата для отчёта. """ roi_jpeg_psnr = ( f"{result.mean_roi_jpeg_psnr_db:.6f}" if result.mean_roi_jpeg_psnr_db is not None else "нет" ) reconstructed_roi_psnr = ( f"{result.mean_reconstructed_roi_psnr_db:.6f}" if result.mean_reconstructed_roi_psnr_db is not None else "нет" ) return ( f"{result.profile_name}: " f"BASE={result.base_width}x{result.base_height}, " f"{result.base_fps:.3f} fps, Q{result.base_quality}, " f"base_frames={result.selected_base_frames}, " f"base={result.base_bitrate_kbps:.6f} kbit/s; " f"ROI=" f"{result.roi_width}x{result.roi_height}, " f"{result.roi_fps:.3f} fps, Q{result.roi_quality}, " f"roi_frames={result.selected_roi_frames}, " f"roi={result.roi_bitrate_kbps:.6f} kbit/s; " f"total={result.total_payload_bitrate_kbps:.6f} kbit/s, " f"baseline_diff=" f"{result.difference_from_baseline_kbps:.6f}, " f"baseline_percent={result.percent_of_baseline:.3f}%, " f"BASE_JPEG_PSNR=" f"{result.mean_base_jpeg_psnr_db:.6f} dB, " f"ROI_JPEG_PSNR={roi_jpeg_psnr} dB, " f"full_reconstructed_PSNR=" f"{result.mean_reconstructed_full_psnr_db:.6f} dB, " f"ROI_reconstructed_PSNR=" f"{reconstructed_roi_psnr} dB" ) def append_result_group( lines: list[str], heading: str, results: list[RoiResult], ) -> None: """ Добавляет в отчёт группу результатов либо отметку об отсутствии. """ lines.append(heading) if not results: lines.append("Нет.") else: lines.extend(format_result_line(result) for result in results) lines.append("") def write_report( results: list[RoiResult], source_width: int, source_height: int, source_fps: float, source_frame_count: int, source_duration_seconds: float, source_file_size_bytes: int, ) -> None: """ Сохраняет подробный UTF-8 отчёт Lab027. """ under_baseline = [ result for result in results if result.total_payload_bitrate_kbps <= BASELINE_BITRATE_KBPS + 1e-9 ] under_100 = [ result for result in results if result.total_payload_bitrate_kbps <= 100.0 ] roi_under_100 = [ result for result in under_100 if result.roi_enabled ] maximum_roi_resolution = ( max( roi_under_100, key=lambda item: ( item.roi_width * item.roi_height, item.mean_reconstructed_roi_psnr_db if item.mean_reconstructed_roi_psnr_db is not None else -math.inf, ), ) if roi_under_100 else None ) best_roi_psnr = ( max( roi_under_100, key=lambda item: ( item.mean_reconstructed_roi_psnr_db if item.mean_reconstructed_roi_psnr_db is not None else -math.inf ), ) if roi_under_100 else None ) best_full_psnr = ( max( under_100, key=lambda item: ( item.mean_reconstructed_full_psnr_db ), ) if under_100 else None ) source_roi = normalized_roi_to_pixels( source_width, source_height, ) lines = [ "Lab027. Многомасштабная передача изображения: " "общий кадр и область интереса", "", "Цель: проверить совместную передачу общего кадра и " "фиксированной ROI при payload около 80–100 кбит/с.", f"Исходное видео: {SOURCE_VIDEO_PATH}", f"Размер файла: {source_file_size_bytes} байт", f"Исходное разрешение: {source_width}×{source_height}", f"Исходный FPS: {source_fps:.6f}", f"Число кадров: {source_frame_count}", f"Длительность: {source_duration_seconds:.6f} с", "", "BASE: полный grayscale-кадр пониженного разрешения " "с собственной частотой и JPEG quality.", "ROI: фиксированная центральная нижняя область исходного " "grayscale-кадра с независимыми разрешением, частотой " "и JPEG quality.", "Нормализованные координаты ROI: " f"x={ROI_X_MIN:.2f}...{ROI_X_MAX:.2f}, " f"y={ROI_Y_MIN:.2f}...{ROI_Y_MAX:.2f}.", "Пиксельные координаты ROI исходного кадра: " f"x={source_roi[0]}...{source_roi[2]}, " f"y={source_roi[1]}...{source_roi[3]}.", "", "Контрольный профиль: 320×180 grayscale, 2 fps, " "JPEG Q40.", f"Контрольный baseline: {BASELINE_BITRATE_KBPS:.6f} " "кбит/с.", "", "Результаты всех девяти профилей:", ] lines.extend(format_result_line(result) for result in results) lines.append("") append_result_group( lines, "Профили до 80.283660 кбит/с:", under_baseline, ) append_result_group( lines, "Профили до 100 кбит/с:", under_100, ) append_result_group( lines, "Профиль с максимальным ROI-разрешением до 100 кбит/с:", ( [maximum_roi_resolution] if maximum_roi_resolution is not None else [] ), ) append_result_group( lines, "Профиль с максимальным ROI PSNR до 100 кбит/с:", [best_roi_psnr] if best_roi_psnr is not None else [], ) append_result_group( lines, "Профиль с максимальным full-frame PSNR " "до 100 кбит/с:", [best_full_psnr] if best_full_psnr is not None else [], ) lines.append("Разделение битрейта BASE/ROI:") for result in results: lines.append( f"{result.profile_name}: " f"BASE={result.base_bitrate_kbps:.6f} кбит/с, " f"ROI={result.roi_bitrate_kbps:.6f} кбит/с, " f"total={result.total_payload_bitrate_kbps:.6f} " "кбит/с." ) lines.extend( [ "", "Фиксированная ROI не гарантирует попадания препятствия " "в область интереса.", "PSNR не заменяет ручную оценку пригодности изображения.", "Нельзя автоматически объявлять профиль безопасным " "для управления ровером.", "Радиопротокол, радиозаголовки, CRC, FEC, " "фрагментация и повторные передачи не учтены.", f"Сравнительное изображение: {CANDIDATE_IMAGE_PATH}", "Следующий шаг: ручная оценка пользователем.", "", ] ) REPORT_PATH.write_text( "\n".join(lines), encoding="utf-8", ) def validate_output_files() -> None: """ Проверяет наличие и ненулевой размер пяти результатов Lab027. """ for path in EXPECTED_OUTPUT_PATHS: if not path.exists(): raise RuntimeError( f"Не создан ожидаемый результат: {path}" ) if path.stat().st_size <= 0: raise RuntimeError( f"Создан пустой результат: {path}" ) def main() -> None: """ Выполняет полный эксперимент Lab027. """ print("Проверка исходного видео Lab027...") ( 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_VIDEO_PATH}") print(f" Размер: {source_file_size_bytes} байт") print( f" Разрешение: {source_width}x{source_height}" ) print(f" FPS: {source_fps:.6f}") print(f" Кадров: {source_frame_count}") print(f" Длительность: {source_duration_seconds:.6f} с") profiles = build_profiles() print(f"Профилей: {len(profiles)}") OUTPUT_DIRECTORY.mkdir( parents=True, exist_ok=True, ) print("Обработка BASE и ROI...") results = process_profiles( source_path=SOURCE_VIDEO_PATH, profiles=profiles, source_width=source_width, source_height=source_height, source_fps=source_fps, source_frame_count=source_frame_count, source_duration_seconds=source_duration_seconds, ) print("Сохранение CSV...") save_csv(results) print("Построение графика битрейта...") save_bitrate_plot(results) print("Построение графика качества...") save_quality_plot(results) print("Создание сравнительного изображения...") save_candidate_image( source_path=SOURCE_VIDEO_PATH, profiles=profiles, results=results, source_width=source_width, source_height=source_height, source_frame_count=source_frame_count, ) print("Сохранение отчёта...") write_report( results=results, source_width=source_width, source_height=source_height, source_fps=source_fps, source_frame_count=source_frame_count, source_duration_seconds=source_duration_seconds, source_file_size_bytes=source_file_size_bytes, ) validate_output_files() print("") print("Результаты:") for result in results: print( f" {result.profile_name}: " f"BASE={result.base_bitrate_kbps:.6f}, " f"ROI={result.roi_bitrate_kbps:.6f}, " f"total={result.total_payload_bitrate_kbps:.6f} " "кбит/с, " f"full PSNR=" f"{result.mean_reconstructed_full_psnr_db:.6f} дБ, " "ROI PSNR=" + ( f"{result.mean_reconstructed_roi_psnr_db:.6f} дБ" if result.mean_reconstructed_roi_psnr_db is not None else "нет" ) ) print("") for path in EXPECTED_OUTPUT_PATHS: print(f" {path} ({path.stat().st_size} байт)") print("") print("Lab027 завершена успешно.") if __name__ == "__main__": main()