signature:
def set_mongoadmin ( self ) :
    """Returns the MongoAdmin object for an app _ label / document _ name style view"""

implementation:
if hasattr ( self , "mongoadmin" ) :
    return None
if not hasattr ( self , "document_name" ) :
    self . set_mongonaut_base ( )
for mongoadmin in self . get_mongoadmins ( ) :
    for model in mongoadmin [ 'obj' ] . models :
        if model . name == self . document_name :
            self . mongoadmin = model . mongoadmin
            break
# TODO change this to use ' finally ' or ' else ' or something
if not hasattr ( self , "mongoadmin" ) :
    raise NoMongoAdminSpecified ( "No MongoAdmin for {0}.{1}" . format ( self . app_label , self . document_name ) )


signature:
def set_permissions_in_context ( self , context = { } ) :
    """Provides permissions for mongoadmin for use in the context"""

implementation:
context [ 'has_view_permission' ] = self . mongoadmin . has_view_permission ( self . request )
context [ 'has_edit_permission' ] = self . mongoadmin . has_edit_permission ( self . request )
context [ 'has_add_permission' ] = self . mongoadmin . has_add_permission ( self . request )
context [ 'has_delete_permission' ] = self . mongoadmin . has_delete_permission ( self . request )
return context


signature:
def process_post_form ( self , success_message = None ) :
    """As long as the form is set on the view this method will validate the form
 and save the submitted data . Only call this if you are posting data .
 The given success _ message will be used with the djanog messages framework
 if the posted data sucessfully submits ."""

implementation:
# When on initial args are given we need to set the base document .
if not hasattr ( self , 'document' ) or self . document is None :
    self . document = self . document_type ( )
self . form = MongoModelForm ( model = self . document_type , instance = self . document , form_post_data = self . request . POST ) . get_form ( )
self . form . is_bound = True
if self . form . is_valid ( ) :
    self . document_map_dict = MongoModelForm ( model = self . document_type ) . create_document_dictionary ( self . document_type )
    self . new_document = self . document_type
    # Used to keep track of embedded documents in lists . Keyed by the list and the number of the
    # document .
    self . embedded_list_docs = { }
    if self . new_document is None :
        messages . error ( self . request , u"Failed to save document" )
    else :
        self . new_document = self . new_document ( )
        for form_key in self . form . cleaned_data . keys ( ) :
            if form_key == 'id' and hasattr ( self , 'document' ) :
                self . new_document . id = self . document . id
                continue
            self . process_document ( self . new_document , form_key , None )
        self . new_document . save ( )
        if success_message :
            messages . success ( self . request , success_message )
return self . form


signature:
def process_document ( self , document , form_key , passed_key ) :
    """Given the form _ key will evaluate the document and set values correctly for
 the document given ."""

implementation:
if passed_key is not None :
    current_key , remaining_key_array = trim_field_key ( document , passed_key )
else :
    current_key , remaining_key_array = trim_field_key ( document , form_key )
key_array_digit = remaining_key_array [ - 1 ] if remaining_key_array and has_digit ( remaining_key_array ) else None
remaining_key = make_key ( remaining_key_array )
if current_key . lower ( ) == 'id' :
    raise KeyError ( u"Mongonaut does not work with models which have fields beginning with id_" )
# Create boolean checks to make processing document easier
is_embedded_doc = ( isinstance ( document . _fields . get ( current_key , None ) , EmbeddedDocumentField ) if hasattr ( document , '_fields' ) else False )
is_list = not key_array_digit is None
key_in_fields = current_key in document . _fields . keys ( ) if hasattr ( document , '_fields' ) else False
# This ensures you only go through each documents keys once , and do not duplicate data
if key_in_fields :
    if is_embedded_doc :
        self . set_embedded_doc ( document , form_key , current_key , remaining_key )
    elif is_list :
        self . set_list_field ( document , form_key , current_key , remaining_key , key_array_digit )
    else :
        value = translate_value ( document . _fields [ current_key ] , self . form . cleaned_data [ form_key ] )
        setattr ( document , current_key , value )


signature:
def set_embedded_doc ( self , document , form_key , current_key , remaining_key ) :
    """Get the existing embedded document if it exists , else created it ."""

implementation:
embedded_doc = getattr ( document , current_key , False )
if not embedded_doc :
    embedded_doc = document . _fields [ current_key ] . document_type_obj ( )
new_key , new_remaining_key_array = trim_field_key ( embedded_doc , remaining_key )
self . process_document ( embedded_doc , form_key , make_key ( new_key , new_remaining_key_array ) )
setattr ( document , current_key , embedded_doc )


signature:
def set_list_field ( self , document , form_key , current_key , remaining_key , key_array_digit ) :
    """1 . Figures out what value the list ought to have
 2 . Sets the list"""

implementation:
document_field = document . _fields . get ( current_key )
# Figure out what value the list ought to have
# None value for ListFields make mongoengine very un - happy
list_value = translate_value ( document_field . field , self . form . cleaned_data [ form_key ] )
if list_value is None or ( not list_value and not bool ( list_value ) ) :
    return None
current_list = getattr ( document , current_key , None )
if isinstance ( document_field . field , EmbeddedDocumentField ) :
    embedded_list_key = u"{0}_{1}" . format ( current_key , key_array_digit )
    # Get the embedded document if it exists , else create it .
    embedded_list_document = self . embedded_list_docs . get ( embedded_list_key , None )
    if embedded_list_document is None :
        embedded_list_document = document_field . field . document_type_obj ( )
    new_key , new_remaining_key_array = trim_field_key ( embedded_list_document , remaining_key )
    self . process_document ( embedded_list_document , form_key , new_key )
    list_value = embedded_list_document
    self . embedded_list_docs [ embedded_list_key ] = embedded_list_document
    if isinstance ( current_list , list ) : # Do not add the same document twice
        if embedded_list_document not in current_list :
            current_list . append ( embedded_list_document )
    else :
        setattr ( document , current_key , [ embedded_list_document ] )
elif isinstance ( current_list , list ) :
    current_list . append ( list_value )
else :
    setattr ( document , current_key , [ list_value ] )


signature:
def with_tz ( request ) :
    """Get the time with TZ enabled"""

implementation:
dt = datetime . now ( )
t = Template ( '{% load tz %}{% localtime on %}{% get_current_timezone as TIME_ZONE %}{{ TIME_ZONE }}{% endlocaltime %}' )
c = RequestContext ( request )
response = t . render ( c )
return HttpResponse ( response )


signature:
def without_tz ( request ) :
    """Get the time without TZ enabled"""

implementation:
t = Template ( '{% load tz %}{% get_current_timezone as TIME_ZONE %}{{ TIME_ZONE }}' )
c = RequestContext ( request )
response = t . render ( c )
return HttpResponse ( response )


signature:
def is_valid_ip ( ip_address ) :
    """Check Validity of an IP address"""

implementation:
try :
    ip = ipaddress . ip_address ( u'' + ip_address )
    return True
except ValueError as e :
    return False


signature:
def is_local_ip ( ip_address ) :
    """Check if IP is local"""

implementation:
try :
    ip = ipaddress . ip_address ( u'' + ip_address )
    return ip . is_loopback
except ValueError as e :
    return None


signature:
def process_request ( self , request ) :
    """If we can get a valid IP from the request ,
 look up that address in the database to get the appropriate timezone
 and activate it .
 Else , use the default ."""

implementation:
if not request :
    return
if not db_loaded :
    load_db ( )
tz = request . session . get ( 'django_timezone' )
if not tz : # use the default timezone ( settings . TIME _ ZONE ) for localhost
    tz = timezone . get_default_timezone ( )
    client_ip = get_ip_address_from_request ( request )
    ip_addrs = client_ip . split ( ',' )
    for ip in ip_addrs :
        if is_valid_ip ( ip ) and not is_local_ip ( ip ) :
            if ':' in ip :
                tz = db_v6 . time_zone_by_addr ( ip )
                break
            else :
                tz = db . time_zone_by_addr ( ip )
                break
if tz :
    timezone . activate ( tz )
    request . session [ 'django_timezone' ] = str ( tz )
    if getattr ( settings , 'AUTH_USER_MODEL' , None ) and getattr ( request , 'user' , None ) :
        detected_timezone . send ( sender = get_user_model ( ) , instance = request . user , timezone = tz )
else :
    timezone . deactivate ( )


signature:
def elastic_query ( model , query , session = None , enabled_fields = None ) :
    """Public method for init the class ElasticQuery
 : model : SQLAlchemy model
 : query : valid string like a ElasticSearch
 : session : SQLAlchemy session * optional
 : enabled _ fields : Fields allowed for make a query * optional"""

implementation:
# TODO : make session to optional
instance = ElasticQuery ( model , query , session , enabled_fields )
return instance . search ( )


signature:
def search ( self ) :
    """This is the most important method"""

implementation:
try :
    filters = json . loads ( self . query )
except ValueError :
    return False
result = self . model_query
if 'filter' in filters . keys ( ) :
    result = self . parse_filter ( filters [ 'filter' ] )
if 'sort' in filters . keys ( ) :
    result = result . order_by ( * self . sort ( filters [ 'sort' ] ) )
return result


signature:
def parse_filter ( self , filters ) :
    """This method process the filters"""

implementation:
for filter_type in filters :
    if filter_type == 'or' or filter_type == 'and' :
        conditions = [ ]
        for field in filters [ filter_type ] :
            if self . is_field_allowed ( field ) :
                conditions . append ( self . create_query ( self . parse_field ( field , filters [ filter_type ] [ field ] ) ) )
        if filter_type == 'or' :
            self . model_query = self . model_query . filter ( or_ ( * conditions ) )
        elif filter_type == 'and' :
            self . model_query = self . model_query . filter ( and_ ( * conditions ) )
    else :
        if self . is_field_allowed ( filter_type ) :
            conditions = self . create_query ( self . parse_field ( filter_type , filters [ filter_type ] ) )
            self . model_query = self . model_query . filter ( conditions )
return self . model_query


signature:
def parse_field ( self , field , field_value ) :
    """Parse the operators and traduce : ES to SQLAlchemy operators"""

implementation:
if type ( field_value ) is dict : # TODO : check operators and emit error
    operator = list ( field_value ) [ 0 ]
    if self . verify_operator ( operator ) is False :
        return "Error: operator does not exist" , operator
    value = field_value [ operator ]
elif type ( field_value ) is unicode :
    operator = u'equals'
    value = field_value
return field , operator , value


signature:
def create_query ( self , attr ) :
    """Mix all values and make the query"""

implementation:
field = attr [ 0 ]
operator = attr [ 1 ]
value = attr [ 2 ]
model = self . model
if '.' in field :
    field_items = field . split ( '.' )
    field_name = getattr ( model , field_items [ 0 ] , None )
    class_name = field_name . property . mapper . class_
    new_model = getattr ( class_name , field_items [ 1 ] )
    return field_name . has ( OPERATORS [ operator ] ( new_model , value ) )
return OPERATORS [ operator ] ( getattr ( model , field , None ) , value )


signature:
def sort ( self , sort_list ) :
    """Sort"""

implementation:
order = [ ]
for sort in sort_list :
    if sort_list [ sort ] == "asc" :
        order . append ( asc ( getattr ( self . model , sort , None ) ) )
    elif sort_list [ sort ] == "desc" :
        order . append ( desc ( getattr ( self . model , sort , None ) ) )
return order


signature:
def sendmail ( self , msg_from , msg_to , msg ) :
    """Remember the recipients ."""

implementation:
SMTP_dummy . msg_from = msg_from
SMTP_dummy . msg_to = msg_to
SMTP_dummy . msg = msg


signature:
def parsemail ( raw_message ) :
    """Parse message headers , then remove BCC header ."""

implementation:
message = email . parser . Parser ( ) . parsestr ( raw_message )
# Detect encoding
detected = chardet . detect ( bytearray ( raw_message , "utf-8" ) )
encoding = detected [ "encoding" ]
print ( ">>> encoding {}" . format ( encoding ) )
for part in message . walk ( ) :
    if part . get_content_maintype ( ) == 'multipart' :
        continue
    part . set_charset ( encoding )
