""" Lab026B. Увеличение разрешения за счёт глубины цвета 4 бита. Исследуются независимые кадры с 4-битной яркостью и 4-битным индексированным цветом. Под индексированным цветом понимается один 4-битный индекс на пиксель и палитра не более чем из 16 цветов. RGB444, потоковые видеокодеки и межкадровое кодирование не применяются. Все промежуточные JPEG, PNG и zlib-представления находятся только в оперативной памяти. Исходное видео открывается только для чтения. """ from __future__ import annotations from dataclasses import dataclass from io import BytesIO from pathlib import Path import csv import math import textwrap import zlib import cv2 import matplotlib import numpy as np from PIL import Image matplotlib.use("Agg") import matplotlib.pyplot as plt # --------------------------------------------------------------------- # Пути # --------------------------------------------------------------------- SOURCE_VIDEO_PATH = Path("data/raw/lab026_rover_source.mp4") OUTPUT_DIRECTORY = Path("data/processed/lab026b") CSV_PATH = OUTPUT_DIRECTORY / "lab026b_profiles.csv" REPORT_PATH = OUTPUT_DIRECTORY / "lab026b_report.txt" RESOLUTION_PLOT_PATH = ( OUTPUT_DIRECTORY / "lab026b_bitrate_vs_resolution.png" ) CODEC_COMPARISON_PATH = ( OUTPUT_DIRECTORY / "lab026b_codec_comparison.png" ) EQUAL_BITRATE_COMPARISON_PATH = ( OUTPUT_DIRECTORY / "lab026b_equal_bitrate_candidates.png" ) EXPECTED_OUTPUT_PATHS = [ CSV_PATH, REPORT_PATH, RESOLUTION_PLOT_PATH, CODEC_COMPARISON_PATH, EQUAL_BITRATE_COMPARISON_PATH, ] # --------------------------------------------------------------------- # Параметры эксперимента # --------------------------------------------------------------------- EXPERIMENT_NAME = "four_bit_resolution_sweep" TARGET_FPS = 2.0 BASELINE_BITRATE_KBPS = 80.283660 FRAME_TIME_EPSILON_SECONDS = 1e-9 FRAME_HEADER_BYTES = 16 FIXED_PALETTE_BYTES = 16 * 3 MAXIMUM_PSNR_DB = 100.0 RESOLUTIONS = [ (320, 180), (400, 225), (448, 252), (480, 270), (560, 315), (640, 360), ] MODE_GRAY8_JPEG = "gray8_jpeg" MODE_GRAY4_JPEG = "gray4_jpeg" MODE_GRAY4_PACKED_ZLIB = "gray4_packed_zlib" MODE_COLOR16_PNG = "color16_png" MODE_COLOR16_PACKED_ZLIB = "color16_packed_zlib" PROFILE_MODES = [ MODE_GRAY8_JPEG, MODE_GRAY4_JPEG, MODE_GRAY4_PACKED_ZLIB, MODE_COLOR16_PNG, MODE_COLOR16_PACKED_ZLIB, ] CSV_FIELD_NAMES = [ "experiment", "profile_name", "mode", "width", "height", "pixel_count", "target_fps", "actual_fps", "bits_per_pixel_before_compression", "maximum_gray_levels", "maximum_colors", "selected_frames", "total_payload_bytes", "mean_frame_bytes", "median_frame_bytes", "p95_frame_bytes", "max_frame_bytes", "payload_bitrate_bps", "payload_bitrate_kbps", "baseline_bitrate_kbps", "difference_from_baseline_kbps", "percent_of_baseline_bitrate", "compression_ratio_from_unpacked_source", "mean_psnr_db", "header_bytes_per_frame", "palette_bytes_per_frame", "notes", ] @dataclass(frozen=True) class FourBitProfile: """ Описывает разрешение и режим одного профиля Lab026B. """ experiment: str profile_name: str mode: str width: int height: int target_fps: float bits_per_pixel_before_compression: int maximum_gray_levels: int maximum_colors: int header_bytes_per_frame: int palette_bytes_per_frame: int notes: str @dataclass(frozen=True) class FourBitResult: """ Содержит измеренные характеристики одного профиля Lab026B. """ experiment: str profile_name: str mode: str width: int height: int pixel_count: int target_fps: float actual_fps: float bits_per_pixel_before_compression: int maximum_gray_levels: int maximum_colors: int selected_frames: int total_payload_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 baseline_bitrate_kbps: float difference_from_baseline_kbps: float percent_of_baseline_bitrate: float compression_ratio_from_unpacked_source: float mean_psnr_db: float header_bytes_per_frame: int palette_bytes_per_frame: int notes: str @dataclass class ProfileAccumulator: """ Накапливает размеры кадров и PSNR при чтении исходного ролика. """ next_sample_time: float payload_sizes: list[int] psnr_values: list[float] maximum_observed_levels: int maximum_observed_colors: int @dataclass(frozen=True) class FrameEncodingResult: """ Описывает результат кодирования одного кадра в памяти. """ payload_size_bytes: int restored_frame: np.ndarray observed_gray_levels: int observed_colors: int def read_video_metadata( source_path: Path, ) -> tuple[int, int, float, int, float, int]: """ Читает параметры исходного видео через OpenCV. """ 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("Получено некорректное разрешение видео.") if source_fps <= 0.0: raise RuntimeError("FPS исходного видео равен нулю.") if frame_count <= 0: raise RuntimeError("Число кадров исходного видео равно нулю.") duration_seconds = frame_count / source_fps source_size_bytes = source_path.stat().st_size if source_size_bytes <= 0: raise RuntimeError("Исходное видео имеет нулевой размер.") return ( width, height, source_fps, frame_count, duration_seconds, source_size_bytes, ) def build_profiles() -> list[FourBitProfile]: """ Формирует 30 профилей: шесть разрешений и пять режимов. """ profiles: list[FourBitProfile] = [] mode_parameters = { MODE_GRAY8_JPEG: { "bits": 8, "gray_levels": 256, "colors": 0, "header": 0, "palette": 0, "notes": "Grayscale 8 bit, JPEG quality 40.", }, MODE_GRAY4_JPEG: { "bits": 4, "gray_levels": 16, "colors": 0, "header": 0, "palette": 0, "notes": ( "Grayscale quantized to 16 levels, " "restored to uint8, JPEG quality 40." ), }, MODE_GRAY4_PACKED_ZLIB: { "bits": 4, "gray_levels": 16, "colors": 0, "header": FRAME_HEADER_BYTES, "palette": 0, "notes": ( "Two 4-bit grayscale indices per byte, " "zlib level 9, 16-byte frame header." ), }, MODE_COLOR16_PNG: { "bits": 4, "gray_levels": 0, "colors": 16, "header": 0, "palette": 0, "notes": ( "Pillow palette image, maximum 16 colors, " "no dithering, optimized indexed PNG." ), }, MODE_COLOR16_PACKED_ZLIB: { "bits": 4, "gray_levels": 0, "colors": 16, "header": FRAME_HEADER_BYTES, "palette": FIXED_PALETTE_BYTES, "notes": ( "Two 4-bit palette indices per byte, " "zlib level 9, 16-byte header and " "fixed 48-byte palette." ), }, } for width, height in RESOLUTIONS: for mode in PROFILE_MODES: parameters = mode_parameters[mode] profiles.append( FourBitProfile( experiment=EXPERIMENT_NAME, profile_name=( f"{width}x{height}_{mode}_2fps" ), mode=mode, width=width, height=height, target_fps=TARGET_FPS, bits_per_pixel_before_compression=( parameters["bits"] ), maximum_gray_levels=( parameters["gray_levels"] ), maximum_colors=parameters["colors"], header_bytes_per_frame=parameters["header"], palette_bytes_per_frame=parameters["palette"], notes=parameters["notes"], ) ) return profiles def should_sample_frame( frame_time_seconds: float, next_sample_time_seconds: float, epsilon_seconds: float = FRAME_TIME_EPSILON_SECONDS, ) -> bool: """ Выбирает кадры по временной шкале, как в основной Lab026. """ return ( frame_time_seconds + epsilon_seconds >= next_sample_time_seconds ) def quantize_gray4( gray_frame: np.ndarray, ) -> tuple[np.ndarray, np.ndarray]: """ Квантует яркость в индексы 0...15 и восстанавливает uint8. """ indices = np.rint( gray_frame.astype(np.float64) / 255.0 * 15.0 ).astype(np.uint8) if int(np.min(indices)) < 0 or int(np.max(indices)) > 15: raise RuntimeError("Индексы gray4 вышли из диапазона 0...15.") if np.unique(indices).size > 16: raise RuntimeError("Gray4 содержит более 16 уровней.") restored = np.rint( indices.astype(np.float64) / 15.0 * 255.0 ).astype(np.uint8) return indices, restored def pack_nibbles(indices: np.ndarray) -> bytes: """ Упаковывает два 4-битных индекса в один байт. Первый пиксель помещается в старшие четыре бита. При нечётном количестве пикселей младший полубайт последнего байта равен нулю. """ flattened = np.asarray( indices, dtype=np.uint8, ).reshape(-1) if flattened.size == 0: raise RuntimeError("Нельзя упаковать пустой массив индексов.") if int(np.min(flattened)) < 0 or int(np.max(flattened)) > 15: raise RuntimeError( "Перед упаковкой обнаружен индекс вне диапазона 0...15." ) if flattened.size % 2 != 0: flattened = np.pad( flattened, (0, 1), mode="constant", constant_values=0, ) packed = ( (flattened[0::2] << 4) | flattened[1::2] ).astype(np.uint8) return packed.tobytes() def unpack_nibbles( packed_bytes: bytes, original_pixel_count: int, ) -> np.ndarray: """ Восстанавливает исходную последовательность 4-битных индексов. """ if original_pixel_count <= 0: raise RuntimeError( "Число восстанавливаемых пикселей должно быть положительным." ) packed = np.frombuffer( packed_bytes, dtype=np.uint8, ) unpacked = np.empty( packed.size * 2, dtype=np.uint8, ) unpacked[0::2] = packed >> 4 unpacked[1::2] = packed & 0x0F if original_pixel_count > unpacked.size: raise RuntimeError( "Упакованных данных недостаточно для восстановления." ) return unpacked[:original_pixel_count].copy() def pillow_quantize_parameters() -> tuple[object, object]: """ Возвращает совместимые enum Pillow для Median Cut без дизеринга. """ quantize_enum = getattr(Image, "Quantize", None) dither_enum = getattr(Image, "Dither", None) method = ( quantize_enum.MEDIANCUT if quantize_enum is not None else Image.MEDIANCUT ) dither = ( dither_enum.NONE if dither_enum is not None else Image.NONE ) return method, dither def quantize_color16( rgb_frame: np.ndarray, ) -> tuple[Image.Image, np.ndarray, np.ndarray, np.ndarray]: """ Создаёт палитровое Pillow-изображение с максимум 16 цветами. Возвращает изображение P, индексы, палитру 16x3 и восстановленный RGB-кадр. """ method, dither = pillow_quantize_parameters() source_image = Image.fromarray(rgb_frame, mode="RGB") palette_image = source_image.quantize( colors=16, method=method, dither=dither, ) if palette_image.mode != "P": raise RuntimeError( "Pillow вернул не палитровое изображение режима P." ) indices = np.asarray( palette_image, dtype=np.uint8, ) if int(np.min(indices)) < 0 or int(np.max(indices)) > 15: raise RuntimeError( "Индексы color16 вышли из диапазона 0...15." ) if np.unique(indices).size > 16: raise RuntimeError("Color16 содержит более 16 цветов.") raw_palette = palette_image.getpalette() if raw_palette is None: raise RuntimeError("Pillow не вернул палитру изображения.") palette_values = list(raw_palette[:FIXED_PALETTE_BYTES]) if len(palette_values) < FIXED_PALETTE_BYTES: palette_values.extend( [0] * (FIXED_PALETTE_BYTES - len(palette_values)) ) palette_rgb = np.asarray( palette_values, dtype=np.uint8, ).reshape(16, 3) restored_rgb = np.asarray( palette_image.convert("RGB"), dtype=np.uint8, ).copy() return ( palette_image, indices, palette_rgb, restored_rgb, ) def encode_gray_jpeg( gray_frame: np.ndarray, jpeg_quality: int = 40, ) -> tuple[int, np.ndarray]: """ Кодирует grayscale JPEG в памяти и возвращает декодированный кадр. """ encoding_succeeded, encoded_jpeg = cv2.imencode( ".jpg", gray_frame, [cv2.IMWRITE_JPEG_QUALITY, jpeg_quality], ) if not encoding_succeeded or encoded_jpeg is None: raise RuntimeError("JPEG encoding завершился ошибкой.") decoded_gray = cv2.imdecode( encoded_jpeg, cv2.IMREAD_GRAYSCALE, ) if decoded_gray is None: raise RuntimeError("JPEG decoding завершился ошибкой.") if decoded_gray.shape != gray_frame.shape: raise RuntimeError( "Размер JPEG после декодирования не совпадает с эталоном." ) return int(encoded_jpeg.size), decoded_gray def encode_palette_png( palette_image: Image.Image, ) -> tuple[int, np.ndarray]: """ Сохраняет палитровый PNG в BytesIO и декодирует его обратно в RGB. """ if palette_image.mode != "P": raise RuntimeError("Для PNG ожидался палитровый режим P.") buffer = BytesIO() palette_image.save( buffer, format="PNG", optimize=True, ) png_bytes = buffer.getvalue() if not png_bytes: raise RuntimeError("Pillow создал пустой PNG payload.") with Image.open(BytesIO(png_bytes)) as decoded_image: if decoded_image.mode != "P": raise RuntimeError( "Сохранённый PNG перестал быть палитровым." ) decoded_rgb = np.asarray( decoded_image.convert("RGB"), dtype=np.uint8, ).copy() return len(png_bytes), decoded_rgb def encode_packed_zlib( indices: np.ndarray, header_bytes: int, palette_bytes: int, ) -> tuple[int, np.ndarray]: """ Упаковывает индексы по два на байт и сжимает zlib level 9. Функция обязательно проверяет точное восстановление pack/unpack. """ original_shape = indices.shape original_count = indices.size packed_bytes = pack_nibbles(indices) unpacked_flat = unpack_nibbles( packed_bytes, original_count, ) original_flat = np.asarray( indices, dtype=np.uint8, ).reshape(-1) if not np.array_equal(unpacked_flat, original_flat): raise RuntimeError("Проверка pack/unpack завершилась ошибкой.") compressed_payload = zlib.compress( packed_bytes, level=9, ) if not compressed_payload: raise RuntimeError("zlib создал пустой payload.") total_payload_bytes = ( header_bytes + palette_bytes + len(compressed_payload) ) unpacked_indices = unpacked_flat.reshape(original_shape) return total_payload_bytes, unpacked_indices def calculate_psnr( reference_frame: np.ndarray, restored_frame: np.ndarray, ) -> float: """ Рассчитывает PSNR в float64 и ограничивает идеальный случай 100 дБ. """ if reference_frame.shape != restored_frame.shape: raise RuntimeError( "PSNR нельзя рассчитать для кадров разных размеров." ) difference = ( reference_frame.astype(np.float64) - restored_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 process_frame_for_profile( source_frame: np.ndarray, profile: FourBitProfile, ) -> FrameEncodingResult: """ Применяет один из пяти фиксированных режимов к исходному кадру. """ resized_bgr = cv2.resize( source_frame, (profile.width, profile.height), interpolation=cv2.INTER_AREA, ) if profile.mode == MODE_GRAY8_JPEG: reference_gray = cv2.cvtColor( resized_bgr, cv2.COLOR_BGR2GRAY, ) payload_size, restored_gray = encode_gray_jpeg( reference_gray, jpeg_quality=40, ) return FrameEncodingResult( payload_size, restored_gray, int(np.unique(reference_gray).size), 0, ) if profile.mode in { MODE_GRAY4_JPEG, MODE_GRAY4_PACKED_ZLIB, }: reference_gray = cv2.cvtColor( resized_bgr, cv2.COLOR_BGR2GRAY, ) indices, quantized_gray = quantize_gray4( reference_gray ) observed_levels = int(np.unique(indices).size) if profile.mode == MODE_GRAY4_JPEG: payload_size, restored_gray = encode_gray_jpeg( quantized_gray, jpeg_quality=40, ) else: payload_size, unpacked_indices = ( encode_packed_zlib( indices, FRAME_HEADER_BYTES, 0, ) ) restored_gray = np.rint( unpacked_indices.astype(np.float64) / 15.0 * 255.0 ).astype(np.uint8) return FrameEncodingResult( payload_size, restored_gray, observed_levels, 0, ) if profile.mode in { MODE_COLOR16_PNG, MODE_COLOR16_PACKED_ZLIB, }: reference_rgb = cv2.cvtColor( resized_bgr, cv2.COLOR_BGR2RGB, ) ( palette_image, indices, palette_rgb, quantized_rgb, ) = quantize_color16(reference_rgb) observed_colors = int(np.unique(indices).size) if profile.mode == MODE_COLOR16_PNG: payload_size, restored_rgb = encode_palette_png( palette_image ) else: payload_size, unpacked_indices = ( encode_packed_zlib( indices, FRAME_HEADER_BYTES, FIXED_PALETTE_BYTES, ) ) restored_rgb = palette_rgb[unpacked_indices] return FrameEncodingResult( payload_size, restored_rgb, 0, observed_colors, ) raise RuntimeError(f"Неизвестный режим: {profile.mode}") def reference_frame_for_psnr( source_frame: np.ndarray, profile: FourBitProfile, ) -> np.ndarray: """ Создаёт эталон исходного resized-кадра для PSNR. """ resized_bgr = cv2.resize( source_frame, (profile.width, profile.height), interpolation=cv2.INTER_AREA, ) if profile.mode.startswith("gray"): return cv2.cvtColor( resized_bgr, cv2.COLOR_BGR2GRAY, ) return cv2.cvtColor( resized_bgr, cv2.COLOR_BGR2RGB, ) def unpacked_reference_bytes( profile: FourBitProfile, ) -> int: """ Возвращает размер эталонного массива до сжатия. Для grayscale это один uint8 на пиксель, для RGB — три uint8. """ channel_count = ( 1 if profile.mode.startswith("gray") else 3 ) return profile.width * profile.height * channel_count def process_profiles( source_path: Path, profiles: list[FourBitProfile], source_fps: float, source_frame_count: int, source_duration_seconds: float, ) -> list[FourBitResult]: """ За один последовательный проход измеряет все 30 профилей. """ if len(profiles) != 30: raise RuntimeError( f"Ожидалось 30 профилей, получено {len(profiles)}." ) accumulators = { profile.profile_name: ProfileAccumulator( next_sample_time=0.0, payload_sizes=[], psnr_values=[], maximum_observed_levels=0, maximum_observed_colors=0, ) 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"Получен пустой кадр {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 reference_frame = reference_frame_for_psnr( source_frame, profile, ) frame_result = process_frame_for_profile( source_frame, profile, ) psnr_db = calculate_psnr( reference_frame, frame_result.restored_frame, ) accumulator.payload_sizes.append( frame_result.payload_size_bytes ) accumulator.psnr_values.append(psnr_db) accumulator.maximum_observed_levels = max( accumulator.maximum_observed_levels, frame_result.observed_gray_levels, ) accumulator.maximum_observed_colors = max( accumulator.maximum_observed_colors, frame_result.observed_colors, ) 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("Не прочитан ни один исходный кадр.") if frame_index != source_frame_count: print( "Предупреждение: прочитано " f"{frame_index} кадров вместо {source_frame_count}." ) results: list[FourBitResult] = [] for profile in profiles: accumulator = accumulators[profile.profile_name] if not accumulator.payload_sizes: raise RuntimeError( f"Нет выбранных кадров: {profile.profile_name}" ) payload_sizes = np.asarray( accumulator.payload_sizes, dtype=np.float64, ) psnr_values = np.asarray( accumulator.psnr_values, dtype=np.float64, ) selected_frames = len(accumulator.payload_sizes) total_payload_bytes = int( sum(accumulator.payload_sizes) ) actual_fps = ( selected_frames / source_duration_seconds ) payload_bitrate_bps = ( total_payload_bytes * 8.0 / source_duration_seconds ) payload_bitrate_kbps = payload_bitrate_bps / 1000.0 difference_from_baseline_kbps = ( payload_bitrate_kbps - BASELINE_BITRATE_KBPS ) percent_of_baseline = ( payload_bitrate_kbps / BASELINE_BITRATE_KBPS * 100.0 ) compression_ratio = ( unpacked_reference_bytes(profile) / float(np.mean(payload_sizes)) ) if profile.maximum_gray_levels == 16: if accumulator.maximum_observed_levels > 16: raise RuntimeError( f"Gray4 превысил 16 уровней: {profile.profile_name}" ) if profile.maximum_colors == 16: if accumulator.maximum_observed_colors > 16: raise RuntimeError( f"Color16 превысил 16 цветов: {profile.profile_name}" ) results.append( FourBitResult( experiment=profile.experiment, profile_name=profile.profile_name, mode=profile.mode, width=profile.width, height=profile.height, pixel_count=profile.width * profile.height, target_fps=profile.target_fps, actual_fps=actual_fps, bits_per_pixel_before_compression=( profile.bits_per_pixel_before_compression ), maximum_gray_levels=( profile.maximum_gray_levels ), maximum_colors=profile.maximum_colors, selected_frames=selected_frames, total_payload_bytes=total_payload_bytes, mean_frame_bytes=float(np.mean(payload_sizes)), median_frame_bytes=float(np.median(payload_sizes)), p95_frame_bytes=float( np.percentile(payload_sizes, 95) ), max_frame_bytes=int(np.max(payload_sizes)), payload_bitrate_bps=payload_bitrate_bps, payload_bitrate_kbps=payload_bitrate_kbps, baseline_bitrate_kbps=BASELINE_BITRATE_KBPS, difference_from_baseline_kbps=( difference_from_baseline_kbps ), percent_of_baseline_bitrate=( percent_of_baseline ), compression_ratio_from_unpacked_source=( compression_ratio ), mean_psnr_db=float(np.mean(psnr_values)), header_bytes_per_frame=( profile.header_bytes_per_frame ), palette_bytes_per_frame=( profile.palette_bytes_per_frame ), notes=profile.notes, ) ) return results def save_csv(results: list[FourBitResult]) -> None: """ Сохраняет 30 результатов Lab026B в 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( { "experiment": result.experiment, "profile_name": result.profile_name, "mode": result.mode, "width": result.width, "height": result.height, "pixel_count": result.pixel_count, "target_fps": f"{result.target_fps:.6f}", "actual_fps": f"{result.actual_fps:.6f}", "bits_per_pixel_before_compression": ( result.bits_per_pixel_before_compression ), "maximum_gray_levels": ( result.maximum_gray_levels ), "maximum_colors": result.maximum_colors, "selected_frames": result.selected_frames, "total_payload_bytes": ( result.total_payload_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}" ), "baseline_bitrate_kbps": ( f"{result.baseline_bitrate_kbps:.6f}" ), "difference_from_baseline_kbps": ( f"{result.difference_from_baseline_kbps:.6f}" ), "percent_of_baseline_bitrate": ( f"{result.percent_of_baseline_bitrate:.6f}" ), "compression_ratio_from_unpacked_source": ( f"{result.compression_ratio_from_unpacked_source:.6f}" ), "mean_psnr_db": ( f"{result.mean_psnr_db:.6f}" ), "header_bytes_per_frame": ( result.header_bytes_per_frame ), "palette_bytes_per_frame": ( result.palette_bytes_per_frame ), "notes": result.notes, } ) def find_candidates( results: list[FourBitResult], ) -> dict[str, list[FourBitResult]]: """ Формирует заданные группы профилей по payload bitrate. """ return { "70_to_90": [ result for result in results if 70.0 <= result.payload_bitrate_kbps <= 90.0 ], "up_to_baseline": [ result for result in results if ( result.payload_bitrate_kbps <= BASELINE_BITRATE_KBPS ) ], "up_to_100": [ result for result in results if result.payload_bitrate_kbps <= 100.0 ], "color16_up_to_100": [ result for result in results if ( result.mode.startswith("color16") and result.payload_bitrate_kbps <= 100.0 ) ], "larger_than_320_up_to_baseline": [ result for result in results if ( result.pixel_count > 320 * 180 and result.payload_bitrate_kbps <= BASELINE_BITRATE_KBPS ) ], } def maximum_resolution_results( results: list[FourBitResult], bitrate_limit_kbps: float, ) -> list[FourBitResult]: """ Возвращает все профили максимального разрешения до лимита. """ matching = [ result for result in results if result.payload_bitrate_kbps <= bitrate_limit_kbps ] if not matching: return [] maximum_pixel_count = max( result.pixel_count for result in matching ) return [ result for result in matching if result.pixel_count == maximum_pixel_count ] def annotate_bitrate_points( axes: plt.Axes, x_values: list[int], y_values: list[float], ) -> None: """ Подписывает значения bitrate над точками. """ for x_value, y_value in zip(x_values, y_values): axes.annotate( f"{y_value:.1f}", (x_value, y_value), xytext=(0, 6), textcoords="offset points", ha="center", fontsize=7, ) def save_resolution_plot( results: list[FourBitResult], ) -> None: """ Строит пять серий bitrate по разрешению и линию baseline. """ figure, axes = plt.subplots(figsize=(13, 7)) x_positions = list(range(len(RESOLUTIONS))) resolution_labels = [ f"{width}×{height}" for width, height in RESOLUTIONS ] for mode in PROFILE_MODES: mode_results = sorted( ( result for result in results if result.mode == mode ), key=lambda result: result.pixel_count, ) y_values = [ result.payload_bitrate_kbps for result in mode_results ] axes.plot( x_positions, y_values, marker="o", label=mode, ) annotate_bitrate_points( axes, x_positions, y_values, ) axes.axhline( BASELINE_BITRATE_KBPS, linestyle="--", label=( f"Baseline {BASELINE_BITRATE_KBPS:.6f} кбит/с" ), ) axes.set_xticks(x_positions, resolution_labels) axes.set_title( "Lab026B. Битрейт 4-битных режимов по разрешению" ) axes.set_xlabel("Разрешение кадра") axes.set_ylabel("JPEG/PNG/zlib payload, кбит/с") axes.grid(True) axes.legend() figure.tight_layout() figure.savefig(RESOLUTION_PLOT_PATH, dpi=160) plt.close(figure) def save_codec_comparison( results: list[FourBitResult], ) -> None: """ Строит столбчатое сравнение пяти режимов при 480x270. """ selected_results = [ result for result in results if result.width == 480 and result.height == 270 ] if len(selected_results) != 5: raise RuntimeError( "Для 480x270 ожидалось пять результатов." ) labels = [result.mode for result in selected_results] bitrates = [ result.payload_bitrate_kbps for result in selected_results ] figure, axes = plt.subplots(figsize=(12, 7)) bars = axes.bar(labels, bitrates) for bar, result in zip(bars, selected_results): axes.annotate( ( f"{result.payload_bitrate_kbps:.1f} кбит/с\n" f"PSNR {result.mean_psnr_db:.1f} дБ" ), ( bar.get_x() + bar.get_width() / 2.0, bar.get_height(), ), xytext=(0, 5), textcoords="offset points", ha="center", fontsize=8, ) axes.axhline( BASELINE_BITRATE_KBPS, linestyle="--", label=( f"Baseline {BASELINE_BITRATE_KBPS:.6f} кбит/с" ), ) axes.set_title("Lab026B. Сравнение режимов при 480×270") axes.set_xlabel("Режим") axes.set_ylabel("Payload bitrate, кбит/с") axes.tick_params(axis="x", labelrotation=20) axes.grid(True, axis="y") axes.legend() figure.tight_layout() figure.savefig(CODEC_COMPARISON_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_index = source_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( f"Не удалось прочитать средний кадр {middle_index}." ) return frame def result_by_profile( results: list[FourBitResult], width: int, height: int, mode: str, ) -> FourBitResult: """ Находит единственный результат по разрешению и режиму. """ matching = [ result for result in results if ( result.width == width and result.height == height and result.mode == mode ) ] if len(matching) != 1: raise RuntimeError( f"Ожидался один результат {width}x{height} {mode}." ) return matching[0] def save_comparison_image( source_path: Path, source_frame_count: int, profiles: list[FourBitProfile], results: list[FourBitResult], ) -> None: """ Создаёт таблицу девяти кадров при близких битрейтах. """ result_lookup = { ( result.width, result.height, result.mode, ): result for result in results } profile_lookup = { ( profile.width, profile.height, profile.mode, ): profile for profile in profiles } panel_keys: list[tuple[int, int, str, str]] = [ (320, 180, MODE_GRAY8_JPEG, ""), (448, 252, MODE_GRAY4_JPEG, ""), (480, 270, MODE_GRAY4_JPEG, ""), (448, 252, MODE_GRAY4_PACKED_ZLIB, ""), (448, 252, MODE_COLOR16_PNG, ""), (448, 252, MODE_COLOR16_PACKED_ZLIB, ""), (480, 270, MODE_COLOR16_PNG, ""), (480, 270, MODE_COLOR16_PACKED_ZLIB, ""), ] best_candidates = [ result for result in results if result.payload_bitrate_kbps <= 100.0 ] if not best_candidates: raise RuntimeError("Нет профиля с bitrate до 100 кбит/с.") best_result = max( best_candidates, key=lambda result: ( result.pixel_count, result.mean_psnr_db, ), ) panel_keys.append( ( best_result.width, best_result.height, best_result.mode, "BEST <=100 kbps", ) ) representative_frame = read_representative_frame( source_path, source_frame_count, ) figure, axes_grid = plt.subplots( 3, 3, figsize=(18, 13), ) for axes, panel_key in zip( axes_grid.ravel(), panel_keys, ): width, height, mode, special_label = panel_key profile = profile_lookup[(width, height, mode)] result = result_lookup[(width, height, mode)] frame_result = process_frame_for_profile( representative_frame, profile, ) displayed_frame = cv2.resize( frame_result.restored_frame, (640, 360), interpolation=cv2.INTER_NEAREST, ) if mode.startswith("gray"): axes.imshow( displayed_frame, cmap="gray", vmin=0, vmax=255, ) level_text = ( f"{profile.maximum_gray_levels} уровней" ) else: axes.imshow(displayed_frame) level_text = ( f"{profile.maximum_colors} цветов" ) title_prefix = ( f"{special_label}\n" if special_label else "" ) axes.set_title( title_prefix + textwrap.fill(mode, width=28) + "\n" + f"{width}×{height}, " + f"{result.payload_bitrate_kbps:.2f} кбит/с\n" + f"PSNR {result.mean_psnr_db:.2f} дБ, " + level_text, fontsize=9, ) axes.axis("off") figure.suptitle( "Lab026B. Сравнение 4-битных профилей", fontsize=15, ) figure.tight_layout() figure.savefig( EQUAL_BITRATE_COMPARISON_PATH, dpi=160, ) plt.close(figure) def format_result_line(result: FourBitResult) -> str: """ Формирует полную строку результата для текстового отчёта. """ return ( f"{result.profile_name}: " f"mode={result.mode}, " f"{result.width}x{result.height}, " f"frames={result.selected_frames}, " f"actual_fps={result.actual_fps:.6f}, " f"total={result.total_payload_bytes} bytes, " f"mean={result.mean_frame_bytes:.3f}, " f"median={result.median_frame_bytes:.3f}, " f"p95={result.p95_frame_bytes:.3f}, " f"max={result.max_frame_bytes}, " f"bitrate={result.payload_bitrate_kbps:.6f} kbit/s, " f"baseline_diff={result.difference_from_baseline_kbps:.6f}, " f"baseline_percent={result.percent_of_baseline_bitrate:.3f}%, " f"compression={result.compression_ratio_from_unpacked_source:.3f}x, " f"PSNR={result.mean_psnr_db:.6f} dB" ) def append_result_group( report_lines: list[str], title: str, results: list[FourBitResult], ) -> None: """ Добавляет в отчёт именованную группу результатов. """ report_lines.extend(["", title]) if not results: report_lines.append("Нет.") return for result in 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_size_bytes: int, results: list[FourBitResult], ) -> None: """ Создаёт полный UTF-8 отчёт Lab026B. """ candidates = find_candidates(results) maximum_to_baseline = maximum_resolution_results( results, BASELINE_BITRATE_KBPS, ) maximum_to_100 = maximum_resolution_results( results, 100.0, ) gray4_results = [ result for result in results if result.mode.startswith("gray4") ] color16_results = [ result for result in results if result.mode.startswith("color16") ] report_lines = [ "Lab026B. Увеличение разрешения кадра " "за счёт глубины цвета 4 бита", "", "Цель: проверить увеличение разрешения при битрейте, " "близком к контрольным 80.283660 кбит/с.", f"Исходное видео: {SOURCE_VIDEO_PATH}", f"Размер исходного видео: {source_size_bytes} байт", f"Исходное разрешение: {source_width}×{source_height}", f"Исходный FPS: {source_fps:.6f}", f"Число кадров: {source_frame_count}", f"Длительность: {source_duration_seconds:.6f} с", "", "4-битная яркость: индекс 0...15, то есть 16 уровней " "серого на пиксель.", "4-битный индексированный цвет: один индекс 0...15 " "на пиксель и палитра максимум из 16 RGB-цветов.", "Это не RGB444: RGB444 требует по 4 бита на каждый " "из трёх каналов, то есть 12 бит на пиксель.", "", "Режимы:", "gray8_jpeg — контрольный grayscale JPEG Q40.", "gray4_jpeg — 16 уровней серого, восстановление uint8, " "JPEG Q40.", "gray4_packed_zlib — 4-битные индексы, два на байт, " "zlib level 9 и заголовок 16 байт.", "color16_png — палитра максимум 16 цветов, без дизеринга, " "индексированный PNG в памяти.", "color16_packed_zlib — 4-битные индексы, zlib level 9, " "заголовок 16 байт и фиксированная палитра 48 байт.", "", f"Контрольный baseline: {BASELINE_BITRATE_KBPS:.6f} кбит/с.", "CRC, FEC, радиозаголовки и пакетирование пока не учтены.", "Статические метрики не учитывают задержку видеоканала.", "", "Полные результаты 30 профилей:", ] for result in results: report_lines.append(format_result_line(result)) append_result_group( report_lines, "Профили 70–90 кбит/с:", candidates["70_to_90"], ) append_result_group( report_lines, "Профили до baseline:", candidates["up_to_baseline"], ) append_result_group( report_lines, "Профили до 100 кбит/с:", candidates["up_to_100"], ) append_result_group( report_lines, "Цветные color16-профили до 100 кбит/с:", candidates["color16_up_to_100"], ) append_result_group( report_lines, ( "Профили больше 320x180 и не выше baseline:" ), candidates["larger_than_320_up_to_baseline"], ) append_result_group( report_lines, "Максимальное разрешение до baseline:", maximum_to_baseline, ) append_result_group( report_lines, "Максимальное разрешение до 100 кбит/с:", maximum_to_100, ) report_lines.extend( [ "", "Сравнение gray4 и color16:", ( "Минимальный gray4 bitrate: " f"{min(result.payload_bitrate_kbps for result in gray4_results):.6f} " "кбит/с." ), ( "Минимальный color16 bitrate: " f"{min(result.payload_bitrate_kbps for result in color16_results):.6f} " "кбит/с." ), "Размер палитры packed color16: 48 байт на кадр.", "Размер компактного заголовка packed-режимов: " "16 байт на кадр.", "", "Нельзя автоматически объявлять профиль безопасным " "для управления ровером по одной статической метрике.", ( "Сравнительное изображение: " f"{EQUAL_BITRATE_COMPARISON_PATH}" ), "Следующий шаг: ручная визуальная оценка пользователем.", ] ) REPORT_PATH.write_text( "\n".join(report_lines), encoding="utf-8", ) def validate_output_files() -> None: """ Проверяет наличие и ненулевой размер пяти результатов. """ for path in EXPECTED_OUTPUT_PATHS: if not path.is_file(): raise RuntimeError(f"Выходной файл не создан: {path}") if path.stat().st_size <= 0: raise RuntimeError( f"Выходной файл имеет нулевой размер: {path}" ) def main() -> None: """ Выполняет полный набор экспериментов Lab026B. """ OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True) print("Чтение метаданных исходного видео...") ( source_width, source_height, source_fps, source_frame_count, source_duration_seconds, source_size_bytes, ) = read_video_metadata(SOURCE_VIDEO_PATH) print() print("Исходные параметры:") print(f" Путь: {SOURCE_VIDEO_PATH}") print(f" Размер: {source_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() 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, ) print("Сохранение CSV...") save_csv(results) print("Построение графика по разрешению...") save_resolution_plot(results) print("Построение сравнения режимов...") save_codec_comparison(results) print("Создание сравнительного изображения...") save_comparison_image( 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_size_bytes=source_size_bytes, results=results, ) validate_output_files() print() print("Результаты профилей 320x180, 448x252 и 480x270:") for result in results: if ( (result.width, result.height) in {(320, 180), (448, 252), (480, 270)} ): print(f" {format_result_line(result)}") print() print("Созданы файлы:") for path in EXPECTED_OUTPUT_PATHS: print(f" {path} ({path.stat().st_size} байт)") print() print("Lab026B выполнена успешно.") print( "Для ручной оценки откройте: " f"{EQUAL_BITRATE_COMPARISON_PATH}" ) if __name__ == "__main__": main()