Skip to content

GStreamer Python plugin

A Python plugin is useful for fast prototyping. For high-performance frame processing or custom Gst.Meta, move the final plugin to C, C++, or Rust.

Posts

Metadata

bus message

appsink metadata

Gst.Event

Source element

Install

1
sudo apt install -y gstreamer1.0-python3-plugin-loader python3-gi python3-gst-1.0

Folder layout

The Python plugin loader searches for Python files inside a python/ directory under GST_PLUGIN_PATH.

1
2
3
my_plugins/
  python/
    minimal_plugin.py

Run GStreamer with:

1
GST_PLUGIN_PATH=$PWD/my_plugins gst-inspect-1.0 minimalfilter

Basic blocks

Imports and init

Load the GStreamer bindings and initialize GStreamer.

1
2
3
4
5
6
7
import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstBase", "1.0")
from gi.repository import Gst, GstBase, GObject

Gst.init(None)

Element class

Most simple filters inherit from GstBase.BaseTransform. It receives one buffer on the sink pad, processes it, and pushes one buffer on the source pad.

1
2
class MinimalFilter(GstBase.BaseTransform):
    pass

Plugin metadata

__gstmetadata__ describes the element shown by gst-inspect-1.0.

1
2
3
4
5
6
__gstmetadata__ = (
    "MinimalFilter",
    "Filter/Video",
    "Minimal Python GStreamer passthrough element",
    "example",
)

Pad templates

Pad templates define what the element can accept and produce. This example accepts any raw video and outputs any raw video.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
__gsttemplates__ = (
    Gst.PadTemplate.new(
        "sink",
        Gst.PadDirection.SINK,
        Gst.PadPresence.ALWAYS,
        Gst.Caps.from_string("video/x-raw"),
    ),
    Gst.PadTemplate.new(
        "src",
        Gst.PadDirection.SRC,
        Gst.PadPresence.ALWAYS,
        Gst.Caps.from_string("video/x-raw"),
    ),
)

Buffer processing

do_transform_ip() means transform in-place. The same buffer continues downstream.

1
2
3
def do_transform_ip(self, buffer):
    print("buffer pts:", buffer.pts)
    return Gst.FlowReturn.OK

Use this method to inspect timestamps, map frame bytes, run a detector, draw on the frame, or post messages.

Register the element

The factory name is the name used in a pipeline.

1
2
GObject.type_register(MinimalFilter)
__gstelementfactory__ = ("minimalfilter", Gst.Rank.NONE, MinimalFilter)

Minimal plugin

Full example:

code/minimal_plugin.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
#!/usr/bin/env python3
import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstBase", "1.0")
from gi.repository import Gst, GstBase, GObject

Gst.init(None)


class MinimalFilter(GstBase.BaseTransform):
    __gstmetadata__ = (
        "MinimalFilter",
        "Filter/Video",
        "Minimal Python GStreamer passthrough element",
        "example",
    )

    __gsttemplates__ = (
        Gst.PadTemplate.new(
            "sink",
            Gst.PadDirection.SINK,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.from_string("video/x-raw"),
        ),
        Gst.PadTemplate.new(
            "src",
            Gst.PadDirection.SRC,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.from_string("video/x-raw"),
        ),
    )

    def do_transform_ip(self, buffer):
        print("buffer pts:", buffer.pts)
        return Gst.FlowReturn.OK


GObject.type_register(MinimalFilter)
__gstelementfactory__ = ("minimalfilter", Gst.Rank.NONE, MinimalFilter)

The plugin folder should look like this:

1
2
3
plugins/
└── python/
    └── minimal_plugin.py

minimal_plugin.py contains the plugin code shown above. GStreamer discovers Python plugins from the python/ subfolder of each directory listed in GST_PLUGIN_PATH.

Test

Put the file under python/minimal_plugin.py, then run:

1
2
# from plugin folder
GST_PLUGIN_PATH=$PWD gst-inspect-1.0 minimalfilter

Run a pipeline:

1
GST_PLUGIN_PATH=$PWD gst-launch-1.0 videotestsrc num-buffers=5 ! minimalfilter ! fakesink

If GStreamer does not find the element, enable plugin loading logs:

1
2
3
GST_PLUGIN_PATH=$PWD \
GST_DEBUG=python:6,pythonplugin:6,GST_PLUGIN_LOADING:5 \
gst-inspect-1.0 minimalfilter

Troubleshooting

If the inspect not found the plugin check if it block list

what is block list

It means GStreamer tried to load that plugin before, the load failed, and GStreamer cached it as “bad” in its registry. After that, it skips the plugin quickly instead of trying to load it every time.

1
gst-inspect-1.0 -b
If the plugin in the block list clear the cache

1
rm ~/.cache/gstreamer-1.0/registry.*.bin