# Extract recipients
addrs = email . utils . getaddresses ( message . get_all ( "TO" , [ ] ) ) + email . utils . getaddresses ( message . get_all ( "CC" , [ ] ) ) + email . utils . getaddresses ( message . get_all ( "BCC" , [ ] ) )
recipients = [ x [ 1 ] for x in addrs ]
message . __delitem__ ( "bcc" )
message . __setitem__ ( 'Date' , email . utils . formatdate ( ) )
sender = message [ "from" ]
return ( message , sender , recipients )


signature:
def _create_boundary ( message ) :
    """Add boundary parameter to multipart message if they are not present ."""

implementation:
if not message . is_multipart ( ) or message . get_boundary ( ) is not None :
    return message
# HACK : Python2 lists do not natively have a ` copy ` method . Unfortunately ,
# due to a bug in the Backport for the email module , the method
# ` Message . set _ boundary ` converts the Message headers into a native list ,
# so that other methods that rely on " copying " the Message headers fail .
# ` Message . set _ boundary ` is called from ` Generator . handle _ multipart ` if the
# message does not already have a boundary present . ( This method itself is
# called from ` Message . as _ string ` . )
# Hence , to prevent ` Message . set _ boundary ` from being called , add a
# boundary header manually .
from future . backports . email . generator import Generator
# pylint : disable = protected - access
boundary = Generator . _make_boundary ( message . policy . linesep )
message . set_param ( 'boundary' , boundary )
return message


signature:
def make_message_multipart ( message ) :
    """Convert a message into a multipart message ."""

implementation:
if not message . is_multipart ( ) :
    multipart_message = email . mime . multipart . MIMEMultipart ( 'alternative' )
    for header_key in set ( message . keys ( ) ) : # Preserve duplicate headers
        values = message . get_all ( header_key , failobj = [ ] )
        for value in values :
            multipart_message [ header_key ] = value
    original_text = message . get_payload ( )
    multipart_message . attach ( email . mime . text . MIMEText ( original_text ) )
    message = multipart_message
# HACK : For Python2 ( see comments in ` _ create _ boundary ` )
message = _create_boundary ( message )
return message


signature:
def convert_markdown ( message ) :
    """Convert markdown in message text to HTML ."""

implementation:
assert message [ 'Content-Type' ] . startswith ( "text/markdown" )
del message [ 'Content-Type' ]
# Convert the text from markdown and then make the message multipart
message = make_message_multipart ( message )
for payload_item in set ( message . get_payload ( ) ) : # Assume the plaintext item is formatted with markdown .
# Add corresponding HTML version of the item as the last part of
# the multipart message ( as per RFC 2046)
    if payload_item [ 'Content-Type' ] . startswith ( 'text/plain' ) :
        original_text = payload_item . get_payload ( )
        html_text = markdown . markdown ( original_text )
        html_payload = future . backports . email . mime . text . MIMEText ( "{}" . format ( html_text ) , "html" , )
        message . attach ( html_payload )
return message


signature:
def addattachments ( message , template_path ) :
    """Add the attachments from the message from the commandline options ."""

implementation:
if 'attachment' not in message :
    return message , 0
message = make_message_multipart ( message )
attachment_filepaths = message . get_all ( 'attachment' , failobj = [ ] )
template_parent_dir = os . path . dirname ( template_path )
for attachment_filepath in attachment_filepaths :
    attachment_filepath = os . path . expanduser ( attachment_filepath . strip ( ) )
    if not attachment_filepath :
        continue
    if not os . path . isabs ( attachment_filepath ) : # Relative paths are relative to the template ' s parent directory
        attachment_filepath = os . path . join ( template_parent_dir , attachment_filepath )
    normalized_path = os . path . abspath ( attachment_filepath )
    # Check that the attachment exists
    if not os . path . exists ( normalized_path ) :
        print ( "Error: can't find attachment " + normalized_path )
        sys . exit ( 1 )
    filename = os . path . basename ( normalized_path )
    with open ( normalized_path , "rb" ) as attachment :
        part = email . mime . application . MIMEApplication ( attachment . read ( ) , Name = filename )
    part . add_header ( 'Content-Disposition' , 'attachment; filename="{}"' . format ( filename ) )
    message . attach ( part )
    print ( ">>> attached {}" . format ( normalized_path ) )
del message [ 'attachment' ]
return message , len ( attachment_filepaths )


signature:
def sendmail ( message , sender , recipients , config_filename ) :
    """Send email message using Python SMTP library ."""

implementation:
# Read config file from disk to get SMTP server host , port , username
if not hasattr ( sendmail , "host" ) :
    config = configparser . RawConfigParser ( )
    config . read ( config_filename )
    sendmail . host = config . get ( "smtp_server" , "host" )
    sendmail . port = config . getint ( "smtp_server" , "port" )
    sendmail . username = config . get ( "smtp_server" , "username" )
    sendmail . security = config . get ( "smtp_server" , "security" )
    print ( ">>> Read SMTP server configuration from {}" . format ( config_filename ) )
    print ( ">>>   host = {}" . format ( sendmail . host ) )
    print ( ">>>   port = {}" . format ( sendmail . port ) )
    print ( ">>>   username = {}" . format ( sendmail . username ) )
    print ( ">>>   security = {}" . format ( sendmail . security ) )
# Prompt for password
if not hasattr ( sendmail , "password" ) :
    if sendmail . security == "Dummy" or sendmail . username == "None" :
        sendmail . password = None
    else :
        prompt = ">>> password for {} on {}: " . format ( sendmail . username , sendmail . host )
        sendmail . password = getpass . getpass ( prompt )
# Connect to SMTP server
if sendmail . security == "SSL/TLS" :
    smtp = smtplib . SMTP_SSL ( sendmail . host , sendmail . port )
elif sendmail . security == "STARTTLS" :
    smtp = smtplib . SMTP ( sendmail . host , sendmail . port )
    smtp . ehlo ( )
    smtp . starttls ( )
    smtp . ehlo ( )
elif sendmail . security == "Never" :
    smtp = smtplib . SMTP ( sendmail . host , sendmail . port )
elif sendmail . security == "Dummy" :
    smtp = smtp_dummy . SMTP_dummy ( )
else :
    raise configparser . Error ( "Unrecognized security type: {}" . format ( sendmail . security ) )
# Send credentials
if sendmail . username != "None" :
    smtp . login ( sendmail . username , sendmail . password )
# Send message . Note that we can ' t use the elegant
# " smtp . send _ message ( message ) " because that ' s python3 only
smtp . sendmail ( sender , recipients , message . as_string ( ) )
smtp . close ( )


signature:
def create_sample_input_files ( template_filename , database_filename , config_filename ) :
    """Create sample template email and database ."""

implementation:
print ( "Creating sample template email {}" . format ( template_filename ) )
if os . path . exists ( template_filename ) :
    print ( "Error: file exists: " + template_filename )
    sys . exit ( 1 )
with io . open ( template_filename , "w" ) as template_file :
    template_file . write ( u"TO: {{email}}\n" u"SUBJECT: Testing mailmerge\n" u"FROM: My Self <myself@mydomain.com>\n" u"\n" u"Hi, {{name}},\n" u"\n" u"Your number is {{number}}.\n" )
print ( "Creating sample database {}" . format ( database_filename ) )
if os . path . exists ( database_filename ) :
    print ( "Error: file exists: " + database_filename )
    sys . exit ( 1 )
with io . open ( database_filename , "w" ) as database_file :
    database_file . write ( u'email,name,number\n' u'myself@mydomain.com,"Myself",17\n' u'bob@bobdomain.com,"Bob",42\n' )
print ( "Creating sample config file {}" . format ( config_filename ) )
if os . path . exists ( config_filename ) :
    print ( "Error: file exists: " + config_filename )
    sys . exit ( 1 )
with io . open ( config_filename , "w" ) as config_file :
    config_file . write ( u"# Example: GMail\n" u"[smtp_server]\n" u"host = smtp.gmail.com\n" u"port = 465\n" u"security = SSL/TLS\n" u"username = YOUR_USERNAME_HERE\n" u"#\n" u"# Example: Wide open\n" u"# [smtp_server]\n" u"# host = open-smtp.example.com\n" u"# port = 25\n" u"# security = Never\n" u"# username = None\n" u"#\n" u"# Example: University of Michigan\n" u"# [smtp_server]\n" u"# host = smtp.mail.umich.edu\n" u"# port = 465\n" u"# security = SSL/TLS\n" u"# username = YOUR_USERNAME_HERE\n" u"#\n" u"# Example: University of Michigan EECS Dept., with STARTTLS security\n" # noqa : E501
    u"# [smtp_server]\n" u"# host = newman.eecs.umich.edu\n" u"# port = 25\n" u"# security = STARTTLS\n" u"# username = YOUR_USERNAME_HERE\n" u"#\n" u"# Example: University of Michigan EECS Dept., with no encryption\n" # noqa : E501
    u"# [smtp_server]\n" u"# host = newman.eecs.umich.edu\n" u"# port = 25\n" u"# security = Never\n" u"# username = YOUR_USERNAME_HERE\n" )
print ( "Edit these files, and then run mailmerge again" )


signature:
def main ( sample = False , dry_run = True , limit = 1 , no_limit = False , database_filename = DATABASE_FILENAME_DEFAULT , template_filename = TEMPLATE_FILENAME_DEFAULT , config_filename = CONFIG_FILENAME_DEFAULT ) :
    """Python API for mailmerge .
 mailmerge 0.1 by Andrew DeOrio < awdeorio @ umich . edu > .
 A simple , command line mail merge tool .
 Render an email template for each line in a CSV database ."""

implementation:
# pylint : disable = too - many - arguments , too - many - locals , too - many - branches
# pylint : disable = too - many - statements
# NOTE : this function needs a refactor , then remove ^ ^ ^
# Create a sample email template and database if there isn ' t one already
if sample :
    create_sample_input_files ( template_filename , database_filename , config_filename , )
    sys . exit ( 0 )
if not os . path . exists ( template_filename ) :
    print ( "Error: can't find template email " + template_filename )
    print ( "Create a sample (--sample) or specify a file (--template)" )
    sys . exit ( 1 )
if not os . path . exists ( database_filename ) :
    print ( "Error: can't find database_filename " + database_filename )
    print ( "Create a sample (--sample) or specify a file (--database)" )
    sys . exit ( 1 )
try : # Read template
    with io . open ( template_filename , "r" ) as template_file :
        content = template_file . read ( ) + u"\n"
        template = jinja2 . Template ( content )
    # Read CSV file database
    database = [ ]
    with io . open ( database_filename , "r" ) as database_file :
        reader = csv . DictReader ( database_file )
        for row in reader :
            database . append ( row )
    # Each row corresponds to one email message
    for i , row in enumerate ( database ) :
        if not no_limit and i >= limit :
            break
        # Fill in template fields using fields from row of CSV file
        raw_message = template . render ( ** row )
        # Parse message headers and detect encoding
        ( message , sender , recipients ) = parsemail ( raw_message )
        # Convert message from markdown to HTML if requested
        if message [ 'Content-Type' ] . startswith ( "text/markdown" ) :
            message = convert_markdown ( message )
        print ( ">>> message {}" . format ( i ) )
        print ( message . as_string ( ) )
        # Add attachments if any
        ( message , num_attachments ) = addattachments ( message , template_filename )
        # Send message
        if dry_run :
            print ( ">>> sent message {} DRY RUN" . format ( i ) )
        else : # Send message
            try :
                sendmail ( message , sender , recipients , config_filename )
            except smtplib . SMTPException as err :
                print ( ">>> failed to send message {}" . format ( i ) )
                timestamp = '{:%Y-%m-%d %H:%M:%S}' . format ( datetime . datetime . now ( ) )
                print ( timestamp , i , err , sep = ' ' , file = sys . stderr )
            else :
                print ( ">>> sent message {}" . format ( i ) )
    # Hints for user
    if num_attachments == 0 :
        print ( ">>> No attachments were sent with the emails." )
    if not no_limit :
        print ( ">>> Limit was {} messages.  " . format ( limit ) + "To remove the limit, use the --no-limit option." )
    if dry_run :
        print ( ( ">>> This was a dry run.  " "To send messages, use the --no-dry-run option." ) )
except jinja2 . exceptions . TemplateError as err :
    print ( ">>> Error in Jinja2 template: {}" . format ( err ) )
    sys . exit ( 1 )
