Download notebook · View source · Open in Colab

Video Frame Search

Video frame search compares a text embedding with embeddings of individual frames. This example stores a cooking video in ApertureDB, extracts frames at regular intervals, embeds them with CLIP, and stores the frame vectors with their positions in the video.

Connect to ApertureDB

ApertureDB Cloud The SDK connects with an API key from your Cloud instance. Obtain the key from Connect > Generate API Key and add it to a .env file in this directory:

APERTUREDB_KEY=your_key_here

Local evaluation: follow the deployment guide, then configure your connection profile.

See client configuration options for all connection methods and server setup options for deployment choices.

%pip install --quiet "aperturedb==0.5.1" "python-dotenv==1.1.1" sentence-transformers Pillow requests

from dotenv import load_dotenv
load_dotenv()

# For local evaluation, configure the profile from your versioned deployment package.
# https://docs.aperturedata.io/administration/deployment/DockerCompose

from aperturedb.CommonLibrary import create_connector

client = create_connector()
response, _ = client.query([{"GetStatus": {}}])
client.print_last_response()

Step: Load the CLIP model

The same CLIP model encodes the extracted frames and the search text.

from sentence_transformers import SentenceTransformer
model = SentenceTransformer("clip-ViT-B-32")
print(f"Embedding dimensions: {model.get_sentence_embedding_dimension()}")

Step: Add a video to ApertureDB

AddVideo stores the source video so extracted frame descriptors can refer to it. A unique run marker scopes frame extraction and cleanup to this example's video.

import requests
from uuid import uuid4

run_id = "video-search-" + uuid4().hex
scope = ["$documentation_run", "==", run_id]

VIDEO_URL = "https://raw.githubusercontent.com/aperture-data/Cookbook/51b3c3d9f495f8fd77108a270b88fe8ab8cc0004/notebooks/simple/data/crepe_flambe.mp4"

resp = requests.get(VIDEO_URL, timeout=30)
resp.raise_for_status()
video_bytes = resp.content
print(f"Downloaded {len(video_bytes):,} bytes")

q = [{
    "AddVideo": {
        "_ref": 1,
        "properties": {
            "documentation_run": run_id,
            "name":    "crepe_flambe",
            "cuisine": "French",
        }
    }
}]
response, _ = client.query(q, [video_bytes])
client.print_last_response()

Step: Extract frames from the video

The example extracts frames at regular intervals. Their frame numbers retain each sample's position in the source video.

# Extract frames at positions 0, 30, 60, 90, 120
FRAME_NUMBERS = [0, 30, 60, 90, 120]

q = [{
    "FindVideo": {
        "constraints": scope,
        "_ref": 1,
        "blobs": False,
    }
}, {
    "ExtractFrames": {
        "video_ref": 1,
        "frame_spec": FRAME_NUMBERS,
    }
}]

response, frame_blobs = client.query(q)
n_frames = response[1]["ExtractFrames"].get("returned", 0)
print(f"Extracted {n_frames} frames")

Step: Create a DescriptorSet for frame embeddings

The descriptor set stores the CLIP vectors with the dimensions and metric expected by the model.

import uuid
SET_NAME = "video_frame_search_" + uuid.uuid4().hex[:12]

client.query([{
    "AddDescriptorSet": {
        "name":       SET_NAME,
        "dimensions": 512,
        "engine":     "Flat",
        "metric":     "CS",
    }
}])
client.print_last_response()

Step: Embed and store each frame

Each frame is decoded as a PIL image and embedded with CLIP. Its Descriptor connects to the video and stores the frame number as a property.

from PIL import Image
from io import BytesIO
import numpy as np

for i, (frame_blob, frame_num) in enumerate(zip(frame_blobs, FRAME_NUMBERS)):
    img = Image.open(BytesIO(frame_blob)).convert("RGB")
    emb = model.encode(img, normalize_embeddings=True).astype("float32")

    q = [{
        "FindVideo": {
            "constraints": scope,
            "_ref": 1,
            "blobs": False,
        }
    }, {
        "AddDescriptor": {
            "set":   SET_NAME,
            "connect": {"ref": 1, "class": "has_frame_embedding"},
            "properties": {
                "frame_number": frame_num,
                "video_name":   "crepe_flambe",
            }
        }
    }]
    client.query(q, [emb.tobytes()])

print(f"Stored embeddings for {len(frame_blobs)} frames")

Step: Text-to-frame search

The text query is encoded in the same space as the frame embeddings. The matched descriptor's frame number identifies its position in the source video.

query_text = "pouring batter into pan"

query_emb = model.encode(query_text, normalize_embeddings=True).astype("float32")

q = [{
    "FindDescriptor": {
        "set":         SET_NAME,
        "k_neighbors": 3,
        "distances":   True,
        "results":     {"all_properties": True},
    }
}]

response, _ = client.query(q, [query_emb.tobytes()])

print(f'Query: "{query_text}"\n')
for entity in response[0]["FindDescriptor"].get("entities", []):
    print(f"  Frame {entity['frame_number']:>4}   distance={entity['_distance']:.4f}")

Step: Cleanup (optional)

Remove this run's descriptor set and source video after the search.

client.query([{"DeleteDescriptorSet": {"with_name": SET_NAME}}])
client.print_last_response()

client.query([{"DeleteVideo": {"constraints": scope}}])
client.print_last_response()