morphbr@565: #vim:ts=4:sw=4:et
morphbr@565: import pygst
morphbr@565: pygst.require("0.10")
morphbr@565: import gst
morphbr@565: import gobject
morphbr@565: import time
morphbr@565: import lib.utils as utils
morphbr@565: import lib.server as server
morphbr@565: 
morphbr@565: from threading import Thread
morphbr@565: 
morphbr@565: __all__ = ("TranscoderGstreamer",)
morphbr@565: 
morphbr@565: class TranscoderGstreamer(server.Transcoder):
morphbr@565:     gstreamer_path = utils.which("gst-launch-0.10")
morphbr@565:     name = "gstreamer"
morphbr@565:     priority = -2
morphbr@565: 
morphbr@565:     # StreamListener()
morphbr@565: 
morphbr@565:     class StreamData:
morphbr@565:         stream_count = 0
morphbr@565: 
morphbr@565:         def __init__(self, log, pipe, abin, vbin, sink):
morphbr@565:             self.log = log
morphbr@565:             self.stream_count += 1
morphbr@565:             self.Id = self.stream_count
morphbr@565:             self.Pipe = pipe
morphbr@565:             self.Abin = abin
morphbr@565:             self.Vbin = vbin
morphbr@565:             self.Sink = sink
morphbr@565:             self.Loop = gobject.MainLoop()
morphbr@565:             self.Ready = False
morphbr@565:         # __init__()
morphbr@565: 
morphbr@565:     # StreamData()
morphbr@565: 
morphbr@565: 
morphbr@572:     def __init__(self, params):
morphbr@565:         server.Transcoder.__init__(self, params)
renatofilho@570:         gobject.threads_init ()
morphbr@565:         # set gstreamer basic options
morphbr@565:         self.connection = None
morphbr@565:         self.addr = None
morphbr@565:         self.ready = False
morphbr@574:         self.log.debug("Params for Gstreamer: %s" % self.params)
morphbr@565:     # __init__()
morphbr@565: 
morphbr@565: 
morphbr@565:     def _create_start_elements(self, uri):
morphbr@574:         self.log.debug("Opening Uri:" + uri)
morphbr@565:         src = gst.element_make_from_uri(gst.URI_SRC, uri, "src")
morphbr@565:         decode = gst.element_factory_make("decodebin", "decode")
renatofilho@570:         mux = gst.element_factory_make("ffmux_mpeg", "mux")
morphbr@565:         sink = gst.element_factory_make("fdsink", "sink")
morphbr@565: 
morphbr@565:         return [src, decode, mux, sink]
morphbr@565:     # _create_start_elements()
morphbr@565: 
morphbr@565: 
morphbr@565:     def _setup_video_encode(self, vbin, width, height):
morphbr@565:         vqueue = gst.element_factory_make("queue", "vqueue")
morphbr@565:         colorspace = gst.element_factory_make("ffmpegcolorspace", "")
morphbr@565:         vrate = gst.element_factory_make("videorate", "vrate")
renatofilho@570:         #vencode = gst.element_factory_make("ffenc_mpeg4", "vencode")
renatofilho@570:         vencode = gst.element_factory_make("ffenc_mpeg1video", "vencode")
morphbr@565:         vqueue_src = gst.element_factory_make("queue", "vqueue_src")
morphbr@565: 
morphbr@565:         vencode.set_property("bitrate", 200)
renatofilho@570:         vencode.set_property ("pass", 2)
renatofilho@570:         vencode.set_property ("quantizer", 5)
morphbr@565: 
morphbr@565:         if None in [vbin, vqueue, vrate, vencode, vqueue_src]:
morphbr@574:             self.log.error("Fail to create video encode elements.")
morphbr@565:             return False
morphbr@565: 
morphbr@565:         vbin.add(vqueue)
morphbr@565:         if int(width) > 0 and int(height) > 0:
morphbr@574:             self.log.debug("Formating output to %d / %d" % ( int(width), int(height)))
morphbr@565: 
morphbr@565:             vscale = gst.element_factory_make("ffvideoscale", "vscale")
morphbr@565: 
morphbr@565:             vbin.add(vscale);
morphbr@565:             if not vqueue.link(vscale):
morphbr@574:                 self.log.error("Fail to link video elements")
morphbr@565:                 return False
morphbr@565: 
morphbr@565:             vbin.add(colorspace)
morphbr@565: 
morphbr@565:             if not vscale.link(colorspace, \
morphbr@565:                 gst.caps_from_string("video/x-raw-yuv,width=(int)%d,height=(int)%d" %(\
morphbr@565:                 int(width), int(height)))):
morphbr@574:                 self.log.error("Fail to link video elements")
morphbr@565:                 return False
morphbr@565:         else:
morphbr@565:             vbin.add(colorspace)
morphbr@565:             vqueue.link(colorspace)
morphbr@565: 
morphbr@565:         vbin.add(vrate, vencode, vqueue_src)
morphbr@565:         if not colorspace.link(vrate):
morphbr@574:             self.log.error("Fail to colorspace with vrate")
morphbr@565:             return False
morphbr@565: 
morphbr@565:         if not vrate.link(vencode, \
morphbr@565:             gst.caps_from_string("video/x-raw-yuv,framerate=(fraction)10/1")):
morphbr@574:             self.log.error("Fail to link vrate element")
morphbr@565:             return False
morphbr@565: 
morphbr@565:         if not vencode.link(vqueue_src):
morphbr@574:             self.log.error("Fail to link video encode with queue")
morphbr@565:             return False
morphbr@565: 
morphbr@565:         vbin.add_pad(gst.GhostPad("sink", vqueue.get_pad("sink")))
morphbr@565:         vbin.add_pad(gst.GhostPad("src", vqueue_src.get_pad("src")))
morphbr@565: 
morphbr@565:         return True
morphbr@565:     # _setup_video_encode()
morphbr@565: 
morphbr@565: 
morphbr@565:     def _setup_audio_encode(self, abin):
morphbr@565:         aqueue = gst.element_factory_make("queue", "aqueue")
morphbr@565:         aconvert = gst.element_factory_make("audioconvert", "aconvert")
renatofilho@570:         #aencode = gst.element_factory_make("queue", "aencode")
renatofilho@570:         #aencode = gst.element_factory_make("ffenc_mp2", "aencode")
renatofilho@570:         aencode = gst.element_factory_make("lame", "aencode")
morphbr@565:         aqueue_src = gst.element_factory_make("queue", "aqueue_src")
morphbr@574: 
renatofilho@570:         if None in [abin, aqueue, aencode, aqueue_src]:
morphbr@574:             self.log.error("Fail to create video encode elements.")
morphbr@565:             return False
morphbr@565: 
morphbr@574:         #aencode.set_property ("bitrate", 32)
renatofilho@570:         #aencode.set_property ("vbr-quality", 2)
morphbr@565: 
renatofilho@570:         abin.add(aqueue, aconvert, aencode, aqueue_src)
renatofilho@570: 
morphbr@574:         self.log.debug("Link queue -> aconvert")
renatofilho@570:         if not aqueue.link (aconvert):
morphbr@574:             self.log.error("Fail to link queue video")
renatofilho@570:             return False
renatofilho@570: 
morphbr@574:         self.log.debug("Link aconvert -> aencode")
renatofilho@570:         if not aconvert.link (aencode):
morphbr@574:             self.log.error("Fail to link video elements")
morphbr@565:             return False
morphbr@565: 
morphbr@574:         self.log.debug("Link aencode -> aqueue_src")
renatofilho@570:         if not aencode.link (aqueue_src):
morphbr@574:             self.log.error("Fail to link aencode -> aqueue_src")
renatofilho@570:             return False
renatofilho@570: 
morphbr@574:         self.log.debug("Link:OK")
renatofilho@570: 
renatofilho@570:         ghost_sink = gst.GhostPad("sink", aqueue.get_pad("sink"))
renatofilho@570:         ghost_src = gst.GhostPad("src", aqueue_src.get_pad("src"))
renatofilho@570:         #ghost_src.set_caps (gst.caps_from_string ("audio/mpeg,mpegversion=(int)1,layer=(int)3,rate=(int)32000"))
renatofilho@570:         abin.add_pad(ghost_sink)
renatofilho@570:         abin.add_pad(ghost_src)
morphbr@565: 
morphbr@565:         return True
morphbr@565:     # _setup_audio_encode()
morphbr@565: 
morphbr@565: 
morphbr@565:     def setup(self, uri, mux, vcodec, vbitrate,
morphbr@565:               fps, acodec, abitrate, width, height, options):
morphbr@565: 
morphbr@565:         ## Pipelines
morphbr@565:         pipe = gst.Pipeline()
morphbr@565:         src, decode, mux, sink = self._create_start_elements(uri)
morphbr@565: 
morphbr@565:         if None in [src, decode, mux, sink]:
morphbr@565:             self.log.info("Problems with while starting basic elements");
morphbr@565:             return False
morphbr@565: 
morphbr@565:         #video encode
morphbr@565:         #queue ! videoscale ! video/x-raw-yuv,width=240,height=144 ! videorate !
morphbr@565:         #ffenc_h263p bitrate=256000 me-method=2 ! rtph263ppay ! udpsink  host=224.0.0.1
morphbr@565:         #port=5000
morphbr@565: 
morphbr@565:         vbin = gst.Bin()
morphbr@565:         if not self._setup_video_encode(vbin, width, height):
morphbr@565:             return False
morphbr@565: 
morphbr@565:         #audio encode
morphbr@565:         #audio/x-raw-int ! queue ! audioconvert ! faac ! rtpmp4gpay !
morphbr@565:         #udpsink name=upd_audio host=224.0.0.1 port=5002
morphbr@565: 
morphbr@565:         abin = gst.Bin()
morphbr@565:         if not self._setup_audio_encode(abin):
morphbr@565:             return False
morphbr@565: 
morphbr@565:         #Finish Pipeline
morphbr@565:         pipe.add(src, decode, abin, vbin, mux, sink)
morphbr@565:         gst.element_link_many(src, decode)
morphbr@565:         gst.element_link_many(mux, sink)
morphbr@565: 
morphbr@565:         #Linking decode with mux
morphbr@565:         mux_audio = mux.get_pad("audio_0")
morphbr@565:         mux_video = mux.get_pad("video_0")
morphbr@565: 
morphbr@565:         audio_pad = abin.get_pad("src")
morphbr@565:         video_pad = vbin.get_pad("src")
morphbr@565: 
morphbr@565:         if audio_pad.link(mux_audio) != gst.PAD_LINK_OK:
morphbr@574:             self.log.error("Fail to link audio with mux")
morphbr@565:             return False
morphbr@565: 
morphbr@565:         if video_pad.link(mux_video) != gst.PAD_LINK_OK:
morphbr@574:             self.log.error("Fail to link audio with mux")
morphbr@565:             return False
morphbr@565: 
morphbr@565:         self.stream_data = self.StreamData(self.log, pipe, abin, vbin, sink)
morphbr@565:         bus = pipe.get_bus()
morphbr@565:         bus.add_signal_watch()
morphbr@565:         bus.connect("message", self.__on_bus_message, self.stream_data)
morphbr@565: 
morphbr@565:         decode.connect("new-decoded-pad", self.__on_decode_new_pad, self.stream_data)
morphbr@565:         decode.connect("unknown-type", self.__on_decode_unknown_type, self.stream_data)
morphbr@565: 
morphbr@565:         self.log.info("Setting PIPELINE state to PAUSED")
morphbr@565:         pipe.set_state(gst.STATE_PAUSED)
morphbr@565:         self.log.info("Running Loop")
morphbr@565:         self.stream_data.Loop.run()
morphbr@565:     # setup()
morphbr@565: 
morphbr@565:     def __on_bus_message(self, bus, message, stream_data):
morphbr@565: 
morphbr@565:         t = message.type
morphbr@574:         self.log.debug("__on_bus_message")
morphbr@565: 
morphbr@565:         if t == gst.MESSAGE_STATE_CHANGED:
morphbr@565:             oldstate = -1
morphbr@565:             newstate = -1
morphbr@565:             pending = -1
morphbr@565: 
morphbr@565:             oldstate, newstate, pending = message.parse_state_changed()
morphbr@565: 
morphbr@565:             if oldstate == gst.STATE_READY and \
morphbr@565:                newstate == gst.STATE_PAUSED and \
morphbr@565:                pending == gst.STATE_VOID_PENDING and \
morphbr@565:                stream_data.Ready == False:
morphbr@565: 
morphbr@565:                 state_changed_status, current_state, pending_state = stream_data.Pipe.get_state()
morphbr@565:                 if current_state == gst.STATE_PAUSED and \
morphbr@565:                     pending_state == gst.STATE_VOID_PENDING:
morphbr@565:                     self.log.info("Pipe paused")
morphbr@565:                     stream_data.Loop.quit()
morphbr@565:                     stream_data.Ready = True
morphbr@565: 
morphbr@565:         elif t == gst.MESSAGE_EOS:
morphbr@565:             self.log.info("Pipe finished")
renatofilho@570:             if stream_data.Ready:
renatofilho@570:                 self.stop()
renatofilho@570:             else:
renatofilho@570:                 stream_data.Loop.quit()
morphbr@565: 
morphbr@565:         elif t == gst.MESSAGE_ERROR:
morphbr@565:             err, debug = message.parse_error()
morphbr@565:             self.log.error("Error: %s %s" %(err, debug))
renatofilho@570:             if stream_data.Ready:
renatofilho@570:                 self.stop()
renatofilho@570:             else:
renatofilho@570:                 stream_data.Loop.quit()
morphbr@565: 
morphbr@565:         return True
morphbr@565:     # __on_bus_message()
morphbr@565: 
morphbr@565:     def __on_decode_unknown_type(self, decode, pad, caps, stream_data):
morphbr@565:         self.log.info("Unknown Type")
morphbr@565:         return None
morphbr@565:     # __on_decode_unknown_type
morphbr@565: 
morphbr@565:     def __on_decode_new_pad(self, decode, pad, arg1, stream_data):
morphbr@565: 
morphbr@565:         caps = pad.get_caps().to_string()
morphbr@574:         self.log.debug("New pad " + caps)
morphbr@565:         if caps.rfind("audio") != -1:
morphbr@565:             apad = stream_data.Abin.get_pad("sink")
morphbr@565:             if pad.link(apad) != gst.PAD_LINK_OK:
morphbr@574:                 self.log.error("Error on link audio pad")
morphbr@565:                 return None
morphbr@565:         elif caps.rfind("video") != -1:
morphbr@565:             vpad = stream_data.Vbin.get_pad("sink")
morphbr@565:             if pad.link(vpad) != gst.PAD_LINK_OK:
morphbr@574:                 self.log.error("Error on link video pad")
morphbr@565:                 return None
morphbr@565:         else:
morphbr@574:             self.log.error("Invalid caps")
morphbr@574:         self.log.debug("Linked")
morphbr@565:     # __on_decode_new_pad
morphbr@565: 
morphbr@565: 
morphbr@565:     def start(self, outfd):
morphbr@565:         params_first = self.params_first
morphbr@565: 
renatofilho@570:         uri = '%s://%s' % (params_first("uri_prefix", ""), params_first("uri_path", ""))
renatofilho@570:         self.setup(uri, params_first("mux", "avi"),
morphbr@565:                    params_first("vcodec", "ffenc_h263p"), params_first("vbitrate", 256000),
morphbr@565:                    params_first("fps", 25),  params_first("acodec", "faac"),
morphbr@565:                    params_first("abitrate", 192000),  params_first("width", 320),
morphbr@565:                    params_first("height", 240), params_first("options", ""))
morphbr@565: 
morphbr@574:         self.log.debug("Play %s", outfd.fileno())
morphbr@565:         self.stream_data.Sink.set_property("fd", outfd.fileno())
morphbr@565:         self.log.info("Setting Pipeline state to PLAYING")
morphbr@565:         self.stream_data.Pipe.set_state(gst.STATE_PLAYING)
morphbr@565: 
morphbr@565:         # keep playing until EOS
renatofilho@570:         self.stream_data.Loop.run()
morphbr@565:         return True
morphbr@565:     # start()
morphbr@565: 
morphbr@565:     def stop(self):
morphbr@565:         self.log.info("Stop stream_data: %s" % self.stream_data)
morphbr@565: 
morphbr@565:         if self.stream_data:
morphbr@565:             self.stream_data.Pipe.set_state(gst.STATE_NULL)
morphbr@565: 
renatofilho@570:         self.stream_data.Ready = False
renatofilho@570:         self.stream_data.Loop.quit ()
morphbr@565:         del self.stream_data
morphbr@565:         self.stream_data = None
renatofilho@573:         time.sleep (2)
morphbr@565:     # stop