except csv . Error as err :
    print ( ">>> Error reading CSV file: {}" . format ( err ) )
    sys . exit ( 1 )
except smtplib . SMTPAuthenticationError as err :
    print ( ">>> Authentication error: {}" . format ( err ) )
    sys . exit ( 1 )
except configparser . Error as err :
    print ( ">>> Error reading config file {}: {}" . format ( config_filename , err ) )
    sys . exit ( 1 )


signature:
def cli ( sample , dry_run , limit , no_limit , database_filename , template_filename , config_filename ) :
    """Command line interface ."""

implementation:
# pylint : disable = too - many - arguments
mailmerge . api . main ( sample = sample , dry_run = dry_run , limit = limit , no_limit = no_limit , database_filename = database_filename , template_filename = template_filename , config_filename = config_filename , )


signature:
def _tailCallback ( f , uid ) :
    """This is the " callable " version of the continuation , which sould only
 be accessible from the inside of the function to be continued . An
 attribute called " C " can be used in order to get back the public
 version of the continuation ( for passing the continuation to another
 function ) ."""

implementation:
def t ( * args ) :
    raise _TailCall ( f , args , uid )
t . C = f
return t


signature:
def with_continuations ( ** c ) :
    """A decorator for defining tail - call optimized functions .
 Example
 @ with _ continuations ( )
 def factorial ( n , k , self = None ) :
 return self ( n - 1 , k * n ) if n > 1 else k
 @ with _ continuations ( )
 def identity ( x , self = None ) :
 return x
 @ with _ continuations ( out = identity )
 def factorial2 ( n , k , self = None , out = None ) :
 return self ( n - 1 , k * n ) if n > 1 else out ( k )
 print ( factorial ( 7,1 ) )
 print ( factorial2(7,1 ) )"""

implementation:
if len ( c ) :
    keys , k = zip ( * c . items ( ) )
else :
    keys , k = tuple ( [ ] ) , tuple ( [ ] )
def d ( f ) :
    return C ( lambda kself , * conts : lambda * args : f ( * args , self = kself , ** dict ( zip ( keys , conts ) ) ) ) ( * k )
return d


signature:
def parse_int_list ( string ) :
    """Parses a string of numbers and ranges into a list of integers . Ranges
 are separated by dashes and inclusive of both the start and end number .
 Example :
 parse _ int _ list ( " 8 9 10,11-13 " ) = = [ 8,9,10,11,12,13]"""

implementation:
integers = [ ]
for comma_part in string . split ( "," ) :
    for substring in comma_part . split ( " " ) :
        if len ( substring ) == 0 :
            continue
        if "-" in substring :
            left , right = substring . split ( "-" )
            left_val = int ( left . strip ( ) )
            right_val = int ( right . strip ( ) )
            integers . extend ( range ( left_val , right_val + 1 ) )
        else :
            integers . append ( int ( substring . strip ( ) ) )
return integers


signature:
def sanitize_params ( method , ** kwargs ) :
    """Request params can be extracted from the ` ` * * kwargs ` `
 Arguments starting with ` _ ` will be stripped from it , so they
 can be used as an argument for the request
 ( eg . " _ headers " → " headers " in the kwargs returned by this
 function while " headers " would be inserted into the parameters
 of the request )
 Parameters
 method : str
 method to use to make the request
 kwargs : dict
 Keywords arguments given to the request
 Returns
 dict
 New requests parameters , correctly formatted"""

implementation:
# items which does not have a key starting with ` _ `
items = [ ( key , value ) for key , value in kwargs . items ( ) if not key . startswith ( "_" ) ]
params , skip_params = { } , False
for key , value in items : # binary data
    if hasattr ( value , 'read' ) or isinstance ( value , bytes ) :
        params [ key ] = value
        # The params won ' t be used to make the signature
        skip_params = True
    # booleans conversion
    elif isinstance ( value , bool ) :
        params [ key ] = "true" if value else "false"
    # iterables conversion
    elif isinstance ( value , iterable ) :
        params [ key ] = "," . join ( map ( str , value ) )
    # skip params with value None
    elif value is None :
        pass
    # the rest is converted to str
    # ( make sure you don ' t send something wrong )
    else :
        params [ key ] = str ( value )
# dict with other items ( + strip " _ " from keys )
kwargs = { key [ 1 : ] : value for key , value in kwargs . items ( ) if key . startswith ( "_" ) }
if method == "post" and not kwargs . get ( 'data' , None ) and params :
    kwargs [ 'data' ] = params
# post requests use the data argument
elif not kwargs . get ( 'params' , None ) and params :
    kwargs [ 'params' ] = params
return kwargs , skip_params


signature:
def _get_base_url ( base_url , api , version ) :
    """create the base url for the api
 Parameters
 base _ url : str
 format of the base _ url using { api } and { version }
 api : str
 name of the api to use
 version : str
 version of the api
 Returns
 str
 the base url of the api you want to use"""

implementation:
format_args = { }
if "{api}" in base_url :
    if api == "" :
        base_url = base_url . replace ( '{api}.' , '' )
    else :
        format_args [ 'api' ] = api
if "{version}" in base_url :
    if version == "" :
        base_url = base_url . replace ( '/{version}' , '' )
    else :
        format_args [ 'version' ] = version
return base_url . format ( api = api , version = version )


signature:
async def request ( self , method , url , future , headers = None , session = None , encoding = None , ** kwargs ) :
    """Make requests to the REST API
 Parameters
 future : asyncio . Future
 Future used to return the response
 method : str
 Method to be used by the request
 url : str
 URL of the resource
 headers : . oauth . PeonyHeaders
 Custom headers ( doesn ' t overwrite ` Authorization ` headers )
 session : aiohttp . ClientSession , optional
 Client session used to make the request
 Returns
 data . PeonyResponse
 Response to the request"""

implementation:
await self . setup
# prepare request arguments , particularly the headers
req_kwargs = await self . headers . prepare_request ( method = method , url = url , headers = headers , proxy = self . proxy , ** kwargs )
if encoding is None :
    encoding = self . encoding
session = session if ( session is not None ) else self . _session
logger . debug ( "making request with parameters: %s" % req_kwargs )
async with session . request ( ** req_kwargs ) as response :
    if response . status < 400 :
        data = await data_processing . read ( response , self . _loads , encoding = encoding )
        future . set_result ( data_processing . PeonyResponse ( data = data , headers = response . headers , url = response . url , request = req_kwargs ) )
    else : # throw exception if status is not 2xx
        await exceptions . throw ( response , loads = self . _loads , encoding = encoding , url = url )


signature:
def stream_request ( self , method , url , headers = None , _session = None , * args , ** kwargs ) :
    """Make requests to the Streaming API
 Parameters
 method : str
 Method to be used by the request
 url : str
 URL of the resource
 headers : dict
 Custom headers ( doesn ' t overwrite ` Authorization ` headers )
 _ session : aiohttp . ClientSession , optional
 The session to use for this specific request , the session
 given as argument of : meth : ` _ _ init _ _ ` is used by default
 Returns
 . stream . StreamResponse
 Stream context for the request"""

implementation: return StreamResponse ( method = method , url = url , client = self , headers = headers , session = _session , proxy = self . proxy , ** kwargs )


signature:
def get_tasks ( self ) :
    """Get the tasks attached to the instance
 Returns
 list
 List of tasks ( : class : ` asyncio . Task ` )"""

implementation:
tasks = self . _get_tasks ( )
tasks . extend ( self . _streams . get_tasks ( self ) )
return tasks


signature:
async def run_tasks ( self ) :
    """Run the tasks attached to the instance"""

implementation:
tasks = self . get_tasks ( )
self . _gathered_tasks = asyncio . gather ( * tasks , loop = self . loop )
try :
    await self . _gathered_tasks
except CancelledError :
    pass


signature:
async def close ( self ) :
    """properly close the client"""

implementation:
tasks = self . _get_close_tasks ( )
if tasks :
    await asyncio . wait ( tasks )
self . _session = None


signature:
async def _get_twitter_configuration ( self ) :
    """create a ` ` twitter _ configuration ` ` attribute with the response
 of the endpoint
 https : / / api . twitter . com / 1.1 / help / configuration . json"""

implementation:
api = self [ 'api' , general . twitter_api_version , ".json" , general . twitter_base_api_url ]
return await api . help . configuration . get ( )


signature:
async def _get_user ( self ) :
    """create a ` ` user ` ` attribute with the response of the endpoint
 https : / / api . twitter . com / 1.1 / account / verify _ credentials . json"""

implementation:
api = self [ 'api' , general . twitter_api_version , ".json" , general . twitter_base_api_url ]
return await api . account . verify_credentials . get ( )


signature:
async def _chunked_upload ( self , media , media_size , path = None , media_type = None , media_category = None , chunk_size = 2 ** 20 , ** params ) :
    """upload media in chunks
 Parameters
 media : file object
 a file object of the media
 media _ size : int
 size of the media
 path : str , optional
 filename of the media
 media _ type : str , optional
 mime type of the media
 media _ category : str , optional
 twitter media category , must be used with ` ` media _ type ` `
 chunk _ size : int , optional
 size of a chunk in bytes
 params : dict , optional
 additional parameters of the request
 Returns
 . data _ processing . PeonyResponse
 Response of the request"""

implementation:
if isinstance ( media , bytes ) :
    media = io . BytesIO ( media )
chunk = media . read ( chunk_size )
is_coro = asyncio . iscoroutine ( chunk )
if is_coro :
    chunk = await chunk
if media_type is None :
    media_metadata = await utils . get_media_metadata ( chunk , path )
    media_type , media_category = media_metadata
elif media_category is None :
    media_category = utils . get_category ( media_type )
response = await self . upload . media . upload . post ( command = "INIT" , total_bytes = media_size , media_type = media_type , media_category = media_category , ** params )
media_id = response [ 'media_id' ]
i = 0
while chunk :
    if is_coro :
        req = self . upload . media . upload . post ( command = "APPEND" , media_id = media_id , media = chunk , segment_index = i )
        chunk , _ = await asyncio . gather ( media . read ( chunk_size ) , req )
    else :
        await self . upload . media . upload . post ( command = "APPEND" , media_id = media_id , media = chunk , segment_index = i )
        chunk = media . read ( chunk_size )
    i += 1
status = await self . upload . media . upload . post ( command = "FINALIZE" , media_id = media_id )
if 'processing_info' in status :
    while status [ 'processing_info' ] . get ( 'state' ) != "succeeded" :
        processing_info = status [ 'processing_info' ]
        if processing_info . get ( 'state' ) == "failed" :
            error = processing_info . get ( 'error' , { } )
            message = error . get ( 'message' , str ( status ) )
            raise exceptions . MediaProcessingError ( data = status , message = message , ** params )
        delay = processing_info [ 'check_after_secs' ]
        await asyncio . sleep ( delay )
        status = await self . upload . media . upload . get ( command = "STATUS" , media_id = media_id , ** params )
return response


signature:
async def upload_media ( self , file_ , media_type = None , media_category = None , chunked = None , size_limit = None , ** params ) :
    """upload a media on twitter
 Parameters
 file _ : str or pathlib . Path or file
 Path to the file or file object
 media _ type : str , optional
 mime type of the media
 media _ category : str , optional
 Twitter ' s media category of the media , must be used with
 ` ` media _ type ` `
 chunked : bool , optional
 If True , force the use of the chunked upload for the media
 size _ limit : int , optional
 If set , the media will be sent using a multipart upload if
 its size is over ` ` size _ limit ` ` bytes
 params : dict
 parameters used when making the request
 Returns
 . data _ processing . PeonyResponse
 Response of the request"""

implementation:
if isinstance ( file_ , str ) :
    url = urlparse ( file_ )
    if url . scheme . startswith ( 'http' ) :
        media = await self . _session . get ( file_ )
    else :
        path = urlparse ( file_ ) . path . strip ( " \"'" )
        media = await utils . execute ( open ( path , 'rb' ) )
elif hasattr ( file_ , 'read' ) or isinstance ( file_ , bytes ) :
    media = file_
else :
    raise TypeError ( "upload_media input must be a file object or a " "filename or binary data or an aiohttp request" )
media_size = await utils . get_size ( media )
if chunked is not None :
    size_test = False
else :
    size_test = await self . _size_test ( media_size , size_limit )
