Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions examples/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,3 +10,4 @@ For more information and examples, visit our [documentation](https://supervision
- [speed estimation](./speed_estimation) by [@SkalskiP](https://github.com/SkalskiP)
- [time in zone](./time_in_zone) by [@SkalskiP](https://github.com/SkalskiP)
- [heatmap and track](./heatmap_and_track/) by [@HinePo](https://github.com/HinePo)
- [twelvelabs video understanding](./twelvelabs_video_understanding/) by [@mohit-twelvelabs](https://github.com/mohit-twelvelabs)
9 changes: 9 additions & 0 deletions examples/twelvelabs_video_understanding/.gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
data/
venv*/
*.pt
*.pth
*.mp4
*.mov
*.png
*.jpg
*.jpeg
71 changes: 71 additions & 0 deletions examples/twelvelabs_video_understanding/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
# twelvelabs video understanding

## 👋 hello

This script pairs per-frame Supervision detections with a whole-video understanding generated by [TwelveLabs](https://twelvelabs.io) Pegasus.

Supervision answers "what object is where, in each frame" with precise, box-level detections. Pegasus answers "what is this video about" with a natural-language summary of the whole clip. Running both on the same video combines fine-grained detections with a high-level narrative — useful for captioning, search, moderation, or building a quick description alongside your annotated output.

## 💻 install

- clone repository and navigate to example directory

```bash
git clone --depth 1 -b develop https://github.com/roboflow/supervision.git
cd supervision/examples/twelvelabs_video_understanding
```

- setup python environment and activate it [optional]

```bash
uv venv
source .venv/bin/activate
```

- install required dependencies

```bash
uv pip install -r requirements.txt
```

- set your TwelveLabs API key (grab a free one at [twelvelabs.io](https://twelvelabs.io))

```bash
export TWELVELABS_API_KEY="<YOUR_API_KEY>"
```

## 🛠️ script arguments

- `--source_video_path`: Required. Path to the source video file to analyze.

- `--prompt` (optional): Instruction passed to Pegasus for the video-level summary. Default is `"Summarize what happens in this video."`.

- `--model_id` (optional): Ultralytics YOLO model id or weights path used for per-frame detection. Default is `"yolov8n.pt"`.

- `--confidence_threshold` (optional): Confidence threshold for detections (0 to 1). Default is `0.3`.

- `--pegasus_model_name` (optional): TwelveLabs Pegasus model to use. Default is `"pegasus1.5"`.

- `--max_tokens` (optional): Maximum number of tokens in the Pegasus response. Default is `512`.

- `--target_video_path` (optional): Path to save the annotated video. If not specified, annotated frames are displayed in real time.

- `--api_key` (optional): TwelveLabs API key. Falls back to the `TWELVELABS_API_KEY` environment variable when omitted.

## ⚙️ run

```bash
python twelvelabs_example.py \
--source_video_path data/video.mp4 \
--prompt "Summarize what happens in this video." \
--confidence_threshold 0.3 \
--target_video_path data/video_result.mp4
```

## © license

This demo integrates two main components, each with its own licensing:

- TwelveLabs: Video understanding is provided by the [TwelveLabs](https://twelvelabs.io) Pegasus model through the `twelvelabs` Python SDK and requires a TwelveLabs API key.

- supervision: The detection and annotation code in this demo is based on the Supervision library, which is licensed under the [MIT license](https://github.com/roboflow/supervision/blob/develop/LICENSE.md).
4 changes: 4 additions & 0 deletions examples/twelvelabs_video_understanding/requirements.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
supervision
ultralytics
twelvelabs
jsonargparse[signatures]
135 changes: 135 additions & 0 deletions examples/twelvelabs_video_understanding/twelvelabs_example.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,135 @@
from __future__ import annotations

import os

import cv2
import numpy as np

import supervision as sv

COLORS = sv.ColorPalette.DEFAULT
BOX_ANNOTATOR = sv.BoxAnnotator(color=COLORS)
LABEL_ANNOTATOR = sv.LabelAnnotator(color=COLORS)


def analyze_video(
source_video_path: str,
prompt: str,
model_name: str = "pegasus1.5",
max_tokens: int = 512,
api_key: str = "",
) -> str:
"""
Generate a natural-language understanding of a whole video with TwelveLabs
Pegasus.

This example base64-encodes the entire video file in memory before sending it to
the API, so keep clips short to avoid high memory usage.

Args:
source_video_path: Path to the source video file to analyze.
prompt: Instruction passed to Pegasus, e.g. "Summarize this video".
model_name: TwelveLabs Pegasus model to use.
max_tokens: Maximum number of tokens in the generated response.
api_key: TwelveLabs API key. Falls back to the `TWELVELABS_API_KEY`
environment variable when empty.

Returns:
The text produced by Pegasus describing the video.
"""
import base64

from twelvelabs import TwelveLabs
from twelvelabs.types.video_context import VideoContext_Base64String

api_key = api_key or os.environ.get("TWELVELABS_API_KEY", "")
if not api_key:
raise ValueError(
"A TwelveLabs API key is required. Pass --api_key or set the "
"TWELVELABS_API_KEY environment variable. Grab a free key at "
"https://twelvelabs.io."
)

with open(source_video_path, "rb") as video_file:
encoded_video = base64.b64encode(video_file.read()).decode("utf-8")

client = TwelveLabs(api_key=api_key)
response = client.analyze(
model_name=model_name,
video=VideoContext_Base64String(base_64_string=encoded_video),
prompt=prompt,
max_tokens=max_tokens,
)
return response.data or ""


def main(
source_video_path: str,
prompt: str = "Summarize what happens in this video.",
model_id: str = "yolov8n.pt",
confidence_threshold: float = 0.3,
pegasus_model_name: str = "pegasus1.5",
max_tokens: int = 512,
target_video_path: str | None = None,
api_key: str = "",
) -> None:
"""
Combine per-frame Supervision detections with a whole-video TwelveLabs
Pegasus understanding.

Supervision answers "what object is where, in each frame" while Pegasus
answers "what is this video about". Running both on the same clip pairs
precise, frame-level detections with a high-level narrative summary.

Args:
source_video_path: Path to the source video file.
prompt: Instruction passed to Pegasus for the video-level summary.
model_id: Ultralytics YOLO model id or weights path for detection.
confidence_threshold: Confidence level for detections (0 to 1).
pegasus_model_name: TwelveLabs Pegasus model to use.
max_tokens: Maximum number of tokens in the Pegasus response.
target_video_path: Optional path to save the annotated video. When
omitted, annotated frames are displayed in a window.
api_key: TwelveLabs API key. Falls back to the `TWELVELABS_API_KEY`
environment variable when empty.
"""
from ultralytics import YOLO

summary = analyze_video(
source_video_path=source_video_path,
prompt=prompt,
model_name=pegasus_model_name,
max_tokens=max_tokens,
api_key=api_key,
)
print("TwelveLabs Pegasus summary:")
print(summary)

model = YOLO(model_id)
video_info = sv.VideoInfo.from_video_path(video_path=source_video_path)
frames_generator = sv.get_video_frames_generator(source_video_path)

def annotate(frame: np.ndarray) -> np.ndarray:
result = model(frame, conf=confidence_threshold, verbose=False)[0]
detections = sv.Detections.from_ultralytics(result)
annotated_frame = BOX_ANNOTATOR.annotate(frame.copy(), detections)
annotated_frame = LABEL_ANNOTATOR.annotate(annotated_frame, detections)
return annotated_frame

if target_video_path is not None:
with sv.VideoSink(target_video_path, video_info) as sink:
for frame in frames_generator:
sink.write_frame(annotate(frame))
else:
for frame in frames_generator:
cv2.imshow("Processed Video", annotate(frame))
if cv2.waitKey(1) & 0xFF == ord("q"):
break
cv2.destroyAllWindows()


if __name__ == "__main__":
from jsonargparse import auto_cli, set_parsing_settings

set_parsing_settings(parse_optionals_as_positionals=True)
auto_cli(main, as_positional=False)