def simpleTest2():

    wf = PypeWorkflow()

    f1 = makePypeLocalFile("test.fa")
    f2 = makePypeLocalFile("ref.fa")
    f3 = makePypeLocalFile("aln.txt", readOnly=False)
    f4 = makePypeLocalFile("aln2.txt", readOnly=False)

    os.system("touch %s" % f1.localFileName)
    os.system("touch %s" % f2.localFileName)
    
    @PypeTask(inputDataObjs={"fasta":f1, "ref":f2},
              outputDataObjs={"aln":f3},
              parameters={"a":10}, **{"b":12})
    def testTask(*argv, **kwargv):
        print("testTask is running")
        for ft, f in testTask.outputDataObjs.iteritems():
            #os.system("touch %s" % f.localFileName)
            runShellCmd(["touch", "%s" % f.localFileName])
            runShellCmd(["sleep", "5" ])

    @PypeTask(inputDataObjs={"fasta":f1, "aln":f3},
              outputDataObjs={"aln2":f4},
              parameters={"a":10}, **{"b":12})
    def testTask2(*argv, **kwargv):
        print("testTask2 is running")
        for ft, f in testTask2.outputDataObjs.iteritems():
            #os.system("touch %s" % f.localFileName)
            runShellCmd(["touch", "%s" % f.localFileName])
        
    #wf.addObjects([f1,f2,f3,f4]) wf.addObjects([testTask, testTask2])
    
    wf.addTasks([testTask, testTask2])

    print (wf.RDFXML)
    print (wf.graphvizDot)

    #aGraph = PypeGraph(wf._RDFGraph) print(aGraph.tSort())

    wf.refreshTargets([f4])

    print("re-touch f1")
    os.system("sleep 1;touch %s;" % f1.localFileName)
    wf.refreshTargets([f4])

    print("re-touch f3")
    os.system("sleep 1;touch %s;" % f3.localFileName)

def create_daligner_tasks(wd, db_prefix, db_file, rdb_build_done, config, pread_aln = False):

    job_id = 0
    tasks = []
    tasks_out = {}
    with open(os.path.join(wd,  "run_jobs.sh")) as f :
        for l in f :
            l = l.strip().split()
            if l[0] == "daligner":
                try:
                    os.makedirs(os.path.join( wd, "./job_%05d" % job_id))
                except OSError:
                    pass
                os.system("cd %s/job_%05d;ln -sf ../.%s.bps .; ln -sf ../.%s.idx .; ln -sf ../%s.db ." % (wd, job_id, db_prefix, db_prefix, db_prefix) )
                job_done = makePypeLocalFile(os.path.abspath( "%s/job_%05d/job_%05d_done" % (wd, job_id, job_id)  ))
                if pread_aln == True:
                    l[0] = "daligner_p"
                parameters =  {"daligner_cmd": " ".join(l),
                               "cwd": os.path.join(wd, "job_%05d" % job_id),
                               "job_id": job_id,
                               "config": config}
                make_daligner_task = PypeTask( inputs = {"rdb_build_done": rdb_build_done},
                                               outputs = {"job_done": job_done},
                                               parameters = parameters,
                                               TaskType = PypeThreadTaskBase,
                                               URL = "task://localhost/d_%05d_%s" % (job_id, db_prefix) )
                daligner_task = make_daligner_task ( run_daligner )
                tasks.append( daligner_task )
                tasks_out[ "ajob_%d" % job_id ] = job_done
                job_id += 1
    return tasks, tasks_out

def create_daligner_tasks(wd, db_prefix, db_file, rdb_build_done, config, pread_aln=False):
    job_id = 0
    tasks = []
    tasks_out = {}

    nblock = 1
    new_db = True
    if os.path.exists(fn(db_file)):
        with open(fn(db_file)) as f:
            for l in f:
                l = l.strip().split()
                if l[0] == "blocks" and l[1] == "=":
                    nblock = int(l[2])
                    new_db = False
                    break

    for pid in xrange(1, nblock + 1):
        support.make_dirs("%s/m_%05d" % (wd, pid))

    with open(os.path.join(wd, "run_jobs.sh")) as f:
        for l in f:
            l = l.strip()
            job_uid = hashlib.md5(l).hexdigest()
            job_uid = job_uid[:8]
            l = l.split()
            if l[0] == "daligner":
                support.make_dirs(os.path.join(wd, "./job_%s" % job_uid))
                call = "cd %s/job_%s;ln -sf ../.%s.bps .; ln -sf ../.%s.idx .; ln -sf ../%s.db ." % (
                    wd,
                    job_uid,
                    db_prefix,
                    db_prefix,
                    db_prefix,
                )
                rc = os.system(call)
                if rc:
                    raise Exception("Failure in system call: %r -> %d" % (call, rc))
                job_done = makePypeLocalFile(os.path.abspath("%s/job_%s/job_%s_done" % (wd, job_uid, job_uid)))
                if pread_aln == True:
                    l[0] = "daligner_p"
                parameters = {
                    "daligner_cmd": " ".join(l),
                    "cwd": os.path.join(wd, "job_%s" % job_uid),
                    "job_uid": job_uid,
                    "config": config,
                    "nblock": nblock,
                    "db_prefix": db_prefix,
                }
                make_daligner_task = PypeTask(
                    inputs={"rdb_build_done": rdb_build_done},
                    outputs={"job_done": job_done},
                    parameters=parameters,
                    TaskType=PypeThreadTaskBase,
                    URL="task://localhost/d_%s_%s" % (job_uid, db_prefix),
                )
                daligner_task = make_daligner_task(run_daligner)
                tasks.append(daligner_task)
                tasks_out["ajob_%s" % job_uid] = job_done
                job_id += 1
    return tasks, tasks_out

