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):
    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]

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.

Typical subclass skeleton::

class MyGstProcessor(BaseGstreamerProcessor):
    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	GstBaseConfig	Runtime configuration.
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 | None	Optional configuration. A plain dict is coerced into a GstBaseConfig. None uses all GstBaseConfig defaults.	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.

config_model() classmethod

Return the stable configuration model for this processor.

Returns:

Type	Description
type[GstBaseConfig]	type[GstBaseConfig]: 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).