""" Lab026. Исследование уменьшения видеопотока для управления ровером. Лабораторная исследует только полезную нагрузку покадрового JPEG. Радиопередача, заголовки пакетов, CRC, FEC и повторные передачи на этом этапе не моделируются. Исходный MP4 открывается только для чтения. Отдельные JPEG-файлы на диск не сохраняются: кодирование и декодирование выполняются в оперативной памяти. """ from __future__ import annotations from dataclasses import dataclass from pathlib import Path import csv import math import textwrap import cv2 import matplotlib import numpy as np # Графики сохраняются в файлы без открытия блокирующих окон. matplotlib.use("Agg") import matplotlib.pyplot as plt # --------------------------------------------------------------------- # Пути к исходным и выходным данным # --------------------------------------------------------------------- SOURCE_VIDEO_PATH = Path("data/raw/lab026_rover_source.mp4") OUTPUT_DIRECTORY = Path("data/processed/lab026") PROFILES_CSV_PATH = OUTPUT_DIRECTORY / "lab026_profiles.csv" REPORT_PATH = OUTPUT_DIRECTORY / "lab026_report.txt" FPS_PLOT_PATH = OUTPUT_DIRECTORY / "lab026_bitrate_vs_fps.png" RESOLUTION_PLOT_PATH = ( OUTPUT_DIRECTORY / "lab026_bitrate_vs_resolution.png" ) QUALITY_PLOT_PATH = ( OUTPUT_DIRECTORY / "lab026_bitrate_vs_jpeg_quality.png" ) CANDIDATE_COMPARISON_PATH = ( OUTPUT_DIRECTORY / "lab026_candidate_profiles.png" ) EXPECTED_OUTPUT_PATHS = [ PROFILES_CSV_PATH, REPORT_PATH, FPS_PLOT_PATH, RESOLUTION_PLOT_PATH, QUALITY_PLOT_PATH, CANDIDATE_COMPARISON_PATH, ] # --------------------------------------------------------------------- # Имена экспериментов и общие параметры # --------------------------------------------------------------------- EXPERIMENT_FPS_SWEEP = "fps_sweep" EXPERIMENT_RESOLUTION_SWEEP = "resolution_sweep" EXPERIMENT_QUALITY_SWEEP = "quality_sweep" EXPERIMENT_CANDIDATES = "candidate_profiles" COLOR_MODE_COLOR = "color" COLOR_MODE_GRAYSCALE = "grayscale" FRAME_TIME_EPSILON_SECONDS = 1e-9 MAXIMUM_PSNR_DB = 100.0 CSV_FIELD_NAMES = [ "experiment", "profile_name", "width", "height", "target_fps", "actual_fps", "color_mode", "jpeg_quality", "source_duration_s", "selected_frames", "total_bytes", "mean_frame_bytes", "median_frame_bytes", "p95_frame_bytes", "max_frame_bytes", "payload_bitrate_bps", "payload_bitrate_kbps", "source_file_bitrate_kbps", "reduction_ratio", "percent_of_source", "mean_psnr_db", ] @dataclass(frozen=True) class VideoProfile: """ Описывает один исследуемый профиль покадрового JPEG. Атрибут target_fps задаёт частоту выбора кадров на временной шкале исходного видео. Разрешение и цветовой режим применяются до JPEG. """ experiment: str profile_name: str width: int height: int target_fps: float color_mode: str jpeg_quality: int @dataclass(frozen=True) class ProfileResult: """ Содержит измеренные характеристики одного видеопрофиля. """ experiment: str profile_name: str width: int height: int target_fps: float actual_fps: float color_mode: str jpeg_quality: int source_duration_s: float selected_frames: int total_bytes: int mean_frame_bytes: float median_frame_bytes: float p95_frame_bytes: float max_frame_bytes: int payload_bitrate_bps: float payload_bitrate_kbps: float source_file_bitrate_kbps: float reduction_ratio: float percent_of_source: float mean_psnr_db: float @dataclass class ProfileAccumulator: """ Хранит промежуточные значения при последовательном чтении видео. """ next_sample_time: float frame_sizes_bytes: list[int] psnr_values_db: list[float] def read_video_metadata( source_path: Path, ) -> tuple[int, int, float, int, float, int, float]: """ Читает основные метаданные исходного видео через OpenCV. Возвращает ширину, высоту, FPS, число кадров, длительность, размер файла и приблизительный файловый битрейт в кбит/с. """ if not source_path.is_file(): 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 source_file_size_bytes = source_path.stat().st_size if source_file_size_bytes <= 0: raise RuntimeError("Исходный видеофайл имеет нулевой размер.") source_file_bitrate_kbps = ( source_file_size_bytes * 8.0 / duration_seconds / 1000.0 ) return ( width, height, source_fps, frame_count, duration_seconds, source_file_size_bytes, source_file_bitrate_kbps, ) def format_number_for_name(value: float) -> str: """ Преобразует частоту кадров в безопасную часть имени профиля. """ if float(value).is_integer(): return str(int(value)) return str(value).replace(".", "_") def build_profiles() -> list[VideoProfile]: """ Формирует профили четырёх экспериментов Lab026. """ profiles: list[VideoProfile] = [] # Эксперимент A: влияние частоты кадров. for target_fps in [30, 10, 5, 2, 1, 0.5, 0.2]: profiles.append( VideoProfile( experiment=EXPERIMENT_FPS_SWEEP, profile_name=( "fps_" f"{format_number_for_name(float(target_fps))}" ), width=640, height=360, target_fps=float(target_fps), color_mode=COLOR_MODE_GRAYSCALE, jpeg_quality=50, ) ) # Эксперимент B: влияние разрешения. for width, height in [ (1280, 720), (960, 540), (640, 360), (480, 270), (320, 180), (160, 90), ]: profiles.append( VideoProfile( experiment=EXPERIMENT_RESOLUTION_SWEEP, profile_name=f"resolution_{width}x{height}", width=width, height=height, target_fps=5.0, color_mode=COLOR_MODE_GRAYSCALE, jpeg_quality=50, ) ) # Эксперимент C: влияние цвета и JPEG quality. for color_mode in [ COLOR_MODE_COLOR, COLOR_MODE_GRAYSCALE, ]: for jpeg_quality in [90, 70, 50, 30, 20]: profiles.append( VideoProfile( experiment=EXPERIMENT_QUALITY_SWEEP, profile_name=( f"quality_{color_mode}_q{jpeg_quality}" ), width=640, height=360, target_fps=5.0, color_mode=color_mode, jpeg_quality=jpeg_quality, ) ) # Эксперимент D: заранее заданные кандидаты радиоканала. profiles.extend( [ VideoProfile( EXPERIMENT_CANDIDATES, "candidate_640x360_gray_5fps_q50", 640, 360, 5.0, COLOR_MODE_GRAYSCALE, 50, ), VideoProfile( EXPERIMENT_CANDIDATES, "candidate_640x360_gray_2fps_q40", 640, 360, 2.0, COLOR_MODE_GRAYSCALE, 40, ), VideoProfile( EXPERIMENT_CANDIDATES, "candidate_320x180_color_5fps_q50", 320, 180, 5.0, COLOR_MODE_COLOR, 50, ), VideoProfile( EXPERIMENT_CANDIDATES, "candidate_320x180_gray_5fps_q50", 320, 180, 5.0, COLOR_MODE_GRAYSCALE, 50, ), VideoProfile( EXPERIMENT_CANDIDATES, "candidate_320x180_gray_2fps_q40", 320, 180, 2.0, COLOR_MODE_GRAYSCALE, 40, ), VideoProfile( EXPERIMENT_CANDIDATES, "candidate_320x180_gray_1fps_q35", 320, 180, 1.0, COLOR_MODE_GRAYSCALE, 35, ), VideoProfile( EXPERIMENT_CANDIDATES, "candidate_320x180_gray_0_5fps_q35", 320, 180, 0.5, COLOR_MODE_GRAYSCALE, 35, ), VideoProfile( EXPERIMENT_CANDIDATES, "candidate_320x180_gray_0_2fps_q35", 320, 180, 0.2, COLOR_MODE_GRAYSCALE, 35, ), VideoProfile( EXPERIMENT_CANDIDATES, "candidate_160x90_gray_1fps_q30", 160, 90, 1.0, COLOR_MODE_GRAYSCALE, 30, ), ] ) return profiles 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 prepare_frame( source_frame: np.ndarray, profile: VideoProfile, ) -> np.ndarray: """ Изменяет разрешение кадра и применяет заданный цветовой режим. """ resized_frame = cv2.resize( source_frame, (profile.width, profile.height), interpolation=cv2.INTER_AREA, ) if profile.color_mode == COLOR_MODE_GRAYSCALE: return cv2.cvtColor( resized_frame, cv2.COLOR_BGR2GRAY, ) if profile.color_mode == COLOR_MODE_COLOR: return resized_frame raise RuntimeError( f"Неизвестный цветовой режим: {profile.color_mode}" ) def encode_decode_jpeg( prepared_frame: np.ndarray, profile: VideoProfile, ) -> tuple[int, np.ndarray]: """ Кодирует и декодирует JPEG только в оперативной памяти. Возвращает размер JPEG в байтах и восстановленное изображение. """ encoding_succeeded, encoded_jpeg = cv2.imencode( ".jpg", prepared_frame, [ cv2.IMWRITE_JPEG_QUALITY, profile.jpeg_quality, ], ) if not encoding_succeeded or encoded_jpeg is None: raise RuntimeError( "JPEG encoding завершился ошибкой для профиля " f"{profile.profile_name}." ) decode_mode = ( cv2.IMREAD_GRAYSCALE if profile.color_mode == COLOR_MODE_GRAYSCALE else cv2.IMREAD_COLOR ) decoded_frame = cv2.imdecode(encoded_jpeg, decode_mode) if decoded_frame is None: raise RuntimeError( "JPEG decoding завершился ошибкой для профиля " f"{profile.profile_name}." ) if decoded_frame.shape != prepared_frame.shape: raise RuntimeError( "Размер восстановленного JPEG не совпадает с эталоном " f"для профиля {profile.profile_name}." ) return int(encoded_jpeg.size), decoded_frame def calculate_psnr( reference_frame: np.ndarray, decoded_frame: np.ndarray, ) -> float: """ Рассчитывает PSNR между подготовленным кадром и JPEG-декодированием. """ if reference_frame.shape != decoded_frame.shape: raise RuntimeError( "Невозможно рассчитать PSNR для разных размеров кадров." ) difference = ( reference_frame.astype(np.float64) - decoded_frame.astype(np.float64) ) mean_squared_error = float(np.mean(difference ** 2)) if mean_squared_error == 0.0: return MAXIMUM_PSNR_DB psnr_db = 10.0 * math.log10( (255.0 ** 2) / mean_squared_error ) return min(psnr_db, MAXIMUM_PSNR_DB) def process_profiles( source_path: Path, profiles: list[VideoProfile], source_fps: float, source_frame_count: int, source_duration_seconds: float, source_file_bitrate_kbps: float, ) -> list[ProfileResult]: """ За один последовательный проход измеряет все профили Lab026. """ if not profiles: raise RuntimeError("Список видеопрофилей пуст.") accumulators = { profile.profile_name: ProfileAccumulator( next_sample_time=0.0, frame_sizes_bytes=[], psnr_values_db=[], ) for profile in profiles } if len(accumulators) != len(profiles): raise RuntimeError("Обнаружены повторяющиеся имена профилей.") 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"OpenCV вернул пустой кадр с индексом {frame_index}." ) frame_time_seconds = frame_index / source_fps for profile in profiles: accumulator = accumulators[profile.profile_name] if not should_sample_frame( frame_time_seconds, accumulator.next_sample_time, ): continue prepared_frame = prepare_frame( source_frame, profile, ) jpeg_size_bytes, decoded_frame = ( encode_decode_jpeg( prepared_frame, profile, ) ) psnr_db = calculate_psnr( prepared_frame, decoded_frame, ) accumulator.frame_sizes_bytes.append( jpeg_size_bytes ) accumulator.psnr_values_db.append(psnr_db) accumulator.next_sample_time += ( 1.0 / profile.target_fps ) 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("OpenCV не прочитал ни одного кадра.") if frame_index != source_frame_count: print( "Предупреждение: фактически прочитанное число кадров " f"({frame_index}) отличается от метаданных " f"({source_frame_count})." ) results: list[ProfileResult] = [] for profile in profiles: accumulator = accumulators[profile.profile_name] if not accumulator.frame_sizes_bytes: raise RuntimeError( "Не выбран ни один кадр для профиля " f"{profile.profile_name}." ) frame_sizes = np.asarray( accumulator.frame_sizes_bytes, dtype=np.float64, ) psnr_values = np.asarray( accumulator.psnr_values_db, dtype=np.float64, ) selected_frames = len(accumulator.frame_sizes_bytes) total_bytes = int( sum(accumulator.frame_sizes_bytes) ) actual_fps = ( selected_frames / source_duration_seconds ) payload_bitrate_bps = ( total_bytes * 8.0 / source_duration_seconds ) payload_bitrate_kbps = payload_bitrate_bps / 1000.0 if payload_bitrate_kbps <= 0.0: raise RuntimeError( "Рассчитан неположительный payload bitrate " f"для профиля {profile.profile_name}." ) reduction_ratio = ( source_file_bitrate_kbps / payload_bitrate_kbps ) percent_of_source = ( payload_bitrate_kbps / source_file_bitrate_kbps * 100.0 ) results.append( ProfileResult( experiment=profile.experiment, profile_name=profile.profile_name, width=profile.width, height=profile.height, target_fps=profile.target_fps, actual_fps=actual_fps, color_mode=profile.color_mode, jpeg_quality=profile.jpeg_quality, source_duration_s=source_duration_seconds, selected_frames=selected_frames, total_bytes=total_bytes, mean_frame_bytes=float(np.mean(frame_sizes)), median_frame_bytes=float(np.median(frame_sizes)), p95_frame_bytes=float( np.percentile(frame_sizes, 95) ), max_frame_bytes=int(np.max(frame_sizes)), payload_bitrate_bps=payload_bitrate_bps, payload_bitrate_kbps=payload_bitrate_kbps, source_file_bitrate_kbps=( source_file_bitrate_kbps ), reduction_ratio=reduction_ratio, percent_of_source=percent_of_source, mean_psnr_db=float(np.mean(psnr_values)), ) ) if not results: raise RuntimeError("Не создан ни один результат Lab026.") return results def save_results_csv( results: list[ProfileResult], ) -> None: """ Сохраняет все измеренные профили в UTF-8 CSV с запятой. """ with PROFILES_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( { "experiment": result.experiment, "profile_name": result.profile_name, "width": result.width, "height": result.height, "target_fps": f"{result.target_fps:.6f}", "actual_fps": f"{result.actual_fps:.6f}", "color_mode": result.color_mode, "jpeg_quality": result.jpeg_quality, "source_duration_s": ( f"{result.source_duration_s:.6f}" ), "selected_frames": result.selected_frames, "total_bytes": result.total_bytes, "mean_frame_bytes": ( f"{result.mean_frame_bytes:.3f}" ), "median_frame_bytes": ( f"{result.median_frame_bytes:.3f}" ), "p95_frame_bytes": ( f"{result.p95_frame_bytes:.3f}" ), "max_frame_bytes": result.max_frame_bytes, "payload_bitrate_bps": ( f"{result.payload_bitrate_bps:.3f}" ), "payload_bitrate_kbps": ( f"{result.payload_bitrate_kbps:.6f}" ), "source_file_bitrate_kbps": ( f"{result.source_file_bitrate_kbps:.6f}" ), "reduction_ratio": ( f"{result.reduction_ratio:.6f}" ), "percent_of_source": ( f"{result.percent_of_source:.6f}" ), "mean_psnr_db": ( f"{result.mean_psnr_db:.6f}" ), } ) def annotate_points( axes: plt.Axes, x_values: list[float], y_values: list[float], ) -> None: """ Подписывает на графике значения payload bitrate. """ for x_value, y_value in zip(x_values, y_values): axes.annotate( f"{y_value:.1f}", (x_value, y_value), xytext=(0, 7), textcoords="offset points", ha="center", fontsize=8, ) def save_fps_plot( results: list[ProfileResult], ) -> None: """ Строит зависимость payload bitrate от целевого FPS. """ fps_results = sorted( ( result for result in results if result.experiment == EXPERIMENT_FPS_SWEEP ), key=lambda result: result.target_fps, ) x_values = [result.target_fps for result in fps_results] y_values = [ result.payload_bitrate_kbps for result in fps_results ] figure, axes = plt.subplots(figsize=(10, 6)) axes.plot(x_values, y_values, marker="o") axes.set_xscale("log") annotate_points(axes, x_values, y_values) axes.set_title( "Lab026. Влияние частоты кадров на JPEG payload bitrate" ) axes.set_xlabel("Целевая частота кадров, кадр/с") axes.set_ylabel("Полезный битрейт JPEG, кбит/с") axes.grid(True) figure.tight_layout() figure.savefig(FPS_PLOT_PATH, dpi=160) plt.close(figure) def save_resolution_plot( results: list[ProfileResult], ) -> None: """ Строит зависимость payload bitrate от числа пикселей кадра. """ resolution_results = sorted( ( result for result in results if result.experiment == EXPERIMENT_RESOLUTION_SWEEP ), key=lambda result: result.width * result.height, ) x_positions = list(range(len(resolution_results))) y_values = [ result.payload_bitrate_kbps for result in resolution_results ] labels = [ f"{result.width}×{result.height}" for result in resolution_results ] figure, axes = plt.subplots(figsize=(11, 6)) axes.plot(x_positions, y_values, marker="o") annotate_points(axes, x_positions, y_values) axes.set_xticks(x_positions, labels) axes.set_title( "Lab026. Влияние разрешения на JPEG payload bitrate" ) axes.set_xlabel("Разрешение кадра") axes.set_ylabel("Полезный битрейт JPEG, кбит/с") axes.grid(True) figure.tight_layout() figure.savefig(RESOLUTION_PLOT_PATH, dpi=160) plt.close(figure) def save_quality_plot( results: list[ProfileResult], ) -> None: """ Строит зависимости bitrate от JPEG quality для цвета и серого. """ figure, axes = plt.subplots(figsize=(10, 6)) for color_mode in [ COLOR_MODE_COLOR, COLOR_MODE_GRAYSCALE, ]: mode_results = sorted( ( result for result in results if ( result.experiment == EXPERIMENT_QUALITY_SWEEP and result.color_mode == color_mode ) ), key=lambda result: result.jpeg_quality, ) x_values = [ result.jpeg_quality for result in mode_results ] y_values = [ result.payload_bitrate_kbps for result in mode_results ] axes.plot( x_values, y_values, marker="o", label=color_mode, ) annotate_points(axes, x_values, y_values) axes.set_title( "Lab026. Влияние JPEG quality на полезный битрейт" ) axes.set_xlabel("JPEG quality") axes.set_ylabel("Полезный битрейт JPEG, кбит/с") axes.grid(True) axes.legend() figure.tight_layout() figure.savefig(QUALITY_PLOT_PATH, dpi=160) plt.close(figure) def read_representative_frame( source_path: Path, source_frame_count: int, ) -> np.ndarray: """ Читает один кадр приблизительно из середины исходного ролика. """ capture = cv2.VideoCapture(str(source_path)) if not capture.isOpened(): raise RuntimeError( f"OpenCV не смог открыть видео: {source_path}" ) middle_frame_index = source_frame_count // 2 try: capture.set( cv2.CAP_PROP_POS_FRAMES, middle_frame_index, ) frame_read, frame = capture.read() finally: capture.release() if not frame_read or frame is None: raise RuntimeError( "Не удалось прочитать репрезентативный кадр " f"с индексом {middle_frame_index}." ) return frame def save_candidate_comparison( source_path: Path, source_frame_count: int, profiles: list[VideoProfile], results: list[ProfileResult], ) -> None: """ Создаёт таблицу восстановленных кадров девяти кандидатов. """ candidate_profiles = [ profile for profile in profiles if profile.experiment == EXPERIMENT_CANDIDATES ] if len(candidate_profiles) != 9: raise RuntimeError( "Для сравнения ожидалось ровно девять " "candidate-профилей." ) results_by_name = { result.profile_name: result for result in results } representative_frame = read_representative_frame( source_path, source_frame_count, ) figure, axes_grid = plt.subplots( 3, 3, figsize=(18, 13), ) axes = axes_grid.ravel() for axes_item, profile in zip( axes, candidate_profiles, ): prepared_frame = prepare_frame( representative_frame, profile, ) _, decoded_frame = encode_decode_jpeg( prepared_frame, profile, ) if profile.color_mode == COLOR_MODE_GRAYSCALE: axes_item.imshow( decoded_frame, cmap="gray", vmin=0, vmax=255, ) else: axes_item.imshow( cv2.cvtColor( decoded_frame, cv2.COLOR_BGR2RGB, ) ) result = results_by_name[profile.profile_name] wrapped_name = textwrap.fill( profile.profile_name, width=34, ) axes_item.set_title( f"{wrapped_name}\n" f"{profile.width}×{profile.height}, " f"{profile.target_fps:g} кадр/с, " f"Q={profile.jpeg_quality}\n" f"{result.payload_bitrate_kbps:.2f} кбит/с, " f"PSNR={result.mean_psnr_db:.2f} дБ", fontsize=9, ) axes_item.axis("off") figure.suptitle( "Lab026. Сравнение candidate-профилей покадрового JPEG", fontsize=15, ) figure.tight_layout() figure.savefig(CANDIDATE_COMPARISON_PATH, dpi=160) plt.close(figure) def format_result_line(result: ProfileResult) -> str: """ Формирует компактную строку результата для текстового отчёта. """ return ( f"{result.profile_name}: " f"{result.width}x{result.height}, " f"{result.target_fps:g} кадр/с, " f"{result.color_mode}, Q={result.jpeg_quality}, " f"кадров={result.selected_frames}, " f"payload={result.payload_bitrate_kbps:.3f} кбит/с, " f"PSNR={result.mean_psnr_db:.3f} дБ, " f"уменьшение={result.reduction_ratio:.2f}x" ) def append_experiment_section( report_lines: list[str], title: str, experiment_name: str, results: list[ProfileResult], ) -> None: """ Добавляет в отчёт все строки одного эксперимента. """ report_lines.extend(["", title]) for result in results: if result.experiment == experiment_name: report_lines.append(format_result_line(result)) def append_threshold_section( report_lines: list[str], threshold_kbps: float, candidate_results: list[ProfileResult], ) -> None: """ Перечисляет candidate-профили не выше заданного битрейта. """ report_lines.extend( [ "", ( "Candidate-профили с payload bitrate " f"не более {threshold_kbps:g} кбит/с:" ), ] ) matching_results = [ result for result in candidate_results if result.payload_bitrate_kbps <= threshold_kbps ] if not matching_results: report_lines.append("Нет.") return for result in matching_results: report_lines.append(format_result_line(result)) def write_report( source_width: int, source_height: int, source_fps: float, source_frame_count: int, source_duration_seconds: float, source_file_size_bytes: int, source_file_bitrate_kbps: float, results: list[ProfileResult], ) -> None: """ Создаёт полный текстовый отчёт Lab026 в кодировке UTF-8. """ report_lines = [ "Lab026. Исследование уменьшения видеопотока " "для радиоуправления ровером", "", "Цель: исследовать уменьшение полезной нагрузки " "видеоканала для передачи через ретранслятор.", f"Исходный файл: {SOURCE_VIDEO_PATH}", f"Размер исходного файла: {source_file_size_bytes} байт", ( f"Исходное разрешение: " f"{source_width}x{source_height}" ), f"Исходный FPS: {source_fps:.6f}", f"Число кадров: {source_frame_count}", f"Длительность: {source_duration_seconds:.6f} с", ( "Приблизительный исходный файловый битрейт: " f"{source_file_bitrate_kbps:.3f} кбит/с" ), "", "Исследуется независимое покадровое JPEG-сжатие. " "JPEG хранится и декодируется только в памяти.", "Рассчитанный bitrate учитывает только JPEG payload.", "Не учитываются заголовки радиопакетов, CRC, FEC, " "преамбула, интервалы, повторы и команды управления.", ] append_experiment_section( report_lines, "Результаты fps_sweep:", EXPERIMENT_FPS_SWEEP, results, ) append_experiment_section( report_lines, "Результаты resolution_sweep:", EXPERIMENT_RESOLUTION_SWEEP, results, ) append_experiment_section( report_lines, "Результаты quality_sweep:", EXPERIMENT_QUALITY_SWEEP, results, ) append_experiment_section( report_lines, "Результаты candidate_profiles:", EXPERIMENT_CANDIDATES, results, ) candidate_results = [ result for result in results if result.experiment == EXPERIMENT_CANDIDATES ] for threshold_kbps in [100.0, 50.0, 20.0, 10.0]: append_threshold_section( report_lines, threshold_kbps, candidate_results, ) minimum_bitrate_result = min( results, key=lambda result: result.payload_bitrate_kbps, ) report_lines.extend( [ "", "Профиль с минимальным измеренным битрейтом:", format_result_line(minimum_bitrate_result), "", "Автоматические метрики не заменяют визуальную оценку " "пригодности изображения для управления ровером.", ( "Сравнительное изображение: " f"{CANDIDATE_COMPARISON_PATH}" ), "Следующий шаг: пользователь должен визуально выбрать " "пригодные candidate-профили.", ] ) REPORT_PATH.write_text( "\n".join(report_lines), encoding="utf-8", ) def validate_output_files() -> None: """ Проверяет наличие и ненулевой размер шести результатов Lab026. """ for output_path in EXPECTED_OUTPUT_PATHS: if not output_path.is_file(): raise RuntimeError( f"Выходной файл не создан: {output_path}" ) if output_path.stat().st_size <= 0: raise RuntimeError( f"Выходной файл имеет нулевой размер: {output_path}" ) def print_candidate_results( results: list[ProfileResult], ) -> None: """ Выводит в консоль полные результаты девяти кандидатов. """ print() print("Результаты candidate_profiles:") for result in results: if result.experiment == EXPERIMENT_CANDIDATES: print(f" {format_result_line(result)}") def main() -> None: """ Выполняет все четыре эксперимента Lab026. """ OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) print("Чтение метаданных исходного видео...") ( source_width, source_height, source_fps, source_frame_count, source_duration_seconds, source_file_size_bytes, source_file_bitrate_kbps, ) = read_video_metadata(SOURCE_VIDEO_PATH) print() print("Параметры исходного видео:") print(f" Путь: {SOURCE_VIDEO_PATH}") print(f" Размер: {source_file_size_bytes} байт") print( f" Разрешение: " f"{source_width}x{source_height}" ) print(f" FPS: {source_fps:.6f}") print(f" Кадров: {source_frame_count}") print(f" Длительность:{source_duration_seconds:.6f} с") print( f" Битрейт: " f"{source_file_bitrate_kbps:.3f} кбит/с" ) profiles = build_profiles() print() print(f"Сформировано профилей: {len(profiles)}") print("Последовательная обработка кадров...") results = process_profiles( source_path=SOURCE_VIDEO_PATH, profiles=profiles, source_fps=source_fps, source_frame_count=source_frame_count, source_duration_seconds=source_duration_seconds, source_file_bitrate_kbps=source_file_bitrate_kbps, ) print() print("Сохранение CSV...") save_results_csv(results) print("Построение графика FPS...") save_fps_plot(results) print("Построение графика разрешений...") save_resolution_plot(results) print("Построение графика JPEG quality...") save_quality_plot(results) print("Создание сравнительного изображения кандидатов...") save_candidate_comparison( source_path=SOURCE_VIDEO_PATH, source_frame_count=source_frame_count, profiles=profiles, results=results, ) print("Создание текстового отчёта...") write_report( 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, source_file_bitrate_kbps=source_file_bitrate_kbps, results=results, ) validate_output_files() print_candidate_results(results) print() print("Созданы файлы:") for output_path in EXPECTED_OUTPUT_PATHS: print( f" {output_path} " f"({output_path.stat().st_size} байт)" ) print() print("Lab026 выполнена успешно.") print( "Для ручной оценки откройте: " f"{CANDIDATE_COMPARISON_PATH}" ) if __name__ == "__main__": main()