def main():
    lfn = 'logging-cfg.json'
    if os.path.exists(lfn):
        logging.config.dictConfig(json.load(open(lfn)))
    else:
        logging.basicConfig()
        logging.getLogger().setLevel(logging.NOTSET)
        try:
            import logging_tree
            logging_tree.printout()
        except ImportError:
            pass
    log.debug('DEBUG LOGGING ON')
    log.warning('Available via env: JOB_TYPE={}, SLEEP_S={}'.format(
        JOB_TYPE, SLEEP_S))
    exitOnFailure=False
    concurrent_jobs=2
    #Workflow = pypeflow.controller.PypeThreadWorkflow
    Workflow = PypeProcWatcherWorkflow
    Workflow.setNumThreadAllowed(concurrent_jobs, concurrent_jobs)
    wf = Workflow(job_type=JOB_TYPE)

    par = dict(sleep_s=SLEEP_S)
    DIR ='mytmp'
    makedirs(DIR)
    f0 = makePypeLocalFile('mytmp/f0')
    f1 = makePypeLocalFile('mytmp/f1')
    make_task = PypeTask(
            #inputs = {'f': f},
            outputs = {'f0': f0},
            parameters = par,
            TaskType = MyFakePypeThreadTaskBase)
    task = make_task(taskrun0)
    wf.addTasks([task])
    make_task = PypeTask(
            inputs = {'f0': f0},
            outputs = {'f1': f1},
            parameters = par,
            TaskType = MyFakePypeThreadTaskBase)
    task = make_task(taskrun1)
    wf.addTasks([task])
    wf.refreshTargets([task])

def create_consensus_tasks(wd, db_prefix, config, p_ids_merge_job_done):
    consensus_tasks = []
    consensus_out ={}
    # Unlike the merge tasks, consensus occurs in a single directory.
    rdir = os.path.join(wd, 'preads')
    mkdir(rdir)
    for p_id, job_done in p_ids_merge_job_done:
        out_file = makePypeLocalFile(os.path.abspath("%s/preads/out.%05d.fasta" % (wd, p_id)))
        out_done = makePypeLocalFile(os.path.abspath("%s/preads/c_%05d_done" % (wd, p_id)))
        parameters =  {"cwd": rdir,
                       "job_id": p_id, 
                       "prefix": db_prefix,
                       "config": config}
        make_c_task = PypeTask(inputs = {"job_done": job_done},
                               outputs = {"out_file": out_file, "out_done": out_done},
                               parameters = parameters,
                               TaskType = PypeThreadTaskBase,
                               URL = "task://localhost/ct_%05d" % p_id)
        c_task = make_c_task(task_run_consensus)
        consensus_tasks.append(c_task)
        consensus_out["cjob_%d" % p_id] = out_done 
    return consensus_tasks, consensus_out

def create_daligner_tasks(run_jobs_fn, wd, db_prefix, db_file, rdb_build_done, config, pread_aln=False):
    job_id = 0
    tasks = []
    tasks_out = {}

    nblock = get_nblock(fn(db_file))

    xform_script = get_script_xformer(pread_aln)

    line_count = 0
    job_descs = get_daligner_job_descriptions(open(run_jobs_fn), db_prefix)
    for desc, bash in job_descs.iteritems():
        # job_uid = hashlib.md5(bash).hexdigest()
        # job_uid = job_uid[:8]
        job_uid = "%08d" % line_count
        line_count += 1

        support.make_dirs(os.path.join(wd, "./job_%s" % job_uid))
        call = "cd %s/job_%s;ln -sf ../.%s.bps .; ln -sf ../.%s.idx .; ln -sf ../%s.db ." % (
            wd,
            job_uid,
            db_prefix,
            db_prefix,
            db_prefix,
        )
        rc = system(call)
        if rc:
            raise Exception("Failure in system call: %r -> %d" % (call, rc))
        job_done = makePypeLocalFile(os.path.abspath("%s/job_%s/job_%s_done" % (wd, job_uid, job_uid)))
        bash = xform_script(bash)
        parameters = {
            "daligner_cmd": bash,
            "cwd": os.path.join(wd, "job_%s" % job_uid),
            "job_uid": job_uid,
            "config": config,
            "nblock": nblock,
            "db_prefix": db_prefix,
        }
        make_daligner_task = PypeTask(
            inputs={"rdb_build_done": rdb_build_done},
            outputs={"job_done": job_done},
            parameters=parameters,
            TaskType=PypeThreadTaskBase,
            URL="task://localhost/d_%s_%s" % (job_uid, db_prefix),
        )
        daligner_task = make_daligner_task(task_run_daligner)
        tasks.append(daligner_task)
        tasks_out["ajob_%s" % job_uid] = job_done
        job_id += 1
    return tasks, tasks_out

