Implementation:NVIDIA NeMo Curator VideoTasks

Knowledge Sources	NVIDIA NeMo Curator
Domains	Video Processing, Data Model, Pipeline Architecture, Data Curation
Last Updated	2026-02-14 00:00 GMT

Overview

Defines the comprehensive data model for video processing, including dataclass definitions for clips, windows, metadata, statistics, and the top-level VideoTask pipeline wrapper.

Description

The video tasks module serves as the central schema definition for the entire video curation pipeline. It defines six dataclasses that represent the hierarchical structure of video processing data:

_Window: An internal dataclass representing a temporal window within a video clip used for captioning. It stores the start and end frame numbers, MP4 bytes for the window, model-specific inputs (Qwen LLM input, X1 model input), generated captions and enhanced captions (keyed by model name), and optional WebP preview bytes. Includes get_major_size() for memory estimation.

Clip: Represents a video segment (clip) identified by a UUID with a time span tuple (start, end). Contains fields for the video buffer, extracted frames (dictionary keyed by extraction signature), motion detection data and scores (motion_score_global_mean, motion_score_per_patch_min_256), aesthetic score, Cosmos Embed1 frames and embeddings, captioning windows, egomotion data, text match results, and error tracking. Provides extract_metadata() which delegates to extract_video_metadata from decoder_utils, a duration property, and get_major_size().

ClipStats: Accumulates pipeline statistics including counts for clips filtered by motion, filtered by aesthetic, passed, transcoded, with embeddings, with captions, and with WebP previews, plus total and maximum clip durations. The combine() method merges two ClipStats instances.

VideoMetadata: A lightweight container for video properties: size (bytes), height, width, framerate, number of frames, duration, video codec, pixel format, audio codec, and bit rate in kilobits.

Video: The top-level container linking an input video path (pathlib.Path) with source bytes, metadata, frame arrays (for TransNetV2 scene detection), lists of clips and filtered clips, chunking information, clip statistics, and error tracking. Key methods include:

• populate_metadata(): Extracts metadata from source_bytes using extract_video_metadata
• weight property: Calculates processing weight normalized to a 5-minute duration, adjusted by the fraction of clips processed
• fraction property: Returns the ratio of processed clips to total clips
• is_10_bit_color(): Heuristic check for 10-bit color depth via pixel format inspection
• has_metadata(): Validates that all essential metadata fields are present

VideoTask: Wraps a Video in the generic Task[Video] interface. Provides validate() (checks file existence) and num_items (always returns 1).

Usage

These dataclasses are used throughout the video curation pipeline. Every processing stage -- from video reading and scene splitting through motion filtering, aesthetic scoring, embedding generation, captioning, and writing -- reads from and writes to instances of these classes. The VideoTask wrapper provides the interface expected by the pipeline's task scheduling infrastructure.

Code Reference

Source Location

• Repository: NeMo-Curator
• File: nemo_curator/tasks/video.py
• Lines: 1-376

Signature

@dataclass
class _Window:
    start_frame: int
    end_frame: int
    mp4_bytes: bytes | None = None
    qwen_llm_input: dict[str, Any] | None = None
    x1_input: Any | None = None
    caption: dict[str, str] = field(default_factory=dict)
    enhanced_caption: dict[str, str] = field(default_factory=dict)
    webp_bytes: bytes | None = None
    def get_major_size(self) -> int: ...

@dataclass
class Clip:
    uuid: UUID
    source_video: str
    span: tuple[float, float]
    buffer: bytes | None = None
    extracted_frames: dict[str, npt.NDArray[np.uint8]] = field(default_factory=dict)
    decoded_motion_data: None = None
    motion_score_global_mean: float | None = None
    motion_score_per_patch_min_256: float | None = None
    aesthetic_score: float | None = None
    cosmos_embed1_frames: npt.NDArray[np.float32] | None = None
    cosmos_embed1_embedding: npt.NDArray[np.float32] | None = None
    windows: list[_Window] = field(default_factory=list)
    egomotion: dict[str, bytes] = field(default_factory=dict)
    cosmos_embed1_text_match: tuple[str, float] | None = None
    errors: dict[str, str] = field(default_factory=dict)
    def extract_metadata(self) -> dict[str, Any] | None: ...
    @property
    def duration(self) -> float: ...
    def get_major_size(self) -> int: ...

@dataclass
class ClipStats:
    num_filtered_by_motion: int = 0
    num_filtered_by_aesthetic: int = 0
    num_passed: int = 0
    num_transcoded: int = 0
    num_with_embeddings: int = 0
    num_with_caption: int = 0
    num_with_webp: int = 0
    total_clip_duration: float = 0.0
    max_clip_duration: float = 0.0
    def combine(self, other: "ClipStats") -> None: ...

