<s> import sys <EOL> from collections import defaultdict <EOL> from compiler import compile <EOL> from inheritance import Family <EOL> import sqlalchemy as sql <EOL> import database <EOL> from gemini_constants import * <EOL> import GeminiQuery <EOL> from functools import wraps <EOL> def compile_decorator ( f ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( f ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> query_string = f ( * args , ** kwargs ) <EOL> if query_string == "<STR_LIT:False>" or query_string == { "<STR_LIT>" : "<STR_LIT:False>" } : <EOL> return None <EOL> if not isinstance ( query_string , dict ) : <EOL> return compile ( query_string , "<STR_LIT>" , "<STR_LIT>" ) <EOL> query_dict = query_string <EOL> for k , stmt in query_dict . iteritems ( ) : <EOL> query_dict [ k ] = compile ( stmt , "<STR_LIT>" , "<STR_LIT>" ) <EOL> return query_dict <EOL> return wrapper <EOL> def get_phred_query ( sample_id , gt_ll , genotype , prefix = "<STR_LIT>" , invert = False ) : <EOL> """<STR_LIT>""" <EOL> assert genotype in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> if not gt_ll : return "<STR_LIT>" <EOL> if hasattr ( sample_id , "<STR_LIT>" ) : <EOL> sample_id = sample_id . sample_id <EOL> sign = [ "<STR_LIT:<>" , "<STR_LIT:>>" ] [ int ( invert ) ] <EOL> s = "<STR_LIT>" . format ( sample_id = sample_id - <NUM_LIT:1> , genotype = genotype , <EOL> gt_ll = gt_ll , sign = sign ) <EOL> return prefix + s <EOL> class Subject ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , row ) : <EOL> self . _set_fields_from_row ( row ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT:\t>" . join ( map ( str , [ self . name , self . paternal_id , <EOL> self . maternal_id , self . phenotype ] ) ) <EOL> def set_father ( self ) : <EOL> self . father = True <EOL> def set_mother ( self ) : <EOL> self . mother = True <EOL> def _set_fields_from_row ( self , row ) : <EOL> self . __dict__ . update ( row ) <EOL> self . phenotype = int ( self . phenotype ) if self . _has_phenotype ( ) else None <EOL> self . _set_affected_status ( ) <EOL> def _has_phenotype ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) and self . phenotype is not None : <EOL> return True <EOL> def _set_affected_status ( self ) : <EOL> pheno = str ( self . phenotype ) <EOL> if pheno == "<STR_LIT:2>" : <EOL> self . affected = True <EOL> elif pheno == "<STR_LIT:1>" : <EOL> self . affected = False <EOL> else : <EOL> self . affected = None <EOL> def get_families ( db , selected_families = None ) : <EOL> """<STR_LIT>""" <EOL> conn , metadata = database . get_session_metadata ( db ) <EOL> families_dict = Family . from_cursor ( conn ) <EOL> if selected_families is not None : <EOL> for family in selected_families . split ( '<STR_LIT:U+002C>' ) : <EOL> if family not in families_dict : <EOL> sys . exit ( "<STR_LIT>" % family ) <EOL> families = [ ] <EOL> for fam in families_dict : <EOL> if selected_families is None or fam in selected_families : <EOL> families . append ( families_dict [ fam ] ) <EOL> return families <EOL> def get_family_dict ( args ) : <EOL> families = defaultdict ( list ) <EOL> subjects = get_subjects ( args ) <EOL> for subject in subjects . values ( ) : <EOL> families [ subject . family_id ] . append ( subject ) <EOL> return families <EOL> def get_subjects ( args , skip_filter = False ) : <EOL> """<STR_LIT>""" <EOL> gq = GeminiQuery . GeminiQuery ( args . db ) <EOL> query = "<STR_LIT>" <EOL> if not skip_filter : <EOL> if hasattr ( args , '<STR_LIT>' ) and args . sample_filter : <EOL> query += args . sample_filter <EOL> res = gq . metadata . tables [ "<STR_LIT>" ] . select ( ) . where ( sql . text ( query ) ) . execute ( ) <EOL> samples_dict = { } <EOL> for row in res : <EOL> subject = Subject ( row ) <EOL> samples_dict [ subject . name ] = subject <EOL> return samples_dict <EOL> def get_subjects_in_family ( args , family ) : <EOL> subjects = get_subjects ( args ) <EOL> family_names = [ f . name for f in family ] <EOL> subject_dict = { } <EOL> for subject in subjects : <EOL> if subject in family_names : <EOL> subject_dict [ subject ] = subjects [ subject ] <EOL> return subject_dict <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> import doctest <EOL> doctest . testmod ( ) </s>
