Overview

Media toolkit for processing video and audio attachments.

This package provides the infrastructure for media processing, including video clip extraction, frame sampling, audio extraction, keyframe extraction, and temporal segmentation.

Submodules

processor -- Processor families for video clip, frame sampling, and audio extraction operations.
segmenter -- Temporal segmenters for splitting media into fixed-duration chunks.
keyframe_extractor -- Keyframe extraction from video streams.

Usage

from gllm_multimodal.media_toolkit import processor, segmenter

`AudioExtractionProcessor()`

Bases: BackendSelectableProcessor[Attachment, Attachment], ABC

Family base for extracting audio tracks from video attachments.

This class serves as a unified entry point for audio extraction 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

Auto (Recommended)

from gllm_multimodal.media_toolkit.processor.audio_extraction_processor import AudioExtractionProcessor
from gllm_inference.schema import Attachment

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

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

FFmpeg Backend

from gllm_multimodal.media_toolkit.processor.audio_extraction_processor import AudioExtractionProcessor
from gllm_inference.schema import Attachment

# Explicitly force the ffmpeg backend
processor = AudioExtractionProcessor.build(backend="ffmpeg")

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

MoviePy Backend

from gllm_multimodal.media_toolkit.processor.audio_extraction_processor import AudioExtractionProcessor
from gllm_inference.schema import Attachment

# Explicitly force the moviepy backend
processor = AudioExtractionProcessor.build(backend="moviepy")

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

`FixedDurationSegmenter(segment_durations, start_time=0.0)`

Bases: BaseSegmenter

Segment attachments using explicit per-segment durations.

segment returns cumulative time windows from segment_durations. materialize clips each window into a separate attachment using a shared GstVideoClipProcessor instance that is created on first use and reused across all subsequent segments and videos.

Initialize the segmenter with manually provided segment durations.

Parameters:

Name	Type	Description	Default
`segment_durations`	`list[float]`	Ordered segment durations in seconds.	required
`start_time`	`float`	Base start time for the first segment. Defaults to 0.0.	`0.0`

Raises:

Type	Description
`ValueError`	If no segment duration is provided, or any duration is non-positive.

`materialize(attachment, segment, segment_index=0)` `async`

Clip and return one attachment for a precomputed segment window.

This method is convenient when segment planning and clip extraction are performed in separate stages, and only selected windows should be materialized.

Parameters:

Name	Type	Description	Default
`attachment`	`Attachment`	Source media attachment to clip.	required
`segment`	`VideoSegment`	Segment boundary plan to materialize.	required
`segment_index`	`int`	Zero-based index used in generated output filenames. Defaults to 0.	`0`

Returns:

Name	Type	Description
`Attachment`	`Attachment`	Clipped attachment with [`VideoSegment`][gllm_core.schema.multimodal.video_caption.VideoSegment] metadata.

`process(attachment, **kwargs)` `async`

Materialize fixed-duration clips from one media attachment.

Unlike calling segment directly, this method returns real clipped attachment outputs with [VideoSegment][gllm_core.schema.multimodal.video_caption.VideoSegment] metadata embedded on each result. It is the main runtime entrypoint when you need files/bytes for every configured duration window, not only boundary plans.

Parameters:

Name	Type	Description	Default
`attachment`	`Attachment`	Source media attachment to split.	required
`**kwargs`	`Any`	Forwarded processing arguments accepted by the base media-toolkit contract.	`{}`

Notes

Delegates shared validation and orchestration to MediaToolkit.process (inherited by BaseSegmenter).

Returns:

Type	Description
`list[Attachment]`	list[Attachment]: One clipped attachment per configured segment window.

`segment(attachment)` `async`

Return computed fixed windows without creating clip attachments.

This is useful for previewing time boundaries (for inspection, logging, or downstream planning) before paying the cost of media clipping.

Parameters:

Name	Type	Description	Default
`attachment`	`Attachment`	Source media attachment. The payload itself is not read by this implementation when computing boundaries.	required

Notes

Delegates shared validation to BaseSegmenter.segment.

Returns:

Type	Description
`list[VideoSegment]`	list[VideoSegment]: Fixed cumulative windows derived from
`list[VideoSegment]`	`segment_durations` and `start_time`.

`FrameSamplingProcessor()`

Bases: BackendSelectableProcessor[Attachment, Attachment], ABC

Family base for resampling video attachments to a target frame rate.

This class serves as a unified entry point for frame sampling 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

Auto (Recommended)

from gllm_multimodal.media_toolkit.processor.frame_sampling_processor import (
    FrameSamplingProcessor,
    GstFrameSamplingConfig,
)
from gllm_inference.schema import Attachment

