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)
│   ├── DeinterlaceProcessor (family base)
│   │   └── FFmpegDeinterlaceProcessor (concrete)
│   ├── FrameExtractionProcessor (family base)
│   │   ├── GstFrameExtractionProcessor (concrete, default)
│   │   └── FFmpegFrameExtractionProcessor (concrete)
│   └── FrameDecodeProcessor (family base)
│       ├── GstFrameDecodeProcessor (concrete, default)
│       └── FFmpegFrameDecodeProcessor (concrete)
├── BaseSegmenter (composite, orchestrates nested processors)
│   ├── FixedDurationSegmenter
│   └── ShotBasedSegmenter
└── BaseKeyframeExtractor (composite, orchestrates nested processors)
    ├── UniformKeyframeExtractor
    └── LDDRKeyframeExtractor

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

Registration — every MediaToolkit subclass is registered by class name in MediaToolkit.registry via __init_subclass__.
Construction — call build_media_toolkit(class_name, backend=..., **kwargs) to instantiate any registered component. (Under the hood, this calls MediaToolkit.build).
Processing — call await processor.process(attachment) to process a single attachment or await processor.process_batch(attachments) for batches.
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.

`name` `property`

Return a stable component name for optional plan metadata.

Returns:

Name	Type	Description
`str`	`str`	Component class name.

`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",
    config={"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>]

Media Toolkit

Architecture overview

Registered subclasses

Processor

Segmenter

KeyframeExtractor

Lifecycle

MediaToolkit()

name property

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

`MediaToolkit()`

`name` `property`

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