Porting video to GStreamer 1.0
[mediagoblin.git] / mediagoblin / media_types / video / transcoders.py
index e08b897c692d0180b01c09cb39d1f9e7119aa300..d53cabc6c25c520fa88477247eb968724cab75bc 100644 (file)
@@ -19,16 +19,18 @@ from __future__ import division
 import os
 import sys
 import logging
-import urllib
 import multiprocessing
-import gobject
+from mediagoblin.media_types.tools import discover
+
+#os.environ['GST_DEBUG'] = '4,python:4'
 
 old_argv = sys.argv
 sys.argv = []
 
-import pygst
-pygst.require('0.10')
-import gst
+import gi
+gi.require_version('Gst', '1.0')
+from gi.repository import GObject, Gst, GstPbutils
+Gst.init(None)
 
 sys.argv = old_argv
 import struct
@@ -37,12 +39,8 @@ try:
 except ImportError:
     import Image
 
-from gst.extend import discoverer
-
 _log = logging.getLogger(__name__)
 
-gobject.threads_init()
-
 CPU_COUNT = 2
 
 try:
@@ -53,57 +51,70 @@ except NotImplementedError:
 os.putenv('GST_DEBUG_DUMP_DOT_DIR', '/tmp')
 
 
-def pixbuf_to_pilbuf(buf):
-    data = list()
-    for i in range(0, len(buf)-4, 4):
-        r, g, b, x = struct.unpack('BBBB', buf[i:i + 4])
-        # XXX: can something be done with the 'X' part of RGBX?
-        data.append((r, g, b))
-    return data
-
 def capture_thumb(video_path, dest_path, width=None, height=None, percent=0.5):
     def pad_added(element, pad, connect_to):
-        caps = pad.get_caps()
-        name = caps[0].get_name()
+        '''This is a callback to dynamically add element to pipeline'''
+        caps = pad.query_caps(None)
+        name = caps.to_string()
         _log.debug('on_pad_added: {0}'.format(name))
         if name.startswith('video') and not connect_to.is_linked():
             pad.link(connect_to)
-    # construct pipeline: uridecodebin ! ffmpegcolorspace ! videoscale ! \
+
+    # construct pipeline: uridecodebin ! videoconvert ! videoscale ! \
     # ! CAPS ! appsink
-    pipeline = gst.Pipeline()
-    uridecodebin = gst.element_factory_make('uridecodebin')
+    pipeline = Gst.Pipeline()
+    uridecodebin = Gst.ElementFactory.make('uridecodebin', None)
     uridecodebin.set_property('uri', 'file://{0}'.format(video_path))
-    ffmpegcolorspace = gst.element_factory_make('ffmpegcolorspace')
+    videoconvert = Gst.ElementFactory.make('videoconvert', None)
     uridecodebin.connect('pad-added', pad_added,
-                         ffmpegcolorspace.get_pad('sink'))
-    videoscale = gst.element_factory_make('videoscale')
-    filter = gst.element_factory_make('capsfilter', 'filter')
+                         videoconvert.get_static_pad('sink'))
+    videoscale = Gst.ElementFactory.make('videoscale', None)
+
     # create caps for video scaling
-    caps_struct = gst.Structure('video/x-raw-rgb')
-    caps_struct.set_value('pixel-aspect-ratio', gst.Fraction(1, 1))
+    caps_struct = Gst.Structure.new_empty('video/x-raw')
+    caps_struct.set_value('pixel-aspect-ratio', Gst.Fraction(1, 1))
+    caps_struct.set_value('format', 'RGB')
     if height:
         caps_struct.set_value('height', height)
     if width:
         caps_struct.set_value('width', width)
-    caps = gst.Caps(caps_struct)
-    filter.set_property('caps', caps)
-    appsink = gst.element_factory_make('appsink')
-    pipeline.add(uridecodebin, ffmpegcolorspace, videoscale, filter, appsink)
-    gst.element_link_many(ffmpegcolorspace, videoscale, filter, appsink)
+    caps = Gst.Caps.new_empty()
+    caps.append_structure(caps_struct)
+
+    # sink everything to memory
+    appsink = Gst.ElementFactory.make('appsink', None)
+    appsink.set_property('caps', caps)
+
+    # add everything to pipeline
+    elements = [uridecodebin, videoconvert, videoscale, appsink]
+    for e in elements:
+        pipeline.add(e)
+    videoconvert.link(videoscale)
+    videoscale.link(appsink)
+
     # pipeline constructed, starting playing, but first some preparations
