Overview

Shared backend infrastructure for media toolkit processors.

This package holds base classes and utilities for backend implementations (GStreamer, FFmpeg CLI).

Exported Classes

• BaseGstreamerProcessor -- Abstract base for GStreamer processors.
• GstBaseConfig -- Base configuration for GStreamer processors.
• EncoderFormatInfo -- Encoder format metadata.
• BaseFFmpegProcessor -- Shared helpers for FFmpeg CLI processors.
• FFmpegBaseConfig -- Minimal shared configuration for FFmpeg processors.
• coerce_config -- Hydrate a constructor config into a specific model type.
• merge_process_config -- Overlay per-call process config onto a constructor model.

BaseFFmpegProcessor(config=None)

Bases: MediaToolkit[Attachment, Attachment], Generic[TConfig]

Shared helpers for processors that shell out to the ffmpeg binary.

Unlike BaseGstreamerProcessor, this base does not require ffmpeg at construction time — availability is checked when a command is run. That keeps import / build paths light when ffmpeg is only needed for optional I/O.

Attributes:

Name	Type	Description
INSTALL_HINT	str	Human-readable install guidance for missing ffmpeg.
config	TConfig	Coerced stable configuration from config_model().

Initialize the FFmpeg helper base and coerce constructor config.

Parameters:

Name	Type	Description	Default
config	dict[str, Any] | BaseModel | None	Stable config dict, family/backend model, or None for defaults.	None

config_model() classmethod

Return the stable configuration model for this processor.

Concrete backends must override with their FFmpeg config type.

Returns:

Type	Description
type[TConfig]	type[TConfig]: The configuration class used at construction time.

BaseGstreamerProcessor(config=None, *, enable_video=True, enable_audio=True)

Bases: MediaToolkit[Attachment, Attachment], Generic[TConfig]

Abstract base class for GStreamer-powered processors.

Subclasses must implement only _execute_pipeline. All common concerns — availability checks, Conda environment configuration, GStreamer initialisation, encoder/element selection, the standard EOS/error bus loop, and temporary-file cleanup — are handled here.

Parameterise with the concrete config model, e.g. BaseGstreamerProcessor[GstFrameDecodeConfig].

Typical subclass skeleton::

class MyGstProcessor(BaseGstreamerProcessor[GstBaseConfig]):
    def __init__(self, my_param, config=None):
        super().__init__(config=config)
        self.my_param = my_param

    async def _process(self, attachment: Attachment) -> Attachment:
        return await self._process_single(attachment)

    async def _execute_pipeline(self, input_path, output_path):
        # build & run YOUR GStreamer pipeline here
        ...

Attributes:

Name	Type	Description
logger		Logger bound to the concrete subclass name.
config	TConfig	Runtime configuration for this processor family.
video_encoder_info	EncoderFormatInfo | None	Selected video encoder format info, or None when video is disabled.
audio_encoder_info	EncoderFormatInfo | None	Selected audio encoder format info, or None when audio is disabled.

Verify GStreamer availability, configure the environment, and select encoders.

Parameters:

Name	Type	Description	Default
config	dict[str, Any] | GstBaseConfig | BaseModel | None	Optional configuration. A plain dict or shared family BaseModel is coerced into config_model. None uses all defaults of that model.	None
enable_video	bool	When True, select a video encoder. Defaults to True.	True
enable_audio	bool	When True, select an audio encoder. Defaults to True.	True

Raises:

Type	Description
RuntimeError	If GStreamer is not installed or cannot be initialised.
RuntimeError	If no suitable video encoder is found in the registry.
TypeError	If config is neither dict, BaseModel, nor None.

config_model() classmethod

Return the stable configuration model for this processor.

Returns:

Type	Description
type[TConfig]	type[TConfig]: The configuration class used at construction time.

EncoderFormatInfo(encoder, muxer, extension, mime) dataclass

Immutable metadata for a selected encoder and its associated muxer/format.

Bundles encoder element name together with the container muxer, file extension, and MIME type so callers never have to juggle four separate string attributes.

FFmpegBaseConfig

Bases: BaseModel

Minimal shared configuration for FFmpeg CLI processors.

Family backends extend this (or their own shared family config) with engine-specific fields. Kept intentionally thin — FFmpeg leaves share process helpers more than stable knobs.

GstBaseConfig

Bases: BaseModel

Minimal shared configuration for all GStreamer-based processors.

Attributes:

Name	Type	Description
timeout	int	Maximum seconds a GStreamer pipeline may run before being forcibly terminated. Defaults to 300 seconds.
audio_passthrough	bool	When True (the default), encoded audio streams are passed directly to the muxer without being decoded and re-encoded. Set to False to force re-encoding via the best available audio encoder.
video_encoder	str | None	Pin a specific GStreamer video encoder element name (e.g. "x264enc"). When set, the candidate-list scan is skipped entirely and this element is used directly. The element must exist in the GStreamer registry. Defaults to None (auto-select from _VIDEO_FORMAT_MAP).
audio_encoder	str | None	Pin a specific GStreamer audio encoder element name (e.g. "voaacenc"). When set, the candidate-list scan is skipped entirely and this element is used directly. The element must exist in the GStreamer registry. Defaults to None (auto-select from _AUDIO_FORMAT_MAP).

coerce_config(config, model)

Coerce a constructor config dict / model / None into model.

Shared family configs (parent BaseModel instances that are not the exact backend model) are re-validated so backend-specific defaults fill in.

Parameters:

Name	Type	Description	Default
config	dict[str, Any] | BaseModel | None	A dict to hydrate, a model instance, or None for defaults.	required
model	type[TModel]	Target constructor config class.	required

Returns:

Name	Type	Description
TModel	TModel	A typed config instance of model.

Raises:

Type	Description
TypeError	If config is neither dict, BaseModel, nor None.

coerce_optional_process_config(process_config, model_cls, *, allow_family=False)

Normalize an optional per-invocation process config.

Parameters:

Name	Type	Description	Default
process_config	ProcessorProcessConfig | dict[str, Any] | None	Per-call config, a dict to hydrate, or None.	required
model_cls	type[TProcess]	Backend process-config class.	required
allow_family	bool	When True, parent family process configs are re-validated into model_cls. When False, a parent instance raises. Defaults to False.	False

Returns:

Type	Description
TProcess | None	TProcess | None: None when omitted; otherwise a model_cls instance.

Raises:

Type	Description
TypeError	If process_config is an unsupported type, or a family instance when allow_family is False.

merge_process_config(stable, process_config, *, process_model, exclude_keys=frozenset(), field_map=None)

Overlay non-None process-config fields onto stable.

Parameters:

Name	Type	Description	Default
stable	TModel	Constructor config to start from.	required
process_config	ProcessorProcessConfig | dict[str, Any] | None	Per-call overrides. None returns stable unchanged.	required
process_model	type[TProcess]	Backend process-config class.	required
exclude_keys	frozenset[str]	Process-config keys skipped during the overlay. Defaults to empty.	frozenset()
field_map	Mapping[str, str] | None	Copy process-config src onto stable key dest (e.g. timestamps → default_timestamps). Defaults to None.	None

Returns:

Name	Type	Description
TModel	TModel	Effective constructor config of the same type as stable.

Raises:

Type	Description
TypeError	If process_config cannot be coerced to process_model.