How to write a GStreamer filter in Python

A guide to drawing bounding boxes using Cairo in a Python GStreamer element

Sep 14, 2024

Writing Gstreamer elements in Python, albeit the performance penalty, is quite easy. There might be use cases where you might have no choice but to write the element in Python. For example if you are dealing with an API that is available only in Python.

In this tutorial we will write the same element we wrote in C, one that draws bounding boxes on our video stream. Something that is useful in Computer Vision and Artificial Intelligence applications.

Let’s get started.

We start with the basic stucture of the element, see what it contains and start filling in functions. In essence all the GStreamer filter is a subclass of GstBaseTransform along with code to register it as a GStreamer element.

Something important to remember is that elements written in Python have to reside inside a directory called python/ in the current directory. The current directory is specified with GST_PLUGIN_PATH=.

Steps

Install dependencies
The template for the basic filter
Add signals
Add code in init, set_caps, transform_ip
Test

Step 1: Install dependencies

We need the package gst-python to be able to use elements written in Python.

sudo apt install gstreamer1.0-python3-plugin-loader
mkdir python
#Clone the repository
git clone https://www.github.com/mndar/writings/

This makes it possible to place the element in a directory named python

mkdir python
touch python/gstbasicoverlaypy.py

Step 2: The template for the basic filter

We start with the basic template for the filter. It contains code that sub-classes GstBaseTransform, adds pads and over rides set_caps and do_transform_ip. The most important function in transform_ip. This is where the GstBuffer is available for us to draw on.

# sudo apt install gstreamer1.0-python3-plugin-loader
import gi
import cairo

gi.require_version('Gst', '1.0')
gi.require_version('GstVideo', '1.0')
from gi.repository import Gst, GLib, GObject, GstVideo, GstBase

Gst.init(None)

class BasicOverlayPy(GstBase.BaseTransform):
    __gstmetadata__ = ('BasicOverlayPy','Filter', \
                      'Description', 'Name')

    __gsttemplates__ = (Gst.PadTemplate.new("src",
                                            Gst.PadDirection.SRC,
                                            Gst.PadPresence.ALWAYS,
                                            GstVideo.video_make_raw_caps([GstVideo.VideoFormat.BGRX, GstVideo.VideoFormat.BGRA, GstVideo.VideoFormat.RGB16])),
                        Gst.PadTemplate.new("sink",
                                            Gst.PadDirection.SINK,
                                            Gst.PadPresence.ALWAYS,
                                            GstVideo.video_make_raw_caps([GstVideo.VideoFormat.BGRX, GstVideo.VideoFormat.BGRA, GstVideo.VideoFormat.RGB16])))

    def __init__(self):
        GstBase.BaseTransform.__init__(self)
    
    def do_transform_ip(self, buf):
        return Gst.FlowReturn.OK
    
    def do_set_caps(self, icaps, ocaps):
        return True

GObject.type_register(BasicOverlayPy)
__gstelementfactory__ = ("basicoverlaypy", Gst.Rank.NONE, BasicOverlayPy)

Step 3: Add Signals

We will add two signals addbox and removebox and emit them from our test application. The signals are addded by filling in an array __gsignals__. Below is the code snippet. The documentation to refer to is https://lazka.github.io/pgi-docs/.

   __gsignals__ = {
        "addbox": (GObject.SignalFlags.RUN_LAST|GObject.SignalFlags.ACTION, GObject.TYPE_NONE, [GObject.TYPE_STRING, GObject.TYPE_STRING, GObject.TYPE_INT, GObject.TYPE_INT, GObject.TYPE_INT, GObject.TYPE_INT]),
        "removebox": (GObject.SignalFlags.RUN_LAST|GObject.SignalFlags.ACTION, GObject.TYPE_NONE, [GObject.TYPE_STRING,]),
    }

Step 4: Add code in init, set_caps, transform_ip

We have to fill in 3 functions init, set_caps and transform_ip. init is the constructor were will initialize variables. set_caps is called when the video stream starts and transform_ip is for drawing on th3e GStreamer buffer. The code snippets for the functions are

def __init__(self):
        GstBase.BaseTransform.__init__(self)
        self.boxlist = []
        self.connect("addbox", BasicOverlayPy.on_addbox)
        self.connect("removebox", BasicOverlayPy.on_removebox)