if isinstance ( media , aiohttp . ClientResponse ) : # send the content of the response
    media = media . content
if chunked or ( size_test and chunked is None ) :
    args = media , media_size , file_ , media_type , media_category
    response = await self . _chunked_upload ( * args , ** params )
else :
    response = await self . upload . media . upload . post ( media = media , ** params )
if not hasattr ( file_ , 'read' ) and not getattr ( media , 'closed' , True ) :
    media . close ( )
return response


signature:
def split_stdout_lines ( stdout ) :
    """Given the standard output from NetMHC / NetMHCpan / NetMHCcons tools ,
 drop all { comments , lines of hyphens , empty lines } and split the
 remaining lines by whitespace ."""

implementation:
# all the NetMHC formats use lines full of dashes before any actual
# binding results
seen_dash = False
for l in stdout . split ( "\n" ) :
    l = l . strip ( )
    # wait for a line like ' - - - - - ' before trying to parse entries
    # have to include multiple dashes here since NetMHC 4.0 sometimes
    # gives negative positions in its " peptide " input mode
    if l . startswith ( "---" ) :
        seen_dash = True
        continue
    if not seen_dash :
        continue
    # ignore empty lines and comments
    if not l or l . startswith ( "#" ) :
        continue
    # beginning of headers in NetMHC
    if any ( l . startswith ( word ) for word in NETMHC_TOKENS ) :
        continue
    yield l . split ( )


signature:
def clean_fields ( fields , ignored_value_indices , transforms ) :
    """Sometimes , NetMHC * has fields that are only populated sometimes , which results
 in different count / indexing of the fields when that happens .
 We handle this by looking for particular strings at particular indices , and
 deleting them .
 Warning : this may result in unexpected behavior sometimes . For example , we
 ignore " SB " and " WB " for NetMHC 3 . x output ; which also means that any line
 with a key called SB or WB will be ignored .
 Also , sometimes NetMHC * will have fields that we want to modify in some
 consistent way , e . g . NetMHCpan3 has 1 - based offsets and all other predictors
 have 0 - based offsets ( and we rely on 0 - based offsets ) . We handle this using
 a map from field index to transform function ."""

implementation:
cleaned_fields = [ ]
for i , field in enumerate ( fields ) :
    if field in ignored_value_indices :
        ignored_index = ignored_value_indices [ field ]
        # Is the value we want to ignore at the index where we ' d ignore it ?
        if ignored_index == i :
            continue
    # transform this field if the index is in transforms , otherwise leave alone
    cleaned_field = transforms [ i ] ( field ) if i in transforms else field
    cleaned_fields . append ( cleaned_field )
return cleaned_fields


signature:
def parse_stdout ( stdout , prediction_method_name , sequence_key_mapping , key_index , offset_index , peptide_index , allele_index , ic50_index , rank_index , log_ic50_index , ignored_value_indices = { } , transforms = { } ) :
    """Generic function for parsing any NetMHC * output , given expected indices
 of values of interest .
 Parameters
 ignored _ value _ indices : dict
 Map from values to the positions we ' ll ignore them at . See clean _ fields .
 transforms : dict
 Map from field index to a transform function to be applied to values in
 that field . See clean _ fields .
 Returns BindingPredictionCollection"""

implementation:
binding_predictions = [ ]
for fields in split_stdout_lines ( stdout ) :
    fields = clean_fields ( fields , ignored_value_indices , transforms )
    offset = int ( fields [ offset_index ] )
    peptide = str ( fields [ peptide_index ] )
    allele = str ( fields [ allele_index ] )
    ic50 = float ( fields [ ic50_index ] )
    rank = float ( fields [ rank_index ] ) if rank_index else 0.0
    log_ic50 = float ( fields [ log_ic50_index ] )
    key = str ( fields [ key_index ] )
    if sequence_key_mapping :
        original_key = sequence_key_mapping [ key ]
    else : # if sequence _ key _ mapping isn ' t provided then let ' s assume it ' s the
    # identity function
        original_key = key
    binding_predictions . append ( BindingPrediction ( source_sequence_name = original_key , offset = offset , peptide = peptide , allele = normalize_allele_name ( allele ) , affinity = ic50 , percentile_rank = rank , log_affinity = log_ic50 , prediction_method_name = prediction_method_name ) )
return binding_predictions


signature:
def parse_netmhc3_stdout ( stdout , prediction_method_name = "netmhc3" , sequence_key_mapping = None ) :
    """Parse the output format for NetMHC 3 . x , which looks like :
 pos peptide logscore affinity ( nM ) Bind Level Protein Name Allele
 0 SIINKFELL 0.437 441 WB A1 HLA - A02:01
 0 SIINKFFFQ 0.206 5411 A2 HLA - A02:01
 1 IINKFFFQQ 0.128 12544 A2 HLA - A02:01
 2 INKFFFQQQ 0.046 30406 A2 HLA - A02:01
 3 NKFFFQQQQ 0.050 29197 A2 HLA - A02:01"""

implementation: return parse_stdout ( stdout = stdout , prediction_method_name = prediction_method_name , sequence_key_mapping = sequence_key_mapping , key_index = 4 , offset_index = 0 , peptide_index = 1 , allele_index = 5 , ic50_index = 3 , rank_index = None , log_ic50_index = 2 , ignored_value_indices = { "WB" : 4 , "SB" : 4 } )


signature:
def parse_netmhc4_stdout ( stdout , prediction_method_name = "netmhc4" , sequence_key_mapping = None ) :
    """# Peptide length 9
 # Rank Threshold for Strong binding peptides 0.500
 # Rank Threshold for Weak binding peptides 2.000
 pos HLA peptide Core Offset I _ pos I _ len D _ pos D _ len iCore Identity 1 - log50k ( aff ) Affinity ( nM ) % Rank BindLevel
 0 HLA - A0201 TMDKSELVQ TMDKSELVQ 0 0 0 0 0 TMDKSELVQ 143B _ BOVIN _ P293 0.051 28676.59 43.00
 1 HLA - A0201 MDKSELVQK MDKSELVQK 0 0 0 0 0 MDKSELVQK 143B _ BOVIN _ P293 0.030 36155.15 70.00
 2 HLA - A0201 DKSELVQKA DKSELVQKA 0 0 0 0 0 DKSELVQKA 143B _ BOVIN _ P293 0.030 36188.42 70.00
 3 HLA - A0201 KSELVQKAK KSELVQKAK 0 0 0 0 0 KSELVQKAK 143B _ BOVIN _ P293 0.032 35203.22 65.00
 4 HLA - A0201 SELVQKAKL SELVQKAKL 0 0 0 0 0 SELVQKAKL 143B _ BOVIN _ P293 0.031 35670.99 65.00
 5 HLA - A0201 ELVQKAKLA ELVQKAKLA 0 0 0 0 0 ELVQKAKLA 143B _ BOVIN _ P293 0.080 21113.07 29.00
 6 HLA - A0201 LVQKAKLAE LVQKAKLAE 0 0 0 0 0 LVQKAKLAE 143B _ BOVIN _ P293 0.027 37257.56 75.00
 7 HLA - A0201 VQKAKLAEQ VQKAKLAEQ 0 0 0 0 0 VQKAKLAEQ 143B _ BOVIN _ P293 0.040 32404.62 55.00
 219 HLA - A0201 QLLRDNLTL QLLRDNLTL 0 0 0 0 0 QLLRDNLTL 143B _ BOVIN _ P293 0.527 167.10 1.50 < = WB"""

implementation: return parse_stdout ( stdout = stdout , prediction_method_name = prediction_method_name , sequence_key_mapping = sequence_key_mapping , key_index = 10 , offset_index = 0 , peptide_index = 2 , allele_index = 1 , ic50_index = 12 , rank_index = 13 , log_ic50_index = 11 )


signature:
def parse_netmhcpan28_stdout ( stdout , prediction_method_name = "netmhcpan" , sequence_key_mapping = None ) :
    """# Affinity Threshold for Strong binding peptides 50.000 ' ,
 # Affinity Threshold for Weak binding peptides 500.000 ' ,
 # Rank Threshold for Strong binding peptides 0.500 ' ,
 # Rank Threshold for Weak binding peptides 2.000 ' ,
 pos HLA peptide Identity 1 - log50k ( aff ) Affinity ( nM ) % Rank BindLevel
 0 HLA - A * 02:03 QQQQQYFPE id0 0.024 38534.25 50.00
 1 HLA - A * 02:03 QQQQYFPEI id0 0.278 2461.53 15.00
 2 HLA - A * 02:03 QQQYFPEIT id0 0.078 21511.53 50.00
 3 HLA - A * 02:03 QQYFPEITH id0 0.041 32176.84 50.00
 4 HLA - A * 02:03 QYFPEITHI id0 0.085 19847.09 32.00
 5 HLA - A * 02:03 YFPEITHII id0 0.231 4123.85 15.00
 6 HLA - A * 02:03 FPEITHIII id0 0.060 26134.28 50.00
 7 HLA - A * 02:03 PEITHIIIA id0 0.034 34524.63 50.00
 8 HLA - A * 02:03 EITHIIIAS id0 0.076 21974.48 50.00
 9 HLA - A * 02:03 ITHIIIASS id0 0.170 7934.26 32.00
 10 HLA - A * 02:03 THIIIASSS id0 0.040 32361.18 50.00
 11 HLA - A * 02:03 HIIIASSSL id0 0.515 189.74 4.00 < = WB"""

implementation:
check_stdout_error ( stdout , "NetMHCpan-2.8" )
return parse_stdout ( stdout = stdout , prediction_method_name = prediction_method_name , sequence_key_mapping = sequence_key_mapping , key_index = 3 , offset_index = 0 , peptide_index = 2 , allele_index = 1 , ic50_index = 5 , rank_index = 6 , log_ic50_index = 4 )


signature:
def parse_netmhcpan3_stdout ( stdout , prediction_method_name = "netmhcpan" , sequence_key_mapping = None ) :
    """# Rank Threshold for Strong binding peptides 0.500
 # Rank Threshold for Weak binding peptides 2.000
 Pos HLA Peptide Core Of Gp Gl Ip Il Icore Identity Score Aff ( nM ) % Rank BindLevel
 1 HLA - B * 18:01 MFCQLAKT MFCQLAKT - 0 0 0 8 1 MFCQLAKT sequence0_0 0.02864 36676.0 45.00
 2 HLA - B * 18:01 FCQLAKTY F - CQLAKTY 0 0 0 1 1 FCQLAKTY sequence0_0 0.07993 21056.5 13.00"""

implementation:
# the offset specified in " pos " ( at index 0 ) is 1 - based instead of 0 - based . we adjust it to be
# 0 - based , as in all the other netmhc predictors supported by this library .
transforms = { 0 : lambda x : int ( x ) - 1 , }
return parse_stdout ( stdout = stdout , prediction_method_name = prediction_method_name , sequence_key_mapping = sequence_key_mapping , key_index = 10 , offset_index = 0 , peptide_index = 2 , allele_index = 1 , ic50_index = 12 , rank_index = 13 , log_ic50_index = 11 , transforms = transforms )


signature:
def parse_netmhcpan4_stdout ( stdout , prediction_method_name = "netmhcpan" , sequence_key_mapping = None ) :
    """# NetMHCpan version 4.0
 # Tmpdir made / var / folders / jc / fyrvcrcs3sb8g4mkdg6nl _ t80000gp / T / / netMHCpanuH3SvY
 # Input is in PEPTIDE format
 # Make binding affinity predictions
 HLA - A02:01 : Distance to training data 0.000 ( using nearest neighbor HLA - A02:01)
 # Rank Threshold for Strong binding peptides 0.500
 # Rank Threshold for Weak binding peptides 2.000
 Pos HLA Peptide Core Of Gp Gl Ip Il Icore Identity Score Aff ( nM ) % Rank BindLevel
 1 HLA - A * 02:01 SIINFEKL SIINF - EKL 0 0 0 5 1 SIINFEKL PEPLIST 0.1141340 14543.1 18.9860
 Protein PEPLIST . Allele HLA - A * 02:01 . Number of high binders 0 . Number of weak binders 0 . Number of peptides 1"""

