#!/usr/bin/env python """ Local-GPU preprocessing: extracts frames, CLIP embeddings, and clips for every video in the benchmark (or every .mp4 under CHNB_VIDEO_DIR if --all-videos). Steps (run all by default; use --step to pick one): frames — uniform + fps frame extraction to data/frames/ clip_embeddings — CLIP features over uniform frames to data/clip_embeddings/ clips — mp4 clip extraction (shot_grouped or fixed) to data/clips/ Usage: python run_server.py python run_server.py --step frames --uniform-n 64 128 256 python run_server.py --step clip_embeddings python run_server.py --step clips --chunk-mode shot_grouped python run_server.py --all-videos """ from __future__ import annotations import argparse import json import subprocess import sys from pathlib import Path _PROJECT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(_PROJECT)) import numpy as np from PIL import Image from tqdm.auto import tqdm from util.benchmark import load_benchmark, unique_movies from util.frames import ( extract_frames_fps, extract_frames_uniform, get_clip_embeddings, ) from util.paths import ( BENCHMARK_PATH, VIDEO_DIR, CLIPS_DIR, get_clip_path, get_clips_dir, get_frames_dir, ensure_dir, ) from util.shots import extract_all_clips def ffprobe_duration(video_path: Path) -> float: try: r = subprocess.run( ['ffprobe', '-v', 'quiet', '-print_format', 'json', '-show_format', str(video_path)], capture_output=True, text=True, ) return float(json.loads(r.stdout)['format']['duration']) except Exception: from decord import VideoReader, cpu vr = VideoReader(str(video_path), ctx=cpu(0), num_threads=1) d = len(vr) / vr.get_avg_fps() del vr return d # ---------------------------------------------------------------- frames def run_frames(movies: list[str], args) -> dict[str, float]: """Extract fps + uniform frames. Returns {vid: duration_sec}.""" durations: dict[str, float] = {} for vid in tqdm(movies, desc='frames'): vpath = VIDEO_DIR / f'{vid}.mp4' if not vpath.exists(): print(f' SKIP (video missing): {vid}') continue dur = ffprobe_duration(vpath) durations[vid] = dur # fps frames if args.fps > 0: d = get_frames_dir(vid, 'fps', args.fps) n_expected = int(np.ceil(dur * args.fps)) n_existing = len(list(d.glob('*.jpg'))) if n_existing < n_expected * 0.96: extract_frames_fps(str(vpath), fps=args.fps, cache_dir=str(d), low_memory=args.low_memory) # uniform frames for n in args.uniform_n: d = get_frames_dir(vid, 'uniform', n) if len(list(d.glob('*.jpg'))) < n: extract_frames_uniform(str(vpath), n, cache_dir=str(d), low_memory=args.low_memory) print(f'Frames: processed {len(durations)} / {len(movies)} movies.') return durations # ---------------------------------------------------------------- CLIP def run_clip_embeddings(movies: list[str], args) -> None: """ Compute CLIP features over EVERY uniform-N frame set in `args.uniform_n`. Idempotent: skips any (vid, n) pair whose .npy already exists. """ for clip_n in sorted(args.uniform_n): print(f'\n=== CLIP embeddings: uniform_{clip_n} ===') for vid in tqdm(movies, desc=f'CLIP (uniform_{clip_n})'): out = get_clip_path(vid, 'uniform', clip_n) if out.exists(): continue uni_dir = get_frames_dir(vid, 'uniform', clip_n) jpgs = sorted(uni_dir.glob('*.jpg')) if not jpgs: continue images = [Image.open(f) for f in jpgs] get_clip_embeddings(images, cache_path=str(out)) print('CLIP embeddings done.') # ---------------------------------------------------------------- clips def run_clips(movies: list[str], durations: dict[str, float], args) -> None: """Extract mp4 clip segments. Respects existing manifests unless --force.""" for chunk_mode in args.chunk_modes: print(f'\n=== extracting clips: {chunk_mode} ===') for vid in tqdm(movies, desc=f'clips ({chunk_mode})'): vpath = VIDEO_DIR / f'{vid}.mp4' if not vpath.exists(): continue dur = durations.get(vid) if dur is None: dur = ffprobe_duration(vpath) if dur <= 0: continue clips_dir = get_clips_dir(vid, chunk_mode) manifest = clips_dir / 'manifest.json' if manifest.exists() and not args.force: try: with open(manifest) as f: m = json.load(f) if all((clips_dir / item['clip']).exists() for item in m): continue except Exception: pass try: clips = extract_all_clips( vid, str(vpath), args.assume_fps, dur, chunk_mode=chunk_mode, with_audio=args.with_audio, ) print(f' {vid}: {len(clips)} clips') except Exception as e: print(f' Error {vid}: {type(e).__name__}: {e!r}') print('Clip extraction done.') # ---------------------------------------------------------------- movies list def resolve_movies(args) -> list[str]: if args.all_videos: mp4s = sorted(VIDEO_DIR.glob('*.mp4')) print(f'Found {len(mp4s)} .mp4 files under {VIDEO_DIR}') return [p.stem for p in mp4s] df = load_benchmark(args.benchmark) movies = unique_movies(df) print(f'Loaded {len(movies)} movies from {args.benchmark}') return movies def main() -> None: ap = argparse.ArgumentParser(description=__doc__.split('\n\n', 1)[0]) ap.add_argument('--step', choices=['frames', 'clip_embeddings', 'clips', 'all'], default='all') ap.add_argument('--benchmark', default=str(BENCHMARK_PATH)) ap.add_argument('--all-videos', action='store_true', help='process every .mp4 under VIDEO_DIR (not just benchmark movies)') ap.add_argument('--uniform-n', type=int, nargs='+', default=[64, 128, 256], help='Pool sizes to extract frames AND compute CLIP embeddings at. ') ap.add_argument('--fps', type=float, default=0.5) ap.add_argument('--low-memory', action='store_true', default=False) ap.add_argument('--chunk-modes', nargs='+', default=['shot_grouped'], choices=['shot_grouped', 'fixed']) ap.add_argument('--with-audio', action='store_true', default=True) ap.add_argument('--no-audio', dest='with_audio', action='store_false') ap.add_argument('--assume-fps', type=float, default=24.0, help='fps used when reading .scenes.txt frame indices (24 is typical for film)') ap.add_argument('--force', action='store_true', help='re-extract even if outputs exist') args = ap.parse_args() movies = resolve_movies(args) durations: dict[str, float] = {} if args.step in ('frames', 'all'): durations = run_frames(movies, args) if args.step in ('clip_embeddings', 'all'): run_clip_embeddings(movies, args) if args.step in ('clips', 'all'): # Need durations; if 'frames' was skipped, compute them lazily if not durations: print('Gathering durations for clip extraction...') for vid in tqdm(movies, desc='ffprobe'): vpath = VIDEO_DIR / f'{vid}.mp4' if vpath.exists(): durations[vid] = ffprobe_duration(vpath) run_clips(movies, durations, args) print('\nDone.') if __name__ == '__main__': main()