"""Обработка видео: извлечение лучшего кадра с распознанным номером.""" from pathlib import Path import cv2 import numpy as np from app.services.pipeline import PlatePipeline def process_video( pipeline: PlatePipeline, video_path: str | Path, frame_step: int = 10, # обрабатываем каждый 10-й кадр max_frames: int = 60, # максимум кадров к обработке (защита от длинных видео) ) -> dict | None: """ Читает видео, прогоняет кадры через пайплайн, возвращает ЛУЧШИЙ результат. Возвращает dict: - frame: np.ndarray (BGR) — лучший кадр - detection: dict — данные распознавания (как из pipeline.process) Или None, если ни в одном кадре номер не найден. "Лучший" = валидный по ГОСТ + максимальная confidence. Если валидных нет — просто максимальная confidence. """ cap = cv2.VideoCapture(str(video_path)) if not cap.isOpened(): raise ValueError("Не удалось открыть видеофайл (неподдерживаемый кодек?)") best = None # лучший валидный по ГОСТ best_any = None # лучший вообще (запасной) frame_idx = 0 processed = 0 try: while processed < max_frames: ret, frame = cap.read() if not ret: break # конец видео # Обрабатываем только каждый N-й кадр if frame_idx % frame_step != 0: frame_idx += 1 continue frame_idx += 1 processed += 1 detections = pipeline.process(frame) for d in detections: if d.get("fallback"): continue conf = d["confidence"] # Запасной кандидат — любой с максимальной уверенностью if best_any is None or conf > best_any["detection"]["confidence"]: best_any = {"frame": frame.copy(), "detection": d} # Приоритетный — валидный по ГОСТ с максимальной уверенностью if d["is_valid_gost"]: if best is None or conf > best["detection"]["confidence"]: best = {"frame": frame.copy(), "detection": d} finally: cap.release() return best if best is not None else best_any def is_video_file(filename: str) -> bool: ext = Path(filename).suffix.lower() return ext in {".mp4", ".avi", ".mov", ".mkv", ".webm", ".m4v"}