How to write Gstreamer plugin with Python?

10 min. read |

Next guide shows steps to write Gstreamer Plugin in Python for any Computer Vision, Image Processing task and use it in standard Gstreamer pipeline from command line. Additionally this post explains how to get/set own properties from implemented Gstreamer plugin.

Requirements

• Ubuntu 18
• Python 3.6
• Gstreamer with Python Bindings. Look at “How to install Gstreamer Python Bindings on Ubuntu 18”
• gstreamer-python
• opencv-python

Code

• gst-python-plugins

Learn how to?

• create basic gstreamer plugins
  • “hello_world” plugin (gstplugin_py)
  • blur filter plugin (gaussian_blur)
• set/get user defined properties for plugin

Use gstreamer plugins

• sources: videotestsrc, filesrc
• transforms: videoconvert, decodebin, capsfilter, videobox
• mixer: videomixer
• sinks: autovideosink, gtksink, fakesink

Preface

Inspired by guide “How to write Gstreamer elements in Python”, where the author shows how to write Audio Source and Filter Elements, I wanted to create simple example on how to write Gstreamer plugins for Computer Vision/Image Processing purpose. Despite mentioned disadvantages of Python implementation of Gstreamer elements it is still faster than any other Python approach in image processing (like OpenCV), and more flexible as you can easily reuse this element in other Gstreamer pipelines without changing any code.

Guide

Extend GstBase.BaseTransform

GstBase.BaseTransform is base class for filter element that does data processing. The main parts or filter element are:

• sink: receiving incoming buffers (sinkpad)
• src: send outcoming buffers (srcpad)
• transform
  • inplace transform (transform_ip)
  • out buffer transform (transform)

Description

Plugin’s description is stored in gstmetadata field as tuple with the following fields:

__gstmetadata__ = (name,         # str
                   transform,    # str
                   description,  # str
                   author)       # str

Note: detailed explanation of each field

This description is displayed when user calls gst-inspect-1.0. For example:

Pads

General scheme of any filter element looks like the following:

Sink and Src are implementations of Gst.Pad. In Pads and Capabilities there is well defined meaning and functions of pads.

To initialize specific pad, – define Gst.PadTemplate first. Gst.PadTemplate describes pad’s name, direction (sink, src), presense (always, sometimes, request), caps. Have a look at the following code to define sink’s Gst.PadTemplate that accepts buffers in Red Green Blue colorspaces format and it’s variations.

# RGB colorspace variations
FORMATS = "{RGBx,BGRx,xRGB,xBGR,RGBA,BGRA,ARGB,ABGR,RGB,BGR}"

Gst.PadTemplate.new("sink",
                    Gst.PadDirection.SINK,
                    Gst.PadPresence.ALWAYS,
                    # Set to target format                                            
                    Gst.Caps.from_string(f"video/x-raw,format={FORMATS}"))

The, let do the same for src pad

Gst.PadTemplate.new("src",
                    Gst.PadDirection.SRC,
                    Gst.PadPresence.ALWAYS,
                    # Set to target format,                                            
                    Gst.Caps.from_string(f"video/x-raw,format={FORMATS}"))

In order to make pad templates visible for plugin, – just define gsttemplates field

__gsttemplates__ = (src_pad_template, sink_pad_template)

Otherwise you’ll get next CRITICAL error:

If everything OK, you should be able to get next information after running gst-inspect-1.0:

Transform

As buffer flows from sink to src we need to override do_transform_ip (for buffer processing in-place) or do_transform (for out-buffer processing and in-buffer remains unchanged) in order to do custom buffer processing and push it to src’s pad.

def do_transform_ip(self, buffer: Gst.Buffer) -> Gst.FlowReturn:
    # BUFFER processing 
    return Gst.FlowReturn.OK

In our case the main purpose of plugin is to blur image. So we need to convert Gst.Buffer to numpy array, make it writable (recap: How to make Gst.Buffer writable) and apply Gaussian Blur (using OpenCV) to it. Look how this could be implemented within do_transform_ip:

def do_transform_ip(self, buffer: Gst.Buffer) -> Gst.FlowReturn:
    try:
        # get sink's caps
        sink_caps = self.sinkpad.get_current_caps()

        # convert Gst.Buffer to np.ndarray
        image = gst_buffer_with_caps_to_ndarray(buffer, sink_caps)

        # apply gaussian blur to image
        image[:] = gaussian_blur(image, self.kernel_size, 
                                  sigma=(self.sigma_x, self.sigma_y))
    except Exception as e:
        logging.error(e)

    return Gst.FlowReturn.OK

Properties

Most of gstreamer plugins have properties that are being initialized by user when specifying commands in terminal. As every Gst.Element is derived from GObject we can easily use it’s __gproperties__ metadata to define own properties. It’s well described in official documentation.

So to create own property, – fill the __gproperties__ dictionary (property-name: tuple) with your own properties and it’s description. Common template is the following:

