Skip to content

Overview

Video clip processor family.

This package provides processors for extracting video clips from longer videos based on time ranges or keyframe boundaries.

Exported Classes

GstVideoClipConfig

Bases: GstBaseConfig

Stable configuration for GstVideoClipProcessor.

Set once at construction. Controls encoder selection, timeout, and other processor behaviour that does not change per attachment.

Attributes:

Name Type Description
default_windows list[tuple[float, float]] | None

Default clipping windows used when GstVideoClipProcessor.process is called without a process_config. None (the default) means windows must be supplied per-call via GstVideoClipProcessConfig. When set, the value is validated by GstVideoClipProcessConfig's field_validator.

validate_default_windows(default_windows) classmethod

Validate the constructor-level default windows, if any.

Parameters:

Name Type Description Default
default_windows list[tuple[float, float]] | None

The candidate default windows to validate.

 required

Returns:

Type Description
list[tuple[float, float]] | None

list[tuple[float, float]] | None: default_windows unchanged
list[tuple[float, float]] | None

when None or when every window is well-ordered.

Raises:

Type Description
ValueError

If default_windows is empty or any window has end <= start. Validation mirrors GstVideoClipProcessConfig's field_validator.

GstVideoClipProcessConfig

Bases: ProcessorProcessConfig

Per-invocation configuration for [process][gllm_multimodal.media_toolkit.media_toolkit.MediaToolkit.process].

Attributes:

Name Type Description
windows list[tuple[float, float]]

Ordered (start_time, end_time) windows in seconds. Each window must satisfy end > start.

validate_windows(windows) classmethod

Validate that every clipping window is non-empty and well ordered.

Parameters:

Name Type Description Default
windows list[tuple[float, float]]

The windows to validate.

 required

Returns:

Type Description
list[tuple[float, float]]

list[tuple[float, float]]: The validated windows.

GstVideoClipProcessor(config=None)

Bases: BaseGstreamerProcessor[GstVideoClipConfig], VideoClipProcessor

Clips a video attachment to one or more [start_time, end_time] windows.

Stable settings belong in GstVideoClipConfig (constructor). Per-call clip windows belong in GstVideoClipProcessConfig (process(..., process_config=...)).

Initialise the clip processor.

Parameters:

Name Type Description Default
config dict[str, Any] | GstVideoClipConfig | None

Optional stable configuration. None uses GstVideoClipConfig defaults.

 None

Raises:

Type Description
ValueError

If config.default_windows is empty or any window has end <= start.

config_model() classmethod

Return the stable configuration model for this processor.

Returns:

Type Description
type[GstVideoClipConfig]

type[GstVideoClipConfig]: The stable configuration model for this processor.

process(attachment, **kwargs) async

Clip one video attachment using resolved time windows.

This method is the canonical caller-facing API for clipping. It accepts stable default windows set at construction time and also supports per-call window overrides through process_config.

Parameters:

Name Type Description Default
attachment Attachment

Input video attachment to clip.

 required
**kwargs Any

Runtime options, typically process_config as GstVideoClipProcessConfig or dict.

 {}
Notes
  1. Delegates shared mimetype validation and dispatch to MediaToolkit.process.

Returns:

Type Description
Attachment | list[Attachment]

Attachment | list[Attachment]: Single clipped attachment when exactly
Attachment | list[Attachment]

one window is used, otherwise a list of clipped attachments in window order.
Example 
processor = GstVideoClipProcessor(
    config={"default_windows": [(0.0, 10.0)]},
)
clips = await processor.process(
    attachment=video_attachment,
    process_config={
        "windows": [(5.0, 12.5), (30.0, 45.0)],
    },
)

process_config_model() classmethod

Return the per-invocation configuration model for this processor.

Returns:

Type Description
type[GstVideoClipProcessConfig]

type[GstVideoClipProcessConfig]: The per-invocation configuration model for this processor.

set_windows(windows)

Update the constructor-level default windows.

Mutates config so process falls back to windows when no process_config is supplied. Prefer passing GstVideoClipProcessConfig to process for per-call overrides.

Parameters:

Name Type Description Default
windows list[tuple[float, float]]

The windows to set.

 required

Returns:

Name Type Description
None None

self.config is replaced with a copy whose
None

default_windows is the validated windows.

Raises:

Type Description
ValueError

If windows is empty or any window has end <= start. Validation reuses GstVideoClipProcessConfig's field_validator.

VideoClipProcessor()

Bases: BackendSelectableProcessor[Attachment, Attachment], ABC

Family base for clipping video attachments to time windows.

This class serves as a unified entry point for video clipping operations. It automatically routes requests to the most appropriate, available backend implementation based on your system environment.

Why use this base class?

  • Portability: Your code will run regardless of which underlying libraries are installed on the host machine.
  • Simplicity: No need to handle fallback logic or conditional imports yourself.
  • Future-proofing: New backends can be added to the library without requiring changes to your application code.

Usage Example

from gllm_multimodal.media_toolkit.processor.video_clip_processor import VideoClipProcessor
from gllm_inference.schema import Attachment

# Instantiates the best available backend automatically
processor = VideoClipProcessor.build()

# Set the target clipping window (start_time, end_time) in seconds
processor.set_windows([(10.0, 20.5)])

attachment = Attachment(url="file:///path/to/video.mp4")
clipped_video = await processor.process(attachment)

from gllm_multimodal.media_toolkit.processor.video_clip_processor import VideoClipProcessor
from gllm_inference.schema import Attachment

# Explicitly force the ffmpeg backend
processor = VideoClipProcessor.build(backend="ffmpeg")
processor.set_windows([(10.0, 20.5)])

attachment = Attachment(url="file:///path/to/video.mp4")
clipped_video = await processor.process(attachment)

from gllm_multimodal.media_toolkit.processor.video_clip_processor import VideoClipProcessor
from gllm_inference.schema import Attachment

# Explicitly force the moviepy backend
processor = VideoClipProcessor.build(backend="moviepy")
processor.set_windows([(10.0, 20.5)])

attachment = Attachment(url="file:///path/to/video.mp4")
clipped_video = await processor.process(attachment)

set_windows(windows) abstractmethod

Configure one or more [start, end] clipping windows for the next call.

Parameters:

Name Type Description Default
windows list[tuple[float, float]]

List of (start, end) tuples in seconds.

 required