1 #!/usr/bin/python  2 # -*- coding: utf-8 -*-  3 # __author__ = "JieYao"  4   5 import os  6 import string  7 import argparse  8 import shutil  9 import json 10 import time 11 import subprocess 12  13 usage = "该程序为框架下Tool的一键化测试工具，支持连续测试多组数据。会提取对应的测试结果文件放到指定目录。\n" 14 parser = argparse.ArgumentParser(description = usage) 15 parser.add_argument("-c", "--code_file", help="源码文件路径", required = True) 16 parser.add_argument("-i", "--id", help="用户ID", required = True) 17 parser.add_argument("-type", "--type", help="工具所属流程和模块，格式如\"meta.beta_diversity.pca\"，\"meta.otunetwork\"", required = True) 18 parser.add_argument("-p", "--para_file", help="参数文件路径,文件中每行代表一次测试，各个参数用tab隔开，以name:value的形式写，不需要引号", required = True) 19 parser.add_argument("-o", "--output_dir", help="重设输出文件路径", required = False, default="./") 20 args = vars(parser.parse_args()) 21  22  23 args[‘code_file‘] = os.path.abspath(args[‘code_file‘]) 24 args[‘para_file‘] = os.path.abspath(args[‘para_file‘]) 25 args[‘output_dir‘] = os.path.abspath(args[‘output_dir‘]) 26  27 try: 28     shutil.copyfile(args[‘code_file‘], "/mnt/ilustre/users/sanger-dev/biocluster/src/mbio/tools/" + args[‘type‘].replace(".", "/") + ".py") 29 except: 30     print  "代码文件copy到%s下时出现错误，请检查type参数" % "/mnt/ilustre/users/sanger-dev/biocluster/src/mbio/tools/" + args[‘type‘].replace(".", "/") + ".py" 31     exit(0) 32  33 if args[‘output_dir‘] and not os.path.exists(args[‘output_dir‘]): 34     os,mkdir(args[‘output_dir‘]) 35  36 para_data =http://www.mamicode.com/ [] 37 for s in open(args[‘para_file‘]).readlines(): 38     s = s.strip().split() 39     if not s: 40         continue 41     tmp_dict = dict() 42     for data in s: 43         position = data.find(":") 44         tmp_dict[data[:position]] = data[position+1 : len(data)] 45     para_data += [tmp_dict] 46  47 option_type = dict() 48 with open(args[‘code_file‘], "r") as tmp_file: 49     for text in tmp_file.readlines(): 50         if "name" in text and "type" in text: 51             text = text.strip() 52             if text[-1] != "}": 53                 text = text[0:-1] 54             try: 55                 data =http://www.mamicode.com/ json.loads(text) 56                 if "type" in data.keys() and "name" in data.keys(): 57                     option_type[data["name"]] = data["type"] 58             except: 59                 continue 60  61 std_option = dict() 62  63 file_name = os.path.split(args[‘code_file‘])[1] 64 json_name = "single_" + os.path.splitext(file_name)[0] + ".json" 65 py_name = args["output_dir"] + ‘/test_single_‘ + file_name 66 date = time.strftime(‘%Y%m%d‘,time.localtime(time.time())) 67 log_file = "/mnt/ilustre/users/sanger-dev/workspace/%s/Single_%s" %(date, args[‘id‘]) 68  69 with open(py_name, "w") as tmp_file: 70     tmp_file.write("#!/mnt/ilustre/users/sanger-dev/app/program/Python/bin/python\n") 71     tmp_file.write("from mbio.workflows.single import SingleWorkflow\n") 72     tmp_file.write("from biocluster.wsheet import Sheet\n") 73     tmp_file.write("wsheet = Sheet(\"%s\")\n" % json_name) 74     tmp_file.write("wf = SingleWorkflow(wsheet)\n") 75     tmp_file.write("wf.run()\n") 76  77 for i in range(len(para_data)): 78     for keys in para_data[i].keys(): 79         if keys in option_type.keys(): 80             if option_type[keys] == "bool": 81                 para_data[i][keys] = bool(para_data[i][keys]) 82             elif option_type[keys] == "int": 83                 para_data[i][keys] = int(para_data[i][keys]) 84             elif option_type[keys] == "float": 85                 para_data[i][keys] = float(para_data[i][keys]) 86             else: 87                 continue 88     std_option[‘type‘] = "tool" 89     std_option[‘name‘] = args[‘type‘] 90     std_option[‘options‘] = para_data[i] 91     std_option[‘id‘] = args[‘id‘] 92  93     with open(json_name, "w") as tmp_file: 94         tmp_file.write(json.dumps(std_option, indent=4)) 95  96     task_ID = "" 97     print "开始第%d组参数的测试。" %(i+1) 98     try: 99         os.system("python %s > /dev/null" % py_name)100         print "第%d组参数测试运行完成" %(i+1)101     except:102         print "第%d组参数测试运行失败" %(i+1)103         continue104     for text in open(log_file + "/log.txt", "r").readlines():105         if "ID" not in text:106             continue107         text = text.strip()[text.find("ID")+3:]108         while len(text)>0 and text[0] not in "1234567890":109             text = text[1:]110         task_ID = text[:5]111     group_name = args[‘output_dir‘] + "/Option_Group_%d_" %(i+1)112     if os.path.exists(group_name + task_ID):113         shutil.rmtree(group_name + task_ID)114     os.mkdir(group_name + task_ID)115     calc_path = log_file + "/" + os.path.splitext(file_name)[0].capitalize()116     for tmp_name in os.listdir(calc_path):117         files = os.path.join(calc_path, tmp_name)118         if os.path.isdir(files):119             os.system("cp -r %s %s" %(files, group_name + task_ID))120         if task_ID in files:121             os.system("cp %s %s" %(files, group_name + task_ID))122     shutil.move(json_name, group_name + task_ID)