signature:
def get_fixed_capabils ( payload ) :
    """strip ( 2 byte ) wlan _ mgt . fixed . capabilities
 : payload : ctypes . structure
 2 byte
 : return : dict
 None in error"""

implementation:
if len ( payload ) != 2 :
    return None
capabils = { }
fix_cap = struct . unpack ( 'H' , payload ) [ 0 ]
cap_bits = format ( fix_cap , '016b' ) [ : : - 1 ]
capabils [ 'ess' ] = int ( cap_bits [ 0 ] )
# Extended Service Set
capabils [ 'ibss' ] = int ( cap_bits [ 1 ] )
# Independent Basic Service Set
capabils [ 'priv' ] = int ( cap_bits [ 4 ] )
# Privacy
capabils [ 'short_preamble' ] = int ( cap_bits [ 5 ] )
# Short Preamble
capabils [ 'pbcc' ] = int ( cap_bits [ 6 ] )
# Packet Binary Convolutional Code
capabils [ 'chan_agility' ] = int ( cap_bits [ 7 ] )
# Channel Agility
capabils [ 'spec_man' ] = int ( cap_bits [ 8 ] )
# Spectrum Management
capabils [ 'short_slot' ] = int ( cap_bits [ 10 ] )
# Short Slot Time
capabils [ 'apsd' ] = int ( cap_bits [ 11 ] )
# Automatic Power Save Delivery
capabils [ 'radio_meas' ] = int ( cap_bits [ 12 ] )
# Radio Measurement
capabils [ 'dss_ofdm' ] = int ( cap_bits [ 13 ] )
# Direct Spread Spectrum
capabils [ 'del_back' ] = int ( cap_bits [ 14 ] )
# Delayed Block Acknowledgement
capabils [ 'imm_back' ] = int ( cap_bits [ 15 ] )
# Immediate Block Acknowledgement
return capabils


signature:
def parse_vendor_ie ( payload ) :
    """parse vendor specific information element
 oui - > organizationally unique identifier
 first 3 bytes of mac addresses
 see : https : / / www . wireshark . org / tools / oui - lookup . html
 strip wlan _ mgt . tag . oui ( 3 bytes ) ,
 wlan _ mgt . tag . vendor . oui . type ( 1 byte )
 wlan _ mgt . tag . vendor . data ( varies )
 : payload : ctypes . structure
 : return : dict
 { ' oui ' : 00-11-22 , ' oui _ type ' : 1 , ' oui _ data ' : ctypes . structure }"""

implementation:
output = { }
oui = struct . unpack ( 'BBB' , payload [ 0 : 3 ] )
oui = b'-' . join ( [ ( '%02x' % o ) . encode ( 'ascii' ) for o in oui ] )
oui_type = struct . unpack ( 'B' , payload [ 3 : 4 ] ) [ 0 ]
oui_data = payload [ 4 : ]
output [ 'oui' ] = oui . upper ( )
output [ 'oui_type' ] = oui_type
output [ 'oui_data' ] = oui_data
return output


signature:
def strip_fixed_params ( payload ) :
    """strip ( 12 byte ) wlan _ mgt . fixed . all
 : payload : ctypes . structure
 : return : int
 timestamp
 : return : int
 beacon interval
 : return : dict
 capabilities"""

implementation:
if len ( payload ) != 12 :
    return None , None , None
idx = 0
timestamp = Management . get_timestamp ( payload [ idx : idx + 8 ] )
idx += 8
interval = Management . get_interval ( payload [ idx : idx + 2 ] )
idx += 2
capabils = Management . get_fixed_capabils ( payload [ idx : idx + 2 ] )
return timestamp , interval , capabils


signature:
def is_valid_mac_oui ( mac_block ) :
    """checks whether mac block is in format of
 00-11-22 or 00:11:22.
 : return : int"""

implementation:
if len ( mac_block ) != 8 :
    return 0
if ':' in mac_block :
    if len ( mac_block . split ( ':' ) ) != 3 :
        return 0
elif '-' in mac_block :
    if len ( mac_block . split ( '-' ) ) != 3 :
        return 0
return 1


signature:
def set_fixed_capabils ( self , capabils ) :
    """set keys of capabils into fields of object
 : capabils : dict"""

implementation:
self . ess = capabils [ 'ess' ]
self . ibss = capabils [ 'ibss' ]
self . priv = capabils [ 'priv' ]
self . short_preamble = capabils [ 'short_preamble' ]
self . pbcc = capabils [ 'pbcc' ]
self . chan_agility = capabils [ 'chan_agility' ]
self . spec_man = capabils [ 'spec_man' ]
self . short_slot = capabils [ 'short_slot' ]
self . apsd = capabils [ 'apsd' ]
self . radio_meas = capabils [ 'radio_meas' ]
self . dss_ofdm = capabils [ 'dss_ofdm' ]
self . del_back = capabils [ 'del_back' ]
self . imm_back = capabils [ 'imm_back' ]


signature:
def get_vendor_ies ( self , mac_block = None , oui_type = None ) :
    """vendor information element querying
 : mac _ block : str
 first 3 bytes of mac addresses in format of
 00-11-22 or 00:11:22 or 001122
 : oui _ type : int
 vendors ie type
 : return : int
 is valid mac _ block format
 -1 is unknown
 : return : dict [ ]
 list of oui information elements
 -1 on error ( invalid v"""

implementation:
vendor_ies = [ ]
if mac_block is not None :
    if Management . is_valid_mac_oui ( mac_block ) :
        mac_block = mac_block . upper ( )
        if ':' in mac_block :
            mac_block . replace ( ':' , '-' )
    else :
        logging . warning ( "invalid oui macblock" )
        return None
for elem in self . tagged_params :
    tag_num = elem [ 'number' ]
    if MNGMT_TAGS [ tag_num ] == 'TAG_VENDOR_SPECIFIC_IE' :
        if mac_block is None :
            vendor_ies . append ( elem )
        elif elem [ 'payload' ] [ 'oui' ] == mac_block . encode ( 'ascii' ) :
            if oui_type is None :
                vendor_ies . append ( elem )
            elif elem [ 'payload' ] [ 'oui_type' ] == oui_type :
                vendor_ies . append ( elem )
return vendor_ies


signature:
def get_shark_field ( self , fields ) :
    """: fields : str [ ]"""

implementation:
out = super ( BACK , self ) . get_shark_field ( fields )
out . update ( { 'acked_seqs' : self . acked_seqs , 'bitmap_str' : self . bitmap_str } )
return out


signature:
def strip_cntrl ( payload ) :
    """strip ( 2 byte ) wlan . ba . control
 : payload : ctypes . structure
 : return : int
 multitid ( tid : traffic indicator )
 : return : int
 ackpolicy"""

implementation:
cntrl = struct . unpack ( 'H' , payload ) [ 0 ]
cntrl_bits = format ( cntrl , '016b' ) [ : : - 1 ]
ackpolicy = int ( cntrl_bits [ 0 ] )
multitid = int ( cntrl_bits [ 1 ] )
return ackpolicy , multitid


signature:
def strip_ssc ( payload ) :
    """strip ( 2 byte ) wlan _ mgt . fixed . ssc
 : payload : ctypes . structure
 : return : int
 ssc _ seq ( starting sequence control sequence )
 : return : int
 ssc _ frag ( starting sequence control fragment number )"""

implementation:
ssc = struct . unpack ( 'H' , payload ) [ 0 ]
ssc_seq = ssc >> 4
ssc_frag = ssc & 0x000f
return ssc_seq , ssc_frag


signature:
def strip_bitmap_str ( payload ) :
    """strip ( 8 byte ) wlan . ba . bm
 : payload : ctypes . structure
 : return : str
 bitmap"""

implementation:
bitmap = struct . unpack ( 'BBBBBBBB' , payload )
bitmap_str = ''
for elem in bitmap :
    bitmap_str += format ( elem , '08b' ) [ : : - 1 ]
return bitmap_str


signature:
def extract_acked_seqs ( bitmap , ssc_seq ) :
    """extracts acknowledged sequences from bitmap and
 starting sequence number .
 : bitmap : str
 : ssc _ seq : int
 : return : int [ ]
 acknowledged sequence numbers"""

implementation:
acked_seqs = [ ]
for idx , val in enumerate ( bitmap ) :
    if int ( val ) == 1 :
        seq = ( ssc_seq + idx ) % 4096
        acked_seqs . append ( seq )
return acked_seqs


signature:
def heartbeat ( ) :
    """Call Heartbeat URL"""

implementation:
print "We got a call heartbeat notification\n"
if request . method == 'POST' :
    print request . form
else :
    print request . args
return "OK"


signature:
def request ( self , path , method = None , data = { } ) :
    """sends a request and gets a response from the Plivo REST API
 path : the URL ( relative to the endpoint URL , after the / v1
 method : the HTTP method to use , defaults to POST
 data : for POST or PUT , a dict of data to send
 returns Plivo response in XML or raises an exception on error"""

implementation:
if not path :
    raise ValueError ( 'Invalid path parameter' )
if method and method not in [ 'GET' , 'POST' , 'DELETE' , 'PUT' ] :
    raise NotImplementedError ( 'HTTP %s method not implemented' % method )
if path [ 0 ] == '/' :
    uri = self . url + path
else :
    uri = self . url + '/' + path
if APPENGINE :
    return json . loads ( self . _appengine_fetch ( uri , data , method ) )
return json . loads ( self . _urllib2_fetch ( uri , data , method ) )


signature:
def reload_config ( self , call_params ) :
    """REST Reload Plivo Config helper"""

implementation:
path = '/' + self . api_version + '/ReloadConfig/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def reload_cache_config ( self , call_params ) :
    """REST Reload Plivo Cache Config helper"""

implementation:
path = '/' + self . api_version + '/ReloadCacheConfig/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def call ( self , call_params ) :
    """REST Call Helper"""

implementation:
path = '/' + self . api_version + '/Call/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def bulk_call ( self , call_params ) :
    """REST BulkCalls Helper"""

implementation:
path = '/' + self . api_version + '/BulkCall/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def group_call ( self , call_params ) :
    """REST GroupCalls Helper"""

implementation:
path = '/' + self . api_version + '/GroupCall/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def transfer_call ( self , call_params ) :
    """REST Transfer Live Call Helper"""

implementation:
path = '/' + self . api_version + '/TransferCall/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def hangup_all_calls ( self ) :
    """REST Hangup All Live Calls Helper"""

implementation:
path = '/' + self . api_version + '/HangupAllCalls/'
method = 'POST'
return self . request ( path , method )


signature:
def hangup_call ( self , call_params ) :
    """REST Hangup Live Call Helper"""

implementation:
path = '/' + self . api_version + '/HangupCall/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def schedule_hangup ( self , call_params ) :
    """REST Schedule Hangup Helper"""

implementation:
path = '/' + self . api_version + '/ScheduleHangup/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def cancel_scheduled_hangup ( self , call_params ) :
    """REST Cancel a Scheduled Hangup Helper"""

implementation:
path = '/' + self . api_version + '/CancelScheduledHangup/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def record_start ( self , call_params ) :
    """REST RecordStart helper"""

implementation:
path = '/' + self . api_version + '/RecordStart/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def record_stop ( self , call_params ) :
    """REST RecordStop"""

implementation:
path = '/' + self . api_version + '/RecordStop/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_mute ( self , call_params ) :
    """REST Conference Mute helper"""

implementation:
path = '/' + self . api_version + '/ConferenceMute/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def play ( self , call_params ) :
    """REST Play something on a Call Helper"""

implementation:
path = '/' + self . api_version + '/Play/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def play_stop ( self , call_params ) :
    """REST PlayStop on a Call Helper"""

implementation:
path = '/' + self . api_version + '/PlayStop/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def schedule_play ( self , call_params ) :
    """REST Schedule playing something on a call Helper"""

implementation:
path = '/' + self . api_version + '/SchedulePlay/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def cancel_scheduled_play ( self , call_params ) :
    """REST Cancel a Scheduled Play Helper"""

implementation:
path = '/' + self . api_version + '/CancelScheduledPlay/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def sound_touch ( self , call_params ) :
    """REST Add soundtouch audio effects to a Call"""

implementation:
path = '/' + self . api_version + '/SoundTouch/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def sound_touch_stop ( self , call_params ) :
    """REST Remove soundtouch audio effects on a Call"""

implementation:
path = '/' + self . api_version + '/SoundTouchStop/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def send_digits ( self , call_params ) :
    """REST Send digits to a Call"""

