Native Video Summarization Pipeline: Processing Frames with SmolVLM2-2.2B

# Python 3.10+ required
python --version

python -m venv smolvlm2-env
supply smolvlm2-env/bin/activate       # macOS / Linux
smolvlm2-envScriptsactivate          # Home windows

# Set up from the secure SmolVLM-2 department -- required for SmolVLM2 help
pip set up git+https://github.com/huggingface/transformers@v4.49.0-SmolVLM-2

# Core dependencies
pip set up torch torchvision --index-url https://obtain.pytorch.org/whl/cu121
pip set up 
  opencv-python 
  Pillow 
  numpy 
  num2words 
  speed up

# Flash Consideration 2 for CUDA -- considerably sooner on NVIDIA GPUs
# Skip this on Apple Silicon and CPU -- it's CUDA-only
pip set up flash-attn --no-build-isolation

# decord -- required for SmolVLM2's native video enter path (utilized in Part 5)
pip set up decord

# device_check.py
# Run: python device_check.py

import torch

def detect_device():
    if torch.cuda.is_available():
        title  = torch.cuda.get_device_name(0)
        vram  = torch.cuda.get_device_properties(0).total_memory / 1e9
        print(f"CUDA: {title} ({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, "keen"
    else:
        print("CPU fallback (gradual -- contemplate Colab T4)")
        return "cpu", torch.float32, "keen"

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

# frame_extractor.py
# Conditions: pip set up opencv-python Pillow numpy
# Utilization: from frame_extractor import FrameExtractor

import cv2
import numpy as np
from PIL import Picture


class FrameExtractor:
    """
    Extracts video frames as PIL Pictures for SmolVLM2 inference.
    Every extracted body is paired with its timestamp in seconds.

    SmolVLM2 makes use of ~81 visible tokens per picture. At 50 frames that's
    roughly 4,050 picture tokens -- the sensible higher restrict earlier than VRAM
    stress impacts era high quality on shopper GPUs.
    """

    MAX_FRAMES = 50

    def __init__(self, max_frames: int = MAX_FRAMES):
        """
        Args:
            max_frames: Laborious cap on extracted frames. Default 50 matches
                        the SmolVLM2 reference pipeline's examined higher restrict.
        """
        self.max_frames = max_frames

    def uniform_sample(self, video_path: str) -> listing[tuple[float, Image.Image]]:
        """
        Extract evenly spaced frames throughout the total video period.
        Finest for: assembly recordings, lectures, tutorials, course content material.

        Returns:
            Listing of (timestamp_seconds, PIL_Image) in chronological order.
        """
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            increase IOError(f"Can't 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)

        # Construct body indices unfold evenly from first to final body
        indices = np.linspace(0, total_frames - 1, n_extract, dtype=int)
        outcomes = []

        for idx in indices:
            cap.set(cv2.CAP_PROP_POS_FRAMES, int(idx))
            ret, body = cap.learn()
            if not ret:
                proceed
            timestamp = spherical(idx / fps, 2)
            rgb = cv2.cvtColor(body, cv2.COLOR_BGR2RGB)
            outcomes.append((timestamp, Picture.fromarray(rgb)))

        cap.launch()
        return outcomes

    def keyframe_sample(
        self, video_path: str, diff_threshold: float = 30.0
    ) -> listing[tuple[float, Image.Image]]:
        """
        Extract frames the place visible content material adjustments considerably.
        Finest for: surveillance, occasion detection, spotlight extraction.

        Makes use of imply absolute pixel distinction between consecutive grayscale frames
        because the change sign. When the diff exceeds diff_threshold, a brand new
        keyframe is recorded.

        Args:
            diff_threshold: Imply pixel distinction to deal with as a scene change.
                            30.0 works for many industrial content material.
                            Decrease = extra delicate, larger = fewer frames.

        Returns:
            Listing of (timestamp_seconds, PIL_Image) in chronological order,
            capped at self.max_frames.
        """
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            increase IOError(f"Can't open video: {video_path}")

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

        whereas len(outcomes) < self.max_frames:
            ret, body = cap.learn()
            if not ret:
                break

            grey = cv2.cvtColor(body, cv2.COLOR_BGR2GRAY)

            if prev_gray is None:
                # All the time seize the primary body as a baseline
                rgb = cv2.cvtColor(body, cv2.COLOR_BGR2RGB)
                outcomes.append((spherical(idx / fps, 2), Picture.fromarray(rgb)))
            else:
                diff = np.imply(np.abs(grey.astype(float) - prev_gray.astype(float)))
                if diff > diff_threshold:
                    rgb = cv2.cvtColor(body, cv2.COLOR_BGR2RGB)
                    outcomes.append((spherical(idx / fps, 2), Picture.fromarray(rgb)))

            prev_gray = grey
            idx += 1

        cap.launch()
        return outcomes

# smolvlm2_loader.py
# Conditions: transformers from v4.49.0-SmolVLM-2 department, torch, flash-attn (CUDA solely)
# Run: python smolvlm2_loader.py your_video.mp4

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

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


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

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

    processor = AutoProcessor.from_pretrained(MODEL_ID)

    mannequin = AutoModelForImageTextToText.from_pretrained(
        MODEL_ID,
        torch_dtype=dtype,
        _attn_implementation=attn,
    ).to(machine)

    mannequin.eval()
    print(f"Mannequin prepared on {machine}")
    return mannequin, processor


def describe_frame(
    mannequin,
    processor,
    body: Picture.Picture,
    immediate: str = "Describe what is occurring on this body intimately. Be aware any textual content, folks, objects, or actions seen.",
    max_new_tokens: int = 256,
) -> str:
    """
    Run SmolVLM2 inference on a single PIL Picture.

    The chat template expects picture content material earlier than textual content content material within the
    message -- this mirrors the coaching knowledge format and is essential
    for dependable output.

    Args:
        body:          A PIL Picture (from FrameExtractor)
        immediate:         What to ask the mannequin about this body
        max_new_tokens: Most response size in tokens

    Returns:
        Mannequin 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 codecs the message and injects visible token placeholders
    input_text = processor.apply_chat_template(
        messages,
        add_generation_prompt=True,
    )

    inputs = processor(
        photographs=[frame],
        textual content=input_text,
        return_tensors="pt",
    ).to(mannequin.machine)

    with torch.no_grad():
        output_ids = mannequin.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            do_sample=False,   # Grasping decoding for constant structured output
        )

    # Decode solely the newly generated tokens -- strip the enter immediate
    new_tokens = output_ids[0][inputs["input_ids"].form[-1]:]
    return processor.decode(new_tokens, skip_special_tokens=True).strip()


# ── Fast sanity test ────────────────────────────────────────────────────────

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

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

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

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

    print(f"nDescribing body at {ts}s...")
    description = describe_frame(mannequin, processor, first_frame)
    print(f"n{description}")

python smolvlm2_loader.py your_video.mp4

# video_summarizer.py
# Conditions: frame_extractor.py and smolvlm2_loader.py in the identical listing
# Run: python video_summarizer.py meeting_recording.mp4 --output abstract.json

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

from frame_extractor import FrameExtractor
from smolvlm2_loader import load_model, describe_frame


# ── Knowledge fashions ───────────────────────────────────────────────────────────────

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

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


# ── Per-frame immediate ──────────────────────────────────────────────────────────

FRAME_PROMPT = """You're analyzing a body from a recorded assembly.

Describe what you see concisely however fully:
- Who or what's seen (folks, whiteboards, screens, slides)
- Any readable textual content (slide titles, whiteboard content material, display content material)
- The obvious exercise (presenting, discussing, writing, listening)

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

# ── Synthesis immediate ──────────────────────────────────────────────────────────

def build_synthesis_prompt(descriptions: listing[FrameDescription], period: float) -> str:
    """Construct the second-pass immediate that synthesizes body descriptions right into a report."""
    frames_text = "n".be part of(
        f"[{int(d.timestamp // 60):02d}:{int(d.timestamp % 60):02d}] {d.description}"
        for d in descriptions
    )
    return f"""Under are time-stamped descriptions of frames from a {period:.0f}-second assembly recording.

{frames_text}

Primarily based on these descriptions, present:

1. NARRATIVE SUMMARY: A 3-5 sentence abstract of what the assembly lined, who participated (if seen), and what selections or conclusions had been reached.

2. ACTION ITEMS: A bullet listing of concrete duties or follow-ups talked about or implied within the assembly. Begin every with a touch (-).

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

Format your response with clear headings for every part."""


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

def parse_action_items(textual content: str) -> listing[str]:
    """Extract bullet-point motion gadgets from the synthesis output."""
    gadgets = []
    for line in textual content.break up("n"):
        stripped = line.strip()
        if re.match(r"^[-*•]s+", stripped) or re.match(r"^d+.s+", stripped):
            clear = re.sub(r"^[-*•d.]+s*", "", stripped).strip()
            if clear and len(clear) > 5:
                gadgets.append(clear)
    return gadgets

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


# ── Fundamental summarizer class ─────────────────────────────────────────────────────

class VideoSummarizer:
    """
    Finish-to-end native video summarizer utilizing SmolVLM2-2.2B.
    Two-pass technique: per-frame descriptions + synthesis narrative.
    Works on scanned, digital, and live-recorded movies alike.
    """

    def __init__(self, batch_size: int = 8):
        """
        Args:
            batch_size: Frames to explain per inference batch.
                        Tune based mostly on VRAM: 8 for 8 GB, 16 for 16 GB.
                        Every body makes use of ~81 visible tokens; decrease batch = much less peak VRAM.
        """
        self.mannequin, 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.launch()
        return spherical(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 protection, "keyframe" for scene adjustments

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

        # ── Cross 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}...")

        # ── Cross 2: Describe every body ────────────────────────────────────
        descriptions: listing[FrameDescription] = []

        for batch_start in vary(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 body at {timestamp}s...")

                desc = describe_frame(
                    self.mannequin,
                    self.processor,
                    img,
                    immediate=FRAME_PROMPT,
                    max_new_tokens=128,   # Preserve body descriptions concise
                )
                descriptions.append(FrameDescription(
                    timestamp=timestamp,
                    frame_index=global_idx,
                    description=desc,
                ))

        # ── Cross 3: Synthesis ──────────────────────────────────────────────
        print("nRunning synthesis go...")

        synthesis_prompt = build_synthesis_prompt(descriptions, period)
        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 photographs on this go
        synthesis_inputs = self.processor(
            textual content=synthesis_text_input,
            return_tensors="pt",
        ).to(self.mannequin.machine)

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

        synthesis_new = synthesis_ids[0][synthesis_inputs["input_ids"].form[-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=period,
            frames_analyzed=len(descriptions),
            frame_descriptions=descriptions,
            narrative_summary=synthesis_output,
            action_items=action_items,
            key_moments=key_moments,
        )

    def to_json(self, abstract: VideoSummary) -> str:
        """Serialize a VideoSummary to formatted JSON."""
        return json.dumps({
            "video":            abstract.video_path,
            "duration_seconds": abstract.duration_seconds,
            "frames_analyzed":  abstract.frames_analyzed,
            "narrative":        abstract.narrative_summary,
            "action_items":     abstract.action_items,
            "key_moments":      abstract.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 level ───────────────────────────────────────────────────────────────

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

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

    output_str = summarizer.to_json(outcome)
    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: {outcome.frames_analyzed}")
    print(f"Motion gadgets discovered: {len(outcome.action_items)}")
    for merchandise in outcome.action_items:
        print(f"  - {merchandise}")

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

# Keyframe sampling -- greatest for occasion detection, surveillance
python video_summarizer.py security_footage.mp4 --mode keyframe --output occasions.json

# Modify batch dimension to 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 assembly centered 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 secure batch dimension to your GPU earlier than working the pipeline

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

    Args:
        vram_gb:          Obtainable GPU VRAM in gigabytes
        tokens_per_frame: Visible tokens per picture (81 for SmolVLM2)

    Returns:
        Secure batch dimension, minimal 1, most 50
    """
    MODEL_GB     = 4.5      # SmolVLM2-2.2B weights in bfloat16
    HEADROOM     = 0.80     # Use at most 80% of complete VRAM
    MB_PER_TOKEN = 0.5 / 1024  # GB per KV token at 2.2B scale (tough)

    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 help for long-video processing

import json

class JSONLWriter:
    """
    Writes body descriptions to a JSONL file as they're produced.
    Permits resume-from-checkpoint on lengthy movies -- if inference fails at
    body 30 of fifty, 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, file: dict):
        """Write one body file and flush instantly to disk."""
        self._fh.write(json.dumps(file, ensure_ascii=False) + "n")
        self._fh.flush()

    def already_processed(self) -> set[int]:
        """Return the set of body indices already within the checkpoint file."""
        processed = set()
        attempt:
            with open(self.path, "r", encoding="utf-8") as f:
                for line in f:
                    file = json.masses(line)
                    processed.add(file.get("frame_index", -1))
        besides FileNotFoundError:
            go
        return processed

    def shut(self):
        self._fh.shut()

    def __enter__(self):
        return self

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

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