Chris@56: '''A high-level interface to the vampyhost extension module, for quickly and easily running Vamp audio analysis plugins on audio files and buffers.''' Chris@56: Chris@56: import vampyhost Chris@75: import frames Chris@75: import load Chris@68: Chris@89: def process_frames_with_plugin(ff, sample_rate, step_size, plugin, outputs): Chris@89: Chris@98: out_indices = dict([(id, plugin.get_output(id)["output_index"]) Chris@98: for id in outputs]) Chris@89: plugin.reset() Chris@89: fi = 0 Chris@89: Chris@89: for f in ff: Chris@89: timestamp = vampyhost.frame_to_realtime(fi, sample_rate) Chris@89: results = plugin.process_block(f, timestamp) Chris@89: # results is a dict mapping output number -> list of feature dicts Chris@89: for o in outputs: Chris@89: ix = out_indices[o] Chris@89: if ix in results: Chris@89: for r in results[ix]: Chris@89: yield { o: r } Chris@89: fi = fi + step_size Chris@89: Chris@89: results = plugin.get_remaining_features() Chris@89: for o in outputs: Chris@89: ix = out_indices[o] Chris@89: if ix in results: Chris@89: for r in results[ix]: Chris@89: yield { o: r } Chris@89: Chris@89: Chris@89: def process(data, sample_rate, key, output = "", parameters = {}): Chris@98: """Process audio data with a Vamp plugin, and make the results from a Chris@98: single plugin output available as a generator. Chris@98: Chris@98: The provided data should be a 1- or 2-dimensional list or NumPy Chris@98: array of floats. If it is 2-dimensional, the first dimension is Chris@98: taken to be the channel count. Chris@98: Chris@98: The returned results will be those calculated by the plugin with Chris@98: the given key and returned through its output with the given Chris@98: output identifier. If the requested output is the empty string, Chris@98: the first output provided by the plugin will be used. Chris@98: Chris@98: If the parameters dict is non-empty, the plugin will be configured Chris@98: by setting its parameters according to the (string) key and Chris@98: (float) value data found in the dict. Chris@98: Chris@98: This function acts as a generator, yielding a sequence of result Chris@98: features as it obtains them. Each feature is represented as a Chris@98: dictionary containing, optionally, timestamp and duration Chris@98: (RealTime objects), label (string), and a 1-dimensional array of Chris@98: float values. Chris@98: """ Chris@89: Chris@89: plugin, step_size, block_size = load.load_and_configure(data, sample_rate, key, parameters) Chris@89: Chris@89: if output == "": Chris@89: output = plugin.get_output(0)["identifier"] Chris@89: Chris@89: ff = frames.frames_from_array(data, step_size, block_size) Chris@89: Chris@89: for r in process_frames_with_plugin(ff, sample_rate, step_size, plugin, [output]): Chris@89: yield r[output] Chris@76: Chris@89: plugin.unload() Chris@89: Chris@76: Chris@98: def process_frames(ff, sample_rate, step_size, key, output = "", parameters = {}): Chris@98: """Process audio data with a Vamp plugin, and make the results from a Chris@98: single plugin output available as a generator. Chris@95: Chris@98: The provided data should be an enumerable sequence of time-domain Chris@98: audio frames, of which each frame is 2-dimensional list or NumPy Chris@98: array of floats. The first dimension is taken to be the channel Chris@98: count, and the second dimension the frame or block size. The Chris@98: step_size argument gives the increment in audio samples from one Chris@98: frame to the next. Each frame must have the same size. Chris@95: Chris@98: The returned results will be those calculated by the plugin with Chris@98: the given key and returned through its output with the given Chris@98: output identifier. If the requested output is the empty string, Chris@98: the first output provided by the plugin will be used. Chris@95: Chris@98: If the parameters dict is non-empty, the plugin will be configured Chris@98: by setting its parameters according to the (string) key and Chris@98: (float) value data found in the dict. Chris@98: Chris@98: This function acts as a generator, yielding a sequence of result Chris@98: features as it obtains them. Each feature is represented as a Chris@98: dictionary containing, optionally, timestamp and duration Chris@98: (RealTime objects), label (string), and a 1-dimensional array of Chris@98: float values. Chris@98: """ Chris@98: Chris@98: plugin = vampyhost.load_plugin(key, sample_rate, Chris@98: vampyhost.ADAPT_INPUT_DOMAIN + Chris@98: vampyhost.ADAPT_BUFFER_SIZE + Chris@98: vampyhost.ADAPT_CHANNEL_COUNT) Chris@98: Chris@98: fi = 0 Chris@98: channels = 0 Chris@98: block_size = 0 Chris@98: Chris@98: if output == "": Chris@98: out_index = 0 Chris@98: else: Chris@98: out_index = plugin.get_output(output)["output_index"] Chris@95: Chris@98: for f in ff: Chris@95: Chris@98: if fi == 0: Chris@98: channels = f.shape[0] Chris@98: block_size = f.shape[1] Chris@98: plugin.set_parameter_values(parameters) Chris@98: if not plugin.initialise(channels, step_size, block_size): Chris@98: raise "Failed to initialise plugin" Chris@98: Chris@98: timestamp = vampyhost.frame_to_realtime(fi, sample_rate) Chris@98: results = plugin.process_block(f, timestamp) Chris@98: # results is a dict mapping output number -> list of feature dicts Chris@98: if out_index in results: Chris@98: for r in results[out_index]: Chris@98: yield r Chris@98: Chris@98: fi = fi + step_size Chris@98: Chris@98: if fi > 0: Chris@98: results = plugin.get_remaining_features() Chris@98: if out_index in results: Chris@98: for r in results[out_index]: Chris@98: yield r Chris@98: Chris@95: plugin.unload() Chris@95: Chris@95: Chris@84: def process_multiple_outputs(data, sample_rate, key, outputs, parameters = {}): Chris@68: #!!! docstring Chris@68: Chris@89: plugin, step_size, block_size = load.load_and_configure(data, sample_rate, key, parameters) Chris@64: Chris@84: ff = frames.frames_from_array(data, step_size, block_size) Chris@64: Chris@89: for r in process_frames_with_plugin(ff, sample_rate, step_size, plugin, outputs): Chris@89: yield r Chris@61: Chris@89: plugin.unload()