def main1(prog_name, input_config_fn, logger_config_fn=None):
    global fc_run_logger
    fc_run_logger = support.setup_logger(logger_config_fn)

    fc_run_logger.info("fc_run started with configuration %s", input_config_fn)
    try:
        config = support.get_dict_from_old_falcon_cfg(support.parse_config(input_config_fn))
    except Exception:
        fc_run_logger.exception('Failed to parse config "{}".'.format(input_config_fn))
        raise
    input_fofn_plf = makePypeLocalFile(config["input_fofn"])
    #Workflow = PypeProcWatcherWorkflow
    wf = PypeProcWatcherWorkflow(job_type=config['job_type'])
    run(wf, config,
            input_fofn_plf=input_fofn_plf,
            setNumThreadAllowed=PypeProcWatcherWorkflow.setNumThreadAllowed)

def create_daligner_tasks(run_jobs_fn, wd, db_prefix, rdb_build_done, config, pread_aln=False):
    tasks = []
    tasks_out = {}
    for job_uid, script in bash.scripts_daligner(run_jobs_fn, db_prefix, rdb_build_done, pread_aln):
        run_dir = "job_%s" %job_uid
        cwd = os.path.join(wd, run_dir)
        job_done_fn = os.path.abspath(os.path.join(cwd, "job_%s_done" %job_uid))
        job_done = makePypeLocalFile(job_done_fn)
        parameters =  {"daligner_script": script,
                       "cwd": cwd,
                       "job_uid": job_uid,
                       "config": config,
                       "db_prefix": db_prefix}
        make_daligner_task = PypeTask(inputs = {"rdb_build_done": rdb_build_done},
                                      outputs = {"job_done": job_done},
                                      parameters = parameters,
                                      TaskType = PypeThreadTaskBase,
                                      URL = "task://localhost/d_%s_%s" %(job_uid, db_prefix))
        daligner_task = make_daligner_task(task_run_daligner)
        tasks.append(daligner_task)
        tasks_out[ "ajob_%s" % job_uid ] = job_done
    return tasks, tasks_out

def create_merge_tasks(run_jobs_fn, wd, db_prefix, input_dep, config):
    merge_tasks = []
    merge_out = {}
    p_ids_merge_job_done = [] # for consensus

    merge_scripts = bash.scripts_merge(config, db_prefix, run_jobs_fn)
    for p_id, merge_script in merge_scripts:
        job_done = makePypeLocalFile(os.path.abspath("%s/m_%05d/m_%05d_done" % (wd, p_id, p_id)))
        parameters =  {"merge_script": merge_script,
                       "cwd": os.path.join(wd, "m_%05d" % p_id),
                       "job_id": p_id,
                       "config": config}
        make_merge_task = PypeTask(inputs = {"input_dep": input_dep},
                                   outputs = {"job_done": job_done},
                                   parameters = parameters,
                                   TaskType = PypeThreadTaskBase,
                                   URL = "task://localhost/m_%05d_%s" % (p_id, db_prefix))
        merge_task = make_merge_task(task_run_las_merge)
        merge_out["mjob_%d" % p_id] = job_done
        merge_tasks.append(merge_task)
        p_ids_merge_job_done.append((p_id, job_done))
    return merge_tasks, merge_out, p_ids_merge_job_done

