comparison ipcluster/test_sonic_annotator_notimeside.py @ 0:e34cf1b6fe09 tip

commit

+# Part of DML (Digital Music Laboratory)
+# Copyright 2014-2015 Daniel Wolff, City University
+ 
+# 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/local/spark-1.0.0-bin-hadoop2/bin/spark-submit
+# -*- coding: utf-8 -*-
+__author__="wolffd"
+__date__ ="$11-Jul-2014 15:31:01$"
+import sys
+import time
+import os
+import hashlib
+from IPython.parallel import Client
+
+# this is the main routine to be submmitted as a spark job
+#
+#
+# Running python applications through ./bin/pyspark is deprecated as of Spark 1.0.
+# Use ./bin/spark-submit <python file> --py-files sonic_annotator_vamp.py
+# you can also provide a zip of all necessary python files
+#
+# @param string audiopath root of the folder structure to be traversed
+# @param string transform_file path to the .n3 turtle file describing the transform
+#def main(audiopath = './',
+#         transform_file = 'silvet_settings.n3',
+#         masterip = '0.0.0.0):
+
+def main(audiopath, transform_path, out_path = ''):
+    print "iPCluster implementation for Vamp processing"
+    
+    # ---
+    # initialise ipcluster
+    # ---
+    #time.sleep(20)
+    rc = Client()
+    nb_core = len(rc.ids)
+    lview = rc.load_balanced_view()
+    lview.block = False # asynch now
+    dview = rc[:]
+    dview.block = True
+    
+    # import libraries
+    with dview.sync_imports():
+        import sys
+        import os
+        import sonic_annotator_vamp
+    
+    # here traverse the file structure
+    data = []
+    count = 0
+    for (dirpath, dirnames, filenames) in os.walk(audiopath):
+        for file in filenames:
+            print '\rChecked %d files' % (count), 
+            count = count + 1
+            if file.endswith(".wav") or file.endswith(".mp3") or file.endswith(".flac"):
+                data.append(os.path.join(dirpath, file).replace('\\','/'))
+    # count jobs
+    njobs = len(data)
+
+    
+    # we now allow
+    if transform_path.endswith(".n3"):
+        transform_files = [transform_path]
+    else:
+        transform_files = []
+        for file in os.listdir(transform_path):
+            if file.endswith(".n3"):
+                transform_files.append(transform_path + file)
+    
+    for transform_file in transform_files:
+        # 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)
+        hash = str(hasher.hexdigest())
+        
+        # create action containing data and parameter file
+        action = [(x,transform_file,hash,out_path) for x in data]
+        
+        # output the current task
+        tpath = os.path.split(transform_file)
+        print "Using " + tpath[1] + " on " + str(njobs) + " files"
+        
+        # ---
+        # do the work!
+        # ---
+        ar = lview.map(sonic_annotator_vamp.transform, action)
+
+        # asynch process output
+        tic = time.time()
+        while True:
+            
+            # update time used
+            toc = time.time()-tic
+            
+            # update progress
+            msgset = set(ar.msg_ids)
+            completed = len(msgset.difference(rc.outstanding))
+            pending = len(msgset.intersection(rc.outstanding))
+            
+            if completed > 0:
+                timerem = ((toc/completed) * pending) / 3600.0
+                print '\rRunning %3.2f hrs: %3.2f percent. %d done, %d pending, approx %3.2f hrs' % (toc / 3600.0, completed/(pending+completed*1.0) * 100.0,completed, pending, timerem),
+            
+            if ar.ready():
+                print '\n'
+                break
+            time.sleep(1)
+            
+        toc = time.time()-tic
+        #print ar.get()
+        print '\rProcessed %d files in %3.2f hours.' % (njobs,toc / 3600.0)
+        print '\n'
+    
+    # output
+    #print(result)
+    #thefile = open(audiopath + tpath[1] + '.txt', 'w')
+    #for item in result:
+    #    thefile.write("%s\n" % item)
+    #close(thefile)
+
+if __name__ == "__main__":
+    if len(sys.argv) >= 3:
+        main(sys.argv[1],sys.argv[2])
+    else:
+        main(audiopath = '/audio', transform_path = 'dml_processing/sonic_annotator/vamp_plugins/bbc_speechmusic.n3', out_path = './')
+