def do_transform_ip(self, buf):
        result, mapinfo = buf.map(Gst.MapFlags.READ|Gst.MapFlags.WRITE)
        
        # create cairo surface
        surface = cairo.ImageSurface.create_for_data(mapinfo.data, cairo.Format.RGB24, self.width, self.height, cairo.Format.RGB24.stride_for_width(self.width))
        # create cairo context
        ctx= cairo.Context(surface)
        # cairo scale
        ctx.scale(self.width, self.height)
        for box in self.boxlist:
            BasicOverlayPy.draw_box(self, ctx, box)

        del ctx
        del surface
        buf.unmap(mapinfo)
        return Gst.FlowReturn.OK

 def do_set_caps(self, icaps, ocaps):
        ret, self.info = GstVideo.video_info_from_caps (icaps)
        print(str(self.info.width)+"x"+str(self.info.height))
        self.width = self.info.width
        self.height = self.info.height;
        return True

Note: Code to the actually drawing of the box is show included above. The element code in the respository contains it. The cairo documentation is avaialble here https://pycairo.readthedocs.io/en/latest/

Step 5: Test

If you have clone my writings repository, you can run the test application that adds two bounding boxes with a 2 second delay and removes one of them with a 6 second delay. Pay close attention to the layout of the directory. The Python element is places in the directory python and not in the current directory.

We’ll run gst-inspect-1.0 to confirm that everything checkout about the element.

$ git clone https://www.github.com/mndar/writings
$ cd writings/filter_py
$ ls
python  testelement.py
$ ls python/
gstbasicoverlaypy.py

$ GST_PLUGIN_PATH=. gst-inspect-1.0 basicoverlaypy
Factory Details:
  Rank                     none (0)
  Long-name                BasicOverlayPy
  Klass                    Filter
  Description              Draw Bounding Boxes
  Author                   Mandar Joshi

Plugin Details:
  Name                     python
  Description              loader for plugins written in python
  Filename                 /usr/lib/x86_64-linux-gnu/gstreamer-1.0/libgstpython.so
  Version                  1.20.1
  License                  LGPL
  Source module            gst-python
  Binary package           GStreamer Python
  Origin URL               http://gstreamer.freedesktop.org

GObject
 +----GInitiallyUnowned
       +----GstObject
             +----GstElement
                   +----GstBaseTransform
                         +----gstbasicoverlaypy+BasicOverlayPy

Pad Templates:
  SINK template: 'sink'
    Availability: Always
    Capabilities:
      video/x-raw
                  width: [ 1, 2147483647 ]
                 height: [ 1, 2147483647 ]
              framerate: [ 0/1, 2147483647/1 ]
                 format: { (string)BGRx, (string)BGRA, (string)RGB16 }
  
  SRC template: 'src'
    Availability: Always
    Capabilities:
      video/x-raw
                  width: [ 1, 2147483647 ]
                 height: [ 1, 2147483647 ]
              framerate: [ 0/1, 2147483647/1 ]
                 format: { (string)BGRx, (string)BGRA, (string)RGB16 }

Element has no clocking capabilities.
Element has no URI handling capabilities.

Pads:
  SINK: 'sink'
    Pad Template: 'sink'
  SRC: 'src'
    Pad Template: 'src'

Element Properties:
  name                : The name of the object
                        flags: readable, writable, 0x2000
                        String. Default: "gstbasicoverlaypy+basicoverlaypy0"
  parent              : The parent of the object
                        flags: readable, writable, 0x2000
                        Object of type "GstObject"
  qos                 : Handle Quality-of-Service events
                        flags: readable, writable
                        Boolean. Default: false

Element Actions:
  "addbox" :  void user_function (GstElement* object,
                                  gchararray arg0,
                                  gchararray arg1,
                                  gint arg2,
                                  gint arg3,
                                  gint arg4,
                                  gint arg5);
  "removebox" :  void user_function (GstElement* object,
                                     gchararray arg0);

$ GST_PLUGIN_PATH=. python3 testelement.py

Conclusion

There you have. A GStreamer element in Python. Elements of other types follow the same steps. You have to subclass a base class, add pads, over ride functions and register the element. That’s it for this tutorial. See you in the next one.