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.ACaps = "" morphbr@565: self.VCaps = "" morphbr@565: self.Ready = False morphbr@565: self.Connection = None morphbr@565: self.Addr = None morphbr@565: # __init__() morphbr@565: morphbr@565: # StreamData() morphbr@565: morphbr@565: morphbr@565: def __init__(self, params): morphbr@565: server.Transcoder.__init__(self, params) morphbr@565: # set gstreamer basic options morphbr@565: self.connection = None morphbr@565: self.addr = None morphbr@565: self.ready = False morphbr@565: self.quit = False morphbr@565: morphbr@565: self.log.info("Params for Gstreamer: %s" % self.params) morphbr@565: # __init__() morphbr@565: morphbr@565: morphbr@565: def _create_start_elements(self, uri): morphbr@565: self.log.info("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") morphbr@565: mux = gst.element_factory_make("avimux", "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") morphbr@565: vencode = gst.element_factory_make("ffenc_mpeg4", "vencode") morphbr@565: vqueue_src = gst.element_factory_make("queue", "vqueue_src") morphbr@565: morphbr@565: vencode.set_property("bitrate", 200) morphbr@565: morphbr@565: if None in [vbin, vqueue, vrate, vencode, vqueue_src]: morphbr@565: self.log.info("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@565: self.log.info(("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@565: self.log.info("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@565: self.log.info("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@565: self.log.info("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@565: self.log.info("Fail to link vrate element") morphbr@565: return False morphbr@565: morphbr@565: if not vencode.link(vqueue_src): morphbr@565: self.log.info("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") morphbr@565: arate = gst.element_factory_make("audioresample", "arate") morphbr@565: aencode = gst.element_factory_make("queue", "aencode") morphbr@565: aqueue_src = gst.element_factory_make("queue", "aqueue_src") morphbr@565: morphbr@565: if None in [abin, aqueue, arate, aencode, aqueue_src]: morphbr@565: self.log.info("Fail to create video encode elements.") morphbr@565: return False morphbr@565: morphbr@565: abin.add(aqueue, aconvert, arate, aencode, aqueue_src) morphbr@565: morphbr@565: if not gst.element_link_many(aqueue, aconvert, arate, aencode, aqueue_src): morphbr@565: self.log.info("Fail to link video elements") morphbr@565: return False morphbr@565: morphbr@565: abin.add_pad(gst.GhostPad("sink", aqueue.get_pad("sink"))) morphbr@565: abin.add_pad(gst.GhostPad("src", aqueue_src.get_pad("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@565: self.log.info("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@565: self.log.info("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@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") morphbr@565: stream_data.Loop.quit() morphbr@565: self.quit = True 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)) morphbr@565: stream_data.Loop.quit() morphbr@565: stream_data.Ready = False 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@565: self.log.info("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@565: self.log.info("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@565: self.log.info("Error on link video pad") morphbr@565: return None morphbr@565: else: morphbr@565: self.log.info("Invalid caps") morphbr@565: self.log.info("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: morphbr@565: self.setup(params_first("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@565: self.log.info("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 morphbr@565: self.log.info("QUIT: %s" % self.quit) morphbr@565: morphbr@565: i = 0 morphbr@565: loop = gobject.MainLoop() morphbr@565: loop.run() morphbr@565: morphbr@565: self.log.info("quit loop") morphbr@565: 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: self.quit = True morphbr@565: morphbr@565: del self.stream_data morphbr@565: self.stream_data = None morphbr@565: # stop