# HG changeset patch # User renatofilho # Date 1175612700 -3600 # Node ID 8edb3b445cacb59d7769cc8e5c02421fc43ec1ca # Parent bf3ac902013e6c4b86b6e1662671a6b18a9e5357 [svn r488] created gstreamer media plugin diff -r bf3ac902013e -r 8edb3b445cac gmyth-stream/plugins/media/gstreamer-rtp.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/gmyth-stream/plugins/media/gstreamer-rtp.py Tue Apr 03 16:05:00 2007 +0100 @@ -0,0 +1,218 @@ +import pygst +pygst.require("0.10") +import gst +import gobject + +class Media: + class StreamData: + stream_count = 0 + + def __init__ (self, pipe, abin, vbin): + + self.stream_count += 1 + self.Id = self.stream_count + self.Pipe = pipe + self.Abin = abin + self.Vbin = vbin + self.Loop = gobject.MainLoop() + self.ACaps = "" + self.VCaps = "" + self.Ready = False + + + def __init__(self, config): + # set gstreamer basic options + self.config = config + self.pipe = None + self.streams = [] + + + def setup(self, filename, mux, vcodec, vbitrate, + fps, acodec, abitrate, width, height, port, options): + + ## Pipelines + self.pipe = gst.Pipeline () + uri = "file://" + filename + print "Opening Uri:" + uri + src = gst.element_make_from_uri (gst.URI_SRC, uri, "src") + if (src is None): + return None + + decode = gst.element_factory_make ("decodebin", "decode") + if (decode is None): + return None + + + #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") + vscale = gst.element_factory_make ("videoscale", "vscale") + vrate = gst.element_factory_make ("videorate", "vrate") + vencode = gst.element_factory_make ("ffenc_mpeg4", "vencode") + vpay = gst.element_factory_make ("rtpmp4vpay", "vpay") + vsink = gst.element_factory_make ("udpsink", "vsink") + + if (None in [vbin, vqueue, vscale, vrate, vencode, vpay, vsink]): + print "Fail to create video encode elements." + return None + + vscale_pad = vscale.get_pad("sink") + if (vscale_pad is None): + print "Fail to get vscale sink pad." + return None + + vscale_caps = gst.caps_from_string ("video/x-raw-yuv, width=%s, height=%s" % (width, height)) + if (vscale_caps is None): + print "Fail to create video caps" + return None + + if (not vscale_pad.set_caps (vscale_caps)): + print "Fail to set video output caps" + return None + + vencode.set_property ("bitrate", 256000) + vencode.set_property ("me-method", 2) + + vsink.set_property ("host", "224.0.0.1") + vsink.set_property ("port", 5000) + + vbin.add (vqueue, vscale, vrate, vencode, vpay, vsink) + if (not gst.element_link_many (vqueue, vscale, vrate, vencode, vpay, vsink)): + print "Fail to link video elements" + return None + + vbin.add_pad (gst.GhostPad ("sink", vqueue.get_pad ("sink"))) + + #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", "vqueue") + aconvert = gst.element_factory_make ("audioconvert", "aconvert") + aencode = gst.element_factory_make ("faac", "aencode") + apay = gst.element_factory_make ("rtpmp4gpay", "apay") + asink = gst.element_factory_make ("udpsink", "asink") + + if (None in [abin, aqueue, aconvert, aencode, apay, asink]): + print "Fail to create video encode elements." + return None + + asink.set_property ("host", "224.0.0.1") + asink.set_property ("port", 5002) + + abin.add (aqueue, aconvert, aencode, apay, asink) + if (not gst.element_link_many (aqueue, aconvert, aencode, apay, asink)): + print "Fail to link video elements" + return None + + abin.add_pad (gst.GhostPad ("sink", aqueue.get_pad ("sink"))) + + self.pipe.add (src, decode, abin, vbin) + gst.element_link_many (src, decode) + + stream_data = self.StreamData (self.pipe, abin, vbin) + + bus = self.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) + + + self.pipe.set_state (gst.STATE_PAUSED) + print "Running Pipe" + stream_data.Loop.run () + print "End run" + + a_caps = stream_data.ACaps + v_caps = stream_data.VCaps + stream_id = stream_data.Id + + self.streams.append (stream_data) + + def play(self): + + print "Trying to play pipeline: %s" % self.pipe + try: + if (self.pipe): + self.pipe.set_state(gst.STATE_PLAYING) + except gobject.GError, e: + print "Error: " + str(e) + + + def stop(self): + + print "Trying to stop pipeline: %s" % self.pipe + try: + if (self.pipeline): + self.pipeline.set_state(gst.STATE_NULL) + except gobject.GError, e: + print "Error: " + str(e) + + 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" + self.__fill_sink_pads (stream_data) + 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 __fill_sink_pads (self, stream_data): + + asink = stream_data.Abin.get_by_name ("asink") + vsink = stream_data.Vbin.get_by_name ("vsink") + + asink_pad = asink.get_pad ("sink") + stream_data.ACaps = asink_pad.get_negotiated_caps().to_string() + print "ACAPS " + stream_data.ACaps + + vsink_pad = vsink.get_pad ("sink") + stream_data.VCaps = vsink_pad.get_negotiated_caps().to_string() + print "ACAPS " + stream_data.VCaps + + + + 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" + + diff -r bf3ac902013e -r 8edb3b445cac gmyth-stream/plugins/media/gstreamer.py --- a/gmyth-stream/plugins/media/gstreamer.py Tue Apr 03 14:58:30 2007 +0100 +++ b/gmyth-stream/plugins/media/gstreamer.py Tue Apr 03 16:05:00 2007 +0100 @@ -2,59 +2,174 @@ pygst.require("0.10") import gst import gobject +import socket class Media: + 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 + def __init__(self, config): - # set gstreamer basic options self.config = config - self.pipe = "" + self.pipe = None + self.streams = [] + self.socket = None + self.connection = None + self.addr = None - def setup(self, filename, mux, vcodec, vbitrate,\ - fps, acodec, abitrate, width, height, port): - - self.filename = filename - self.mux = mux - self.vcodec = vcodec - self.vbitrate = int(vbitrate) - self.fps = int(fps) - self.acodec = acodec - self.abitrate = int(abitrate) - self.width = int(width) - self.height = int(height) - - self.port = int(port) + def setup(self, filename, mux, vcodec, vbitrate, + fps, acodec, abitrate, width, height, port, options): ## Pipelines + self.pipe = gst.Pipeline () + uri = "file://" + filename + print "Opening Uri:" + uri + src = gst.element_make_from_uri (gst.URI_SRC, uri, "src") + if (src is None): + return None + + decode = gst.element_factory_make ("decodebin", "decode") + if (decode is None): + return None - #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 + mux = gst.element_factory_make ("avimux", "mux") + if (mux is None): + return None + sink = gst.element_factory_make ("fdsink", "sink") + if (sink is None): + return None - #audio/x-raw-int ! queue ! audioconvert ! faac ! rtpmp4gpay\ - #! udpsink name=upd_audio host=224.0.0.1 port=5002 + #Create socket + self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.socket.bind(('', int (port))) + + #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") + vscale = gst.element_factory_make ("videoscale", "vscale") + vrate = gst.element_factory_make ("videorate", "vrate") + vencode = gst.element_factory_make ("ffenc_h263p", "vencode") - self.pipe = "filesrc location=%s ! decodebin name=d ! queue ! videoscale !"\ - "video/x-raw-yuv,width=(int)%d,height=(int)%d ! ffenc_h263p bitrate=%d"\ - " me-method=2 ! rtph263ppay ! udpsink host=224.0.0.1 port=%d d. ! "\ - "queue ! audioconvert ! faac ! rtpmp4gpay ! udpsink name=udp_audio "\ - "host=224.0.0.1 port=%d" % (self.filename, self.width, self.height,\ - self.vbitrate, self.port, self.port+2) + if (None in [vbin, vqueue, vscale, vrate, vencode]): + print "Fail to create video encode elements." + return None - #self.pipe = "filesrc location=/tmp/mpg/cpm.mpg ! decodebin ! ffmpegcolorspace ! ximagesink" + vscale_pad = vscale.get_pad("sink") + if (vscale_pad is None): + print "Fail to get vscale sink pad." + return None - self.pipeline = gst.parse_launch(self.pipe) + vscale_caps = gst.caps_from_string ("video/x-raw-yuv, width=%s, height=%s" % (width, height)) + if (vscale_caps is None): + print "Fail to create video caps" + return None + if (not vscale_pad.set_caps (vscale_caps)): + print "Fail to set video output caps" + return None + + vbin.add (vqueue, vscale, vrate, vencode) + if (not gst.element_link_many (vqueue, vscale, vrate, vencode)): + print "Fail to link video elements" + return None + + vbin.add_pad (gst.GhostPad ("sink", vqueue.get_pad ("sink"))) + vbin.add_pad (gst.GhostPad ("src", vencode.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", "vqueue") + aconvert = gst.element_factory_make ("audioconvert", "aconvert") + aencode = gst.element_factory_make ("ffenc_ac3", "aencode") + + if (None in [abin, aqueue, aconvert, aencode]): + print "Fail to create video encode elements." + return None + + abin.add (aqueue, aconvert, aencode) + if (not gst.element_link_many (aqueue, aconvert, aencode)): + print "Fail to link video elements" + return None + + abin.add_pad (gst.GhostPad ("sink", aqueue.get_pad ("sink"))) + abin.add_pad (gst.GhostPad ("src", aencode.get_pad ("src"))) + + #Finish Pipeline + + self.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 (self.pipe, abin, vbin, sink) + + bus = self.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) + + + self.pipe.set_state (gst.STATE_PAUSED) + print "Running Pipe" + stream_data.Loop.run () + print "End run" + + a_caps = stream_data.ACaps + v_caps = stream_data.VCaps + stream_id = stream_data.Id + + self.streams.append (stream_data) def play(self): print "Trying to play pipeline: %s" % self.pipe try: - if (self.pipeline): - self.pipeline.set_state(gst.STATE_PLAYING) + if (self.pipe): + print "Waiting for connection" + self.socket.listen(1) + print "Connection Requested" + #Create socket + self.connection, self.addr = self.socket.accept () + + stream_data = self.streams[0] + stream_data.Sink.set_property ("fd", self.connection.fileno()); + print "Connected" + + self.pipe.set_state(gst.STATE_PLAYING) except gobject.GError, e: print "Error: " + str(e) @@ -64,7 +179,57 @@ print "Trying to stop pipeline: %s" % self.pipe try: if (self.pipeline): + self.connection.close () self.pipeline.set_state(gst.STATE_NULL) except gobject.GError, e: print "Error: " + str(e) + 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" + +