@dataclass
class VideoMetadata:
    size: int | None = None
    height: int | None = None
    width: int | None = None
    framerate: float | None = None
    num_frames: int | None = None
    duration: float | None = None
    video_codec: str | None = None
    pixel_format: str | None = None
    audio_codec: str | None = None
    bit_rate_k: int | None = None

@dataclass
class Video:
    input_video: pathlib.Path
    source_bytes: bytes | None = None
    metadata: VideoMetadata = field(default_factory=VideoMetadata)
    frame_array: npt.NDArray[np.uint8] | None = None
    clips: list[Clip] = field(default_factory=list)
    filtered_clips: list[Clip] = field(default_factory=list)
    num_total_clips: int = 0
    num_clip_chunks: int = 0
    clip_chunk_index: int = 0
    clip_stats: ClipStats = field(default_factory=ClipStats)
    errors: dict[str, str] = field(default_factory=dict)
    def populate_metadata(self) -> None: ...
    @property
    def fraction(self) -> float: ...
    @property
    def weight(self) -> float: ...
    def get_major_size(self) -> int: ...
    def has_metadata(self) -> bool: ...
    def is_10_bit_color(self) -> bool | None: ...
    @property
    def input_path(self) -> str: ...

@dataclass
class VideoTask(Task[Video]):
    data: Video = field(default_factory=Video)
    def validate(self) -> bool: ...
    @property
    def num_items(self) -> int: ...

Import

from nemo_curator.tasks.video import Video, VideoTask, Clip, ClipStats, VideoMetadata

I/O Contract

Inputs

Name	Type	Required	Description
input_video	pathlib.Path	Yes	Path to the input video file for the Video dataclass.
source_bytes	None	No	Raw video bytes; required by populate_metadata() to extract video properties.
uuid	UUID	Yes	Unique identifier for a Clip instance.
source_video	str	Yes	Source video identifier for a Clip.
span	tuple[float, float]	Yes	Start and end timestamps (in seconds) for a Clip.
start_frame	int	Yes	Start frame number for a _Window.
end_frame	int	Yes	End frame number for a _Window.

Outputs

Name	Type	Description
metadata	None	Video metadata dictionary from Clip.extract_metadata() containing width, height, framerate, num_frames, video_codec, and num_bytes.
duration	float	Clip duration in seconds from the Clip.duration property.
weight	float	Processing weight from Video.weight, normalized to a 5-minute reference duration.
fraction	float	Ratio of processed clips to total clips from Video.fraction.
is_10_bit_color	None	Whether the video uses 10-bit color depth, or None if pixel format is unknown.
get_major_size	int	Estimated total memory size in bytes of the object and its contents.

Usage Examples

Creating a Video Task

import pathlib
from nemo_curator.tasks.video import Video, VideoTask

# Create a Video object from a file path
video = Video(input_video=pathlib.Path("/data/videos/sample.mp4"))

# Wrap it in a VideoTask for pipeline processing
task = VideoTask(data=video)

# Validate the task (checks file existence)
if task.validate():
    print(f"Task is valid, items: {task.num_items}")

Working with Clips

from uuid import uuid4
from nemo_curator.tasks.video import Clip

# Create a clip representing a 5-second segment
clip = Clip(
    uuid=uuid4(),
    source_video="/data/videos/sample.mp4",
    span=(10.0, 15.0),
)

print(f"Clip duration: {clip.duration} seconds")
print(f"Memory size: {clip.get_major_size()} bytes")

# Extract metadata if buffer is populated
clip.buffer = open("/data/videos/sample.mp4", "rb").read()
metadata = clip.extract_metadata()
if metadata:
    print(f"Resolution: {metadata['width']}x{metadata['height']}")

Populating Video Metadata

import pathlib
from nemo_curator.tasks.video import Video

video = Video(input_video=pathlib.Path("/data/videos/sample.mp4"))
video.source_bytes = open("/data/videos/sample.mp4", "rb").read()

# Extract and populate all metadata fields
video.populate_metadata()

print(f"Resolution: {video.metadata.width}x{video.metadata.height}")
print(f"Duration: {video.metadata.duration}s")
print(f"FPS: {video.metadata.framerate}")
print(f"Codec: {video.metadata.video_codec}")
print(f"10-bit color: {video.is_10_bit_color()}")

Aggregating Clip Statistics

from nemo_curator.tasks.video import ClipStats

stats_a = ClipStats(num_passed=10, num_filtered_by_motion=3, total_clip_duration=50.0)
stats_b = ClipStats(num_passed=8, num_filtered_by_motion=2, total_clip_duration=40.0)

# Combine statistics from multiple workers
stats_a.combine(stats_b)
print(f"Total passed: {stats_a.num_passed}")          # 18
print(f"Total filtered by motion: {stats_a.num_filtered_by_motion}")  # 5
print(f"Total clip duration: {stats_a.total_clip_duration}")  # 90.0

Related Pages

• Environment:NVIDIA_NeMo_Curator_Python_Linux_Base