-    if pipeline.set_state(gst.STATE_PAUSED) == gst.STATE_CHANGE_FAILURE:
-        _log.warning('state change failed')
-    pipeline.get_state()
-    duration = pipeline.query_duration(gst.FORMAT_TIME, None)[0]
-    if duration == gst.CLOCK_TIME_NONE:
+    # seek to 50% of the file is required
+    pipeline.set_state(Gst.State.PAUSED)
+    # timeout of 3 seconds below was set experimentally
+    state = pipeline.get_state(Gst.SECOND * 3)
+    if state[0] != Gst.StateChangeReturn.SUCCESS:
+        _log.warning('state change failed, {0}'.format(state))
+        return
+
+    # get duration
+    (success, duration) = pipeline.query_duration(Gst.Format.TIME)
+    if not success:
         _log.warning('query_duration failed')
-        duration = 0  # XXX
+        return
+
     seek_to = int(duration * int(percent * 100) / 100)
     _log.debug('Seeking to {0} of {1}'.format(
-            seek_to / gst.SECOND, duration / gst.SECOND))
-    seek = pipeline.seek_simple(gst.FORMAT_TIME, gst.SEEK_FLAG_FLUSH, seek_to)
+            float(seek_to) / Gst.SECOND, float(duration) / Gst.SECOND))
+    seek = pipeline.seek_simple(Gst.Format.TIME, Gst.SeekFlags.FLUSH, seek_to)
     if not seek:
         _log.warning('seek failed')
+        return
+
     # get sample, retrieve it's format and save
     sample = appsink.emit("pull-preroll")
     if not sample:
@@ -112,16 +123,20 @@ def capture_thumb(video_path, dest_path, width=None, height=None, percent=0.5):
     caps = sample.get_caps()
     if not caps:
         _log.warning('could not get snapshot format')
+        return
     structure = caps.get_structure(0)
     (success, width) = structure.get_int('width')
     (success, height) = structure.get_int('height')
     buffer = sample.get_buffer()
+
+    # get the image from the buffer and save it to disk
     im = Image.frombytes('RGB', (width, height),
                          buffer.extract_dup(0, buffer.get_size()))
     im.save(dest_path)
     _log.info('thumbnail saved to {0}'.format(dest_path))
+
     # cleanup
-    pipeline.set_state(gst.STATE_NULL)
+    pipeline.set_state(Gst.State.NULL)
 
 
 class VideoTranscoder(object):
@@ -130,16 +145,12 @@ class VideoTranscoder(object):
 
     Transcodes the SRC video file to a VP8 WebM video file at DST
 
-     - Does the same thing as VideoThumbnailer, but produces a WebM vp8
-       and vorbis video file.
-     - The VideoTranscoder exceeds the VideoThumbnailer in the way
-       that it was refined afterwards and therefore is done more
-       correctly.
+     - Produces a WebM vp8 and vorbis video file.
     '''
     def __init__(self):
         _log.info('Initializing VideoTranscoder...')
         self.progress_percentage = None
-        self.loop = gobject.MainLoop()
+        self.loop = GObject.MainLoop()
 
     def transcode(self, src, dst, **kwargs):
         '''
@@ -172,152 +183,85 @@ class VideoTranscoder(object):
         if not type(self.destination_dimensions) == tuple:
             raise Exception('dimensions must be tuple: (width, height)')
 
-        self._setup()
-        self._run()
-
-    # XXX: This could be a static method.
-    def discover(self, src):
-        '''
-        Discover properties about a media file
-        '''
-        _log.info('Discovering {0}'.format(src))
-
-        self.source_path = src
-        self._setup_discover(discovered_callback=self.__on_discovered)
-
-        self.discoverer.discover()
-
-        self.loop.run()
-        if hasattr(self, '_discovered_data'):
-            return self._discovered_data.__dict__
-        else:
-            return None
-
-    def __on_discovered(self, data, is_media):
-        _log.debug('Discovered: {0}'.format(data))
-        if not is_media:
-            self.__stop()
-            raise Exception('Could not discover {0}'.format(self.source_path))
-
-        self._discovered_data = data
-
-        self.__stop_mainloop()
-
-    def _setup(self):
-        self._setup_discover()
         self._setup_pipeline()
