<s> """<STR_LIT>""" <EOL> from bx . bbi . bigbed_file import BigBedFile <EOL> from bx . bbi . bigwig_file import BigWigFile <EOL> from bx . intervals . io import StrandFormatError <EOL> import numpy as np <EOL> import subprocess <EOL> import pysam <EOL> import pybedtools <EOL> import os <EOL> import sys <EOL> from textwrap import dedent <EOL> strand_lookup = { <NUM_LIT:16> : '<STR_LIT:->' , <NUM_LIT:0> : '<STR_LIT:+>' } <EOL> class BaseAdapter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fn ) : <EOL> self . fn = fn <EOL> self . fileobj = None <EOL> self . fileobj = self . make_fileobj ( ) <EOL> def __getitem__ ( self , key ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def make_fileobj ( self ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> class BamAdapter ( BaseAdapter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fn ) : <EOL> super ( BamAdapter , self ) . __init__ ( fn ) <EOL> def make_fileobj ( self ) : <EOL> return pysam . Samfile ( self . fn , '<STR_LIT:rb>' ) <EOL> def __getitem__ ( self , key ) : <EOL> iterator = self . fileobj . fetch ( <EOL> str ( key . chrom ) , <EOL> key . start , <EOL> key . stop ) <EOL> for r in iterator : <EOL> start = r . pos <EOL> curr_end = r . pos <EOL> for op , bp in r . cigar : <EOL> start = curr_end <EOL> curr_end += bp <EOL> if op == <NUM_LIT:0> : <EOL> interval = pybedtools . Interval ( <EOL> self . fileobj . references [ r . rname ] , <EOL> start , <EOL> curr_end , <EOL> strand = strand_lookup [ r . flag & <NUM_LIT> ] ) <EOL> interval . file_type = '<STR_LIT>' <EOL> yield interval <EOL> class BedAdapter ( BaseAdapter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fn ) : <EOL> super ( BedAdapter , self ) . __init__ ( fn ) <EOL> def make_fileobj ( self ) : <EOL> obj = pybedtools . BedTool ( self . fn ) <EOL> if not obj . _tabixed ( ) : <EOL> obj = obj . sort ( ) . tabix ( in_place = False , force = False , is_sorted = True ) <EOL> self . fn = obj . fn <EOL> return obj <EOL> def __getitem__ ( self , key ) : <EOL> bt = self . fileobj . tabix_intervals ( <EOL> '<STR_LIT>' % ( key . chrom , key . start , key . stop ) ) <EOL> for i in bt : <EOL> yield i <EOL> del bt <EOL> class BigBedAdapter ( BaseAdapter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fn ) : <EOL> super ( BigBedAdapter , self ) . __init__ ( fn ) <EOL> def make_fileobj ( self ) : <EOL> return BigBedFile ( open ( self . fn ) ) <EOL> def __getitem__ ( self , key ) : <EOL> chrom = key . chrom <EOL> start = key . start <EOL> stop = key . end <EOL> try : <EOL> bx_intervals = self . fileobj . get ( chrom , start , stop ) <EOL> except StrandFormatError : <EOL> raise NotImplementedError ( dedent ( <EOL> """<STR_LIT>""" . format ( self . fn ) ) ) <EOL> if bx_intervals is None : <EOL> raise StopIteration <EOL> for i in bx_intervals : <EOL> interval = pybedtools . create_interval_from_list ( i . fields ) <EOL> interval . file_type = '<STR_LIT>' <EOL> yield interval <EOL> class BigWigAdapter ( BaseAdapter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fn ) : <EOL> super ( BigWigAdapter , self ) . __init__ ( fn ) <EOL> def make_fileobj ( self ) : <EOL> return self . fn <EOL> def __getitem__ ( self , key ) : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" % self . __class__ . __name__ ) <EOL> def summarize ( self , interval , bins = None , method = '<STR_LIT>' , <EOL> function = '<STR_LIT>' ) : <EOL> orig = np . geterr ( ) [ '<STR_LIT>' ] <EOL> np . seterr ( invalid = '<STR_LIT:ignore>' ) <EOL> if ( bins is None ) or ( method == '<STR_LIT>' ) : <EOL> bw = BigWigFile ( open ( self . fn ) ) <EOL> s = bw . get_as_array ( <EOL> interval . chrom , <EOL> interval . start , <EOL> interval . stop , ) <EOL> if s is None : <EOL> s = np . zeros ( ( interval . stop - interval . start , ) ) <EOL> else : <EOL> s [ np . isnan ( s ) ] = <NUM_LIT:0> <EOL> elif method == '<STR_LIT>' : <EOL> if function in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> return self . ucsc_summarize ( interval , bins , function = function ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> bw = BigWigFile ( open ( self . fn ) ) <EOL> s = bw . summarize ( <EOL> interval . chrom , <EOL> interval . start , <EOL> interval . stop , bins ) <EOL> if s is None : <EOL> s = np . zeros ( ( bins , ) ) <EOL> else : <EOL> if function == '<STR_LIT>' : <EOL> s = s . sum_data <EOL> if function == '<STR_LIT>' : <EOL> s = s . sum_data / s . valid_count <EOL> s [ np . isnan ( s ) ] = <NUM_LIT:0> <EOL> if function == '<STR_LIT>' : <EOL> s = s . min_val <EOL> s [ np . isinf ( s ) ] = <NUM_LIT:0> <EOL> if function == '<STR_LIT>' : <EOL> s = s . max_val <EOL> s [ np . isinf ( s ) ] = <NUM_LIT:0> <EOL> if function == '<STR_LIT>' : <EOL> s = ( s . sum_squares / s . valid_count ) <EOL> s [ np . isnan ( s ) ] = <NUM_LIT:0> <EOL> np . seterr ( divide = orig ) <EOL> return s <EOL> def ucsc_summarize ( self , interval , bins = None , function = '<STR_LIT>' ) : <EOL> if bins is None : <EOL> bins = len ( interval ) <EOL> y = np . zeros ( bins ) <EOL> cmds = [ <EOL> '<STR_LIT>' , <EOL> self . fn , <EOL> interval . chrom , <EOL> str ( interval . start ) , <EOL> str ( interval . stop ) , <EOL> str ( bins ) , <EOL> '<STR_LIT>' % function ] <EOL> p = subprocess . Popen ( <EOL> cmds , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , <EOL> ) <EOL> def gen ( ) : <EOL> try : <EOL> for line in p . stdout : <EOL> yield line <EOL> finally : <EOL> if p . poll ( ) is None : <EOL> return <EOL> else : <EOL> p . wait ( ) <EOL> err = p . stderr . read ( ) . strip ( ) <EOL> if p . returncode not in ( <NUM_LIT:0> , None ) : <EOL> if err . startswith ( '<STR_LIT>' ) : <EOL> return <EOL> raise ValueError ( <EOL> "<STR_LIT>" % <EOL> ( '<STR_LIT:U+0020>' . join ( cmds ) , p . stderr . read ( ) ) ) <EOL> if len ( err ) != <NUM_LIT:0> : <EOL> sys . stderr . write ( err ) <EOL> for line in gen ( ) : <EOL> for i , x in enumerate ( line . split ( '<STR_LIT:\t>' ) ) : <EOL> try : <EOL> y [ i ] = float ( x ) <EOL> except ValueError : <EOL> pass <EOL> return np . array ( y ) </s>
