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
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 = processor(attachment)
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 = processor(attachment)
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 = processor(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. |
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
from gllm_multimodal.media_toolkit.processor.frame_sampling_processor import FrameSamplingProcessor
from gllm_inference.schema import Attachment
# Instantiates the best available backend automatically
processor = FrameSamplingProcessor.build(target_fps=2)
attachment = Attachment(url="file:///path/to/video.mp4")
sampled_video = processor(attachment)
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", target_fps=2)
attachment = Attachment(url="file:///path/to/video.mp4")
sampled_video = processor(attachment)
from gllm_multimodal.media_toolkit.processor.frame_sampling_processor import FrameSamplingProcessor
from gllm_inference.schema import Attachment
# Explicitly force the cv2 backend
processor = FrameSamplingProcessor.build(backend="cv2", target_fps=2)
attachment = Attachment(url="file:///path/to/video.mp4")
sampled_video = processor(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.
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 subclass by class name automatically.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Any
|
Extra class declaration kwargs. |
{}
|
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. |
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
MediaToolkit.build("AudioExtractionProcessor", backend="gstreamer")
resolves and returns a concrete backend class
(e.g. GstAudioExtractionProcessor).
MediaToolkit.build("FixedDurationSegmenter", backend="gstreamer", segment_durations=[2.0])
creates the segmenter and stores backend preference for nested
processor resolution.
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. |
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
|
|
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. |
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 |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
T_out |
T_out
|
The result of the processing. |
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 |
{}
|
Returns:
| Type | Description |
|---|---|
list[T_out]
|
list[T_out]: The result of the batch processing. |
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 = processor(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 = processor(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 = processor(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 |