diff -r 000000000000 -r 367d791aeb57 gmyth-stream/server/0.1/plugins/media/gstreamer.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/gmyth-stream/server/0.1/plugins/media/gstreamer.py Thu May 17 14:07:27 2007 +0100 @@ -0,0 +1,290 @@ +#vim:ts=4:sw=4:et +import pygst +pygst.require("0.10") +import gst +import gobject +import socket +import time +from threading import Thread + +class Media: + class StreamListener(Thread): + def __init__ (self, stream_data): + Thread.__init__(self) + self.stream = stream_data + print "Thread Created" + + def run (self): + #Create socket + print "Waiting connection" + self.stream.Socket.listen(1) + self.stream.Connection, self.stream.Addr = self.stream.Socket.accept () + print "Connection requested" + self.stream.Sink.set_property ("fd", self.stream.Connection.fileno()) + self.stream.Pipe.set_state(gst.STATE_PLAYING) + print "PLAYING" + + + class StreamData: + stream_count = 0 + + def __init__ (self, pipe, abin, vbin, sink): + self.stream_count += 1 + self.Id = self.stream_count + self.Pipe = pipe + self.Abin = abin + self.Vbin = vbin + self.Sink = sink + self.Loop = gobject.MainLoop() + self.ACaps = "" + self.VCaps = "" + self.Ready = False + self.Socket = None + self.Connection = None + self.Addr = None + + def __init__(self, config): + # set gstreamer basic options + self.config = config + self.streams = [] + self.socket = None + self.connection = None + self.addr = None + self.ready = False + self.current = None + + + def setup(self, uri, mux, vcodec, vbitrate, + fps, acodec, abitrate, width, height, port, options): + + ## Pipelines + pipe = gst.Pipeline () + print "Opening Uri:" + uri + src = gst.element_make_from_uri (gst.URI_SRC, uri, "src") + #src = gst.element_factory_make ("gnomevfssrc", "src") + src.set_property ("location", uri) + if (src is None): + print "Fail to create src element" + return None + + print ("Create source") + decode = gst.element_factory_make ("decodebin", "decode") + if (decode is None): + print "Fail to create decodebin" + return None + + print ("Create source") + mux = gst.element_factory_make ("avimux", "mux") + if (mux is None): + print "Fail to create mux" + return None + + sink = gst.element_factory_make ("fdsink", "sink") + if (sink is None): + print "Fail to create fdsink" + return None + + print ("Create source") + + #video encode + #queue ! videoscale ! video/x-raw-yuv,width=240,height=144 ! videorate ! ffenc_h263p bitrate=256000 me-method=2 ! rtph263ppay ! udpsink host=224.0.0.1 port=5000 + vbin = gst.Bin () + vqueue = gst.element_factory_make ("queue", "vqueue") + colorspace = gst.element_factory_make ("ffmpegcolorspace", "") + vrate = gst.element_factory_make ("videorate", "vrate") + vencode = gst.element_factory_make ("ffenc_mpeg4", "vencode") + #vencode = gst.element_factory_make ("ffenc_msmpeg4v1", "vencode") + vqueue_src = gst.element_factory_make ("queue", "vqueue_src") + + #if (int(vbitrate) > 0): + vencode.set_property ("bitrate", 200) + #vencode.set_property ("quant-type", 1) + vencode.set_property ("pass", 2) + vencode.set_property ("quantizer", 5) + #vencode.set_property ("me-method", 1) + + + if (None in [vbin, vqueue, vrate, vencode, vqueue_src]): + print "Fail to create video encode elements." + return None + + vbin.add (vqueue) + if ((int(width) > 0) and (int(height) > 0)): + print ("formating output to %d / %d" % (int(width), int(height))) + + vscale = gst.element_factory_make ("ffvideoscale", "vscale") + + vbin.add (vscale); + if (not vqueue.link (vscale)): + print "Fail to link video elements" + return None + + vbin.add (colorspace) + + if (not vscale.link (colorspace, \ + gst.caps_from_string ("video/x-raw-yuv,width=(int)%d,height=(int)%d" % (int(width), int(height))))): + print "Fail to link video elements" + return None + else: + vbin.add (colorspace) + vqueue.link (colorspace) + + vbin.add (vrate, vencode, vqueue_src) + if (not colorspace.link (vrate)): + print "Fail to colorspace with vrate" + return None + + + if (not vrate.link (vencode, \ + gst.caps_from_string ("video/x-raw-yuv,framerate=(fraction)10/1"))): + print "Fail to link vrate element" + return None + + if (not vencode.link (vqueue_src)): + print "Fail to link video encode with queue" + return None + + vbin.add_pad (gst.GhostPad ("sink", vqueue.get_pad ("sink"))) + vbin.add_pad (gst.GhostPad ("src", vqueue_src.get_pad ("src"))) + + #audio encode + #audio/x-raw-int ! queue ! audioconvert ! faac ! rtpmp4gpay ! udpsink name=upd_audio host=224.0.0.1 port=5002 + abin = gst.Bin () + aqueue = gst.element_factory_make ("queue", "aqueue") + aconvert = gst.element_factory_make ("audioconvert", "aconvert") + arate = gst.element_factory_make ("audioresample", "arate") + #aencode = gst.element_factory_make ("ffenc_ac3", "aencode") + aencode = gst.element_factory_make ("queue", "aencode") + #aencode = gst.element_factory_make ("lame", "aencode") + #aencode = gst.element_factory_make ("ffenc_mp2", "aencode") + aqueue_src = gst.element_factory_make ("queue", "aqueue_src") + + if (None in [abin, aqueue, arate, aencode, aqueue_src]): + print "Fail to create video encode elements." + return None + + abin.add (aqueue, aconvert, arate, aencode, aqueue_src) + if (not gst.element_link_many (aqueue, aconvert, arate, aencode, aqueue_src)): + print "Fail to link video elements" + return None + + abin.add_pad (gst.GhostPad ("sink", aqueue.get_pad ("sink"))) + abin.add_pad (gst.GhostPad ("src", aqueue_src.get_pad ("src"))) + + #Finish Pipeline + pipe.add (src, decode, abin, vbin, mux, sink) + gst.element_link_many (src, decode) + gst.element_link_many (mux, sink) + + #Linking decode with mux + mux_audio = mux.get_pad ("audio_0") + mux_video = mux.get_pad ("video_0") + + audio_pad = abin.get_pad ("src") + video_pad = vbin.get_pad ("src") + + if (audio_pad.link (mux_audio) != gst.PAD_LINK_OK): + print "Fail to link audio with mux" + return None + + if (video_pad.link (mux_video) != gst.PAD_LINK_OK): + print "Fail to link audio with mux" + return None + + stream_data = self.StreamData (pipe, abin, vbin, sink) + bus = pipe.get_bus() + bus.add_signal_watch() + bus.connect ("message", self.__on_bus_message, stream_data) + + decode.connect("new-decoded-pad", self.__on_decode_new_pad, stream_data) + decode.connect("unknown-type", self.__on_decode_unknown_type, stream_data) + + print ("Create source") + pipe.set_state (gst.STATE_PAUSED) + print "Running Pipe" + stream_data.Loop.run () + print "End run" + + + #Create socket + stream_data.Socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + print "Bind on port %d" % port + stream_data.Socket.bind(('', int (port))) + self.streams.append (stream_data) + return (True, "") + + def play(self): + print "Play" + stream = self.streams[0] + self.current = self.StreamListener(stream) + self.current.start () + time.sleep (1) + return (True, "") + + def stop(self): + self.current.join () + self.current = None + stream = self.streams[0] + stream.Pipe.set_state(gst.STATE_NULL) + del (stream.Pipe) + stream.Pipe = None + stream.Abin = None + stream.Vbin = None + stream.Sink = None + if (stream.Connection != None): + stream.Connection.close () + + self.streams = [] + time.sleep (5) + return (True, "") + + + def __on_bus_message (self, bus, message, stream_data): + + t = message.type + if (t == gst.MESSAGE_STATE_CHANGED): + oldstate = -1 + newstate = -1 + pending = -1 + oldstate, newstate, pending = message.parse_state_changed () + if ((oldstate == gst.STATE_READY) and \ + (newstate == gst.STATE_PAUSED) and \ + (pending == gst.STATE_VOID_PENDING) and \ + (stream_data.Ready == False)): + state_changed_status, current_state, pending_state = stream_data.Pipe.get_state () + if ((current_state == gst.STATE_PAUSED) and \ + (pending_state == gst.STATE_VOID_PENDING)): + print "Pipe paused" + stream_data.Loop.quit () + stream_data.Ready = True + elif (t == gst.MESSAGE_ERROR): + err, debug = message.parse_error() + print "Error: %s" % err, debug + stream_data.Loop.quit () + stream_data.Ready = False + + return True + + def __on_decode_unknown_type (self, decode, pad, caps, stream_data): + + print "Unknown Type" + return None + + def __on_decode_new_pad (self, decode, pad, arg1, stream_data): + + caps = pad.get_caps().to_string() + print "New pad " + caps + if (caps.rfind ("audio") != -1): + apad = stream_data.Abin.get_pad ("sink") + if (pad.link (apad) != gst.PAD_LINK_OK): + print "Error on link audio pad" + return None + elif (caps.rfind ("video") != -1): + vpad = stream_data.Vbin.get_pad ("sink") + if (pad.link (vpad) != gst.PAD_LINK_OK): + print "Error on link video pad" + return None + else: + print "Invalid caps" + print "Linked" +