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