gmyth-stream/server/0.2/plugins/transcoders/gstreamer.py
author renatofilho
Thu Apr 19 16:14:41 2007 +0100 (2007-04-19)
branchtrunk
changeset 573 f4927ca53da7
parent 572 0bc04f7e67e6
child 574 2dfea125c76c
permissions -rw-r--r--
[svn r578] * GMyth-Streamer:
- Fixed problems with gstreamer plugin
     1 #vim:ts=4:sw=4:et
     2 import pygst
     3 pygst.require("0.10")
     4 import gst
     5 import gobject
     6 import time
     7 import lib.utils as utils
     8 import lib.server as server
     9 
    10 from threading import Thread
    11 
    12 __all__ = ("TranscoderGstreamer",)
    13 
    14 class TranscoderGstreamer(server.Transcoder):
    15     gstreamer_path = utils.which("gst-launch-0.10")
    16     name = "gstreamer"
    17     priority = -2
    18 
    19     # StreamListener()
    20 
    21     class StreamData:
    22         stream_count = 0
    23 
    24         def __init__(self, log, pipe, abin, vbin, sink):
    25             self.log = log
    26             self.stream_count += 1
    27             self.Id = self.stream_count
    28             self.Pipe = pipe
    29             self.Abin = abin
    30             self.Vbin = vbin
    31             self.Sink = sink
    32             self.Loop = gobject.MainLoop()
    33             self.ACaps = ""
    34             self.VCaps = ""
    35             self.Ready = False
    36             self.Connection = None
    37             self.Addr = None
    38         # __init__()
    39 
    40     # StreamData()
    41 
    42 
    43     def __init__(self, params):
    44         server.Transcoder.__init__(self, params)
    45         gobject.threads_init ()
    46         # set gstreamer basic options
    47         self.connection = None
    48         self.addr = None
    49         self.ready = False
    50 
    51         self.log.info("Params for Gstreamer: %s" % self.params)
    52     # __init__()
    53 
    54 
    55     def _create_start_elements(self, uri):
    56         self.log.info("Opening Uri:" + uri)
    57         src = gst.element_make_from_uri(gst.URI_SRC, uri, "src")
    58         decode = gst.element_factory_make("decodebin", "decode")
    59         mux = gst.element_factory_make("ffmux_mpeg", "mux")
    60         sink = gst.element_factory_make("fdsink", "sink")
    61 
    62         return [src, decode, mux, sink]
    63     # _create_start_elements()
    64 
    65 
    66     def _setup_video_encode(self, vbin, width, height):
    67         vqueue = gst.element_factory_make("queue", "vqueue")
    68         colorspace = gst.element_factory_make("ffmpegcolorspace", "")
    69         vrate = gst.element_factory_make("videorate", "vrate")
    70         #vencode = gst.element_factory_make("ffenc_mpeg4", "vencode")
    71         vencode = gst.element_factory_make("ffenc_mpeg1video", "vencode")
    72         vqueue_src = gst.element_factory_make("queue", "vqueue_src")
    73 
    74         vencode.set_property("bitrate", 200)
    75         vencode.set_property ("pass", 2)
    76         vencode.set_property ("quantizer", 5)
    77 
    78         if None in [vbin, vqueue, vrate, vencode, vqueue_src]:
    79             self.log.info("Fail to create video encode elements.")
    80             return False
    81 
    82         vbin.add(vqueue)
    83         if int(width) > 0 and int(height) > 0:
    84             self.log.info(("Formating output to %d / %d" %(int(width), int(height))))
    85 
    86             vscale = gst.element_factory_make("ffvideoscale", "vscale")
    87 
    88             vbin.add(vscale);
    89             if not vqueue.link(vscale):
    90                 self.log.info("Fail to link video elements")
    91                 return False
    92 
    93             vbin.add(colorspace)
    94 
    95             if not vscale.link(colorspace, \
    96                 gst.caps_from_string("video/x-raw-yuv,width=(int)%d,height=(int)%d" %(\
    97                 int(width), int(height)))):
    98                 self.log.info("Fail to link video elements")
    99                 return False
   100         else:
   101             vbin.add(colorspace)
   102             vqueue.link(colorspace)
   103 
   104         vbin.add(vrate, vencode, vqueue_src)
   105         if not colorspace.link(vrate):
   106             self.log.info("Fail to colorspace with vrate")
   107             return False
   108 
   109         if not vrate.link(vencode, \
   110             gst.caps_from_string("video/x-raw-yuv,framerate=(fraction)10/1")):
   111             self.log.info("Fail to link vrate element")
   112             return False
   113 
   114         if not vencode.link(vqueue_src):
   115             self.log.info("Fail to link video encode with queue")
   116             return False
   117 
   118         vbin.add_pad(gst.GhostPad("sink", vqueue.get_pad("sink")))
   119         vbin.add_pad(gst.GhostPad("src", vqueue_src.get_pad("src")))
   120 
   121         return True
   122     # _setup_video_encode()
   123 
   124 
   125     def _setup_audio_encode(self, abin):
   126         aqueue = gst.element_factory_make("queue", "aqueue")
   127         aconvert = gst.element_factory_make("audioconvert", "aconvert")
   128         #aencode = gst.element_factory_make("queue", "aencode")
   129         #aencode = gst.element_factory_make("ffenc_mp2", "aencode")
   130         aencode = gst.element_factory_make("lame", "aencode")
   131         aqueue_src = gst.element_factory_make("queue", "aqueue_src")
   132        
   133         if None in [abin, aqueue, aencode, aqueue_src]:
   134             self.log.info("Fail to create video encode elements.")
   135             return False
   136 
   137         #aencode.set_property ("bitrate", 32)            
   138         #aencode.set_property ("vbr-quality", 2)
   139 
   140         abin.add(aqueue, aconvert, aencode, aqueue_src)
   141 
   142         self.log.info ("Link queue -> aconvert")
   143         if not aqueue.link (aconvert):
   144             self.log.info("Fail to link queue video")
   145             return False
   146 
   147         self.log.info ("Link aconvert -> aencode")
   148         if not aconvert.link (aencode):
   149             self.log.info("Fail to link video elements")
   150             return False
   151 
   152         self.log.info ("Link aencode -> aqueue_src")
   153         if not aencode.link (aqueue_src):
   154             self.log.info("Fail to link aencode -> aqueue_src")
   155             return False
   156 
   157         self.log.info ("Link:OK")
   158 
   159         ghost_sink = gst.GhostPad("sink", aqueue.get_pad("sink"))
   160         ghost_src = gst.GhostPad("src", aqueue_src.get_pad("src"))
   161         #ghost_src.set_caps (gst.caps_from_string ("audio/mpeg,mpegversion=(int)1,layer=(int)3,rate=(int)32000"))
   162         abin.add_pad(ghost_sink)
   163         abin.add_pad(ghost_src)
   164 
   165         return True
   166     # _setup_audio_encode()
   167 
   168 
   169     def setup(self, uri, mux, vcodec, vbitrate,
   170               fps, acodec, abitrate, width, height, options):
   171 
   172         ## Pipelines
   173         pipe = gst.Pipeline()
   174         src, decode, mux, sink = self._create_start_elements(uri)
   175 
   176         if None in [src, decode, mux, sink]:
   177             self.log.info("Problems with while starting basic elements");
   178             return False
   179 
   180         #video encode
   181         #queue ! videoscale ! video/x-raw-yuv,width=240,height=144 ! videorate !
   182         #ffenc_h263p bitrate=256000 me-method=2 ! rtph263ppay ! udpsink  host=224.0.0.1
   183         #port=5000
   184 
   185         vbin = gst.Bin()
   186         if not self._setup_video_encode(vbin, width, height):
   187             return False
   188 
   189         #audio encode
   190         #audio/x-raw-int ! queue ! audioconvert ! faac ! rtpmp4gpay !
   191         #udpsink name=upd_audio host=224.0.0.1 port=5002
   192 
   193         abin = gst.Bin()
   194         if not self._setup_audio_encode(abin):
   195             return False
   196 
   197         #Finish Pipeline
   198         pipe.add(src, decode, abin, vbin, mux, sink)
   199         gst.element_link_many(src, decode)
   200         gst.element_link_many(mux, sink)
   201 
   202         #Linking decode with mux
   203         mux_audio = mux.get_pad("audio_0")
   204         mux_video = mux.get_pad("video_0")
   205 
   206         audio_pad = abin.get_pad("src")
   207         video_pad = vbin.get_pad("src")
   208 
   209         if audio_pad.link(mux_audio) != gst.PAD_LINK_OK:
   210             self.log.info("Fail to link audio with mux")
   211             return False
   212 
   213         if video_pad.link(mux_video) != gst.PAD_LINK_OK:
   214             self.log.info("Fail to link audio with mux")
   215             return False
   216 
   217         self.stream_data = self.StreamData(self.log, pipe, abin, vbin, sink)
   218         bus = pipe.get_bus()
   219         bus.add_signal_watch()
   220         bus.connect("message", self.__on_bus_message, self.stream_data)
   221 
   222         decode.connect("new-decoded-pad", self.__on_decode_new_pad, self.stream_data)
   223         decode.connect("unknown-type", self.__on_decode_unknown_type, self.stream_data)
   224 
   225         self.log.info("Setting PIPELINE state to PAUSED")
   226         pipe.set_state(gst.STATE_PAUSED)
   227         self.log.info("Running Loop")
   228         self.stream_data.Loop.run()
   229     # setup()
   230 
   231     def __on_bus_message(self, bus, message, stream_data):
   232 
   233         t = message.type
   234         self.log.info ("__on_bus_message")
   235 
   236         if t == gst.MESSAGE_STATE_CHANGED:
   237             oldstate = -1
   238             newstate = -1
   239             pending = -1
   240 
   241             oldstate, newstate, pending = message.parse_state_changed()
   242 
   243             if oldstate == gst.STATE_READY and \
   244                newstate == gst.STATE_PAUSED and \
   245                pending == gst.STATE_VOID_PENDING and \
   246                stream_data.Ready == False:
   247 
   248                 state_changed_status, current_state, pending_state = stream_data.Pipe.get_state()
   249                 if current_state == gst.STATE_PAUSED and \
   250                     pending_state == gst.STATE_VOID_PENDING:
   251                     self.log.info("Pipe paused")
   252                     stream_data.Loop.quit()
   253                     stream_data.Ready = True
   254 
   255         elif t == gst.MESSAGE_EOS:
   256             self.log.info("Pipe finished")
   257             if stream_data.Ready:
   258                 self.stop()
   259             else:
   260                 stream_data.Loop.quit()
   261 
   262         elif t == gst.MESSAGE_ERROR:
   263             err, debug = message.parse_error()
   264             self.log.error("Error: %s %s" %(err, debug))
   265             if stream_data.Ready:
   266                 self.stop()
   267             else:
   268                 stream_data.Loop.quit()
   269 
   270         return True
   271     # __on_bus_message()
   272 
   273     def __on_decode_unknown_type(self, decode, pad, caps, stream_data):
   274         self.log.info("Unknown Type")
   275         return None
   276     # __on_decode_unknown_type
   277 
   278     def __on_decode_new_pad(self, decode, pad, arg1, stream_data):
   279 
   280         caps = pad.get_caps().to_string()
   281         self.log.info("New pad " + caps)
   282         if caps.rfind("audio") != -1:
   283             apad = stream_data.Abin.get_pad("sink")
   284             if pad.link(apad) != gst.PAD_LINK_OK:
   285                 self.log.info("Error on link audio pad")
   286                 return None
   287         elif caps.rfind("video") != -1:
   288             vpad = stream_data.Vbin.get_pad("sink")
   289             if pad.link(vpad) != gst.PAD_LINK_OK:
   290                 self.log.info("Error on link video pad")
   291                 return None
   292         else:
   293             self.log.info("Invalid caps")
   294         self.log.info("Linked")
   295     # __on_decode_new_pad
   296 
   297 
   298     def start(self, outfd):
   299         params_first = self.params_first
   300 
   301         uri = '%s://%s' % (params_first("uri_prefix", ""), params_first("uri_path", ""))
   302         self.setup(uri, params_first("mux", "avi"),
   303                    params_first("vcodec", "ffenc_h263p"), params_first("vbitrate", 256000),
   304                    params_first("fps", 25),  params_first("acodec", "faac"),
   305                    params_first("abitrate", 192000),  params_first("width", 320),
   306                    params_first("height", 240), params_first("options", ""))
   307 
   308         self.log.info("Play %s", outfd.fileno())
   309         self.stream_data.Sink.set_property("fd", outfd.fileno())
   310         self.log.info("Setting Pipeline state to PLAYING")
   311         self.stream_data.Pipe.set_state(gst.STATE_PLAYING)
   312 
   313         # keep playing until EOS
   314         self.stream_data.Loop.run()
   315         self.log.info("quit loop")
   316 
   317         return True
   318     # start()
   319 
   320     def stop(self):
   321         self.log.info("Stop stream_data: %s" % self.stream_data)
   322 
   323         if self.stream_data:
   324             self.stream_data.Pipe.set_state(gst.STATE_NULL)
   325 
   326         self.stream_data.Ready = False
   327         self.stream_data.Loop.quit ()
   328         del self.stream_data
   329         self.stream_data = None
   330         time.sleep (2)
   331     # stop