view pyspark/sonic-annotator-notimeside/sonic_annotator_vamp.py @ 0:e34cf1b6fe09 tip

commit

# Part of DML (Digital Music Laboratory)
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
# 
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
# 
# You should have received a copy of the GNU General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA

#!/usr/bin/python
# -*- coding: utf-8 -*-
"""
Created on Fri Oct 11 13:22:37 2013

@author: thomas
"""

from __future__ import division

import sys
import shutil
import os
import errno
import subprocess
import time 
import random
import hashlib

# uses a separate console process to achieve the file conversion
def vamp_host_process(argslist):
    #"""Call sonic annotator"""

    vamp_host = 'sonic-annotator'
    command = [vamp_host]
    command.extend(argslist)

    # which sa version?
    #p = subprocess.Popen([vamp_host, '-v'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    #print(p.stdout.read())


    #stdout = subprocess.check_output(command, stderr=subprocess.STDOUT)
    time.sleep(random.random()*1.0)
    p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    text = p.stdout.read()
    retcode = p.wait()
    if (retcode==0):
        print "Finished."
        return 1
    else:
        print "Error " + text
        return 0
    
    #time.sleep(random.random()*1.0)
    #print command
    #subprocess.call(command,shell=True)
    return 1


# processes the given file using a vamp plugin and sonic annotator
# @param string wav_file file to be processed
def transform(wav_file = 'sweep.mp3',
              transform_file = 'bbc_speechmusic.n3'):

    # get transform hash
    BLOCKSIZE = 65536
    hasher = hashlib.sha1()
    with open(transform_file, 'rb') as afile:
        buf = afile.read(BLOCKSIZE)
        while len(buf) > 0:
            hasher.update(buf)
            buf = afile.read(BLOCKSIZE)
    tpath = os.path.split(transform_file)
    
    # prepare output directory:
    # get audio file subpath
    # create directory _Features for output if doesnt exist
    spath = os.path.split(wav_file)
    hash = str(hasher.hexdigest())
    featpath = spath[0] + '/_Analysis/' + tpath[1] + "_" + hash[:5]
    
    #if not os.path.exists(featpath):
    #    os.makedirs(featpath)
    print 'Creating directory' + featpath
    try:
        os.makedirs(featpath)
    except OSError as exception:
        if exception.errno!= errno.EEXIST:
            raise   
    
    # copy transform file into directory
    shutil.copy(transform_file, featpath + '/' +  tpath[1][:-3] + '_' + hash[:5] + '.n3')

    #./sonic-annotator -t silvet_settings.n3 input.wav -w csv
    # prepare arguments
    args = ['-t', transform_file, wav_file, '-w', 'csv', '-w', 'rdf',
            '--rdf-basedir',featpath,'--csv-basedir',featpath, '--rdf-many-files', '--rdf-append']
    #args = ['-t', transform_file, wav_file, '-w', 'csv', '--csv-force','--csv-basedir',featpath]


    print "Analysing " + wav_file
    
    result = vamp_host_process(args)
    # execute vamp host
    return [wav_file, result]


# entry function only for testing
if __name__ == "__main__":
    if len(sys.argv) >= 2:
        transform(sys.argv[1])
    else:
        transform()