__gproperties__ = {"property-name": (type,  # GObject.TYPE_*
                                     short description, # str
                                     full desctiption,  # str
                                     min_value,  # any
                                     max_value,  # any
                                     default_value,  # any
                                     flags) # GObject.ParamFlags

Then properties for gaussian_blur plugin (sigmaX, sigmaY, kernel) could be defined by the following lines of code:

 __gproperties__ = {       
        "kernel": (GObject.TYPE_INT64,
                   "Kernel Size",
                   "Gaussian Kernel Size",
                   1,
                   GLib.MAXINT,
                   DEFAULT_KERNEL_SIZE,
                   GObject.ParamFlags.READWRITE
                   ),
    
        "sigmaX": (GObject.TYPE_FLOAT,
                   "Standart deviation in X",
                   "Gaussian kernel standard deviation in X direction",
                   1.0,
                   GLib.MAXFLOAT,
                   DEFAULT_SIGMA_X,
                   GObject.ParamFlags.READWRITE
                   ),

        "sigmaY": (GObject.TYPE_FLOAT,
                   "Standart deviation in Y",
                   "Gaussian kernel standard deviation in Y direction",
                   1.0,
                   GLib.MAXFLOAT,
                   DEFAULT_SIGMA_Y,
                   GObject.ParamFlags.READWRITE
                   ),
    }

To use own properties just override do_get_property method and include implementation for custom properties handle.

 def do_get_property(self, prop: GObject.GParamSpec):
    if prop.name == 'kernel':
        return self.kernel_size
    elif prop.name == 'sigmaX':
        return self.sigma_x
    elif prop.name == 'sigmaY':
        return self.sigma_y
    else:
        raise AttributeError('unknown property %s' % prop.name)

And override do_set_property as well:

def do_set_property(self, prop: GObject.GParamSpec, value):
    if prop.name == 'kernel':
        self.kernel_size = value
    elif prop.name == 'sigmaX':
        self.sigma_x = value
    elif prop.name == 'sigmaY':
        self.sigma_y = value
    else:
        raise AttributeError('unknown property %s' % prop.name)

With gst-inspect-1.0 defined properties above looks like the following:

Registration

In order to use gstreamer plugin from command line, just put next two lines in the end of the file:

GObject.type_register(class_type)
__gstelementfactory__ = (name,        # str
                         rank,        # Gst.Rank 
                         class_type)  # class type

For example, for gaussian_blur plugin previous code is going to be similar to the following:

GObject.type_register(GstGaussianBlur)  # register type
__gstelementfactory__ = ("gaussian_blur",
                         Gst.Rank.NONE, 
                         GstGaussianBlur)

Run examples

Get repository

First clone the repository and setup environment:

git clone https://github.com/jackersson/gst-python-plugins.git
cd gst-python-plugins

python3 -m venv venv
source venv/bin/activate
pip install -U wheel pip setuptools

pip install -r requirements.txt

Before running examples, export the following GST_PLUGIN_PATH, so gstreamer can locate newly created plugins:

export GST_PLUGIN_PATH=$GST_PLUGIN_PATH:$PWD/venv/lib/gstreamer-1.0/:$PWD/gst/

Note: Explanation for GST_PLUGIN_PATH export

• $GST_PLUGIN_PATH: previous path with plugin’s libs
• $PWD/venv/lib/gstreamer-1.0/: path were gst-python was installed (–prefix value) and contains libgstpython*.so
• $PWD/gst/: path with python/ directory which contains plugins implementation (*.py)

Launch simple command, to check how gstplugin_py works.

GST_DEBUG=python:6 gst-launch-1.0 videotestsrc ! gstplugin_py int-prop=100 float-prop=0.2 bool-prop=True str-prop="set" ! fakesink

Notice that the properties “int-prop“, “float-prop“, “bool-prop“, “str-prop“, “pyobject-prop” successfully set to values specified by command line.

Next, let check gaussian_blur plugin:

gst-launch-1.0 videotestsrc ! gaussian_blur kernel=9 sigmaX=5.0 sigmaY=5.0 ! videoconvert ! autovideosink

Note: To see the difference I used basic video mixing pipeline to put two streams (with/without blur) in one window (Look at gstreamer cheat sheet, to learn more about classic pipelines)

gst-launch-1.0 videomixer name=mixer ! videoconvert ! \
autovideosink videotestsrc ! \
video/x-raw,format=RGBA,width=1280,height=720 ! \
gaussian_blur kernel=9 sigmaX = 5.0 sigmaY=5.0 ! \
videobox left=-1280 ! mixer. videotestsrc ! \
video/x-raw,format=RGBA,width=1280,height=720 ! \
videobox left=0 ! mixer.

Or let us do the same with a video:

gst-launch-1.0 videomixer name=mixer ! videoconvert \
! autovideosink \
filesrc location=video.mp4 ! decodebin ! \
tee name=t ! queue ! videoconvert ! \
gaussian_blur kernel=9 sigmaX=5.0 sigmaY=5.0 ! \
videobox left=-1280 ! mixer. t. ! queue ! \
videobox left=0 ! mixer.

The other option could be to do the same but directly from youtube video using “How to watch Youtube videos with Gstreamer”.

Troubleshooting

1. Remove gstreamer’s cache to reload plugins and display error messages when there is a problem.

rm -rf ~/.cache/gstreamer-1.0/

2. Check GST_PLUGIN_PATH export as described above if you are receiving:

No such element or plugin 'myplugin'

Conclusion

With gstreamer python bindings you can easily implement plugins for image processing (gaussian_blur). We learned basic steps to implement filter plugin from GstBase.BaseTransform:

• description
• pads
• transform
• properties
• registration

Hope everything works as expected 😉 In case of troubles with running code leave comments or open an issue in Github.