# Instantiates the best available backend automatically
processor = FrameSamplingProcessor.build(
    config=GstFrameSamplingConfig(default_target_fps=2)
)

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

FFmpeg Backend

from gllm_multimodal.media_toolkit.processor.frame_sampling_processor import FrameSamplingProcessor
from gllm_inference.schema import Attachment

# Explicitly force the ffmpeg backend
processor = FrameSamplingProcessor.build(backend="ffmpeg")

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

`MediaToolkit()`

Bases: ABC, Generic[T_in, T_out]

Base abstraction for all media toolkit processing components.

This class provides the shared lifecycle and registry behavior used by both: - concrete leaf processors (e.g. backend-specific audio/video processors), and - composite components (e.g. segmenters, keyframe extractors) that orchestrate nested processors.

Key responsibilities: - auto-register subclasses by class name for class-name-based construction via build; - provide consistent input validation against supported_mimetypes; - define async processing contracts through process and process_batch.

Contributor guidance: - inherit this class directly for concrete processors with custom behavior; - inherit BackendSelectableProcessor when one logical processor family maps to multiple backend implementations; - inherit composite bases (e.g. BaseSegmenter) for orchestration-style components.

Example

Building a processor by class name

from gllm_multimodal.media_toolkit.media_toolkit import MediaToolkit

processor = MediaToolkit.build("AudioExtractionProcessor", backend="gstreamer")
result = await processor.process(video_attachment)

Checking mimetype support

if processor.is_supported(attachment):
    result = await processor.process(attachment)

Listing registered processors

print(list(MediaToolkit.registry.keys()))
# ['GstAudioExtractionProcessor', 'GstVideoClipProcessor', ...]

Initialize processor logging.

`registry = {}` `class-attribute`

Global class-name registry used by build.

Maps each registered subclass name to its concrete class type, enabling string-based construction such as MediaToolkit.build("AudioExtractionProcessor").

`supported_mimetypes = ['/']` `class-attribute`

MIME types this processor accepts (supports wildcards, e.g. 'video/*').

Defaults to ['*/*'] (accept all). Override as a class attribute in subclasses.

`__init_subclass__(**kwargs)`

Register every concrete subclass into the global registry.

This hook is triggered automatically by Python whenever a class inherits from MediaToolkit (directly or indirectly). Registration happens at class definition/import time, so classes become immediately discoverable by build without manual setup.

Registration key

The subclass' __name__ (e.g. "AudioExtractionProcessor").
The value stored is the subclass type itself.

Why uniqueness is enforced

build(class_name=...) uses this registry for class resolution.
Duplicate class names would silently shadow earlier classes and could route builds to unintended implementations.
To prevent that ambiguity, duplicate keys raise TypeError.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Extra class declaration keyword arguments forwarded to parent `__init_subclass__` implementations.	`{}`

Raises:

Type	Description
`TypeError`	If a subclass with the same class name is already registered, preventing silent dispatch to the wrong implementation.

`available_backends_for(class_name)` `classmethod`

Return backend keys registered for a processor family class name.

Parameters:

Name	Type	Description	Default
`class_name`	`str`	Registered processor family class name.	required

Returns:

Type	Description
`list[str]`	list[str]: Available backend keys. Empty when the class is unknown or not a backend-selectable family base.

Raises:

Type	Description
`ValueError`	If the class name is unknown.

Example

backends = MediaToolkit.available_backends_for("VideoClipProcessor")
print(backends)

Output:

["gstreamer", "ffmpeg"]

`build(class_name, backend=None, **kwargs)` `classmethod`

Build a processor by class name.

Family abstract classes (e.g. AudioExtractionProcessor) resolve a concrete backend implementation via backend. Composite components (segmenters, keyframe extractors) store backend on the instance for nested resolution.

Parameters:

Name	Type	Description	Default
`class_name`	`str`	Registered subclass name.	required
`backend`	`str \| MediaBackend \| None`	Backend key for family classes, or nested processor preference for composite instances. Defaults to None.	`None`
`**kwargs`	`Any`	Constructor kwargs passed to the processor class.	`{}`

Returns:

Name	Type	Description
`MediaToolkit`	`MediaToolkit`	Instantiated processor.

Raises:

Type	Description
`ValueError`	If the class name is unknown.

Example

proc = MediaToolkit.build("AudioExtractionProcessor", backend="gstreamer")
seg = MediaToolkit.build(
    "FixedDurationSegmenter",
    backend="gstreamer",
    segment_durations=[2.0],
)
print(proc)
print(seg)

Output:

<GstAudioExtractionProcessor instance>
<FixedDurationSegmenter instance>

`build_from_registry(backend=None, **kwargs)` `classmethod`

Instantiate this registered class.

