ilia
6a78c84bcd
- capture/: frame sampling at configurable fps, ffmpeg audio extraction (WAV, whisper-ready) - vision/: changed-pixel motion scoring, stable/moving segmentation, contour + perspective document detection, Laplacian sharpness scoring, optional CLAHE enhancement - pipeline: two-pass detect (motion timeline, then best-frame crop per stable window) writing crops, debug frames, report.json, and motion_scores.csv - CLI: probe / detect / extract-audio subcommands - config.yaml with tunable thresholds; placeholder packages for events, transcribe, ocr, naming, pdf, review_cli - synthetic sample video generator + 16 unit tests
92 lines
2.6 KiB
Python
92 lines
2.6 KiB
Python
"""Frame extraction from a pre-recorded video file."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Iterator
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
|
|
@dataclass
|
|
class VideoMeta:
|
|
path: str
|
|
fps: float
|
|
frame_count: int
|
|
duration_s: float
|
|
width: int
|
|
height: int
|
|
|
|
|
|
@dataclass
|
|
class Frame:
|
|
index: int # frame index in the source video (native fps)
|
|
t: float # timestamp in seconds
|
|
image: np.ndarray # BGR
|
|
|
|
|
|
def probe_video(path: str | Path) -> VideoMeta:
|
|
path = Path(path)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Video not found: {path}")
|
|
cap = cv2.VideoCapture(str(path))
|
|
if not cap.isOpened():
|
|
raise ValueError(f"Could not open video (unsupported codec?): {path}")
|
|
try:
|
|
fps = cap.get(cv2.CAP_PROP_FPS) or 0.0
|
|
frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
|
|
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
|
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
|
finally:
|
|
cap.release()
|
|
if fps <= 0:
|
|
raise ValueError(f"Video reports invalid fps ({fps}): {path}")
|
|
return VideoMeta(
|
|
path=str(path),
|
|
fps=fps,
|
|
frame_count=frame_count,
|
|
duration_s=frame_count / fps if frame_count > 0 else 0.0,
|
|
width=width,
|
|
height=height,
|
|
)
|
|
|
|
|
|
def iter_frames(
|
|
path: str | Path,
|
|
sample_fps: float = 8.0,
|
|
resize_width: int | None = None,
|
|
) -> Iterator[Frame]:
|
|
"""Sequentially decode the video, yielding frames at ~sample_fps.
|
|
|
|
Decoding is sequential (no seeking) so it works reliably across codecs.
|
|
If resize_width is set, frames are downscaled to that width preserving
|
|
aspect ratio (used for motion analysis; crops use full resolution).
|
|
"""
|
|
meta = probe_video(path)
|
|
step = max(1, round(meta.fps / sample_fps))
|
|
|
|
cap = cv2.VideoCapture(str(path))
|
|
try:
|
|
index = 0
|
|
while True:
|
|
ok = cap.grab()
|
|
if not ok:
|
|
break
|
|
if index % step == 0:
|
|
ok, image = cap.retrieve()
|
|
if not ok:
|
|
break
|
|
if resize_width is not None and image.shape[1] > resize_width:
|
|
scale = resize_width / image.shape[1]
|
|
image = cv2.resize(
|
|
image,
|
|
(resize_width, max(1, round(image.shape[0] * scale))),
|
|
interpolation=cv2.INTER_AREA,
|
|
)
|
|
yield Frame(index=index, t=index / meta.fps, image=image)
|
|
index += 1
|
|
finally:
|
|
cap.release()
|