def run(wf, config,
        input_fofn_plf,
        setNumThreadAllowed,
        ):
    """
    Preconditions (for now):
    * fc_run_logger
    * run_support.logger
    """
    rawread_dir = os.path.abspath("./0-rawreads")
    pread_dir = os.path.abspath("./1-preads_ovl")
    falcon_asm_dir  = os.path.abspath("./2-asm-falcon")
    script_dir = os.path.abspath("./scripts")
    sge_log_dir = os.path.abspath("./sge_log")

    for d in (rawread_dir, pread_dir, falcon_asm_dir, script_dir, sge_log_dir):
        support.make_dirs(d)

    exitOnFailure=config['stop_all_jobs_on_failure'] # only matter for parallel jobs
    concurrent_jobs = config["pa_concurrent_jobs"]
    setNumThreadAllowed(concurrent_jobs, concurrent_jobs)

    rawread_fofn_plf = makePypeLocalFile(os.path.join(rawread_dir, os.path.basename(config["input_fofn"])))
    make_fofn_abs_task = PypeTask(inputs = {"i_fofn": input_fofn_plf},
                                  outputs = {"o_fofn": rawread_fofn_plf},
                                  parameters = {},
                                  TaskType = MyFakePypeThreadTaskBase)
    fofn_abs_task = make_fofn_abs_task(task_make_fofn_abs_raw)

    wf.addTasks([fofn_abs_task])
    wf.refreshTargets([fofn_abs_task])

    if config["input_type"] == "raw":
        #### import sequences into daligner DB
        sleep_done = makePypeLocalFile( os.path.join( rawread_dir, "sleep_done") )
        rdb_build_done = makePypeLocalFile( os.path.join( rawread_dir, "rdb_build_done") )
        run_jobs = makePypeLocalFile( os.path.join( rawread_dir, "run_jobs.sh") )
        parameters = {"work_dir": rawread_dir,
                      "sge_option": config["sge_option_da"],
                      "config": config}

        length_cutoff_plf = makePypeLocalFile(os.path.join(rawread_dir, "length_cutoff"))
        raw_reads_db_plf = makePypeLocalFile(os.path.join(rawread_dir, "%s.db" % "raw_reads"))
        make_build_rdb_task = PypeTask(inputs = {"input_fofn": rawread_fofn_plf},
                                      outputs = {"rdb_build_done": rdb_build_done,
                                                 "raw_reads_db": raw_reads_db_plf,
                                                 "length_cutoff": length_cutoff_plf,
                                                 "run_jobs": run_jobs,
                                      },
                                      parameters = parameters,
                                      TaskType = MyFakePypeThreadTaskBase)
        build_rdb_task = make_build_rdb_task(task_build_rdb)

        wf.addTasks([build_rdb_task])
        wf.refreshTargets([rdb_build_done])

        raw_reads_nblock = support.get_nblock(fn(raw_reads_db_plf))
        #### run daligner
        daligner_tasks, daligner_out = create_daligner_tasks(fn(run_jobs), rawread_dir, "raw_reads", rdb_build_done, config)

        wf.addTasks(daligner_tasks)
        r_da_done = makePypeLocalFile( os.path.join( rawread_dir, "da_done") )

        parameters =  {
                "nblock": raw_reads_nblock,
        }
        make_daligner_gather = PypeTask(
                   inputs = daligner_out,
                   outputs =  {"da_done":r_da_done},
                   parameters = parameters,
                   TaskType = MyFakePypeThreadTaskBase,
                   URL = "task://localhost/rda_check" )
        check_r_da_task = make_daligner_gather(task_daligner_gather)
        wf.addTask(check_r_da_task)
        wf.refreshTargets(exitOnFailure=exitOnFailure)

        merge_tasks, merge_out, p_ids_merge_job_done = create_merge_tasks(fn(run_jobs), rawread_dir, "raw_reads", r_da_done, config)
        wf.addTasks( merge_tasks )
        wf.refreshTargets(exitOnFailure=exitOnFailure)

        if config["target"] == "overlapping":
            sys.exit(0)
        consensus_tasks, consensus_out = create_consensus_tasks(rawread_dir, "raw_reads", config, p_ids_merge_job_done)
        wf.addTasks( consensus_tasks )

        r_cns_done = makePypeLocalFile( os.path.join( rawread_dir, "cns_done") )
        pread_fofn = makePypeLocalFile( os.path.join( pread_dir,  "input_preads.fofn" ) )

        @PypeTask( inputs = consensus_out,
                   outputs =  {"cns_done":r_cns_done, "pread_fofn": pread_fofn},
                   TaskType = MyFakePypeThreadTaskBase,
                   URL = "task://localhost/cns_check" )
        def check_r_cns_task(self):
            with open(fn(self.pread_fofn),  "w") as f:
                fn_list =  glob.glob("%s/preads/out*.fasta" % rawread_dir)
                fn_list.sort()
                for fa_fn in fn_list:
                    print >>f, fa_fn
            system("touch %s" % fn(self.cns_done))
        wf.addTask(check_r_cns_task)