Subclasses override this hook to customize registry-based construction (e.g. backend-selectable families resolve a concrete backend; composites store backend for nested processor resolution).

Parameters:

Name	Type	Description	Default
`backend`	`str \| MediaBackend \| None`	Backend key forwarded to subclass overrides. Ignored by the base implementation.	`None`
`**kwargs`	`Any`	Constructor kwargs passed to the processor class.	`{}`

Returns:

Name	Type	Description
`MediaToolkit`	`MediaToolkit`	Instantiated processor.

Example

# Called indirectly by MediaToolkit.build(...)
processor = SomeRegisteredProcessor.build_from_registry(custom_flag=True)
print(processor)

Output:

<SomeRegisteredProcessor instance>

`is_supported(attachment)`

Return whether the attachment's mimetype is accepted by this processor.

Callers can use this to check compatibility before calling process or process_batch, avoiding a ValueError.

Parameters:

Name	Type	Description	Default
`attachment`	`Attachment`	The attachment to check.	required

Returns:

Name	Type	Description
`bool`	`bool`	`True` if the attachment's mimetype matches any entry in `supported_mimetypes` (including wildcards). `True` is also returned when the attachment has no mimetype set.

Example

ok = processor.is_supported(Attachment(mime_type="video/mp4"))
print(ok)

Output:

True

`list_available_backends()` `classmethod`

Return backend keys when this class supports backend selection.

Returns:

Type	Description
`list[str]`	list[str]: Available backend keys. Empty for classes that are not backend-selectable family bases.

Example

Base MediaToolkit

Base classes are not backend-selectable.

print(MediaToolkit.list_available_backends())

Output:

[]

Backend-selectable family

Family classes expose registered backend keys.

from gllm_multimodal.media_toolkit.processor.video_clip_processor import VideoClipProcessor

backends = VideoClipProcessor.list_available_backends()
print(backends)

Output:

["gstreamer", "ffmpeg", "moviepy"]  # depends on registered backends

`process(attachment, **kwargs)` `async`

Process a single attachment (or perform an aggregation on a list) and return the result.

Parameters:

Name	Type	Description	Default
`attachment`	`T_in`	The attachment or list of attachments to process.	required
`**kwargs`	`Any`	Additional keyword arguments forwarded to `_process`.	`{}`

Returns:

Name	Type	Description
`T_out`	`T_out`	The result of the processing.

Example

result = await processor.process(attachment)
print(result)

Output:

<processed attachment or transformed output>

`process_batch(attachments, **kwargs)` `async`

Process a batch of attachments.

Parameters:

Name	Type	Description	Default
`attachments`	`list[T_in]`	The batch of attachments to process.	required
`**kwargs`	`Any`	Additional keyword arguments forwarded to `process`.	`{}`

Returns:

Type	Description
`list[T_out]`	list[T_out]: The result of the batch processing.

Example

results = await processor.process_batch([attachment_1, attachment_2])
print(results)

Output:

[<result_1>, <result_2>]

`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

Auto (Recommended)

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)

FFmpeg Backend

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)

MoviePy Backend

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

Overview

Submodules

Usage

AudioExtractionProcessor()

FixedDurationSegmenter(segment_durations, start_time=0.0)

materialize(attachment, segment, segment_index=0) async

process(attachment, **kwargs) async

segment(attachment) async

FrameSamplingProcessor()

MediaToolkit()

registry = {} class-attribute

supported_mimetypes = ['*/*'] class-attribute

__init_subclass__(**kwargs)

available_backends_for(class_name) classmethod

build(class_name, backend=None, **kwargs) classmethod

build_from_registry(backend=None, **kwargs) classmethod

is_supported(attachment)

list_available_backends() classmethod

process(attachment, **kwargs) async

process_batch(attachments, **kwargs) async

VideoClipProcessor()

set_windows(windows) abstractmethod

`AudioExtractionProcessor()`

`FixedDurationSegmenter(segment_durations, start_time=0.0)`

`materialize(attachment, segment, segment_index=0)` `async`

`process(attachment, **kwargs)` `async`

`segment(attachment)` `async`

`FrameSamplingProcessor()`

`MediaToolkit()`

`registry = {}` `class-attribute`

`supported_mimetypes = ['/']` `class-attribute`

`__init_subclass__(**kwargs)`

`available_backends_for(class_name)` `classmethod`

`build(class_name, backend=None, **kwargs)` `classmethod`

`build_from_registry(backend=None, **kwargs)` `classmethod`

`is_supported(attachment)`

`list_available_backends()` `classmethod`

`process(attachment, **kwargs)` `async`

`process_batch(attachments, **kwargs)` `async`

`VideoClipProcessor()`

`set_windows(windows)` `abstractmethod`