SmolVLM2-2.2B を用いたフレーム処理によるローカル動画要約パイプライン

# Python 3.10+ required
python --version

python -m venv smolvlm2-env
source smolvlm2-env/bin/activate       # macOS / Linux
smolvlm2-env\Scripts\activate          # Windows

# Install from the stable SmolVLM-2 branch -- required for SmolVLM2 support
pip install git+https://github.com/huggingface/transformers@v4.49.0-SmolVLM-2

# Core dependencies
pip install torch torchvision --index-url https://download.pytorch.org/whl/cu121
pip install \
  opencv-python \
  Pillow \
  numpy \
  num2words \
  accelerate

# Flash Attention 2 for CUDA -- significantly faster on NVIDIA GPUs
# Skip this on Apple Silicon and CPU -- it is CUDA-only
pip install flash-attn --no-build-isolation

# decord -- required for SmolVLM2's native video input path (used in Section 5)
pip install decord

# device_check.py
# Run: python device_check.py

import torch

def detect_device():
    if torch.cuda.is_available():
        name  = torch.cuda.get_device_name(0)
        vram  = torch.cuda.get_device_properties(0).total_memory / 1e9
        print(f"CUDA: {name} ({vram:.1f} GB VRAM)")
        return "cuda", torch.bfloat16, "flash_attention_2"
    elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
        print("Apple Silicon MPS detected")
        return "mps", torch.float16, "eager"
    else:
        print("CPU fallback (slow -- consider Colab T4)")
        return "cpu", torch.float32, "eager"

if __name__ == "__main__":
    device, dtype, attn = detect_device()
    print(f"Device: {device} | dtype: {dtype} | attn: {attn}")

python device_check.py

# frame_extractor.py
# Prerequisites: pip install opencv-python Pillow numpy
# Usage: from frame_extractor import FrameExtractor

import cv2
import numpy as np
from PIL import Image

class FrameExtractor:
    """
    Extracts video frames as PIL Images for SmolVLM2 inference.
    Each extracted frame is paired with its timestamp in seconds.

    SmolVLM2 uses ~81 visual tokens per image. At 50 frames that is
    roughly 4,050 image tokens -- the practical upper limit before VRAM
    pressure affects generation quality on consumer GPUs.
    """

    MAX_FRAMES = 50

    def __init__(self, max_frames: int = MAX_FRAMES):
        """
        Args:
            max_frames: Hard cap on extracted frames. Default 50 matches
                        the SmolVLM2 reference pipeline's tested upper limit.
        """
        self.max_frames = max_frames

    def uniform_sample(self, video_path: str) -> list[tuple[float, Image.Image]]:
        """
        Extract evenly spaced frames across the full video duration.
        Best for: meeting recordings, lectures, tutorials, course content.

        Returns:
            List of (timestamp_seconds, PIL_Image) in chronological order.
        """
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            raise IOError(f"Cannot open video: {video_path}")

        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        fps          = cap.get(cv2.CAP_PROP_FPS) or 30.0
        n_extract    = min(self.max_frames, total_frames)

        # Build frame indices spread evenly from first to last frame
        indices = np.linspace(0, total_frames - 1, n_extract, dtype=int)
        results = []

        for idx in indices:
            cap.set(cv2.CAP_PROP_POS_FRAMES, int(idx))
            ret, frame = cap.read()
            if not ret:
                continue
            timestamp = round(idx / fps, 2)
            rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            results.append((timestamp, Image.fromarray(rgb)))

        cap.release()
        return results

    def keyframe_sample(
        self, video_path: str, diff_threshold: float = 30.0
    ) -> list[tuple[float, Image.Image]]:
        """
        Extract frames where visual content changes significantly.
        Best for: surveillance, event detection, highlight extraction.

        Uses mean absolute pixel difference between consecutive grayscale frames
        as the change signal. When the diff exceeds diff_threshold, a new
        keyframe is recorded.

        Args:
            diff_threshold: Mean pixel difference to treat as a scene change.
                            30.0 works for most commercial content.
                            Lower = more sensitive, higher = fewer frames.

        Returns:
            List of (timestamp_seconds, PIL_Image) in chronological order,
            capped at self.max_frames.
        """
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            raise IOError(f"Cannot open video: {video_path}")

        fps       = cap.get(cv2.CAP_PROP_FPS) or 30.0
        results   = []
        prev_gray = None
        idx       = 0

        while len(results) < self.max_frames:
            ret, frame = cap.read()
            if not ret:
                break

            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

            if prev_gray is None:
                # Always capture the first frame as a baseline
                rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                results.append((round(idx / fps, 2), Image.fromarray(rgb)))
            else:
                diff = np.mean(np.abs(gray.astype(float) - prev_gray.astype(float)))
                if diff > diff_threshold:
                    rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                    results.append((round(idx / fps, 2), Image.fromarray(rgb)))

            prev_gray = gray
            idx += 1

        cap.release()
        return results