implementation:
# Output format is compatible with netmhcpan3 , but netmhcpan 4.0 must be
# called with the - BA flag , so it gives affinity predictions , not mass - spec
# elution likelihoods .
return parse_netmhcpan3_stdout ( stdout = stdout , prediction_method_name = prediction_method_name , sequence_key_mapping = sequence_key_mapping )


signature:
def _parse_iedb_response ( response ) :
    """Take the binding predictions returned by IEDB ' s web API
 and parse them into a DataFrame
 Expect response to look like :
 allele seq _ num start end length peptide ic50 percentile _ rank
 HLA - A * 01:01 1 2 10 9 LYNTVATLY 2145.70 3.7
 HLA - A * 01:01 1 5 13 9 TVATLYCVH 2216.49 3.9
 HLA - A * 01:01 1 7 15 9 ATLYCVHQR 2635.42 5.1
 HLA - A * 01:01 1 4 12 9 NTVATLYCV 6829.04 20
 HLA - A * 01:01 1 1 9 9 SLYNTVATL 8032.38 24
 HLA - A * 01:01 1 8 16 9 TLYCVHQRI 8853.90 26
 HLA - A * 01:01 1 3 11 9 YNTVATLYC 9865.62 29
 HLA - A * 01:01 1 6 14 9 VATLYCVHQ 27575.71 58
 HLA - A * 01:01 1 10 18 9 YCVHQRIDV 48929.64 74
 HLA - A * 01:01 1 9 17 9 LYCVHQRID 50000.00 75"""

implementation:
if len ( response ) == 0 :
    raise ValueError ( "Empty response from IEDB!" )
df = pd . read_csv ( io . BytesIO ( response ) , delim_whitespace = True , header = 0 )
# pylint doesn ' t realize that df is a DataFrame , so tell is
assert type ( df ) == pd . DataFrame
df = pd . DataFrame ( df )
if len ( df ) == 0 :
    raise ValueError ( "No binding predictions in response from IEDB: %s" % ( response , ) )
required_columns = [ "allele" , "peptide" , "ic50" , "start" , "end" , ]
for column in required_columns :
    if column not in df . columns :
        raise ValueError ( "Response from IEDB is missing '%s' column: %s. Full " "response:\n%s" % ( column , df . ix [ 0 ] , response ) )
# since IEDB has allowed multiple column names for percentile rank ,
# we ' re defensively normalizing all of them to just ' rank '
df = df . rename ( columns = { "percentile_rank" : "rank" , "percentile rank" : "rank" } )
return df


signature:
def _query_iedb ( request_values , url ) :
    """Call into IEDB ' s web API for MHC binding prediction using request dictionary
 with fields :
 - " method "
 - " length "
 - " sequence _ text "
 - " allele "
 Parse the response into a DataFrame ."""

implementation:
data = urlencode ( request_values )
req = Request ( url , data . encode ( "ascii" ) )
response = urlopen ( req ) . read ( )
return _parse_iedb_response ( response )


signature:
def predict_subsequences ( self , sequence_dict , peptide_lengths = None ) :
    """Given a dictionary mapping unique keys to amino acid sequences ,
 run MHC binding predictions on all candidate epitopes extracted from
 sequences and return a EpitopeCollection .
 Parameters
 fasta _ dictionary : dict or string
 Mapping of protein identifiers to protein amino acid sequences .
 If string then converted to dictionary ."""

implementation:
sequence_dict = check_sequence_dictionary ( sequence_dict )
peptide_lengths = self . _check_peptide_lengths ( peptide_lengths )
# take each mutated sequence in the dataframe
# and general MHC binding scores for all k - mer substrings
binding_predictions = [ ]
expected_peptides = set ( [ ] )
normalized_alleles = [ ]
for key , amino_acid_sequence in sequence_dict . items ( ) :
    for l in peptide_lengths :
        for i in range ( len ( amino_acid_sequence ) - l + 1 ) :
            expected_peptides . add ( amino_acid_sequence [ i : i + l ] )
    self . _check_peptide_inputs ( expected_peptides )
    for allele in self . alleles : # IEDB MHCII predictor expects DRA1 to be omitted .
        allele = normalize_allele_name ( allele , omit_dra1 = True )
        normalized_alleles . append ( allele )
        request = self . _get_iedb_request_params ( amino_acid_sequence , allele )
        logger . info ( "Calling IEDB (%s) with request %s" , self . url , request )
        response_df = _query_iedb ( request , self . url )
        for _ , row in response_df . iterrows ( ) :
            binding_predictions . append ( BindingPrediction ( source_sequence_name = key , offset = row [ 'start' ] - 1 , allele = row [ 'allele' ] , peptide = row [ 'peptide' ] , affinity = row [ 'ic50' ] , percentile_rank = row [ 'rank' ] , prediction_method_name = "iedb-" + self . prediction_method ) )
self . _check_results ( binding_predictions , alleles = normalized_alleles , peptides = expected_peptides )
return BindingPredictionCollection ( binding_predictions )


signature:
def get_args ( func , skip = 0 ) :
    """Hackish way to get the arguments of a function
 Parameters
 func : callable
 Function to get the arguments from
 skip : int , optional
 Arguments to skip , defaults to 0 set it to 1 to skip the
 ` ` self ` ` argument of a method .
 Returns
 tuple
 Function ' s arguments"""

implementation:
code = getattr ( func , '__code__' , None )
if code is None :
    code = func . __call__ . __code__
return code . co_varnames [ skip : code . co_argcount ]


signature:
def log_error ( msg = None , exc_info = None , logger = None , ** kwargs ) :
    """log an exception and its traceback on the logger defined
 Parameters
 msg : str , optional
 A message to add to the error
 exc _ info : tuple
 Information about the current exception
 logger : logging . Logger
 logger to use"""

implementation:
if logger is None :
    logger = _logger
if not exc_info :
    exc_info = sys . exc_info ( )
if msg is None :
    msg = ""
exc_class , exc_msg , _ = exc_info
if all ( info is not None for info in exc_info ) :
    logger . error ( msg , exc_info = exc_info )


signature:
async def get_media_metadata ( data , path = None ) :
    """Get all the file ' s metadata and read any kind of file object
 Parameters
 data : bytes
 first bytes of the file ( the mimetype shoudl be guessed from the
 file headers
 path : str , optional
 path to the file
 Returns
 str
 The mimetype of the media
 str
 The category of the media on Twitter"""

implementation:
if isinstance ( data , bytes ) :
    media_type = await get_type ( data , path )
else :
    raise TypeError ( "get_metadata input must be a bytes" )
media_category = get_category ( media_type )
_logger . info ( "media_type: %s, media_category: %s" % ( media_type , media_category ) )
return media_type , media_category


signature:
async def get_size ( media ) :
    """Get the size of a file
 Parameters
 media : file object
 The file object of the media
 Returns
 int
 The size of the file"""

implementation:
if hasattr ( media , 'seek' ) :
    await execute ( media . seek ( 0 , os . SEEK_END ) )
    size = await execute ( media . tell ( ) )
    await execute ( media . seek ( 0 ) )
elif hasattr ( media , 'headers' ) :
    size = int ( media . headers [ 'Content-Length' ] )
elif isinstance ( media , bytes ) :
    size = len ( media )
else :
    raise TypeError ( "Can't get size of media of type:" , type ( media ) . __name__ )
_logger . info ( "media size: %dB" % size )
return size


signature:
async def get_type ( media , path = None ) :
    """Parameters
 media : file object
 A file object of the image
 path : str , optional
 The path to the file
 Returns
 str
 The mimetype of the media
 str
 The category of the media on Twitter"""

implementation:
if magic :
    if not media :
        raise TypeError ( "Media data is empty" )
    _logger . debug ( "guessing mimetype using magic" )
    media_type = mime . from_buffer ( media [ : 1024 ] )
else :
    media_type = None
    if path :
        _logger . debug ( "guessing mimetype using built-in module" )
        media_type = mime . guess_type ( path ) [ 0 ]
    if media_type is None :
        msg = ( "Could not guess the mimetype of the media.\n" "Please consider installing python-magic\n" "(pip3 install peony-twitter[magic])" )
        raise RuntimeError ( msg )
return media_type


signature:
def set_debug ( ) :
    """activates error messages , useful during development"""

implementation:
logging . basicConfig ( level = logging . WARNING )
peony . logger . setLevel ( logging . DEBUG )


signature:
def clone_with_updates ( self , ** kwargs ) :
    """Returns new BindingPrediction with updated fields"""

implementation:
fields_dict = self . to_dict ( )
fields_dict . update ( kwargs )
return BindingPrediction ( ** fields_dict )


signature:
def NetMHCpan ( alleles , program_name = "netMHCpan" , process_limit = - 1 , default_peptide_lengths = [ 9 ] , extra_flags = [ ] ) :
    """This function wraps NetMHCpan28 and NetMHCpan3 to automatically detect which class
 to use , with the help of the miraculous and strange ' - - version ' netmhcpan argument ."""

implementation:
# convert to str since Python3 returns a ` bytes ` object .
# The ' _ MHCTOOLS _ VERSION _ SNIFFING ' here is meaningless , but it is necessary
# to call ` netmhcpan - - version ` with some argument , otherwise it hangs .
with open ( os . devnull , 'w' ) as devnull :
    output = check_output ( [ program_name , "--version" , "_MHCTOOLS_VERSION_SNIFFING" ] , stderr = devnull )
output_str = output . decode ( "ascii" , "ignore" )
common_kwargs = { "alleles" : alleles , "default_peptide_lengths" : default_peptide_lengths , "program_name" : program_name , "process_limit" : process_limit , "extra_flags" : extra_flags , }
if "NetMHCpan version 2.8" in output_str :
    return NetMHCpan28 ( ** common_kwargs )
elif "NetMHCpan version 3.0" in output_str :
    return NetMHCpan3 ( ** common_kwargs )
elif "NetMHCpan version 4.0" in output_str :
    return NetMHCpan4 ( ** common_kwargs )
else :
    raise RuntimeError ( "This software expects NetMHCpan version 2.8, 3.0, or 4.0" )


signature:
def get_data ( self , response ) :
    """Get the data from the response"""

implementation:
if self . _response_list :
    return response
elif self . _response_key is None :
    if hasattr ( response , "items" ) :
        for key , data in response . items ( ) :
            if ( hasattr ( data , "__getitem__" ) and not hasattr ( data , "items" ) and len ( data ) > 0 and 'id' in data [ 0 ] ) :
                self . _response_key = key
                return data
    else :
        self . _response_list = True
        return response
else :
    return response [ self . _response_key ]
raise NoDataFound ( response = response , url = self . request . get_url ( ) )


signature:
async def call_on_response ( self , data ) :
    """Try to fill the gaps and strip last tweet from the response
 if its id is that of the first tweet of the last response
 Parameters
 data : list
 The response data"""

implementation:
since_id = self . kwargs . get ( self . param , 0 ) + 1
if self . fill_gaps :
    if data [ - 1 ] [ 'id' ] != since_id :
        max_id = data [ - 1 ] [ 'id' ] - 1
        responses = with_max_id ( self . request ( ** self . kwargs , max_id = max_id ) )
        async for tweets in responses :
            data . extend ( tweets )
    if data [ - 1 ] [ 'id' ] == self . last_id :
        data = data [ : - 1 ]
if not data and not self . force :
    raise StopAsyncIteration
await self . set_param ( data )


signature:
async def get_oauth_token ( consumer_key , consumer_secret , callback_uri = "oob" ) :
    """Get a temporary oauth token
 Parameters
 consumer _ key : str
 Your consumer key
 consumer _ secret : str
 Your consumer secret
 callback _ uri : str , optional
 Callback uri , defaults to ' oob '
 Returns
 dict
 Temporary tokens"""

implementation:
client = BasePeonyClient ( consumer_key = consumer_key , consumer_secret = consumer_secret , api_version = "" , suffix = "" )
response = await client . api . oauth . request_token . post ( _suffix = "" , oauth_callback = callback_uri )
return parse_token ( response )


signature:
async def get_oauth_verifier ( oauth_token ) :
    """Open authorize page in a browser ,
 print the url if it didn ' t work
 Arguments
 oauth _ token : str
 The oauth token received in : func : ` get _ oauth _ token `
 Returns
 str
 The PIN entered by the user"""