-
-    def _run(self):
-        _log.info('Discovering...')
-        self.discoverer.discover()
-        _log.info('Done')
-
-        _log.debug('Initializing MainLoop()')
-        self.loop.run()
-
-    def _setup_discover(self, **kw):
-        _log.debug('Setting up discoverer')
-        self.discoverer = discoverer.Discoverer(self.source_path)
-
-        # Connect self.__discovered to the 'discovered' event
-        self.discoverer.connect(
-            'discovered',
-            kw.get('discovered_callback', self.__discovered))
-
-    def __discovered(self, data, is_media):
-        '''
-        Callback for media discoverer.
-        '''
-        if not is_media:
-            self.__stop()
-            raise Exception('Could not discover {0}'.format(self.source_path))
-
-        _log.debug('__discovered, data: {0}'.format(data.__dict__))
-
-        self.data = data
-
-        # Launch things that should be done after discovery
+        self.data = discover(self.source_path)
         self._link_elements()
         self.__setup_videoscale_capsfilter()
-
-        # Tell the transcoding pipeline to start running
-        self.pipeline.set_state(gst.STATE_PLAYING)
+        self.pipeline.set_state(Gst.State.PLAYING)
         _log.info('Transcoding...')
+        _log.debug('Initializing MainLoop()')
+        self.loop.run()
+
 
     def _setup_pipeline(self):
         _log.debug('Setting up transcoding pipeline')
         # Create the pipeline bin.
-        self.pipeline = gst.Pipeline('VideoTranscoderPipeline')
+        self.pipeline = Gst.Pipeline.new('VideoTranscoderPipeline')
 
         # Create all GStreamer elements, starting with
         # filesrc & decoder
-        self.filesrc = gst.element_factory_make('filesrc', 'filesrc')
+        self.filesrc = Gst.ElementFactory.make('filesrc', 'filesrc')
         self.filesrc.set_property('location', self.source_path)
         self.pipeline.add(self.filesrc)
 
-        self.decoder = gst.element_factory_make('decodebin2', 'decoder')
-        self.decoder.connect('new-decoded-pad', self._on_dynamic_pad)
+        self.decoder = Gst.ElementFactory.make('decodebin', 'decoder')
+        self.decoder.connect('pad-added', self._on_dynamic_pad)
         self.pipeline.add(self.decoder)
 
         # Video elements
-        self.videoqueue = gst.element_factory_make('queue', 'videoqueue')
+        self.videoqueue = Gst.ElementFactory.make('queue', 'videoqueue')
         self.pipeline.add(self.videoqueue)
 
-        self.videorate = gst.element_factory_make('videorate', 'videorate')
+        self.videorate = Gst.ElementFactory.make('videorate', 'videorate')
         self.pipeline.add(self.videorate)
 
-        self.ffmpegcolorspace = gst.element_factory_make(
-            'ffmpegcolorspace', 'ffmpegcolorspace')
-        self.pipeline.add(self.ffmpegcolorspace)
+        self.videoconvert = Gst.ElementFactory.make('videoconvert',
+                                                    'videoconvert')
+        self.pipeline.add(self.videoconvert)
 
-        self.videoscale = gst.element_factory_make('ffvideoscale', 'videoscale')
-        #self.videoscale.set_property('method', 2)  # I'm not sure this works
-        #self.videoscale.set_property('add-borders', 0)
+        self.videoscale = Gst.ElementFactory.make('videoscale', 'videoscale')
         self.pipeline.add(self.videoscale)
 
-        self.capsfilter = gst.element_factory_make('capsfilter', 'capsfilter')
+        self.capsfilter = Gst.ElementFactory.make('capsfilter', 'capsfilter')
         self.pipeline.add(self.capsfilter)
 
-        self.vp8enc = gst.element_factory_make('vp8enc', 'vp8enc')
-        self.vp8enc.set_property('quality', self.vp8_quality)
+        self.vp8enc = Gst.ElementFactory.make('vp8enc', 'vp8enc')
         self.vp8enc.set_property('threads', self.vp8_threads)
-        self.vp8enc.set_property('max-latency', 25)
         self.pipeline.add(self.vp8enc)
 
         # Audio elements
-        self.audioqueue = gst.element_factory_make('queue', 'audioqueue')
+        self.audioqueue = Gst.ElementFactory.make('queue', 'audioqueue')
         self.pipeline.add(self.audioqueue)
 
-        self.audiorate = gst.element_factory_make('audiorate', 'audiorate')
+        self.audiorate = Gst.ElementFactory.make('audiorate', 'audiorate')
         self.audiorate.set_property('tolerance', 80000000)
         self.pipeline.add(self.audiorate)
 