# smolvlm2_loader.py
# Prerequisites: transformers from v4.49.0-SmolVLM-2 branch, torch, flash-attn (CUDA only)
# Run: python smolvlm2_loader.py your_video.mp4

import sys
import torch
from PIL import Image
from transformers import AutoProcessor, AutoModelForImageTextToText

MODEL_ID = "HuggingFaceTB/SmolVLM2-2.2B-Instruct"

def load_model():
    """
    Load SmolVLM2-2.2B and its processor.
    Automatically selects Flash Attention 2 on CUDA, eager mode elsewhere.
    First run downloads ~4.5 GB of weights to ~/.cache/huggingface/hub.
    """
    if torch.cuda.is_available():
        dtype  = torch.bfloat16
        device = "cuda"
        attn   = "flash_attention_2"
    elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
        dtype  = torch.float16
        device = "mps"
        attn   = "eager"
    else:
        dtype  = torch.float32
        device = "cpu"
        attn   = "eager"

    print(f"Loading {MODEL_ID} on {device}...")

    processor = AutoProcessor.from_pretrained(MODEL_ID)

    model = AutoModelForImageTextToText.from_pretrained(
        MODEL_ID,
        torch_dtype=dtype,
        _attn_implementation=attn,
    ).to(device)

    model.eval()
    print(f"Model ready on {device}")
    return model, processor

def describe_frame(
    model,
    processor,
    frame: Image.Image,
    prompt: str = "Describe what is happening in this frame in detail. Note any text, people, objects, or actions visible.",
    max_new_tokens: int = 256,
) -> str:
    """
    Run SmolVLM2 inference on a single PIL Image.

    The chat template expects image content before text content in the
    message -- this mirrors the training data format and is important
    for reliable output.

    Args:
        frame:          A PIL Image (from FrameExtractor)
        prompt:         What to ask the model about this frame
        max_new_tokens: Maximum response length in tokens

    Returns:
        Model response as a plain string
    """
    messages = [
        {
            "role": "user",
            "content": [
                # Image placed before text -- matches SmolVLM2 training format
                {"type": "image"},
                {"type": "text", "text": prompt},
            ],
        }
    ]

    # apply_chat_template formats the message and injects visual token placeholders
    input_text = processor.apply_chat_template(
        messages,
        add_generation_prompt=True,
    )

    inputs = processor(
        images=[frame],
        text=input_text,
        return_tensors="pt",
    ).to(model.device)

    with torch.no_grad():
        output_ids = model.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            do_sample=False,   # Greedy decoding for consistent structured output
        )

    # Decode only the newly generated tokens -- strip the input prompt
    new_tokens = output_ids[0][inputs["input_ids"].shape[-1]:]
    return processor.decode(new_tokens, skip_special_tokens=True).strip()

# ── Quick sanity check ────────────────────────────────────────────────────────

if __name__ == "__main__":
    from frame_extractor import FrameExtractor

    if len(sys.argv) < 2:
        print("Usage: python smolvlm2_loader.py ")
        sys.exit(1)

    model, processor = load_model()
    extractor = FrameExtractor(max_frames=5)

    frames = extractor.uniform_sample(sys.argv[1])
    ts, first_frame = frames[0]

    print(f"\nDescribing frame at {ts}s...")
    description = describe_frame(model, processor, first_frame)
    print(f"\n{description}")

python smolvlm2_loader.py your_video.mp4

