custom gst-python plugin is not working in python app
I have created a custom gst-python plugin for blurring the incoming frame.
This is the plugin code for my blur plugin:
import gi
gi.require_version("Gst", "1.0")
gi.require_version("GstBase", "1.0")
from gi.repository import Gst, GObject, GstBase, GLib
import numpy as np
import cv2
Gst.init(None)
DEFAULT_KERNEL_SIZE = 11
class Blur(GstBase.BaseTransform):
"""
Gstreamer Python Plugin for blurring incoming video buffers.
Args:
GstBase.BaseTransform: The base class for the plugin.
Methods:
do_get_property: Retrieves the value of a specified property.
do_set_property: Sets the value of a specified property.
do_set_caps: Extracts information from the input capabilities.
do_transform_ip: Implements the functionality for blurring the input GstBuffer.
"""
GST_PLUGIN_NAME = "blur"
# Plugin’s description is stored in gstmetadata field as tuple
# gst-inspect-1.0 blur
__gstmetadata__ = (
"Blur Plugin", # Long-name
"Blur Plugin", # Klass
"Plugin for blurring incoming video buffers", # Description
"vishal" # Author
"any queries @ vishalkmr01123@gmail.com", # Author
)
# Pad Template Definition
# Used for defining the capabilities of the inputs and outputs.
# Description of a pad that the element will create and use. it contains:
# - A short name for the pad.
# - Pad direction.
# - Existence property. This indicates whether the pad exists always
# (an "always" pad), only in some cases (a "sometimes" pad) or only if
# the application requested such a pad (a "request" pad).
# - Supported types/formats by this element (capabilities).
FORMATS = "{RGB,BGR}" # Allow only these format
src_format = Gst.Caps.from_string(f"video/x-raw,format={FORMATS}")
sink_format = Gst.Caps.from_string(f"video/x-raw,format={FORMATS}")
src_pad_template = Gst.PadTemplate.new(
"src",
Gst.PadDirection.SRC,
Gst.PadPresence.ALWAYS,
src_format
)
sink_pad_template = Gst.PadTemplate.new(
"sink",
Gst.PadDirection.SINK,
Gst.PadPresence.ALWAYS,
sink_format
)
__gsttemplates__ = (src_pad_template, sink_pad_template)
# Installing various properties for the plugin
__gproperties__ = {
"kernel-size": (int, # type
"Kernel Size", # nick
"Gaussian Kernel Size", # blurb
1, # min
GLib.MAXINT, # max
DEFAULT_KERNEL_SIZE, # default
GObject.ParamFlags.READWRITE # flags
),
}
def __init__(self):
"""
Initializes the Blur class object.
This is common to all instances of the Blur class.
Used to initialise the class only once.
"""
# Calling the __init__() method of the parent class.
super().__init__()
# Set default values for properties
self.kernel_size= DEFAULT_KERNEL_SIZE
def do_get_property(self, prop: GObject.ParamSpec) -> any:
"""
Retrieves the value of the specified property.
Args:
prop (GObject.ParamSpec): The property object containing information
about the property name, type, and flags.
Returns:
The value of the requested property.
Raises:
AttributeError: If the provided property name does not match any
known properties.
"""
if prop.name == 'kernel-size':
return self.kernel_size
else:
raise AttributeError(f"Unknown Property {prop.name}")
def do_set_property(self, prop: GObject.ParamSpec, value: any) -> None:
"""
Set the values of the properties in the GstBlink class based on
the provided property name and value.
Args:
prop (GObject.ParamSpec): The property object containing information
about the property name, type, and flags.
value (any): The value to be set for the property.
Raises:
AttributeError: If the name does not match any known properties.
Returns:
None
"""
if prop.name == 'kernel-size':
# Ensuring the kernel size is odd
if value%2 == 0:
self.kernel_size = value + 1
else:
self.kernel_size = value
else:
raise AttributeError(f"Unknown Property {prop.name}")
def do_set_caps(self, incaps: Gst.Caps, outcaps: Gst.Caps) -> bool:
"""
Extracts information about the width, height, format, framerate, pixel
aspect ratio, interlace mode, colorimetry, and chroma site from the
input caps.
Args:
incaps (Gst.Caps): The input capabilities (formats) of the pad.
outcaps (Gst.Caps): The output capabilities (formats) of the pad.
Returns:
bool
"""
struct = incaps.get_structure(0)
self.width = struct.get_int("width").value
self.height = struct.get_int("height").value
self.format = struct.get_string("format")
self.framerate = struct.get_fraction("framerate")
self.pixel_aspect_ratio = struct.get_fraction("pixel-aspect-ratio")
self.interlace_mode = struct.get_string("interlace-mode")
self.colorimetry = struct.get_string("colorimetry")
self.chroma_site = struct.get_string("chroma-site")
caps_string="\n"
caps_string += "#"*100
caps_string += f"\nFormat: {self.format}"
caps_string += f"\nHeight: {self.height}"
caps_string += f"\nWidth: {self.width}"
caps_string += f"\nFramerate: {self.framerate}"
caps_string += f"\nPixel Aspect Ratio: {self.pixel_aspect_ratio}"
caps_string += f"\nInterlace Mode: {self.interlace_mode}"
caps_string += f"\nColorimetry: {self.colorimetry}"
caps_string += f"\nChroma Site: {self.chroma_site}\n"
caps_string += "#"*100
Gst.info(caps_string)
return True
def do_transform_ip(self, gst_buffer: Gst.Buffer) -> Gst.FlowReturn:
"""
This method implements the functionality for blurring the input GstBuffer.
Changes can be made to the input GstBuffer directly (inplace) to
obtain the output GstBuffer.
Args:
gst_buffer (Gst.Buffer): The input buffer to be processed.
Returns:
Gst.FlowReturn.OK: If the buffer is successfully processed.
Gst.FlowReturn.ERROR: If mapping error is encountered.
"""
try:
with gst_buffer.map(Gst.MapFlags.READ | Gst.MapFlags.WRITE) as info:
# Fetch the RGB/BGR image
image = np.ndarray((self.height, self.width, 3), buffer=info.data, dtype=np.uint8)
# Blur RGB/BGR frame
image[:,:,:] = cv2.GaussianBlur(image, (self.kernel_size, self.kernel_size), 0)
return Gst.FlowReturn.OK
except Gst.MapError as e:
Gst.error("Mapping error: %s" % e)
return Gst.FlowReturn.ERROR
# Register the element factories and other features
# The value of this attribute should be a tuple consisting:
# - Factory-name for the plugin
# - Plugin Rank
# - Class that implements the element.
GObject.type_register(Blur)
__gstelementfactory__ = (
Blur.GST_PLUGIN_NAME,
Gst.Rank.PRIMARY,
Blur
)
The plugin can be used to blur the whole incoming buffer if used in the below pipeline:
gst-launch-1.0 filesrc location=sample.mp4 ! qtdemux ! h264parse ! avdec_h264 ! videoconvert ! blur kernel-size=50 ! videoconvert ! autovideosink
Expected Behavior:
The created blur plugin can be used as follows:
- inside a gst-launch pipeline.
- inside a C-based GStreamer app where a plugin element is created using gst_element_factory_make
- inside a Python-based GStreamer app where Gst.parse_launch(pipeline_str) is used to create the pipeline
- inside a Python-based GStreamer app where gst elements are created using Gst.ElementFactory.make. Then we add/link them to the pipeline.
Observed Behavior:
My plugin is working fine in the below scenario where the incoming video buffers are blurred:
- When the plugin is used in the gst-launch pipeline.
- When the plugin element is created using gst_element_factory_make inside a C-based GStreamer app.
- When the plugin is used inside a Python app where Gst.parse_launch(pipeline_str) is used to create the pipeline
However when i try to create the gst-python plugin blur using Gst.ElementFactory.make and add it to the pipeline then the plugin is not blurring the incoming buffer. Although no error was seen on the console seems like transform_ip function is not getting called.
Here is the code for my Python-based gstreamer app where i am creating and adding blur plugin into pipeline
import sys
import gi
gi.require_version('Gst', '1.0')
from gi.repository import GLib, Gst
Gst.init(None)
def bus_call(bus, message, loop):
t = message.type
if t == Gst.MessageType.EOS:
sys.stdout.write("End-of-stream\n")
loop.quit()
elif t==Gst.MessageType.WARNING:
err, debug = message.parse_warning()
sys.stderr.write("Warning: %s: %s\n" % (err, debug))
elif t == Gst.MessageType.ERROR:
err, debug = message.parse_error()
sys.stderr.write("Error: %s: %s\n" % (err, debug))
loop.quit()
return True
def demuxer_pad_added(demuxer, pad, element):
"""
Link element to qtdemux dynamically
"""
if pad.name == 'video_0':
demuxer.link(element)
# Create a pipeline
pipeline = Gst.Pipeline()
# Create elements
source = Gst.ElementFactory.make("filesrc", "source")
source.set_property("location", "1.mp4")
qtdemux = Gst.ElementFactory.make("qtdemux", "qtdemux")
h264parse = Gst.ElementFactory.make("h264parse", "h264parse")
avdec_h264 = Gst.ElementFactory.make("avdec_h264", "avdec_h264")
videoconvert = Gst.ElementFactory.make("videoconvert", "videoconvert")
custom_plugin = Gst.ElementFactory.make("blur", "blur")
custom_plugin.set_property('kernel-size', 50)
videoconvert1 = Gst.ElementFactory.make("videoconvert", "videoconvert1")
sink = Gst.ElementFactory.make("autovideosink", "sink")
# Add elements to the pipeline
pipeline.add(source)
pipeline.add(qtdemux)
pipeline.add(h264parse)
pipeline.add(avdec_h264)
pipeline.add(videoconvert)
pipeline.add(custom_plugin)
pipeline.add(videoconvert1)
pipeline.add(sink)
# Link elements in the pipeline
if not source.link(qtdemux):
print("Failed to link source and qtdemux elements")
exit(1)
if not h264parse.link(avdec_h264):
print("Failed to link h264parse and avdec_h264 elements")
exit(1)
if not avdec_h264.link(videoconvert):
print("Failed to link avdec_h264 and videoconvert elements")
exit(1)
if not videoconvert.link(custom_plugin):
print("Failed to link videoconvert and custom_plugin elements")
exit(1)
if not custom_plugin.link(videoconvert1):
print("Failed to link custom_plugin and videoconvert1 elements")
exit(1)
if not videoconvert1.link(sink):
print("Failed to link videoconvert1 and sink elements")
exit(1)
qtdemux.connect("pad-added", demuxer_pad_added, h264parse)
# Start the main loop
loop = GLib.MainLoop()
bus = pipeline.get_bus()
bus.add_signal_watch()
bus.connect ("message", bus_call, loop)
pipeline.set_state(Gst.State.PLAYING)
try:
loop.run()
except KeyboardInterrupt:
pass
# Stop the pipeline and cleanup
pipeline.set_state(Gst.State.NULL)
The blur plugin is recognised by gst-inspect and works fine on gst-launch.
Debugging shows that when i create an element in the Python app Gst.ElementFactory.make("blur", "blur") then only __init__ & do_set_property methods are called. Methods do_set_caps & tranform_ip are not getting executed.
Setup
- Operating System: ubuntu 22.04
- Device: Computer
- GStreamer Version: 1.20.3
My doubt is why blur (custom gst-python plugin) is not working well with my Python-based gstreamer app, although it is working fine on gst-launch and some other scenarios i mentioned above.