Base Gstreamer Processor

Abstract base class for all GStreamer-based processors.

Handles GStreamer availability detection, conditional import, environment configuration, runtime initialisation, encoder/element selection, the standard EOS/error bus loop, and temporary-file cleanup — so that concrete sub-classes only need to implement their own pipeline topology.

Architecture

BaseGstreamerProcessor
├── GStreamer availability check (GST_AVAILABLE flag)
├── Encoder selection (_select_encoder, _select_video_encoder, _select_audio_encoder)
├── Pipeline execution (_execute_pipeline — abstract, subclass implements)
├── File transcoding (_run_file_transcode — shared recipe)
└── Temp file management (_write_input_tempfile, _cleanup)

Subclass contract

Subclasses must implement only _execute_pipeline. All common concerns are handled by this base class.

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

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

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

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.

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