# video_summarizer.py
# Prerequisites: frame_extractor.py and smolvlm2_loader.py in the same directory
# Run: python video_summarizer.py meeting_recording.mp4 --output summary.json

import re
import json
import argparse
from dataclasses import dataclass, field
import cv2
import torch

from frame_extractor import FrameExtractor
from smolvlm2_loader import load_model, describe_frame

# ── Data models ───────────────────────────────────────────────────────────────

@dataclass
class FrameDescription:
    timestamp: float
    frame_index: int
    description: str

@dataclass
class VideoSummary:
    video_path: str
    duration_seconds: float
    frames_analyzed: int
    frame_descriptions: list[FrameDescription]
    narrative_summary: str
    action_items: list[str] = field(default_factory=list)
    key_moments: list[dict] = field(default_factory=list)

# ── Per-frame prompt ──────────────────────────────────────────────────────────

FRAME_PROMPT = """You are analyzing a frame from a recorded meeting.

Describe what you see concisely but completely:
- Who or what is visible (people, whiteboards, screens, slides)
- Any readable text (slide titles, whiteboard content, screen content)
- The apparent activity (presenting, discussing, writing, listening)

Keep your response to 2-3 sentences."""

# ── Synthesis prompt ──────────────────────────────────────────────────────────

def build_synthesis_prompt(descriptions: list[FrameDescription], duration: float) -> str:
    """Build the second-pass prompt that synthesizes frame descriptions into a report."""
    frames_text = "\n".join(
        f"[{int(d.timestamp // 60):02d}:{int(d.timestamp % 60):02d}] {d.description}"
        for d in descriptions
    )
    return f"""Below are time-stamped descriptions of frames from a {duration:.0f}-second meeting recording.

{frames_text}

Based on these descriptions, provide:

1. NARRATIVE SUMMARY: A 3-5 sentence summary of what the meeting covered, who participated (if visible), and what decisions or conclusions were reached.

2. ACTION ITEMS: A bullet list of concrete tasks or follow-ups mentioned or implied in the meeting. Start each with a dash (-).

3. KEY MOMENTS: A bullet list of the 3-5 most significant moments with their timestamps in [MM:SS] format.

Format your response with clear headings for each section."""

# ── Output parser ─────────────────────────────────────────────────────────────

def parse_action_items(text: str) -> list[str]:
    """Extract bullet-point action items from the synthesis output."""
    items = []
    for line in text.split("\n"):
        stripped = line.strip()
        if re.match(r"^[-*•]\s+", stripped) or re.match(r"^\d+\.\s+", stripped):
            clean = re.sub(r"^[-*•\d.]+\s*", "", stripped).strip()
            if clean and len(clean) > 5:
                items.append(clean)
    return items

def parse_key_moments(text: str) -> list[dict]:
    """Extract key moments with timestamps from the synthesis output."""
    moments = []
    pattern = re.compile(r"\[(\d{2}:\d{2})\]\s*(.+)")
    for match in pattern.finditer(text):
        moments.append({
            "timestamp_label": match.group(1),
            "description": match.group(2).strip()
        })
    return moments

# ── Main summarizer class ─────────────────────────────────────────────────────

