Video Understanding

import cv2
import numpy as np
from typing import Literal

def extract_frames(
    video_path: str,
    strategy: Literal["uniform", "keyframe", "scene"] = "uniform",
    target_frames: int = 16,
    scene_threshold: float = 30.0,
) -> list[tuple[np.ndarray, float]]:
    """Extract frames from video with timestamp.

    Returns list of (frame, timestamp_seconds) tuples.
    """
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    duration = total_frames / fps

    if strategy == "uniform":
        # Sample evenly across the video
        indices = np.linspace(0, total_frames - 1, target_frames, dtype=int)
        frames = []
        for idx in indices:
            cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
            ret, frame = cap.read()
            if ret:
                frames.append((frame, idx / fps))
        cap.release()
        return frames

    elif strategy == "keyframe":
        # Extract frames with significant visual change
        frames = []
        prev_hist = None
        frame_idx = 0

        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            # Compute color histogram
            hist = cv2.calcHist([frame], [0, 1, 2], None,
                               [8, 8, 8], [0, 256, 0, 256, 0, 256])
            hist = cv2.normalize(hist, hist).flatten()

            if prev_hist is None:
                frames.append((frame, frame_idx / fps))
            else:
                diff = cv2.compareHist(prev_hist, hist, cv2.HISTCMP_CHISQR)
                if diff > scene_threshold:
                    frames.append((frame, frame_idx / fps))
            prev_hist = hist
            frame_idx += 1

        cap.release()
        # If too many, subsample uniformly
        if len(frames) > target_frames:
            indices = np.linspace(0, len(frames) - 1, target_frames, dtype=int)
            frames = [frames[i] for i in indices]
        return frames

    elif strategy == "scene":
        # Use ffmpeg scene detection (shell out)
        import subprocess, json
        cmd = [
            "ffprobe", "-v", "quiet", "-select_streams", "v",
            "-show_frames", "-show_entries", "frame=pts_time,pict_type",
            "-of", "json", video_path
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)
        scene_data = json.loads(result.stdout)
        # Filter I-frames (scene changes)
        i_frames = [
            float(f["pts_time"]) for f in scene_data.get("frames", [])
            if f.get("pict_type") == "I"
        ]
        # Sample from scene boundaries
        timestamps = i_frames[:target_frames] if len(i_frames) <= target_frames \
            else [i_frames[i] for i in np.linspace(0, len(i_frames)-1, target_frames, dtype=int)]

        frames = []
        for ts in timestamps:
            cap.set(cv2.CAP_PROP_POS_MSEC, ts * 1000)
            ret, frame = cap.read()
            if ret:
                frames.append((frame, ts))
        cap.release()
        return frames

import base64
from openai import OpenAI

def encode_frame(frame: np.ndarray) -> str:
    """Encode frame as base64 JPEG for API consumption."""
    _, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
    return base64.b64encode(buffer).decode('utf-8')

def analyze_video_gpt4o(
    frames: list[tuple[np.ndarray, float]],
    question: str,
    model: str = "gpt-4o",
) -> str:
    """Analyze video frames with GPT-4o.

    Includes timestamps in the prompt so the model can
    reference specific moments in its response.
    """
    client = OpenAI()

    # Build the message content with timestamped frames
    content = [
        {
            "type": "text",
            "text": (
                f"You are analyzing a video. I'm providing {len(frames)} frames "
                f"sampled from the video, each labeled with its timestamp.\n\n"
                f"Question: {question}"
            ),
        }
    ]

    for frame, timestamp in frames:
        # Add timestamp label before each frame
        minutes = int(timestamp // 60)
        seconds = timestamp % 60
        content.append({
            "type": "text",
            "text": f"[{minutes}:{seconds:05.2f}]"
        })
        content.append({
            "type": "image_url",
            "image_url": {
                "url": f"data:image/jpeg;base64,{encode_frame(frame)}",
                "detail": "low"  # Use "high" for fine detail, costs 4x more
            }
        })

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": content}],
        max_tokens=2000,
    )
    return response.choices[0].message.content

# Usage
frames = extract_frames("lecture.mp4", strategy="uniform", target_frames=16)
summary = analyze_video_gpt4o(
    frames,
    "Summarize the key points discussed in this presentation. "
    "Reference timestamps when the topic changes."
)

import google.generativeai as genai
import time

genai.configure(api_key="YOUR_API_KEY")

def analyze_video_gemini(video_path: str, question: str) -> str:
    """Analyze video natively with Gemini.

    Gemini handles frame sampling internally at ~1 fps.
    Supports up to 1 hour of video in Gemini 2.5 Pro.
    """
    model = genai.GenerativeModel("gemini-2.5-pro")

    # Upload video file (supports mp4, mov, avi, mkv, webm)
    video_file = genai.upload_file(video_path)

    # Wait for server-side processing (transcoding + frame extraction)
    while video_file.state.name == "PROCESSING":
        time.sleep(5)
        video_file = genai.get_file(video_file.name)

    if video_file.state.name == "FAILED":
        raise ValueError(f"Video processing failed: {video_file.state.name}")

    # Analyze — Gemini sees frames + audio natively
    response = model.generate_content(
        [video_file, question],
        generation_config=genai.GenerationConfig(
            temperature=0.2,
            max_output_tokens=4000,
        ),
    )
    return response.text

# Usage — no frame extraction needed
result = analyze_video_gemini(
    "meeting_recording.mp4",
    "List all action items discussed in this meeting with timestamps."
)

import whisper
import subprocess
from dataclasses import dataclass

@dataclass
class VideoAnalysis:
    transcript: str
    visual_description: str
    combined_summary: str
    timestamps: list[dict]