-        self.audioconvert = gst.element_factory_make('audioconvert', 'audioconvert')
+        self.audioconvert = Gst.ElementFactory.make('audioconvert', 'audioconvert')
         self.pipeline.add(self.audioconvert)
 
-        self.audiocapsfilter = gst.element_factory_make('capsfilter',
-                                                        'audiocapsfilter')
-        audiocaps = ['audio/x-raw-float']
-        self.audiocapsfilter.set_property(
-            'caps',
-            gst.caps_from_string(
-                ','.join(audiocaps)))
+        self.audiocapsfilter = Gst.ElementFactory.make('capsfilter',
+                                                       'audiocapsfilter')
+        audiocaps = Gst.Caps.new_empty()
+        audiocaps_struct = Gst.Structure.new_empty('audio/x-raw')
+        audiocaps.append_structure(audiocaps_struct)
+        self.audiocapsfilter.set_property('caps', audiocaps)
         self.pipeline.add(self.audiocapsfilter)
 
-        self.vorbisenc = gst.element_factory_make('vorbisenc', 'vorbisenc')
+        self.vorbisenc = Gst.ElementFactory.make('vorbisenc', 'vorbisenc')
         self.vorbisenc.set_property('quality', self.vorbis_quality)
         self.pipeline.add(self.vorbisenc)
 
         # WebMmux & filesink
-        self.webmmux = gst.element_factory_make('webmmux', 'webmmux')
+        self.webmmux = Gst.ElementFactory.make('webmmux', 'webmmux')
         self.pipeline.add(self.webmmux)
 
-        self.filesink = gst.element_factory_make('filesink', 'filesink')
+        self.filesink = Gst.ElementFactory.make('filesink', 'filesink')
         self.filesink.set_property('location', self.destination_path)
         self.pipeline.add(self.filesink)
 
         # Progressreport
