Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/usr/bin/python3
#-*- coding: utf-8 -*-
import logging
import timeit
import traceback
import time
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GObject, GLib
GObject.threads_init()
Gst.init(None)
class GstPluginPy(Gst.Element):
__gstmeta__ = ("gstplugin_py",
"Gst Plugin Python Implementation",
"gst.Element wraps processing model written in Python",
"DataAI")
__gstmetadata__ = __gstmeta__
_srctemplate = Gst.PadTemplate.new('src', Gst.PadDirection.SRC,
Gst.PadPresence.ALWAYS,
Gst.Caps.from_string("video/x-raw,format=RGB"))
_sinktemplate = Gst.PadTemplate.new('sink', Gst.PadDirection.SINK,
Gst.PadPresence.ALWAYS,
Gst.Caps.from_string("video/x-raw,format=RGB"))
__gsttemplates__ = (_srctemplate, _sinktemplate)
__gproperties__ = {
"model": (GObject.TYPE_PYOBJECT,
"model",
"Contains model that implements IDataTransform",
GObject.ParamFlags.READWRITE)
}
def __init__(self):
Gst.Element.__init__(self)
self.sinkpad = Gst.Pad.new_from_template(self._sinktemplate, 'sink')
self.sinkpad.set_chain_function_full(self.chainfunc, None)
self.sinkpad.set_event_function_full(self.eventfunc, None)
self.add_pad(self.sinkpad)
self.srcpad = Gst.Pad.new_from_template(self._srctemplate, 'src')
self.srcpad.set_event_function_full(self.srceventfunc, None)
self.srcpad.set_query_function_full(self.srcqueryfunc, None)
self.add_pad(self.srcpad)
self.model = None
def chainfunc(self, pad, parent, buffer):
try:
if self.model is not None:
item = {
"pad": pad,
"buffer": buffer,
"timeout": 0.01
}
self.model.process(**item)
except Exception as e:
logging.error(e)
traceback.print_exc()
return self.srcpad.push(buffer)
def do_get_property(self, prop):
if prop.name == 'model':
return self.model
else:
raise AttributeError('unknown property %s' % prop.name)
def do_set_property(self, prop, value):
if prop.name == 'model':
self.model = value
else:
raise AttributeError('unknown property %s' % prop.name)
def eventfunc(self, pad, parent, event):
return self.srcpad.push_event(event)
def srcqueryfunc(self, pad, object, query):
return self.sinkpad.query(query)
def srceventfunc(self, pad, parent, event):
return self.sinkpad.push_event(event)
def register(class_info):
def init(plugin, plugin_impl, plugin_name):
type_to_register = GObject.type_register(plugin_impl)
return Gst.Element.register(plugin, plugin_name, 0, type_to_register)
# Parameters explanation
# https://lazka.github.io/pgi-docs/Gst-1.0/classes/Plugin.html#Gst.Plugin.register_static
version = '14.1'
gstlicense = 'LGPL'
origin = ''
source = class_info.__gstmeta__[1]
package = class_info.__gstmeta__[0]
name = class_info.__gstmeta__[0]
description = class_info.__gstmeta__[2]
init_function = lambda plugin : init(plugin, class_info, name)
if not Gst.Plugin.register_static(Gst.VERSION_MAJOR, Gst.VERSION_MINOR,
name, description,
init_function, version, gstlicense,
source, package, origin):
raise ImportError("Plugin {} not registered".format(name))
return True
register(GstPluginPy)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.