Source Code for Module timeside.analyzer.aubio_temporal

  1  # -*- coding: utf-8 -*- 
  2  # 
  3  # Copyright (c) 2013 Paul Brossier <piem@piem.org> 
  4   
  5  # This file is part of TimeSide. 
  6   
  7  # TimeSide is free software: you can redistribute it and/or modify 
  8  # it under the terms of the GNU General Public License as published by 
  9  # the Free Software Foundation, either version 2 of the License, or 
 10  # (at your option) any later version. 
 11   
 12  # TimeSide is distributed in the hope that it will be useful, 
 13  # but WITHOUT ANY WARRANTY; without even the implied warranty of 
 14  # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 
 15  # GNU General Public License for more details. 
 16   
 17  # You should have received a copy of the GNU General Public License 
 18  # along with TimeSide.  If not, see <http://www.gnu.org/licenses/>. 
 19   
 20  # Author: Paul Brossier <piem@piem.org> 
 21   
 22  from timeside.core import implements, interfacedoc 
 23  from timeside.analyzer.core import Analyzer 
 24  from timeside.api import IAnalyzer 
 25  from preprocessors import downmix_to_mono, frames_adapter 
 26  from aubio import onset, tempo 
 27   
 28  import numpy 
 29   
 30   
 31 -class AubioTemporal(Analyzer): 
 32      implements(IAnalyzer) 
 33   
 34 -    def __init__(self): 
 35          super(AubioTemporal, self).__init__() 
 36          self.input_blocksize = 1024 
 37          self.input_stepsize = 256 
 38   
 39      @interfacedoc 
 40 -    def setup(self, 
 41                channels=None, 
 42                samplerate=None, 
 43                blocksize=None, 
 44                totalframes=None): 
 45          super(AubioTemporal, self).setup( 
 46              channels, samplerate, blocksize, totalframes) 
 47          self.o = onset( 
 48              "default", self.input_blocksize, self.input_stepsize, samplerate) 
 49          self.t = tempo( 
 50              "default", self.input_blocksize, self.input_stepsize, samplerate) 
 51          self.block_read = 0 
 52          self.onsets = [] 
 53          self.beats = [] 
 54          self.beat_confidences = [] 
 55   
 56      @staticmethod 
 57      @interfacedoc 
 58 -    def id(): 
 59          return "aubio_temporal" 
 60   
 61      @staticmethod 
 62      @interfacedoc 
 63 -    def name(): 
 64          return "onsets (aubio)" 
 65   
 66      @staticmethod 
 67      @interfacedoc 
 68 -    def unit(): 
 69          return "" 
 70   
 71 -    def __str__(self): 
 72          return "%s %s" % (str(self.value), self.unit()) 
 73   
 74   
 75      @downmix_to_mono 
 76      @frames_adapter 
 77 -    def process(self, frames, eod=False): 
 78          if self.o(frames): 
 79              self.onsets += [self.o.get_last_s()] 
 80          if self.t(frames): 
 81              self.beats += [self.t.get_last_s()] 
 82              self.beat_confidences += [self.t.get_confidence()] 
 83          self.block_read += 1 
 84          return frames, eod 
 85   
 86 -    def post_process(self): 
 87   
 88          #--------------------------------- 
 89          #  Onsets: Event (time, "Onset") 
 90          #--------------------------------- 
 91          onsets = self.new_result(data_mode='label', time_mode='event') 
 92          onsets.id_metadata.id += '.' + 'onset' 
 93          onsets.id_metadata.name += ' ' + 'Onset' 
 94          onsets.id_metadata.unit = 's' 
 95          onsets.data_object.time = self.onsets 
 96          onsets.data_object.label = numpy.ones(len(self.onsets)) 
 97          onsets.label_metadata.label = {1: 'Onset'} 
 98   
 99          self.process_pipe.results.add(onsets) 
100   
101          #--------------------------------- 
102          #  Onset Rate: Segment (time, duration, value) 
103          #--------------------------------- 
104          onsetrate = self.new_result(data_mode='value', time_mode='segment') 
105          onsetrate.id_metadata.id += '.' + "onset_rate" 
106          onsetrate.id_metadata.name = " " + "Onset Rate" 
107          onsetrate.id_metadata.unit = "bpm" 
108          if len(self.onsets) > 1: 
109              periods = numpy.diff(self.onsets) 
110              periods = numpy.append(periods, periods[-1]) 
111              onsetrate.data_object.time = self.onsets 
112              onsetrate.data_object.duration = periods 
113              onsetrate.data_object.value = 60. / periods 
114          else: 
115              onsetrate.data_object.value = [] 
116              onsetrate.data_object.time = [] 
117   
118          self.process_pipe.results.add(onsetrate) 
119   
120          #--------------------------------- 
121          #  Beats: Event (time, "Beat") 
122          #--------------------------------- 
123          beats = self.new_result(data_mode='label', time_mode='event') 
124          beats.id_metadata.id += '.' + "beat" 
125          beats.id_metadata.name += " " + "Beats" 
126          beats.id_metadata.unit = "s" 
127          beats.data_object.time = self.beats 
128          beats.data_object.label = numpy.ones(len(self.beats)) 
129          beats.label_metadata.label = {1: 'Beat'} 
130   
131          self.process_pipe.results.add(beats) 
132   
133          #--------------------------------- 
134          #  Beat confidences: Event (time, value) 
135          #--------------------------------- 
136          beat_confidences = self.new_result(data_mode='value', time_mode='event') 
137          beat_confidences.id_metadata.id += '.' + "beat_confidence" 
138          beat_confidences.id_metadata.name += " " + "Beat confidences" 
139          beat_confidences.id_metadata.unit = None 
140          beat_confidences.data_object.time = self.beats 
141          beat_confidences.data_object.value = self.beat_confidences 
142   
143          self.process_pipe.results.add(beat_confidences) 
144   
145          #--------------------------------- 
146          #  BPM: Segment (time, duration, value) 
147          #--------------------------------- 
148          bpm = self.new_result(data_mode='value', time_mode='segment') 
149          bpm.id_metadata.id += '.' + "bpm" 
150          bpm.id_metadata.name += ' ' + "bpm" 
151          bpm.id_metadata.unit = "bpm" 
152          if len(self.beats) > 1: 
153              periods = numpy.diff(self.beats) 
154              periods = numpy.append(periods, periods[-1]) 
155              bpm.data_object.time = self.beats 
156              bpm.data_object.duration = periods 
157              bpm.data_object.value = 60. / periods 
158          else: 
159              bpm.data_object.value = [] 
160   
161          self.process_pipe.results.add(bpm) 
162