Added --nproc to MakeDb

NEWS.rst

@@ -4,7 +4,11 @@ Release Notes
 Version 1.3.4dev:  Unreleased
 -------------------------------------------------------------------------------
 
-+ Added the flag ``--partial`` to MakeDb igblast-aa to pass incomplete alignment
+MakeDb:
+
++ ``MakeDb.py`` is now parallelized across input files. The ``--nproc`` argument 
+   can be used to restrict the resources used.
++ Added the flag ``--partial`` to ``MakeDb igblast-aa`` to pass incomplete alignment
   results. As ``igblastp`` (as at igblast 1.22.0) only uses the V germline 
   database, all sequences will be missing the ``junction`` and ``j_call`` fields 
   and be considered incomplete. Specifying ``--partial`` will allow these 

bin/MakeDb.py

@@ -16,6 +16,8 @@
 from textwrap import dedent
 from time import time
 from Bio import SeqIO
+from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures import as_completed
 
 # Presto and changeo imports
 from presto.Annotation import parseAnnotation
@@ -423,7 +425,7 @@ def _partial(rec):
 
 def parseIMGT(aligner_file, seq_file=None, repo=None, cellranger_file=None, validate='strict', asis_id=True,
               extended=False, format=default_format, out_file=None, out_args=default_out_args,
-              imgt_id_len=default_imgt_id_len):
+              imgt_id_len=default_imgt_id_len, nproc=None):
     """
     Main for IMGT aligned sample sequences.
 
@@ -438,6 +440,7 @@ def parseIMGT(aligner_file, seq_file=None, repo=None, cellranger_file=None, vali
       out_file (str): output file name. Automatically generated from the input file if None.
       out_args (dict): common output argument dictionary from parseCommonArgs.
       imgt_id_len (int): maximum character length of sequence identifiers reported by IMGT/HighV-QUEST.
+      nproc (int): number of processes to use for parallel processing.
 
     Returns:
       dict: names of the 'pass' and 'fail' output files.
@@ -448,6 +451,7 @@ def parseIMGT(aligner_file, seq_file=None, repo=None, cellranger_file=None, vali
     log['COMMAND'] = 'imgt'
     log['ALIGNER_FILE'] = aligner_file
     log['SEQ_FILE'] = os.path.basename(seq_file) if seq_file else ''
+    log['NPROC'] = nproc
     log['ASIS_ID'] = asis_id
     log['VALIDATE'] = validate
     log['EXTENDED'] = extended
@@ -518,7 +522,7 @@ def parseIMGT(aligner_file, seq_file=None, repo=None, cellranger_file=None, vali
 
 def parseIgBLAST(aligner_file, seq_file, repo, amino_acid=False, cellranger_file=None, validate='strict',
                  asis_id=True, asis_calls=False, extended=False, regions='default', infer_junction=False,
-                 format='changeo', out_file=None, out_args=default_out_args):
+                 format='changeo', out_file=None, out_args=default_out_args, nproc=None):
     """
     Main for IgBLAST aligned sample sequences.
 