implementation:
path = '/' + self . api_version + '/SendDigits/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_unmute ( self , call_params ) :
    """REST Conference Unmute helper"""

implementation:
path = '/' + self . api_version + '/ConferenceUnmute/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_kick ( self , call_params ) :
    """REST Conference Kick helper"""

implementation:
path = '/' + self . api_version + '/ConferenceKick/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_hangup ( self , call_params ) :
    """REST Conference Hangup helper"""

implementation:
path = '/' + self . api_version + '/ConferenceHangup/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_deaf ( self , call_params ) :
    """REST Conference Deaf helper"""

implementation:
path = '/' + self . api_version + '/ConferenceDeaf/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_undeaf ( self , call_params ) :
    """REST Conference Undeaf helper"""

implementation:
path = '/' + self . api_version + '/ConferenceUndeaf/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_record_start ( self , call_params ) :
    """REST Conference RecordStart helper"""

implementation:
path = '/' + self . api_version + '/ConferenceRecordStart/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_record_stop ( self , call_params ) :
    """REST Conference RecordStop"""

implementation:
path = '/' + self . api_version + '/ConferenceRecordStop/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_play ( self , call_params ) :
    """REST Conference Play helper"""

implementation:
path = '/' + self . api_version + '/ConferencePlay/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_speak ( self , call_params ) :
    """REST Conference Speak helper"""

implementation:
path = '/' + self . api_version + '/ConferenceSpeak/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_list ( self , call_params ) :
    """REST Conference List Helper"""

implementation:
path = '/' + self . api_version + '/ConferenceList/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def conference_list_members ( self , call_params ) :
    """REST Conference List Members Helper"""

implementation:
path = '/' + self . api_version + '/ConferenceListMembers/'
method = 'POST'
return self . request ( path , method , call_params )


signature:
def _xml ( self , root ) :
    """Return an XML element representing this element"""

implementation:
element = root . createElement ( self . name )
# Add attributes
keys = self . attrs . keys ( )
keys . sort ( )
for a in keys :
    element . setAttribute ( a , self . attrs [ a ] )
if self . body :
    text = root . createTextNode ( self . body )
    element . appendChild ( text )
for c in self . elements :
    element . appendChild ( c . _xml ( root ) )
return element


signature:
def validateRequest ( self , uri , postVars , expectedSignature ) :
    """validate a request from plivo
 uri : the full URI that Plivo requested on your server
 postVars : post vars that Plivo sent with the request
 expectedSignature : signature in HTTP X - Plivo - Signature header
 returns true if the request passes validation , false if not"""

implementation:
# append the POST variables sorted by key to the uri
s = uri
for k , v in sorted ( postVars . items ( ) ) :
    s += k + v
# compute signature and compare signatures
return ( base64 . encodestring ( hmac . new ( self . auth_token , s , sha1 ) . digest ( ) ) . strip ( ) == expectedSignature )


signature:
def DFS_prefix ( self , root = None ) :
    """Depth - first search .
 . . seealso : :
 ` Wikipedia DFS descritpion < http : / / en . wikipedia . org / wiki / Depth - first _ search > ` _
 : param root : first to start the search
 : return : list of nodes"""

implementation:
if not root :
    root = self . _root
return self . _DFS_prefix ( root )


signature:
def BFS ( self , root = None ) :
    """Breadth - first search .
 . . seealso : :
 ` Wikipedia BFS descritpion < http : / / en . wikipedia . org / wiki / Breadth - first _ search > ` _
 : param root : first to start the search
 : return : list of nodes"""

implementation:
if not root :
    root = self . root ( )
queue = deque ( )
queue . append ( root )
nodes = [ ]
while len ( queue ) > 0 :
    x = queue . popleft ( )
    nodes . append ( x )
    for child in x . children ( ) :
        queue . append ( child )
return nodes


signature:
def add_node ( self , label ) :
    '''Return a node with label . Create node if label is new'''

implementation:
try :
    n = self . _nodes [ label ]
except KeyError :
    n = Node ( )
    n [ 'label' ] = label
    self . _nodes [ label ] = n
return n


signature:
def add_edge ( self , n1_label , n2_label , directed = False ) :
    """Get or create edges using get _ or _ create _ node"""

implementation:
n1 = self . add_node ( n1_label )
n2 = self . add_node ( n2_label )
e = Edge ( n1 , n2 , directed )
self . _edges . append ( e )
return e


signature:
def parse_dom ( dom ) :
    """Parse dom into a Graph .
 : param dom : dom as returned by minidom . parse or minidom . parseString
 : return : A Graph representation"""

implementation:
root = dom . getElementsByTagName ( "graphml" ) [ 0 ]
graph = root . getElementsByTagName ( "graph" ) [ 0 ]
name = graph . getAttribute ( 'id' )
g = Graph ( name )
# # Get attributes
# attributes = [ ]
# for attr in root . getElementsByTagName ( " key " ) :
# attributes . append ( attr )
# Get nodes
for node in graph . getElementsByTagName ( "node" ) :
    n = g . add_node ( id = node . getAttribute ( 'id' ) )
    for attr in node . getElementsByTagName ( "data" ) :
        if attr . firstChild :
            n [ attr . getAttribute ( "key" ) ] = attr . firstChild . data
        else :
            n [ attr . getAttribute ( "key" ) ] = ""
# Get edges
for edge in graph . getElementsByTagName ( "edge" ) :
    source = edge . getAttribute ( 'source' )
    dest = edge . getAttribute ( 'target' )
    # source / target attributes refer to IDs : http : / / graphml . graphdrawing . org / xmlns / 1.1 / graphml - structure . xsd
    e = g . add_edge_by_id ( source , dest )
    for attr in edge . getElementsByTagName ( "data" ) :
        if attr . firstChild :
            e [ attr . getAttribute ( "key" ) ] = attr . firstChild . data
        else :
            e [ attr . getAttribute ( "key" ) ] = ""
return g


signature:
def parse_string ( self , string ) :
    """Parse a string into a Graph .
 : param string : String that is to be passed into Grapg
 : return : Graph"""

implementation:
dom = minidom . parseString ( string )
return self . parse_dom ( dom )


signature:
def node ( self , node ) :
    """Return the other node"""

implementation:
if node == self . node1 :
    return self . node2
elif node == self . node2 :
    return self . node1
else :
    return None


signature:
def consume ( self , ** kwargs ) :
    """Return a generator that yields whenever a message is waiting in the
 queue . Will block otherwise . Example :
 > > > for msg in queue . consume ( timeout = 1 ) :
 . . . print msg
 my message
 another message
 : param kwargs : any arguments that : meth : ` ~ hotqueue . HotQueue . get ` can
 accept ( : attr : ` block ` will default to ` ` True ` ` if not given )"""

implementation:
kwargs . setdefault ( 'block' , True )
try :
    while True :
        msg = self . get ( ** kwargs )
        if msg is None :
            break
        yield msg
except KeyboardInterrupt :
    print ;
    return


signature:
def get ( self , block = False , timeout = None ) :
    """Return a message from the queue . Example :
 > > > queue . get ( )
 ' my message '
 > > > queue . get ( )
 ' another message '
 : param block : whether or not to wait until a msg is available in
 the queue before returning ; ` ` False ` ` by default
 : param timeout : when using : attr : ` block ` , if no msg is available
 for : attr : ` timeout ` in seconds , give up and return ` ` None ` `"""

implementation:
if block :
    if timeout is None :
        timeout = 0
    msg = self . __redis . blpop ( self . key , timeout = timeout )
    if msg is not None :
        msg = msg [ 1 ]
else :
    msg = self . __redis . lpop ( self . key )
if msg is not None and self . serializer is not None :
    msg = self . serializer . loads ( msg )
return msg


signature:
def put ( self , * msgs ) :
    """Put one or more messages onto the queue . Example :
 > > > queue . put ( " my message " )
 > > > queue . put ( " another message " )
 To put messages onto the queue in bulk , which can be significantly
 faster if you have a large number of messages :
 > > > queue . put ( " my message " , " another message " , " third message " )"""

implementation:
if self . serializer is not None :
    msgs = map ( self . serializer . dumps , msgs )
self . __redis . rpush ( self . key , * msgs )


signature:
def worker ( self , * args , ** kwargs ) :
    """Decorator for using a function as a queue worker . Example :
 > > > @ queue . worker ( timeout = 1)
 . . . def printer ( msg ) :
 . . . print msg
 > > > printer ( )
 my message
 another message
 You can also use it without passing any keyword arguments :
 > > > @ queue . worker
 . . . def printer ( msg ) :
 . . . print msg
 > > > printer ( )
 my message
 another message
 : param kwargs : any arguments that : meth : ` ~ hotqueue . HotQueue . get ` can
 accept ( : attr : ` block ` will default to ` ` True ` ` if not given )"""

implementation:
def decorator ( worker ) :
    @ wraps ( worker )
    def wrapper ( * args ) :
        for msg in self . consume ( ** kwargs ) :
            worker ( * args + ( msg , ) )
    return wrapper
if args :
    return decorator ( * args )
return decorator


signature:
def _arg_parser ( ) :
    """Factory for creating the argument parser"""

implementation:
description = "Converts a completezip to a litezip"
parser = argparse . ArgumentParser ( description = description )
verbose_group = parser . add_mutually_exclusive_group ( )
verbose_group . add_argument ( '-v' , '--verbose' , action = 'store_true' , dest = 'verbose' , default = None , help = "increase verbosity" )
verbose_group . add_argument ( '-q' , '--quiet' , action = 'store_false' , dest = 'verbose' , default = None , help = "print nothing to stdout or stderr" )
parser . add_argument ( 'location' , help = "Location of the unpacked litezip" )
return parser


signature:
def to_bytes ( self , previous : bytes ) :
    """Complex code ahead . Comments have been added in as needed ."""

implementation:
# First , validate the lengths .
if len ( self . conditions ) != len ( self . body ) :
    raise exc . CompileError ( "Conditions and body length mismatch!" )
bc = b""
prev_len = len ( previous )
# Loop over the conditions and bodies
for condition , body in zip ( self . conditions , self . body ) : # Generate the conditional data .
    cond_bytecode = condition . to_bytecode ( previous )
    bc += cond_bytecode
    # Complex calculation . First , generate the bytecode for all tokens in the body . Then
    # we calculate the len ( ) of that . We create a POP _ JUMP _ IF _ FALSE operation that jumps
    # to the instructions after the body code + 3 for the pop call . This is done for all
    # chained IF calls , as if it was an elif call . Else calls are not possible to be
    # auto - generated , but it is possible to emulate them using an elif call that checks
    # for the opposite of the above IF .
    # Call the _ compile _ func method from compiler to compile the body .
    body_bc = compiler . compile_bytecode ( body )
    bdyl = len ( body_bc )
    # Add together the lengths .
    gen_len = prev_len + len ( cond_bytecode ) + bdyl + 1
    # Generate the POP _ JUMP _ IF _ FALSE instruction
    bc += generate_simple_call ( tokens . POP_JUMP_IF_FALSE , gen_len )
    # Add the body _ bc
    bc += body_bc
    # Update previous _ len
    prev_len = len ( previous ) + len ( bc )
return bc


signature:
def to_bytes_35 ( self , previous : bytes ) :
    """A to - bytes specific to Python 3.5 and below ."""

implementation:
# Calculations ahead .
bc = b""
# Calculate the length of the iterator .
it_bc = util . generate_bytecode_from_obb ( self . iterator , previous )
bc += it_bc
# Push a get _ iter on .
bc += util . generate_bytecode_from_obb ( tokens . GET_ITER , b"" )
prev_len = len ( previous ) + len ( bc )
# Calculate the bytecode for the body .
body_bc = b""
for op in self . _body : # Add padding bytes to the bytecode to allow if blocks to work .
    padded_bc = previous
    # Add padding for SETUP _ LOOP
    padded_bc += b"\x00\x00\x00"
    padded_bc += bc
    # Add padding for FOR _ ITER
    padded_bc += b"\x00\x00\x00"
    # Add previous body
    padded_bc += body_bc
    body_bc += util . generate_bytecode_from_obb ( op , padded_bc )
