#! /usr/bin/env python
#
# Test rganalysis element. seems to work...
# Copyright (C) 2007 Michael Urman
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of version 2 of the GNU General Public License as
# published by the Free Software Foundation.
#
from os.path import realpath
import sys
import pygst
pygst.require('0.10')
import gst
class Analyze(object):
def analyze(self, files):
self.pipe = gst.Pipeline("pipe")
self.filesrc = gst.element_factory_make("filesrc", "source")
self.pipe.add(self.filesrc)
self.decode = gst.element_factory_make("decodebin", "decode")
self.decode.connect('new-decoded-pad', self.new_decoded_pad)
self.decode.connect('removed-decoded-pad', self.removed_decoded_pad)
self.pipe.add(self.decode)
self.filesrc.link(self.decode)
self.convert = gst.element_factory_make("audioconvert", "convert")
self.pipe.add(self.convert)
self.resample = gst.element_factory_make("audioresample", "resample")
self.pipe.add(self.resample)
self.convert.link(self.resample)
self.analysis = gst.element_factory_make("rganalysis", "analysis")
self.analysis.set_property("num-tracks", len(files))
self.pipe.add(self.analysis)
self.resample.link(self.analysis)
self.sink = gst.element_factory_make("fakesink", "sink")
self.pipe.add(self.sink)
self.analysis.link(self.sink)
bus = self.pipe.get_bus()
bus.add_signal_watch()
bus.connect("message::tag", self.bus_message_tag)
self.data = {}
for file in files:
self.current_song = file
self.filesrc.set_property("location", realpath(file))
self.pipe.set_state(gst.STATE_PLAYING)
self.analysis.set_locked_state(False)
while True:
message = bus.poll(-1, -1)
if message.type == gst.MESSAGE_EOS:
self.analysis.set_locked_state(True)
self.pipe.set_state(gst.STATE_NULL)
break
self.analysis.set_locked_state(False)
def new_decoded_pad(self, dbin, pad, islast):
print 'new decoded pad', dbin, pad, islast
pad.link(self.convert.get_pad("sink"))
def removed_decoded_pad(self, dbin, pad):
pad.unlink(self.convert.get_pad("sink"))
def bus_message_tag(self, bus, message):
if message.src != self.analysis:
return
tags = message.parse_tag()
try:
self.data[self.current_song] = tags[gst.TAG_TRACK_PEAK], tags[gst.TAG_TRACK_GAIN]
except KeyError: pass
try:
self.data[None] = tags[gst.TAG_ALBUM_PEAK], tags[gst.TAG_ALBUM_GAIN]
except KeyError: pass
analyzer = Analyze()
analyzer.analyze(sys.argv[1:])
from pprint import pprint
pprint(analyzer.data)
