1
INDEX_NAME = ctx.params.get("index_name")
2
3
INDEXES_URL = f"{API_URL}/indexes"
4
5
headers = {
6
    "x-api-key": API_KEY
7
}
8
9
so = []
10
11
if ctx.params.get("visual"):
12
    so.append("visual")
13
if ctx.params.get("logo"):
14
    so.append("logo")
15
if ctx.params.get("text_in_video"):
16
    so.append("text_in_video")
17
if ctx.params.get("conversation"):
18
    so.append("conversation")
19
20
data = {
21
"engine_id": "marengo2.7",
22
"index_options": so,
23
"index_name": INDEX_NAME,
24
}
25
26
response = requests.post(INDEXES_URL, headers=headers, json=data)

1
TASKS_URL = f"{API_URL}/tasks"
2
3
videos = target_view
4
for sample in videos:
5
    if sample.metadata.duration < 4:
6
        continue
7
    else:
8
        file_name = sample.filepath.split("/")[-1] 
9
        file_path = sample.filepath 
10
        file_stream = open(file_path,"rb")
11
    
12
        headers = {
13
            "x-api-key": API_KEY
14
        }
15
    
16
        data = {
17
            "index_id": INDEX_ID, 
18
            "language": "en"
19
        }
20
    
21
        file_param=[
22
            ("video_file", (file_name, file_stream, "application/octet-stream")),]
23
    
24
        response = requests.post(TASKS_URL, headers=headers, data=data, files=file_param)
25
        TASK_ID = response.json().get("_id")
26
        print (f"Status code: {response.status_code}")
27
        pprint (response.json())
28
    
29
        TASK_STATUS_URL = f"{API_URL}/tasks/{TASK_ID}"
30
        while True:
31
            response = requests.get(TASK_STATUS_URL, headers=headers)
32
            STATUS = response.json().get("status")
33
            if STATUS == "ready":
34
                break
35
            time.sleep(10)
36
        
37
        VIDEO_ID = response.json().get('video_id')
38
        sample["TwelveLabs " + INDEX_NAME] = VIDEO_ID
39
        sample.save()

1
SEARCH_URL = f"{API_URL}/search"
2
3
headers = {
4
"x-api-key": API_KEY
5
}
6
7
so = []
8
9
if ctx.params.get("visual"):
10
    so.append("visual")
11
if ctx.params.get("logo"):
12
    so.append("logo")
13
if ctx.params.get("text_in_video"):
14
    so.append("text_in_video")
15
if ctx.params.get("conversation"):
16
    so.append("conversation")
17
18
data = {
19
"query": prompt,
20
"index_id": INDEX_ID,
21
"search_options": so,
22
}
23
24
response = requests.post(SEARCH_URL, headers=headers, json=data)
25
video_ids = [entry['video_id'] for entry in response.json()['data']]
26
print(response.json())
27
samples = []
28
view1 = target_view.select_by("TwelveLabs " + INDEX_NAME, video_ids,ordered=True)
29
start = [entry['start'] for entry in response.json()['data']]
30
end = [entry['end'] for entry in response.json()['data']]
31
if "results" in ctx.dataset.get_field_schema().keys():
32
    ctx.dataset.delete_sample_field("results")
33
34
i=0
35
for sample in view1:
36
    support = [int(start[i]*sample.metadata.frame_rate)+1 ,int(end[i]*sample.metadata.frame_rate)+1]
37
    sample["results"] = fo.TemporalDetection(label=prompt, support=tuple(support))
38
    sample.save()
39
40
view2 = view1.to_clips("results")
41
ctx.trigger("set_view", {"view": view2._serialize()})
42
43
return {}