# Add a JUMP _ ABSOLUTE
body_bc += util . generate_simple_call ( tokens . JUMP_ABSOLUTE , prev_len + 3 )
# Add a POP _ TOP
body_bc += util . generate_bytecode_from_obb ( tokens . POP_BLOCK , b"" )
# Calculate the right lengths .
# Add a FOR _ ITER , using len ( body _ bc )
body_bc = util . generate_simple_call ( tokens . FOR_ITER , len ( body_bc ) - 1 ) + body_bc
# Add the SETUP _ LOOP call
bc = util . generate_simple_call ( tokens . SETUP_LOOP , prev_len + len ( body_bc ) - 6 ) + bc + body_bc
return bc


signature:
def to_bytes_36 ( self , previous : bytes ) :
    """A to - bytes specific to Python 3.6 and above ."""

implementation:
# Calculations ahead .
bc = b""
# Calculate the length of the iterator .
it_bc = util . generate_bytecode_from_obb ( self . iterator , previous )
bc += it_bc
bc += util . ensure_instruction ( tokens . GET_ITER )


signature:
def validate_content ( * objs ) :
    """Runs the correct validator for given ` obj ` ects . Assumes all same type"""

implementation:
from . main import Collection , Module
validator = { Collection : cnxml . validate_collxml , Module : cnxml . validate_cnxml , } [ type ( objs [ 0 ] ) ]
return validator ( * [ obj . file for obj in objs ] )


signature:
def validate_litezip ( struct ) :
    """Validate the given litezip as ` struct ` .
 Returns a list of validation messages ."""

implementation:
msgs = [ ]
def _fmt_err ( err ) :
    return ( Path ( err . filename ) , "{}:{} -- {}: {}" . format ( * ( err [ 1 : ] ) ) )
obj_by_type = { }
for obj in struct :
    if not is_valid_identifier ( obj . id ) :
        msg = ( obj . file . parent , "{} is not a valid identifier" . format ( obj . id ) , )
        logger . info ( "{}: {}" . format ( * msg ) )
        msgs . append ( msg )
    obj_by_type . setdefault ( type ( obj ) , [ ] ) . append ( obj )
for obtype in obj_by_type :
    content_msgs = list ( [ _fmt_err ( err ) for err in validate_content ( * obj_by_type [ obtype ] ) ] )
    for msg in content_msgs :
        logger . info ( "{}: {}" . format ( * msg ) )
    msgs . extend ( content_msgs )
return msgs


signature:
def ensure_instruction ( instruction : int ) -> bytes :
    """Wraps an instruction to be Python 3.6 + compatible . This does nothing on Python 3.5 and below .
 This is most useful for operating on bare , single - width instructions such as
 ` ` RETURN _ FUNCTION ` ` in a version portable way .
 : param instruction : The instruction integer to use .
 : return : A safe bytes object , if applicable ."""

implementation:
if PY36 :
    return instruction . to_bytes ( 2 , byteorder = "little" )
else :
    return instruction . to_bytes ( 1 , byteorder = "little" )


signature:
def pack_value ( index : int ) -> bytes :
    """Small helper value to pack an index value into bytecode .
 This is used for version compat between 3.5 - and 3.6 +
 : param index : The item to pack .
 : return : The packed item ."""

implementation:
if PY36 :
    return index . to_bytes ( 1 , byteorder = "little" )
else :
    return index . to_bytes ( 2 , byteorder = "little" )


signature:
def generate_simple_call ( opcode : int , index : int ) :
    """Generates a simple call , with an index for something .
 : param opcode : The opcode to generate .
 : param index : The index to use as an argument .
 : return :"""

implementation:
bs = b""
# add the opcode
bs += opcode . to_bytes ( 1 , byteorder = "little" )
# Add the index
if isinstance ( index , int ) :
    if PY36 :
        bs += index . to_bytes ( 1 , byteorder = "little" )
    else :
        bs += index . to_bytes ( 2 , byteorder = "little" )
else :
    bs += index
return bs


signature:
def generate_bytecode_from_obb ( obb : object , previous : bytes ) -> bytes :
    """Generates a bytecode from an object .
 : param obb : The object to generate .
 : param previous : The previous bytecode to use when generating subobjects .
 : return : The generated bytecode ."""

implementation:
# Generates bytecode from a specified object , be it a validator or an int or bytes even .
if isinstance ( obb , pyte . superclasses . _PyteOp ) :
    return obb . to_bytes ( previous )
elif isinstance ( obb , ( pyte . superclasses . _PyteAugmentedComparator , pyte . superclasses . _PyteAugmentedValidator . _FakeMathematicalOP ) ) :
    return obb . to_bytes ( previous )
elif isinstance ( obb , pyte . superclasses . _PyteAugmentedValidator ) :
    obb . validate ( )
    return obb . to_load ( )
