<s> import sqlite3 <EOL> import sys <EOL> import os <EOL> import gzip <EOL> from collections import OrderedDict <EOL> GWASCAT_COLS = ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:date>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> GWASCAT_TYPES = ( "<STR_LIT:text>" , "<STR_LIT>" , "<STR_LIT:text>" , "<STR_LIT:text>" , "<STR_LIT:text>" , "<STR_LIT:text>" , <EOL> "<STR_LIT:text>" , "<STR_LIT:text>" , "<STR_LIT:text>" , "<STR_LIT:text>" , "<STR_LIT:text>" , <EOL> "<STR_LIT:text>" , "<STR_LIT>" , "<STR_LIT:text>" , "<STR_LIT:text>" , "<STR_LIT:text>" , <EOL> "<STR_LIT:text>" , "<STR_LIT:text>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT:text>" , "<STR_LIT:text>" , "<STR_LIT>" , "<STR_LIT:text>" , <EOL> "<STR_LIT:text>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:text>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:text>" , "<STR_LIT:text>" , "<STR_LIT:text>" ) <EOL> TBL_COLS = list ( GWASCAT_COLS ) <EOL> TBL_TYPES = list ( GWASCAT_TYPES ) <EOL> pos = TBL_COLS . index ( "<STR_LIT>" ) <EOL> TBL_COLS . insert ( pos + <NUM_LIT:1> , "<STR_LIT>" ) <EOL> TBL_COLS [ pos ] = "<STR_LIT>" <EOL> TBL_TYPES . insert ( pos + <NUM_LIT:1> , "<STR_LIT:text>" ) <EOL> TBL = OrderedDict ( zip ( TBL_COLS , TBL_TYPES ) ) <EOL> db_filename = sys . argv [ <NUM_LIT:1> ] <EOL> db_tablename = sys . argv [ <NUM_LIT:2> ] <EOL> gwas_filename = sys . argv [ <NUM_LIT:3> ] <EOL> reader = gzip . open if gwas_filename . endswith ( '<STR_LIT>' ) else open <EOL> TBL = OrderedDict ( zip ( TBL_COLS , TBL_TYPES ) ) <EOL> TBL_TYPESTR = "<STR_LIT>" . join ( [ "<STR_LIT>" % ( k , v ) for k , v in TBL . items ( ) ] ) <EOL> TYPES = { "<STR_LIT:text>" : str , "<STR_LIT>" : int , "<STR_LIT>" : float } <EOL> def cleanfield ( x ) : <EOL> """<STR_LIT>""" <EOL> if x is None : <EOL> return x <EOL> x = x . strip ( ) <EOL> if x . upper ( ) in ( "<STR_LIT:->" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) or len ( x ) == <NUM_LIT:0> : <EOL> return None <EOL> return x <EOL> conn = sqlite3 . connect ( db_filename ) <EOL> conn . text_factory = str <EOL> c = conn . cursor ( ) <EOL> TBL_SCHEMA = "<STR_LIT>" % ( db_tablename , TBL_TYPESTR ) <EOL> c . execute ( TBL_SCHEMA ) <EOL> header = True <EOL> for line in reader ( gwas_filename ) : <EOL> if header : <EOL> header = False <EOL> continue <EOL> fields = line . strip ( ) . split ( "<STR_LIT:\t>" ) <EOL> coerced_fields = list ( ) <EOL> for field , col , ftype in zip ( fields , GWASCAT_COLS , GWASCAT_TYPES ) : <EOL> if ftype is None : <EOL> continue <EOL> if col == "<STR_LIT>" : <EOL> snp , _ , allele = field . partition ( "<STR_LIT:->" ) <EOL> allele = None if allele . upper ( ) not in "<STR_LIT>" else allele . upper ( ) . strip ( ) <EOL> coerced_fields . extend ( ( snp . strip ( ) , allele ) ) <EOL> continue <EOL> if col in ( "<STR_LIT>" , "<STR_LIT:date>" ) : <EOL> month , day , year = field . split ( '<STR_LIT:/>' ) <EOL> coerced_fields . append ( "<STR_LIT>" % ( year , month , day ) ) <EOL> continue <EOL> try : <EOL> typefun = TYPES . get ( ftype , None ) <EOL> if typefun is not None : <EOL> cfield = cleanfield ( field ) <EOL> if cfield is None : <EOL> coerced_fields . append ( None ) <EOL> else : <EOL> coerced_fields . append ( typefun ( cfield ) ) <EOL> else : <EOL> coerced_fields . append ( cleanfield ( field ) ) <EOL> except : <EOL> coerced_fields . append ( None ) <EOL> placeholders = [ "<STR_LIT:?>" ] * len ( TBL_COLS ) <EOL> query = "<STR_LIT>" % ( db_tablename , "<STR_LIT:U+002CU+0020>" . join ( TBL_COLS ) , "<STR_LIT:U+002CU+0020>" . join ( placeholders ) ) <EOL> c . execute ( query , coerced_fields ) <EOL> conn . commit ( ) <EOL> c = conn . cursor ( ) </s>
