[SCM] morituri/master: * morituri/common/encode.py: Add a task to read tags. * morituri/test/test_common_encode.py: Add a test.

[js at users.alioth.debian.org]

The following commit has been merged in the master branch:
commit a787676239a11b5a0689a4b2ca22a4f320d26f9f
Author: Thomas Vander Stichele <thomas (at) apestaart (dot) org>
Date:   Tue Apr 13 07:37:01 2010 +0000

    	* morituri/common/encode.py:
    	  Add a task to read tags.
    	* morituri/test/test_common_encode.py:
    	  Add a test.

diff --git a/ChangeLog b/ChangeLog
index 9bac061..f3b329c 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,10 @@
+2010-04-13  Thomas Vander Stichele  <thomas at apestaart dot org>
+
+	* morituri/common/encode.py:
+	  Add a task to read tags.
+	* morituri/test/test_common_encode.py:
+	  Add a test.
+
 2010-04-11  Thomas Vander Stichele  <thomas at apestaart dot org>
 
 	* morituri/common/encode.py:
diff --git a/morituri/common/encode.py b/morituri/common/encode.py
index 3a89131..a55419f 100644
--- a/morituri/common/encode.py
+++ b/morituri/common/encode.py
@@ -272,3 +272,82 @@ class EncodeTask(task.Task):
             self.peak = math.sqrt(math.pow(10, self._peakdB / 10.0))
         else:
             self.warning('No peak found, something went wrong!')
+
+class TagReadTask(task.Task):
+    """
+    I am a task that reads tags.
+
+    @ivar  taglist: the tag list read from the file.
+    @type  taglist: L{gst.TagList}
+    """
+
+    logCategory = 'TagReadTask'
+
+    description = 'Reading Tags'
+
+    taglist = None
+
+    def __init__(self, path):
+        """
+        """
+        assert type(path) is unicode, "inpath %r is not unicode" % inpath
+        
+        self._path = path
+
+    def start(self, runner):
+        task.Task.start(self, runner)
+
+        # here to avoid import gst eating our options
+        import gst
+
+        self._pipeline = gst.parse_launch('''
+            filesrc location="%s" !
+            decodebin name=decoder !
+            fakesink''' % (
+                common.quoteParse(self._path).encode('utf-8')))
+
+        self.debug('pausing pipeline')
+        self._pipeline.set_state(gst.STATE_PAUSED)
+        self._pipeline.get_state()
+        self.debug('paused pipeline')
+
+        # add eos handling
+        bus = self._pipeline.get_bus()
+        bus.add_signal_watch()
+        bus.connect('message::eos', self._message_eos_cb)
+
+        # set up tag callbacks
+        bus.connect('message::tag', self._message_tag_cb)
+
+        self.debug('scheduling setting to play')
+        # since set_state returns non-False, adding it as timeout_add
+        # will repeatedly call it, and block the main loop; so
+        #   gobject.timeout_add(0L, self._pipeline.set_state, gst.STATE_PLAYING)
+        # would not work.
+
+        def play():
+            self._pipeline.set_state(gst.STATE_PLAYING)
+            return False
+        self.runner.schedule(0, play)
+
+        #self._pipeline.set_state(gst.STATE_PLAYING)
+        self.debug('scheduled setting to play')
+
+    def _message_eos_cb(self, bus, message):
+        self.debug('eos, scheduling stop')
+        self.runner.schedule(0, self.stop)
+
+    def _message_tag_cb(self, bus, message):
+        taglist = message.parse_tag()
+        self.taglist = taglist
+
+    def stop(self):
+        # here to avoid import gst eating our options
+        import gst
+
+        self.debug('stopping')
+        self.debug('setting state to NULL')
+        self._pipeline.set_state(gst.STATE_NULL)
+        self.debug('set state to NULL')
+        task.Task.stop(self)
+
diff --git a/morituri/test/test_common_encode.py b/morituri/test/test_common_encode.py
index 9f41ad8..383b32c 100644
--- a/morituri/test/test_common_encode.py
+++ b/morituri/test/test_common_encode.py
@@ -39,3 +39,13 @@ class PathTestCase(common.TestCase):
 
     def testDoubleQuote(self):
         self._testSuffix(u'.morituri.test_encode.12" edit')
+
+class TagReadTestCase(common.TestCase):
+    def testRead(self):
+        path = os.path.join(os.path.dirname(__file__), u'track.flac')
+        self.runner = task.SyncRunner(verbose=False)
+        t = encode.TagReadTask(path)
+        self.runner.run(t)
+        self.failUnless(t.taglist)
+        self.assertEquals(t.taglist['audio-codec'], 'FLAC')
+        self.assertEquals(t.taglist['description'], 'audiotest wave')

-- 
morituri packaging