1862 lines
54 KiB
Python
1862 lines
54 KiB
Python
"""
|
||
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()
|