implementation:
url = "https://api.twitter.com/oauth/authorize?oauth_token=" + oauth_token
try :
    browser = webbrowser . open ( url )
    await asyncio . sleep ( 2 )
    if not browser :
        raise RuntimeError
except RuntimeError :
    print ( "could not open a browser\ngo here to enter your PIN: " + url )
verifier = input ( "\nEnter your PIN: " )
return verifier


signature:
async def get_access_token ( consumer_key , consumer_secret , oauth_token , oauth_token_secret , oauth_verifier , ** kwargs ) :
    """get the access token of the user
 Parameters
 consumer _ key : str
 Your consumer key
 consumer _ secret : str
 Your consumer secret
 oauth _ token : str
 OAuth token from : func : ` get _ oauth _ token `
 oauth _ token _ secret : str
 OAuth token secret from : func : ` get _ oauth _ token `
 oauth _ verifier : str
 OAuth verifier from : func : ` get _ oauth _ verifier `
 Returns
 dict
 Access tokens"""

implementation:
client = BasePeonyClient ( consumer_key = consumer_key , consumer_secret = consumer_secret , access_token = oauth_token , access_token_secret = oauth_token_secret , api_version = "" , suffix = "" )
response = await client . api . oauth . access_token . get ( _suffix = "" , oauth_verifier = oauth_verifier )
return parse_token ( response )


signature:
async def async_oauth_dance ( consumer_key , consumer_secret , callback_uri = "oob" ) :
    """OAuth dance to get the user ' s access token
 Parameters
 consumer _ key : str
 Your consumer key
 consumer _ secret : str
 Your consumer secret
 callback _ uri : str
 Callback uri , defaults to ' oob '
 Returns
 dict
 Access tokens"""

implementation:
token = await get_oauth_token ( consumer_key , consumer_secret , callback_uri )
oauth_verifier = await get_oauth_verifier ( token [ 'oauth_token' ] )
token = await get_access_token ( consumer_key , consumer_secret , oauth_verifier = oauth_verifier , ** token )
token = dict ( consumer_key = consumer_key , consumer_secret = consumer_secret , access_token = token [ 'oauth_token' ] , access_token_secret = token [ 'oauth_token_secret' ] )
return token


signature:
def parse_token ( response ) :
    """parse the responses containing the tokens
 Parameters
 response : str
 The response containing the tokens
 Returns
 dict
 The parsed tokens"""

implementation:
items = response . split ( "&" )
items = [ item . split ( "=" ) for item in items ]
return { key : value for key , value in items }


signature:
def oauth_dance ( consumer_key , consumer_secret , oauth_callback = "oob" , loop = None ) :
    """OAuth dance to get the user ' s access token
 It calls async _ oauth _ dance and create event loop of not given
 Parameters
 consumer _ key : str
 Your consumer key
 consumer _ secret : str
 Your consumer secret
 oauth _ callback : str
 Callback uri , defaults to ' oob '
 loop : event loop
 asyncio event loop
 Returns
 dict
 Access tokens"""

implementation:
loop = asyncio . get_event_loop ( ) if loop is None else loop
coro = async_oauth_dance ( consumer_key , consumer_secret , oauth_callback )
return loop . run_until_complete ( coro )


signature:
def oauth2_dance ( consumer_key , consumer_secret , loop = None ) :
    """oauth2 dance
 Parameters
 consumer _ key : str
 Your consumer key
 consumer _ secret : str
 Your consumer secret
 loop : event loop , optional
 event loop to use
 Returns
 str
 Bearer token"""

implementation:
loop = asyncio . get_event_loop ( ) if loop is None else loop
client = BasePeonyClient ( consumer_key = consumer_key , consumer_secret = consumer_secret , auth = oauth . OAuth2Headers )
loop . run_until_complete ( client . headers . sign ( ) )
return client . headers . token


signature:
def predict ( self , sequences ) :
    """Return netChop predictions for each position in each sequence .
 Parameters
 sequences : list of string
 Amino acid sequences to predict cleavage for
 Returns
 list of list of float
 The i ' th list corresponds to the i ' th sequence . Each list gives
 the cleavage probability for each position in the sequence ."""

implementation:
with tempfile . NamedTemporaryFile ( suffix = ".fsa" , mode = "w" ) as input_fd :
    for ( i , sequence ) in enumerate ( sequences ) :
        input_fd . write ( "> %d\n" % i )
        input_fd . write ( sequence )
        input_fd . write ( "\n" )
    input_fd . flush ( )
    try :
        output = subprocess . check_output ( [ "netChop" , input_fd . name ] )
    except subprocess . CalledProcessError as e :
        logging . error ( "Error calling netChop: %s:\n%s" % ( e , e . output ) )
        raise
parsed = self . parse_netchop ( output )
assert len ( parsed ) == len ( sequences ) , "Expected %d results but got %d" % ( len ( sequences ) , len ( parsed ) )
assert [ len ( x ) for x in parsed ] == [ len ( x ) for x in sequences ]
return parsed


signature:
def parse_netchop ( netchop_output ) :
    """Parse netChop stdout ."""

implementation:
line_iterator = iter ( netchop_output . decode ( ) . split ( "\n" ) )
scores = [ ]
for line in line_iterator :
    if "pos" in line and 'AA' in line and 'score' in line :
        scores . append ( [ ] )
        if "----" not in next ( line_iterator ) :
            raise ValueError ( "Dashes expected" )
        line = next ( line_iterator )
        while '-------' not in line :
            score = float ( line . split ( ) [ 3 ] )
            scores [ - 1 ] . append ( score )
            line = next ( line_iterator )
return scores


signature:
def to_dataframe ( self , columns = BindingPrediction . fields + ( "length" , ) ) :
    """Converts collection of BindingPrediction objects to DataFrame"""

implementation: return pd . DataFrame . from_records ( [ tuple ( [ getattr ( x , name ) for name in columns ] ) for x in self ] , columns = columns )


signature:
def NetMHC ( alleles , default_peptide_lengths = [ 9 ] , program_name = "netMHC" ) :
    """This function wraps NetMHC3 and NetMHC4 to automatically detect which class
 to use . Currently based on running the ' - h ' command and looking for
 discriminating substrings between the versions ."""

implementation:
# run NetMHC ' s help command and parse discriminating substrings out of
# the resulting str output
with open ( os . devnull , 'w' ) as devnull :
    help_output = check_output ( [ program_name , "-h" ] , stderr = devnull )
help_output_str = help_output . decode ( "ascii" , "ignore" )
substring_to_netmhc_class = { "-listMHC" : NetMHC4 , "--Alleles" : NetMHC3 , }
successes = [ ]
for substring , netmhc_class in substring_to_netmhc_class . items ( ) :
    if substring in help_output_str :
        successes . append ( netmhc_class )
if len ( successes ) > 1 :
    raise SystemError ( "Command %s is valid for multiple NetMHC versions. " "This is likely an mhctools bug." % program_name )
if len ( successes ) == 0 :
    raise SystemError ( "Command %s is not a valid way of calling any NetMHC software." % program_name )
netmhc_class = successes [ 0 ]
return netmhc_class ( alleles = alleles , default_peptide_lengths = default_peptide_lengths , program_name = program_name )


signature:
def predict_peptides ( self , peptides ) :
    """Predict MHC affinity for peptides ."""

implementation:
# importing locally to avoid slowing down CLI applications which
# don ' t use MHCflurry
from mhcflurry . encodable_sequences import EncodableSequences
binding_predictions = [ ]
encodable_sequences = EncodableSequences . create ( peptides )
for allele in self . alleles :
    predictions_df = self . predictor . predict_to_dataframe ( encodable_sequences , allele = allele )
    for ( _ , row ) in predictions_df . iterrows ( ) :
        binding_prediction = BindingPrediction ( allele = allele , peptide = row . peptide , affinity = row . prediction , percentile_rank = ( row . prediction_percentile if 'prediction_percentile' in row else nan ) , prediction_method_name = "mhcflurry" )
        binding_predictions . append ( binding_prediction )
return BindingPredictionCollection ( binding_predictions )


signature:
def seq_to_str ( obj , sep = "," ) :
    """Given a sequence convert it to a comma separated string .
 If , however , the argument is a single object , return its string
 representation ."""

implementation:
if isinstance ( obj , string_classes ) :
    return obj
elif isinstance ( obj , ( list , tuple ) ) :
    return sep . join ( [ str ( x ) for x in obj ] )
else :
    return str ( obj )


signature:
def convert ( img , formats ) :
    """Convert the image to all the formats specified
 Parameters
 img : PIL . Image . Image
 The image to convert
 formats : list
 List of all the formats to use
 Returns
 io . BytesIO
 A file object containing the converted image"""

implementation:
media = None
min_size = 0
for kwargs in formats :
    f = io . BytesIO ( )
    if img . mode == "RGBA" and kwargs [ 'format' ] != "PNG" : # convert to RGB if picture is too large as a png
    # this implies that the png format is the first in ` formats `
        if min_size < 5 * 1024 ** 2 :
            continue
        else :
            img . convert ( 'RGB' )
    img . save ( f , ** kwargs )
    size = f . tell ( )
    if media is None or size < min_size :
        if media is not None :
            media . close ( )
        media = f
        min_size = size
    else :
        f . close ( )
return media


signature:
def optimize_media ( file_ , max_size , formats ) :
    """Optimize an image
 Resize the picture to the ` ` max _ size ` ` , defaulting to the large
 photo size of Twitter in : meth : ` PeonyClient . upload _ media ` when
 used with the ` ` optimize _ media ` ` argument .
 Parameters
 file _ : file object
 the file object of an image
 max _ size : : obj : ` tuple ` or : obj : ` list ` of : obj : ` int `
 a tuple in the format ( width , height ) which is maximum size of
 the picture returned by this function
 formats : : obj ` list ` or : obj : ` tuple ` of : obj : ` dict `
 a list of all the formats to convert the picture to
 Returns
 file
 The smallest file created in this function"""

implementation:
if not PIL :
    msg = ( "Pillow must be installed to optimize a media\n" "$ pip3 install Pillow" )
    raise RuntimeError ( msg )
