diff --git a/video_contact_sheet/README.md b/video_contact_sheet/README.md new file mode 100644 index 0000000..d032094 --- /dev/null +++ b/video_contact_sheet/README.md @@ -0,0 +1,19 @@ +# video_contact_sheet + +Generate visually rich contact-sheet thumbnails (aka filmstrips) for any number +of videos—handy for quick QA or cataloging. + +```bash +# single file +python -m video_contact_sheet.cli demo.mp4 -o out + +# entire folder, 8 threads +python -m video_contact_sheet.cli /videos -o out --threads 8 --cols 6 +``` + +## Features +Scene-change detection for “interesting” keyframes +Multithreaded extraction using OpenCV + ffmpeg +Footer shows duration / resolution / codec +Pure-Python, works on Windows/Linux/macOS + diff --git a/video_contact_sheet/__init__.py b/video_contact_sheet/__init__.py new file mode 100644 index 0000000..3a81b7c --- /dev/null +++ b/video_contact_sheet/__init__.py @@ -0,0 +1,10 @@ +""" +video_contact_sheet +~~~~~~~~~~~~~~~~~~~ +Generate key-frame contact sheets (filmstrip grids) from videos. + +>>> python -m video_contact_sheet.cli --help +""" + +# __init__.py +__version__ = "0.1.0" \ No newline at end of file diff --git a/video_contact_sheet/cli.py b/video_contact_sheet/cli.py new file mode 100644 index 0000000..a54b9fe --- /dev/null +++ b/video_contact_sheet/cli.py @@ -0,0 +1,50 @@ +""" +Command-line tool: bulk process folders / single files; supports multi-threading. +""" +from __future__ import annotations + +import click +from pathlib import Path +from tqdm import tqdm + +from .core import video_to_contact_sheet +from .utils import parallel_map + +@click.command() +@click.argument("inputs", nargs=-1, type=click.Path(exists=True, path_type=Path)) +@click.option("-o", "--out-dir", type=click.Path(path_type=Path), default="sheets") +@click.option("--max-frames", default=15, show_default=True) +@click.option("--cols", default=5, show_default=True, help="Columns in grid") +@click.option("--scene-thresh", default=30.0, show_default=True, help="Scene-change threshold (higher = fewer frames)") +@click.option("--threads", default=4, show_default=True, help="Parallel workers") +def main( + inputs: tuple[Path], + out_dir: Path, + max_frames: int, + cols: int, + scene_thresh: float, + threads: int, +): + """Generate contact sheets for videos or folders of videos.""" + + vids = [] + for p in inputs: + if p.is_dir(): + vids.extend(list(p.rglob("*.mp4"))) + else: + vids.append(p) + if not vids: + click.echo("No videos found.", err=True) + raise SystemExit(1) + + click.echo(f"Processing {len(vids)} video(s)…") + task = lambda v: video_to_contact_sheet( + v, out_dir, max_frames=max_frames, cols=cols, scene_thresh=scene_thresh + ) + for _ in tqdm(parallel_map(task, vids, max_workers=threads), total=len(vids)): + pass + click.echo(f"Done! Saved to {out_dir}") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/video_contact_sheet/core.py b/video_contact_sheet/core.py new file mode 100644 index 0000000..1c7d46a --- /dev/null +++ b/video_contact_sheet/core.py @@ -0,0 +1,115 @@ +""" +Core logic: Keyframe extraction, contact table splicing. +""" +from __future__ import annotations + +import math +import cv2 +import numpy as np +from pathlib import Path +from PIL import Image, ImageDraw, ImageFont +from typing import List, Dict, Tuple + +from .utils import ffprobe_metadata + + +def extract_keyframes( + video_path: Path, max_frames: int = 15, scene_thresh: float = 30.0 +) -> List[np.ndarray]: + """ + Return up to *max_frames* key frames (BGR ndarray). + Use HSV histogram difference for simple scene change detection. + """ + cap = cv2.VideoCapture(str(video_path)) + if not cap.isOpened(): + raise RuntimeError(f"Cannot open {video_path}") + + frames: List[np.ndarray] = [] + prev_hist = None + + while len(frames) < max_frames: + ok, frame = cap.read() + if not ok: + break + hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV) + hist = cv2.calcHist([hsv], [0, 1], None, [50, 60], [0, 180, 0, 256]) + hist = cv2.normalize(hist, hist).flatten() + + if prev_hist is None: + frames.append(frame) + else: + diff = cv2.compareHist(prev_hist, hist, cv2.HISTCMP_BHATTACHARYYA) + if diff * 100 > scene_thresh: # scale for intuition + frames.append(frame) + prev_hist = hist + + cap.release() + if not frames: + raise RuntimeError("No frames extracted") + return frames + + +FONT = ImageFont.load_default() + + +def make_contact_sheet( + frames: List[np.ndarray], + metadata: Dict, + cols: int = 5, + margin: int = 8, +) -> Image.Image: + + rows = math.ceil(len(frames) / cols) + h, w, _ = frames[0].shape + sheet_w = w * cols + margin * (cols + 1) + sheet_h = h * rows + margin * (rows + 1) + 60 + canvas = Image.new("RGB", (sheet_w, sheet_h), "black") + + for idx, frame in enumerate(frames): + r = idx // cols + c = idx % cols + x = margin + c * (w + margin) + y = margin + r * (h + margin) + canvas.paste(Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)), (x, y)) + + draw = ImageDraw.Draw(canvas) + text = ( + f"{metadata['file']} | {metadata['duration']:.1f}s " + f"| {metadata['width']}x{metadata['height']} | {metadata['codec']}" + ) + bbox = draw.textbbox((0, 0), text, font=FONT) + tw = bbox[2] - bbox[0] + th = bbox[3] - bbox[1] + draw.text(((sheet_w - tw) // 2, sheet_h - th - 10), text, fill="white", font=FONT) + return canvas + + +def video_to_contact_sheet( + video_path: Path, + out_dir: Path, + max_frames: int = 15, + cols: int = 5, + scene_thresh: float = 30.0, + quality: int = 85, +) -> Path: + + frames = extract_keyframes(video_path, max_frames=max_frames, scene_thresh=scene_thresh) + meta = _collect_meta(video_path) + sheet = make_contact_sheet(frames, meta, cols=cols) + + out_dir.mkdir(parents=True, exist_ok=True) + out_path = out_dir / f"{video_path.stem}_sheet.jpg" + sheet.save(out_path, "JPEG", quality=quality) + return out_path + + +def _collect_meta(video_path: Path) -> Dict: + info = ffprobe_metadata(video_path) + v_stream = next(s for s in info["streams"] if s["codec_type"] == "video") + return dict( + file=video_path.name, + duration=float(info["format"]["duration"]), + width=int(v_stream["width"]), + height=int(v_stream["height"]), + codec=v_stream["codec_name"], + ) \ No newline at end of file diff --git a/video_contact_sheet/requirements.txt b/video_contact_sheet/requirements.txt new file mode 100644 index 0000000..0acdd06 --- /dev/null +++ b/video_contact_sheet/requirements.txt @@ -0,0 +1,6 @@ +opencv-python>=4.9 +ffmpeg-python>=0.2 +Pillow>=10.3 +tqdm>=4.66 +click>=8.1 +pytest>=8.2 \ No newline at end of file diff --git a/video_contact_sheet/utils.py b/video_contact_sheet/utils.py new file mode 100644 index 0000000..820db27 --- /dev/null +++ b/video_contact_sheet/utils.py @@ -0,0 +1,35 @@ +""" +Shared utilities: ffprobe metadata, parallel helpers, paths. +""" +from __future__ import annotations + +import json +import subprocess +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Any, Dict, Iterable, List + +FFPROBE_CMD = [ + "ffprobe", + "-v", + "error", + "-print_format", + "json", + "-show_format", + "-show_streams", +] + + +def ffprobe_metadata(path: Path) -> Dict[str, Any]: + """Return ffprobe JSON metadata for *path*.""" + proc = subprocess.run( + FFPROBE_CMD + [str(path)], capture_output=True, text=True, check=True + ) + return json.loads(proc.stdout) + + +def parallel_map(func, iterable: Iterable, max_workers: int = 4) -> List: + """Lightweight ThreadPool map that preserves order.""" + with ThreadPoolExecutor(max_workers=max_workers) as pool: + futures = [pool.submit(func, item) for item in iterable] + return [f.result() for f in futures] \ No newline at end of file