--- /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.process_pipe.results.add(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 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.process_pipe.results.add(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 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
+
+
+class AubioPitch(Analyzer):
+
+ """Aubio Pitch estimation analyzer"""
+ implements(IAnalyzer) # TODO check if needed with inheritance
+
+ def __init__(self):
+ super(AubioPitch, self).__init__()
+ self.input_blocksize = 2048
+ self.input_stepsize = self.input_blocksize / 2
+
+ @interfacedoc
+ def setup(self, channels=None, samplerate=None,
+ blocksize=None, totalframes=None):
+ super(AubioPitch, self).setup(channels,
+ samplerate,
+ blocksize,
+ totalframes)
+ 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.process_pipe.results.add(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.process_pipe.results.add(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 timeside.core import Processor, implements, interfacedoc, FixedSizeInputAdapter
+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.process_pipe.results.add(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 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.label_metadata.label = {1: 'Onset'}
+
+ self.process_pipe.results.add(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.process_pipe.results.add(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.label_metadata.label = {1: 'Beat'}
+
+ self.process_pipe.results.add(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.process_pipe.results.add(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.process_pipe.results.add(bpm)
projects / timeside.git / commitdiff