elif isinstance ( obb , int ) :
    return obb . to_bytes ( ( obb . bit_length ( ) + 7 ) // 8 , byteorder = "little" ) or b''
elif isinstance ( obb , bytes ) :
    return obb
else :
    raise TypeError ( "`{}` was not a valid bytecode-encodable item" . format ( obb ) )


signature:
def _get_const_info ( const_index , const_list ) :
    """Helper to get optional details about const references
 Returns the dereferenced constant and its repr if the constant
 list is defined .
 Otherwise returns the constant index and its repr ( ) ."""

implementation:
argval = const_index
if const_list is not None :
    try :
        argval = const_list [ const_index ]
    except IndexError :
        raise ValidationError ( "Consts value out of range: {}" . format ( const_index ) ) from None
return argval , repr ( argval )


signature:
def _get_name_info ( name_index , name_list ) :
    """Helper to get optional details about named references
 Returns the dereferenced name as both value and repr if the name
 list is defined .
 Otherwise returns the name index and its repr ( ) ."""

implementation:
argval = name_index
if name_list is not None :
    try :
        argval = name_list [ name_index ]
    except IndexError :
        raise ValidationError ( "Names value out of range: {}" . format ( name_index ) ) from None
    argrepr = argval
else :
    argrepr = repr ( argval )
return argval , argrepr


signature:
def add_to_parser ( parser , usage , ignore_existing = False ) :
    """Add arguments described in the usage string to the
 parser . View more details at the < create _ parser > docs
 Args :
 parser ( ArgumentParser ) : parser to add arguments to
 usage ( str ) : Usage string in the format described above
 ignore _ existing ( bool ) : Ignore any arguments that have already been defined"""

implementation:
usage = '\n' + usage
first_line = [ i for i in usage . split ( '\n' ) if i ] [ 0 ]
indent = ' ' * ( len ( first_line ) - len ( first_line . lstrip ( ' ' ) ) )
usage = usage . replace ( '\n' + indent , '\n' )
usage = usage . replace ( '\n...' , '' )
defaults = { }
description , * descriptors = usage . split ( '\n:' )
description = description . replace ( '\n' , ' ' ) . strip ( )
if description and ( not parser . description or not ignore_existing ) :
    parser . description = description
for descriptor in descriptors :
    try :
        options , * info = descriptor . split ( '\n' )
        info = ' ' . join ( i for i in info if i ) . replace ( '    ' , '' )
        if options . count ( ' ' ) == 1 :
            if options [ 0 ] == '-' :
                short , long = options . split ( ' ' )
                var_name = long . strip ( '-' ) . replace ( '-' , '_' )
                parser . add_argument ( short , long , dest = var_name , action = 'store_true' , help = info )
                defaults [ var_name ] = False
            else :
                short , typ = options . split ( ' ' )
                parser . add_argument ( short , type = types [ typ ] , help = info )
        else :
            short , long , typ , default = options . split ( ' ' )
            info = info . rstrip ( ) + '. Default: ' + default
            default = '' if default == '-' else default
            parser . add_argument ( short , long , type = types [ typ ] , default = default , help = info )
    except ArgumentError :
        if not ignore_existing :
            raise
    except Exception as e :
        print ( e . __class__ . __name__ + ': ' + str ( e ) )
        print ( 'While parsing:' )
        print ( descriptor )
        raise ValueError ( 'Failed to create parser from usage string' )


signature:
def compile_bytecode ( code : list ) -> bytes :
    """Compiles Pyte objects into a bytecode list .
 : param code : A list of objects to compile .
 : return : The computed bytecode ."""

implementation:
bc = b""
for i , op in enumerate ( code ) :
    try : # Get the bytecode .
        if isinstance ( op , _PyteOp ) or isinstance ( op , _PyteAugmentedComparator ) :
            bc_op = op . to_bytes ( bc )
        elif isinstance ( op , int ) :
            bc_op = op . to_bytes ( 1 , byteorder = "little" )
        elif isinstance ( op , bytes ) :
            bc_op = op
        else :
            raise CompileError ( "Could not compile code of type {}" . format ( type ( op ) ) )
        bc += bc_op
    except Exception as e :
        print ( "Fatal compiliation error on operator {i} ({op})." . format ( i = i , op = op ) )
        raise e
return bc


signature:
def _simulate_stack ( code : list ) -> int :
    """Simulates the actions of the stack , to check safety .
 This returns the maximum needed stack ."""

implementation:
max_stack = 0
curr_stack = 0
def _check_stack ( ins ) :
    if curr_stack < 0 :
        raise CompileError ( "Stack turned negative on instruction: {}" . format ( ins ) )
    if curr_stack > max_stack :
        return curr_stack
# Iterate over the bytecode .
for instruction in code :
    assert isinstance ( instruction , dis . Instruction )
    if instruction . arg is not None :
        try :
            effect = dis . stack_effect ( instruction . opcode , instruction . arg )
        except ValueError as e :
            raise CompileError ( "Invalid opcode `{}` when compiling" . format ( instruction . opcode ) ) from e
    else :
        try :
            effect = dis . stack_effect ( instruction . opcode )
        except ValueError as e :
            raise CompileError ( "Invalid opcode `{}` when compiling" . format ( instruction . opcode ) ) from e
    curr_stack += effect
    # Re - check the stack .
    _should_new_stack = _check_stack ( instruction )
    if _should_new_stack :
        max_stack = _should_new_stack
return max_stack


signature:
def compile ( code : list , consts : list , names : list , varnames : list , func_name : str = "<unknown, compiled>" , arg_count : int = 0 , kwarg_defaults : Tuple [ Any ] = ( ) , use_safety_wrapper : bool = True ) :
    """Compiles a set of bytecode instructions into a working function , using Python ' s bytecode
 compiler .
 : param code : A list of bytecode instructions .
 : param consts : A list of constants to compile into the function .
 : param names : A list of names to compile into the function .
 : param varnames : A list of ` ` varnames ` ` to compile into the function .
 : param func _ name : The name of the function to use .
 : param arg _ count : The number of arguments this function takes . Must be ` ` < = len ( varnames ) ` ` .
 : param kwarg _ defaults : A tuple of defaults for kwargs .
 : param use _ safety _ wrapper : Use the safety wrapper ? This hijacks SystemError to print better stack traces ."""

implementation:
varnames = tuple ( varnames )
consts = tuple ( consts )
names = tuple ( names )
# Flatten the code list .
code = util . flatten ( code )
if arg_count > len ( varnames ) :
    raise CompileError ( "arg_count > len(varnames)" )
if len ( kwarg_defaults ) > len ( varnames ) :
    raise CompileError ( "len(kwarg_defaults) > len(varnames)" )
# Compile it .
bc = compile_bytecode ( code )
dis . dis ( bc )
# Check for a final RETURN _ VALUE .
if PY36 : # TODO : Add Python 3.6 check
    pass
else :
    if bc [ - 1 ] != tokens . RETURN_VALUE :
        raise CompileError ( "No default RETURN_VALUE. Add a `pyte.tokens.RETURN_VALUE` to the end of your " "bytecode if you don't need one." )
# Set default flags
flags = 1 | 2 | 64
frame_data = inspect . stack ( ) [ 1 ]
if sys . version_info [ 0 : 2 ] > ( 3 , 3 ) : # Validate the stack .
    stack_size = _simulate_stack ( dis . _get_instructions_bytes ( bc , constants = consts , names = names , varnames = varnames ) )
else :
    warnings . warn ( "Cannot check stack for safety." )
    stack_size = 99
# Generate optimization warnings .
_optimize_warn_pass ( dis . _get_instructions_bytes ( bc , constants = consts , names = names , varnames = varnames ) )
obb = types . CodeType ( arg_count , # Varnames - used for arguments .
0 , # Kwargs are not supported yet
len ( varnames ) , # co _ nlocals - > Non - argument local variables
stack_size , # Auto - calculated
flags , # 67 is default for a normal function .
bc , # co _ code - use the bytecode we generated .
consts , # co _ consts
names , # co _ names , used for global calls .
varnames , # arguments
frame_data [ 1 ] , # use < unknown , compiled >
func_name , # co _ name
frame_data [ 2 ] , # co _ firstlineno , ignore this .
b'' , # https : / / svn . python . org / projects / python / trunk / Objects / lnotab _ notes . txt
( ) , # freevars - no idea what this does
( ) # cellvars - used for nested functions - we don ' t use these .
)
# Update globals
f_globals = frame_data [ 0 ] . f_globals
# Create a function type .
f = types . FunctionType ( obb , f_globals )
f . __name__ = func_name
f . __defaults__ = kwarg_defaults
if use_safety_wrapper :
    def __safety_wrapper ( * args , ** kwargs ) :
        try :
            return f ( * args , ** kwargs )
        except SystemError as e :
            if 'opcode' not in ' ' . join ( e . args ) : # Re - raise any non opcode related errors .
                raise
            msg = "Bytecode exception!" "\nFunction {} returned an invalid opcode." "\nFunction dissection:\n\n" . format ( f . __name__ )
            # dis sucks and always prints to stdout
            # so we capture it
            file = io . StringIO ( )
            with contextlib . redirect_stdout ( file ) :
                dis . dis ( f )
            msg += file . getvalue ( )
            raise SystemError ( msg ) from e
    returned_func = __safety_wrapper
    returned_func . wrapped = f
else :
    returned_func = f
# return the func
return returned_func


signature:
def _parse_document_id ( elm_tree ) :
    """Given the parsed xml to an ` ElementTree ` ,
 parse the id from the content ."""

implementation:
xpath = '//md:content-id/text()'
return [ x for x in elm_tree . xpath ( xpath , namespaces = COLLECTION_NSMAP ) ] [ 0 ]


signature:
def _find_resources ( directory , excludes = [ ] ) :
    """Return a list of resource paths from the directory .
 Ignore records via the list of ` excludes ` ,
 which are callables that take a file parameter ( as a ` Path ` instance ) ."""

implementation: return sorted ( [ r for r in directory . glob ( '*' ) if True not in [ e ( r ) for e in excludes ] ] )


signature:
def parse_module ( path , excludes = None ) :
    """Parse the file structure to a data structure given the path to
 a module directory ."""

implementation:
file = path / MODULE_FILENAME
if not file . exists ( ) :
    raise MissingFile ( file )
id = _parse_document_id ( etree . parse ( file . open ( ) ) )
excludes = excludes or [ ]
excludes . extend ( [ lambda filepath : filepath . name == MODULE_FILENAME , ] )
resources_paths = _find_resources ( path , excludes = excludes )
resources = tuple ( _resource_from_path ( res ) for res in resources_paths )
return Module ( id , file , resources )


signature:
def parse_collection ( path , excludes = None ) :
    """Parse a file structure to a data structure given the path to
 a collection directory ."""

implementation:
file = path / COLLECTION_FILENAME
if not file . exists ( ) :
    raise MissingFile ( file )
id = _parse_document_id ( etree . parse ( file . open ( ) ) )
excludes = excludes or [ ]
excludes . extend ( [ lambda filepath : filepath . name == COLLECTION_FILENAME , lambda filepath : filepath . is_dir ( ) , ] )
resources_paths = _find_resources ( path , excludes = excludes )
resources = tuple ( _resource_from_path ( res ) for res in resources_paths )
return Collection ( id , file , resources )


signature:
def parse_litezip ( path ) :
    """Parse a litezip file structure to a data structure given the path
 to the litezip directory ."""

implementation:
struct = [ parse_collection ( path ) ]
struct . extend ( [ parse_module ( x ) for x in path . iterdir ( ) if x . is_dir ( ) and x . name . startswith ( 'm' ) ] )
return tuple ( sorted ( struct ) )


signature:
def convert_completezip ( path ) :
    """Converts a completezip file structure to a litezip file structure .
 Returns a litezip data structure ."""

implementation:
for filepath in path . glob ( '**/index_auto_generated.cnxml' ) :
    filepath . rename ( filepath . parent / 'index.cnxml' )
    logger . debug ( 'removed {}' . format ( filepath ) )
for filepath in path . glob ( '**/index.cnxml.html' ) :
    filepath . unlink ( )
return parse_litezip ( path )


signature:
def _get_instructions_bytes ( code , varnames = None , names = None , constants = None , cells = None , linestarts = None , line_offset = 0 ) :
    """Iterate over the instructions in a bytecode string .
 Generates a sequence of Instruction namedtuples giving the details of each
 opcode . Additional information about the code ' s runtime environment
 ( e . g . variable names , constants ) can be specified using optional
 arguments ."""

implementation:
labels = dis . findlabels ( code )
extended_arg = 0
starts_line = None
free = None
# enumerate ( ) is not an option , since we sometimes process
# multiple elements on a single pass through the loop
n = len ( code )
i = 0
while i < n :
    op = code [ i ]
    offset = i
    if linestarts is not None :
        starts_line = linestarts . get ( i , None )
        if starts_line is not None :
            starts_line += line_offset
    is_jump_target = i in labels
    i = i + 1
    arg = None
    argval = None
    argrepr = ''
    if op >= dis . HAVE_ARGUMENT :
        arg = code [ i ] + code [ i + 1 ] * 256 + extended_arg
        extended_arg = 0
        i = i + 2
        if op == dis . EXTENDED_ARG :
            extended_arg = arg * 65536
        # Set argval to the dereferenced value of the argument when
        # availabe , and argrepr to the string representation of argval .
        # _ disassemble _ bytes needs the string repr of the
        # raw name index for LOAD _ GLOBAL , LOAD _ CONST , etc .
        argval = arg
        if op in dis . hasconst :
            argval , argrepr = dis . _get_const_info ( arg , constants )
        elif op in dis . hasname :
            argval , argrepr = dis . _get_name_info ( arg , names )
        elif op in dis . hasjrel :
            argval = i + arg
            argrepr = "to " + repr ( argval )
        elif op in dis . haslocal :
            argval , argrepr = dis . _get_name_info ( arg , varnames )
        elif op in dis . hascompare :
            argval = dis . cmp_op [ arg ]
            argrepr = argval
        elif op in dis . hasfree :
            argval , argrepr = dis . _get_name_info ( arg , cells )
        elif op in dis . hasnargs :
            argrepr = "%d positional, %d keyword pair" % ( code [ i - 2 ] , code [ i - 1 ] )
    yield dis . Instruction ( dis . opname [ op ] , op , arg , argval , argrepr , offset , starts_line , is_jump_target )


signature:
def _disassemble ( self , lineno_width = 3 , mark_as_current = False ) :
    """Format instruction details for inclusion in disassembly output
 * lineno _ width * sets the width of the line number field ( 0 omits it )
 * mark _ as _ current * inserts a ' - - > ' marker arrow as part of the line"""

implementation:
fields = [ ]
# Column : Source code line number
if lineno_width :
    if self . starts_line is not None :
        lineno_fmt = "%%%dd" % lineno_width
        fields . append ( lineno_fmt % self . starts_line )
    else :
        fields . append ( ' ' * lineno_width )
# Column : Current instruction indicator
if mark_as_current :
    fields . append ( '-->' )
else :
    fields . append ( '   ' )
# Column : Jump target marker
if self . is_jump_target :
    fields . append ( '>>' )
else :
    fields . append ( '  ' )
# Column : Instruction offset from start of code sequence
fields . append ( repr ( self . offset ) . rjust ( 4 ) )
# Column : Opcode name
fields . append ( self . opname . ljust ( 20 ) )
# Column : Opcode argument
if self . arg is not None :
    fields . append ( repr ( self . arg ) . rjust ( 5 ) )
    # Column : Opcode argument details
    if self . argrepr :
        fields . append ( '(' + self . argrepr + ')' )
return ' ' . join ( fields ) . rstrip ( )


signature:
def intersection ( l1 , l2 ) :
    '''Returns intersection of two lists . Assumes the lists are sorted by start positions'''

implementation:
if len ( l1 ) == 0 or len ( l2 ) == 0 :
    return [ ]
out = [ ]
l2_pos = 0
for l in l1 :
    while l2_pos < len ( l2 ) and l2 [ l2_pos ] . end < l . start :
        l2_pos += 1
    if l2_pos == len ( l2 ) :
        break
    while l2_pos < len ( l2 ) and l . intersects ( l2 [ l2_pos ] ) :
        out . append ( l . intersection ( l2 [ l2_pos ] ) )
        l2_pos += 1
    l2_pos = max ( 0 , l2_pos - 1 )
return out


signature:
def merge_overlapping_in_list ( l ) :
    '''Sorts list , merges any overlapping intervals , and also adjacent intervals . e . g .
 [0,1 ] , [ 1,2 ] would be merge to [ 0 , . 2 ] .'''

implementation:
i = 0
l . sort ( )
while i < len ( l ) - 1 :
    u = l [ i ] . union ( l [ i + 1 ] )
    if u is not None :
        l [ i ] = u
        l . pop ( i + 1 )
    else :
        i += 1


signature:
def remove_contained_in_list ( l ) :
    '''Sorts list in place , then removes any intervals that are completely
 contained inside another interval'''

implementation:
i = 0
l . sort ( )
while i < len ( l ) - 1 :
    if l [ i + 1 ] . contains ( l [ i ] ) :
        l . pop ( i )
    elif l [ i ] . contains ( l [ i + 1 ] ) :
        l . pop ( i + 1 )
    else :
        i += 1


signature:
def distance_to_point ( self , p ) :
    '''Returns the distance from the point to the interval . Zero if the point lies inside the interval .'''

implementation:
if self . start <= p <= self . end :
    return 0
else :
    return min ( abs ( self . start - p ) , abs ( self . end - p ) )


signature:
def intersects ( self , i ) :
    '''Returns true iff this interval intersects the interval i'''

implementation: return self . start <= i . end and i . start <= self . end


signature:
def contains ( self , i ) :
    '''Returns true iff this interval contains the interval i'''

implementation: return self . start <= i . start and i . end <= self . end


signature:
def union ( self , i ) :
    '''If intervals intersect , returns their union , otherwise returns None'''

implementation:
if self . intersects ( i ) or self . end + 1 == i . start or i . end + 1 == self . start :
    return Interval ( min ( self . start , i . start ) , max ( self . end , i . end ) )
else :
    return None


signature:
def union_fill_gap ( self , i ) :
    '''Like union , but ignores whether the two intervals intersect or not'''

implementation: return Interval ( min ( self . start , i . start ) , max ( self . end , i . end ) )


signature:
def intersection ( self , i ) :
    '''If intervals intersect , returns their intersection , otherwise returns None'''

implementation:
if self . intersects ( i ) :
    return Interval ( max ( self . start , i . start ) , min ( self . end , i . end ) )
else :
    return None


signature:
def file_reader ( fname , read_quals = False ) :
    '''Iterates over a FASTA or FASTQ file , yielding the next sequence in the file until there are no more sequences'''

implementation:
f = utils . open_file_read ( fname )
line = f . readline ( )
phylip_regex = re . compile ( '^\s*[0-9]+\s+[0-9]+$' )
gbk_regex = re . compile ( '^LOCUS\s+\S' )
if line . startswith ( '>' ) :
    seq = Fasta ( )
    previous_lines [ f ] = line
elif line . startswith ( '##gff-version 3' ) :
    seq = Fasta ( )
    # if a GFF file , need to skip past all the annotation
    # and get to the fasta sequences at the end of the file
    while not line . startswith ( '>' ) :
        line = f . readline ( )
        if not line :
            utils . close ( f )
            raise Error ( 'No sequences found in GFF file "' + fname + '"' )
    seq = Fasta ( )
    previous_lines [ f ] = line
elif line . startswith ( 'ID   ' ) and line [ 5 ] != ' ' :
    seq = Embl ( )
    previous_lines [ f ] = line
elif gbk_regex . search ( line ) :
    seq = Embl ( )
    previous_lines [ f ] = line
elif line . startswith ( '@' ) :
    seq = Fastq ( )
    previous_lines [ f ] = line
elif phylip_regex . search ( line ) : # phylip format could be interleaved or not , need to look at next
# couple of lines to figure that out . Don ' t expect these files to
# be too huge , so just store all the sequences in memory
    number_of_seqs , bases_per_seq = line . strip ( ) . split ( )
    number_of_seqs = int ( number_of_seqs )
    bases_per_seq = int ( bases_per_seq )
    got_blank_line = False
    first_line = line
    seq_lines = [ ]
    while 1 :
        line = f . readline ( )
        if line == '' :
            break
        elif line == '\n' :
            got_blank_line = True
        else :
            seq_lines . append ( line . rstrip ( ) )
    utils . close ( f )
    if len ( seq_lines ) == 1 or len ( seq_lines ) == number_of_seqs :
        sequential = True
    elif seq_lines [ 0 ] [ 10 ] != ' ' and seq_lines [ 1 ] [ 10 ] == ' ' :
        sequential = True
    else :
        sequential = False
    # if the 11th char of second sequence line is a space , then the file is sequential , e . g . :
    # GAGCCCGGGC AATACAGGGT AT
    # as opposed to :
    # Salmo gairAAGCCTTGGC AGTGCAGGGT
    if sequential :
        current_id = None
        current_seq = ''
        for line in seq_lines :
            if len ( current_seq ) == bases_per_seq or len ( current_seq ) == 0 :
                if current_id is not None :
                    yield Fasta ( current_id , current_seq . replace ( '-' , '' ) )
                current_seq = ''
                current_id , new_bases = line [ 0 : 10 ] . rstrip ( ) , line . rstrip ( ) [ 10 : ]
            else :
                new_bases = line . rstrip ( )
            current_seq += new_bases . replace ( ' ' , '' )
        yield Fasta ( current_id , current_seq . replace ( '-' , '' ) )
    else : # seaview files start all seqs at pos > = 12 . Other files start
    # their sequence at the start of the line
        if seq_lines [ number_of_seqs + 1 ] [ 0 ] == ' ' :
            first_gap_pos = seq_lines [ 0 ] . find ( ' ' )
            end_of_gap = first_gap_pos
            while seq_lines [ 0 ] [ end_of_gap ] == ' ' :
                end_of_gap += 1
            first_seq_base = end_of_gap
        else :
            first_seq_base = 10
        seqs = [ ]
        for i in range ( number_of_seqs ) :
            name , bases = seq_lines [ i ] [ 0 : first_seq_base ] . rstrip ( ) , seq_lines [ i ] [ first_seq_base : ]
            seqs . append ( Fasta ( name , bases ) )
        for i in range ( number_of_seqs , len ( seq_lines ) ) :
            seqs [ i % number_of_seqs ] . seq += seq_lines [ i ]
        for fa in seqs :
            fa . seq = fa . seq . replace ( ' ' , '' ) . replace ( '-' , '' )
            yield fa
    return
elif line == '' :
    utils . close ( f )
    return
else :
    utils . close ( f )
    raise Error ( 'Error determining file type from file "' + fname + '". First line is:\n' + line . rstrip ( ) )
try :
    while seq . get_next_from_file ( f , read_quals ) :
        yield seq
finally :
    utils . close ( f )


signature:
def subseq ( self , start , end ) :
    '''Returns Fasta object with the same name , of the bases from start to end , but not including end'''

implementation: return Fasta ( self . id , self . seq [ start : end ] )


signature:
def split_capillary_id ( self ) :
    '''Gets the prefix and suffix of an name of a capillary read , e . g . xxxxx . p1k or xxxx . q1k . Returns a tuple ( prefix , suffx )'''

implementation:
try :
    a = self . id . rsplit ( '.' , 1 )
    if a [ 1 ] . startswith ( 'p' ) :
        dir = 'fwd'
    elif a [ 1 ] . startswith ( 'q' ) :
        dir = 'rev'
    else :
        dir = 'unk'
    return { 'prefix' : a [ 0 ] , 'dir' : dir , 'suffix' : a [ 1 ] }
except :
    raise Error ( 'Error in split_capillary_id() on ID' , self . id )


signature:
def expand_nucleotides ( self ) :
    '''Assumes sequence is nucleotides . Returns list of all combinations of redundant nucleotides . e . g . R is A or G , so CRT would have combinations CAT and CGT'''

implementation:
s = list ( self . seq )
for i in range ( len ( s ) ) :
    if s [ i ] in redundant_nts :
        s [ i ] = '' . join ( redundant_nts [ s [ i ] ] )
seqs = [ ]
for x in itertools . product ( * s ) :
    seqs . append ( Fasta ( self . id + '.' + str ( len ( seqs ) + 1 ) , '' . join ( x ) ) )
return seqs


signature:
def strip_illumina_suffix ( self ) :
    '''Removes any trailing / 1 or / 2 off the end of the name'''

implementation:
if self . id . endswith ( '/1' ) or self . id . endswith ( '/2' ) :
    self . id = self . id [ : - 2 ]


signature:
def is_all_Ns ( self , start = 0 , end = None ) :
    '''Returns true if the sequence is all Ns ( upper or lower case )'''

implementation:
if end is not None :
    if start > end :
        raise Error ( 'Error in is_all_Ns. Start coord must be <= end coord' )
    end += 1
else :
    end = len ( self )
if len ( self ) == 0 :
    return False
else :
    return re . search ( '[^Nn]' , self . seq [ start : end ] ) is None


signature:
def add_insertions ( self , skip = 10 , window = 1 , test = False ) :
    '''Adds a random base within window bases around every skip bases . e . g . skip = 10 , window = 1 means a random base added somwhere in theintervals [ 9,11 ] , [ 19,21 ] . . .'''

implementation:
assert 2 * window < skip
new_seq = list ( self . seq )
for i in range ( len ( self ) - skip , 0 , - skip ) :
    pos = random . randrange ( i - window , i + window + 1 )
    base = random . choice ( [ 'A' , 'C' , 'G' , 'T' ] )
    if test :
        base = 'N'
    new_seq . insert ( pos , base )
self . seq = '' . join ( new_seq )


signature:
def replace_bases ( self , old , new ) :
    '''Replaces all occurrences of ' old ' with ' new ' '''

implementation: self . seq = self . seq . replace ( old , new )


signature:
def replace_interval ( self , start , end , new ) :
    '''Replaces the sequence from start to end with the sequence " new "'''

implementation:
if start > end or start > len ( self ) - 1 or end > len ( self ) - 1 :
    raise Error ( 'Error replacing bases ' + str ( start ) + '-' + str ( end ) + ' in sequence ' + self . id )
self . seq = self . seq [ 0 : start ] + new + self . seq [ end + 1 : ]


signature:
def gaps ( self , min_length = 1 ) :
    '''Finds the positions of all gaps in the sequence that are at least min _ length long . Returns a list of Intervals . Coords are zero - based'''

implementation:
gaps = [ ]
regex = re . compile ( 'N+' , re . IGNORECASE )
for m in regex . finditer ( self . seq ) :
    if m . span ( ) [ 1 ] - m . span ( ) [ 0 ] + 1 >= min_length :
        gaps . append ( intervals . Interval ( m . span ( ) [ 0 ] , m . span ( ) [ 1 ] - 1 ) )
return gaps


signature:
def contig_coords ( self ) :
    '''Finds coords of contigs , i . e . everything that ' s not a gap ( N or n ) . Returns a list of Intervals . Coords are zero - based'''

implementation:
# contigs are the opposite of gaps , so work out the coords from the gap coords
gaps = self . gaps ( )
if len ( gaps ) == 0 :
    return [ intervals . Interval ( 0 , len ( self ) - 1 ) ]
coords = [ 0 ]
for g in gaps :
    if g . start == 0 :
        coords = [ g . end + 1 ]
    else :
        coords += [ g . start - 1 , g . end + 1 ]
if coords [ - 1 ] < len ( self ) :
    coords . append ( len ( self ) - 1 )
return [ intervals . Interval ( coords [ i ] , coords [ i + 1 ] ) for i in range ( 0 , len ( coords ) - 1 , 2 ) ]


signature:
def orfs ( self , frame = 0 , revcomp = False ) :
    '''Returns a list of ORFs that the sequence has , starting on the given
 frame . Each returned ORF is an interval . Interval object .
 If revomp = True , then finds the ORFs of the reverse complement
 of the sequence .'''

implementation:
assert frame in [ 0 , 1 , 2 ]
if revcomp :
    self . revcomp ( )
aa_seq = self . translate ( frame = frame ) . seq . rstrip ( 'X' )
if revcomp :
    self . revcomp ( )
orfs = _orfs_from_aa_seq ( aa_seq )
for i in range ( len ( orfs ) ) :
    if revcomp :
        start = len ( self ) - ( orfs [ i ] . end * 3 + 3 ) - frame
        end = len ( self ) - ( orfs [ i ] . start * 3 ) - 1 - frame
    else :
        start = orfs [ i ] . start * 3 + frame
        end = orfs [ i ] . end * 3 + 2 + frame
    orfs [ i ] = intervals . Interval ( start , end )
return orfs


signature:
def all_orfs ( self , min_length = 300 ) :
    '''Finds all open reading frames in the sequence , that are at least as
 long as min _ length . Includes ORFs on the reverse strand .
 Returns a list of ORFs , where each element is a tuple :
 ( interval . Interval , bool )
 where bool = True means on the reverse strand'''

implementation:
orfs = [ ]
for frame in [ 0 , 1 , 2 ] :
    for revcomp in [ False , True ] :
        orfs . extend ( [ ( t , revcomp ) for t in self . orfs ( frame = frame , revcomp = revcomp ) if len ( t ) >= min_length ] )
return sorted ( orfs , key = lambda t : t [ 0 ] )


signature:
def is_complete_orf ( self ) :
    '''Returns true iff length is > = 6 , is a multiple of 3 , and there is exactly one stop codon in the sequence and it is at the end'''

implementation:
if len ( self ) % 3 != 0 or len ( self ) < 6 :
    return False
orfs = self . orfs ( )
complete_orf = intervals . Interval ( 0 , len ( self ) - 1 )
for orf in orfs :
    if orf == complete_orf :
        return True
return False


signature:
def looks_like_gene ( self ) :
    '''Returns true iff : length > = 6 , length is a multiple of 3 , first codon is start , last codon is a stop and has no other stop codons'''

implementation: return self . is_complete_orf ( ) and len ( self ) >= 6 and len ( self ) % 3 == 0 and self . seq [ 0 : 3 ] . upper ( ) in genetic_codes . starts [ genetic_code ]


signature:
def make_into_gene ( self ) :
    '''Tries to make into a gene sequence . Tries all three reading frames and both strands . Returns a tuple ( new sequence , strand , frame ) if it was successful . Otherwise returns None .'''

implementation:
for reverse in [ True , False ] :
    for frame in range ( 3 ) :
        new_seq = copy . copy ( self )
        if reverse :
            new_seq . revcomp ( )
        new_seq . seq = new_seq [ frame : ]
        if len ( new_seq ) % 3 :
            new_seq . seq = new_seq . seq [ : - ( len ( new_seq ) % 3 ) ]
        new_aa_seq = new_seq . translate ( )
        if len ( new_aa_seq ) >= 2 and new_seq [ 0 : 3 ] in genetic_codes . starts [ genetic_code ] and new_aa_seq [ - 1 ] == '*' and '*' not in new_aa_seq [ : - 1 ] :
            strand = '-' if reverse else '+'
            return new_seq , strand , frame
return None


signature:
def trim ( self , start , end ) :
    '''Removes first ' start ' / ' end ' bases off the start / end of the sequence'''

implementation: self . seq = self . seq [ start : len ( self . seq ) - end ]


signature:
def to_Fastq ( self , qual_scores ) :
    '''Returns a Fastq object . qual _ scores expected to be a list of numbers , like you would get in a . qual file'''

implementation:
if len ( self ) != len ( qual_scores ) :
    raise Error ( 'Error making Fastq from Fasta, lengths differ.' , self . id )
return Fastq ( self . id , self . seq , '' . join ( [ chr ( max ( 0 , min ( x , 93 ) ) + 33 ) for x in qual_scores ] ) )


signature:
def search ( self , search_string ) :
    '''Finds every occurrence ( including overlapping ones ) of the search _ string , including on the reverse strand . Returns a list where each element is a tuple ( position , strand ) where strand is in [ ' - ' , ' + ' ] . Positions are zero - based'''

implementation:
seq = self . seq . upper ( )
search_string = search_string . upper ( )
pos = 0
found = seq . find ( search_string , pos )
hits = [ ]
while found != - 1 :
    hits . append ( ( found , '+' ) )
    pos = found + 1
    found = seq . find ( search_string , pos )
pos = 0
search_string = Fasta ( 'x' , search_string )
search_string . revcomp ( )
search_string = search_string . seq
found = seq . find ( search_string , pos )
while found != - 1 :
    hits . append ( ( found , '-' ) )
    pos = found + 1
    found = seq . find ( search_string , pos )
return hits


signature:
def translate ( self , frame = 0 ) :
    '''Returns a Fasta sequence , translated into amino acids . Starts translating from ' frame ' , where frame expected to be 0,1 or 2'''

implementation: return Fasta ( self . id , '' . join ( [ genetic_codes . codes [ genetic_code ] . get ( self . seq [ x : x + 3 ] . upper ( ) , 'X' ) for x in range ( frame , len ( self ) - 1 - frame , 3 ) ] ) )


signature:
def gc_content ( self , as_decimal = True ) :
    """Returns the GC content for the sequence .
 Notes :
 This method ignores N when calculating the length of the sequence .
 It does not , however ignore other ambiguous bases . It also only
 includes the ambiguous base S ( G or C ) . In this sense the method is
 conservative with its calculation .
 Args :
 as _ decimal ( bool ) : Return the result as a decimal . Setting to False
 will return as a percentage . i . e for the sequence GCAT it will
 return 0.5 by default and 50.00 if set to False .
 Returns :
 float : GC content calculated as the number of G , C , and S divided
 by the number of ( non - N ) bases ( length ) ."""

implementation:
gc_total = 0.0
num_bases = 0.0
n_tuple = tuple ( 'nN' )
accepted_bases = tuple ( 'cCgGsS' )
# counter sums all unique characters in sequence . Case insensitive .
for base , count in Counter ( self . seq ) . items ( ) : # dont count N in the number of bases
    if base not in n_tuple :
        num_bases += count
        if base in accepted_bases : # S is a G or C
            gc_total += count
gc_content = gc_total / num_bases
if not as_decimal : # return as percentage
    gc_content *= 100
return gc_content


signature:
def subseq ( self , start , end ) :
    '''Returns Fastq object with the same name , of the bases from start to end , but not including end'''

implementation: return Fastq ( self . id , self . seq [ start : end ] , self . qual [ start : end ] )


implementation:
super ( ) . trim ( start , end )
self . qual = self . qual [ start : len ( self . qual ) - end ]


signature:
def trim_Ns ( self ) :
    '''Removes any leading or trailing N or n characters from the sequence'''

implementation:
# get index of first base that is not an N
i = 0
while i < len ( self ) and self . seq [ i ] in 'nN' :
    i += 1
# strip off start of sequence and quality
self . seq = self . seq [ i : ]
self . qual = self . qual [ i : ]
# strip the ends
self . seq = self . seq . rstrip ( 'Nn' )
self . qual = self . qual [ : len ( self . seq ) ]


signature:
def replace_interval ( self , start , end , new , qual_string ) :
    '''Replaces the sequence from start to end with the sequence " new "'''

implementation:
if len ( new ) != len ( qual_string ) :
    raise Error ( 'Length of new seq and qual string in replace_interval() must be equal. Cannot continue' )
super ( ) . replace_interval ( start , end , new )
self . qual = self . qual [ 0 : start ] + qual_string + self . qual [ end + 1 : ]


signature:
def translate ( self ) :
    '''Returns a Fasta sequence , translated into amino acids . Starts translating from ' frame ' , where frame expected to be 0,1 or 2'''

implementation:
fa = super ( ) . translate ( )
return Fastq ( fa . id , fa . seq , 'I' * len ( fa . seq ) )


signature:
def acgtn_only ( infile , outfile ) :
    '''Replace every non - acgtn ( case insensitve ) character with an N'''

implementation:
f = utils . open_file_write ( outfile )
for seq in sequences . file_reader ( infile ) :
    seq . replace_non_acgt ( )
    print ( seq , file = f )
utils . close ( f )


signature:
def caf_to_fastq ( infile , outfile , min_length = 0 , trim = False ) :
    '''Convert a CAF file to fastq . Reads shorter than min _ length are not output . If clipping information is in the CAF file ( with a line Clipping QUAL . . . ) and trim = True , then trim the reads'''

implementation:
caf_reader = caf . file_reader ( infile )
fout = utils . open_file_write ( outfile )
for c in caf_reader :
    if trim :
        if c . clip_start is not None and c . clip_end is not None :
            c . seq . seq = c . seq . seq [ c . clip_start : c . clip_end + 1 ]
            c . seq . qual = c . seq . qual [ c . clip_start : c . clip_end + 1 ]
        else :
            print ( 'Warning: no clipping info for sequence' , c . id , file = sys . stderr )
    if len ( c . seq ) >= min_length :
        print ( c . seq , file = fout )
utils . close ( fout )


signature:
def count_sequences ( infile ) :
    '''Returns the number of sequences in a file'''

implementation:
seq_reader = sequences . file_reader ( infile )
n = 0
for seq in seq_reader :
    n += 1
return n


signature:
def interleave ( infile_1 , infile_2 , outfile , suffix1 = None , suffix2 = None ) :
    '''Makes interleaved file from two sequence files . If used , will append suffix1 onto end
 of every sequence name in infile _ 1 , unless it already ends with suffix1 . Similar for sufffix2.'''

implementation:
seq_reader_1 = sequences . file_reader ( infile_1 )
seq_reader_2 = sequences . file_reader ( infile_2 )
f_out = utils . open_file_write ( outfile )
for seq_1 in seq_reader_1 :
    try :
        seq_2 = next ( seq_reader_2 )
    except :
        utils . close ( f_out )
        raise Error ( 'Error getting mate for sequence' , seq_1 . id , ' ... cannot continue' )
    if suffix1 is not None and not seq_1 . id . endswith ( suffix1 ) :
        seq_1 . id += suffix1
    if suffix2 is not None and not seq_2 . id . endswith ( suffix2 ) :
        seq_2 . id += suffix2
    print ( seq_1 , file = f_out )
    print ( seq_2 , file = f_out )
try :
    seq_2 = next ( seq_reader_2 )
except :
    seq_2 = None
if seq_2 is not None :
    utils . close ( f_out )
    raise Error ( 'Error getting mate for sequence' , seq_2 . id , ' ... cannot continue' )
utils . close ( f_out )


signature:
def make_random_contigs ( contigs , length , outfile , name_by_letters = False , prefix = '' , seed = None , first_number = 1 ) :
    '''Makes a multi fasta file of random sequences , all the same length'''

implementation:
random . seed ( a = seed )
fout = utils . open_file_write ( outfile )
letters = list ( 'ABCDEFGHIJKLMNOPQRSTUVWXYZ' )
letters_index = 0
for i in range ( contigs ) :
    if name_by_letters :
        name = letters [ letters_index ]
        letters_index += 1
        if letters_index == len ( letters ) :
            letters_index = 0
    else :
        name = str ( i + first_number )
    fa = sequences . Fasta ( prefix + name , '' . join ( [ random . choice ( 'ACGT' ) for x in range ( length ) ] ) )
    print ( fa , file = fout )
utils . close ( fout )


signature:
def mean_length ( infile , limit = None ) :
    '''Returns the mean length of the sequences in the input file . By default uses all sequences . To limit to the first N sequences , use limit = N'''

implementation:
total = 0
count = 0
seq_reader = sequences . file_reader ( infile )
for seq in seq_reader :
    total += len ( seq )
    count += 1
    if limit is not None and count >= limit :
        break
assert count > 0
return total / count


signature:
def merge_to_one_seq ( infile , outfile , seqname = 'union' ) :
    '''Takes a multi fasta or fastq file and writes a new file that contains just one sequence , with the original sequences catted together , preserving their order'''

implementation:
seq_reader = sequences . file_reader ( infile )
seqs = [ ]
for seq in seq_reader :
    seqs . append ( copy . copy ( seq ) )
new_seq = '' . join ( [ seq . seq for seq in seqs ] )
if type ( seqs [ 0 ] ) == sequences . Fastq :
    new_qual = '' . join ( [ seq . qual for seq in seqs ] )
    seqs [ : ] = [ ]
    merged = sequences . Fastq ( seqname , new_seq , new_qual )
else :
    merged = sequences . Fasta ( seqname , new_seq )
    seqs [ : ] = [ ]
f = utils . open_file_write ( outfile )
print ( merged , file = f )
utils . close ( f )


signature:
def scaffolds_to_contigs ( infile , outfile , number_contigs = False ) :
    '''Makes a file of contigs from scaffolds by splitting at every N .
 Use number _ contigs = True to add . 1 , . 2 , etc onto end of each
 contig , instead of default to append coordinates .'''

implementation:
seq_reader = sequences . file_reader ( infile )
fout = utils . open_file_write ( outfile )
for seq in seq_reader :
    contigs = seq . contig_coords ( )
    counter = 1
    for contig in contigs :
        if number_contigs :
            name = seq . id + '.' + str ( counter )
            counter += 1
        else :
            name = '.' . join ( [ seq . id , str ( contig . start + 1 ) , str ( contig . end + 1 ) ] )
        print ( sequences . Fasta ( name , seq [ contig . start : contig . end + 1 ] ) , file = fout )
utils . close ( fout )


signature:
def sort_by_size ( infile , outfile , smallest_first = False ) :
    '''Sorts input sequence file by biggest sequence first , writes sorted output file . Set smallest _ first = True to have smallest first'''

implementation:
seqs = { }
file_to_dict ( infile , seqs )
seqs = list ( seqs . values ( ) )
seqs . sort ( key = lambda x : len ( x ) , reverse = not smallest_first )
fout = utils . open_file_write ( outfile )
for seq in seqs :
    print ( seq , file = fout )
utils . close ( fout )


signature:
def sort_by_name ( infile , outfile ) :
    '''Sorts input sequence file by sort - d - k1,1 , writes sorted output file .'''

implementation:
seqs = { }
file_to_dict ( infile , seqs )
# seqs = list ( seqs . values ( ) )
# seqs . sort ( )
fout = utils . open_file_write ( outfile )
for name in sorted ( seqs ) :
    print ( seqs [ name ] , file = fout )
utils . close ( fout )


signature:
def to_fastg ( infile , outfile , circular = None ) :
    '''Writes a FASTG file in SPAdes format from input file . Currently only whether or not a sequence is circular is supported . Put circular = set of ids , or circular = filename to make those sequences circular in the output . Puts coverage = 1 on all contigs'''

implementation:
if circular is None :
    to_circularise = set ( )
elif type ( circular ) is not set :
    f = utils . open_file_read ( circular )
    to_circularise = set ( [ x . rstrip ( ) for x in f . readlines ( ) ] )
    utils . close ( f )
else :
    to_circularise = circular
seq_reader = sequences . file_reader ( infile )
fout = utils . open_file_write ( outfile )
nodes = 1
for seq in seq_reader :
    new_id = '_' . join ( [ 'NODE' , str ( nodes ) , 'length' , str ( len ( seq ) ) , 'cov' , '1' , 'ID' , seq . id ] )
    if seq . id in to_circularise :
        seq . id = new_id + ':' + new_id + ';'
        print ( seq , file = fout )
        seq . revcomp ( )
        seq . id = new_id + "':" + new_id + "';"
        print ( seq , file = fout )
    else :
        seq . id = new_id + ';'
        print ( seq , file = fout )
        seq . revcomp ( )
        seq . id = new_id + "';"
        print ( seq , file = fout )
    nodes += 1
utils . close ( fout )


signature:
def length_offsets_from_fai ( fai_file ) :
    '''Returns a dictionary of positions of the start of each sequence , as
 if all the sequences were catted into one sequence .
 eg if file has three sequences , seq1 10bp , seq2 30bp , seq3 20bp , then
 the output would be : { ' seq1 ' : 0 , ' seq2 ' : 10 , ' seq3 ' : 40}'''

implementation:
positions = { }
total_length = 0
f = utils . open_file_read ( fai_file )
for line in f :
    try :
        ( name , length ) = line . rstrip ( ) . split ( ) [ : 2 ]
        length = int ( length )
    except :
        raise Error ( 'Error reading the following line of fai file ' + fai_file + '\n' + line )
    positions [ name ] = total_length
    total_length += length
utils . close ( f )
return positions


signature:
def split_by_base_count ( infile , outfiles_prefix , max_bases , max_seqs = None ) :
    '''Splits a fasta / q file into separate files , file size determined by number of bases .
 Puts < = max _ bases in each split file The exception is a single sequence > = max _ bases
 is put in its own file . This does not split sequences .'''

implementation:
seq_reader = sequences . file_reader ( infile )
base_count = 0
file_count = 1
seq_count = 0
fout = None
if max_seqs is None :
    max_seqs = float ( 'inf' )
for seq in seq_reader :
    if base_count == 0 :
        fout = utils . open_file_write ( outfiles_prefix + '.' + str ( file_count ) )
        file_count += 1
    if base_count + len ( seq ) > max_bases or seq_count >= max_seqs :
        if base_count == 0 :
            print ( seq , file = fout )
            utils . close ( fout )
        else :
            utils . close ( fout )
            fout = utils . open_file_write ( outfiles_prefix + '.' + str ( file_count ) )
            print ( seq , file = fout )
            base_count = len ( seq )
            file_count += 1
            seq_count = 1
    else :
        base_count += len ( seq )
        seq_count += 1
        print ( seq , file = fout )
utils . close ( fout )


signature:
def split_by_fixed_size ( infile , outfiles_prefix , chunk_size , tolerance , skip_if_all_Ns = False ) :
    '''Splits fasta / q file into separate files , with up to ( chunk _ size + tolerance ) bases in each file'''

implementation:
file_count = 1
coords = [ ]
small_sequences = [ ]
# sequences shorter than chunk _ size
seq_reader = sequences . file_reader ( infile )
f_coords = utils . open_file_write ( outfiles_prefix + '.coords' )
for seq in seq_reader :
    if skip_if_all_Ns and seq . is_all_Ns ( ) :
        continue
    if len ( seq ) < chunk_size :
        small_sequences . append ( copy . copy ( seq ) )
    elif len ( seq ) <= chunk_size + tolerance :
        f = utils . open_file_write ( outfiles_prefix + '.' + str ( file_count ) )
        print ( seq , file = f )
        utils . close ( f )
        file_count += 1
    else : # make list of chunk coords
        chunks = [ ( x , x + chunk_size ) for x in range ( 0 , len ( seq ) , chunk_size ) ]
        if chunks [ - 1 ] [ 1 ] - 1 > len ( seq ) :
            chunks [ - 1 ] = ( chunks [ - 1 ] [ 0 ] , len ( seq ) )
        if len ( chunks ) > 1 and ( chunks [ - 1 ] [ 1 ] - chunks [ - 1 ] [ 0 ] ) <= tolerance :
            chunks [ - 2 ] = ( chunks [ - 2 ] [ 0 ] , chunks [ - 1 ] [ 1 ] )
            chunks . pop ( )
        # write one output file per chunk
        offset = 0
        for chunk in chunks :
            if not ( skip_if_all_Ns and seq . is_all_Ns ( start = chunk [ 0 ] , end = chunk [ 1 ] - 1 ) ) :
                f = utils . open_file_write ( outfiles_prefix + '.' + str ( file_count ) )
                chunk_id = seq . id + ':' + str ( chunk [ 0 ] + 1 ) + '-' + str ( chunk [ 1 ] )
                print ( sequences . Fasta ( chunk_id , seq [ chunk [ 0 ] : chunk [ 1 ] ] ) , file = f )
                print ( chunk_id , seq . id , offset , sep = '\t' , file = f_coords )
                utils . close ( f )
                file_count += 1
            offset += chunk [ 1 ] - chunk [ 0 ]
# write files of small sequences
if len ( small_sequences ) :
    f = utils . open_file_write ( outfiles_prefix + '.' + str ( file_count ) )
    file_count += 1
    base_count = 0
    for seq in small_sequences :
        if base_count > 0 and base_count + len ( seq ) > chunk_size + tolerance :
            utils . close ( f )
            f = utils . open_file_write ( outfiles_prefix + '.' + str ( file_count ) )
            file_count += 1
            base_count = 0
        print ( seq , file = f )
        base_count += len ( seq )
    utils . close ( f )


signature:
def split_by_fixed_size_onefile ( infile , outfile , chunk_size , tolerance , skip_if_all_Ns = False ) :
    '''Splits each sequence in infile into chunks of fixed size , last chunk can be up to
 ( chunk _ size + tolerance ) in length'''

implementation:
seq_reader = sequences . file_reader ( infile )
f_out = utils . open_file_write ( outfile )
for seq in seq_reader :
    for i in range ( 0 , len ( seq ) , chunk_size ) :
        if i + chunk_size + tolerance >= len ( seq ) :
            end = len ( seq )
        else :
            end = i + chunk_size
        subseq = seq . subseq ( i , end )
        if not ( skip_if_all_Ns and subseq . is_all_Ns ( ) ) :
            subseq . id += '.' + str ( i + 1 ) + '_' + str ( end )
            print ( subseq , file = f_out )
        if end == len ( seq ) :
            break
utils . close ( f_out )


signature:
def stats_from_fai ( infile ) :
    '''Returns dictionary of length stats from an fai file . Keys are : longest , shortest , mean , total _ length , N50 , number'''

implementation:
f = utils . open_file_read ( infile )
try :
    lengths = sorted ( [ int ( line . split ( '\t' ) [ 1 ] ) for line in f ] , reverse = True )
except :
    raise Error ( 'Error getting lengths from fai file ' + infile )
utils . close ( f )
stats = { }
if len ( lengths ) > 0 :
    stats [ 'longest' ] = max ( lengths )
    stats [ 'shortest' ] = min ( lengths )
    stats [ 'total_length' ] = sum ( lengths )
    stats [ 'mean' ] = stats [ 'total_length' ] / len ( lengths )
    stats [ 'number' ] = len ( lengths )
    cumulative_length = 0
    for length in lengths :
        cumulative_length += length
        if cumulative_length >= 0.5 * stats [ 'total_length' ] :
            stats [ 'N50' ] = length
            break
else :
    stats = { x : 0 for x in ( 'longest' , 'shortest' , 'mean' , 'N50' , 'total_length' , 'number' ) }
return stats


signature:
def to_boulderio ( infile , outfile ) :
    '''Converts input sequence file into a " Boulder - IO format " , as used by primer3'''

implementation:
seq_reader = sequences . file_reader ( infile )
f_out = utils . open_file_write ( outfile )
for sequence in seq_reader :
    print ( "SEQUENCE_ID=" + sequence . id , file = f_out )
    print ( "SEQUENCE_TEMPLATE=" + sequence . seq , file = f_out )
    print ( "=" , file = f_out )
utils . close ( f_out )


signature:
def salted_hmac ( key_salt , value , secret = None ) :
    """Returns the HMAC - HASH of ' value ' , using a key generated from key _ salt and a
 secret ( which defaults to settings . SECRET _ KEY ) .
 A different key _ salt should be passed in for every application of HMAC .
 : type key _ salt : any
 : type value : any
 : type secret : any
 : rtype : HMAC"""

implementation:
if secret is None :
    secret = settings . SECRET_KEY
key_salt = force_bytes ( key_salt )
secret = force_bytes ( secret )
# We need to generate a derived key from our base key . We can do this by
# passing the key _ salt and our base key through a pseudo - random function and
# SHA1 works nicely .
digest = hashes . Hash ( settings . CRYPTOGRAPHY_DIGEST , backend = settings . CRYPTOGRAPHY_BACKEND )
digest . update ( key_salt + secret )
key = digest . finalize ( )
# If len ( key _ salt + secret ) > sha _ constructor ( ) . block _ size , the above
# line is redundant and could be replaced by key = key _ salt + secret , since
# the hmac module does the same thing for keys longer than the block size .
# However , we need to ensure that we * always * do this .
h = HMAC ( key , settings . CRYPTOGRAPHY_DIGEST , backend = settings . CRYPTOGRAPHY_BACKEND )
h . update ( force_bytes ( value ) )
return h


signature:
def pbkdf2 ( password , salt , iterations , dklen = 0 , digest = None ) :
    """Implements PBKDF2 with the same API as Django ' s existing
 implementation , using cryptography .
 : type password : any
 : type salt : any
 : type iterations : int
 : type dklen : int
 : type digest : cryptography . hazmat . primitives . hashes . HashAlgorithm"""

implementation:
if digest is None :
    digest = settings . CRYPTOGRAPHY_DIGEST
if not dklen :
    dklen = digest . digest_size
password = force_bytes ( password )
salt = force_bytes ( salt )
kdf = PBKDF2HMAC ( algorithm = digest , length = dklen , salt = salt , iterations = iterations , backend = settings . CRYPTOGRAPHY_BACKEND )
return kdf . derive ( password )


signature:
def encrypt ( self , data ) :
    """: type data : any
 : rtype : any"""

implementation:
data = force_bytes ( data )
iv = os . urandom ( 16 )
return self . _encrypt_from_parts ( data , iv )


signature:
def _encrypt_from_parts ( self , data , iv ) :
    """: type data : bytes
 : type iv : bytes
 : rtype : any"""

implementation:
padder = padding . PKCS7 ( algorithms . AES . block_size ) . padder ( )
padded_data = padder . update ( data ) + padder . finalize ( )
encryptor = Cipher ( algorithms . AES ( self . _encryption_key ) , modes . CBC ( iv ) , self . _backend ) . encryptor ( )
ciphertext = encryptor . update ( padded_data ) + encryptor . finalize ( )
return self . _signer . sign ( iv + ciphertext )


signature:
def decrypt ( self , data , ttl = None ) :
    """: type data : bytes
 : type ttl : int
 : rtype : bytes"""

implementation:
data = self . _signer . unsign ( data , ttl )
iv = data [ : 16 ]
ciphertext = data [ 16 : ]
decryptor = Cipher ( algorithms . AES ( self . _encryption_key ) , modes . CBC ( iv ) , self . _backend ) . decryptor ( )
plaintext_padded = decryptor . update ( ciphertext )
try :
    plaintext_padded += decryptor . finalize ( )
except ValueError :
    raise InvalidToken
# Remove padding
unpadder = padding . PKCS7 ( algorithms . AES . block_size ) . unpadder ( )
unpadded = unpadder . update ( plaintext_padded )
try :
    unpadded += unpadder . finalize ( )
except ValueError :
    raise InvalidToken
return unpadded


signature:
def get_encrypted_field ( base_class ) :
    """A get or create method for encrypted fields , we cache the field in
 the module to avoid recreation . This also allows us to always return
 the same class reference for a field .
 : type base _ class : models . Field [ T ]
 : rtype : models . Field [ EncryptedMixin , T ]"""

implementation:
assert not isinstance ( base_class , models . Field )
field_name = 'Encrypted' + base_class . __name__
if base_class not in FIELD_CACHE :
    FIELD_CACHE [ base_class ] = type ( field_name , ( EncryptedMixin , base_class ) , { 'base_class' : base_class , } )
return FIELD_CACHE [ base_class ]


signature:
def encrypt ( base_field , key = None , ttl = None ) :
    """A decorator for creating encrypted model fields .
 : type base _ field : models . Field [ T ]
 : param bytes key : This is an optional argument .
 Allows for specifying an instance specific encryption key .
 : param int ttl : This is an optional argument .
 The amount of time in seconds that a value can be stored for . If the
 time to live of the data has passed , it will become unreadable .
 The expired value will return an : class : ` Expired ` object .
 : rtype : models . Field [ EncryptedMixin , T ]"""

implementation:
if not isinstance ( base_field , models . Field ) :
    assert key is None
    assert ttl is None
    return get_encrypted_field ( base_field )
name , path , args , kwargs = base_field . deconstruct ( )
kwargs . update ( { 'key' : key , 'ttl' : ttl } )
return get_encrypted_field ( base_field . __class__ ) ( * args , ** kwargs )


signature:
def value_to_string ( self , obj ) :
    """Pickled data is serialized as base64"""

implementation:
value = self . value_from_object ( obj )
return b64encode ( self . _dump ( value ) ) . decode ( 'ascii' )


signature:
def dumps ( obj , key = None , salt = 'django.core.signing' , serializer = JSONSerializer , compress = False ) :
    """Returns URL - safe , sha1 signed base64 compressed JSON string . If key is
 None , settings . SECRET _ KEY is used instead .
 If compress is True ( not the default ) checks if compressing using zlib can
 save some space . Prepends a ' . ' to signify compression . This is included
 in the signature , to protect against zip bombs .
 Salt can be used to namespace the hash , so that a signed string is
 only valid for a given namespace . Leaving this at the default
 value or re - using a salt value across different parts of your
 application without good cause is a security risk .
 The serializer is expected to return a bytestring ."""

implementation:
data = serializer ( ) . dumps ( obj )
# Flag for if it ' s been compressed or not
is_compressed = False
if compress : # Avoid zlib dependency unless compress is being used
    compressed = zlib . compress ( data )
    if len ( compressed ) < ( len ( data ) - 1 ) :
        data = compressed
        is_compressed = True
base64d = b64_encode ( data )
if is_compressed :
    base64d = b'.' + base64d
return TimestampSigner ( key , salt = salt ) . sign ( base64d )


signature:
def signature ( self , value ) :
    """: type value : any
 : rtype : HMAC"""

implementation:
h = HMAC ( self . key , self . digest , backend = settings . CRYPTOGRAPHY_BACKEND )
h . update ( force_bytes ( value ) )
return h


signature:
def sign ( self , value ) :
    """: type value : any
 : rtype : bytes"""

implementation:
payload = struct . pack ( '>cQ' , self . version , int ( time . time ( ) ) )
payload += force_bytes ( value )
return payload + self . signature ( payload ) . finalize ( )


signature:
def unsign ( self , signed_value , ttl = None ) :
    """Retrieve original value and check it wasn ' t signed more
 than max _ age seconds ago .
 : type signed _ value : bytes
 : type ttl : int | datetime . timedelta"""

implementation:
h_size , d_size = struct . calcsize ( '>cQ' ) , self . digest . digest_size
fmt = '>cQ%ds%ds' % ( len ( signed_value ) - h_size - d_size , d_size )
try :
    version , timestamp , value , sig = struct . unpack ( fmt , signed_value )
except struct . error :
    raise BadSignature ( 'Signature is not valid' )
if version != self . version :
    raise BadSignature ( 'Signature version not supported' )
if ttl is not None :
    if isinstance ( ttl , datetime . timedelta ) :
        ttl = ttl . total_seconds ( )
    # Check timestamp is not older than ttl
    age = abs ( time . time ( ) - timestamp )
    if age > ttl + _MAX_CLOCK_SKEW :
        raise SignatureExpired ( 'Signature age %s > %s seconds' % ( age , ttl ) )
try :
    self . signature ( signed_value [ : - d_size ] ) . verify ( sig )
except InvalidSignature :
    raise BadSignature ( 'Signature "%s" does not match' % binascii . b2a_base64 ( sig ) )
return value


signature:
def get_version ( version = None ) :
    """Returns a PEP 386 - compliant version number from VERSION ."""

implementation:
version = get_complete_version ( version )
# Now build the two parts of the version number :
# main = X . Y [ . Z ]
# sub = . devN - for pre - alpha releases
# | { a | b | c } N - for alpha , beta and rc releases
main = get_main_version ( version )
sub = ''
if version [ 3 ] == 'alpha' and version [ 4 ] == 0 :
    git_changeset = get_git_changeset ( )
    if git_changeset :
        sub = '.dev%s' % git_changeset
elif version [ 3 ] != 'final' :
    mapping = { 'alpha' : 'a' , 'beta' : 'b' , 'rc' : 'c' }
    sub = mapping [ version [ 3 ] ] + str ( version [ 4 ] )
return str ( main + sub )


signature:
def get_complete_version ( version = None ) :
    """Returns a tuple of the django _ cryptography version . If version
 argument is non - empty , then checks for correctness of the tuple
 provided ."""

implementation:
if version is None :
    from django_cryptography import VERSION as version
else :
    assert len ( version ) == 5
    assert version [ 3 ] in ( 'alpha' , 'beta' , 'rc' , 'final' )
return version


signature:
def enumeration ( * args ) :
    """Return a value check function which raises a value error if the value is not
 in a pre - defined enumeration of values .
 If you pass in a list , tuple or set as the single argument , it is assumed
 that the list / tuple / set defines the membership of the enumeration .
 If you pass in more than on argument , it is assumed the arguments themselves
 define the enumeration ."""

implementation:
assert len ( args ) > 0 , 'at least one argument is required'
if len ( args ) == 1 : # assume the first argument defines the membership
    members = args [ 0 ]
else : # assume the arguments are the members
    members = args
def checker ( value ) :
    if value not in members :
        raise ValueError ( value )
return checker


signature:
def match_pattern ( regex ) :
    """Return a value check function which raises a ValueError if the value does
 not match the supplied regular expression , see also ` re . match ` ."""

implementation:
prog = re . compile ( regex )
def checker ( v ) :
    result = prog . match ( v )
    if result is None :
        raise ValueError ( v )
return checker


signature:
def search_pattern ( regex ) :
    """Return a value check function which raises a ValueError if the supplied
 regular expression does not match anywhere in the value , see also
 ` re . search ` ."""

implementation:
prog = re . compile ( regex )
def checker ( v ) :
    result = prog . search ( v )
    if result is None :
        raise ValueError ( v )
return checker


signature:
def number_range_inclusive ( min , max , type = float ) :
    """Return a value check function which raises a ValueError if the supplied
 value when cast as ` type ` is less than ` min ` or greater than ` max ` ."""

implementation:
def checker ( v ) :
    if type ( v ) < min or type ( v ) > max :
        raise ValueError ( v )
return checker


signature:
def number_range_exclusive ( min , max , type = float ) :
    """Return a value check function which raises a ValueError if the supplied
 value when cast as ` type ` is less than or equal to ` min ` or greater than
 or equal to ` max ` ."""

implementation:
def checker ( v ) :
    if type ( v ) <= min or type ( v ) >= max :
        raise ValueError ( v )
return checker


signature:
def datetime_range_inclusive ( min , max , format ) :
    """Return a value check function which raises a ValueError if the supplied
 value when converted to a datetime using the supplied ` format ` string is
 less than ` min ` or greater than ` max ` ."""

implementation:
dmin = datetime . strptime ( min , format )
dmax = datetime . strptime ( max , format )
def checker ( v ) :
    dv = datetime . strptime ( v , format )
    if dv < dmin or dv > dmax :
        raise ValueError ( v )
return checker


signature:
def write_problems ( problems , file , summarize = False , limit = 0 ) :
    """Write problems as restructured text to a file ( or stdout / stderr ) ."""

implementation:
w = file . write
# convenience variable
w ( """
=================
Validation Report
=================
""" )
counts = dict ( )
# store problem counts per problem code
total = 0
for i , p in enumerate ( problems ) :
    if limit and i >= limit :
        break
        # bail out
    if total == 0 and not summarize :
        w ( """
Problems
========
""" )
    total += 1
    code = p [ 'code' ]
    if code in counts :
        counts [ code ] += 1
    else :
        counts [ code ] = 1
    if not summarize :
        ptitle = '\n%s - %s\n' % ( p [ 'code' ] , p [ 'message' ] )
        w ( ptitle )
        underline = ''
        for i in range ( len ( ptitle . strip ( ) ) ) :
            underline += '-'
        underline += '\n'
        w ( underline )
        for k in sorted ( p . viewkeys ( ) - set ( [ 'code' , 'message' , 'context' ] ) ) :
            w ( ':%s: %s\n' % ( k , p [ k ] ) )
        if 'context' in p :
            c = p [ 'context' ]
            for k in sorted ( c . viewkeys ( ) ) :
                w ( ':%s: %s\n' % ( k , c [ k ] ) )
w ( """
Summary
=======

Found %s%s problem%s in total.

""" % ( 'at least ' if limit else '' , total , 's' if total != 1 else '' ) )
for code in sorted ( counts . viewkeys ( ) ) :
    w ( ':%s: %s\n' % ( code , counts [ code ] ) )
return total


signature:
def add_header_check ( self , code = HEADER_CHECK_FAILED , message = MESSAGES [ HEADER_CHECK_FAILED ] ) :
    """Add a header check , i . e . , check whether the header record is consistent
 with the expected field names .
 Arguments
 ` code ` - problem code to report if the header record is not valid ,
 defaults to ` HEADER _ CHECK _ FAILED `
 ` message ` - problem message to report if a value is not valid"""

implementation:
t = code , message
self . _header_checks . append ( t )


signature:
def add_record_length_check ( self , code = RECORD_LENGTH_CHECK_FAILED , message = MESSAGES [ RECORD_LENGTH_CHECK_FAILED ] , modulus = 1 ) :
    """Add a record length check , i . e . , check whether the length of a record is
 consistent with the number of expected fields .
 Arguments
 ` code ` - problem code to report if a record is not valid , defaults to
 ` RECORD _ LENGTH _ CHECK _ FAILED `
 ` message ` - problem message to report if a record is not valid
 ` modulus ` - apply the check to every nth record , defaults to 1 ( check
 every record )"""

implementation:
t = code , message , modulus
self . _record_length_checks . append ( t )


signature:
def add_value_check ( self , field_name , value_check , code = VALUE_CHECK_FAILED , message = MESSAGES [ VALUE_CHECK_FAILED ] , modulus = 1 ) :
    """Add a value check function for the specified field .
 Arguments
 ` field _ name ` - the name of the field to attach the value check function
 to
 ` value _ check ` - a function that accepts a single argument ( a value ) and
 raises a ` ValueError ` if the value is not valid
 ` code ` - problem code to report if a value is not valid , defaults to
 ` VALUE _ CHECK _ FAILED `
 ` message ` - problem message to report if a value is not valid
 ` modulus ` - apply the check to every nth record , defaults to 1 ( check
 every record )"""

implementation:
# guard conditions
assert field_name in self . _field_names , 'unexpected field name: %s' % field_name
assert callable ( value_check ) , 'value check must be a callable function'
t = field_name , value_check , code , message , modulus
self . _value_checks . append ( t )


signature:
def add_value_predicate ( self , field_name , value_predicate , code = VALUE_PREDICATE_FALSE , message = MESSAGES [ VALUE_PREDICATE_FALSE ] , modulus = 1 ) :
    """Add a value predicate function for the specified field .
 N . B . , everything you can do with value predicates can also be done with
 value check functions , whether you use one or the other is a matter of
 style .
 Arguments
 ` field _ name ` - the name of the field to attach the value predicate
 function to
 ` value _ predicate ` - a function that accepts a single argument ( a value )
 and returns False if the value is not valid
 ` code ` - problem code to report if a value is not valid , defaults to
 ` VALUE _ PREDICATE _ FALSE `
 ` message ` - problem message to report if a value is not valid
 ` modulus ` - apply the check to every nth record , defaults to 1 ( check
 every record )"""

implementation:
assert field_name in self . _field_names , 'unexpected field name: %s' % field_name
assert callable ( value_predicate ) , 'value predicate must be a callable function'
t = field_name , value_predicate , code , message , modulus
self . _value_predicates . append ( t )


signature:
def add_record_check ( self , record_check , modulus = 1 ) :
    """Add a record check function .
 Arguments
 ` record _ check ` - a function that accepts a single argument ( a record as
 a dictionary of values indexed by field name ) and raises a
 ` RecordError ` if the record is not valid
 ` modulus ` - apply the check to every nth record , defaults to 1 ( check
 every record )"""

implementation:
assert callable ( record_check ) , 'record check must be a callable function'
t = record_check , modulus
self . _record_checks . append ( t )


signature:
def add_record_predicate ( self , record_predicate , code = RECORD_PREDICATE_FALSE , message = MESSAGES [ RECORD_PREDICATE_FALSE ] , modulus = 1 ) :
    """Add a record predicate function .
 N . B . , everything you can do with record predicates can also be done with
 record check functions , whether you use one or the other is a matter of
 style .
 Arguments
 ` record _ predicate ` - a function that accepts a single argument ( a record
 as a dictionary of values indexed by field name ) and returns False if
 the value is not valid
 ` code ` - problem code to report if a record is not valid , defaults to
 ` RECORD _ PREDICATE _ FALSE `
 ` message ` - problem message to report if a record is not valid
 ` modulus ` - apply the check to every nth record , defaults to 1 ( check
 every record )"""

implementation:
assert callable ( record_predicate ) , 'record predicate must be a callable function'
t = record_predicate , code , message , modulus
self . _record_predicates . append ( t )


signature:
def add_unique_check ( self , key , code = UNIQUE_CHECK_FAILED , message = MESSAGES [ UNIQUE_CHECK_FAILED ] ) :
    """Add a unique check on a single column or combination of columns .
 Arguments
 ` key ` - a single field name ( string ) specifying a field in which all
 values are expected to be unique , or a sequence of field names ( tuple
 or list of strings ) specifying a compound key
 ` code ` - problem code to report if a record is not valid , defaults to
 ` UNIQUE _ CHECK _ FAILED `
 ` message ` - problem message to report if a record is not valid"""

implementation:
if isinstance ( key , basestring ) :
    assert key in self . _field_names , 'unexpected field name: %s' % key
else :
    for f in key :
        assert f in self . _field_names , 'unexpected field name: %s' % key
t = key , code , message
self . _unique_checks . append ( t )