Lab027: optimize synchronized ROI video stream

This commit is contained in:
LittleSam129
2026-07-24 17:54:27 +03:00
parent 6d49036066
commit 78258b5355
17 changed files with 8300 additions and 0 deletions

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,925 @@
"""
Lab027. Динамическое превью временного обновления BASE и ROI.
Скрипт не повторяет расчёт benchmark Lab027 и не изменяет его
результаты. Он создаёт одно игнорируемое Git preview-видео с четырьмя
фиксированными профилями. Для каждого профиля независимо удерживаются
последние декодированные JPEG-состояния BASE и ROI.
Отдельные JPEG-файлы не сохраняются: кодирование и декодирование
выполняются только в памяти средствами OpenCV.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
import cv2
import numpy as np
SOURCE_VIDEO_PATH = Path("data/raw/lab026_rover_source.mp4")
PREVIEW_DIRECTORY = Path("data/raw/lab027_previews")
MP4_PREVIEW_PATH = (
PREVIEW_DIRECTORY / "lab027_roi_temporal_preview.mp4"
)
AVI_PREVIEW_PATH = (
PREVIEW_DIRECTORY / "lab027_roi_temporal_preview.avi"
)
OUTPUT_FPS = 30.0
PANEL_WIDTH = 640
PANEL_HEIGHT = 360
OUTPUT_WIDTH = PANEL_WIDTH * 2
OUTPUT_HEIGHT = PANEL_HEIGHT * 2
ROI_X_MIN = 0.20
ROI_X_MAX = 0.80
ROI_Y_MIN = 0.42
ROI_Y_MAX = 1.00
FRAME_TIME_EPSILON_SECONDS = 1e-9
NEW_UPDATE_LABEL_SECONDS = 0.15
@dataclass(frozen=True)
class PreviewProfile:
"""
Описывает один из четырёх фиксированных preview-профилей.
"""
name: str
base_width: int
base_height: int
base_fps: float
base_quality: int
roi_enabled: bool
roi_width: int
roi_height: int
roi_fps: float
roi_quality: int
measured_payload_kbps: float
@dataclass
class ProfileState:
"""
Хранит независимое временное состояние BASE и ROI профиля.
"""
latest_base: np.ndarray | None = None
latest_roi: np.ndarray | None = None
next_base_time: float = 0.0
next_roi_time: float = 0.0
last_base_update_time: float = 0.0
last_roi_update_time: float = 0.0
base_update_count: int = 0
roi_update_count: int = 0
def read_video_metadata(
source_path: Path,
) -> tuple[int, int, float, int, float]:
"""
Читает и проверяет параметры исходного видео.
"""
if not source_path.exists():
raise RuntimeError(
f"Исходный видеофайл отсутствует: {source_path}"
)
capture = cv2.VideoCapture(str(source_path))
if not capture.isOpened():
raise RuntimeError(
f"OpenCV не смог открыть видео: {source_path}"
)
try:
width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT))
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 fps <= 0.0:
raise RuntimeError("FPS исходного видео равен нулю.")
if frame_count <= 0:
raise RuntimeError("Число кадров исходного видео равно нулю.")
duration_seconds = frame_count / fps
return width, height, fps, frame_count, duration_seconds
def build_profiles() -> list[PreviewProfile]:
"""
Создаёт ровно четыре согласованных preview-профиля.
"""
profiles = [
PreviewProfile(
name="baseline",
base_width=320,
base_height=180,
base_fps=2.0,
base_quality=40,
roi_enabled=False,
roi_width=0,
roi_height=0,
roi_fps=0.0,
roi_quality=0,
measured_payload_kbps=80.27,
),
PreviewProfile(
name="ROI normal",
base_width=240,
base_height=135,
base_fps=1.0,
base_quality=25,
roi_enabled=True,
roi_width=320,
roi_height=180,
roi_fps=2.0,
roi_quality=35,
measured_payload_kbps=93.28,
),
PreviewProfile(
name="ROI economy",
base_width=160,
base_height=90,
base_fps=1.0,
base_quality=25,
roi_enabled=True,
roi_width=320,
roi_height=180,
roi_fps=2.0,
roi_quality=35,
measured_payload_kbps=85.19,
),
PreviewProfile(
name="ROI degraded channel",
base_width=240,
base_height=135,
base_fps=1.0,
base_quality=20,
roi_enabled=True,
roi_width=320,
roi_height=180,
roi_fps=1.0,
roi_quality=30,
measured_payload_kbps=50.49,
),
]
if len(profiles) != 4:
raise RuntimeError(
"Preview Lab027 должно содержать ровно четыре профиля."
)
return profiles
def normalized_roi_to_pixels(
width: int,
height: int,
) -> tuple[int, int, int, int]:
"""
Переводит фиксированные нормализованные координаты ROI в пиксели.
Результат задаёт полуоткрытый прямоугольник:
x_min, y_min, x_max, y_max.
"""
if width <= 0 or height <= 0:
raise ValueError("Размер кадра должен быть положительным.")
x_min = int(round(width * ROI_X_MIN))
x_max = int(round(width * ROI_X_MAX))
y_min = int(round(height * ROI_Y_MIN))
y_max = int(round(height * ROI_Y_MAX))
x_min = min(max(x_min, 0), width - 1)
x_max = min(max(x_max, x_min + 1), width)
y_min = min(max(y_min, 0), height - 1)
y_max = min(max(y_max, y_min + 1), height)
return x_min, y_min, x_max, y_max
def should_update(
current_time: float,
next_update_time: float,
epsilon_seconds: float = FRAME_TIME_EPSILON_SECONDS,
) -> bool:
"""
Проверяет наступление времени очередного обновления потока.
"""
return (
current_time + epsilon_seconds
>= next_update_time
)
def encode_decode_grayscale_jpeg(
gray_frame: np.ndarray,
jpeg_quality: int,
) -> np.ndarray:
"""
Кодирует grayscale JPEG в памяти и декодирует его обратно.
"""
if gray_frame.ndim != 2:
raise RuntimeError(
"JPEG preview должен получать grayscale-кадр."
)
encoding_ok, encoded = cv2.imencode(
".jpg",
gray_frame,
[cv2.IMWRITE_JPEG_QUALITY, jpeg_quality],
)
if not encoding_ok or encoded is None or encoded.size == 0:
raise RuntimeError("OpenCV не смог закодировать JPEG.")
decoded = cv2.imdecode(
encoded,
cv2.IMREAD_GRAYSCALE,
)
if decoded is None or decoded.shape != gray_frame.shape:
raise RuntimeError(
"Декодированный JPEG имеет некорректный размер."
)
return decoded
def encode_decode_base(
source_frame: np.ndarray,
profile: PreviewProfile,
) -> np.ndarray:
"""
Формирует очередное декодированное состояние BASE.
"""
source_gray = cv2.cvtColor(
source_frame,
cv2.COLOR_BGR2GRAY,
)
resized_base = cv2.resize(
source_gray,
(profile.base_width, profile.base_height),
interpolation=cv2.INTER_AREA,
)
return encode_decode_grayscale_jpeg(
resized_base,
profile.base_quality,
)
def encode_decode_roi(
source_frame: np.ndarray,
source_roi: tuple[int, int, int, int],
profile: PreviewProfile,
) -> np.ndarray:
"""
Вырезает ROI исходного BGR-кадра и формирует JPEG-состояние.
"""
if not profile.roi_enabled:
raise RuntimeError(
"Нельзя кодировать ROI для отключённого ROI-профиля."
)
x_min, y_min, x_max, y_max = source_roi
roi_bgr = source_frame[y_min:y_max, x_min:x_max]
if roi_bgr.size == 0:
raise RuntimeError("Вырезана пустая область ROI.")
roi_gray = cv2.cvtColor(
roi_bgr,
cv2.COLOR_BGR2GRAY,
)
resized_roi = cv2.resize(
roi_gray,
(profile.roi_width, profile.roi_height),
interpolation=cv2.INTER_AREA,
)
return encode_decode_grayscale_jpeg(
resized_roi,
profile.roi_quality,
)
def reconstruct_panel(
profile: PreviewProfile,
state: ProfileState,
panel_roi: tuple[int, int, int, int],
) -> np.ndarray:
"""
Восстанавливает одну grayscale-панель размером 640x360.
"""
if state.latest_base is None:
raise RuntimeError("BASE-состояние ещё не создано.")
reconstructed = cv2.resize(
state.latest_base,
(PANEL_WIDTH, PANEL_HEIGHT),
interpolation=cv2.INTER_LINEAR,
)
if profile.roi_enabled:
if state.latest_roi is None:
raise RuntimeError("ROI-состояние ещё не создано.")
x_min, y_min, x_max, y_max = panel_roi
resized_roi = cv2.resize(
state.latest_roi,
(x_max - x_min, y_max - y_min),
interpolation=cv2.INTER_LINEAR,
)
reconstructed[y_min:y_max, x_min:x_max] = resized_roi
return reconstructed
def draw_text_line(
image: np.ndarray,
text: str,
y_position: int,
text_color: tuple[int, int, int] = (255, 255, 255),
) -> None:
"""
Рисует одну ASCII-строку служебной информации OpenCV.
"""
cv2.putText(
image,
text,
(10, y_position),
cv2.FONT_HERSHEY_SIMPLEX,
0.46,
text_color,
1,
cv2.LINE_AA,
)
def draw_panel_information(
reconstructed_gray: np.ndarray,
profile: PreviewProfile,
state: ProfileState,
current_time: float,
panel_roi: tuple[int, int, int, int],
) -> np.ndarray:
"""
Добавляет рамку ROI, параметры и индикаторы обновления панели.
"""
panel = cv2.cvtColor(
reconstructed_gray,
cv2.COLOR_GRAY2BGR,
)
if profile.roi_enabled:
cv2.rectangle(
panel,
(panel_roi[0], panel_roi[1]),
(panel_roi[2] - 1, panel_roi[3] - 1),
(0, 255, 255),
2,
)
overlay = panel.copy()
cv2.rectangle(
overlay,
(0, 0),
(PANEL_WIDTH - 1, 142),
(0, 0, 0),
thickness=-1,
)
cv2.addWeighted(
overlay,
0.72,
panel,
0.28,
0.0,
panel,
)
base_age = max(
0.0,
current_time - state.last_base_update_time,
)
roi_age = max(
0.0,
current_time - state.last_roi_update_time,
)
base_is_new = (
base_age <= NEW_UPDATE_LABEL_SECONDS
)
roi_is_new = (
profile.roi_enabled
and roi_age <= NEW_UPDATE_LABEL_SECONDS
)
draw_text_line(panel, profile.name, 20)
draw_text_line(
panel,
(
f"BASE: {profile.base_width}x{profile.base_height}, "
f"{profile.base_fps:g} fps, Q{profile.base_quality}"
),
41,
)
if profile.roi_enabled:
roi_text = (
f"ROI: {profile.roi_width}x{profile.roi_height}, "
f"{profile.roi_fps:g} fps, Q{profile.roi_quality}"
)
roi_age_text = f"{roi_age:.3f} s"
else:
roi_text = "ROI: disabled"
roi_age_text = "disabled"
draw_text_line(panel, roi_text, 62)
draw_text_line(
panel,
(
f"payload={profile.measured_payload_kbps:.2f} kbps, "
f"source time={current_time:.3f} s"
),
83,
)
draw_text_line(
panel,
(
f"since BASE={base_age:.3f} s, "
f"since ROI={roi_age_text}"
),
104,
)
update_labels: list[str] = []
if base_is_new:
update_labels.append("NEW BASE")
if roi_is_new:
update_labels.append("NEW ROI")
if update_labels:
draw_text_line(
panel,
" | ".join(update_labels),
130,
text_color=(0, 255, 0),
)
return panel
def compose_grid(panels: list[np.ndarray]) -> np.ndarray:
"""
Объединяет четыре панели в сетку 2x2 размером 1280x720.
"""
if len(panels) != 4:
raise RuntimeError(
"Для сетки preview требуется ровно четыре панели."
)
for panel in panels:
if panel.shape != (PANEL_HEIGHT, PANEL_WIDTH, 3):
raise RuntimeError(
f"Некорректный размер панели: {panel.shape}"
)
top_row = np.hstack((panels[0], panels[1]))
bottom_row = np.hstack((panels[2], panels[3]))
grid = np.vstack((top_row, bottom_row))
if grid.shape != (OUTPUT_HEIGHT, OUTPUT_WIDTH, 3):
raise RuntimeError(
f"Некорректный размер сетки: {grid.shape}"
)
return grid
def open_preview_writer() -> tuple[cv2.VideoWriter, Path, bool]:
"""
Открывает MP4 writer либо разрешённый fallback AVI/MJPG.
"""
mp4_writer = cv2.VideoWriter(
str(MP4_PREVIEW_PATH),
cv2.VideoWriter_fourcc(*"mp4v"),
OUTPUT_FPS,
(OUTPUT_WIDTH, OUTPUT_HEIGHT),
True,
)
if mp4_writer.isOpened():
return mp4_writer, MP4_PREVIEW_PATH, False
mp4_writer.release()
if MP4_PREVIEW_PATH.exists():
MP4_PREVIEW_PATH.unlink()
avi_writer = cv2.VideoWriter(
str(AVI_PREVIEW_PATH),
cv2.VideoWriter_fourcc(*"MJPG"),
OUTPUT_FPS,
(OUTPUT_WIDTH, OUTPUT_HEIGHT),
True,
)
if not avi_writer.isOpened():
avi_writer.release()
if AVI_PREVIEW_PATH.exists():
AVI_PREVIEW_PATH.unlink()
raise RuntimeError(
"OpenCV не смог открыть ни MP4, ни AVI writer."
)
return avi_writer, AVI_PREVIEW_PATH, True
def create_preview(
source_path: Path,
profiles: list[PreviewProfile],
source_width: int,
source_height: int,
source_fps: float,
expected_frame_count: int,
) -> tuple[Path, bool, int, list[ProfileState]]:
"""
Создаёт preview-видео с одним выходным кадром на исходный кадр.
"""
if len(profiles) != 4:
raise RuntimeError(
"Ожидалось ровно четыре preview-профиля."
)
PREVIEW_DIRECTORY.mkdir(
parents=True,
exist_ok=True,
)
capture = cv2.VideoCapture(str(source_path))
if not capture.isOpened():
raise RuntimeError(
f"OpenCV не смог открыть видео: {source_path}"
)
writer, preview_path, fallback_used = open_preview_writer()
source_roi = normalized_roi_to_pixels(
source_width,
source_height,
)
panel_roi = normalized_roi_to_pixels(
PANEL_WIDTH,
PANEL_HEIGHT,
)
states = [
ProfileState()
for _ in profiles
]
frame_index = 0
try:
while True:
frame_read, source_frame = capture.read()
if not frame_read:
break
if source_frame is None:
raise RuntimeError(
f"Получен пустой кадр {frame_index}."
)
if (
source_frame.shape[1] != source_width
or source_frame.shape[0] != source_height
):
raise RuntimeError(
"Размер кадра отличается от метаданных."
)
current_time = frame_index / source_fps
panels: list[np.ndarray] = []
for profile, state in zip(profiles, states):
if should_update(
current_time,
state.next_base_time,
):
state.latest_base = encode_decode_base(
source_frame,
profile,
)
state.last_base_update_time = current_time
state.next_base_time += 1.0 / profile.base_fps
state.base_update_count += 1
if (
profile.roi_enabled
and should_update(
current_time,
state.next_roi_time,
)
):
state.latest_roi = encode_decode_roi(
source_frame,
source_roi,
profile,
)
state.last_roi_update_time = current_time
state.next_roi_time += 1.0 / profile.roi_fps
state.roi_update_count += 1
reconstructed = reconstruct_panel(
profile,
state,
panel_roi,
)
panels.append(
draw_panel_information(
reconstructed,
profile,
state,
current_time,
panel_roi,
)
)
writer.write(compose_grid(panels))
frame_index += 1
if (
frame_index % 100 == 0
or frame_index == expected_frame_count
):
print(
f" Written frames: "
f"{frame_index}/{expected_frame_count}"
)
except Exception:
capture.release()
writer.release()
if preview_path.exists():
preview_path.unlink()
raise
finally:
capture.release()
writer.release()
if frame_index != expected_frame_count:
if preview_path.exists():
preview_path.unlink()
raise RuntimeError(
"Число записанных кадров не совпало с исходным: "
f"{frame_index} != {expected_frame_count}."
)
return (
preview_path,
fallback_used,
frame_index,
states,
)
def read_frame_at(
capture: cv2.VideoCapture,
frame_index: int,
) -> tuple[bool, tuple[int, ...] | None]:
"""
Читает один кадр preview по индексу для итоговой проверки.
"""
capture.set(
cv2.CAP_PROP_POS_FRAMES,
frame_index,
)
frame_read, frame = capture.read()
if not frame_read or frame is None:
return False, None
return True, frame.shape
def verify_preview(
preview_path: Path,
source_frame_count: int,
source_duration_seconds: float,
) -> tuple[
int,
int,
float,
int,
float,
tuple[bool, tuple[int, ...] | None],
tuple[bool, tuple[int, ...] | None],
tuple[bool, tuple[int, ...] | None],
]:
"""
Проверяет контейнер, геометрию, FPS, длительность и три кадра.
"""
if not preview_path.exists():
raise RuntimeError(
f"Preview отсутствует: {preview_path}"
)
if preview_path.stat().st_size <= 0:
raise RuntimeError("Preview имеет нулевой размер.")
capture = cv2.VideoCapture(str(preview_path))
if not capture.isOpened():
raise RuntimeError(
f"OpenCV не смог открыть preview: {preview_path}"
)
try:
width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = float(capture.get(cv2.CAP_PROP_FPS))
frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
if fps <= 0.0:
raise RuntimeError("FPS preview равен нулю.")
duration_seconds = frame_count / fps
first_frame = read_frame_at(capture, 0)
middle_frame = read_frame_at(
capture,
frame_count // 2,
)
last_frame = read_frame_at(
capture,
frame_count - 1,
)
finally:
capture.release()
if width != OUTPUT_WIDTH or height != OUTPUT_HEIGHT:
raise RuntimeError(
f"Некорректное разрешение preview: {width}x{height}."
)
if frame_count != source_frame_count:
raise RuntimeError(
"Число кадров preview не совпало с исходным."
)
allowed_duration_difference = (
1.0 / fps + FRAME_TIME_EPSILON_SECONDS
)
if (
abs(duration_seconds - source_duration_seconds)
> allowed_duration_difference
):
raise RuntimeError(
"Длительность preview отличается более чем на один кадр."
)
for label, frame_result in [
("первый", first_frame),
("средний", middle_frame),
("последний", last_frame),
]:
if not frame_result[0]:
raise RuntimeError(
f"Не удалось прочитать {label} кадр preview."
)
return (
width,
height,
fps,
frame_count,
duration_seconds,
first_frame,
middle_frame,
last_frame,
)
def main() -> None:
"""
Создаёт и проверяет динамическое preview Lab027.
"""
print("Reading source video metadata...")
(
source_width,
source_height,
source_fps,
source_frame_count,
source_duration_seconds,
) = read_video_metadata(SOURCE_VIDEO_PATH)
print(f" Source: {SOURCE_VIDEO_PATH}")
print(f" Resolution: {source_width}x{source_height}")
print(f" FPS: {source_fps:.6f}")
print(f" Frames: {source_frame_count}")
print(f" Duration: {source_duration_seconds:.6f} s")
profiles = build_profiles()
print("Creating dynamic preview...")
(
preview_path,
fallback_used,
written_frame_count,
states,
) = create_preview(
source_path=SOURCE_VIDEO_PATH,
profiles=profiles,
source_width=source_width,
source_height=source_height,
source_fps=source_fps,
expected_frame_count=source_frame_count,
)
print("Verifying preview...")
(
preview_width,
preview_height,
preview_fps,
preview_frame_count,
preview_duration_seconds,
first_frame,
middle_frame,
last_frame,
) = verify_preview(
preview_path,
source_frame_count,
source_duration_seconds,
)
print("")
print(f"Preview path: {preview_path}")
print(f"Fallback used: {fallback_used}")
print(f"File size: {preview_path.stat().st_size} bytes")
print(
f"Preview resolution: "
f"{preview_width}x{preview_height}"
)
print(f"Preview FPS: {preview_fps:.6f}")
print(f"Written frames: {written_frame_count}")
print(f"Verified frames: {preview_frame_count}")
print(
f"Preview duration: "
f"{preview_duration_seconds:.6f} s"
)
print(f"First frame: {first_frame}")
print(f"Middle frame: {middle_frame}")
print(f"Last frame: {last_frame}")
print("")
print("Profile update counts:")
for profile, state in zip(profiles, states):
print(
f" {profile.name}: "
f"BASE={state.base_update_count}, "
f"ROI={state.roi_update_count}"
)
print("")
print("Lab027 temporal preview completed successfully.")
if __name__ == "__main__":
main()

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff