Media Toolkit

Base abstractions and factory entrypoint for the media toolkit.

Defines MediaToolkit, the shared base for leaf processors, backend-selectable processor families, and composite orchestration components (segmenters, keyframe extractors). Subclasses auto-register by class name for construction via build.

Architecture overview

MediaToolkit (abstract base)
├── BackendSelectableProcessor (abstract, adds backend registry)
│   ├── AudioExtractionProcessor (family base)
│   │   └── GstAudioExtractionProcessor (concrete)
│   ├── VideoClipProcessor (family base)
│   │   └── GstVideoClipProcessor (concrete)
│   └── FrameSamplingProcessor (family base)
│       ├── GstFrameSamplingProcessor (concrete)
│       └── FfmpegFrameSamplingProcessor (concrete)
├── BaseSegmenter (composite, orchestrates nested processors)
│   └── FixedDurationSegmenter
└── BaseKeyframeExtractor (composite, orchestrates nested processors)
    └── UniformKeyframeExtractor

Registered subclasses

Every MediaToolkit subclass self-registers on definition. There are three logical families — Processor, Segmenter, and KeyframeExtractor — each with their own base class and API page.

Processor

Leaf components that receive a single Attachment and produce one or more Attachment outputs. Backend-selectable families (audio extraction, video clipping, frame sampling) automatically resolve the best available backend at construction time.

See BackendSelectableProcessor for the full API.

from gllm_multimodal.builder.media_toolkit_builder import build_media_toolkit

# Extract audio track from a video
processor = build_media_toolkit("AudioExtractionProcessor")
result = await processor.process(video_attachment)

# Sample frames at 2 fps using GStreamer
processor = build_media_toolkit("FrameSamplingProcessor", backend="gstreamer")
frames = await processor.process(video_attachment)

Segmenter

Composite components that split a media Attachment into a list of fixed- duration or content-aware clips. Segmenters own their nested VideoClipProcessor and forward the active backend to it automatically.

See BaseSegmenter for the full API.

from gllm_multimodal.builder.media_toolkit_builder import build_media_toolkit

segmenter = build_media_toolkit(
    "FixedDurationSegmenter",
    backend="gstreamer",
    segment_duration=30.0,
)
clips = await segmenter.process(video_attachment)

KeyframeExtractor

Composite components that extract representative frames from a video clip. They orchestrate a nested FrameSamplingProcessor and a VideoClipProcessor internally.

See BaseKeyframeExtractor for the full API.

from gllm_multimodal.builder.media_toolkit_builder import build_media_toolkit

extractor = build_media_toolkit(
    "UniformKeyframeExtractor",
    backend="gstreamer",
    num_frames=8,
)
keyframes = await extractor.process(video_attachment)

Lifecycle

1. Registration — every MediaToolkit subclass is registered by class name in MediaToolkit.registry via __init_subclass__.
2. Construction — call build_media_toolkit(class_name, backend=..., **kwargs) to instantiate any registered component. (Under the hood, this calls MediaToolkit.build).
3. Processing — call await processor.process(attachment) to process a single attachment or await processor.process_batch(attachments) for batches.
4. Backend selection — backend-selectable families resolve a concrete backend at construction time; composites store the backend for nested processor resolution.

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	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.

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 _process.	{}

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 process.	{}

Returns:

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