7 import lib.utils as utils
8 import lib.server as server
10 from threading import Thread
12 __all__ = ("TranscoderGstreamer",)
14 class TranscoderGstreamer(server.Transcoder):
15 gstreamer_path = utils.which("gst-launch-0.10")
24 def __init__(self, log, pipe, abin, vbin, sink):
26 self.stream_count += 1
27 self.Id = self.stream_count
32 self.Loop = gobject.MainLoop()
36 self.Connection = None
43 def __init__(self, params):
44 server.Transcoder.__init__(self, params)
45 gobject.threads_init ()
46 # set gstreamer basic options
47 self.connection = None
51 self.log.info("Params for Gstreamer: %s" % self.params)
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")
62 return [src, decode, mux, sink]
63 # _create_start_elements()
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")
74 vencode.set_property("bitrate", 200)
75 vencode.set_property ("pass", 2)
76 vencode.set_property ("quantizer", 5)
78 if None in [vbin, vqueue, vrate, vencode, vqueue_src]:
79 self.log.info("Fail to create video encode elements.")
83 if int(width) > 0 and int(height) > 0:
84 self.log.info(("Formating output to %d / %d" %(int(width), int(height))))
86 vscale = gst.element_factory_make("ffvideoscale", "vscale")
89 if not vqueue.link(vscale):
90 self.log.info("Fail to link video elements")
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")
102 vqueue.link(colorspace)
104 vbin.add(vrate, vencode, vqueue_src)
105 if not colorspace.link(vrate):
106 self.log.info("Fail to colorspace with vrate")
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")
114 if not vencode.link(vqueue_src):
115 self.log.info("Fail to link video encode with queue")
118 vbin.add_pad(gst.GhostPad("sink", vqueue.get_pad("sink")))
119 vbin.add_pad(gst.GhostPad("src", vqueue_src.get_pad("src")))
122 # _setup_video_encode()
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")
133 if None in [abin, aqueue, aencode, aqueue_src]:
134 self.log.info("Fail to create video encode elements.")
137 #aencode.set_property ("bitrate", 32)
138 #aencode.set_property ("vbr-quality", 2)
140 abin.add(aqueue, aconvert, aencode, aqueue_src)
142 self.log.info ("Link queue -> aconvert")
143 if not aqueue.link (aconvert):
144 self.log.info("Fail to link queue video")
147 self.log.info ("Link aconvert -> aencode")
148 if not aconvert.link (aencode):
149 self.log.info("Fail to link video elements")
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")
157 self.log.info ("Link:OK")
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)
166 # _setup_audio_encode()
169 def setup(self, uri, mux, vcodec, vbitrate,
170 fps, acodec, abitrate, width, height, options):
173 pipe = gst.Pipeline()
174 src, decode, mux, sink = self._create_start_elements(uri)
176 if None in [src, decode, mux, sink]:
177 self.log.info("Problems with while starting basic elements");
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
186 if not self._setup_video_encode(vbin, width, height):
190 #audio/x-raw-int ! queue ! audioconvert ! faac ! rtpmp4gpay !
191 #udpsink name=upd_audio host=224.0.0.1 port=5002
194 if not self._setup_audio_encode(abin):
198 pipe.add(src, decode, abin, vbin, mux, sink)
199 gst.element_link_many(src, decode)
200 gst.element_link_many(mux, sink)
202 #Linking decode with mux
203 mux_audio = mux.get_pad("audio_0")
204 mux_video = mux.get_pad("video_0")
206 audio_pad = abin.get_pad("src")
207 video_pad = vbin.get_pad("src")
209 if audio_pad.link(mux_audio) != gst.PAD_LINK_OK:
210 self.log.info("Fail to link audio with mux")
213 if video_pad.link(mux_video) != gst.PAD_LINK_OK:
214 self.log.info("Fail to link audio with mux")
217 self.stream_data = self.StreamData(self.log, pipe, abin, vbin, sink)
219 bus.add_signal_watch()
220 bus.connect("message", self.__on_bus_message, self.stream_data)
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)
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()
231 def __on_bus_message(self, bus, message, stream_data):
234 self.log.info ("__on_bus_message")
236 if t == gst.MESSAGE_STATE_CHANGED:
241 oldstate, newstate, pending = message.parse_state_changed()
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:
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
255 elif t == gst.MESSAGE_EOS:
256 self.log.info("Pipe finished")
257 if stream_data.Ready:
260 stream_data.Loop.quit()
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:
268 stream_data.Loop.quit()
273 def __on_decode_unknown_type(self, decode, pad, caps, stream_data):
274 self.log.info("Unknown Type")
276 # __on_decode_unknown_type
278 def __on_decode_new_pad(self, decode, pad, arg1, stream_data):
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")
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")
293 self.log.info("Invalid caps")
294 self.log.info("Linked")
295 # __on_decode_new_pad
298 def start(self, outfd):
299 params_first = self.params_first
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", ""))
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)
313 # keep playing until EOS
314 self.stream_data.Loop.run()
315 self.log.info("quit loop")
321 self.log.info("Stop stream_data: %s" % self.stream_data)
324 self.stream_data.Pipe.set_state(gst.STATE_NULL)
326 self.stream_data.Ready = False
327 self.stream_data.Loop.quit ()
329 self.stream_data = None