parse.py 文件源码

def main(args):
    """
    Parse IgBLAST output
    """
    n = 0
    if args.vdatabase:
        with SequenceReader(args.vdatabase) as fr:
            v_database = {record.name: record.sequence.upper() for record in fr}
    else:
        v_database = None

    detected_cdr3s = 0
    writer = TableWriter(sys.stdout)
    with SequenceReader(args.fasta) as sequences, xopen(args.igblast) as igblast:
        parser = ExtendedIgBlastParser(sequences, igblast, v_database)
        for record in parser:
            n += 1
            d = record.asdict()
            if d['CDR3_aa']:
                detected_cdr3s += 1
            if args.rename is not None:
                d['name'] = "{}seq{}".format(args.rename, n)
            try:
                writer.write(d)
            except IOError as e:
                if e.errno == errno.EPIPE:
                    sys.exit(1)
                raise
    logger.info('%d IgBLAST assignments parsed and written', n)
    logger.info('CDR3s detected in %.1f%% of all sequences', detected_cdr3s / n * 100)

    if args.stats:
        stats = {'total': n, 'detected_cdr3s': detected_cdr3s}
        with open(args.stats, 'w') as f:
            json.dump(stats, f)
            print(file=f)