Use Case: Conference Slide Extraction

!pip install videodb

import videodb

# Set your API key
api_key = "your_api_key"

# Connect to VideoDB
conn = videodb.connect(api_key=api_key)
coll = conn.get_collection()

# Upload a video by URL
video = coll.upload(url="https://www.youtube.com/watch?v=IEe-5VOv0Js")

import re

transcript_understanding = video.understand(
    analyzers=[{"type": "spoken_words", "name": "transcript"}],
)
transcript_understanding.wait_until_complete()
transcript_analyzer = transcript_understanding.get_analyzer("transcript")
transcript_output = transcript_analyzer.get_output()
sentence_records = []
sentence_words = []
for scene in transcript_output["scenes"]:
    for word in scene["data"]["words"]:
        sentence_words.append(word)
        if re.search(r"""[.!?]+["'”’)\]]*$""", word["text"]):
            sentence_records.append(
                {
                    "start": sentence_words[0]["start"],
                    "end": sentence_words[-1]["end"],
                    "text": " ".join(item["text"] for item in sentence_words),
                }
            )
            sentence_words = []
if sentence_words:
    sentence_records.append(
        {
            "start": sentence_words[0]["start"],
            "end": sentence_words[-1]["end"],
            "text": " ".join(item["text"] for item in sentence_words),
        }
    )
transcript_index = video.index(
    name="conference_transcript_sentences",
    source=sentence_records,
    use_for=["query"],
    fields={"filter": ["text"]},
)
transcript_index.wait_until_complete()

print(f"✅ Transcript index is ready with {len(sentence_records)} timed sentences!")

from IPython.display import Image, display
import requests


# Helper function to preview extracted scenes
def display_scenes(scenes, images=True):
    for scene in scenes:
        print(f"{scene.id} : {scene.start}-{scene.end}")
        if images:
            for frame in scene.frames:
                image = Image(requests.get(frame.url, stream=True).content)
                display(image)
        print("----")


scene_collection_default = video.extract_scenes()
display_scenes(scene_collection_default.scenes)

from videodb import SceneExtractionType

scene_collection = video.extract_scenes(
    extraction_type=SceneExtractionType.shot_based,
    extraction_config={"threshold": 10},
)
display_scenes(scene_collection.scenes)

slide_prompt = (
    "Give the content written on the slides, output None if it isn't the slides."
)

for scene in scene_collection.scenes[20:23]:
    description = scene.describe(slide_prompt)
    print(f"{scene.id} : {scene.start}-{scene.end}")
    print(description)
    print("-----")

def display_scene_index(scene_index):
    for scene in scene_index:
        print(f"{scene['start']} - {scene['end']}")
        print(scene["description"])
        print("----")


slide_prompt = (
    "Give the content written on the slides, output None if it isn't the slides."
)
slide_segmentation = {"type": "shot", "threshold": 10}
slide_sampling = {"strategy": "uniform", "frame_count": 1}
slide_schema = {"description": "string"}

slide_understanding = video.understand(
    segmentation=slide_segmentation,
    analyzers=[
        {
            "type": "vlm",
            "name": "slides",
            "sampling": slide_sampling,
            "config": {"prompt": slide_prompt, "schema": slide_schema},
        }
    ],
)
slide_understanding.wait_until_complete()

slides_analyzer = slide_understanding.get_analyzer("slides")
slides_output = slides_analyzer.get_output()

scene_index = [
    {
        "start": float(segment["start"]),
        "end": float(segment["end"]),
        "description": segment["data"]["description"],
    }
    for segment in slides_output["scenes"]
]
scene_index.sort(key=lambda scene: (scene["start"], scene["end"]))
display_scene_index(scene_index)

def simple_filter_scenes(time_ranges, scene_dicts):
    def is_in_range(scene, range_start, range_end):
        scene_start = scene["start"]
        scene_end = scene["end"]
        return scene_start < range_end and range_start < scene_end

    filtered_scenes = []
    for start, end in time_ranges:
        filtered_scenes.extend(
            scene
            for scene in scene_dicts
            if is_in_range(scene, start, end)
        )

    # Remove duplicates while preserving order
    seen = set()
    return [
        scene
        for scene in filtered_scenes
        if not (
            (scene["start"], scene["end"]) in seen
            or seen.add((scene["start"], scene["end"]))
        )
    ]

def search_pipeline(query, video):
    search_result = video.query(
        index_id=transcript_index.index_id,
        filter=[{"field": "text", "op": "contains", "value": query}],
        limit=100,
        sort=[("start", "asc")],
    )
    time_ranges = [(shot.start, shot.end) for shot in search_result.get_shots()]
    final_result = simple_filter_scenes(time_ranges, scene_index)
    result_text = "\n\n".join(item["description"] for item in final_result
                               if item.get("description", "").lower().strip() != "none")
    result_timeline = [(item.get("start"), item.get("end")) for item in final_result]
    return result_text, result_timeline

from videodb import play_stream

query = "hard and fast rule"

result_text, result_timeline = search_pipeline(query, video)

# Truncate result_text before printing if it's too long
max_chars = 500
if len(result_text) > max_chars:
    print(result_text[:max_chars] + "...\n<TRUNCATED>")
else:
    print(result_text)

# Get the video's actual length from the video object
video_length = video.length

# Clip the end times in result_timeline to not exceed video_length
clipped_result_timeline = [
    (start, min(end, video_length)) for start, end in result_timeline
]

stream_link = video.generate_stream(clipped_result_timeline)
play_stream(stream_link)

IT'S ALL IN THE DETAILS

- Prefer American English for naming
- Avoid payment-industry jargon
- Timestamp fields should use <verbed_at>
- Amount properties should also provide a currency
- API resources with IDs are top-level
- New API resources should be retrieved and listed one way
- API resource mutations should be reflected in API responses
- Use nested structures for future extensibility
- Prefer enums to booleans for new properties
- Use a type field for polymorphic objects
- Use verbs for properties with side effects
- Use top-level namespaces for product APIs
- Evaluate new features in the Dashboard before building an API
- Use simple, unambiguous language
- Always paginate unbounded lists
- Iterate on designs with beta users with the feature behind a gate

API Review: [Insert Title Here]
Ziec: Gavel jar, link for API review join creation

Gavel block
To ping PM when Q/A and other stakeholders have

Summary
(Please include a short description of the change you would like to make...)

https://go/terminal-dogfooding-instructions

Stripe! Thanks a ton for your help in dogfooding ahead of our Terminal GA launch!

We’re very close to launching Terminal in public beta and then GA.

And we could use your help! There are a ton of different use cases of these integrations to test and polish, and we want to stress test what we've made and ship these paths and accompanying docs and dashboard flows in a developer-friendly as possible in the time we have before the Terminal GA launch.

If you’re arriving at this doc after having signed up to dogfood, continue to Steps below.
If you haven’t signed up yet, please signup here, and we’ll get back to you when you have a slot your test.

Steps
1. If you’re dogfooding remotely and haven’t received instructions on ordering hardware or attending demos, please email terminal-dogfooding@stripe.com or ping in #terminal-dogfooding

2. If you're dogfooding the iOS or Android SDK, you'll need to set up your environment as described in the links...