self.source = ts_samples["sweep.wav"]
def testOnGuitar(self):
- "runs on guitar"
+ "runs on C4_scale"
self.source = ts_samples["C4_scale.wav"]
def tearDown(self):
from timeside.decoder.file import FileDecoder
from timeside import _WITH_YAAFE
if _WITH_YAAFE:
- from timeside.analyzer.yaafe import Yaafe
+ from timeside.analyzer.externals.yaafe import Yaafe
import os
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2013 Paul Brossier <piem@piem.org>
-
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-# Author: Paul Brossier <piem@piem.org>
-from __future__ import absolute_import
-
-from ...core import implements, interfacedoc
-from ..core import Analyzer
-from ...api import IAnalyzer
-from ..preprocessors import downmix_to_mono, frames_adapter
-from aubio import filterbank, pvoc
-
-
-class AubioMelEnergy(Analyzer):
-
- """Aubio Mel Energy analyzer"""
- implements(IAnalyzer)
-
- def __init__(self):
- super(AubioMelEnergy, self).__init__()
- self.input_blocksize = 1024
- self.input_stepsize = self.input_blocksize / 4
-
- @interfacedoc
- def setup(self, channels=None, samplerate=None,
- blocksize=None, totalframes=None):
- super(AubioMelEnergy, self).setup(
- channels, samplerate, blocksize, totalframes)
- self.n_filters = 40
- self.n_coeffs = 13
- self.pvoc = pvoc(self.input_blocksize, self.input_stepsize)
- self.melenergy = filterbank(self.n_filters, self.input_blocksize)
- self.melenergy.set_mel_coeffs_slaney(samplerate)
- self.block_read = 0
- self.melenergy_results = []
-
- @staticmethod
- @interfacedoc
- def id():
- return "aubio_melenergy"
-
- @staticmethod
- @interfacedoc
- def name():
- return "Mel Energy (aubio)"
-
- @staticmethod
- @interfacedoc
- def unit():
- return ""
-
- @downmix_to_mono
- @frames_adapter
- def process(self, frames, eod=False):
-
- fftgrain = self.pvoc(frames)
- self.melenergy_results.append(self.melenergy(fftgrain))
- self.block_read += 1
- return frames, eod
-
- def post_process(self):
- melenergy = self.new_result(data_mode='value', time_mode='framewise')
- melenergy.parameters = dict(n_filters=self.n_filters,
- n_coeffs=self.n_coeffs)
- melenergy.data_object.value = self.melenergy_results
- self.add_result(melenergy)
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2013 Paul Brossier <piem@piem.org>
-
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-# Author: Paul Brossier <piem@piem.org>
-from __future__ import absolute_import
-
-from timeside.core import implements, interfacedoc
-from timeside.analyzer.core import Analyzer
-from timeside.api import IAnalyzer
-from timeside.analyzer.preprocessors import downmix_to_mono, frames_adapter
-
-import numpy
-from aubio import mfcc, pvoc
-
-
-class AubioMfcc(Analyzer):
-
- """Aubio MFCC analyzer"""
- implements(IAnalyzer)
-
- def __init__(self):
- super(AubioMfcc, self).__init__()
- self.input_blocksize = 1024
- self.input_stepsize = self.input_blocksize / 4
-
- @interfacedoc
- def setup(self, channels=None, samplerate=None,
- blocksize=None, totalframes=None):
- super(AubioMfcc, self).setup(
- channels, samplerate, blocksize, totalframes)
- self.n_filters = 40
- self.n_coeffs = 13
- self.pvoc = pvoc(self.input_blocksize, self.input_stepsize)
- self.mfcc = mfcc(self.input_blocksize,
- self.n_filters,
- self.n_coeffs,
- samplerate)
- self.block_read = 0
- self.mfcc_results = numpy.zeros([self.n_coeffs, ])
-
- @staticmethod
- @interfacedoc
- def id():
- return "aubio_mfcc"
-
- @staticmethod
- @interfacedoc
- def name():
- return "MFCC (aubio)"
-
- @staticmethod
- @interfacedoc
- def unit():
- return ""
-
- @downmix_to_mono
- @frames_adapter
- def process(self, frames, eod=False):
- fftgrain = self.pvoc(frames)
- coeffs = self.mfcc(fftgrain)
- self.mfcc_results = numpy.vstack((self.mfcc_results, coeffs))
- self.block_read += 1
- return frames, eod
-
- def post_process(self):
- mfcc = self.new_result(data_mode='value', time_mode='framewise')
- mfcc.parameters = dict(n_filters=self.n_filters,
- n_coeffs=self.n_coeffs)
- mfcc.data_object.value = self.mfcc_results
- self.add_result(mfcc)
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2013 Paul Brossier <piem@piem.org>
-
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-# Author: Paul Brossier <piem@piem.org>
-from __future__ import absolute_import
-
-from timeside.core import implements, interfacedoc
-from timeside.analyzer.core import Analyzer
-from timeside.api import IAnalyzer
-from timeside.analyzer.preprocessors import downmix_to_mono, frames_adapter
-from aubio import pitch
-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__()
-
- self._blocksize_s = blocksize_s
- self._stepsize_s = stepsize_s
-
- @interfacedoc
- def setup(self, channels=None, samplerate=None,
- blocksize=None, totalframes=None):
- super(AubioPitch, self).setup(channels,
- samplerate,
- blocksize,
- totalframes)
-
-
- # Frame parameters setup
- if self._blocksize_s:
- self.input_blocksize = nextpow2(self._blocksize_s * samplerate)
- else:
- self.input_blocksize = 2048
-
- if self._stepsize_s:
- self.input_stepsize = nextpow2(self._stepsize_s * samplerate)
- else:
- self.input_stepsize = self.input_blocksize / 2
-
- # Aubio Pitch set-up
- self.aubio_pitch = pitch(
- "default", self.input_blocksize, self.input_stepsize,
- samplerate)
- self.aubio_pitch.set_unit("freq")
- self.block_read = 0
- self.pitches = []
- self.pitch_confidences = []
-
- @staticmethod
- @interfacedoc
- def id():
- return "aubio_pitch"
-
- @staticmethod
- @interfacedoc
- def name():
- return "f0 (aubio)"
-
- @staticmethod
- @interfacedoc
- def unit():
- return "Hz"
-
- def __str__(self):
- return "pitch values"
-
- @downmix_to_mono
- @frames_adapter
- def process(self, frames, eod=False):
- #time = self.block_read * self.input_stepsize * 1. / self.samplerate()
- self.pitches += [self.aubio_pitch(frames)[0]]
- self.pitch_confidences += [
- np.nan_to_num(self.aubio_pitch.get_confidence())]
- self.block_read += 1
- return frames, eod
-
- def post_process(self):
- pitch = self.new_result(data_mode='value', time_mode='framewise')
-
- # parameters : None # TODO check with Piem "default" and "freq" in
- # setup
-
- pitch.id_metadata.id += '.' + "pitch"
- pitch.id_metadata.name += ' ' + "pitch"
- pitch.id_metadata.unit = "Hz"
- pitch.data_object.value = self.pitches
- self.add_result(pitch)
-
- pitch_confidence = self.new_result(
- data_mode='value', time_mode='framewise')
- pitch_confidence.id_metadata.id += '.' + "pitch_confidence"
- pitch_confidence.id_metadata.name += ' ' + "pitch confidence"
- pitch_confidence.id_metadata.unit = None
- pitch_confidence.data_object.value = self.pitch_confidences
- self.add_result(pitch_confidence)
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2013 Paul Brossier <piem@piem.org>
-
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-# Author: Paul Brossier <piem@piem.org>
-from __future__ import absolute_import
-
-from timeside.core import implements, interfacedoc
-from timeside.analyzer.core import Analyzer
-from timeside.api import IAnalyzer
-from timeside.analyzer.preprocessors import downmix_to_mono, frames_adapter
-
-from aubio import specdesc, pvoc
-
-
-class AubioSpecdesc(Analyzer):
-
- """Aubio Spectral Descriptors collection analyzer"""
- implements(IAnalyzer)
-
- def __init__(self):
- super(AubioSpecdesc, self).__init__()
- self.input_blocksize = 1024
- self.input_stepsize = self.input_blocksize / 4
-
- @interfacedoc
- def setup(self, channels=None, samplerate=None,
- blocksize=None, totalframes=None):
- super(
- AubioSpecdesc,
- self).setup(
- channels,
- samplerate,
- blocksize,
- totalframes)
- self.block_read = 0
- self.pvoc = pvoc(self.input_blocksize, self.input_stepsize)
- self.methods = [
- 'default', 'energy', 'hfc', 'complex', 'phase', 'specdiff', 'kl',
- 'mkl', 'specflux', 'centroid', 'slope', 'rolloff', 'spread', 'skewness',
- 'kurtosis', 'decrease']
- self.specdesc = {}
- self.specdesc_results = {}
- for method in self.methods:
- self.specdesc[method] = specdesc(method, self.input_blocksize)
- self.specdesc_results[method] = []
-
- @staticmethod
- @interfacedoc
- def id():
- return "aubio_specdesc"
-
- @staticmethod
- @interfacedoc
- def name():
- return "Spectral Descriptor (aubio)"
-
- @staticmethod
- @interfacedoc
- def unit():
- return ""
-
- @downmix_to_mono
- @frames_adapter
- def process(self, frames, eod=False):
- fftgrain = self.pvoc(frames)
- for method in self.methods:
- self.specdesc_results[method] += [
- self.specdesc[method](fftgrain)[0]]
- return frames, eod
-
- def post_process(self):
-
- # For each method store results in container
- for method in self.methods:
- res_specdesc = self.new_result(data_mode='value',
- time_mode='framewise')
- # Set metadata
- res_specdesc.id_metadata.id += '.' + method
- res_specdesc.id_metadata.name = ' ' + method
- res_specdesc.data_object.value = self.specdesc_results[method]
-
- self.add_result(res_specdesc)
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2013 Paul Brossier <piem@piem.org>
-
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-# Author: Paul Brossier <piem@piem.org>
-from __future__ import absolute_import
-
-from timeside.core import implements, interfacedoc
-from timeside.analyzer.core import Analyzer
-from timeside.api import IAnalyzer
-from timeside.analyzer.preprocessors import downmix_to_mono, frames_adapter
-from aubio import onset, tempo
-
-import numpy
-
-
-class AubioTemporal(Analyzer):
-
- """Aubio Temporal analyzer"""
- implements(IAnalyzer)
-
- def __init__(self):
- super(AubioTemporal, self).__init__()
- self.input_blocksize = 1024
- self.input_stepsize = 256
-
- @interfacedoc
- def setup(self,
- channels=None,
- samplerate=None,
- blocksize=None,
- totalframes=None):
- super(AubioTemporal, self).setup(
- channels, samplerate, blocksize, totalframes)
- self.o = onset(
- "default", self.input_blocksize, self.input_stepsize, samplerate)
- self.t = tempo(
- "default", self.input_blocksize, self.input_stepsize, samplerate)
- self.block_read = 0
- self.onsets = []
- self.beats = []
- self.beat_confidences = []
-
- @staticmethod
- @interfacedoc
- def id():
- return "aubio_temporal"
-
- @staticmethod
- @interfacedoc
- def name():
- return "onsets (aubio)"
-
- @staticmethod
- @interfacedoc
- def unit():
- return ""
-
- def __str__(self):
- return "%s %s" % (str(self.value), self.unit())
-
- @downmix_to_mono
- @frames_adapter
- def process(self, frames, eod=False):
- if self.o(frames):
- self.onsets += [self.o.get_last_s()]
- if self.t(frames):
- self.beats += [self.t.get_last_s()]
- self.beat_confidences += [self.t.get_confidence()]
- self.block_read += 1
- return frames, eod
-
- def post_process(self):
-
- #---------------------------------
- # Onsets: Event (time, "Onset")
- #---------------------------------
- onsets = self.new_result(data_mode='label', time_mode='event')
- onsets.id_metadata.id += '.' + 'onset'
- onsets.id_metadata.name += ' ' + 'Onset'
- onsets.id_metadata.unit = 's'
- onsets.data_object.time = self.onsets
- onsets.data_object.label = numpy.ones(len(self.onsets))
- onsets.data_object.label_metadata.label = {1: 'Onset'}
-
- self.add_result(onsets)
-
- #---------------------------------
- # Onset Rate: Segment (time, duration, value)
- #---------------------------------
- onsetrate = self.new_result(data_mode='value', time_mode='segment')
- onsetrate.id_metadata.id += '.' + "onset_rate"
- onsetrate.id_metadata.name = " " + "Onset Rate"
- onsetrate.id_metadata.unit = "bpm"
- if len(self.onsets) > 1:
- periods = numpy.diff(self.onsets)
- periods = numpy.append(periods, periods[-1])
- onsetrate.data_object.time = self.onsets
- onsetrate.data_object.duration = periods
- onsetrate.data_object.value = 60. / periods
- else:
- onsetrate.data_object.value = []
- onsetrate.data_object.time = []
-
- self.add_result(onsetrate)
-
- #---------------------------------
- # Beats: Event (time, "Beat")
- #---------------------------------
- beats = self.new_result(data_mode='label', time_mode='event')
- beats.id_metadata.id += '.' + "beat"
- beats.id_metadata.name += " " + "Beats"
- beats.id_metadata.unit = "s"
- beats.data_object.time = self.beats
- beats.data_object.label = numpy.ones(len(self.beats))
- beats.data_object.label_metadata.label = {1: 'Beat'}
-
- self.add_result(beats)
-
- #---------------------------------
- # Beat confidences: Event (time, value)
- #---------------------------------
- beat_confidences = self.new_result(
- data_mode='value', time_mode='event')
- beat_confidences.id_metadata.id += '.' + "beat_confidence"
- beat_confidences.id_metadata.name += " " + "Beat confidences"
- beat_confidences.id_metadata.unit = None
- beat_confidences.data_object.time = self.beats
- beat_confidences.data_object.value = self.beat_confidences
-
- self.add_result(beat_confidences)
-
- #---------------------------------
- # BPM: Segment (time, duration, value)
- #---------------------------------
- bpm = self.new_result(data_mode='value', time_mode='segment')
- bpm.id_metadata.id += '.' + "bpm"
- bpm.id_metadata.name += ' ' + "bpm"
- bpm.id_metadata.unit = "bpm"
- if len(self.beats) > 1:
- periods = numpy.diff(self.beats)
- periods = numpy.append(periods, periods[-1])
- bpm.data_object.time = self.beats
- bpm.data_object.duration = periods
- bpm.data_object.value = 60. / periods
- else:
- bpm.data_object.value = []
-
- self.add_result(bpm)
--- /dev/null
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2013 Paul Brossier <piem@piem.org>
+
+# This file is part of TimeSide.
+
+# TimeSide is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+
+# TimeSide is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
+
+# Author: Paul Brossier <piem@piem.org>
+from __future__ import absolute_import
+
+from ...core import implements, interfacedoc
+from ..core import Analyzer
+from ...api import IAnalyzer
+from ..preprocessors import downmix_to_mono, frames_adapter
+from aubio import filterbank, pvoc
+
+
+class AubioMelEnergy(Analyzer):
+
+ """Aubio Mel Energy analyzer"""
+ implements(IAnalyzer)
+
+ def __init__(self):
+ super(AubioMelEnergy, self).__init__()
+ self.input_blocksize = 1024
+ self.input_stepsize = self.input_blocksize / 4
+
+ @interfacedoc
+ def setup(self, channels=None, samplerate=None,
+ blocksize=None, totalframes=None):
+ super(AubioMelEnergy, self).setup(
+ channels, samplerate, blocksize, totalframes)
+ self.n_filters = 40
+ self.n_coeffs = 13
+ self.pvoc = pvoc(self.input_blocksize, self.input_stepsize)
+ self.melenergy = filterbank(self.n_filters, self.input_blocksize)
+ self.melenergy.set_mel_coeffs_slaney(samplerate)
+ self.block_read = 0
+ self.melenergy_results = []
+
+ @staticmethod
+ @interfacedoc
+ def id():
+ return "aubio_melenergy"
+
+ @staticmethod
+ @interfacedoc
+ def name():
+ return "Mel Energy (aubio)"
+
+ @staticmethod
+ @interfacedoc
+ def unit():
+ return ""
+
+ @downmix_to_mono
+ @frames_adapter
+ def process(self, frames, eod=False):
+
+ fftgrain = self.pvoc(frames)
+ self.melenergy_results.append(self.melenergy(fftgrain))
+ self.block_read += 1
+ return frames, eod
+
+ def post_process(self):
+ melenergy = self.new_result(data_mode='value', time_mode='framewise')
+ melenergy.parameters = dict(n_filters=self.n_filters,
+ n_coeffs=self.n_coeffs)
+ melenergy.data_object.value = self.melenergy_results
+ self.add_result(melenergy)
--- /dev/null
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2013 Paul Brossier <piem@piem.org>
+
+# This file is part of TimeSide.
+
+# TimeSide is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+
+# TimeSide is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
+
+# Author: Paul Brossier <piem@piem.org>
+from __future__ import absolute_import
+
+from timeside.core import implements, interfacedoc
+from timeside.analyzer.core import Analyzer
+from timeside.api import IAnalyzer
+from timeside.analyzer.preprocessors import downmix_to_mono, frames_adapter
+
+import numpy
+from aubio import mfcc, pvoc
+
+
+class AubioMfcc(Analyzer):
+
+ """Aubio MFCC analyzer"""
+ implements(IAnalyzer)
+
+ def __init__(self):
+ super(AubioMfcc, self).__init__()
+ self.input_blocksize = 1024
+ self.input_stepsize = self.input_blocksize / 4
+
+ @interfacedoc
+ def setup(self, channels=None, samplerate=None,
+ blocksize=None, totalframes=None):
+ super(AubioMfcc, self).setup(
+ channels, samplerate, blocksize, totalframes)
+ self.n_filters = 40
+ self.n_coeffs = 13
+ self.pvoc = pvoc(self.input_blocksize, self.input_stepsize)
+ self.mfcc = mfcc(self.input_blocksize,
+ self.n_filters,
+ self.n_coeffs,
+ samplerate)
+ self.block_read = 0
+ self.mfcc_results = numpy.zeros([self.n_coeffs, ])
+
+ @staticmethod
+ @interfacedoc
+ def id():
+ return "aubio_mfcc"
+
+ @staticmethod
+ @interfacedoc
+ def name():
+ return "MFCC (aubio)"
+
+ @staticmethod
+ @interfacedoc
+ def unit():
+ return ""
+
+ @downmix_to_mono
+ @frames_adapter
+ def process(self, frames, eod=False):
+ fftgrain = self.pvoc(frames)
+ coeffs = self.mfcc(fftgrain)
+ self.mfcc_results = numpy.vstack((self.mfcc_results, coeffs))
+ self.block_read += 1
+ return frames, eod
+
+ def post_process(self):
+ mfcc = self.new_result(data_mode='value', time_mode='framewise')
+ mfcc.parameters = dict(n_filters=self.n_filters,
+ n_coeffs=self.n_coeffs)
+ mfcc.data_object.value = self.mfcc_results
+ self.add_result(mfcc)
--- /dev/null
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2013 Paul Brossier <piem@piem.org>
+
+# This file is part of TimeSide.
+
+# TimeSide is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+
+# TimeSide is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
+
+# Author: Paul Brossier <piem@piem.org>
+from __future__ import absolute_import
+
+from timeside.core import implements, interfacedoc
+from timeside.analyzer.core import Analyzer
+from timeside.api import IAnalyzer
+from timeside.analyzer.preprocessors import downmix_to_mono, frames_adapter
+from aubio import pitch
+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__()
+
+ self._blocksize_s = blocksize_s
+ self._stepsize_s = stepsize_s
+
+ @interfacedoc
+ def setup(self, channels=None, samplerate=None,
+ blocksize=None, totalframes=None):
+ super(AubioPitch, self).setup(channels,
+ samplerate,
+ blocksize,
+ totalframes)
+
+
+ # Frame parameters setup
+ if self._blocksize_s:
+ self.input_blocksize = nextpow2(self._blocksize_s * samplerate)
+ else:
+ self.input_blocksize = 2048
+
+ if self._stepsize_s:
+ self.input_stepsize = nextpow2(self._stepsize_s * samplerate)
+ else:
+ self.input_stepsize = self.input_blocksize / 2
+
+ # Aubio Pitch set-up
+ self.aubio_pitch = pitch(
+ "default", self.input_blocksize, self.input_stepsize,
+ samplerate)
+ self.aubio_pitch.set_unit("freq")
+ self.block_read = 0
+ self.pitches = []
+ self.pitch_confidences = []
+
+ @staticmethod
+ @interfacedoc
+ def id():
+ return "aubio_pitch"
+
+ @staticmethod
+ @interfacedoc
+ def name():
+ return "f0 (aubio)"
+
+ @staticmethod
+ @interfacedoc
+ def unit():
+ return "Hz"
+
+ def __str__(self):
+ return "pitch values"
+
+ @downmix_to_mono
+ @frames_adapter
+ def process(self, frames, eod=False):
+ #time = self.block_read * self.input_stepsize * 1. / self.samplerate()
+ self.pitches += [self.aubio_pitch(frames)[0]]
+ self.pitch_confidences += [
+ np.nan_to_num(self.aubio_pitch.get_confidence())]
+ self.block_read += 1
+ return frames, eod
+
+ def post_process(self):
+ pitch = self.new_result(data_mode='value', time_mode='framewise')
+
+ # parameters : None # TODO check with Piem "default" and "freq" in
+ # setup
+
+ pitch.id_metadata.id += '.' + "pitch"
+ pitch.id_metadata.name += ' ' + "pitch"
+ pitch.id_metadata.unit = "Hz"
+ pitch.data_object.value = self.pitches
+ self.add_result(pitch)
+
+ pitch_confidence = self.new_result(
+ data_mode='value', time_mode='framewise')
+ pitch_confidence.id_metadata.id += '.' + "pitch_confidence"
+ pitch_confidence.id_metadata.name += ' ' + "pitch confidence"
+ pitch_confidence.id_metadata.unit = None
+ pitch_confidence.data_object.value = self.pitch_confidences
+ self.add_result(pitch_confidence)
--- /dev/null
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2013 Paul Brossier <piem@piem.org>
+
+# This file is part of TimeSide.
+
+# TimeSide is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+
+# TimeSide is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
+
+# Author: Paul Brossier <piem@piem.org>
+from __future__ import absolute_import
+
+from timeside.core import implements, interfacedoc
+from timeside.analyzer.core import Analyzer
+from timeside.api import IAnalyzer
+from timeside.analyzer.preprocessors import downmix_to_mono, frames_adapter
+
+from aubio import specdesc, pvoc
+
+
+class AubioSpecdesc(Analyzer):
+
+ """Aubio Spectral Descriptors collection analyzer"""
+ implements(IAnalyzer)
+
+ def __init__(self):
+ super(AubioSpecdesc, self).__init__()
+ self.input_blocksize = 1024
+ self.input_stepsize = self.input_blocksize / 4
+
+ @interfacedoc
+ def setup(self, channels=None, samplerate=None,
+ blocksize=None, totalframes=None):
+ super(
+ AubioSpecdesc,
+ self).setup(
+ channels,
+ samplerate,
+ blocksize,
+ totalframes)
+ self.block_read = 0
+ self.pvoc = pvoc(self.input_blocksize, self.input_stepsize)
+ self.methods = [
+ 'default', 'energy', 'hfc', 'complex', 'phase', 'specdiff', 'kl',
+ 'mkl', 'specflux', 'centroid', 'slope', 'rolloff', 'spread', 'skewness',
+ 'kurtosis', 'decrease']
+ self.specdesc = {}
+ self.specdesc_results = {}
+ for method in self.methods:
+ self.specdesc[method] = specdesc(method, self.input_blocksize)
+ self.specdesc_results[method] = []
+
+ @staticmethod
+ @interfacedoc
+ def id():
+ return "aubio_specdesc"
+
+ @staticmethod
+ @interfacedoc
+ def name():
+ return "Spectral Descriptor (aubio)"
+
+ @staticmethod
+ @interfacedoc
+ def unit():
+ return ""
+
+ @downmix_to_mono
+ @frames_adapter
+ def process(self, frames, eod=False):
+ fftgrain = self.pvoc(frames)
+ for method in self.methods:
+ self.specdesc_results[method] += [
+ self.specdesc[method](fftgrain)[0]]
+ return frames, eod
+
+ def post_process(self):
+
+ # For each method store results in container
+ for method in self.methods:
+ res_specdesc = self.new_result(data_mode='value',
+ time_mode='framewise')
+ # Set metadata
+ res_specdesc.id_metadata.id += '.' + method
+ res_specdesc.id_metadata.name = ' ' + method
+ res_specdesc.data_object.value = self.specdesc_results[method]
+
+ self.add_result(res_specdesc)
--- /dev/null
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2013 Paul Brossier <piem@piem.org>
+
+# This file is part of TimeSide.
+
+# TimeSide is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+
+# TimeSide is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
+
+# Author: Paul Brossier <piem@piem.org>
+from __future__ import absolute_import
+
+from timeside.core import implements, interfacedoc
+from timeside.analyzer.core import Analyzer
+from timeside.api import IAnalyzer
+from timeside.analyzer.preprocessors import downmix_to_mono, frames_adapter
+from aubio import onset, tempo
+
+import numpy
+
+
+class AubioTemporal(Analyzer):
+
+ """Aubio Temporal analyzer"""
+ implements(IAnalyzer)
+
+ def __init__(self):
+ super(AubioTemporal, self).__init__()
+ self.input_blocksize = 1024
+ self.input_stepsize = 256
+
+ @interfacedoc
+ def setup(self,
+ channels=None,
+ samplerate=None,
+ blocksize=None,
+ totalframes=None):
+ super(AubioTemporal, self).setup(
+ channels, samplerate, blocksize, totalframes)
+ self.o = onset(
+ "default", self.input_blocksize, self.input_stepsize, samplerate)
+ self.t = tempo(
+ "default", self.input_blocksize, self.input_stepsize, samplerate)
+ self.block_read = 0
+ self.onsets = []
+ self.beats = []
+ self.beat_confidences = []
+
+ @staticmethod
+ @interfacedoc
+ def id():
+ return "aubio_temporal"
+
+ @staticmethod
+ @interfacedoc
+ def name():
+ return "onsets (aubio)"
+
+ @staticmethod
+ @interfacedoc
+ def unit():
+ return ""
+
+ def __str__(self):
+ return "%s %s" % (str(self.value), self.unit())
+
+ @downmix_to_mono
+ @frames_adapter
+ def process(self, frames, eod=False):
+ if self.o(frames):
+ self.onsets += [self.o.get_last_s()]
+ if self.t(frames):
+ self.beats += [self.t.get_last_s()]
+ self.beat_confidences += [self.t.get_confidence()]
+ self.block_read += 1
+ return frames, eod
+
+ def post_process(self):
+
+ #---------------------------------
+ # Onsets: Event (time, "Onset")
+ #---------------------------------
+ onsets = self.new_result(data_mode='label', time_mode='event')
+ onsets.id_metadata.id += '.' + 'onset'
+ onsets.id_metadata.name += ' ' + 'Onset'
+ onsets.id_metadata.unit = 's'
+ onsets.data_object.time = self.onsets
+ onsets.data_object.label = numpy.ones(len(self.onsets))
+ onsets.data_object.label_metadata.label = {1: 'Onset'}
+
+ self.add_result(onsets)
+
+ #---------------------------------
+ # Onset Rate: Segment (time, duration, value)
+ #---------------------------------
+ onsetrate = self.new_result(data_mode='value', time_mode='segment')
+ onsetrate.id_metadata.id += '.' + "onset_rate"
+ onsetrate.id_metadata.name = " " + "Onset Rate"
+ onsetrate.id_metadata.unit = "bpm"
+ if len(self.onsets) > 1:
+ periods = numpy.diff(self.onsets)
+ periods = numpy.append(periods, periods[-1])
+ onsetrate.data_object.time = self.onsets
+ onsetrate.data_object.duration = periods
+ onsetrate.data_object.value = 60. / periods
+ else:
+ onsetrate.data_object.value = []
+ onsetrate.data_object.time = []
+
+ self.add_result(onsetrate)
+
+ #---------------------------------
+ # Beats: Event (time, "Beat")
+ #---------------------------------
+ beats = self.new_result(data_mode='label', time_mode='event')
+ beats.id_metadata.id += '.' + "beat"
+ beats.id_metadata.name += " " + "Beats"
+ beats.id_metadata.unit = "s"
+ beats.data_object.time = self.beats
+ beats.data_object.label = numpy.ones(len(self.beats))
+ beats.data_object.label_metadata.label = {1: 'Beat'}
+
+ self.add_result(beats)
+
+ #---------------------------------
+ # Beat confidences: Event (time, value)
+ #---------------------------------
+ beat_confidences = self.new_result(
+ data_mode='value', time_mode='event')
+ beat_confidences.id_metadata.id += '.' + "beat_confidence"
+ beat_confidences.id_metadata.name += " " + "Beat confidences"
+ beat_confidences.id_metadata.unit = None
+ beat_confidences.data_object.time = self.beats
+ beat_confidences.data_object.value = self.beat_confidences
+
+ self.add_result(beat_confidences)
+
+ #---------------------------------
+ # BPM: Segment (time, duration, value)
+ #---------------------------------
+ bpm = self.new_result(data_mode='value', time_mode='segment')
+ bpm.id_metadata.id += '.' + "bpm"
+ bpm.id_metadata.name += ' ' + "bpm"
+ bpm.id_metadata.unit = "bpm"
+ if len(self.beats) > 1:
+ periods = numpy.diff(self.beats)
+ periods = numpy.append(periods, periods[-1])
+ bpm.data_object.time = self.beats
+ bpm.data_object.duration = periods
+ bpm.data_object.value = 60. / periods
+ else:
+ bpm.data_object.value = []
+
+ self.add_result(bpm)
--- /dev/null
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2013 Paul Brossier <piem@piem.org>
+
+# This file is part of TimeSide.
+
+# TimeSide is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+
+# TimeSide is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
+
+# Author: Paul Brossier <piem@piem.org>
+
+from timeside.core import implements, interfacedoc
+from timeside.analyzer.core import Analyzer
+from timeside.api import IAnalyzer
+from timeside.tools.parameters import HasTraits, List
+
+import subprocess
+import numpy as np
+
+
+def simple_host_process(argslist):
+ """Call vamp-simple-host"""
+
+ vamp_host = 'vamp-simple-host'
+ command = [vamp_host]
+ command.extend(argslist)
+ # try ?
+ stdout = subprocess.check_output(
+ command, stderr=subprocess.STDOUT).splitlines()
+
+ return stdout
+
+
+# Raise an exception if Vamp Host is missing
+from timeside.exceptions import VampImportError
+try:
+ simple_host_process(['-v'])
+ WITH_VAMP = True
+except OSError:
+ WITH_VAMP = False
+ raise VampImportError
+
+
+class VampSimpleHost(Analyzer):
+
+ """Vamp plugins library interface analyzer"""
+
+ implements(IAnalyzer)
+
+ class _Param(HasTraits):
+ plugin_list = List
+
+ def __init__(self, plugin_list=None):
+ super(VampSimpleHost, self).__init__()
+ if plugin_list is None:
+ plugin_list = self.get_plugins_list()
+ #plugin_list = [['vamp-example-plugins', 'percussiononsets', 'detectionfunction']]
+
+ self.plugin_list = plugin_list
+
+ @interfacedoc
+ def setup(self, channels=None, samplerate=None,
+ blocksize=None, totalframes=None):
+ super(VampSimpleHost, self).setup(
+ channels, samplerate, blocksize, totalframes)
+
+ @staticmethod
+ @interfacedoc
+ def id():
+ return "vamp_simple_host"
+
+ @staticmethod
+ @interfacedoc
+ def name():
+ return "Vamp Plugins host"
+
+ @staticmethod
+ @interfacedoc
+ def unit():
+ return ""
+
+ def process(self, frames, eod=False):
+ pass
+ return frames, eod
+
+ def post_process(self):
+ #plugin = 'vamp-example-plugins:amplitudefollower:amplitude'
+
+ wavfile = self.mediainfo()['uri'].split('file://')[-1]
+
+ for plugin_line in self.plugin_list:
+
+ plugin = ':'.join(plugin_line)
+ (time, duration, value) = self.vamp_plugin(plugin, wavfile)
+ if value is None:
+ return
+
+ if duration is not None:
+ plugin_res = self.new_result(
+ data_mode='value', time_mode='segment')
+ plugin_res.data_object.duration = duration
+ else:
+ plugin_res = self.new_result(
+ data_mode='value', time_mode='event')
+
+ plugin_res.data_object.time = time
+ plugin_res.data_object.value = value
+
+# Fix strat, duration issues if audio is a segment
+# if self.mediainfo()['is_segment']:
+# start_index = np.floor(self.mediainfo()['start'] *
+# self.result_samplerate /
+# self.result_stepsize)
+#
+# stop_index = np.ceil((self.mediainfo()['start'] +
+# self.mediainfo()['duration']) *
+# self.result_samplerate /
+# self.result_stepsize)
+#
+# fixed_start = (start_index * self.result_stepsize /
+# self.result_samplerate)
+# fixed_duration = ((stop_index - start_index) * self.result_stepsize /
+# self.result_samplerate)
+#
+# plugin_res.audio_metadata.start = fixed_start
+# plugin_res.audio_metadata.duration = fixed_duration
+#
+# value = value[start_index:stop_index + 1]
+ plugin_res.id_metadata.id += '.' + '.'.join(plugin_line[1:])
+ plugin_res.id_metadata.name += ' ' + \
+ ' '.join(plugin_line[1:])
+
+ self.add_result(plugin_res)
+
+ @staticmethod
+ def vamp_plugin(plugin, wavfile):
+
+ args = [plugin, wavfile]
+
+ stdout = simple_host_process(args) # run vamp-simple-host
+
+ stderr = stdout[0:8] # stderr containing file and process information
+ res = stdout[8:] # stdout containg the feature data
+
+ if len(res) == 0:
+ return ([], [], [])
+
+ # Parse stderr to get blocksize and stepsize
+ blocksize_info = stderr[4]
+
+ import re
+ # Match agianst pattern 'Using block size = %d, step size = %d'
+ m = re.match(
+ 'Using block size = (\d+), step size = (\d+)', blocksize_info)
+
+ blocksize = int(m.groups()[0])
+ stepsize = int(m.groups()[1])
+ # Get the results
+
+ value = np.asfarray([line.split(': ')[1].split(' ')
+ for line in res if (len(line.split(': ')) > 1)])
+ time = np.asfarray([r.split(':')[0].split(',')[0] for r in res])
+
+ time_len = len(res[0].split(':')[0].split(','))
+ if time_len == 1:
+ # event
+ duration = None
+ elif time_len == 2:
+ # segment
+ duration = np.asfarray(
+ [r.split(':')[0].split(',')[1] for r in res])
+
+ return (time, duration, value)
+
+ @staticmethod
+ def get_plugins_list():
+ arg = ['--list-outputs']
+ stdout = simple_host_process(arg)
+
+ return [line.split(':')[1:] for line in stdout]
--- /dev/null
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2013 Thomas Fillon <thomas@parisson.com>
+
+# This file is part of TimeSide.
+
+# TimeSide is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+
+# TimeSide is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
+
+# Author : Thomas Fillon <thomas@parisson.com>
+"""
+Module Yaafe Analyzer
+"""
+
+from timeside.core import implements, interfacedoc
+from timeside.analyzer.core import Analyzer
+from timeside.api import IAnalyzer
+
+import yaafelib
+import numpy
+from timeside.analyzer.preprocessors import downmix_to_mono
+from timeside.tools.parameters import HasTraits, ListUnicode, Float
+
+
+class Yaafe(Analyzer):
+ """Yaafe feature extraction library interface analyzer"""
+ implements(IAnalyzer)
+
+ # Define Parameters
+ class _Param(HasTraits):
+
+ feature_plan = ListUnicode
+ input_samplerate = Float
+
+ def __init__(self, feature_plan=None, input_samplerate=32000):
+ super(Yaafe, self).__init__()
+
+ if input_samplerate is None:
+ self.input_samplerate = 0
+ else:
+ 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.data_flow)
+ self.yaafe_engine.reset()
+ #self.input_samplerate = samplerate
+ #self.input_blocksize = blocksize
+
+ @property
+ def force_samplerate(self):
+ return self.input_samplerate
+
+ @staticmethod
+ @interfacedoc
+ def id():
+ return "yaafe"
+
+ @staticmethod
+ @interfacedoc
+ def name():
+ return "Yaafe Descriptor"
+
+ @staticmethod
+ @interfacedoc
+ def unit():
+ return ''
+
+ @downmix_to_mono
+ def process(self, frames, eod=False):
+ # do process things...
+ # Convert to float64and reshape
+ # for compatibility with Yaafe engine
+ yaafe_frames = frames.astype(numpy.float64).reshape(1, -1)
+
+ # write audio array on 'audio' input
+ self.yaafe_engine.writeInput('audio', yaafe_frames)
+ # process available data
+ self.yaafe_engine.process()
+ if eod:
+ # flush yaafe engine to process remaining data
+ self.yaafe_engine.flush()
+
+ return frames, eod
+
+ def post_process(self):
+ # Get feature extraction results from yaafe
+ featNames = self.yaafe_engine.getOutputs().keys()
+ if len(featNames) == 0:
+ raise KeyError('Yaafe engine did not return any feature')
+ for featName in featNames:
+
+ result = self.new_result(data_mode='value', time_mode='framewise')
+ result.id_metadata.id += '.' + featName
+ result.id_metadata.name += ' ' + featName
+ # Read Yaafe Results
+ result.data_object.value = self.yaafe_engine.readOutput(featName)
+
+ yaafe_metadata = self.yaafe_engine.getOutputs()[featName]
+ result.data_object.frame_metadata.blocksize = yaafe_metadata['frameLength']
+ result.data_object.frame_metadata.stepsize = yaafe_metadata['sampleStep']
+ result.data_object.frame_metadata.samplerate = yaafe_metadata['sampleRate']
+
+ # Store results in Container
+ if len(result.data_object.value):
+ self.add_result(result)
from timeside.api import IAnalyzer
import numpy
from timeside.analyzer.preprocessors import frames_adapter
-from timeside.analyzer.aubio.aubio_pitch import AubioPitch
+from timeside.analyzer.externals.aubio_pitch import AubioPitch
class IRITMonopoly(Analyzer):
from timeside.api import IAnalyzer
import timeside
-from ..tools.parameters import Enum, HasTraits, Float, Tuple
+from timeside.tools.parameters import Enum, HasTraits, Float, Tuple
import numpy as np
import pickle
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2013 Paul Brossier <piem@piem.org>
-
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-# Author: Paul Brossier <piem@piem.org>
-
-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
-
-
-def simple_host_process(argslist):
- """Call vamp-simple-host"""
-
- vamp_host = 'vamp-simple-host'
- command = [vamp_host]
- command.extend(argslist)
- # try ?
- stdout = subprocess.check_output(
- command, stderr=subprocess.STDOUT).splitlines()
-
- return stdout
-
-
-# Raise an exception if Vamp Host is missing
-from ..exceptions import VampImportError
-try:
- simple_host_process(['-v'])
- WITH_VAMP = True
-except OSError:
- WITH_VAMP = False
- raise VampImportError
-
-
-class VampSimpleHost(Analyzer):
-
- """Vamp plugins library interface analyzer"""
-
- implements(IAnalyzer)
-
- class _Param(HasTraits):
- plugin_list = List
-
- def __init__(self, plugin_list=None):
- super(VampSimpleHost, self).__init__()
- if plugin_list is None:
- plugin_list = self.get_plugins_list()
- #plugin_list = [['vamp-example-plugins', 'percussiononsets', 'detectionfunction']]
-
- self.plugin_list = plugin_list
-
- @interfacedoc
- def setup(self, channels=None, samplerate=None,
- blocksize=None, totalframes=None):
- super(VampSimpleHost, self).setup(
- channels, samplerate, blocksize, totalframes)
-
- @staticmethod
- @interfacedoc
- def id():
- return "vamp_simple_host"
-
- @staticmethod
- @interfacedoc
- def name():
- return "Vamp Plugins host"
-
- @staticmethod
- @interfacedoc
- def unit():
- return ""
-
- def process(self, frames, eod=False):
- pass
- return frames, eod
-
- def post_process(self):
- #plugin = 'vamp-example-plugins:amplitudefollower:amplitude'
-
- wavfile = self.mediainfo()['uri'].split('file://')[-1]
-
- for plugin_line in self.plugin_list:
-
- plugin = ':'.join(plugin_line)
- (time, duration, value) = self.vamp_plugin(plugin, wavfile)
- if value is None:
- return
-
- if duration is not None:
- plugin_res = self.new_result(
- data_mode='value', time_mode='segment')
- plugin_res.data_object.duration = duration
- else:
- plugin_res = self.new_result(
- data_mode='value', time_mode='event')
-
- plugin_res.data_object.time = time
- plugin_res.data_object.value = value
-
-# Fix strat, duration issues if audio is a segment
-# if self.mediainfo()['is_segment']:
-# start_index = np.floor(self.mediainfo()['start'] *
-# self.result_samplerate /
-# self.result_stepsize)
-#
-# stop_index = np.ceil((self.mediainfo()['start'] +
-# self.mediainfo()['duration']) *
-# self.result_samplerate /
-# self.result_stepsize)
-#
-# fixed_start = (start_index * self.result_stepsize /
-# self.result_samplerate)
-# fixed_duration = ((stop_index - start_index) * self.result_stepsize /
-# self.result_samplerate)
-#
-# plugin_res.audio_metadata.start = fixed_start
-# plugin_res.audio_metadata.duration = fixed_duration
-#
-# value = value[start_index:stop_index + 1]
- plugin_res.id_metadata.id += '.' + '.'.join(plugin_line[1:])
- plugin_res.id_metadata.name += ' ' + \
- ' '.join(plugin_line[1:])
-
- self.add_result(plugin_res)
-
- @staticmethod
- def vamp_plugin(plugin, wavfile):
-
- args = [plugin, wavfile]
-
- stdout = simple_host_process(args) # run vamp-simple-host
-
- stderr = stdout[0:8] # stderr containing file and process information
- res = stdout[8:] # stdout containg the feature data
-
- if len(res) == 0:
- return ([], [], [])
-
- # Parse stderr to get blocksize and stepsize
- blocksize_info = stderr[4]
-
- import re
- # Match agianst pattern 'Using block size = %d, step size = %d'
- m = re.match(
- 'Using block size = (\d+), step size = (\d+)', blocksize_info)
-
- blocksize = int(m.groups()[0])
- stepsize = int(m.groups()[1])
- # Get the results
-
- value = np.asfarray([line.split(': ')[1].split(' ')
- for line in res if (len(line.split(': ')) > 1)])
- time = np.asfarray([r.split(':')[0].split(',')[0] for r in res])
-
- time_len = len(res[0].split(':')[0].split(','))
- if time_len == 1:
- # event
- duration = None
- elif time_len == 2:
- # segment
- duration = np.asfarray(
- [r.split(':')[0].split(',')[1] for r in res])
-
- return (time, duration, value)
-
- @staticmethod
- def get_plugins_list():
- arg = ['--list-outputs']
- stdout = simple_host_process(arg)
-
- return [line.split(':')[1:] for line in stdout]
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2013 Thomas Fillon <thomas@parisson.com>
-
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-# Author : Thomas Fillon <thomas@parisson.com>
-"""
-Module Yaafe Analyzer
-"""
-
-from timeside.core import implements, interfacedoc
-from timeside.analyzer.core import Analyzer
-from timeside.api import IAnalyzer
-
-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)
-
- # Define Parameters
- class _Param(HasTraits):
-
- feature_plan = ListUnicode
- input_samplerate = Float
-
- def __init__(self, feature_plan=None, input_samplerate=32000):
- super(Yaafe, self).__init__()
-
- if input_samplerate is None:
- self.input_samplerate = 0
- else:
- 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.data_flow)
- self.yaafe_engine.reset()
- #self.input_samplerate = samplerate
- #self.input_blocksize = blocksize
-
- @property
- def force_samplerate(self):
- return self.input_samplerate
-
- @staticmethod
- @interfacedoc
- def id():
- return "yaafe"
-
- @staticmethod
- @interfacedoc
- def name():
- return "Yaafe Descriptor"
-
- @staticmethod
- @interfacedoc
- def unit():
- return ''
-
- @downmix_to_mono
- def process(self, frames, eod=False):
- # do process things...
- # Convert to float64and reshape
- # for compatibility with Yaafe engine
- yaafe_frames = frames.astype(numpy.float64).reshape(1, -1)
-
- # write audio array on 'audio' input
- self.yaafe_engine.writeInput('audio', yaafe_frames)
- # process available data
- self.yaafe_engine.process()
- if eod:
- # flush yaafe engine to process remaining data
- self.yaafe_engine.flush()
-
- return frames, eod
-
- def post_process(self):
- # Get feature extraction results from yaafe
- featNames = self.yaafe_engine.getOutputs().keys()
- if len(featNames) == 0:
- raise KeyError('Yaafe engine did not return any feature')
- for featName in featNames:
-
- result = self.new_result(data_mode='value', time_mode='framewise')
- result.id_metadata.id += '.' + featName
- result.id_metadata.name += ' ' + featName
- # Read Yaafe Results
- result.data_object.value = self.yaafe_engine.readOutput(featName)
-
- yaafe_metadata = self.yaafe_engine.getOutputs()[featName]
- result.data_object.frame_metadata.blocksize = yaafe_metadata['frameLength']
- result.data_object.frame_metadata.stepsize = yaafe_metadata['sampleStep']
- result.data_object.frame_metadata.samplerate = yaafe_metadata['sampleRate']
-
- # Store results in Container
- if len(result.data_object.value):
- self.add_result(result)
"Check Vamp host availability"
try:
- from ..analyzer import vamp_plugin
+ from timeside.analyzer.externals import vamp_plugin
except VampImportError:
warnings.warn('Vamp host is not available', ImportWarning,
stacklevel=2)
projects / timeside.git / commitdiff