img = PIL . Image . open ( file_ )
# resize the picture ( defaults to the ' large ' photo size of Twitter
# in peony . PeonyClient . upload _ media )
ratio = max ( hw / max_hw for hw , max_hw in zip ( img . size , max_size ) )
if ratio > 1 :
    size = tuple ( int ( hw // ratio ) for hw in img . size )
    img = img . resize ( size , PIL . Image . ANTIALIAS )
media = convert ( img , formats )
# do not close a file opened by the user
# only close if a filename was given
if not hasattr ( file_ , 'read' ) :
    img . close ( )
return media


signature:
def create_input_peptides_files ( peptides , max_peptides_per_file = None , group_by_length = False ) :
    """Creates one or more files containing one peptide per line ,
 returns names of files ."""

implementation:
if group_by_length :
    peptide_lengths = { len ( p ) for p in peptides }
    peptide_groups = { l : [ ] for l in peptide_lengths }
    for p in peptides :
        peptide_groups [ len ( p ) ] . append ( p )
else :
    peptide_groups = { "" : peptides }
file_names = [ ]
for key , group in peptide_groups . items ( ) :
    n_peptides = len ( group )
    if not max_peptides_per_file :
        max_peptides_per_file = n_peptides
    input_file = None
    for i , p in enumerate ( group ) :
        if i % max_peptides_per_file == 0 :
            if input_file is not None :
                file_names . append ( input_file . name )
                input_file . close ( )
            input_file = make_writable_tempfile ( prefix_number = i // max_peptides_per_file , prefix_name = key , suffix = ".txt" )
        input_file . write ( "%s\n" % p )
    if input_file is not None :
        file_names . append ( input_file . name )
        input_file . close ( )
return file_names


signature:
def _check_peptide_lengths ( self , peptide_lengths = None ) :
    """If peptide lengths not specified , then try using the default
 lengths associated with this predictor object . If those aren ' t
 a valid non - empty sequence of integers , then raise an exception .
 Otherwise return the peptide lengths ."""

implementation:
if not peptide_lengths :
    peptide_lengths = self . default_peptide_lengths
if not peptide_lengths :
    raise ValueError ( ( "Must either provide 'peptide_lengths' argument " "or set 'default_peptide_lengths" ) )
if isinstance ( peptide_lengths , int ) :
    peptide_lengths = [ peptide_lengths ]
require_iterable_of ( peptide_lengths , int )
for peptide_length in peptide_lengths :
    if ( self . min_peptide_length is not None and peptide_length < self . min_peptide_length ) :
        raise ValueError ( "Invalid peptide length %d, shorter than min %d" % ( peptide_length , self . min_peptide_length ) )
    elif ( self . max_peptide_length is not None and peptide_length > self . max_peptide_length ) :
        raise ValueError ( "Invalid peptide length %d, longer than max %d" % ( peptide_length , self . max_peptide_length ) )
return peptide_lengths


signature:
def _check_peptide_inputs ( self , peptides ) :
    """Check peptide sequences to make sure they are valid for this predictor ."""

implementation:
require_iterable_of ( peptides , string_types )
check_X = not self . allow_X_in_peptides
check_lower = not self . allow_lowercase_in_peptides
check_min_length = self . min_peptide_length is not None
min_length = self . min_peptide_length
check_max_length = self . max_peptide_length is not None
max_length = self . max_peptide_length
for p in peptides :
    if not p . isalpha ( ) :
        raise ValueError ( "Invalid characters in peptide '%s'" % p )
    elif check_X and "X" in p :
        raise ValueError ( "Invalid character 'X' in peptide '%s'" % p )
    elif check_lower and not p . isupper ( ) :
        raise ValueError ( "Invalid lowercase letters in peptide '%s'" % p )
    elif check_min_length and len ( p ) < min_length :
        raise ValueError ( "Peptide '%s' too short (%d chars), must be at least %d" % ( p , len ( p ) , min_length ) )
    elif check_max_length and len ( p ) > max_length :
        raise ValueError ( "Peptide '%s' too long (%d chars), must be at least %d" % ( p , len ( p ) , max_length ) )


signature:
def predict_subsequences ( self , sequence_dict , peptide_lengths = None ) :
    """Given a dictionary mapping sequence names to amino acid strings ,
 and an optional list of peptide lengths , returns a
 BindingPredictionCollection ."""

implementation:
if isinstance ( sequence_dict , string_types ) :
    sequence_dict = { "seq" : sequence_dict }
elif isinstance ( sequence_dict , ( list , tuple ) ) :
    sequence_dict = { seq : seq for seq in sequence_dict }
peptide_lengths = self . _check_peptide_lengths ( peptide_lengths )
# convert long protein sequences to set of peptides and
# associated sequence name / offsets that each peptide may have come
# from
peptide_set = set ( [ ] )
peptide_to_name_offset_pairs = defaultdict ( list )
for name , sequence in sequence_dict . items ( ) :
    for peptide_length in peptide_lengths :
        for i in range ( len ( sequence ) - peptide_length + 1 ) :
            peptide = sequence [ i : i + peptide_length ]
            peptide_set . add ( peptide )
            peptide_to_name_offset_pairs [ peptide ] . append ( ( name , i ) )
peptide_list = sorted ( peptide_set )
binding_predictions = self . predict_peptides ( peptide_list )
# create BindingPrediction objects with sequence name and offset
results = [ ]
for binding_prediction in binding_predictions :
    for name , offset in peptide_to_name_offset_pairs [ binding_prediction . peptide ] :
        results . append ( binding_prediction . clone_with_updates ( source_sequence_name = name , offset = offset ) )
self . _check_results ( results , peptides = peptide_set , alleles = self . alleles )
return BindingPredictionCollection ( results )


signature:
def _check_hla_alleles ( alleles , valid_alleles = None ) :
    """Given a list of HLA alleles and an optional list of valid
 HLA alleles , return a set of alleles that we will pass into
 the MHC binding predictor ."""

implementation:
require_iterable_of ( alleles , string_types , "HLA alleles" )
# Don ' t run the MHC predictor twice for homozygous alleles ,
# only run it for unique alleles
alleles = { normalize_allele_name ( allele . strip ( ) . upper ( ) ) for allele in alleles }
if valid_alleles : # For some reason netMHCpan drops the ' * ' in names , so
# ' HLA - A * 03:01 ' becomes ' HLA - A03:01'
    missing_alleles = [ allele for allele in alleles if allele not in valid_alleles ]
    if len ( missing_alleles ) > 0 :
        raise UnsupportedAllele ( "Unsupported HLA alleles: %s" % missing_alleles )
return list ( alleles )


signature:
async def _connect ( self ) :
    """Connect to the stream
 Returns
 asyncio . coroutine
 The streaming response"""

implementation:
logger . debug ( "connecting to the stream" )
await self . client . setup
if self . session is None :
    self . session = self . client . _session
kwargs = await self . client . headers . prepare_request ( ** self . kwargs )
request = self . client . error_handler ( self . session . request )
return await request ( timeout = 0 , ** kwargs )


signature:
async def connect ( self ) :
    """Create the connection
 Returns
 self
 Raises
 exception . PeonyException
 On a response status in 4xx that are not status 420 or 429
 Also on statuses in 1xx or 3xx since this should not be the status
 received here"""

implementation:
with async_timeout . timeout ( self . timeout ) :
    self . response = await self . _connect ( )
if self . response . status in range ( 200 , 300 ) :
    self . _error_timeout = 0
    self . state = NORMAL
elif self . response . status == 500 :
    self . state = DISCONNECTION
elif self . response . status in range ( 501 , 600 ) :
    self . state = RECONNECTION
elif self . response . status in ( 420 , 429 ) :
    self . state = ENHANCE_YOUR_CALM
else :
    logger . debug ( "raising error during stream connection" )
    raise await exceptions . throw ( self . response , loads = self . client . _loads , url = self . kwargs [ 'url' ] )
logger . debug ( "stream state: %d" % self . state )


signature:
async def init_restart ( self , error = None ) :
    """Restart the stream on error
 Parameters
 error : bool , optional
 Whether to print the error or not"""

implementation:
if error :
    utils . log_error ( logger = logger )
if self . state == DISCONNECTION :
    if self . _error_timeout < MAX_DISCONNECTION_TIMEOUT :
        self . _error_timeout += DISCONNECTION_TIMEOUT
    logger . info ( "The stream was disconnected, will reconnect in %ss" % self . _error_timeout )
elif self . state == RECONNECTION :
    if self . _error_timeout < RECONNECTION_TIMEOUT :
        self . _error_timeout = RECONNECTION_TIMEOUT
    elif self . _error_timeout < MAX_RECONNECTION_TIMEOUT :
        self . _error_timeout *= 2
    logger . info ( "Could not connect to the stream, reconnection in %ss" % self . _error_timeout )
elif self . state == ENHANCE_YOUR_CALM :
    if self . _error_timeout < ENHANCE_YOUR_CALM_TIMEOUT :
        self . _error_timeout = ENHANCE_YOUR_CALM_TIMEOUT
    else :
        self . _error_timeout *= 2
    logger . warning ( "Enhance Your Calm response received from Twitter. " "If you didn't restart your program frenetically " "then there is probably something wrong with it. " "Make sure you are not opening too many connections" " to the endpoint you are currently using by " "checking out Twitter's Streaming API " "documentation: " "https://dev.twitter.com/streaming/overview\n" "The stream will restart in %ss." % self . _error_timeout )
elif self . state == EOF :
    pass
    # no timeout
else :
    raise RuntimeError ( "Incorrect state: %d" % self . state )
self . _reconnecting = True
return { 'reconnecting_in' : self . _error_timeout , 'error' : error }


signature:
async def restart_stream ( self ) :
    """Restart the stream on error"""

implementation:
await self . response . release ( )
await asyncio . sleep ( self . _error_timeout )
await self . connect ( )
logger . info ( "Reconnected to the stream" )
self . _reconnecting = False
return { 'stream_restart' : True }


signature:
def with_prefix ( self , prefix , strict = False ) :
    """decorator to handle commands with prefixes
 Parameters
 prefix : str
 the prefix of the command
 strict : bool , optional
 If set to True the command must be at the beginning
 of the message . Defaults to False .
 Returns
 function
 a decorator that returns an : class : ` EventHandler ` instance"""

implementation:
def decorated ( func ) :
    return EventHandler ( func = func , event = self . event , prefix = prefix , strict = strict )
return decorated


signature:
def envelope ( self ) :
    """returns an : class : ` Event ` that can be used for site streams"""

implementation:
def enveloped_event ( data ) :
    return 'for_user' in data and self . _func ( data . get ( 'message' ) )
return self . __class__ ( enveloped_event , self . __name__ )


signature:
async def set_tz ( self ) :
    """set the environment timezone to the timezone
 set in your twitter settings"""

implementation:
settings = await self . api . account . settings . get ( )
tz = settings . time_zone . tzinfo_name
os . environ [ 'TZ' ] = tz
time . tzset ( )


signature:
def run_command ( args , ** kwargs ) :
    """Given a list whose first element is a command name , followed by arguments ,
 execute it and show timing info ."""

implementation:
assert len ( args ) > 0
start_time = time . time ( )
process = AsyncProcess ( args , ** kwargs )
process . wait ( )
elapsed_time = time . time ( ) - start_time
logger . info ( "%s took %0.4f seconds" , args [ 0 ] , elapsed_time )


signature:
def run_multiple_commands_redirect_stdout ( multiple_args_dict , print_commands = True , process_limit = - 1 , polling_freq = 0.5 , ** kwargs ) :
    """Run multiple shell commands in parallel , write each of their
 stdout output to files associated with each command .
 Parameters
 multiple _ args _ dict : dict
 A dictionary whose keys are files and values are args list .
 Run each args list as a subprocess and write stdout to the
 corresponding file .
 print _ commands : bool
 Print shell commands before running them .
 process _ limit : int
 Limit the number of concurrent processes to this number . 0
 if there is no limit , - 1 to use max number of processors
 polling _ freq : int
 Number of seconds between checking for done processes , if
 we have a process limit"""

implementation:
assert len ( multiple_args_dict ) > 0
assert all ( len ( args ) > 0 for args in multiple_args_dict . values ( ) )
assert all ( hasattr ( f , 'name' ) for f in multiple_args_dict . keys ( ) )
if process_limit < 0 :
    logger . debug ( "Using %d processes" % cpu_count ( ) )
    process_limit = cpu_count ( )
start_time = time . time ( )
processes = Queue ( maxsize = process_limit )
def add_to_queue ( process ) :
    process . start ( )
    if print_commands :
        handler = logging . FileHandler ( process . redirect_stdout_file . name )
        handler . setLevel ( logging . DEBUG )
        logger . addHandler ( handler )
        logger . debug ( " " . join ( process . args ) )
        logger . removeHandler ( handler )
    processes . put ( process )
for f , args in multiple_args_dict . items ( ) :
    p = AsyncProcess ( args , redirect_stdout_file = f , ** kwargs )
    if not processes . full ( ) :
        add_to_queue ( p )
    else :
        while processes . full ( ) : # Are there any done processes ?
            to_remove = [ ]
            for possibly_done in processes . queue :
                if possibly_done . poll ( ) is not None :
                    possibly_done . wait ( )
                    to_remove . append ( possibly_done )
            # Remove them from the queue and stop checking
            if to_remove :
                for process_to_remove in to_remove :
                    processes . queue . remove ( process_to_remove )
                break
            # Check again in a second if there weren ' t
            time . sleep ( polling_freq )
        add_to_queue ( p )
# Wait for all the rest of the processes
while not processes . empty ( ) :
    processes . get ( ) . wait ( )
elapsed_time = time . time ( ) - start_time
logger . info ( "Ran %d commands in %0.4f seconds" , len ( multiple_args_dict ) , elapsed_time )


signature:
def _determine_supported_alleles ( command , supported_allele_flag ) :
    """Try asking the commandline predictor ( e . g . netMHCpan )
 which alleles it supports ."""

implementation:
try : # convert to str since Python3 returns a ` bytes ` object
    supported_alleles_output = check_output ( [ command , supported_allele_flag ] )
    supported_alleles_str = supported_alleles_output . decode ( "ascii" , "ignore" )
    assert len ( supported_alleles_str ) > 0 , '%s returned empty allele list' % command
    supported_alleles = set ( [ ] )
    for line in supported_alleles_str . split ( "\n" ) :
        line = line . strip ( )
        if not line . startswith ( '#' ) and len ( line ) > 0 :
            try : # We need to normalize these alleles ( the output of the predictor
            # when it lists its supported alleles ) so that they are comparable with
            # our own alleles .
                supported_alleles . add ( normalize_allele_name ( line ) )
            except AlleleParseError as error :
                logger . info ( "Skipping allele %s: %s" , line , error )
                continue
    if len ( supported_alleles ) == 0 :
        raise ValueError ( "Unable to determine supported alleles" )
    return supported_alleles
except Exception as e :
    logger . exception ( e )
    raise SystemError ( "Failed to run %s %s. Possibly an incorrect executable version?" % ( command , supported_allele_flag ) )


signature:
def loads ( json_data , encoding = "utf-8" , ** kwargs ) :
    """Custom loads function with an object _ hook and automatic decoding
 Parameters
 json _ data : str
 The JSON data to decode
 * args
 Positional arguments , passed to : func : ` json . loads `
 encoding : : obj : ` str ` , optional
 The encoding of the bytestring
 * * kwargs
 Keyword arguments passed to : func : ` json . loads `
 Returns
 : obj : ` dict ` or : obj : ` list `
 Decoded json data"""

implementation:
if isinstance ( json_data , bytes ) :
    json_data = json_data . decode ( encoding )
return json . loads ( json_data , object_hook = JSONData , ** kwargs )


signature:
async def read ( response , loads = loads , encoding = None ) :
    """read the data of the response
 Parameters
 response : aiohttp . ClientResponse
 response
 loads : callable
 json loads function
 encoding : : obj : ` str ` , optional
 character encoding of the response , if set to None
 aiohttp should guess the right encoding
 Returns
 : obj : ` bytes ` , : obj : ` str ` , : obj : ` dict ` or : obj : ` list `
 the data returned depends on the response"""

implementation:
ctype = response . headers . get ( 'Content-Type' , "" ) . lower ( )
try :
    if "application/json" in ctype :
        logger . info ( "decoding data as json" )
        return await response . json ( encoding = encoding , loads = loads )
    if "text" in ctype :
        logger . info ( "decoding data as text" )
        return await response . text ( encoding = encoding )
except ( UnicodeDecodeError , json . JSONDecodeError ) as exc :
    data = await response . read ( )
    raise exceptions . PeonyDecodeError ( response = response , data = data , exception = exc )
return await response . read ( )


signature:
def doc ( func ) :
    """Find the message shown when someone calls the help command
 Parameters
 func : function
 the function
 Returns
 str
 The help message for this command"""

implementation:
stripped_chars = " \t"
if hasattr ( func , '__doc__' ) :
    docstring = func . __doc__ . lstrip ( " \n\t" )
    if "\n" in docstring :
        i = docstring . index ( "\n" )
        return docstring [ : i ] . rstrip ( stripped_chars )
    elif docstring :
        return docstring . rstrip ( stripped_chars )
return ""


signature:
def permission_check ( data , command_permissions , command = None , permissions = None ) :
    """Check the permissions of the user requesting a command
 Parameters
 data : dict
 message data
 command _ permissions : dict
 permissions of the command , contains all the roles as key and users
 with these permissions as values
 command : function
 the command that is run
 permissions : tuple or list
 a list of permissions for the command
 Returns
 bool
 True if the user has the right permissions , False otherwise"""

implementation:
if permissions :
    pass
elif command :
    if hasattr ( command , 'permissions' ) :
        permissions = command . permissions
    else :
        return True
        # true if no permission is required
else :
    msg = "{name} must be called with command or permissions argument"
    raise RuntimeError ( msg . format ( name = "_permission_check" ) )
return any ( data [ 'sender' ] [ 'id' ] in command_permissions [ permission ] for permission in permissions if permission in command_permissions )


signature:
def main ( args_list = None ) :
    """Script to make pMHC binding predictions from amino acid sequences .
 Usage example :
 mhctools
 - - sequence SFFPIQQQQQAAALLLI - - sequence SILQQQAQAQQAQAASSSC - - extract - subsequences - - mhc - predictor netmhc - - mhc - alleles HLA - A0201 H2 - Db - - mhc - predictor netmhc - - output - csv epitope . csv"""

implementation:
args = parse_args ( args_list )
binding_predictions = run_predictor ( args )
df = binding_predictions . to_dataframe ( )
logger . info ( '\n%s' , df )
if args . output_csv :
    df . to_csv ( args . output_csv , index = False )
    print ( "Wrote: %s" % args . output_csv )


signature:
def _prepare_drb_allele_name ( self , parsed_beta_allele ) :
    """Assume that we ' re dealing with a human DRB allele
 which NetMHCIIpan treats differently because there is
 little population diversity in the DR - alpha gene"""

implementation:
if "DRB" not in parsed_beta_allele . gene :
    raise ValueError ( "Unexpected allele %s" % parsed_beta_allele )
return "%s_%s%s" % ( parsed_beta_allele . gene , parsed_beta_allele . allele_family , parsed_beta_allele . allele_code )


signature:
def prepare_allele_name ( self , allele_name ) :
    """netMHCIIpan has some unique requirements for allele formats ,
 expecting the following forms :
 - DRB1_0101 ( for non - alpha / beta pairs )
 - HLA - DQA10501 - DQB10636 ( for alpha and beta pairs )
 Other than human class II alleles , the only other alleles that
 netMHCIIpan accepts are the following mouse alleles :
 - H - 2 - IAb
 - H - 2 - IAd"""

implementation:
parsed_alleles = parse_classi_or_classii_allele_name ( allele_name )
if len ( parsed_alleles ) == 1 :
    allele = parsed_alleles [ 0 ]
    if allele . species == "H-2" :
        return "%s-%s%s" % ( allele . species , allele . gene , allele . allele_code )
    return self . _prepare_drb_allele_name ( allele )
else :
    alpha , beta = parsed_alleles
    if "DRA" in alpha . gene :
        return self . _prepare_drb_allele_name ( beta )
    return "HLA-%s%s%s-%s%s%s" % ( alpha . gene , alpha . allele_family , alpha . allele_code , beta . gene , beta . allele_family , beta . allele_code )


signature:
def get_error ( data ) :
    """return the error if there is a corresponding exception"""

implementation:
if isinstance ( data , dict ) :
    if 'errors' in data :
        error = data [ 'errors' ] [ 0 ]
    else :
        error = data . get ( 'error' , None )
    if isinstance ( error , dict ) :
        if error . get ( 'code' ) in errors :
            return error


signature:
async def throw ( response , loads = None , encoding = None , ** kwargs ) :
    """Get the response data if possible and raise an exception"""

implementation:
if loads is None :
    loads = data_processing . loads
data = await data_processing . read ( response , loads = loads , encoding = encoding )
error = get_error ( data )
if error is not None :
    exception = errors [ error [ 'code' ] ]
    raise exception ( response = response , error = error , data = data , ** kwargs )
if response . status in statuses :
    exception = statuses [ response . status ]
    raise exception ( response = response , data = data , ** kwargs )
# raise PeonyException if no specific exception was found
raise PeonyException ( response = response , data = data , ** kwargs )


signature:
def code ( self , code ) :
    """Decorator to associate a code to an exception"""

implementation:
def decorator ( exception ) :
    self [ code ] = exception
    return exception
return decorator


signature:
async def prepare_request ( self , method , url , headers = None , skip_params = False , proxy = None , ** kwargs ) :
    """prepare all the arguments for the request
 Parameters
 method : str
 HTTP method used by the request
 url : str
 The url to request
 headers : dict , optional
 Additionnal headers
 proxy : str
 proxy of the request
 skip _ params : bool
 Don ' t use the parameters to sign the request
 Returns
 dict
 Parameters of the request correctly formatted"""

implementation:
if method . lower ( ) == "post" :
    key = 'data'
else :
    key = 'params'
if key in kwargs and not skip_params :
    request_params = { key : kwargs . pop ( key ) }
else :
    request_params = { }
request_params . update ( dict ( method = method . upper ( ) , url = url ) )
coro = self . sign ( ** request_params , skip_params = skip_params , headers = headers )
request_params [ 'headers' ] = await utils . execute ( coro )
request_params [ 'proxy' ] = proxy
kwargs . update ( request_params )
return kwargs


signature:
def _user_headers ( self , headers = None ) :
    """Make sure the user doesn ' t override the Authorization header"""

implementation:
h = self . copy ( )
if headers is not None :
    keys = set ( headers . keys ( ) )
    if h . get ( 'Authorization' , False ) :
        keys -= { 'Authorization' }
    for key in keys :
        h [ key ] = headers [ key ]
return h


signature:
def process_keys ( func ) :
    """Raise error for keys that are not strings
 and add the prefix if it is missing"""

implementation:
@ wraps ( func )
def decorated ( self , k , * args ) :
    if not isinstance ( k , str ) :
        msg = "%s: key must be a string" % self . __class__ . __name__
        raise ValueError ( msg )
    if not k . startswith ( self . prefix ) :
        k = self . prefix + k
    return func ( self , k , * args )
return decorated


signature:
def _get ( self , text ) :
    """Analyze the text to get the right function
 Parameters
 text : str
 The text that could call a function"""

implementation:
if self . strict :
    match = self . prog . match ( text )
    if match :
        cmd = match . group ( )
        if cmd in self :
            return cmd
else :
    words = self . prog . findall ( text )
    for word in words :
        if word in self :
            return word


signature:
async def run ( self , * args , data ) :
    """run the function you want"""

implementation:
cmd = self . _get ( data . text )
try :
    if cmd is not None :
        command = self [ cmd ] ( * args , data = data )
        return await peony . utils . execute ( command )
except :
    fmt = "Error occurred while running function {cmd}:"
    peony . utils . log_error ( fmt . format ( cmd = cmd ) )


signature:
def get_cartesian ( r , theta ) :
    """Given a radius and theta , return the cartesian ( x , y ) coordinates ."""

implementation:
x = r * np . sin ( theta )
y = r * np . cos ( theta )
return x , y


signature:
def simplified_edges ( self ) :
    """A generator for getting all of the edges without consuming extra
 memory ."""

implementation:
for group , edgelist in self . edges . items ( ) :
    for u , v , d in edgelist :
        yield ( u , v )


signature:
def initialize_major_angle ( self ) :
    """Computes the major angle : 2pi radians / number of groups ."""

implementation:
num_groups = len ( self . nodes . keys ( ) )
self . major_angle = 2 * np . pi / num_groups


signature:
def initialize_minor_angle ( self ) :
    """Computes the minor angle : 2pi radians / 3 * number of groups ."""

implementation:
num_groups = len ( self . nodes . keys ( ) )
self . minor_angle = 2 * np . pi / ( 6 * num_groups )


signature:
def plot_radius ( self ) :
    """Computes the plot radius : maximum of length of each list of nodes ."""

implementation:
plot_rad = 0
for group , nodelist in self . nodes . items ( ) :
    proposed_radius = len ( nodelist ) * self . scale
    if proposed_radius > plot_rad :
        plot_rad = proposed_radius
return plot_rad + self . internal_radius


signature:
def has_edge_within_group ( self , group ) :
    """Checks whether there are within - group edges or not ."""

implementation:
assert group in self . nodes . keys ( ) , "{0} not one of the group of nodes" . format ( group )
nodelist = self . nodes [ group ]
for n1 , n2 in self . simplified_edges ( ) :
    if n1 in nodelist and n2 in nodelist :
        return True


signature:
def plot_axis ( self , rs , theta ) :
    """Renders the axis ."""

implementation:
xs , ys = get_cartesian ( rs , theta )
self . ax . plot ( xs , ys , 'black' , alpha = 0.3 )


signature:
def plot_nodes ( self , nodelist , theta , group ) :
    """Plots nodes to screen ."""

implementation:
for i , node in enumerate ( nodelist ) :
    r = self . internal_radius + i * self . scale
    x , y = get_cartesian ( r , theta )
    circle = plt . Circle ( xy = ( x , y ) , radius = self . dot_radius , color = self . node_colormap [ group ] , linewidth = 0 )
    self . ax . add_patch ( circle )


signature:
def group_theta ( self , group ) :
    """Computes the theta along which a group ' s nodes are aligned ."""

implementation:
for i , g in enumerate ( self . nodes . keys ( ) ) :
    if g == group :
        break
return i * self . major_angle