festifun/backend/festival4d/frames.py
m3ultra bb542960c2 Lane B (M2 + M8 geometry): COLMAP reconstruction, geometry math, graceful degradation
Implements spec M2 (frame sampling, COLMAP orchestration, TXT-model parsing, scene
normalization, pose interpolation, PLY + DB export) and the M8 geometry functions.

geometry.py (M8): slerp_pose (shortest-arc, double-cover), ray_from_pixel (COLMAP
+y-down back-projection), triangulate_rays (closest-point + parallel guard),
nearest_point_on_ray (in-front radius cylinder). Frozen colmap_to_threejs untouched.

frames.py (M2): variance-of-Laplacian sharpness + windowed sharpest-frame sampling.

sfm.py (M2): hand-written COLMAP images/cameras/points3D parsers; normalize_scene
(centroid->0, camera sphere r->10, up->+Y) as one similarity transform over points+poses;
interpolate_poses (slerp+lerp, no extrapolation); build-aware COLMAP CLI orchestration
(3.x/4.x option detection, CPU SIFT, single-camera-per-folder, undistort->TXT, largest
component by images.bin header count); run_reconstruct with full graceful degradation
(COLMAP absent / <60% / <2 videos / no frames / missing files -> poses left untouched).

Tests: 61 pass (24 foundation + 37 lane-B) with independent oracles (scipy Slerp,
projection-inverse, closed-form geometry) and every degradation/DB-safety branch covered.
Validated end-to-end against real COLMAP 4.1.0 (6 components, largest selected, weak ->
graceful degradation, exit 0).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-16 08:40:46 +10:00

89 lines
3.4 KiB
Python

"""Sharpness-aware frame sampling for SfM (spec M2, lane B).
STUB — lane B fills the bodies; the public signatures below are frozen (``sfm.py`` calls
these). Sample candidate frames ~2/s per video; within each 0.5 s window keep the sharpest
frame (variance of Laplacian via OpenCV). Write JPEGs named ``{video_id}_{frame_idx}.jpg``
into ``config.FRAMES_DIR`` (one subfolder per video for COLMAP's single-camera-per-folder).
"""
from __future__ import annotations
import logging
from pathlib import Path
import numpy as np
log = logging.getLogger("festival4d.frames")
def sharpness(image) -> float:
"""Variance of the Laplacian of an image (OpenCV) — higher is sharper.
Accepts a BGR or grayscale ``ndarray`` (as returned by ``cv2.VideoCapture.read``).
The variance of the Laplacian is the standard focus/blur measure: a sharp frame has
strong high-frequency edges (high variance), a blurred one is smooth (low variance).
"""
import cv2
img = np.asarray(image)
if img.ndim == 3:
img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
return float(cv2.Laplacian(img, cv2.CV_64F).var())
def sample_frames(video_path: Path, video_id: int, out_dir: Path,
target_fps: float = 2.0, window_s: float = 0.5) -> list[Path]:
"""Sample the sharpest frame per ``window_s`` window from ``video_path`` (spec M2).
Decodes every frame, buckets them into non-overlapping ``window_s``-second windows, and
keeps the single sharpest frame (variance of Laplacian) in each window. This yields
~``1 / window_s`` frames per second (≈ ``target_fps`` at the defaults), biased toward the
in-focus frames COLMAP needs. Writes JPEGs named ``{video_id}_{frame_idx}.jpg`` (the
original decoded ``frame_idx``, so ``t_video = frame_idx / fps``) into ``out_dir``.
Returns the written JPEG paths, ordered by frame index.
"""
import cv2
out_dir = Path(out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
cap = cv2.VideoCapture(str(video_path))
if not cap.isOpened():
raise RuntimeError(f"could not open video for frame sampling: {video_path}")
try:
fps = float(cap.get(cv2.CAP_PROP_FPS))
if not np.isfinite(fps) or fps <= 0.0:
log.warning("frames: %s reported fps=%s; falling back to 30.0", video_path, fps)
fps = 30.0
bucket_s = float(window_s) if window_s and window_s > 0 else 0.5
# window_idx -> (sharpness, frame_idx, frame_bgr). Only the current best per window
# is retained, so memory stays ~O(number of windows), not O(frames).
best: dict[int, tuple[float, int, np.ndarray]] = {}
frame_idx = 0
while True:
ok, frame = cap.read()
if not ok:
break
t = frame_idx / fps
win = int(t / bucket_s)
s = sharpness(frame)
cur = best.get(win)
if cur is None or s > cur[0]:
best[win] = (s, frame_idx, frame)
frame_idx += 1
finally:
cap.release()
written: list[Path] = []
for win in sorted(best):
_, fidx, frame = best[win]
path = out_dir / f"{video_id}_{fidx}.jpg"
if not cv2.imwrite(str(path), frame):
raise RuntimeError(f"failed to write frame JPEG: {path}")
written.append(path)
log.info("frames: %s -> %d frames sampled into %s", Path(video_path).name,
len(written), out_dir)
return written