comparison galaxy-tools/biobank/tools/check_update_parents_data.py @ 0:ba6cf6ede027 draft default tip

Uploaded

0:ba6cf6ede027

import sys, csv, argparse, logging, os
from collections import Counter

from bl.vl.kb import KnowledgeBase as KB
import bl.vl.utils.ome_utils as vlu
from bl.vl.utils import get_logger, LOG_LEVELS


def make_parser():
    parser = argparse.ArgumentParser(description='check data that will be passed to the update_parents tool')
    parser.add_argument('--logfile', type=str, help='log file (default=stderr)')
    parser.add_argument('--loglevel', type=str, choices=LOG_LEVELS,
                        help='logging level (default=INFO)', default='INFO')
    parser.add_argument('-H', '--host', type=str, help='omero hostname')
    parser.add_argument('-U', '--user', type=str, help='omero user')
    parser.add_argument('-P', '--passwd', type=str, help='omero password')
    parser.add_argument('--in_file', type=str, required=True,
                        help='input file')
    parser.add_argument('--out_file', type=str, required=True,
                        help='output file')
    return parser


def check_row(row, individuals_map, kb, logger):
    logger.debug('Checking record %r' % row)
    try:
        ind = individuals_map[row['individual']]
        logger.info('%s is a valid Individual ID' % ind.id)
        if row['father'] != 'None':
            father = individuals_map[row['father']]
            logger.info('%s is a valid Individual ID' % father.id)
            check_gender(father, kb.Gender.MALE)
            logger.info('Gender check passed')
        else:
            logger.info('None value, no check required')
        if row['mother'] != 'None':
            mother = individuals_map[row['mother']]
            logger.info('%s is a valid Individual ID' % mother.id)
            check_gender(mother, kb.Gender.FEMALE)
            logger.info('Gender check passed')
        else:
            logger.info('None value, no check required')
        return True
    except KeyError, ke:
        logger.error('%s is not a valid Individual ID, rejecting row' % ke)
        return False
    except ValueError, ve:
        logger.error(ve)
        return False


def check_gender(individual, gender):
    if individual.gender.enum_label() != gender.enum_label():
        raise ValueError('Gender for individual %s is %s, expected %s, rejecting row' % (individual.id,
                                                                                         individual.gender.enum_label(),
                                                                                         gender.enum_label()))
    else:
        pass


def main(argv):
    parser = make_parser()
    args = parser.parse_args(argv)

    logger = get_logger('check_update_parents_data', level=args.loglevel,
                        filename=args.logfile)

    try:
        host = args.host or vlu.ome_host()
        user = args.user or vlu.ome_user()
        passwd = args.passwd or vlu.ome_passwd()
    except ValueError, ve:
        logger.critical(ve)
        sys.exit(ve)

    kb = KB(driver='omero')(host, user, passwd)

    logger.info('Preloading all individuals from the system')
    inds = kb.get_objects(kb.Individual)
    logger.info('%d individuals loaded' % len(inds))
    inds_lookup = {}
    for i in inds:
        inds_lookup[i.id] = i

    with open(args.in_file) as infile, open(args.out_file, 'w') as outfile:
        reader = csv.DictReader(infile, delimiter='\t')
        records = list(reader)
        logger.info('Check for duplicated in \'individual\' column')
        recs_by_ind = {}
        for rec in records:
            recs_by_ind.setdefault(rec['individual'], []).append(rec)
        ct = Counter()
        duplicated = []
        for k,v in recs_by_ind.iteritems():
            if len(v) > 1:
                duplicated.append(k)
        for dupl in duplicated:
            logger.info('Individual %s is a duplicated' % dupl)
            for r in recs_by_ind.pop(dupl):
                logger.info('Removing record %r' % r)
        good_records = sum(recs_by_ind.itervalues(), [])
        logger.info('Duplicated check completed')
        writer = csv.DictWriter(outfile, reader.fieldnames, delimiter='\t')
        writer.writeheader()
        logger.info('Checking records')
        for row in good_records:
            if check_row(row, inds_lookup, kb, logger):
                writer.writerow(row)
                logger.debug('Record %r written in output file' % row)
        logger.info('Records check completed')

if __name__ == '__main__':
    main(sys.argv[1:])