1
from typing import List
2
3
from twelvelabs import TwelveLabs
4
from twelvelabs.types import VideoSegment
5
from twelvelabs.embed import TasksStatusResponse
6
7
client = TwelveLabs(api_key="<YOUR_API_KEY>")
8
9
task = client.embed.tasks.create(
10
    model_name="marengo3.0", video_url="<YOUR_VIDEO_URL>")
11
print(f"Created video embedding task: id={task.id}") # Or use video_file to upload a file from the local file system
12
13
14
def on_task_update(task: TasksStatusResponse):
15
    print(f"  Status={task.status}")
16
17
18
status = client.embed.tasks.wait_for_done(task.id, callback=on_task_update)
19
print(f"Embedding done: {status.status}")
20
21
task = client.embed.tasks.retrieve(
22
    task_id=task.id, embedding_option=["visual", "audio", "transcription"])
23
24
25
def print_segments(segments: List[VideoSegment], max_elements: int = 5):
26
    for segment in segments:
27
        print(f"  embedding_scope={segment.embedding_scope} embedding_option={segment.embedding_option} start_offset_sec={segment.start_offset_sec} end_offset_sec={segment.end_offset_sec}")
28
        first_few = segment.float_[:max_elements]
29
        print(
30
            f"  embeddings: [{', '.join(str(x) for x in first_few)}...] (total: {len(segment.float_)} values)"
31
        )
32
33
34
if task.video_embedding is not None and task.video_embedding.segments is not None:
35
    print_segments(task.video_embedding.segments)