<s> '''<STR_LIT>''' <EOL> import re <EOL> import collections <EOL> import itertools <EOL> from CGAT import IOTools as IOTools <EOL> class CuffCompareValues : <EOL> def __init__ ( self , vals ) : <EOL> assert len ( vals ) == <NUM_LIT:4> <EOL> try : <EOL> self . sn = float ( vals [ <NUM_LIT:0> ] ) <EOL> except ValueError : <EOL> self . sn = None <EOL> try : <EOL> self . sp = float ( vals [ <NUM_LIT:1> ] ) <EOL> except ValueError : <EOL> self . sp = None <EOL> try : <EOL> self . fsn = float ( vals [ <NUM_LIT:2> ] ) <EOL> except ValueError : <EOL> self . fsn = None <EOL> try : <EOL> self . fsp = float ( vals [ <NUM_LIT:3> ] ) <EOL> except ValueError : <EOL> self . fsp = None <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT:\t>" . join ( map ( IOTools . val2str , <EOL> ( self . sn , <EOL> self . sp , <EOL> self . fsn , <EOL> self . fsp ) ) ) <EOL> @ classmethod <EOL> def getHeaders ( cls ) : <EOL> return ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> class CuffCompareResult : <EOL> def __init__ ( self ) : <EOL> ( self . baselevel , <EOL> self . exonlevel , <EOL> self . intronlevel , <EOL> self . intronchainlevel , <EOL> self . transcriptlevel , <EOL> self . locuslevel , <EOL> self . missedexons_counts , <EOL> self . missedexons_total , <EOL> self . novelexons_counts , <EOL> self . novelexons_total , <EOL> self . missedintrons_counts , <EOL> self . missedintrons_total , <EOL> self . novelintrons_counts , <EOL> self . novelintrons_total , <EOL> self . missedloci_counts , <EOL> self . missedloci_total , <EOL> self . novelloci_counts , <EOL> self . novelloci_total , <EOL> self . matchingintronchains_count , <EOL> self . matchingloci_count , <EOL> self . query , <EOL> self . query_loci , <EOL> self . query_multi_exon , <EOL> self . reference , <EOL> self . reference_loci , <EOL> self . reference_multi_exon , <EOL> self . loci_multi_exon , <EOL> self . loci_transcripts , <EOL> ) = [ None ] * <NUM_LIT> <EOL> def fromLines ( self , lines ) : <EOL> '''<STR_LIT>''' <EOL> self . is_empty = True <EOL> for line in lines : <EOL> if line . startswith ( "<STR_LIT:#>" ) : <EOL> d = line [ <NUM_LIT:1> : - <NUM_LIT:1> ] . strip ( ) <EOL> if d . startswith ( "<STR_LIT>" ) : <EOL> self . query , self . query_loci , self . query_multi_exon = map ( int , re . match ( <EOL> "<STR_LIT>" , d ) . groups ( ) ) <EOL> elif d . startswith ( "<STR_LIT>" ) : <EOL> self . reference , self . reference_loci , self . reference_multi_exon = map ( int , re . match ( <EOL> "<STR_LIT>" , d ) . groups ( ) ) <EOL> elif d . startswith ( "<STR_LIT:(>" ) : <EOL> self . loci_multi_exon , self . loci_transcripts = re . match ( <EOL> "<STR_LIT>" , d ) . groups ( ) <EOL> self . loci_multi_exon = int ( self . loci_multi_exon ) <EOL> self . loci_transcripts = float ( self . loci_transcripts ) <EOL> continue <EOL> line = line [ : - <NUM_LIT:1> ] . strip ( ) <EOL> if not line : <EOL> continue <EOL> try : <EOL> tag , data = line . split ( "<STR_LIT::>" ) <EOL> except ValueError : <EOL> raise ValueError ( "<STR_LIT>" % line ) <EOL> tag = re . sub ( "<STR_LIT>" , "<STR_LIT>" , tag ) . lower ( ) <EOL> if tag . startswith ( "<STR_LIT>" ) or tag . startswith ( "<STR_LIT>" ) : <EOL> counts , total = map ( <EOL> int , re . match ( "<STR_LIT>" , data ) . groups ( ) ) <EOL> setattr ( self , "<STR_LIT>" % tag , counts ) <EOL> setattr ( self , "<STR_LIT>" % tag , total ) <EOL> elif tag . startswith ( "<STR_LIT>" ) : <EOL> counts = int ( data ) <EOL> setattr ( self , "<STR_LIT>" % tag , counts ) <EOL> elif tag . startswith ( "<STR_LIT>" ) : <EOL> break <EOL> else : <EOL> values = data . strip ( ) . split ( ) <EOL> setattr ( self , tag , CuffCompareValues ( values ) ) <EOL> self . is_empty = False <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT:\t>" . join ( map ( str , ( <EOL> self . baselevel , <EOL> self . exonlevel , <EOL> self . intronlevel , <EOL> self . intronchainlevel , <EOL> self . transcriptlevel , <EOL> self . locuslevel , <EOL> self . missedexons_counts , <EOL> self . missedexons_total , <EOL> self . novelexons_counts , <EOL> self . novelexons_total , <EOL> self . missedintrons_counts , <EOL> self . missedintrons_total , <EOL> self . novelintrons_counts , <EOL> self . novelintrons_total , <EOL> self . missedloci_counts , <EOL> self . missedloci_total , <EOL> self . novelloci_counts , <EOL> self . novelloci_total , <EOL> self . matchingintronchains_count , <EOL> self . matchingloci_count , <EOL> self . query , <EOL> self . query_loci , <EOL> self . query_multi_exon , <EOL> self . reference , <EOL> self . reference_loci , <EOL> self . reference_multi_exon , <EOL> self . loci_multi_exon , <EOL> self . loci_transcripts <EOL> ) ) ) <EOL> @ classmethod <EOL> def getHeaders ( cls ) : <EOL> m = ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> a = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> b = ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> c = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> return [ "<STR_LIT>" % ( x , y ) for x , y in itertools . product ( m , CuffCompareValues . getHeaders ( ) ) ] + [ "<STR_LIT>" % ( x , y , z ) for x , y , z in itertools . product ( a , b , c ) ] + [ "<STR_LIT>" , "<STR_LIT>" ] + [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] <EOL> def parseTranscriptComparison ( infile ) : <EOL> '''<STR_LIT>''' <EOL> firstline = infile . readline ( ) <EOL> result = collections . defaultdict ( dict ) <EOL> tracks = [ ] <EOL> def __blocker ( infile ) : <EOL> blocks = { } <EOL> contig , block = None , [ ] <EOL> y = False <EOL> for line in infile : <EOL> if line . startswith ( "<STR_LIT>" ) : <EOL> if block : <EOL> blocks [ contig ] = block <EOL> if y : <EOL> yield dataset , blocks <EOL> blocks = { } <EOL> y = False <EOL> contig = re . match ( <EOL> "<STR_LIT>" , line ) . groups ( ) [ <NUM_LIT:0> ] <EOL> block = [ ] <EOL> continue <EOL> elif line . startswith ( "<STR_LIT>" ) : <EOL> if block : <EOL> blocks [ contig ] = block <EOL> if y : <EOL> yield dataset , blocks <EOL> blocks = { } <EOL> y = False <EOL> dataset = re . match ( <EOL> "<STR_LIT>" , line ) . groups ( ) [ <NUM_LIT:0> ] <EOL> contig = "<STR_LIT:all>" <EOL> block = [ ] <EOL> y = True <EOL> continue <EOL> block . append ( line ) <EOL> blocks [ contig ] = block <EOL> yield dataset , blocks <EOL> for track , blocks in __blocker ( infile ) : <EOL> print track <EOL> tracks . append ( track ) <EOL> for contig , block in blocks . iteritems ( ) : <EOL> r = CuffCompareResult ( ) <EOL> r . fromLines ( block ) <EOL> result [ track ] [ contig ] = r <EOL> return tracks , result <EOL> Locus = collections . namedtuple ( <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> Tracking = collections . namedtuple ( <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> TranscriptInfo = collections . namedtuple ( <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> def iterate_tracking ( infile ) : <EOL> '''<STR_LIT>''' <EOL> for line in infile : <EOL> data = line [ : - <NUM_LIT:1> ] . split ( "<STR_LIT:\t>" ) <EOL> try : <EOL> transfrag_id , locus_id , transcript_id , code = [ <EOL> x . strip ( ) for x in data [ : <NUM_LIT:4> ] ] <EOL> except ValueError : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % str ( data ) ) <EOL> if transcript_id == "<STR_LIT:->" : <EOL> ref_gene_id , ref_transcript_id = "<STR_LIT>" , "<STR_LIT>" <EOL> else : <EOL> ref_gene_id , ref_transcript_id = transcript_id . split ( "<STR_LIT:|>" ) <EOL> transcripts = [ ] <EOL> for c in data [ <NUM_LIT:4> : ] : <EOL> if c == "<STR_LIT:->" : <EOL> transcripts . append ( None ) <EOL> else : <EOL> for cc in c . split ( "<STR_LIT:U+002C>" ) : <EOL> try : <EOL> ( gene_id , transcript_id , fmi , fpkm , conf_lo , <EOL> conf_hi , cov , length ) = cc . split ( "<STR_LIT:|>" ) <EOL> except ValueError : <EOL> raise ValueError ( "<STR_LIT>" % cc ) <EOL> ( fpkm , conf_lo , conf_hi , cov ) = map ( <EOL> float , ( fpkm , conf_lo , conf_hi , cov ) ) <EOL> if length == "<STR_LIT:->" : <EOL> length = <NUM_LIT:0> <EOL> ( fmi , length ) = map ( int , ( fmi , length ) ) <EOL> transcripts . append ( TranscriptInfo . _make ( <EOL> ( gene_id , transcript_id , fmi , fpkm , conf_lo , conf_hi , cov , length ) ) ) <EOL> yield Tracking . _make ( ( transfrag_id , locus_id , ref_gene_id , ref_transcript_id , code , transcripts ) ) <EOL> def iterate_locus ( infile ) : <EOL> '''<STR_LIT>''' <EOL> for line in infile : <EOL> data = line [ : - <NUM_LIT:1> ] . split ( "<STR_LIT:\t>" ) <EOL> locus_id , pos = data [ : <NUM_LIT:2> ] <EOL> contig , strand , start , end = re . match ( <EOL> "<STR_LIT>" , pos ) . groups ( ) <EOL> start , end = map ( int , ( start , end ) ) <EOL> if strand not in "<STR_LIT>" : <EOL> strand = "<STR_LIT>" <EOL> transcripts = [ ] <EOL> for c in data [ <NUM_LIT:2> : ] : <EOL> if c == "<STR_LIT:->" : <EOL> transcripts . append ( ( ) ) <EOL> else : <EOL> transcripts . append ( c . split ( "<STR_LIT:U+002C>" ) ) <EOL> yield Locus . _make ( ( locus_id , contig , strand , start , end , transcripts [ <NUM_LIT:1> ] , transcripts [ <NUM_LIT:1> : ] ) ) </s>