@@ -536,6 +540,7 @@ def parseIgBLAST(aligner_file, seq_file, repo, amino_acid=False, cellranger_file
       format (str): output format. one of 'changeo' or 'airr'.
       out_file (str): output file name. Automatically generated from the input file if None.
       out_args (dict): common output argument dictionary from parseCommonArgs.
+      nproc (int): number of processes to use for parallel processing.
 
     Returns:
       dict: names of the 'pass' and 'fail' output files.
@@ -546,6 +551,7 @@ def parseIgBLAST(aligner_file, seq_file, repo, amino_acid=False, cellranger_file
     log['COMMAND'] = 'igblast-aa' if amino_acid else 'igblast'
     log['ALIGNER_FILE'] = os.path.basename(aligner_file)
     log['SEQ_FILE'] = os.path.basename(seq_file)
+    log['NPROC'] = nproc
     log['ASIS_ID'] = asis_id
     log['ASIS_CALLS'] = asis_calls
     log['VALIDATE'] = validate
@@ -611,7 +617,7 @@ def parseIgBLAST(aligner_file, seq_file, repo, amino_acid=False, cellranger_file
 
 
 def parseIHMM(aligner_file, seq_file, repo, cellranger_file=None, validate='strict', asis_id=True,
-              extended=False, format=default_format, out_file=None, out_args=default_out_args):
+              extended=False, format=default_format, out_file=None, out_args=default_out_args, nproc=None):
     """
     Main for iHMMuneAlign aligned sample sequences.
 
@@ -625,6 +631,7 @@ def parseIHMM(aligner_file, seq_file, repo, cellranger_file=None, validate='stri
       format (str): output format. One of 'changeo' or 'airr'.
       out_file (str): output file name. Automatically generated from the input file if None.
       out_args (dict): common output argument dictionary from parseCommonArgs.
+      nproc (int): number of processes to use for parallel processing.
 
     Returns:
       dict: names of the 'pass' and 'fail' output files.
@@ -635,6 +642,7 @@ def parseIHMM(aligner_file, seq_file, repo, cellranger_file=None, validate='stri
     log['COMMAND'] = 'ihmm'
     log['ALIGNER_FILE'] = os.path.basename(aligner_file)
     log['SEQ_FILE'] = os.path.basename(seq_file)
+    log['NPROC'] = nproc
     log['ASIS_ID'] = asis_id
     log['VALIDATE'] = validate
     log['EXTENDED'] = extended
@@ -691,7 +699,7 @@ def parseIHMM(aligner_file, seq_file, repo, cellranger_file=None, validate='stri
 
 
 def numberAIRR(aligner_file, repo=None, format=default_format,
-               out_file=None, out_args=default_out_args):
+               out_file=None, out_args=default_out_args, nproc=None):
     """
     Inserts IMGT numbering into V fields
 
@@ -701,6 +709,7 @@ def numberAIRR(aligner_file, repo=None, format=default_format,
       format (str): output format.
       out_file (str): output file name. Automatically generated from the input file if None.
       out_args (dict): common output argument dictionary from parseCommonArgs.
+      nproc (int): number of processes to use for parallel processing.
 
     Returns:
       str: output file name.
@@ -709,6 +718,7 @@ def numberAIRR(aligner_file, repo=None, format=default_format,
     log['START'] = 'MakeDb'
     log['COMMAND'] = 'number'
     log['ALIGNER_FILE'] = os.path.basename(aligner_file)
+    log['NPROC'] = nproc
     printLog(log)
 
     # Define format operators
@@ -859,7 +869,7 @@ def getArgParser():
     subparsers.required = True
 
     # Parent parser
-    parser_parent = getCommonArgParser(db_in=False)
+    parser_parent = getCommonArgParser(db_in=False, multiproc=True)
 
     # igblastn output parser
     parser_igblast = subparsers.add_parser('igblast', parents=[parser_parent],
@@ -1094,13 +1104,18 @@ def getArgParser():
     if 'seq_files' in args_dict and not args_dict['seq_files']:
         args_dict['asis_id'] = True
 
+    # Extract nproc before deleting, use parser default if not specified
+    nproc_default = parser.get_default('nproc')
+    nproc = args_dict.get('nproc', nproc_default)
+
     # Delete
     if 'aligner_files' in args_dict: del args_dict['aligner_files']
     if 'seq_files' in args_dict: del args_dict['seq_files']
     if 'cellranger_files' in args_dict: del args_dict['cellranger_files']
     if 'out_files' in args_dict: del args_dict['out_files']
     if 'command' in args_dict: del args_dict['command']
     if 'func' in args_dict: del args_dict['func']
+    if 'nproc' in args_dict: del args_dict['nproc']
 
 
     ds=[]
@@ -1109,6 +1124,7 @@ def getArgParser():
     for i, f in enumerate(args.__dict__['aligner_files']):
         tmp_dict=args_dict.copy()
         tmp_dict['aligner_file'] = f
+        tmp_dict['nproc'] = nproc
         tmp_dict['out_file'] = args.__dict__['out_files'][i] \
                                 if args.__dict__['out_files'] else None
         if 'seq_files' in args.__dict__:
@@ -1120,10 +1136,7 @@ def getArgParser():
         ds.append(tmp_dict)
 
     # now do the loop in parallel
-    from concurrent.futures import ProcessPoolExecutor
-    from concurrent.futures import as_completed
-
-    with ProcessPoolExecutor() as executor:
+    with ProcessPoolExecutor(max_workers=nproc) as executor:
         futures = [executor.submit(args.func, **d) for d in ds]
         for future in as_completed(futures):
             result = future.result()

changeo/Commandline.py

@@ -119,7 +119,7 @@ def getCommonArgParser(db_in=True, db_out=True, out_file=True, failed=True, log=
     if multiproc:
         group.add_argument('--nproc', action='store', dest='nproc', type=int, default=mp.cpu_count(),
                            help='''The number of simultaneous computational processes to execute
-                                (CPU cores to utilized).''')
+                                (CPU cores to utilize).''')
 
     return parser