#.........这里部分代码省略.........

    sge_log_dir = os.path.abspath("./sge_log")

    for d in (dist_map_dir, fasta_dir, pa_dir, script_dir, celera_asm_dir,  sge_log_dir):
        try:
            os.makedirs(d)
        except:
            pass

    config = get_config(sys.argv[1])
    concurrent_jobs = config["concurrent_jobs"]
    PypeThreadWorkflow.setNumThreadAllowed(concurrent_jobs, concurrent_jobs)
    wf = PypeThreadWorkflow()


    #### Task to convert bas.h5 and bax.h5 to fasta files, it will generates two fofn files for the queries and targets
    input_h5_fofn = makePypeLocalFile(os.path.abspath( config["input_fofn_fn"] ))
    query_fa_fofn = makePypeLocalFile( os.path.join( fasta_dir, "queries.fofn" ) )
    target_fa_fofn = makePypeLocalFile( os.path.join( fasta_dir, "targets.fofn" ) )
    fasta_dump_done = makePypeLocalFile(os.path.abspath( os.path.join( fasta_dir, "fasta_dump_done") ) )
    parameters = {"fasta_dir": fasta_dir,
                  "min_length": config["length_cutoff"],
                  "min_read_score": config["RQ_threshold"]}

    @PypeTask(inputs = {"input_fofn": input_h5_fofn},
              outputs = {"fasta_dump_done": fasta_dump_done, 
                         "target_fa_fofn": target_fa_fofn,
                         "query_fa_fofn":  query_fa_fofn},
              parameters = parameters,
              TaskType = PypeThreadTaskBase)
    def h5fofn_to_fasta(self):
        os.system("h5fofn_to_fasta.py %s %s --min_length 500 --min_seed_length %d --min_read_score %f" %\

def main1(prog_name, input_config_fn, logger_config_fn=None):
    global fc_run_logger
    fc_run_logger = support.setup_logger(logger_config_fn)

    fc_run_logger.info( "fc_run started with configuration %s", input_config_fn ) 
    config = support.get_config(support.parse_config(input_config_fn))
    rawread_dir = os.path.abspath("./0-rawreads")
    pread_dir = os.path.abspath("./1-preads_ovl")
    falcon_asm_dir  = os.path.abspath("./2-asm-falcon")
    script_dir = os.path.abspath("./scripts")
    sge_log_dir = os.path.abspath("./sge_log")

    for d in (rawread_dir, pread_dir, falcon_asm_dir, script_dir, sge_log_dir):
        support.make_dirs(d)

    concurrent_jobs = config["pa_concurrent_jobs"]
    PypeThreadWorkflow.setNumThreadAllowed(concurrent_jobs, concurrent_jobs)
    wf = PypeThreadWorkflow()

    input_fofn_plf = makePypeLocalFile(os.path.basename(config["input_fofn_fn"]))
    rawread_fofn_plf = makePypeLocalFile(os.path.join(rawread_dir, os.path.basename(config["input_fofn_fn"])))
    make_fofn_abs_task = PypeTask(inputs = {"i_fofn": input_fofn_plf},
                                  outputs = {"o_fofn": rawread_fofn_plf},
                                  parameters = {},
                                  TaskType = PypeThreadTaskBase)
    fofn_abs_task = make_fofn_abs_task(task_make_fofn_abs_raw)

    wf.addTasks([fofn_abs_task])
    wf.refreshTargets([fofn_abs_task])

    if config["input_type"] == "raw":
        #### import sequences into daligner DB
        sleep_done = makePypeLocalFile( os.path.join( rawread_dir, "sleep_done") )
        rdb_build_done = makePypeLocalFile( os.path.join( rawread_dir, "rdb_build_done") ) 
        run_jobs = makePypeLocalFile( os.path.join( rawread_dir, "run_jobs.sh") ) 
        parameters = {"work_dir": rawread_dir,
                      "config": config}

        make_build_rdb_task = PypeTask(inputs = {"input_fofn": rawread_fofn_plf},
                                      outputs = {"rdb_build_done": rdb_build_done,
                                                 "run_jobs": run_jobs}, 
                                      parameters = parameters,
                                      TaskType = PypeThreadTaskBase)
        build_rdb_task = make_build_rdb_task(task_build_rdb)

        wf.addTasks([build_rdb_task])
        wf.refreshTargets([rdb_build_done]) 

        db_file = makePypeLocalFile(os.path.join( rawread_dir, "%s.db" % "raw_reads" ))
        #### run daligner
        daligner_tasks, daligner_out = create_daligner_tasks(fn(run_jobs), rawread_dir, "raw_reads", db_file, rdb_build_done, config) 

        wf.addTasks(daligner_tasks)
        #wf.refreshTargets(updateFreq = 60) # larger number better for more jobs

        r_da_done = makePypeLocalFile( os.path.join( rawread_dir, "da_done") )

        @PypeTask( inputs = daligner_out, 
                   outputs =  {"da_done":r_da_done},
                   TaskType = PypeThreadTaskBase,
                   URL = "task://localhost/rda_check" )
        def check_r_da_task(self):
            os.system("touch %s" % fn(self.da_done))
        
        wf.addTask(check_r_da_task)
        wf.refreshTargets(updateFreq = wait_time) # larger number better for more jobs, need to call to run jobs here or the # of concurrency is changed
        
        concurrent_jobs = config["cns_concurrent_jobs"]
        PypeThreadWorkflow.setNumThreadAllowed(concurrent_jobs, concurrent_jobs)
        merge_tasks, merge_out, consensus_tasks, consensus_out = create_merge_tasks(fn(run_jobs), rawread_dir, "raw_reads", r_da_done, config)
        wf.addTasks( merge_tasks )
        if config["target"] == "overlapping":
            wf.refreshTargets(updateFreq = wait_time) # larger number better for more jobs, need to call to run jobs here or the # of concurrency is changed
            sys.exit(0)
        wf.addTasks( consensus_tasks )

        r_cns_done = makePypeLocalFile( os.path.join( rawread_dir, "cns_done") )
        pread_fofn = makePypeLocalFile( os.path.join( pread_dir,  "input_preads.fofn" ) )

        @PypeTask( inputs = consensus_out, 
                   outputs =  {"cns_done":r_cns_done, "pread_fofn": pread_fofn},
                   TaskType = PypeThreadTaskBase,
                   URL = "task://localhost/cns_check" )
        def check_r_cns_task(self):
            with open(fn(self.pread_fofn),  "w") as f:
                fn_list =  glob.glob("%s/preads/out*.fasta" % rawread_dir)
                fn_list.sort()
                for fa_fn in fn_list:
                    print >>f, fa_fn
            os.system("touch %s" % fn(self.cns_done))

        wf.addTask(check_r_cns_task)
        wf.refreshTargets(updateFreq = wait_time) # larger number better for more jobs

    if config["target"] == "pre-assembly":
        sys.exit(0)

    # build pread database
    if config["input_type"] == "preads":
        pread_fofn = makePypeLocalFile(os.path.join(pread_dir, os.path.basename(config["input_fofn_fn"])))
#.........这里部分代码省略.........

def simpleTest():

    wf = PypeWorkflow() 
    
    # f1 and f2 are the mock input files
    f1 = makePypeLocalFile("test.fa")
    f2 = makePypeLocalFile("ref.fa")
    
    # f3 is the object of the expected output of the "testTask"
    f3 = makePypeLocalFile("aln.txt", readOnly=False)

    # create the mock files
    os.system("touch %s" % f1.localFileName)
    os.system("touch %s" % f2.localFileName)
   
    # the testTask will take f1 (as "testTask.fasta") and f2 (as "testTask.ref") and generate f3 (as "testTask.aln")
    @PypeTask(inputDataObjs={"fasta":f1, "ref":f2},
              outputDataObjs={"aln":f3},
              parameters={"a":10}, **{"b":12})
    def testTask(*argv, **kwargv):
        print("testTask is running")
        print("fasta input filename is %s" %  testTask.fasta.localFileName)
        for ft, f in testTask.outputDataObjs.iteritems():
            #os.system("touch %s" % f.localFileName)
            runShellCmd(["touch", "%s" % f.localFileName])
            runShellCmd(["sleep", "5" ])

    # the testTask will take f1 (as "testTask.fasta") and f3 (as "testTask.aln") and generate f4 (as "testTask.aln2")
    f4 = makePypeLocalFile("aln2.txt", readOnly=False)
    @PypeTask(inputDataObjs={"fasta":f1, "aln":f3},
              outputDataObjs={"aln2":f4},
              parameters={"a":10}, **{"b":12})
    def testTask2(*argv, **kwargv):
        print("testTask2 is running")
        for ft, f in testTask2.outputDataObjs.iteritems():
            #os.system("touch %s" % f.localFileName)
            runShellCmd(["touch", "%s" % f.localFileName])
    
    # one can add objects one by one to the workflow
    #wf.addObjects([f1,f2,f3,f4]) 
    #wf.addObjects([testTask, testTask2])
   
    # or, one can add the "tasks" into the workflow, the input and output data objects will be added automatically
    wf.addTasks([testTask, testTask2])

    #print out the RDFXML file that represents the workflow
    print (wf.RDFXML)
    #a graphviz dot for rendering the dependency graph if one
    print (wf.graphvizDot)

    # execute the workflow until f4 is updated
    wf.refreshTargets([f4])

    # mock the case that f1 is updated
    print("re-touch f1")
    os.system("sleep 1;touch %s;" % f1.localFileName)
    wf.refreshTargets([f4])

    # mock the case that f3 is updated
    print("re-touch f3")
    os.system("sleep 1;touch %s;" % f3.localFileName)

def testDistributed(runmode, cleanup):
    logger.info("test start")
    baseDir = "."
    import random
    random.seed(1984)
    #PypeThreadWorkflow.setNumThreadAllowed(20,20)
    #wf = PypeThreadWorkflow()
    PypeMPWorkflow.setNumThreadAllowed(20,20)
    wf = PypeMPWorkflow()
    allTasks = []
    for layer in range(5):
        fN = random.randint(3,7)
        fin = [None] * fN
        fout = [None] * fN
        fmut = [None] * fN
        for w in range(fN):
            fin[w] = makePypeLocalFile(baseDir + "/testdata/testfile_l%d_w%d.dat" % (layer, w) )
            fout[w] = makePypeLocalFile(baseDir + "/testdata/testfile_l%d_w%d.dat" % (layer+1, w) )
            fmut[w] = makePypeLocalFile(baseDir + "/testdata/m_testfile_l%d_w%d.dat" % (layer+1, w) )
            #wf.addObjects([fin[w], fout[w], fmut[w]])

        for w in range(fN):
            inputDataObjs = {}
            outputDataObjs = {}
            mutableDataObjs = {}
            for i in range(5):
                inputDataObjs["infile%d" % i] = random.choice(fin)

            i = 0
            for obj in random.sample(fmut,2):
                #mutableDataObjs["outfile%d" % i] = obj
                i += 1
            outputDataObjs["outfile%d" % i] = fout[w]

            shellCmd = "sleep 1\n" + "\n".join([ "echo %d %d ...  >> %s" % (layer, w, of.localFileName) for of in outputDataObjs.values() ]) + "\nsleep 10"
            shellCmd += "sleep 1\n" + "\n".join([ "echo %d %d ...  >> %s" % (layer, w, of.localFileName) for of in mutableDataObjs.values() ]) + "\nsleep 10"
            shellFileName = baseDir + "/testdata/task_l%d_w%d.sh" % (layer, w)
            shfile = open(shellFileName, 'w')
            print(shellCmd, file=shfile)
            shfile.close()

            if runmode == "internal":
                def t1(self):
                    runShellCmd(["sleep", "%d" % random.randint(0,20) ])

                    for of in self.outputDataObjs.values():
                        runShellCmd(["touch", of.localFileName])

                task = PypeTask(inputDataObjs = inputDataObjs,
                                outputDataObjs = outputDataObjs, 
                                mutableDataObjs = mutableDataObjs,
                                URL="task://internal/task_l%d_w%d" % (layer, w), 
                                TaskType=PypeThreadTaskBase) ( t1 )

            elif runmode == "localshell":
                task = PypeShellTask(inputDataObjs = inputDataObjs,
                                     outputDataObjs = outputDataObjs, 
                                     mutableDataObjs = mutableDataObjs,
                                     URL="task://localshell/task_l%d_w%d" % (layer, w), 
                                     TaskType=PypeThreadTaskBase) ( "%s" % shellFileName )

            elif runmode == "sge": 
                task = PypeSGETask(inputDataObjs = inputDataObjs,
                                   outputDataObjs = outputDataObjs, 
                                   mutableDataObjs = mutableDataObjs,
                                   URL="task://sge/task_l%d_w%d" % (layer, w), 
                                   TaskType=PypeThreadTaskBase) ( "%s" % shellFileName )

            elif runmode == "mixed":
                #distributed = random.choice( (False, True) )
                distributed = True if w % 3 == 0 else False
                task = PypeDistributibleTask(inputDataObjs = inputDataObjs,
                                   outputDataObjs = outputDataObjs,
                                   mutableDataObjs = mutableDataObjs,
                                   URL="task://sge/task_l%d_w%d" % (layer, w), 
                                   distributed=distributed,
                                   TaskType=PypeThreadTaskBase) ( "%s" % shellFileName )

            wf.addTasks([task])
            allTasks.append(task)

    for URL in wf._pypeObjects:
        prereqJobURLs = [str(u) for u in wf._RDFGraph.transitive_objects(URIRef(URL), pypeNS["prereq"])
                                        if isinstance(wf._pypeObjects[str(u)], PypeLocalFile) and str(u) != URL ]
        if len(prereqJobURLs) == 0:
            if cleanup == "1":
                os.system("echo start > %s" % wf._pypeObjects[URL].localFileName)
            pass
    wf.refreshTargets(allTasks)
    dotFile = open("test.dot","w")
    #print >>dotFile, wf.graphvizShortNameDot
    print(wf.graphvizDot, file=dotFile)
    dotFile.close()
    dotFile = open("test_short_name.dot","w")
    print(wf.graphvizShortNameDot, file=dotFile)
    dotFile.close()
    rdfFile = open("test.rdf","w")
    print(wf.RDFXML, file=rdfFile)
    rdfFile.close()
    if runmode != "internal":
#.........这里部分代码省略.........

def generate_read_to_contig_map(rawread_dir=rawread_dir, pread_dir=pread_dir, asm_dir=asm_dir):

    read_map_dir = os.path.abspath(os.path.join(asm_dir, "read_maps"))
    make_dirs(read_map_dir)

    PypeMPWorkflow.setNumThreadAllowed(12, 12)
    wf = PypeMPWorkflow()

    rawread_db = makePypeLocalFile(os.path.join(rawread_dir, "raw_reads.db"))
    rawread_id_file = makePypeLocalFile(os.path.join(rawread_dir, "raw_read_ids"))

    @PypeTask(
        inputs={"rawread_db": rawread_db},
        outputs={"rawread_id_file": rawread_id_file},
        TaskType=PypeThreadTaskBase,
        URL="task://localhost/dump_rawread_ids",
    )
    def dump_rawread_ids(self):
        rawread_db = fn(self.rawread_db)
        rawread_id_file = fn(self.rawread_id_file)
        os.system("DBshow -n %s | tr -d '>' | awk '{print $1}' > %s" % (rawread_db, rawread_id_file))

    wf.addTask(dump_rawread_ids)

    pread_db = makePypeLocalFile(os.path.join(pread_dir, "preads.db"))
    pread_id_file = makePypeLocalFile(os.path.join(pread_dir, "pread_ids"))

    @PypeTask(
        inputs={"pread_db": pread_db},
        outputs={"pread_id_file": pread_id_file},
        TaskType=PypeThreadTaskBase,
        URL="task://localhost/dump_pread_ids",
    )
    def dump_pread_ids(self):
        pread_db = fn(self.pread_db)
        pread_id_file = fn(self.pread_id_file)
        os.system("DBshow -n %s | tr -d '>' | awk '{print $1}' > %s" % (pread_db, pread_id_file))

    wf.addTask(dump_pread_ids)

    all_raw_las_files = {}
    for las_fn in glob.glob(os.path.join(rawread_dir, "raw_reads.*.las")):
        idx = las_fn.split("/")[-1]  # well, we will use regex someday to parse to get the number
        idx = int(idx.split(".")[1])
        las_file = makePypeLocalFile(las_fn)
        all_raw_las_files["r_las_%s" % idx] = las_file

    all_pread_las_files = {}
    for las_fn in glob.glob(os.path.join(pread_dir, "preads.*.las")):
        idx = las_fn.split("/")[-1]  # well, we will use regex someday to parse to get the number
        idx = int(idx.split(".")[1])
        las_file = makePypeLocalFile(las_fn)
        all_pread_las_files["p_las_%s" % idx] = las_file

    wf.refreshTargets()  # block

    sg_edges_list = makePypeLocalFile(os.path.join(asm_dir, "sg_edges_list"))
    utg_data = makePypeLocalFile(os.path.join(asm_dir, "utg_data"))
    ctg_paths = makePypeLocalFile(os.path.join(asm_dir, "ctg_paths"))

    inputs = {
        "rawread_id_file": rawread_id_file,
        "pread_id_file": pread_id_file,
        "sg_edges_list": sg_edges_list,
        "utg_data": utg_data,
        "ctg_paths": ctg_paths,
    }

    read_to_contig_map = makePypeLocalFile(os.path.join(read_map_dir, "read_to_contig_map"))

    @PypeTask(
        inputs=inputs,
        outputs={"read_to_contig_map": read_to_contig_map},
        TaskType=PypeThreadTaskBase,
        URL="task://localhost/get_ctg_read_map",
    )
    def generate_read_to_ctg_map(self):
        rawread_id_file = fn(self.rawread_id_file)
        pread_id_file = fn(self.pread_id_file)
        read_to_contig_map = fn(self.read_to_contig_map)

        pread_did_to_rid = open(pread_id_file).read().split("\n")
        rid_to_oid = open(rawread_id_file).read().split("\n")

        asm_G = AsmGraph(fn(self.sg_edges_list), fn(self.utg_data), fn(self.ctg_paths))

        pread_to_contigs = {}

        with open(read_to_contig_map, "w") as f:
            for ctg in asm_G.ctg_data:
                if ctg[-1] == "R":
                    continue
                ctg_g = asm_G.get_sg_for_ctg(ctg)
                for n in ctg_g.nodes():
                    pid = int(n.split(":")[0])

                    rid = pread_did_to_rid[pid].split("/")[1]
                    rid = int(int(rid) / 10)
                    oid = rid_to_oid[rid]
                    k = (pid, rid, oid)
#.........这里部分代码省略.........

def main1(prog_name, input_config_fn, logger_config_fn=None):
    setup_logger(logger_config_fn)

    fc_run_logger.info( "fc_run started with configuration %s", input_config_fn ) 
    config = get_config(parse_config(input_config_fn))
    rawread_dir = os.path.abspath("./0-rawreads")
    pread_dir = os.path.abspath("./1-preads_ovl")
    falcon_asm_dir  = os.path.abspath("./2-asm-falcon")
    script_dir = os.path.abspath("./scripts")
    sge_log_dir = os.path.abspath("./sge_log")

    for d in (rawread_dir, pread_dir, falcon_asm_dir, script_dir, sge_log_dir):
        make_dirs(d)

    concurrent_jobs = config["pa_concurrent_jobs"]
    PypeThreadWorkflow.setNumThreadAllowed(concurrent_jobs, concurrent_jobs)
    wf = PypeThreadWorkflow()

    input_fofn_plf = makePypeLocalFile(os.path.basename(config["input_fofn_fn"]))
    rawread_fofn_plf = makePypeLocalFile(os.path.join(rawread_dir, os.path.basename(config["input_fofn_fn"])))
    make_fofn_abs_task = PypeTask(inputs = {"i_fofn": input_fofn_plf},
                                  outputs = {"o_fofn": rawread_fofn_plf},
                                  parameters = {},
                                  TaskType = PypeThreadTaskBase)
    fofn_abs_task = make_fofn_abs_task(make_fofn_abs_raw)
    wf.addTasks([fofn_abs_task])
    wf.refreshTargets([fofn_abs_task])

    if config["input_type"] == "raw":
        #### import sequences into daligner DB
        sleep_done = makePypeLocalFile( os.path.join( rawread_dir, "sleep_done") )
        rdb_build_done = makePypeLocalFile( os.path.join( rawread_dir, "rdb_build_done") ) 
        parameters = {"work_dir": rawread_dir,
                      "config": config}

        make_build_rdb_task = PypeTask(inputs = {"input_fofn": rawread_fofn_plf},
                                      outputs = {"rdb_build_done": rdb_build_done}, 
                                      parameters = parameters,
                                      TaskType = PypeThreadTaskBase)

        build_rdb_task = make_build_rdb_task(build_rdb)

        wf.addTasks([build_rdb_task])
        wf.refreshTargets([rdb_build_done]) 
        

        db_file = makePypeLocalFile(os.path.join( rawread_dir, "%s.db" % "raw_reads" ))
        #### run daligner
        daligner_tasks, daligner_out = create_daligner_tasks( rawread_dir, "raw_reads", db_file, rdb_build_done, config) 

        wf.addTasks(daligner_tasks)
        #wf.refreshTargets(updateFreq = 60) # larger number better for more jobs

        r_da_done = makePypeLocalFile( os.path.join( rawread_dir, "da_done") )

        @PypeTask( inputs = daligner_out, 
                   outputs =  {"da_done":r_da_done},
                   TaskType = PypeThreadTaskBase,
                   URL = "task://localhost/rda_check" )
        def check_r_da_task(self):
            os.system("touch %s" % fn(self.da_done))
        
        wf.addTask(check_r_da_task)
        wf.refreshTargets(updateFreq = wait_time) # larger