nodes: 0, module; 1, function_definition; 2, function_name:parallel_deblur; 3, parameters; 4, block; 5, identifier:inputs; 6, identifier:params; 7, identifier:pos_ref_db_fp; 8, identifier:neg_ref_dp_fp; 9, default_parameter; 10, expression_statement; 11, expression_statement; 12, expression_statement; 13, comment:# remove the irrelevant parameters; 14, expression_statement; 15, expression_statement; 16, expression_statement; 17, for_statement; 18, comment:# add the ref_db_fp (since it may be not present in the; 19, comment:# original command parameters); 20, if_statement; 21, if_statement; 22, expression_statement; 23, expression_statement; 24, expression_statement; 25, expression_statement; 26, expression_statement; 27, for_statement; 28, identifier:jobs_to_start; 29, integer:1; 30, comment:"""Dispatch execution over a pool of processors
  
      This code was adopted from the American Gut project:
      https://github.com/biocore/American-Gut/blob/master/americangut/parallel.py
  
      Parameters
      ----------
      inputs : iterable of str
          File paths to input per-sample sequence files
      params : list of str
          list of CLI parameters supplied to the deblur workflow
          (argv - first 2 are 'deblur','workflow' and are ignored)
      pos_ref_db_fp : list of str
          the indexed positive (16s) sortmerna database
          (created in the main thread)
      neg_ref_db_fp : list of str
          the indexed negative (artifacts) sortmerna database
          (created in the main thread)
      jobs_to_start : int, optional
          The number of processors on the local system to use
  
      Returns
      -------
      all_result_paths : list
          list of expected output files
      """; 31, assignment; 32, call; 33, assignment; 34, assignment; 35, assignment; 36, identifier:carg; 37, subscript; 38, block; 39, identifier:pos_ref_db_fp; 40, block; 41, identifier:neg_ref_dp_fp; 42, block; 43, call; 44, assignment; 45, call; 46, assignment; 47, call; 48, pattern_list; 49, call; 50, block; 51, identifier:logger; 52, call; 53, attribute; 54, argument_list; 55, identifier:remove_param_list; 56, list; 57, identifier:skipnext; 58, False; 59, identifier:newparams; 60, list; 61, identifier:params; 62, slice; 63, if_statement; 64, if_statement; 65, expression_statement; 66, expression_statement; 67, expression_statement; 68, expression_statement; 69, expression_statement; 70, expression_statement; 71, expression_statement; 72, attribute; 73, argument_list; 74, identifier:functor; 75, call; 76, attribute; 77, argument_list; 78, identifier:pool; 79, call; 80, attribute; 81, argument_list; 82, identifier:stdout; 83, identifier:stderr; 84, identifier:es; 85, attribute; 86, argument_list; 87, if_statement; 88, attribute; 89, argument_list; 90, identifier:logger; 91, identifier:info; 92, binary_operator:'parallel deblur started for %d inputs' % len(inputs); 93, string; 94, string; 95, string; 96, string; 97, string; 98, integer:2; 99, identifier:skipnext; 100, block; 101, comparison_operator:carg in remove_param_list; 102, block; 103, call; 104, assignment; 105, call; 106, call; 107, assignment; 108, call; 109, call; 110, identifier:logger; 111, identifier:debug; 112, binary_operator:'ready for functor %s' % newparams; 113, identifier:partial; 114, argument_list; 115, identifier:logger; 116, identifier:debug; 117, binary_operator:'ready for pool %d jobs' % jobs_to_start; 118, attribute; 119, argument_list; 120, identifier:logger; 121, identifier:debug; 122, string; 123, identifier:pool; 124, identifier:map; 125, identifier:functor; 126, identifier:inputs; 127, comparison_operator:es != 0; 128, block; 129, identifier:logging; 130, identifier:getLogger; 131, identifier:__name__; 132, string; 133, call; 134, string_content:-O; 135, string_content:--jobs-to-start; 136, string_content:--seqs-fp; 137, string_content:--pos-ref-db-fp; 138, string_content:--neg-ref-db-fp; 139, expression_statement; 140, continue_statement; 141, identifier:carg; 142, identifier:remove_param_list; 143, expression_statement; 144, continue_statement; 145, attribute; 146, argument_list; 147, identifier:new_pos_ref_db_fp; 148, call; 149, attribute; 150, argument_list; 151, attribute; 152, argument_list; 153, identifier:new_neg_ref_db_fp; 154, call; 155, attribute; 156, argument_list; 157, attribute; 158, argument_list; 159, string; 160, identifier:newparams; 161, identifier:run_functor; 162, identifier:deblur_system_call; 163, identifier:newparams; 164, string; 165, identifier:jobs_to_start; 166, identifier:mp; 167, identifier:Pool; 168, keyword_argument; 169, string_content:almost running...; 170, identifier:es; 171, integer:0; 172, raise_statement; 173, string_content:parallel deblur started for %d inputs; 174, identifier:len; 175, argument_list; 176, assignment; 177, assignment; 178, identifier:newparams; 179, identifier:append; 180, identifier:carg; 181, attribute; 182, argument_list; 183, identifier:newparams; 184, identifier:append; 185, string; 186, identifier:newparams; 187, identifier:append; 188, identifier:new_pos_ref_db_fp; 189, attribute; 190, argument_list; 191, identifier:newparams; 192, identifier:append; 193, string; 194, identifier:newparams; 195, identifier:append; 196, identifier:new_neg_ref_db_fp; 197, string_content:ready for functor %s; 198, string_content:ready for pool %d jobs; 199, identifier:processes; 200, identifier:jobs_to_start; 201, call; 202, identifier:inputs; 203, identifier:skipnext; 204, False; 205, identifier:skipnext; 206, True; 207, string; 208, identifier:join; 209, identifier:pos_ref_db_fp; 210, string_content:--pos-ref-db-fp; 211, string; 212, identifier:join; 213, identifier:neg_ref_dp_fp; 214, string_content:--neg-ref-db-fp; 215, identifier:RuntimeError; 216, argument_list; 217, string_content:,; 218, string_content:,; 219, binary_operator:"stdout: %s\nstderr: %s\nexit: %d" % (stdout,
                                                                       stderr,
                                                                       es); 220, string:"stdout: %s\nstderr: %s\nexit: %d"; 221, tuple; 222, identifier:stdout; 223, identifier:stderr; 224, identifier:es
code: def parallel_deblur(inputs, params,
                      pos_ref_db_fp, neg_ref_dp_fp, jobs_to_start=1):
      """Dispatch execution over a pool of processors
  
      This code was adopted from the American Gut project:
      https://github.com/biocore/American-Gut/blob/master/americangut/parallel.py
  
      Parameters
      ----------
      inputs : iterable of str
          File paths to input per-sample sequence files
      params : list of str
          list of CLI parameters supplied to the deblur workflow
          (argv - first 2 are 'deblur','workflow' and are ignored)
      pos_ref_db_fp : list of str
          the indexed positive (16s) sortmerna database
          (created in the main thread)
      neg_ref_db_fp : list of str
          the indexed negative (artifacts) sortmerna database
          (created in the main thread)
      jobs_to_start : int, optional
          The number of processors on the local system to use
  
      Returns
      -------
      all_result_paths : list
          list of expected output files
      """
      logger = logging.getLogger(__name__)
      logger.info('parallel deblur started for %d inputs' % len(inputs))
  
      # remove the irrelevant parameters
      remove_param_list = ['-O', '--jobs-to-start', '--seqs-fp',
                           '--pos-ref-db-fp', '--neg-ref-db-fp']
      skipnext = False
      newparams = []
      for carg in params[2:]:
          if skipnext:
              skipnext = False
              continue
          if carg in remove_param_list:
              skipnext = True
              continue
          newparams.append(carg)
  
      # add the ref_db_fp (since it may be not present in the
      # original command parameters)
      if pos_ref_db_fp:
          new_pos_ref_db_fp = ','.join(pos_ref_db_fp)
          newparams.append('--pos-ref-db-fp')
          newparams.append(new_pos_ref_db_fp)
      if neg_ref_dp_fp:
          new_neg_ref_db_fp = ','.join(neg_ref_dp_fp)
          newparams.append('--neg-ref-db-fp')
          newparams.append(new_neg_ref_db_fp)
  
      logger.debug('ready for functor %s' % newparams)
      functor = partial(run_functor, deblur_system_call, newparams)
      logger.debug('ready for pool %d jobs' % jobs_to_start)
      pool = mp.Pool(processes=jobs_to_start)
      logger.debug('almost running...')
      for stdout, stderr, es in pool.map(functor, inputs):
          if es != 0:
              raise RuntimeError("stdout: %s\nstderr: %s\nexit: %d" % (stdout,
                                                                       stderr,
                                                                       es))