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@112: import vamp.load
Chris@112: import vamp.process
Chris@112: import vamp.frames
Chris@89: 
Chris@93: import numpy as np
Chris@93: 
Chris@94: def get_feature_step_time(sample_rate, step_size, output_desc):
Chris@111:     if output_desc["sampleType"] == vampyhost.ONE_SAMPLE_PER_STEP:
Chris@94:         return vampyhost.frame_to_realtime(step_size, sample_rate)
Chris@111:     elif output_desc["sampleType"] == vampyhost.FIXED_SAMPLE_RATE:
Chris@111:         return vampyhost.RealTime('seconds', 1.0 / output_desc["sampleRate"])
Chris@94:     else:
Chris@94:         return 1
Chris@72: 
Chris@85: def timestamp_features(sample_rate, step_size, output_desc, features):
Chris@85:     n = -1
Chris@111:     if output_desc["sampleType"] == vampyhost.ONE_SAMPLE_PER_STEP:
Chris@85:         for f in features:
Chris@85:             n = n + 1
Chris@85:             t = vampyhost.frame_to_realtime(n * step_size, sample_rate)
Chris@85:             f["timestamp"] = t
Chris@85:             yield f
Chris@111:     elif output_desc["sampleType"] == vampyhost.FIXED_SAMPLE_RATE:
Chris@111:         output_rate = output_desc["sampleRate"]
Chris@85:         for f in features:
Chris@85:             if "has_timestamp" in f:
Chris@85:                 n = int(f["timestamp"].to_float() * output_rate + 0.5)
Chris@85:             else:
Chris@85:                 n = n + 1
Chris@85:             f["timestamp"] = vampyhost.RealTime('seconds', float(n) / output_rate)
Chris@85:             yield f
Chris@85:     else:
Chris@85:         for f in features:
Chris@85:             yield f
Chris@72: 
Chris@93: def fill_timestamps(results, sample_rate, step_size, output_desc):
Chris@93: 
Chris@93:     output = output_desc["identifier"]
Chris@93:     
Chris@93:     selected = [ r[output] for r in results ]
Chris@93: 
Chris@93:     stamped = timestamp_features(sample_rate, step_size, output_desc, selected)
Chris@93: 
Chris@93:     for s in stamped:
Chris@93:         yield s
Chris@93: 
Chris@93: def deduce_shape(output_desc):
Chris@111:     if output_desc["hasDuration"]:
Chris@93:         return "individual"
Chris@111:     if output_desc["sampleType"] == vampyhost.VARIABLE_SAMPLE_RATE:
Chris@93:         return "individual"
Chris@111:     if not output_desc["hasFixedBinCount"]:
Chris@93:         return "individual"
Chris@111:     if output_desc["binCount"] == 0:
Chris@93:         return "individual"
Chris@111:     if output_desc["binCount"] == 1:
Chris@94:         return "vector"
Chris@94:     return "matrix"
Chris@93: 
Chris@88: 
Chris@95: def reshape(results, sample_rate, step_size, output_desc):
Chris@71: 
Chris@95:     output = output_desc["identifier"]
Chris@93:     shape = deduce_shape(output_desc)
Chris@94:     out_step = get_feature_step_time(sample_rate, step_size, output_desc)
Chris@89: 
Chris@93:     if shape == "vector":
Chris@94:         rv = ( out_step,
Chris@96:                np.array([r[output]["values"][0] for r in results], np.float32) )
Chris@93:     elif shape == "matrix":
Chris@97:         #!!! todo: check that each feature has the right number of bins?
Chris@97:         outseq = [r[output]["values"] for r in results]
Chris@97:         rv = ( out_step, np.array(outseq, np.float32) )
Chris@93:     else:
Chris@93:         rv = list(fill_timestamps(results, sample_rate, step_size, output_desc))
Chris@89: 
Chris@95:     return rv
Chris@95: 
Chris@95:         
Chris@96: def collect(data, sample_rate, key, output = "", parameters = {}):
Chris@99:     """Process audio data with a Vamp plugin, and make the results from a
Chris@99:     single plugin output available as a single structure.
Chris@95: 
Chris@99:     The provided data should be a 1- or 2-dimensional list or NumPy
Chris@99:     array of floats. If it is 2-dimensional, the first dimension is
Chris@99:     taken to be the channel count.
Chris@99: 
Chris@99:     The returned results will be those calculated by the plugin with
Chris@99:     the given key and returned through its output with the given
Chris@99:     output identifier. If the requested output is the empty string,
Chris@99:     the first output provided by the plugin will be used.
Chris@99: 
Chris@99:     If the parameters dict is non-empty, the plugin will be configured
Chris@99:     by setting its parameters according to the (string) key and
Chris@99:     (float) value data found in the dict.
Chris@99: 
Chris@99:     The structure in which the results are returned depends upon the
Chris@99:     output descriptor for the requested plugin output, as follows:
Chris@99: 
Chris@99:     * If the plugin output emits single-valued features at a fixed
Chris@99:       sample-rate, then this function will return a tuple of step time
Chris@99:       (the time in seconds between consecutive feature values) and a
Chris@99:       one-dimensional NumPy array of feature values. An example of
Chris@99:       such a feature might be a loudness curve against time.
Chris@99: 
Chris@99:     * If the plugin output emits multiple-valued features, with an
Chris@99:       equal number of bins per feature, at a fixed sample-rate, then
Chris@99:       this function will return a tuple of step time (the time in
Chris@99:       seconds between consecutive feature values) and a
Chris@99:       two-dimensional NumPy array of feature values. An example of
Chris@99:       such a feature might be a spectrogram.
Chris@99: 
Chris@99:     * Otherwise this function will return a list of features, where
Chris@99:       each feature is represented as a dictionary containing a
Chris@99:       timestamp (always) and a duration (optionally), a label
Chris@99:       (string), and a 1-dimensional array of float values.
Chris@99: 
Chris@99:     If you would prefer to obtain features as they are calculated
Chris@99:     (where the plugin supports this) and with the format in which the
Chris@99:     plugin returns them, via an asynchronous generator function, use
Chris@99:     vamp.process() instead.
Chris@99:     """
Chris@100: 
Chris@112:     plugin, step_size, block_size = vamp.load.load_and_configure(data, sample_rate, key, parameters)
Chris@95: 
Chris@95:     if output == "":
Chris@95:         output_desc = plugin.get_output(0)
Chris@95:         output = output_desc["identifier"]
Chris@95:     else:
Chris@95:         output_desc = plugin.get_output(output)
Chris@95: 
Chris@112:     ff = vamp.frames.frames_from_array(data, step_size, block_size)
Chris@95: 
Chris@112:     results = vamp.process.process_with_initialised_plugin(ff, sample_rate, step_size, plugin, [output])
Chris@95: 
Chris@95:     rv = reshape(results, sample_rate, step_size, output_desc)
Chris@95: 
Chris@89:     plugin.unload()
Chris@93:     return rv
Chris@93: