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@68: ###!!! todo: move all the real code out of __init__.py
Chris@68: 
Chris@56: import vampyhost
Chris@56: import numpy
Chris@56: 
Chris@56: def listPlugins():
Chris@56:     return vampyhost.listPlugins()
Chris@56: 
Chris@56: def framesFromArray(arr, stepSize, frameSize):
Chris@56:     """Generate a list of frames of size frameSize, extracted from the input array arr at stepSize intervals"""
Chris@56:     # presumably such a function exists in many places, but I need practice
Chris@56:     assert(stepSize > 0)
Chris@56:     if arr.ndim == 1: # turn 1d into 2d array with 1 channel
Chris@56:         arr = numpy.reshape(arr, (1, arr.shape[0]))
Chris@56:     assert(arr.ndim == 2)
Chris@56:     n = arr.shape[1]
Chris@56:     i = 0
Chris@56:     while (i < n):
Chris@56:         frame = arr[:, i : i + frameSize]
Chris@56:         w = frame.shape[1]
Chris@56:         if (w < frameSize):
Chris@56:             pad = numpy.zeros((frame.shape[0], frameSize - w))
Chris@56:             frame = numpy.concatenate((frame, pad), 1)
Chris@56:         yield frame
Chris@56:         i = i + stepSize
Chris@56: 
Chris@68: 
Chris@68: def loadAndConfigureFor(data, samplerate, key, parameters):
Chris@58:     plug = vampyhost.loadPlugin(key, samplerate,
Chris@58:                                 vampyhost.AdaptInputDomain +
Chris@58:                                 vampyhost.AdaptChannelCount)
Chris@64: 
Chris@68:     plug.setParameterValues(parameters)
Chris@68: 
Chris@68:     stepSize = plug.getPreferredStepSize()
Chris@68:     blockSize = plug.getPreferredBlockSize()
Chris@68: 
Chris@68:     if blockSize == 0:
Chris@68:         blockSize = 1024
Chris@68:     if stepSize == 0:
Chris@68:         stepSize = blockSize ##!!! or blockSize/2, but check this with input domain adapter
Chris@68: 
Chris@68:     channels = 1
Chris@68:     if data.ndim > 1:
Chris@68:         channels = data.shape[0]
Chris@68: 
Chris@68:     plug.initialise(channels, stepSize, blockSize)
Chris@68:     return (plug, stepSize, blockSize)
Chris@68: 
Chris@68: 
Chris@68: def process(data, samplerate, key, parameters = {}, outputs = []):
Chris@68: #!!! docstring
Chris@68: 
Chris@68:     plug, stepSize, blockSize = loadAndConfigureFor(data, samplerate, key, parameters)
Chris@68: 
Chris@64:     plugOuts = plug.getOutputs()
Chris@68:     if plugOuts == []:
Chris@68:         return
Chris@64: 
Chris@64:     outIndices = dict(zip([o["identifier"] for o in plugOuts],
Chris@64:                           range(0, len(plugOuts))))  # id -> n
Chris@64: 
Chris@64:     for o in outputs:
Chris@64:         assert o in outIndices
Chris@64: 
Chris@66:     if outputs == []:
Chris@67:         outputs = [plugOuts[0]["identifier"]]
Chris@66: 
Chris@56:     ff = framesFromArray(data, stepSize, blockSize)
Chris@61:     fi = 0
Chris@64: 
Chris@71:     #!!! should we fill in the correct timestamps here?
Chris@71: 
Chris@61:     for f in ff:
Chris@64:         results = plug.processBlock(f, vampyhost.frame2RealTime(fi, samplerate))
Chris@64:         # results is a dict mapping output number -> list of feature dicts
Chris@71:         print("results = " + str(results))
Chris@66:         for o in outputs:
Chris@66:             if outIndices[o] in results:
Chris@66:                 for r in results[outIndices[o]]:
Chris@70:                     yield { o: r }
Chris@61:         fi = fi + stepSize
Chris@61: 
Chris@66:     results = plug.getRemainingFeatures()
Chris@66:     for o in outputs:
Chris@66:         if outIndices[o] in results:
Chris@66:             for r in results[outIndices[o]]:
Chris@70:                 yield { o: r }
Chris@66: 
Chris@66:     plug.unload()
Chris@66: 
Chris@66: 
Chris@72: ##!!!
Chris@72: ##
Chris@72: ## We could also devise a generator for the timestamps that need
Chris@72: ## filling: provide the output type & rate and get back a timestamp
Chris@72: ## generator
Chris@72: ##
Chris@72: ##!!!
Chris@72: 
Chris@72: 
Chris@71: def collect(data, samplerate, key, parameters = {}, output = ""):
Chris@71:     
Chris@71:     plug, stepSize, blockSize = loadAndConfigureFor(data, samplerate, key, parameters)
Chris@71: 
Chris@71:     plugOuts = plug.getOutputs()
Chris@71:     if plugOuts == []:
Chris@71:         return
Chris@71: 
Chris@71:     outNo = -1
Chris@71:     for n, o in zip(range(0, len(plugOuts)), plugOuts):
Chris@71:         if output == "" or o["identifier"] == output:
Chris@71:             outNo = n
Chris@71:             break
Chris@71: 
Chris@71:     assert outNo >= 0 #!!! todo proper error reporting
Chris@71: 
Chris@71:     ff = framesFromArray(data, stepSize, blockSize)
Chris@71:     fi = 0
Chris@71: 
Chris@71:     #!!! todo!
Chris@71: 
Chris@71:     plug.unload()
Chris@71:     
Chris@71:     return {}
Chris@71: 
Chris@71: