mirror of
https://gitlab.freedesktop.org/gstreamer/gstreamer.git
synced 2024-06-01 07:08:12 +00:00
139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
#!/usr/bin/env python
|
|
|
|
import sys
|
|
import traceback
|
|
from math import pi
|
|
|
|
import pygtk
|
|
pygtk.require ("2.0")
|
|
import gobject
|
|
gobject.threads_init()
|
|
|
|
import pygst
|
|
pygst.require('0.10')
|
|
import gst
|
|
|
|
import cairo
|
|
|
|
WIDTH, HEIGHT = 640, 480
|
|
FRAMES = 300
|
|
FRAMERATE = 15
|
|
|
|
class PyGstBufferDraw(gst.Element):
|
|
_sinkpadtemplate = gst.PadTemplate ("sink",
|
|
gst.PAD_SINK,
|
|
gst.PAD_ALWAYS,
|
|
gst.caps_from_string ("video/x-raw-rgb,bpp=32,depth=32,blue_mask=-16777216,green_mask=16711680, red_mask=65280, alpha_mask=255,width=[ 1, 2147483647 ],height=[ 1, 2147483647 ],framerate=[ 0/1, 2147483647/1 ]"))
|
|
_srcpadtemplate = gst.PadTemplate ("src",
|
|
gst.PAD_SRC,
|
|
gst.PAD_ALWAYS,
|
|
gst.caps_from_string ("video/x-raw-rgb,bpp=32,depth=32,blue_mask=-16777216,green_mask=16711680, red_mask=65280, alpha_mask=255,width=[ 1, 2147483647 ],height=[ 1, 2147483647 ],framerate=[ 0/1, 2147483647/1 ]"))
|
|
|
|
def __init__(self):
|
|
gst.Element.__init__(self)
|
|
|
|
self.sinkpad = gst.Pad(self._sinkpadtemplate, "sink")
|
|
self.sinkpad.set_chain_function(self.chainfunc)
|
|
self.sinkpad.set_event_function(self.eventfunc)
|
|
self.sinkpad.set_getcaps_function(gst.Pad.proxy_getcaps)
|
|
self.sinkpad.set_setcaps_function(gst.Pad.proxy_setcaps)
|
|
self.add_pad (self.sinkpad)
|
|
|
|
self.srcpad = gst.Pad(self._srcpadtemplate, "src")
|
|
|
|
self.srcpad.set_event_function(self.srceventfunc)
|
|
self.srcpad.set_query_function(self.srcqueryfunc)
|
|
self.srcpad.set_getcaps_function(gst.Pad.proxy_getcaps)
|
|
self.srcpad.set_setcaps_function(gst.Pad.proxy_setcaps)
|
|
self.add_pad (self.srcpad)
|
|
|
|
def chainfunc(self, pad, buffer):
|
|
try:
|
|
outbuf = buffer.copy_on_write ()
|
|
self.draw_on (outbuf)
|
|
return self.srcpad.push (outbuf)
|
|
except:
|
|
return GST_FLOW_ERROR
|
|
|
|
def eventfunc(self, pad, event):
|
|
return self.srcpad.push_event (event)
|
|
|
|
def srcqueryfunc (self, pad, query):
|
|
return self.sinkpad.query (query)
|
|
def srceventfunc (self, pad, event):
|
|
return self.sinkpad.push_event (event)
|
|
|
|
def draw_on (self, buf):
|
|
try:
|
|
caps = buf.get_caps()
|
|
width = caps[0]['width']
|
|
height = caps[0]['height']
|
|
framerate = caps[0]['framerate']
|
|
surface = cairo.ImageSurface.create_for_data (buf, cairo.FORMAT_ARGB32, width, height, 4 * width)
|
|
ctx = cairo.Context(surface)
|
|
except:
|
|
print "Failed to create cairo surface for buffer"
|
|
traceback.print_exc()
|
|
return
|
|
|
|
try:
|
|
center_x = width/4
|
|
center_y = 3*height/4
|
|
|
|
# draw a circle
|
|
radius = float (min (width, height)) * 0.25
|
|
ctx.set_source_rgba (0.0, 0.0, 0.0, 0.9)
|
|
ctx.move_to (center_x, center_y)
|
|
ctx.arc (center_x, center_y, radius, 0, 2.0*pi)
|
|
ctx.close_path()
|
|
ctx.fill()
|
|
ctx.set_source_rgba (1.0, 1.0, 1.0, 1.0)
|
|
ctx.set_font_size(0.3 * radius)
|
|
txt = "Hello World"
|
|
extents = ctx.text_extents (txt)
|
|
ctx.move_to(center_x - extents[2]/2, center_y + extents[3]/2)
|
|
ctx.text_path(txt)
|
|
ctx.fill()
|
|
|
|
except:
|
|
print "Failed cairo render"
|
|
traceback.print_exc()
|
|
|
|
gobject.type_register(PyGstBufferDraw)
|
|
|
|
pipe = gst.Pipeline()
|
|
vt = gst.element_factory_make ("videotestsrc")
|
|
cf = gst.element_factory_make ("capsfilter")
|
|
c1 = PyGstBufferDraw()
|
|
color = gst.element_factory_make ("ffmpegcolorspace")
|
|
scale = gst.element_factory_make ("videoscale")
|
|
q1 = gst.element_factory_make ("queue")
|
|
sink = gst.element_factory_make ("autovideosink")
|
|
|
|
caps = gst.caps_from_string ("video/x-raw-rgb,width=%d,height=%d,framerate=%d/1" % (WIDTH, HEIGHT, FRAMERATE))
|
|
cf.set_property ("caps", caps)
|
|
|
|
vt.set_property ("num-buffers", FRAMES)
|
|
|
|
pipe.add (vt, cf, c1, q1, color, scale, sink)
|
|
gst.element_link_many (vt, cf, c1, q1, color, scale, sink)
|
|
|
|
def on_eos (bus, msg):
|
|
mainloop.quit()
|
|
|
|
bus = pipe.get_bus()
|
|
bus.add_signal_watch()
|
|
bus.connect('message::eos', on_eos)
|
|
|
|
pipe.set_state (gst.STATE_PLAYING)
|
|
|
|
mainloop = gobject.MainLoop()
|
|
try:
|
|
mainloop.run()
|
|
except:
|
|
pass
|
|
|
|
pipe.set_state (gst.STATE_NULL)
|
|
pipe.get_state (gst.CLOCK_TIME_NONE)
|
|
|