diff --git a/NEWS.rst b/NEWS.rst index ce4136f..53a9391 100644 --- a/NEWS.rst +++ b/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 diff --git a/bin/MakeDb.py b/bin/MakeDb.py index 4073cfc..ea7bfa4 100755 --- a/bin/MakeDb.py +++ b/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,6 +1104,10 @@ 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'] @@ -1101,6 +1115,7 @@ def getArgParser(): 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() diff --git a/changeo/Commandline.py b/changeo/Commandline.py index ce1c478..c1a7c67 100755 --- a/changeo/Commandline.py +++ b/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