utils/video.py

import logging
import os
import shutil
import subprocess
import tempfile
from typing import List, Tuple

import cv2
import numpy as np


def extract_frames(video_path: str) -> Tuple[List[np.ndarray], float, int, int]:
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise ValueError("Unable to open video.")

    fps = cap.get(cv2.CAP_PROP_FPS) or 0.0
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    frames: List[np.ndarray] = []
    success, frame = cap.read()
    while success:
        frames.append(frame)
        success, frame = cap.read()

    cap.release()

    if not frames:
        raise ValueError("Video decode produced zero frames.")

    return frames, fps, width, height


def _transcode_with_ffmpeg(src_path: str, dst_path: str) -> None:
    cmd = [
        "ffmpeg",
        "-y",
        "-i",
        src_path,
        "-c:v",
        "libx264",
        "-preset",
        "veryfast",
        "-pix_fmt",
        "yuv420p",
        "-movflags",
        "+faststart",
        dst_path,
    ]
    process = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
    if process.returncode != 0:
        raise RuntimeError(process.stderr.decode("utf-8", errors="ignore"))


def write_video(frames: List[np.ndarray], output_path: str, fps: float, width: int, height: int) -> None:
    if not frames:
        raise ValueError("No frames available for writing.")
    temp_fd, temp_path = tempfile.mkstemp(prefix="raw_", suffix=".mp4")
    os.close(temp_fd)
    writer = cv2.VideoWriter(temp_path, cv2.VideoWriter_fourcc(*"mp4v"), fps or 1.0, (width, height))
    if not writer.isOpened():
        os.remove(temp_path)
        raise ValueError("Failed to open VideoWriter.")

    for frame in frames:
        writer.write(frame)

    writer.release()
    try:
        _transcode_with_ffmpeg(temp_path, output_path)
        logging.debug("Transcoded video to H.264 for browser compatibility.")
        os.remove(temp_path)
    except FileNotFoundError:
        logging.warning("ffmpeg not found; serving fallback MP4V output.")
        shutil.move(temp_path, output_path)
    except RuntimeError as exc:
        logging.warning("ffmpeg transcode failed (%s); serving fallback MP4V output.", exc)
        shutil.move(temp_path, output_path)