-        self.progressreport = gst.element_factory_make(
+        self.progressreport = Gst.ElementFactory.make(
             'progressreport', 'progressreport')
         # Update every second
         self.progressreport.set_property('update-freq', 1)
@@ -336,48 +280,41 @@ class VideoTranscoder(object):
         # 'new-decoded-pad' which links decoded src pads to either a video
         # or audio sink
         self.filesrc.link(self.decoder)
-
-        # Link all the video elements in a row to webmmux
-        gst.element_link_many(
-            self.videoqueue,
-            self.videorate,
-            self.ffmpegcolorspace,
-            self.videoscale,
-            self.capsfilter,
-            self.vp8enc,
-            self.webmmux)
+        # link the rest
+        self.videoqueue.link(self.videorate)
+        self.videorate.link(self.videoconvert)
+        self.videoconvert.link(self.videoscale)
+        self.videoscale.link(self.capsfilter)
+        self.capsfilter.link(self.vp8enc)
+        self.vp8enc.link(self.webmmux)
 
         if self.data.is_audio:
-            # Link all the audio elements in a row to webmux
-            gst.element_link_many(
-                self.audioqueue,
-                self.audiorate,
-                self.audioconvert,
-                self.audiocapsfilter,
-                self.vorbisenc,
-                self.webmmux)
-
-        gst.element_link_many(
-            self.webmmux,
-            self.progressreport,
-            self.filesink)
+            # Link all the audio elements in a row to webmmux
+            self.audioqueue.link(self.audiorate)
+            self.audiorate.link(self.audioconvert)
+            self.audioconvert.link(self.audiocapsfilter)
+            self.audiocapsfilter.link(self.vorbisenc)
+            self.vorbisenc.link(self.webmmux)
+        self.webmmux.link(self.progressreport)
+        self.progressreport.link(self.filesink)
 
         # Setup the message bus and connect _on_message to the pipeline
         self._setup_bus()
 
-    def _on_dynamic_pad(self, dbin, pad, islast):
+    def _on_dynamic_pad(self, dbin, pad):
         '''
-        Callback called when ``decodebin2`` has a pad that we can connect to
+        Callback called when ``decodebin`` has a pad that we can connect to
         '''
         # Intersect the capabilities of the video sink and the pad src
         # Then check if they have no common capabilities.
-        if self.ffmpegcolorspace.get_pad_template('sink')\
-                .get_caps().intersect(pad.get_caps()).is_empty():
+        if (self.videorate.get_static_pad('sink').get_pad_template()
+                .get_caps().intersect(pad.query_caps()).is_empty()):
             # It is NOT a video src pad.
-            pad.link(self.audioqueue.get_pad('sink'))
+            pad.link(self.audioqueue.get_static_pad('sink'))
         else:
             # It IS a video src pad.
-            pad.link(self.videoqueue.get_pad('sink'))
+            _log.debug('linking video to the pad dynamically')
+            pad.link(self.videoqueue.get_static_pad('sink'))
 
     def _setup_bus(self):
         self.bus = self.pipeline.get_bus()
@@ -388,73 +325,53 @@ class VideoTranscoder(object):
         '''
         Sets up the output format (width, height) for the video
         '''
-        caps = ['video/x-raw-yuv', 'pixel-aspect-ratio=1/1', 'framerate=30/1']
-
-        if self.data.videoheight > self.data.videowidth:
-            # Whoa! We have ourselves a portrait video!
-            caps.append('height={0}'.format(
-                    self.destination_dimensions[1]))
+        caps_struct = Gst.Structure.new_empty('video/x-raw')
+        caps_struct.set_value('pixel-aspect-ratio', Gst.Fraction(1, 1))
+        caps_struct.set_value('framerate', Gst.Fraction(30, 1))
+        video_info = self.data.get_video_streams()[0]
+        if video_info.get_height() > video_info.get_width():
+            # portrait
+            caps_struct.set_value('height', self.destination_dimensions[1])
         else:
-            # It's a landscape, phew, how normal.
-            caps.append('width={0}'.format(
-                    self.destination_dimensions[0]))
-
-        self.capsfilter.set_property(
-            'caps',
-            gst.caps_from_string(
-                ','.join(caps)))
+            # landscape
+            caps_struct.set_value('width', self.destination_dimensions[0])
+        caps = Gst.Caps.new_empty()
+        caps.append_structure(caps_struct)
+        self.capsfilter.set_property('caps', caps)
 
     def _on_message(self, bus, message):
         _log.debug((bus, message, message.type))
-
-        t = message.type
-
-        if message.type == gst.MESSAGE_EOS:
-            self._discover_dst_and_stop()
+        if message.type == Gst.MessageType.EOS:
+            self.dst_data = discover(self.destination_path)
+            self.__stop()
             _log.info('Done')
-
-        elif message.type == gst.MESSAGE_ELEMENT:
-            if message.structure.get_name() == 'progress':
-                data = dict(message.structure)
+        elif message.type == Gst.MessageType.ELEMENT:
+            if message.has_name('progress'):
+                structure = message.get_structure()
                 # Update progress state if it has changed
-                if self.progress_percentage != data.get('percent'):
-                    self.progress_percentage = data.get('percent')
+                (success, percent) = structure.get_int('percent')
+                if self.progress_percentage != percent and success:
+                    self.progress_percentage = percent
                     if self._progress_callback:
-                        self._progress_callback(data.get('percent'))
-
-                    _log.info('{percent}% done...'.format(
-                            percent=data.get('percent')))
-                _log.debug(data)
-
-        elif t == gst.MESSAGE_ERROR:
-            _log.error((bus, message))
+                        self._progress_callback(percent)
+                    _log.info('{percent}% done...'.format(percent=percent))
+        elif message.type == Gst.MessageType.ERROR:
+            _log.error('Got error: {0}'.format(message.parse_error()))
             self.__stop()
 
-    def _discover_dst_and_stop(self):
-        self.dst_discoverer = discoverer.Discoverer(self.destination_path)
-
-        self.dst_discoverer.connect('discovered', self.__dst_discovered)
-
-        self.dst_discoverer.discover()
-
-    def __dst_discovered(self, data, is_media):
-        self.dst_data = data
-
-        self.__stop()
-
     def __stop(self):
         _log.debug(self.loop)
 
         if hasattr(self, 'pipeline'):
             # Stop executing the pipeline
-            self.pipeline.set_state(gst.STATE_NULL)
+            self.pipeline.set_state(Gst.State.NULL)
 
         # This kills the loop, mercifully
-        gobject.idle_add(self.__stop_mainloop)
+        GObject.idle_add(self.__stop_mainloop)
 
     def __stop_mainloop(self):
         '''
-        Wrapper for gobject.MainLoop.quit()
+        Wrapper for GObject.MainLoop.quit()
 
         This wrapper makes us able to see if self.loop.quit has been called
         '''