diff -r 000000000000 -r f5ef83bbe8b5 gmyth-stream/server/0.2/plugins/transcoders/gstreamer.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/gmyth-stream/server/0.2/plugins/transcoders/gstreamer.py Wed Apr 18 23:13:26 2007 +0100 @@ -0,0 +1,305 @@ +#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.ACaps = "" + self.VCaps = "" + self.Ready = False + self.Connection = None + self.Addr = None + # __init__() + + # StreamData() + + + def __init__(self, params): + server.Transcoder.__init__(self, params) + # set gstreamer basic options + self.connection = None + self.addr = None + self.ready = False + self.quit = False + + self.log.info("Params for Gstreamer: %s" % self.params) + # __init__() + + + def _create_start_elements(self, uri): + self.log.info("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("avimux", "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") + vqueue_src = gst.element_factory_make("queue", "vqueue_src") + + vencode.set_property("bitrate", 200) + + if None in [vbin, vqueue, vrate, vencode, vqueue_src]: + self.log.info("Fail to create video encode elements.") + return False + + vbin.add(vqueue) + if int(width) > 0 and int(height) > 0: + self.log.info(("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.info("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.info("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.info("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.info("Fail to link vrate element") + return False + + if not vencode.link(vqueue_src): + self.log.info("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") + arate = gst.element_factory_make("audioresample", "arate") + aencode = gst.element_factory_make("queue", "aencode") + aqueue_src = gst.element_factory_make("queue", "aqueue_src") + + if None in [abin, aqueue, arate, aencode, aqueue_src]: + self.log.info("Fail to create video encode elements.") + return False + + abin.add(aqueue, aconvert, arate, aencode, aqueue_src) + + if not gst.element_link_many(aqueue, aconvert, arate, aencode, aqueue_src): + self.log.info("Fail to link video elements") + return False + + abin.add_pad(gst.GhostPad("sink", aqueue.get_pad("sink"))) + abin.add_pad(gst.GhostPad("src", aqueue_src.get_pad("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.info("Fail to link audio with mux") + return False + + if video_pad.link(mux_video) != gst.PAD_LINK_OK: + self.log.info("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 + + 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") + stream_data.Loop.quit() + self.quit = True + + elif t == gst.MESSAGE_ERROR: + err, debug = message.parse_error() + self.log.error("Error: %s %s" %(err, debug)) + stream_data.Loop.quit() + stream_data.Ready = False + + 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.info("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.info("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.info("Error on link video pad") + return None + else: + self.log.info("Invalid caps") + self.log.info("Linked") + # __on_decode_new_pad + + + def start(self, outfd): + params_first = self.params_first + + self.setup(params_first("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.info("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.log.info("QUIT: %s" % self.quit) + + i = 0 + loop = gobject.MainLoop() + loop.run() + + self.log.info("quit loop") + + 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.quit = True + + del self.stream_data + self.stream_data = None + # stop