Source Code for Module timeside.analyzer.vamp_plugin

  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   
 26  import subprocess 
 27  import numpy as np 
 28   
 29   
 30 -class VampSimpleHost(Analyzer): 
 31      implements(IAnalyzer) 
 32   
 33 -    def __init__(self, plugin_list=None): 
 34          super(VampSimpleHost, self).__init__() 
 35          if plugin_list is None: 
 36              plugin_list = self.get_plugins_list() 
 37              #plugin_list = [['vamp-example-plugins', 'percussiononsets', 'detectionfunction']] 
 38   
 39          self.plugin_list = plugin_list 
 40   
 41      @interfacedoc 
 42 -    def setup(self, channels=None, samplerate=None, 
 43                blocksize=None, totalframes=None): 
 44          super(VampSimpleHost, self).setup( 
 45              channels, samplerate, blocksize, totalframes) 
 46   
 47      @staticmethod 
 48      @interfacedoc 
 49 -    def id(): 
 50          return "vamp_simple_host" 
 51   
 52      @staticmethod 
 53      @interfacedoc 
 54 -    def name(): 
 55          return "Vamp Plugins host" 
 56   
 57      @staticmethod 
 58      @interfacedoc 
 59 -    def unit(): 
 60          return "" 
 61   
 62 -    def process(self, frames, eod=False): 
 63          pass 
 64          return frames, eod 
 65   
 66 -    def post_process(self): 
 67          #plugin = 'vamp-example-plugins:amplitudefollower:amplitude' 
 68   
 69          wavfile = self.mediainfo()['uri'].split('file://')[-1] 
 70   
 71          for plugin_line in self.plugin_list: 
 72   
 73              plugin = ':'.join(plugin_line) 
 74              (time, duration, value) = self.vamp_plugin(plugin, wavfile) 
 75              if value is None: 
 76                  return 
 77   
 78              if duration is not None: 
 79                  plugin_res = self.new_result(data_mode='value', time_mode='segment') 
 80                  plugin_res.data_object.duration = duration 
 81              else: 
 82                  plugin_res = self.new_result(data_mode='value', time_mode='event') 
 83   
 84              plugin_res.data_object.time = time 
 85              plugin_res.data_object.value = value 
 86   
 87   
 88  #            # Fix strat, duration issues if audio is a segment 
 89  #            if self.mediainfo()['is_segment']: 
 90  #                start_index = np.floor(self.mediainfo()['start'] * 
 91  #                                       self.result_samplerate / 
 92  #                                       self.result_stepsize) 
 93  # 
 94  #                stop_index = np.ceil((self.mediainfo()['start'] + 
 95  #                                      self.mediainfo()['duration']) * 
 96  #                                     self.result_samplerate / 
 97  #                                     self.result_stepsize) 
 98  # 
 99  #                fixed_start = (start_index * self.result_stepsize / 
100  #                               self.result_samplerate) 
101  #                fixed_duration = ((stop_index - start_index) * self.result_stepsize / 
102  #                                  self.result_samplerate) 
103  # 
104  #                plugin_res.audio_metadata.start = fixed_start 
105  #                plugin_res.audio_metadata.duration = fixed_duration 
106  # 
107  #                value = value[start_index:stop_index + 1] 
108   
109              plugin_res.id_metadata.id += '.' + '.'.join(plugin_line[1:]) 
110              plugin_res.id_metadata.name += ' ' + \ 
111                  ' '.join(plugin_line[1:]) 
112   
113              self.process_pipe.results.add(plugin_res) 
114   
115      @staticmethod 
116 -    def vamp_plugin(plugin, wavfile): 
117   
118          args = [plugin, wavfile] 
119   
120          stdout = VampSimpleHost.SimpleHostProcess(args)  # run vamp-simple-host 
121   
122          stderr = stdout[0:8]  # stderr containing file and process information 
123          res = stdout[8:]  # stdout containg the feature data 
124   
125          if len(res) == 0: 
126              return ([], [], []) 
127   
128          # Parse stderr to get blocksize and stepsize 
129          blocksize_info = stderr[4] 
130   
131          import re 
132          # Match agianst pattern 'Using block size = %d, step size = %d' 
133          m = re.match( 
134              'Using block size = (\d+), step size = (\d+)', blocksize_info) 
135   
136          blocksize = int(m.groups()[0]) 
137          stepsize = int(m.groups()[1]) 
138          # Get the results 
139   
140          value = np.asfarray([line.split(': ')[1].split(' ') for line in res if (len(line.split(': ')) > 1)]) 
141          time = np.asfarray([r.split(':')[0].split(',')[0] for r in res]) 
142   
143          time_len = len(res[0].split(':')[0].split(',')) 
144          if time_len == 1: 
145              # event 
146              duration = None 
147          elif time_len == 2: 
148              # segment 
149              duration = np.asfarray([r.split(':')[0].split(',')[1] for r in res]) 
150   
151          return (time, duration, value) 
152   
153      @staticmethod 
154 -    def get_plugins_list(): 
155          arg = ['--list-outputs'] 
156          stdout = VampSimpleHost.SimpleHostProcess(arg) 
157   
158          return [line.split(':')[1:] for line in stdout] 
159   
160      @staticmethod 
161 -    def SimpleHostProcess(argslist): 
162          """Call vamp-simple-host""" 
163   
164          vamp_host = 'vamp-simple-host' 
165          command = [vamp_host] 
166          command.extend(argslist) 
167          # try ? 
168          stdout = subprocess.check_output( 
169              command, stderr=subprocess.STDOUT).splitlines() 
170   
171          return stdout 
172