1359 lines
40 KiB
Python
1359 lines
40 KiB
Python
"""
|
||
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()
|