class VideoSummarizer:
    """
    End-to-end local video summarizer using SmolVLM2-2.2B.
    Two-pass strategy: per-frame descriptions + synthesis narrative.
    Works on scanned, digital, and live-recorded videos alike.
    """

    def __init__(self, batch_size: int = 8):
        """
        Args:
            batch_size: Frames to describe per inference batch.
                        Tune based on VRAM: 8 for 8 GB, 16 for 16 GB.
                        Each frame uses ~81 visual tokens; lower batch = less peak VRAM.
        """
        self.model, self.processor = load_model()
        self.extractor = FrameExtractor(max_frames=50)
        self.batch_size = batch_size

    def _get_duration(self, video_path: str) -> float:
        cap = cv2.VideoCapture(video_path)
        frames = cap.get(cv2.CAP_PROP_FRAME_COUNT)
        fps    = cap.get(cv2.CAP_PROP_FPS) or 30.0
        cap.release()
        return round(frames / fps, 2)

    def summarize(self, video_path: str, mode: str = "uniform") -> VideoSummary:
        """
        Summarize a video file.

        Args:
            video_path: Path to the video file (mp4, avi, mov, mkv)
            mode:       "uniform" for even coverage, "keyframe" for scene changes

        Returns:
            VideoSummary with per-frame descriptions, narrative, and action items
        """
        duration = self._get_duration(video_path)
        print(f"Video: {video_path} ({duration:.0f}s)")

        # ── Pass 1: Extract frames ─────────────────────────────────────────
        if mode == "keyframe":
            frames = self.extractor.keyframe_sample(video_path)
        else:
            frames = self.extractor.uniform_sample(video_path)

        print(f"Extracted {len(frames)} frames -- describing in batches of {self.batch_size}...")

        # ── Pass 2: Describe each frame ────────────────────────────────────
        descriptions: list[FrameDescription] = []

        for batch_start in range(0, len(frames), self.batch_size):
            batch = frames[batch_start : batch_start + self.batch_size]

            for local_idx, (timestamp, img) in enumerate(batch):
                global_idx = batch_start + local_idx
                print(f"  [{global_idx + 1}/{len(frames)}] Describing frame at {timestamp}s...")

                desc = describe_frame(
                    self.model,
                    self.processor,
                    img,
                    prompt=FRAME_PROMPT,
                    max_new_tokens=128,   # Keep frame descriptions concise
                )
                descriptions.append(FrameDescription(
                    timestamp=timestamp,
                    frame_index=global_idx,
                    description=desc,
                ))

        # ── Pass 3: Synthesis ──────────────────────────────────────────────
        print("\nRunning synthesis pass...")

        synthesis_prompt = build_synthesis_prompt(descriptions, duration)
        synthesis_messages = [
            {
                "role": "user",
                "content": [{"type": "text", "text": synthesis_prompt}],
            }
        ]
        synthesis_text_input = self.processor.apply_chat_template(
            synthesis_messages,
            add_generation_prompt=True,
        )
        # Synthesis is text-only -- no images in this pass
        synthesis_inputs = self.processor(
            text=synthesis_text_input,
            return_tensors="pt",
        ).to(self.model.device)

        with torch.no_grad():
            synthesis_ids = self.model.generate(
                **synthesis_inputs,
                max_new_tokens=512,
                do_sample=False,
            )

        synthesis_new = synthesis_ids[0][synthesis_inputs["input_ids"].shape[-1]:]
        synthesis_output = self.processor.decode(synthesis_new, skip_special_tokens=True).strip()

        action_items = parse_action_items(synthesis_output)
        key_moments  = parse_key_moments(synthesis_output)

        return VideoSummary(
            video_path=video_path,
            duration_seconds=duration,
            frames_analyzed=len(descriptions),
            frame_descriptions=descriptions,
            narrative_summary=synthesis_output,
            action_items=action_items,
            key_moments=key_moments,
        )

    def to_json(self, summary: VideoSummary) -> str:
        """Serialize a VideoSummary to formatted JSON."""
        return json.dumps({
            "video":            summary.video_path,
            "duration_seconds": summary.duration_seconds,
            "frames_analyzed":  summary.frames_analyzed,
            "narrative":        summary.narrative_summary,
            "action_items":     summary.action_items,
            "key_moments":      summary.key_moments,
            "frame_descriptions": [
                {
                    "timestamp":    d.timestamp,
                    "timestamp_label": f"{int(d.timestamp // 60):02d}:{int(d.timestamp % 60):02d}",
                    "description":  d.description,
                }
                for d in summary.frame_descriptions
            ],
        }, indent=2, ensure_ascii=False)

# ── Entry point ───────────────────────────────────────────────────────────────

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Summarize a video with SmolVLM2-2.2B")
    parser.add_argument("video",   help="Path to the input video file")
    parser.add_argument("--output", default="summary.json", help="Output JSON file path")
    parser.add_argument("--mode",   default="uniform", choices=["uniform", "keyframe"])
    parser.add_argument("--batch-size", type=int, default=8)
    args = parser.parse_args()

    summarizer = VideoSummarizer(batch_size=args.batch_size)
    result = summarizer.summarize(args.video, mode=args.mode)

    output_str = summarizer.to_json(result)
    with open(args.output, "w", encoding="utf-8") as f:
        f.write(output_str)

    print(f"\nSummary saved to {args.output}")
    print(f"Frames analyzed: {result.frames_analyzed}")
    print(f"Action items found: {len(result.action_items)}")
    for item in result.action_items:
        print(f"  - {item}")

