Also add a test to chack if analyzer __init__ arguments are also defined as traits parameters.
#! /usr/bin/env python
-from unit_timeside import *
+from unit_timeside import unittest, TestRunner
from timeside.decoder.file import FileDecoder
from timeside import _WITH_YAAFE
if _WITH_YAAFE:
from timeside.analyzer.yaafe import Yaafe
- from yaafelib import DataFlow, FeaturePlan
import os
class TestYaafe(unittest.TestCase):
def setUp(self):
- self.sample_rate = 16000
+ self.samplerate = 16000
def testOnSweepWithFeaturePlan(self):
- "runs on sweep and define feature plan manually"
- self.source = os.path.join (os.path.dirname(__file__), "samples", "sweep.wav")
+ "runs on sweep"
+ self.source = os.path.join(os.path.dirname(__file__),
+ "samples", "sweep.wav")
# Setup Yaafe Analyzer
# Define Yaafe Feature Plan
- fp = FeaturePlan(sample_rate=self.sample_rate)
- # add feature definitions manually
- fp.addFeature('mfcc: MFCC blockSize=512 stepSize=256')
- fp.addFeature('mfcc_d1: MFCC blockSize=512 stepSize=256 > Derivate DOrder=1')
- fp.addFeature('mfcc_d2: MFCC blockSize=512 stepSize=256 > Derivate DOrder=2')
+ fp = ['mfcc: MFCC blockSize=512 stepSize=256',
+ 'mfcc_d1: MFCC blockSize=512 stepSize=256 > Derivate DOrder=1',
+ 'mfcc_d2: MFCC blockSize=512 stepSize=256 > Derivate DOrder=2']
# Setup a new Yaafe TimeSide analyzer
# from FeaturePlan
- self.analyzer = Yaafe(fp)
+ self.analyzer = Yaafe(feature_plan=fp,
+ input_samplerate=self.samplerate)
# Expected Results
self.result_length = 3
- def testOnGuitarWithFeaturePlanFromFile(self):
- "runs on guitar and load Yaafe feature plan from file"
- self.source = os.path.join (os.path.dirname(__file__), "samples", "guitar.wav")
- # Setup Yaafe Analyzer
- # Load Yaafe Feature Plan
- fp = FeaturePlan(sample_rate=self.sample_rate)
- fp_file = os.path.join (os.path.dirname(__file__), "yaafe_config", "yaafeFeaturePlan")
-
- fp.loadFeaturePlan(fp_file)
- # Setup a new Yaafe TimeSide analyzer
- # from FeaturePlan
- self.analyzer = Yaafe(fp)
-
- # Expected Results
- self.result_length = 3
-
- def testOnGuitarWithDataFlow(self):
- "runs on guitar and load Yaafe dataflow from file"
- self.source = os.path.join (os.path.dirname(__file__), "samples", "guitar.wav")
- # Setup Yaafe Analyzer
- # Load DataFlow from file
- df = DataFlow()
- df_file = os.path.join (os.path.dirname(__file__), "yaafe_config", "yaafeDataFlow")
- df.load(df_file)
-
- # Setup a new Yaafe TimeSide analyzer
- # from DataFlow
- self.analyzer = Yaafe(df)
-
- # Expected Results
- self.result_length = 5
-
def tearDown(self):
decoder = FileDecoder(self.source)
- decoder.output_samplerate = self.sample_rate
+ decoder.output_samplerate = self.samplerate
(decoder | self.analyzer).run()
results = self.analyzer.results
self.assertEquals(self.result_length, len(results))
- #print results
- #print results.to_yaml()
- #print results.to_json()
- #print results.to_xml()
+ # print results
+ # print results.to_yaml()
+ # print results.to_json()
+ # print results.to_xml()
if __name__ == '__main__':
unittest.main(testRunner=TestRunner())
import numpy as np
from timeside.analyzer.utils import nextpow2
+from ...tools.parameters import Float, HasTraits
+
+
class AubioPitch(Analyzer):
"""Aubio Pitch estimation analyzer"""
implements(IAnalyzer) # TODO check if needed with inheritance
+ # Define Parameters
+ class _Param(HasTraits):
+ blocksize_s = Float
+ stepsize_s = Float
+
def __init__(self, blocksize_s=None, stepsize_s=None):
super(AubioPitch, self).__init__()
Properties:
'''
@interfacedoc
- def __init__(self, save_lab=False):
+ def __init__(self):
super(IRITStartSeg, self).__init__()
- self._save_lab = save_lab
-
self._buffer = BufferTable()
# self.energy = []
self.max_energy = 0.002*2
self.min_overlap = 20
self.threshold = 0.12
+
@interfacedoc
def setup(self, channels=None, samplerate=None,
blocksize=None, totalframes=None):
label = {0: 'Start', 1: 'Session'}
- if self._save_lab:
- with open('out.lab', 'w') as f:
- for s in selected_segs:
- f.write(
- '%.2f\t%.2f\t%s\n' %
- (s[0] * step, s[1] * step, label[s[2]]))
-
- with open('cand.lab', 'w') as f:
- for s in candidates:
- f.write('%.2f\t%.2f\t%f\n' % (s[0] * step,
- s[1] * step,
- s[2]))
-
segs = self.new_result(data_mode='label', time_mode='segment')
segs.id_metadata.id += '.' + 'segments'
segs.id_metadata.name += ' ' + 'Segments'
from numpy.fft import rfft
from scipy.signal import firwin, lfilter
+from ..tools.parameters import Float, HasTraits
+
class IRITSpeech4Hz(Analyzer):
implements(IAnalyzer)
+ # Define Parameters
+ class _Param(HasTraits):
+ medfilt_duration = Float()
+
@interfacedoc
def __init__(self, medfilt_duration=5):
super(IRITSpeech4Hz, self).__init__()
from timeside.api import IAnalyzer
import timeside
+from ..tools.parameters import Enum, HasTraits
+
import yaafelib
import numpy as np
import pickle
"""
implements(IAnalyzer)
+ # Define Parameters
+ class _Param(HasTraits):
+ sad_model = Enum('etape', 'maya')
+
def __init__(self, sad_model='etape'):
"""
Parameters:
super(LimsiSad, self).__init__()
# feature extraction defition
- spec = yaafelib.FeaturePlan(sample_rate=16000)
- spec.addFeature(
- 'mfcc: MFCC CepsIgnoreFirstCoeff=0 blockSize=1024 stepSize=256')
- spec.addFeature(
- 'mfccd1: MFCC CepsIgnoreFirstCoeff=0 blockSize=1024 stepSize=256 > Derivate DOrder=1')
- spec.addFeature(
- 'mfccd2: MFCC CepsIgnoreFirstCoeff=0 blockSize=1024 stepSize=256 > Derivate DOrder=2')
- spec.addFeature('zcr: ZCR blockSize=1024 stepSize=256')
- parent_analyzer = get_processor('yaafe')(spec)
- self.parents['yaafe'] = parent_analyzer
+ feature_plan = ['mfcc: MFCC CepsIgnoreFirstCoeff=0 blockSize=1024 stepSize=256',
+ 'mfccd1: MFCC CepsIgnoreFirstCoeff=0 blockSize=1024 stepSize=256 > Derivate DOrder=1',
+ 'mfccd2: MFCC CepsIgnoreFirstCoeff=0 blockSize=1024 stepSize=256 > Derivate DOrder=2',
+ 'zcr: ZCR blockSize=1024 stepSize=256']
+ yaafe_analyzer = get_processor('yaafe')
+ self.parents['yaafe'] = yaafe_analyzer(feature_plan=feature_plan,
+ input_samplerate=16000)
# informative parameters
# these are not really taken into account by the system
if sad_model not in ['etape', 'maya']:
raise ValueError(
"argument sad_model %s not supported. Supported values are 'etape' or 'maya'" % sad_model)
+ self.sad_model = sad_model
picfname = os.path.join(
timeside.__path__[0], 'analyzer', 'trained_models', 'limsi_sad_%s.pkl' % sad_model)
self.gmms = pickle.load(open(picfname, 'rb'))
import numpy as np
from numpy import pi as Pi
from scipy import signal
+from ..tools.parameters import Int, HasTraits
class OnsetDetectionFunction(Analyzer):
"""Onset Detection Function analyzer"""
implements(IAnalyzer)
- def __init__(self, blocksize=1024, stepsize=None):
+ # Define Parameters
+ class _Param(HasTraits):
+ input_blocksize = Int()
+ input_stepsize = Int()
+
+ def __init__(self, input_blocksize=1024, input_stepsize=None):
super(OnsetDetectionFunction, self).__init__()
- self.input_blocksize = blocksize
- if stepsize:
- self.input_stepsize = stepsize
+ self.input_blocksize = input_blocksize
+ if input_stepsize:
+ self.input_stepsize = input_stepsize
else:
- self.input_stepsize = blocksize / 2
+ self.input_stepsize = input_blocksize / 2
self.parents['spectrogram'] = Spectrogram(
- blocksize=self.input_blocksize,
- stepsize=self.input_stepsize)
+ input_blocksize=self.input_blocksize,
+ input_stepsize=self.input_stepsize)
@interfacedoc
def setup(self, channels=None, samplerate=None,
from timeside.analyzer.core import Analyzer
from timeside.api import IAnalyzer
from timeside.analyzer.preprocessors import downmix_to_mono, frames_adapter
-from ..tools.parameters import Unicode, Int, HasTraits
+from ..tools.parameters import Int, HasTraits
import numpy as np
# Define Parameters
class _Param(HasTraits):
- FFT_SIZE = Int()
+ fft_size = Int()
input_blocksize = Int()
input_stepsize = Int()
- def __init__(self, blocksize=2048, stepsize=None, fft_size=None):
+ def __init__(self, input_blocksize=2048, input_stepsize=None,
+ fft_size=None):
super(Spectrogram, self).__init__()
- self.input_blocksize = blocksize
- if stepsize:
- self.input_stepsize = stepsize
+ self.input_blocksize = input_blocksize
+ if input_stepsize:
+ self.input_stepsize = input_stepsize
else:
- self.input_stepsize = blocksize // 2
+ self.input_stepsize = input_blocksize // 2
if not fft_size:
- self.FFT_SIZE = blocksize
+ self.fft_size = input_blocksize
else:
- self.FFT_SIZE = fft_size
+ self.fft_size = fft_size
self.values = []
@downmix_to_mono
@frames_adapter
def process(self, frames, eod=False):
- self.values.append(np.abs(np.fft.rfft(frames, self.FFT_SIZE)))
+ self.values.append(np.abs(np.fft.rfft(frames, self.fft_size)))
return frames, eod
def post_process(self):
spectrogram = self.new_result(data_mode='value', time_mode='framewise')
- spectrogram.parameters = {'FFT_SIZE': self.FFT_SIZE}
+ spectrogram.parameters = {'fft_size': self.fft_size}
spectrogram.data_object.value = self.values
nb_freq = spectrogram.data_object.value.shape[1]
spectrogram.data_object.y_value = (np.arange(0, nb_freq) *
- self.samplerate() / self.FFT_SIZE)
+ self.samplerate() / self.fft_size)
self.add_result(spectrogram)
from timeside.core import implements, interfacedoc
from timeside.analyzer.core import Analyzer
from timeside.api import IAnalyzer
+from ..tools.parameters import HasTraits, List
import subprocess
import numpy as np
implements(IAnalyzer)
+ class _Param(HasTraits):
+ plugin_list = List
+
def __init__(self, plugin_list=None):
super(VampSimpleHost, self).__init__()
if plugin_list is None:
import yaafelib
import numpy
from timeside.analyzer.preprocessors import downmix_to_mono
+from ..tools.parameters import HasTraits, ListUnicode, Float
class Yaafe(Analyzer):
"""Yaafe feature extraction library interface analyzer"""
implements(IAnalyzer)
- def __init__(self, yaafeSpecification=None):
+ # Define Parameters
+ class _Param(HasTraits):
+
+ feature_plan = ListUnicode
+ input_samplerate = Float
+
+ def __init__(self, feature_plan=None, input_samplerate=32000):
super(Yaafe, self).__init__()
- # Check arguments
- if yaafeSpecification is None:
- yaafeSpecification = yaafelib.FeaturePlan(sample_rate=32000)
- # add feature definitions manually
- yaafeSpecification.addFeature(
- 'mfcc: MFCC blockSize=512 stepSize=256')
-
- if isinstance(yaafeSpecification, yaafelib.DataFlow):
- self.dataFlow = yaafeSpecification
- elif isinstance(yaafeSpecification, yaafelib.FeaturePlan):
- self.featurePlan = yaafeSpecification
- self.dataFlow = self.featurePlan.getDataFlow()
+ if input_samplerate is None:
+ self.input_samplerate = 0
else:
- raise TypeError("'%s' Type must be either '%s' or '%s'" %
- (str(yaafeSpecification),
- str(yaafelib.DataFlow),
- str(yaafelib.FeaturePlan)))
+ self.input_samplerate = input_samplerate
+
+ # Check arguments
+ if feature_plan is None:
+ feature_plan = ['mfcc: MFCC blockSize=512 stepSize=256']
+
+ self.feature_plan = feature_plan
self.yaafe_engine = None
@interfacedoc
def setup(self, channels=None, samplerate=None,
blocksize=None, totalframes=None):
super(Yaafe, self).setup(channels, samplerate, blocksize, totalframes)
+
+ # Setup Yaafe Feature plan and Dataflow
+ yaafe_feature_plan = yaafelib.FeaturePlan(sample_rate=samplerate)
+ for feat in self.feature_plan:
+ yaafe_feature_plan.addFeature(feat)
+
+ self.data_flow = yaafe_feature_plan.getDataFlow()
+
# Configure a YAAFE engine
self.yaafe_engine = yaafelib.Engine()
- self.yaafe_engine.load(self.dataFlow)
+ self.yaafe_engine.load(self.data_flow)
self.yaafe_engine.reset()
- self.input_samplerate = samplerate
- self.input_blocksize = blocksize
+ #self.input_samplerate = samplerate
+ #self.input_blocksize = blocksize
+
+ @property
+ def force_samplerate(self):
+ return self.input_samplerate
@staticmethod
@interfacedoc
self.process_pipe = None
self.UUID = uuid.uuid4()
+ self.input_channels = 0
+ self.input_samplerate = 0
+ self.input_blocksize = 0
+ self.input_stepsize = 0
+
+
@interfacedoc
def setup(self, channels=None, samplerate=None, blocksize=None,
totalframes=None):
# If empty Set default values for input_* attributes
# may be setted by the processor during __init__()
- if not hasattr(self, 'input_channels'):
+ if not self.input_channels:
self.input_channels = self.source_channels
- if not hasattr(self, 'input_samplerate'):
+ if not self.input_samplerate:
self.input_samplerate = self.source_samplerate
- if not hasattr(self, 'input_blocksize'):
+ if not self.input_blocksize:
self.input_blocksize = self.source_blocksize
- if not hasattr(self, 'input_stepsize'):
+ if not self.input_stepsize:
self.input_stepsize = self.source_blocksize
# Check samplerate specification if any
# Thomas Fillon <thomas at parisson.com>
-from traits.api import HasTraits, Unicode, Int, Float, Range
+from traits.api import HasTraits, Unicode, Int, Float, Range, Enum
+from traits.api import ListUnicode, List
from traits.api import TraitError
import simplejson as json
TRAIT_TYPES = {Unicode: 'str',
Int: 'int',
Float: 'float',
- Range: 'range'}
+ Range: 'range',
+ Enum: 'enum',
+ ListUnicode: 'list of str',
+ List: 'list'}
class HasParam(object):
projects / timeside.git / commitdiff