<s> import unittest <EOL> from testlib import testutil , PygrTestProgram <EOL> import ConfigParser <EOL> import os <EOL> import string <EOL> import sys <EOL> from pygr . mapping import Collection <EOL> import pygr . Data <EOL> try : <EOL> import hashlib <EOL> except ImportError : <EOL> import md5 as hashlib <EOL> config = ConfigParser . ConfigParser ( { '<STR_LIT>' : '<STR_LIT:.>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> config . read ( [ os . path . join ( os . path . expanduser ( '<STR_LIT>' ) , '<STR_LIT>' ) , <EOL> os . path . join ( os . path . expanduser ( '<STR_LIT>' ) , '<STR_LIT>' ) , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> msaDir = config . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> seqDir = config . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> smallSampleKey = config . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> testInputDB = config . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> testInputDir = config . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> testOutputBaseDir = config . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if smallSampleKey : <EOL> smallSamplePostfix = '<STR_LIT:_>' + smallSampleKey <EOL> else : <EOL> smallSamplePostfix = '<STR_LIT>' <EOL> docStringDict = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> msaSpeciesList = [ '<STR_LIT>' , '<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>' , '<STR_LIT>' ] <EOL> class PygrBuildNLMSAMegabase ( unittest . TestCase ) : <EOL> def setUp ( self , testDir = None ) : <EOL> '''<STR_LIT>''' <EOL> import random <EOL> tmpList = [ c for c in '<STR_LIT>' ] <EOL> random . shuffle ( tmpList ) <EOL> testDir = os . path . join ( testOutputBaseDir , '<STR_LIT>' + '<STR_LIT>' . join ( tmpList ) ) <EOL> if testDir is None : <EOL> testDir = '<STR_LIT>' + '<STR_LIT>' . join ( tmpList ) <EOL> try : <EOL> os . mkdir ( testDir ) <EOL> testDir = os . path . realpath ( testDir ) <EOL> except : <EOL> raise IOError <EOL> self . path = testDir <EOL> try : <EOL> tmpFileName = os . path . join ( testDir , '<STR_LIT>' ) <EOL> open ( tmpFileName , '<STR_LIT:w>' ) . write ( '<STR_LIT:A>' * <NUM_LIT> * <NUM_LIT> ) <EOL> except : <EOL> raise IOError <EOL> pygr . Data . update ( self . path ) <EOL> from pygr import seqdb <EOL> for orgstr in msaSpeciesList : <EOL> genome = seqdb . BlastDB ( os . path . join ( seqDir , orgstr ) ) <EOL> genome . __doc__ = docStringDict [ orgstr ] <EOL> pygr . Data . addResource ( '<STR_LIT>' + orgstr , genome ) <EOL> pygr . Data . save ( ) <EOL> def copyFile ( self , filename ) : <EOL> newname = os . path . join ( self . path , os . path . basename ( filename ) ) <EOL> open ( newname , '<STR_LIT:w>' ) . write ( open ( filename , '<STR_LIT:r>' ) . read ( ) ) <EOL> return newname <EOL> def tearDown ( self ) : <EOL> '<STR_LIT>' <EOL> for dirpath , subdirs , files in os . walk ( self . path , topdown = False ) : <EOL> for filename in files : <EOL> os . remove ( os . path . join ( dirpath , filename ) ) <EOL> os . rmdir ( dirpath ) <EOL> pygr . Data . update ( None ) <EOL> class Build_Test ( PygrBuildNLMSAMegabase ) : <EOL> def test_seqdb ( self ) : <EOL> '<STR_LIT>' <EOL> l = pygr . Data . dir ( '<STR_LIT>' ) <EOL> preList = [ '<STR_LIT>' + orgstr for orgstr in msaSpeciesList ] <EOL> assert l == preList <EOL> def test_collectionannot ( self ) : <EOL> '<STR_LIT>' <EOL> from pygr import seqdb , cnestedlist , sqlgraph <EOL> dm2 = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> exon_slices = Collection ( <EOL> filename = os . path . join ( self . path , '<STR_LIT>' ) , <EOL> intKeys = True , mode = '<STR_LIT>' , writeback = False ) <EOL> exon_db = seqdb . AnnotationDB ( exon_slices , dm2 , <EOL> sliceAttrDict = dict ( id = <NUM_LIT:0> , exon_id = <NUM_LIT:1> , <EOL> orientation = <NUM_LIT:2> , <EOL> gene_id = <NUM_LIT:3> , start = <NUM_LIT:4> , <EOL> stop = <NUM_LIT:5> ) ) <EOL> msa = cnestedlist . NLMSA ( os . path . join ( self . path , <EOL> '<STR_LIT>' ) , '<STR_LIT:w>' , <EOL> pairwiseMode = True , bidirectional = False ) <EOL> for lines in open ( os . path . join ( testInputDir , <EOL> '<STR_LIT>' <EOL> % smallSamplePostfix ) , <EOL> '<STR_LIT:r>' ) . xreadlines ( ) : <EOL> row = [ x for x in lines . split ( '<STR_LIT:\t>' ) ] <EOL> row [ <NUM_LIT:1> ] = int ( row [ <NUM_LIT:1> ] ) <EOL> exon_slices [ row [ <NUM_LIT:1> ] ] = row <EOL> exon = exon_db [ row [ <NUM_LIT:1> ] ] <EOL> msa . addAnnotation ( exon ) <EOL> exon_db . clear_cache ( ) <EOL> exon_slices . close ( ) <EOL> msa . build ( ) <EOL> exon_db . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , exon_db ) <EOL> msa . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , msa ) <EOL> exon_schema = pygr . Data . ManyToManyRelation ( dm2 , exon_db , <EOL> bindAttrs = ( '<STR_LIT>' , ) ) <EOL> exon_schema . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addSchema ( '<STR_LIT>' , exon_schema ) <EOL> splice_slices = Collection ( <EOL> filename = os . path . join ( self . path , '<STR_LIT>' ) , <EOL> intKeys = True , mode = '<STR_LIT>' , writeback = False ) <EOL> splice_db = seqdb . AnnotationDB ( splice_slices , dm2 , <EOL> sliceAttrDict = dict ( id = <NUM_LIT:0> , splice_id = <NUM_LIT:1> , <EOL> orientation = <NUM_LIT:2> , <EOL> gene_id = <NUM_LIT:3> , start = <NUM_LIT:4> , <EOL> stop = <NUM_LIT:5> ) ) <EOL> msa = cnestedlist . NLMSA ( os . path . join ( self . path , <EOL> '<STR_LIT>' ) , '<STR_LIT:w>' , <EOL> pairwiseMode = True , bidirectional = False ) <EOL> for lines in open ( os . path . join ( testInputDir , <EOL> '<STR_LIT>' <EOL> % smallSamplePostfix ) , <EOL> '<STR_LIT:r>' ) . xreadlines ( ) : <EOL> row = [ x for x in lines . split ( '<STR_LIT:\t>' ) ] <EOL> row [ <NUM_LIT:1> ] = int ( row [ <NUM_LIT:1> ] ) <EOL> splice_slices [ row [ <NUM_LIT:1> ] ] = row <EOL> splice = splice_db [ row [ <NUM_LIT:1> ] ] <EOL> msa . addAnnotation ( splice ) <EOL> splice_db . clear_cache ( ) <EOL> splice_slices . close ( ) <EOL> msa . build ( ) <EOL> splice_db . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , splice_db ) <EOL> msa . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , msa ) <EOL> splice_schema = pygr . Data . ManyToManyRelation ( dm2 , splice_db , <EOL> bindAttrs = ( '<STR_LIT>' , ) ) <EOL> splice_schema . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addSchema ( '<STR_LIT>' , splice_schema ) <EOL> ucsc_slices = Collection ( <EOL> filename = os . path . join ( self . path , '<STR_LIT>' ) , <EOL> intKeys = True , mode = '<STR_LIT>' , writeback = False ) <EOL> ucsc_db = seqdb . AnnotationDB ( ucsc_slices , dm2 , <EOL> sliceAttrDict = dict ( id = <NUM_LIT:0> , ucsc_id = <NUM_LIT:1> , <EOL> orientation = <NUM_LIT:2> , <EOL> gene_id = <NUM_LIT:3> , start = <NUM_LIT:4> , <EOL> stop = <NUM_LIT:5> ) ) <EOL> msa = cnestedlist . NLMSA ( os . path . join ( self . path , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT:w>' , pairwiseMode = True , bidirectional = False ) <EOL> for lines in open ( os . path . join ( testInputDir , <EOL> '<STR_LIT>' <EOL> % smallSamplePostfix ) , <EOL> '<STR_LIT:r>' ) . xreadlines ( ) : <EOL> row = [ x for x in lines . split ( '<STR_LIT:\t>' ) ] <EOL> row [ <NUM_LIT:1> ] = int ( row [ <NUM_LIT:1> ] ) <EOL> ucsc_slices [ row [ <NUM_LIT:1> ] ] = row <EOL> ucsc = ucsc_db [ row [ <NUM_LIT:1> ] ] <EOL> msa . addAnnotation ( ucsc ) <EOL> ucsc_db . clear_cache ( ) <EOL> ucsc_slices . close ( ) <EOL> msa . build ( ) <EOL> ucsc_db . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , <EOL> ucsc_db ) <EOL> msa . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , <EOL> msa ) <EOL> ucsc_schema = pygr . Data . ManyToManyRelation ( dm2 , ucsc_db , <EOL> bindAttrs = ( '<STR_LIT>' , ) ) <EOL> ucsc_schema . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addSchema ( '<STR_LIT>' , <EOL> ucsc_schema ) <EOL> pygr . Data . save ( ) <EOL> pygr . Data . clear_cache ( ) <EOL> dm2 = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> exonmsa = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> splicemsa = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> conservedmsa = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> exons = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> splices = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> mostconserved = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> msa = cnestedlist . NLMSA ( os . path . join ( msaDir , '<STR_LIT>' ) , '<STR_LIT:r>' , <EOL> trypath = [ seqDir ] ) <EOL> exonAnnotFileName = os . path . join ( testInputDir , <EOL> '<STR_LIT>' <EOL> % smallSamplePostfix ) <EOL> intronAnnotFileName = os . path . join ( testInputDir , <EOL> '<STR_LIT>' <EOL> % smallSamplePostfix ) <EOL> newexonAnnotFileName = os . path . join ( self . path , '<STR_LIT>' ) <EOL> newintronAnnotFileName = os . path . join ( self . path , '<STR_LIT>' ) <EOL> tmpexonAnnotFileName = self . copyFile ( exonAnnotFileName ) <EOL> tmpintronAnnotFileName = self . copyFile ( intronAnnotFileName ) <EOL> if smallSampleKey : <EOL> chrList = [ smallSampleKey ] <EOL> else : <EOL> chrList = dm2 . seqLenDict . keys ( ) <EOL> chrList . sort ( ) <EOL> outfile = open ( newexonAnnotFileName , '<STR_LIT:w>' ) <EOL> for chrid in chrList : <EOL> slice = dm2 [ chrid ] <EOL> try : <EOL> ex1 = exonmsa [ slice ] <EOL> except KeyError : <EOL> continue <EOL> else : <EOL> exlist1 = [ ( ix . exon_id , ix ) for ix in ex1 . keys ( ) ] <EOL> exlist1 . sort ( ) <EOL> for ixx , exon in exlist1 : <EOL> saveList = [ ] <EOL> tmp = exon . sequence <EOL> tmpexon = exons [ exon . exon_id ] <EOL> tmpslice = tmpexon . sequence <EOL> wlist1 = '<STR_LIT>' , chrid , tmpexon . exon_id , tmpexon . gene_id , tmpslice . start , tmpslice . stop <EOL> try : <EOL> out1 = conservedmsa [ tmp ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> elementlist = [ ( ix . ucsc_id , ix ) for ix in out1 . keys ( ) ] <EOL> elementlist . sort ( ) <EOL> for iyy , element in elementlist : <EOL> if element . stop - element . start < <NUM_LIT:100> : <EOL> continue <EOL> score = int ( string . split ( element . gene_id , '<STR_LIT:=>' ) [ <NUM_LIT:1> ] ) <EOL> if score < <NUM_LIT:100> : <EOL> continue <EOL> tmp2 = element . sequence <EOL> tmpelement = mostconserved [ element . ucsc_id ] <EOL> tmpslice2 = tmpelement . sequence <EOL> wlist2 = wlist1 + ( tmpelement . ucsc_id , <EOL> tmpelement . gene_id , <EOL> tmpslice2 . start , tmpslice2 . stop ) <EOL> slicestart , sliceend = max ( tmp . start , tmp2 . start ) , min ( tmp . stop , tmp2 . stop ) <EOL> tmp1 = msa . seqDict [ '<STR_LIT>' + chrid ] [ slicestart : <EOL> sliceend ] <EOL> edges = msa [ tmp1 ] . edges ( ) <EOL> for src , dest , e in edges : <EOL> if src . stop - src . start < <NUM_LIT:100> : <EOL> continue <EOL> palign , pident = e . pAligned ( ) , e . pIdentity ( ) <EOL> if palign < <NUM_LIT> or pident < <NUM_LIT> : <EOL> continue <EOL> palign , pident = '<STR_LIT>' % palign , '<STR_LIT>' % pident <EOL> wlist3 = wlist2 + ( ( ~ msa . seqDict ) [ src ] , <EOL> str ( src ) , src . start , <EOL> src . stop , <EOL> ( ~ msa . seqDict ) [ dest ] , <EOL> str ( dest ) , dest . start , <EOL> dest . stop , palign , pident ) <EOL> saveList . append ( '<STR_LIT:\t>' . join ( map ( str , wlist3 ) ) <EOL> + '<STR_LIT:\n>' ) <EOL> saveList . sort ( ) <EOL> for saveline in saveList : <EOL> outfile . write ( saveline ) <EOL> outfile . close ( ) <EOL> md5old = hashlib . md5 ( ) <EOL> md5old . update ( open ( tmpexonAnnotFileName , '<STR_LIT:r>' ) . read ( ) ) <EOL> md5new = hashlib . md5 ( ) <EOL> md5new . update ( open ( newexonAnnotFileName , '<STR_LIT:r>' ) . read ( ) ) <EOL> assert md5old . digest ( ) == md5new . digest ( ) <EOL> outfile = open ( newintronAnnotFileName , '<STR_LIT:w>' ) <EOL> for chrid in chrList : <EOL> slice = dm2 [ chrid ] <EOL> try : <EOL> sp1 = splicemsa [ slice ] <EOL> except : <EOL> continue <EOL> else : <EOL> splist1 = [ ( ix . splice_id , ix ) for ix in sp1 . keys ( ) ] <EOL> splist1 . sort ( ) <EOL> for ixx , splice in splist1 : <EOL> saveList = [ ] <EOL> tmp = splice . sequence <EOL> tmpsplice = splices [ splice . splice_id ] <EOL> tmpslice = tmpsplice . sequence <EOL> wlist1 = '<STR_LIT>' , chrid , tmpsplice . splice_id , tmpsplice . gene_id , tmpslice . start , tmpslice . stop <EOL> try : <EOL> out1 = conservedmsa [ tmp ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> elementlist = [ ( ix . ucsc_id , ix ) for ix in out1 . keys ( ) ] <EOL> elementlist . sort ( ) <EOL> for iyy , element in elementlist : <EOL> if element . stop - element . start < <NUM_LIT:100> : <EOL> continue <EOL> score = int ( string . split ( element . gene_id , '<STR_LIT:=>' ) [ <NUM_LIT:1> ] ) <EOL> if score < <NUM_LIT:100> : <EOL> continue <EOL> tmp2 = element . sequence <EOL> tmpelement = mostconserved [ element . ucsc_id ] <EOL> tmpslice2 = tmpelement . sequence <EOL> wlist2 = wlist1 + ( tmpelement . ucsc_id , <EOL> tmpelement . gene_id , <EOL> tmpslice2 . start , tmpslice2 . stop ) <EOL> slicestart , sliceend = max ( tmp . start , tmp2 . start ) , min ( tmp . stop , tmp2 . stop ) <EOL> tmp1 = msa . seqDict [ '<STR_LIT>' + chrid ] [ slicestart : <EOL> sliceend ] <EOL> edges = msa [ tmp1 ] . edges ( ) <EOL> for src , dest , e in edges : <EOL> if src . stop - src . start < <NUM_LIT:100> : <EOL> continue <EOL> palign , pident = e . pAligned ( ) , e . pIdentity ( ) <EOL> if palign < <NUM_LIT> or pident < <NUM_LIT> : <EOL> continue <EOL> palign , pident = '<STR_LIT>' % palign , '<STR_LIT>' % pident <EOL> wlist3 = wlist2 + ( ( ~ msa . seqDict ) [ src ] , <EOL> str ( src ) , src . start , <EOL> src . stop , <EOL> ( ~ msa . seqDict ) [ dest ] , <EOL> str ( dest ) , dest . start , <EOL> dest . stop , palign , pident ) <EOL> saveList . append ( '<STR_LIT:\t>' . join ( map ( str , wlist3 ) ) <EOL> + '<STR_LIT:\n>' ) <EOL> saveList . sort ( ) <EOL> for saveline in saveList : <EOL> outfile . write ( saveline ) <EOL> outfile . close ( ) <EOL> md5old = hashlib . md5 ( ) <EOL> md5old . update ( open ( tmpintronAnnotFileName , '<STR_LIT:r>' ) . read ( ) ) <EOL> md5new = hashlib . md5 ( ) <EOL> md5new . update ( open ( newintronAnnotFileName , '<STR_LIT:r>' ) . read ( ) ) <EOL> assert md5old . digest ( ) == md5new . digest ( ) <EOL> def test_mysqlannot ( self ) : <EOL> '<STR_LIT>' <EOL> from pygr import seqdb , cnestedlist , sqlgraph <EOL> dm2 = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> exon_slices = sqlgraph . SQLTableClustered ( <EOL> '<STR_LIT>' % ( testInputDB , <EOL> smallSamplePostfix ) , <EOL> clusterKey = '<STR_LIT>' , maxCache = <NUM_LIT:0> ) <EOL> exon_db = seqdb . AnnotationDB ( exon_slices , dm2 , <EOL> sliceAttrDict = dict ( id = '<STR_LIT>' , <EOL> gene_id = '<STR_LIT:name>' , <EOL> exon_id = '<STR_LIT>' ) ) <EOL> msa = cnestedlist . NLMSA ( os . path . join ( self . path , <EOL> '<STR_LIT>' ) , '<STR_LIT:w>' , <EOL> pairwiseMode = True , bidirectional = False ) <EOL> for id in exon_db : <EOL> msa . addAnnotation ( exon_db [ id ] ) <EOL> exon_db . clear_cache ( ) <EOL> exon_slices . clear_cache ( ) <EOL> msa . build ( ) <EOL> exon_db . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , exon_db ) <EOL> msa . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , msa ) <EOL> exon_schema = pygr . Data . ManyToManyRelation ( dm2 , exon_db , <EOL> bindAttrs = ( '<STR_LIT>' , ) ) <EOL> exon_schema . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addSchema ( '<STR_LIT>' , exon_schema ) <EOL> splice_slices = sqlgraph . SQLTableClustered ( <EOL> '<STR_LIT>' % ( testInputDB , <EOL> smallSamplePostfix ) , <EOL> clusterKey = '<STR_LIT>' , maxCache = <NUM_LIT:0> ) <EOL> splice_db = seqdb . AnnotationDB ( splice_slices , dm2 , <EOL> sliceAttrDict = dict ( id = '<STR_LIT>' , <EOL> gene_id = '<STR_LIT:name>' , <EOL> splice_id = '<STR_LIT>' ) ) <EOL> msa = cnestedlist . NLMSA ( os . path . join ( self . path , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT:w>' , pairwiseMode = True , bidirectional = False ) <EOL> for id in splice_db : <EOL> msa . addAnnotation ( splice_db [ id ] ) <EOL> splice_db . clear_cache ( ) <EOL> splice_slices . clear_cache ( ) <EOL> msa . build ( ) <EOL> splice_db . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , splice_db ) <EOL> msa . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , msa ) <EOL> splice_schema = pygr . Data . ManyToManyRelation ( dm2 , splice_db , <EOL> bindAttrs = ( '<STR_LIT>' , ) ) <EOL> splice_schema . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addSchema ( '<STR_LIT>' , <EOL> splice_schema ) <EOL> ucsc_slices = sqlgraph . SQLTableClustered ( <EOL> '<STR_LIT>' % ( testInputDB , <EOL> smallSamplePostfix ) , <EOL> clusterKey = '<STR_LIT>' , maxCache = <NUM_LIT:0> ) <EOL> ucsc_db = seqdb . AnnotationDB ( ucsc_slices , dm2 , <EOL> sliceAttrDict = dict ( id = '<STR_LIT>' , <EOL> gene_id = '<STR_LIT:name>' , <EOL> ucsc_id = '<STR_LIT>' ) ) <EOL> msa = cnestedlist . NLMSA ( os . path . join ( self . path , <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT:w>' , pairwiseMode = True , bidirectional = False ) <EOL> for id in ucsc_db : <EOL> msa . addAnnotation ( ucsc_db [ id ] ) <EOL> ucsc_db . clear_cache ( ) <EOL> ucsc_slices . clear_cache ( ) <EOL> msa . build ( ) <EOL> ucsc_db . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( '<STR_LIT>' , <EOL> ucsc_db ) <EOL> msa . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addResource ( <EOL> '<STR_LIT>' , msa ) <EOL> ucsc_schema = pygr . Data . ManyToManyRelation ( dm2 , ucsc_db , <EOL> bindAttrs = ( '<STR_LIT>' , ) ) <EOL> ucsc_schema . __doc__ = '<STR_LIT>' <EOL> pygr . Data . addSchema ( '<STR_LIT>' , <EOL> ucsc_schema ) <EOL> pygr . Data . save ( ) <EOL> pygr . Data . clear_cache ( ) <EOL> dm2 = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> exonmsa = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> splicemsa = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> conservedmsa = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> exons = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> splices = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> mostconserved = pygr . Data . getResource ( '<STR_LIT>' ) <EOL> msa = cnestedlist . NLMSA ( os . path . join ( msaDir , '<STR_LIT>' ) , '<STR_LIT:r>' , <EOL> trypath = [ seqDir ] ) <EOL> exonAnnotFileName = os . path . join ( testInputDir , <EOL> '<STR_LIT>' <EOL> % smallSamplePostfix ) <EOL> intronAnnotFileName = os . path . join ( testInputDir , <EOL> '<STR_LIT>' <EOL> % smallSamplePostfix ) <EOL> newexonAnnotFileName = os . path . join ( self . path , '<STR_LIT>' ) <EOL> newintronAnnotFileName = os . path . join ( self . path , '<STR_LIT>' ) <EOL> tmpexonAnnotFileName = self . copyFile ( exonAnnotFileName ) <EOL> tmpintronAnnotFileName = self . copyFile ( intronAnnotFileName ) <EOL> if smallSampleKey : <EOL> chrList = [ smallSampleKey ] <EOL> else : <EOL> chrList = dm2 . seqLenDict . keys ( ) <EOL> chrList . sort ( ) <EOL> outfile = open ( newexonAnnotFileName , '<STR_LIT:w>' ) <EOL> for chrid in chrList : <EOL> slice = dm2 [ chrid ] <EOL> try : <EOL> ex1 = exonmsa [ slice ] <EOL> except KeyError : <EOL> continue <EOL> else : <EOL> exlist1 = [ ( ix . exon_id , ix ) for ix in ex1 . keys ( ) ] <EOL> exlist1 . sort ( ) <EOL> for ixx , exon in exlist1 : <EOL> saveList = [ ] <EOL> tmp = exon . sequence <EOL> tmpexon = exons [ exon . exon_id ] <EOL> tmpslice = tmpexon . sequence <EOL> wlist1 = '<STR_LIT>' , chrid , tmpexon . exon_id , tmpexon . gene_id , tmpslice . start , tmpslice . stop <EOL> try : <EOL> out1 = conservedmsa [ tmp ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> elementlist = [ ( ix . ucsc_id , ix ) for ix in out1 . keys ( ) ] <EOL> elementlist . sort ( ) <EOL> for iyy , element in elementlist : <EOL> if element . stop - element . start < <NUM_LIT:100> : <EOL> continue <EOL> score = int ( string . split ( element . gene_id , '<STR_LIT:=>' ) [ <NUM_LIT:1> ] ) <EOL> if score < <NUM_LIT:100> : <EOL> continue <EOL> tmp2 = element . sequence <EOL> tmpelement = mostconserved [ element . ucsc_id ] <EOL> tmpslice2 = tmpelement . sequence <EOL> wlist2 = wlist1 + ( tmpelement . ucsc_id , <EOL> tmpelement . gene_id , <EOL> tmpslice2 . start , tmpslice2 . stop ) <EOL> slicestart , sliceend = max ( tmp . start , tmp2 . start ) , min ( tmp . stop , tmp2 . stop ) <EOL> tmp1 = msa . seqDict [ '<STR_LIT>' + chrid ] [ slicestart : <EOL> sliceend ] <EOL> edges = msa [ tmp1 ] . edges ( ) <EOL> for src , dest , e in edges : <EOL> if src . stop - src . start < <NUM_LIT:100> : <EOL> continue <EOL> palign , pident = e . pAligned ( ) , e . pIdentity ( ) <EOL> if palign < <NUM_LIT> or pident < <NUM_LIT> : <EOL> continue <EOL> palign , pident = '<STR_LIT>' % palign , '<STR_LIT>' % pident <EOL> wlist3 = wlist2 + ( ( ~ msa . seqDict ) [ src ] , <EOL> str ( src ) , src . start , <EOL> src . stop , <EOL> ( ~ msa . seqDict ) [ dest ] , <EOL> str ( dest ) , dest . start , <EOL> dest . stop , palign , pident ) <EOL> saveList . append ( '<STR_LIT:\t>' . join ( map ( str , wlist3 ) ) <EOL> + '<STR_LIT:\n>' ) <EOL> saveList . sort ( ) <EOL> for saveline in saveList : <EOL> outfile . write ( saveline ) <EOL> outfile . close ( ) <EOL> md5old = hashlib . md5 ( ) <EOL> md5old . update ( open ( tmpexonAnnotFileName , '<STR_LIT:r>' ) . read ( ) ) <EOL> md5new = hashlib . md5 ( ) <EOL> md5new . update ( open ( newexonAnnotFileName , '<STR_LIT:r>' ) . read ( ) ) <EOL> assert md5old . digest ( ) == md5new . digest ( ) <EOL> outfile = open ( newintronAnnotFileName , '<STR_LIT:w>' ) <EOL> for chrid in chrList : <EOL> slice = dm2 [ chrid ] <EOL> try : <EOL> sp1 = splicemsa [ slice ] <EOL> except : <EOL> continue <EOL> else : <EOL> splist1 = [ ( ix . splice_id , ix ) for ix in sp1 . keys ( ) ] <EOL> splist1 . sort ( ) <EOL> for ixx , splice in splist1 : <EOL> saveList = [ ] <EOL> tmp = splice . sequence <EOL> tmpsplice = splices [ splice . splice_id ] <EOL> tmpslice = tmpsplice . sequence <EOL> wlist1 = '<STR_LIT>' , chrid , tmpsplice . splice_id , tmpsplice . gene_id , tmpslice . start , tmpslice . stop <EOL> try : <EOL> out1 = conservedmsa [ tmp ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> elementlist = [ ( ix . ucsc_id , ix ) for ix in out1 . keys ( ) ] <EOL> elementlist . sort ( ) <EOL> for iyy , element in elementlist : <EOL> if element . stop - element . start < <NUM_LIT:100> : <EOL> continue <EOL> score = int ( string . split ( element . gene_id , '<STR_LIT:=>' ) [ <NUM_LIT:1> ] ) <EOL> if score < <NUM_LIT:100> : <EOL> continue <EOL> tmp2 = element . sequence <EOL> tmpelement = mostconserved [ element . ucsc_id ] <EOL> tmpslice2 = tmpelement . sequence <EOL> wlist2 = wlist1 + ( tmpelement . ucsc_id , <EOL> tmpelement . gene_id , <EOL> tmpslice2 . start , tmpslice2 . stop ) <EOL> slicestart , sliceend = max ( tmp . start , tmp2 . start ) , min ( tmp . stop , tmp2 . stop ) <EOL> tmp1 = msa . seqDict [ '<STR_LIT>' + chrid ] [ slicestart : <EOL> sliceend ] <EOL> edges = msa [ tmp1 ] . edges ( ) <EOL> for src , dest , e in edges : <EOL> if src . stop - src . start < <NUM_LIT:100> : <EOL> continue <EOL> palign , pident = e . pAligned ( ) , e . pIdentity ( ) <EOL> if palign < <NUM_LIT> or pident < <NUM_LIT> : <EOL> continue <EOL> palign , pident = '<STR_LIT>' % palign , '<STR_LIT>' % pident <EOL> wlist3 = wlist2 + ( ( ~ msa . seqDict ) [ src ] , <EOL> str ( src ) , src . start , <EOL> src . stop , <EOL> ( ~ msa . seqDict ) [ dest ] , <EOL> str ( dest ) , dest . start , <EOL> dest . stop , palign , pident ) <EOL> saveList . append ( '<STR_LIT:\t>' . join ( map ( str , wlist3 ) ) <EOL> + '<STR_LIT:\n>' ) <EOL> saveList . sort ( ) <EOL> for saveline in saveList : <EOL> outfile . write ( saveline ) <EOL> outfile . close ( ) <EOL> md5old = hashlib . md5 ( ) <EOL> md5old . update ( open ( tmpintronAnnotFileName , '<STR_LIT:r>' ) . read ( ) ) <EOL> md5new = hashlib . md5 ( ) <EOL> md5new . update ( open ( newintronAnnotFileName , '<STR_LIT:r>' ) . read ( ) ) <EOL> assert md5old . digest ( ) == md5new . digest ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> PygrTestProgram ( verbosity = <NUM_LIT:2> ) </s>