# Uniform sampling (default) -- best for meetings and lectures
python video_summarizer.py meeting_2026_06_14.mp4 --output meeting_summary.json

# Keyframe sampling -- best for event detection, surveillance
python video_summarizer.py security_footage.mp4 --mode keyframe --output events.json

# Adjust batch size for your VRAM (8 for 8 GB VRAM, 16 for 16 GB)
python video_summarizer.py long_lecture.mp4 --batch-size 4 --output lecture.json

{
  "video": "meeting_2026_06_14.mp4",
  "duration_seconds": 3247.0,
  "frames_analyzed": 50,
  "narrative": "The meeting focused on Q3 product planning ...",
  "action_items": [
    "Finalize API design document by end of June",
    "Schedule testing sprint kickoff for July 1",
    "Share updated Gantt chart with stakeholders"
  ],
  "key_moments": [
    {"timestamp_label": "00:00", "description": "Team introductions and agenda overview"},
    {"timestamp_label": "12:30", "description": "API architecture diagram reviewed on screen"},
    {"timestamp_label": "41:15", "description": "Action items summarized on whiteboard"}
  ]
}

# vram_calculator.py
# Estimate a safe batch size for your GPU before running the pipeline

def compute_batch_size(vram_gb: float, tokens_per_frame: int = 81) -> int:
    """
    Estimate frames per inference batch for a given VRAM budget.

    Args:
        vram_gb:          Available GPU VRAM in gigabytes
        tokens_per_frame: Visual tokens per image (81 for SmolVLM2)

    Returns:
        Safe batch size, minimum 1, maximum 50
    """
    MODEL_GB     = 4.5      # SmolVLM2-2.2B weights in bfloat16
    HEADROOM     = 0.80     # Use at most 80% of total VRAM
    MB_PER_TOKEN = 0.5 / 1024  # GB per KV token at 2.2B scale (rough)

    usable_gb        = vram_gb * HEADROOM
    inference_budget = max(0.0, usable_gb - MODEL_GB)
    frames           = int(inference_budget / (tokens_per_frame * MB_PER_TOKEN))

    return max(1, min(frames, 50))

if __name__ == "__main__":
    for vram in [6.0, 8.0, 12.0, 16.0, 24.0]:
        print(f"  {vram:.0f} GB VRAM → batch_size = {compute_batch_size(vram)}")

  6 GB  VRAM → batch_size = 16
  8 GB  VRAM → batch_size = 30
 12 GB  VRAM → batch_size = 50
 16 GB  VRAM → batch_size = 50
 24 GB  VRAM → batch_size = 50

# jsonl_writer.py -- drop-in checkpoint support for long-video processing

import json

class JSONLWriter:
    """
    Writes frame descriptions to a JSONL file as they are produced.
    Enables resume-from-checkpoint on long videos -- if inference fails at
    frame 30 of 50, re-read the JSONL and skip already-processed frames.
    """
    def __init__(self, path: str):
        self.path = path
        self._fh  = open(path, "a", encoding="utf-8")  # Append mode for resume

    def write(self, record: dict):
        """Write one frame record and flush immediately to disk."""
        self._fh.write(json.dumps(record, ensure_ascii=False) + "\n")
        self._fh.flush()

    def already_processed(self) -> set[int]:
        """Return the set of frame indices already in the checkpoint file."""
        processed = set()
        try:
            with open(self.path, "r", encoding="utf-8") as f:
                for line in f:
                    record = json.loads(line)
                    processed.add(record.get("frame_index", -1))
        except FileNotFoundError:
            pass
        return processed

    def close(self):
        self._fh.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

def timestamp_label(seconds: float) -> str:
    """Convert decimal seconds to MM:SS or HH:MM:SS label."""
    total = int(seconds)
    h, remainder = divmod(total, 3600)
    m, s = divmod(remainder, 60)
    if h > 0:
        return f"{h:02d}:{m:02d}:{s:02d}"
    return f"{m:02d}:{s:02d}"