def extract_audio(video_path: str) -> str:
    """Extract audio track from video using ffmpeg."""
    audio_path = video_path.rsplit('.', 1)[0] + '.wav'
    subprocess.run([
        'ffmpeg', '-y', '-i', video_path,
        '-vn', '-acodec', 'pcm_s16le',
        '-ar', '16000', '-ac', '1',
        audio_path
    ], check=True, capture_output=True)
    return audio_path

def transcribe_with_timestamps(audio_path: str) -> dict:
    """Transcribe audio with word-level timestamps."""
    model = whisper.load_model("base")
    result = model.transcribe(
        audio_path,
        word_timestamps=True,
        language="en",
    )
    return result

def full_video_analysis(video_path: str, question: str) -> VideoAnalysis:
    """Combine visual and audio understanding.

    Pipeline:
    1. Extract frames (uniform, 1 fps)
    2. Extract + transcribe audio (Whisper)
    3. Analyze frames with GPT-4o (visual)
    4. Synthesize visual + audio with final LLM call
    """
    # Visual: extract and analyze frames
    frames = extract_frames(video_path, strategy="uniform", target_frames=16)
    visual_desc = analyze_video_gpt4o(
        frames,
        "Describe what you see in each frame. Note any text, "
        "people, actions, objects, and scene changes."
    )

    # Audio: extract and transcribe
    audio_path = extract_audio(video_path)
    transcript_result = transcribe_with_timestamps(audio_path)
    transcript = transcript_result["text"]

    # Collect segment timestamps
    timestamps = [
        {"start": seg["start"], "end": seg["end"], "text": seg["text"]}
        for seg in transcript_result.get("segments", [])
    ]

    # Synthesis: combine both modalities
    client = OpenAI()
    synthesis = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a video analyst. You receive a visual description "
                    "of video frames and an audio transcript. Synthesize both "
                    "into a coherent analysis. Reference specific timestamps."
                ),
            },
            {
                "role": "user",
                "content": (
                    f"VISUAL DESCRIPTION:\n{visual_desc}\n\n"
                    f"AUDIO TRANSCRIPT:\n{transcript}\n\n"
                    f"QUESTION: {question}"
                ),
            },
        ],
    )

    return VideoAnalysis(
        transcript=transcript,
        visual_description=visual_desc,
        combined_summary=synthesis.choices[0].message.content,
        timestamps=timestamps,
    )

# Usage
analysis = full_video_analysis(
    "product_demo.mp4",
    "What features are demonstrated and what claims are made about each?"
)
print(analysis.combined_summary)

from twelvelabs import TwelveLabs

client = TwelveLabs(api_key="YOUR_API_KEY")

# Create an index (a searchable video collection)
index = client.index.create(
    name="product_demos",
    engines=[{
        "name": "marengo2.7",  # Video understanding engine
        "options": ["visual", "conversation", "text_in_video", "logo"],
    }],
)

# Upload videos to the index
task = client.task.create(
    index_id=index.id,
    file="demo_video.mp4",
)
task.wait_for_done()  # Processing: ~1 min per 1 min of video

# Natural language search — returns timestamps
results = client.search.query(
    index_id=index.id,
    query_text="moment where the speaker demonstrates the API",
    options=["visual", "conversation"],
)

for clip in results.data:
    print(f"[{clip.start:.1f}s - {clip.end:.1f}s] "
          f"score={clip.score:.3f} | {clip.video_id}")

# Generate text from a specific segment
summary = client.generate.text(
    video_id=task.video_id,
    prompt="Summarize the key features shown in this segment",
    temperature=0.2,
)

# Build your own video search with CLIP + vector DB
import torch
from transformers import CLIPModel, CLIPProcessor
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct

# Initialize CLIP
model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")

# Initialize vector database
qdrant = QdrantClient(":memory:")  # or url="http://localhost:6333"
qdrant.create_collection(
    collection_name="video_frames",
    vectors_config=VectorParams(size=768, distance=Distance.COSINE),
)

def index_video(video_path: str, video_id: str):
    """Index a video by embedding frames into Qdrant."""
    frames = extract_frames(video_path, strategy="uniform", target_frames=60)

    points = []
    for i, (frame, timestamp) in enumerate(frames):
        # Convert BGR (OpenCV) to RGB (PIL)
        rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        inputs = processor(images=rgb_frame, return_tensors="pt")

        with torch.no_grad():
            embedding = model.get_image_features(**inputs)
            embedding = embedding / embedding.norm(dim=-1, keepdim=True)

        points.append(PointStruct(
            id=hash(f"{video_id}_{i}"),
            vector=embedding[0].numpy().tolist(),
            payload={"video_id": video_id, "timestamp": timestamp, "frame_idx": i},
        ))

    qdrant.upsert(collection_name="video_frames", points=points)

def search_video(query: str, top_k: int = 5):
    """Search indexed videos with natural language."""
    inputs = processor(text=query, return_tensors="pt", padding=True)
    with torch.no_grad():
        text_embedding = model.get_text_features(**inputs)
        text_embedding = text_embedding / text_embedding.norm(dim=-1, keepdim=True)

    results = qdrant.search(
        collection_name="video_frames",
        query_vector=text_embedding[0].numpy().tolist(),
        limit=top_k,
    )
    return [
        {"video_id": r.payload["video_id"],
         "timestamp": r.payload["timestamp"],
         "score": r.score}
        for r in results
    ]

# Index and search
index_video("keynote.mp4", "keynote_2024")
hits = search_video("slide showing revenue growth chart")
# Returns: [{"video_id": "keynote_2024", "timestamp": 847.3, "score": 0.31}, ...]