signature:
def get_notification_subject_line ( course_name , template_configuration = None ) :
    """Get a subject line for a notification email .
 The method is designed to fail in a " smart " way ; if we can ' t render a
 database - backed subject line template , then we ' ll fall back to a template
 saved in the Django settings ; if we can ' t render _ that _ one , then we ' ll
 fall through to a friendly string written into the code .
 One example of a failure case in which we want to fall back to a stock template
 would be if a site admin entered a subject line string that contained a template
 tag that wasn ' t available , causing a KeyError to be raised .
 Arguments :
 course _ name ( str ) : Course name to be rendered into the string
 template _ configuration : A database - backed object with a stored subject line template"""

implementation:
stock_subject_template = _ ( 'You\'ve been enrolled in {course_name}!' )
default_subject_template = getattr ( settings , 'ENTERPRISE_ENROLLMENT_EMAIL_DEFAULT_SUBJECT_LINE' , stock_subject_template , )
if template_configuration is not None and template_configuration . subject_line :
    final_subject_template = template_configuration . subject_line
else :
    final_subject_template = default_subject_template
try :
    return final_subject_template . format ( course_name = course_name )
except KeyError :
    pass
try :
    return default_subject_template . format ( course_name = course_name )
except KeyError :
    return stock_subject_template . format ( course_name = course_name )


signature:
def send_email_notification_message ( user , enrolled_in , enterprise_customer , email_connection = None ) :
    """Send an email notifying a user about their enrollment in a course .
 Arguments :
 user : Either a User object or a PendingEnterpriseCustomerUser that we can use
 to get details for the email
 enrolled _ in ( dict ) : The dictionary contains details of the enrollable object
 ( either course or program ) that the user enrolled in . This MUST contain
 a ` name ` key , and MAY contain the other following keys :
 - url : A human - friendly link to the enrollable ' s home page
 - type : Either ` course ` or ` program ` at present
 - branding : A special name for what the enrollable " is " ; for example ,
 " MicroMasters " would be the branding for a " MicroMasters Program "
 - start : A datetime object indicating when the enrollable will be available .
 enterprise _ customer : The EnterpriseCustomer that the enrollment was created using .
 email _ connection : An existing Django email connection that can be used without
 creating a new connection for each individual message"""

implementation:
if hasattr ( user , 'first_name' ) and hasattr ( user , 'username' ) : # PendingEnterpriseCustomerUsers don ' t have usernames or real names . We should
# template slightly differently to make sure weird stuff doesn ' t happen .
    user_name = user . first_name
    if not user_name :
        user_name = user . username
else :
    user_name = None
# Users have an ` email ` attribute ; PendingEnterpriseCustomerUsers have ` user _ email ` .
if hasattr ( user , 'email' ) :
    user_email = user . email
elif hasattr ( user , 'user_email' ) :
    user_email = user . user_email
else :
    raise TypeError ( _ ( '`user` must have one of either `email` or `user_email`.' ) )
msg_context = { 'user_name' : user_name , 'enrolled_in' : enrolled_in , 'organization_name' : enterprise_customer . name , }
try :
    enterprise_template_config = enterprise_customer . enterprise_enrollment_template
except ( ObjectDoesNotExist , AttributeError ) :
    enterprise_template_config = None
plain_msg , html_msg = build_notification_message ( msg_context , enterprise_template_config )
subject_line = get_notification_subject_line ( enrolled_in [ 'name' ] , enterprise_template_config )
from_email_address = get_configuration_value_for_site ( enterprise_customer . site , 'DEFAULT_FROM_EMAIL' , default = settings . DEFAULT_FROM_EMAIL )
return mail . send_mail ( subject_line , plain_msg , from_email_address , [ user_email ] , html_message = html_msg , connection = email_connection )


signature:
def get_enterprise_customer ( uuid ) :
    """Get the ` ` EnterpriseCustomer ` ` instance associated with ` ` uuid ` ` .
 : param uuid : The universally unique ID of the enterprise customer .
 : return : The ` ` EnterpriseCustomer ` ` instance , or ` ` None ` ` if it doesn ' t exist ."""

implementation:
EnterpriseCustomer = apps . get_model ( 'enterprise' , 'EnterpriseCustomer' )
# pylint : disable = invalid - name
try :
    return EnterpriseCustomer . objects . get ( uuid = uuid )
    # pylint : disable = no - member
except EnterpriseCustomer . DoesNotExist :
    return None


signature:
def get_enterprise_customer_for_user ( auth_user ) :
    """Return enterprise customer instance for given user .
 Some users are associated with an enterprise customer via ` EnterpriseCustomerUser ` model ,
 1 . if given user is associated with any enterprise customer , return enterprise customer .
 2 . otherwise return ` None ` .
 Arguments :
 auth _ user ( contrib . auth . User ) : Django User
 Returns :
 ( EnterpriseCustomer ) : enterprise customer associated with the current user ."""

implementation:
EnterpriseCustomerUser = apps . get_model ( 'enterprise' , 'EnterpriseCustomerUser' )
# pylint : disable = invalid - name
try :
    return EnterpriseCustomerUser . objects . get ( user_id = auth_user . id ) . enterprise_customer
    # pylint : disable = no - member
except EnterpriseCustomerUser . DoesNotExist :
    return None


signature:
def get_enterprise_customer_user ( user_id , enterprise_uuid ) :
    """Return the object for EnterpriseCustomerUser .
 Arguments :
 user _ id ( str ) : user identifier
 enterprise _ uuid ( UUID ) : Universally unique identifier for the enterprise customer .
 Returns :
 ( EnterpriseCustomerUser ) : enterprise customer user record"""

implementation:
EnterpriseCustomerUser = apps . get_model ( 'enterprise' , 'EnterpriseCustomerUser' )
# pylint : disable = invalid - name
try :
    return EnterpriseCustomerUser . objects . get ( # pylint : disable = no - member
    enterprise_customer__uuid = enterprise_uuid , user_id = user_id )
except EnterpriseCustomerUser . DoesNotExist :
    return None


signature:
def get_course_track_selection_url ( course_run , query_parameters ) :
    """Return track selection url for the given course .
 Arguments :
 course _ run ( dict ) : A dictionary containing course run metadata .
 query _ parameters ( dict ) : A dictionary containing query parameters to be added to course selection url .
 Raises :
 ( KeyError ) : Raised when course run dict does not have ' key ' key .
 Returns :
 ( str ) : Course track selection url ."""

implementation:
try :
    course_root = reverse ( 'course_modes_choose' , kwargs = { 'course_id' : course_run [ 'key' ] } )
except KeyError :
    LOGGER . exception ( "KeyError while parsing course run data.\nCourse Run: \n[%s]" , course_run , )
    raise
url = '{}{}' . format ( settings . LMS_ROOT_URL , course_root )
course_run_url = update_query_parameters ( url , query_parameters )
return course_run_url


signature:
def update_query_parameters ( url , query_parameters ) :
    """Return url with updated query parameters .
 Arguments :
 url ( str ) : Original url whose query parameters need to be updated .
 query _ parameters ( dict ) : A dictionary containing query parameters to be added to course selection url .
 Returns :
 ( slug ) : slug identifier for the identity provider that can be used for identity verification of
 users associated the enterprise customer of the given user ."""

implementation:
scheme , netloc , path , query_string , fragment = urlsplit ( url )
url_params = parse_qs ( query_string )
# Update url query parameters
url_params . update ( query_parameters )
return urlunsplit ( ( scheme , netloc , path , urlencode ( sorted ( url_params . items ( ) ) , doseq = True ) , fragment ) , )


signature:
def filter_audit_course_modes ( enterprise_customer , course_modes ) :
    """Filter audit course modes out if the enterprise customer has not enabled the ' Enable audit enrollment ' flag .
 Arguments :
 enterprise _ customer : The EnterpriseCustomer that the enrollment was created using .
 course _ modes : iterable with dictionaries containing a required ' mode ' key"""

implementation:
audit_modes = getattr ( settings , 'ENTERPRISE_COURSE_ENROLLMENT_AUDIT_MODES' , [ 'audit' ] )
if not enterprise_customer . enable_audit_enrollment :
    return [ course_mode for course_mode in course_modes if course_mode [ 'mode' ] not in audit_modes ]
return course_modes


signature:
def get_enterprise_customer_or_404 ( enterprise_uuid ) :
    """Given an EnterpriseCustomer UUID , return the corresponding EnterpriseCustomer or raise a 404.
 Arguments :
 enterprise _ uuid ( str ) : The UUID ( in string form ) of the EnterpriseCustomer to fetch .
 Returns :
 ( EnterpriseCustomer ) : The EnterpriseCustomer given the UUID ."""

implementation:
EnterpriseCustomer = apps . get_model ( 'enterprise' , 'EnterpriseCustomer' )
# pylint : disable = invalid - name
try :
    enterprise_uuid = UUID ( enterprise_uuid )
    return EnterpriseCustomer . objects . get ( uuid = enterprise_uuid )
    # pylint : disable = no - member
except ( TypeError , ValueError , EnterpriseCustomer . DoesNotExist ) :
    LOGGER . error ( 'Unable to find enterprise customer for UUID: [%s]' , enterprise_uuid )
    raise Http404


signature:
def get_cache_key ( ** kwargs ) :
    """Get MD5 encoded cache key for given arguments .
 Here is the format of key before MD5 encryption .
 key1 : value1 _ _ key2 : value2 . . .
 Example :
 > > > get _ cache _ key ( site _ domain = " example . com " , resource = " enterprise " )
 # Here is key format for above call
 # " site _ domain : example . com _ _ resource : enterprise "
 a54349175618ff1659dee0978e3149ca
 Arguments :
 * * kwargs : Key word arguments that need to be present in cache key .
 Returns :
 An MD5 encoded key uniquely identified by the key word arguments ."""

implementation:
key = '__' . join ( [ '{}:{}' . format ( item , value ) for item , value in iteritems ( kwargs ) ] )
return hashlib . md5 ( key . encode ( 'utf-8' ) ) . hexdigest ( )


signature:
def traverse_pagination ( response , endpoint ) :
    """Traverse a paginated API response .
 Extracts and concatenates " results " ( list of dict ) returned by DRF - powered
 APIs .
 Arguments :
 response ( Dict ) : Current response dict from service API
 endpoint ( slumber Resource object ) : slumber Resource object from edx - rest - api - client
 Returns :
 list of dict ."""

implementation:
results = response . get ( 'results' , [ ] )
next_page = response . get ( 'next' )
while next_page :
    querystring = parse_qs ( urlparse ( next_page ) . query , keep_blank_values = True )
    response = endpoint . get ( ** querystring )
    results += response . get ( 'results' , [ ] )
    next_page = response . get ( 'next' )
return results


signature:
def ungettext_min_max ( singular , plural , range_text , min_val , max_val ) :
    """Return grammatically correct , translated text based off of a minimum and maximum value .
 Example :
 min = 1 , max = 1 , singular = ' { } hour required for this course ' , plural = ' { } hours required for this course '
 output = ' 1 hour required for this course '
 min = 2 , max = 2 , singular = ' { } hour required for this course ' , plural = ' { } hours required for this course '
 output = ' 2 hours required for this course '
 min = 2 , max = 4 , range _ text = ' { } - { } hours required for this course '
 output = ' 2-4 hours required for this course '
 min = None , max = 2 , plural = ' { } hours required for this course '
 output = ' 2 hours required for this course '
 Expects ` ` range _ text ` ` to already have a translation function called on it .
 Returns :
 ` ` None ` ` if both of the input values are ` ` None ` ` .
 ` ` singular ` ` formatted if both are equal or one of the inputs , but not both , are ` ` None ` ` , and the value is 1.
 ` ` plural ` ` formatted if both are equal or one of its inputs , but not both , are ` ` None ` ` , and the value is > 1.
 ` ` range _ text ` ` formatted if min ! = max and both are valid values ."""

implementation:
if min_val is None and max_val is None :
    return None
if min_val == max_val or min_val is None or max_val is None : # pylint : disable = translation - of - non - string
    return ungettext ( singular , plural , min_val or max_val ) . format ( min_val or max_val )
return range_text . format ( min_val , max_val )


signature:
def format_price ( price , currency = '$' ) :
    """Format the price to have the appropriate currency and digits . .
 : param price : The price amount .
 : param currency : The currency for the price .
 : return : A formatted price string , i . e . ' $ 10 ' , ' $ 10.52 ' ."""

implementation:
if int ( price ) == price :
    return '{}{}' . format ( currency , int ( price ) )
return '{}{:0.2f}' . format ( currency , price )


signature:
def get_configuration_value_for_site ( site , key , default = None ) :
    """Get the site configuration value for a key , unless a site configuration does not exist for that site .
 Useful for testing when no Site Configuration exists in edx - enterprise or if a site in LMS doesn ' t have
 a configuration tied to it .
 : param site : A Site model object
 : param key : The name of the value to retrieve
 : param default : The default response if there ' s no key in site config or settings
 : return : The value located at that key in the site configuration or settings file ."""

implementation:
if hasattr ( site , 'configuration' ) :
    return site . configuration . get_value ( key , default )
return default


signature:
def get_configuration_value ( val_name , default = None , ** kwargs ) :
    """Get a configuration value , or fall back to ` ` default ` ` if it doesn ' t exist .
 Also takes a ` type ` argument to guide which particular upstream method to use when trying to retrieve a value .
 Current types include :
 - ` url ` to specifically get a URL ."""

implementation:
if kwargs . get ( 'type' ) == 'url' :
    return get_url ( val_name ) or default if callable ( get_url ) else default
return configuration_helpers . get_value ( val_name , default , ** kwargs ) if configuration_helpers else default


signature:
def get_request_value ( request , key , default = None ) :
    """Get the value in the request , either through query parameters or posted data , from a key .
 : param request : The request from which the value should be gotten .
 : param key : The key to use to get the desired value .
 : param default : The backup value to use in case the input key cannot help us get the value .
 : return : The value we ' re looking for ."""

implementation:
if request . method in [ 'GET' , 'DELETE' ] :
    return request . query_params . get ( key , request . data . get ( key , default ) )
return request . data . get ( key , request . query_params . get ( key , default ) )


signature:
def track_event ( user_id , event_name , properties ) :
    """Emit a track event to segment ( and forwarded to GA ) for some parts of the Enterprise workflows ."""

implementation:
# Only call the endpoint if the import was successful .
if segment :
    segment . track ( user_id , event_name , properties )


signature:
def track_enrollment ( pathway , user_id , course_run_id , url_path = None ) :
    """Emit a track event for enterprise course enrollment ."""

implementation: track_event ( user_id , 'edx.bi.user.enterprise.onboarding' , { 'pathway' : pathway , 'url_path' : url_path , 'course_run_id' : course_run_id , } )


signature:
def is_course_run_enrollable ( course_run ) :
    """Return true if the course run is enrollable , false otherwise .
 We look for the following criteria :
 - end is greater than now OR null
 - enrollment _ start is less than now OR null
 - enrollment _ end is greater than now OR null"""

implementation:
now = datetime . datetime . now ( pytz . UTC )
end = parse_datetime_handle_invalid ( course_run . get ( 'end' ) )
enrollment_start = parse_datetime_handle_invalid ( course_run . get ( 'enrollment_start' ) )
enrollment_end = parse_datetime_handle_invalid ( course_run . get ( 'enrollment_end' ) )
return ( not end or end > now ) and ( not enrollment_start or enrollment_start < now ) and ( not enrollment_end or enrollment_end > now )


signature:
def is_course_run_upgradeable ( course_run ) :
    """Return true if the course run has a verified seat with an unexpired upgrade deadline , false otherwise ."""

implementation:
now = datetime . datetime . now ( pytz . UTC )
for seat in course_run . get ( 'seats' , [ ] ) :
    if seat . get ( 'type' ) == 'verified' :
        upgrade_deadline = parse_datetime_handle_invalid ( seat . get ( 'upgrade_deadline' ) )
        return not upgrade_deadline or upgrade_deadline > now
return False


signature:
def get_closest_course_run ( course_runs ) :
    """Return course run with start date closest to now ."""

implementation:
if len ( course_runs ) == 1 :
    return course_runs [ 0 ]
now = datetime . datetime . now ( pytz . UTC )
# course runs with no start date should be considered last .
never = now - datetime . timedelta ( days = 3650 )
return min ( course_runs , key = lambda x : abs ( get_course_run_start ( x , never ) - now ) )


signature:
def get_active_course_runs ( course , users_all_enrolled_courses ) :
    """Return active course runs ( user is enrolled in ) of the given course .
 This function will return the course _ runs of ' course ' which have
 active enrollment by looking into ' users _ all _ enrolled _ courses '"""

implementation:
# User ' s all course _ run ids in which he has enrolled .
enrolled_course_run_ids = [ enrolled_course_run [ 'course_details' ] [ 'course_id' ] for enrolled_course_run in users_all_enrolled_courses if enrolled_course_run [ 'is_active' ] and enrolled_course_run . get ( 'course_details' ) ]
return [ course_run for course_run in course [ 'course_runs' ] if course_run [ 'key' ] in enrolled_course_run_ids ]


signature:
def get_current_course_run ( course , users_active_course_runs ) :
    """Return the current course run on the following conditions .
 - If user has active course runs ( already enrolled ) then return course run with closest start date
 Otherwise it will check the following logic :
 - Course run is enrollable ( see is _ course _ run _ enrollable )
 - Course run has a verified seat and the upgrade deadline has not expired .
 - Course run start date is closer to now than any other enrollable / upgradeable course runs .
 - If no enrollable / upgradeable course runs , return course run with most recent start date ."""

implementation:
current_course_run = None
filtered_course_runs = [ ]
all_course_runs = course [ 'course_runs' ]
if users_active_course_runs :
    current_course_run = get_closest_course_run ( users_active_course_runs )
else :
    for course_run in all_course_runs :
        if is_course_run_enrollable ( course_run ) and is_course_run_upgradeable ( course_run ) :
            filtered_course_runs . append ( course_run )
    if not filtered_course_runs : # Consider all runs if there were not any enrollable / upgradeable ones .
        filtered_course_runs = all_course_runs
    if filtered_course_runs :
        current_course_run = get_closest_course_run ( filtered_course_runs )
return current_course_run


signature:
def strip_html_tags ( text , allowed_tags = None ) :
    """Strip all tags from a string except those tags provided in ` allowed _ tags ` parameter .
 Args :
 text ( str ) : string to strip html tags from
 allowed _ tags ( list ) : allowed list of html tags
 Returns : a string without html tags"""

implementation:
if text is None :
    return
if allowed_tags is None :
    allowed_tags = ALLOWED_TAGS
return bleach . clean ( text , tags = allowed_tags , attributes = [ 'id' , 'class' , 'style' , 'href' , 'title' ] , strip = True )


signature:
def parse_course_key ( course_identifier ) :
    """Return the serialized course key given either a course run ID or course key ."""

implementation:
try :
    course_run_key = CourseKey . from_string ( course_identifier )
except InvalidKeyError : # Assume we already have a course key .
    return course_identifier
return quote_plus ( ' ' . join ( [ course_run_key . org , course_run_key . course ] ) )


signature:
def create_roles ( apps , schema_editor ) :
    """Create the enterprise roles if they do not already exist ."""

implementation:
SystemWideEnterpriseRole = apps . get_model ( 'enterprise' , 'SystemWideEnterpriseRole' )
SystemWideEnterpriseRole . objects . update_or_create ( name = ENTERPRISE_OPERATOR_ROLE )


signature:
def delete_roles ( apps , schema_editor ) :
    """Delete the enterprise roles ."""

implementation:
SystemWideEnterpriseRole = apps . get_model ( 'enterprise' , 'SystemWideEnterpriseRole' )
SystemWideEnterpriseRole . objects . filter ( name__in = [ ENTERPRISE_OPERATOR_ROLE ] ) . delete ( )


signature:
def lrs ( self ) :
    """LRS client instance to be used for sending statements ."""

implementation: return RemoteLRS ( version = self . lrs_configuration . version , endpoint = self . lrs_configuration . endpoint , auth = self . lrs_configuration . authorization_header , )


signature:
def save_statement ( self , statement ) :
    """Save xAPI statement .
 Arguments :
 statement ( EnterpriseStatement ) : xAPI Statement to send to the LRS .
 Raises :
 ClientError : If xAPI statement fails to save ."""

implementation:
response = self . lrs . save_statement ( statement )
if not response :
    raise ClientError ( 'EnterpriseXAPIClient request failed.' )


signature:
def get_learner_data_records ( self , enterprise_enrollment , completed_date = None , grade = None , is_passing = False ) :
    """Return a SapSuccessFactorsLearnerDataTransmissionAudit with the given enrollment and course completion data .
 If completed _ date is None and the learner isn ' t passing , then course completion has not been met .
 If no remote ID can be found , return None ."""

implementation:
completed_timestamp = None
course_completed = False
if completed_date is not None :
    completed_timestamp = parse_datetime_to_epoch_millis ( completed_date )
    course_completed = is_passing
sapsf_user_id = enterprise_enrollment . enterprise_customer_user . get_remote_id ( )
if sapsf_user_id is not None :
    SapSuccessFactorsLearnerDataTransmissionAudit = apps . get_model ( # pylint : disable = invalid - name
    'sap_success_factors' , 'SapSuccessFactorsLearnerDataTransmissionAudit' )
    # We return two records here , one with the course key and one with the course run id , to account for
    # uncertainty about the type of content ( course vs . course run ) that was sent to the integrated channel .
    return [ SapSuccessFactorsLearnerDataTransmissionAudit ( enterprise_course_enrollment_id = enterprise_enrollment . id , sapsf_user_id = sapsf_user_id , course_id = parse_course_key ( enterprise_enrollment . course_id ) , course_completed = course_completed , completed_timestamp = completed_timestamp , grade = grade , ) , SapSuccessFactorsLearnerDataTransmissionAudit ( enterprise_course_enrollment_id = enterprise_enrollment . id , sapsf_user_id = sapsf_user_id , course_id = enterprise_enrollment . course_id , course_completed = course_completed , completed_timestamp = completed_timestamp , grade = grade , ) , ]
else :
    LOGGER . debug ( 'No learner data was sent for user [%s] because an SAP SuccessFactors user ID could not be found.' , enterprise_enrollment . enterprise_customer_user . username )


signature:
def unlink_learners ( self ) :
    """Iterate over each learner and unlink inactive SAP channel learners .
 This method iterates over each enterprise learner and unlink learner
 from the enterprise if the learner is marked inactive in the related
 integrated channel ."""

implementation:
sap_inactive_learners = self . client . get_inactive_sap_learners ( )
enterprise_customer = self . enterprise_configuration . enterprise_customer
if not sap_inactive_learners :
    LOGGER . info ( 'Enterprise customer {%s} has no SAPSF inactive learners' , enterprise_customer . name )
    return
provider_id = enterprise_customer . identity_provider
tpa_provider = get_identity_provider ( provider_id )
if not tpa_provider :
    LOGGER . info ( 'Enterprise customer {%s} has no associated identity provider' , enterprise_customer . name )
    return None
for sap_inactive_learner in sap_inactive_learners :
    social_auth_user = get_user_from_social_auth ( tpa_provider , sap_inactive_learner [ 'studentID' ] )
    if not social_auth_user :
        continue
    try : # Unlink user email from related Enterprise Customer
        EnterpriseCustomerUser . objects . unlink_user ( enterprise_customer = enterprise_customer , user_email = social_auth_user . email , )
    except ( EnterpriseCustomerUser . DoesNotExist , PendingEnterpriseCustomerUser . DoesNotExist ) :
        LOGGER . info ( 'Learner with email {%s} is not associated with Enterprise Customer {%s}' , social_auth_user . email , enterprise_customer . name )


signature:
def has_implicit_access_to_dashboard ( user , obj ) : # pylint : disable = unused - argument
    """Check that if request user has implicit access to ` ENTERPRISE _ DASHBOARD _ ADMIN _ ROLE ` feature role .
 Returns :
 boolean : whether the request user has access or not"""

implementation:
request = get_request_or_stub ( )
decoded_jwt = get_decoded_jwt_from_request ( request )
return request_user_has_implicit_access_via_jwt ( decoded_jwt , ENTERPRISE_DASHBOARD_ADMIN_ROLE )


signature:
def has_implicit_access_to_catalog ( user , obj ) : # pylint : disable = unused - argument
    """Check that if request user has implicit access to ` ENTERPRISE _ CATALOG _ ADMIN _ ROLE ` feature role .
 Returns :
 boolean : whether the request user has access or not"""

implementation:
request = get_request_or_stub ( )
decoded_jwt = get_decoded_jwt_from_request ( request )
return request_user_has_implicit_access_via_jwt ( decoded_jwt , ENTERPRISE_CATALOG_ADMIN_ROLE , obj )


signature:
def has_implicit_access_to_enrollment_api ( user , obj ) : # pylint : disable = unused - argument
    """Check that if request user has implicit access to ` ENTERPRISE _ ENROLLMENT _ API _ ADMIN _ ROLE ` feature role .
 Returns :
 boolean : whether the request user has access or not"""

implementation:
request = get_request_or_stub ( )
decoded_jwt = get_decoded_jwt_from_request ( request )
return request_user_has_implicit_access_via_jwt ( decoded_jwt , ENTERPRISE_ENROLLMENT_API_ADMIN_ROLE , obj )


signature:
def transform_language_code ( code ) :
    """Transform ISO language code ( e . g . en - us ) to the language name expected by SAPSF ."""

implementation:
if code is None :
    return 'English'
components = code . split ( '-' , 2 )
language_code = components [ 0 ]
try :
    country_code = components [ 1 ]
except IndexError :
    country_code = '_'
language_family = SUCCESSFACTORS_OCN_LANGUAGE_CODES . get ( language_code )
if not language_family :
    return 'English'
return language_family . get ( country_code , language_family [ '_' ] )


signature:
def ecommerce_coupon_url ( self , instance ) :
    """Instance is EnterpriseCustomer . Return e - commerce coupon urls ."""

implementation:
if not instance . entitlement_id :
    return "N/A"
return format_html ( 'View coupon "{id}" details' , base_url = settings . ECOMMERCE_PUBLIC_URL_ROOT , id = instance . entitlement_id )


signature:
def dropHistoricalTable ( apps , schema_editor ) :
    """Drops the historical sap _ success _ factors table named herein ."""

implementation:
table_name = 'sap_success_factors_historicalsapsuccessfactorsenterprisecus80ad'
if table_name in connection . introspection . table_names ( ) :
    migrations . DeleteModel ( name = table_name , )


signature:
def handle ( self , * args , ** options ) :
    """Transmit the learner data for the EnterpriseCustomer ( s ) to the active integration channels ."""

implementation:
# Ensure that we were given an api _ user name , and that User exists .
api_username = options [ 'api_user' ]
try :
    User . objects . get ( username = api_username )
except User . DoesNotExist :
    raise CommandError ( _ ( 'A user with the username {username} was not found.' ) . format ( username = api_username ) )
# Transmit the learner data to each integrated channel
for integrated_channel in self . get_integrated_channels ( options ) :
    transmit_learner_data . delay ( api_username , integrated_channel . channel_code ( ) , integrated_channel . pk )


signature:
def _login ( self , email , password ) :
    """Login to pybotvac account using provided email and password .
 : param email : email for pybotvac account
 : param password : Password for pybotvac account
 : return :"""

implementation:
response = requests . post ( urljoin ( self . ENDPOINT , 'sessions' ) , json = { 'email' : email , 'password' : password , 'platform' : 'ios' , 'token' : binascii . hexlify ( os . urandom ( 64 ) ) . decode ( 'utf8' ) } , headers = self . _headers )
response . raise_for_status ( )
access_token = response . json ( ) [ 'access_token' ]
self . _headers [ 'Authorization' ] = 'Token token=%s' % access_token


signature:
def refresh_maps ( self ) :
    """Get information about maps of the robots .
 : return :"""

implementation:
for robot in self . robots :
    resp2 = ( requests . get ( urljoin ( self . ENDPOINT , 'users/me/robots/{}/maps' . format ( robot . serial ) ) , headers = self . _headers ) )
    resp2 . raise_for_status ( )
    self . _maps . update ( { robot . serial : resp2 . json ( ) } )


signature:
def refresh_robots ( self ) :
    """Get information about robots connected to account .
 : return :"""

implementation:
resp = requests . get ( urljoin ( self . ENDPOINT , 'dashboard' ) , headers = self . _headers )
resp . raise_for_status ( )
for robot in resp . json ( ) [ 'robots' ] :
    if robot [ 'mac_address' ] is None :
        continue
    # Ignore robots without mac - address
    try :
        self . _robots . add ( Robot ( name = robot [ 'name' ] , serial = robot [ 'serial' ] , secret = robot [ 'secret_key' ] , traits = robot [ 'traits' ] , endpoint = robot [ 'nucleo_url' ] ) )
    except requests . exceptions . HTTPError :
        print ( "Your '{}' robot is offline." . format ( robot [ 'name' ] ) )
        continue
self . refresh_persistent_maps ( )
for robot in self . _robots :
    robot . has_persistent_maps = robot . serial in self . _persistent_maps


signature:
def get_map_image ( url , dest_path = None ) :
    """Return a requested map from a robot .
 : return :"""

implementation:
image = requests . get ( url , stream = True , timeout = 10 )
if dest_path :
    image_url = url . rsplit ( '/' , 2 ) [ 1 ] + '-' + url . rsplit ( '/' , 1 ) [ 1 ]
    image_filename = image_url . split ( '?' ) [ 0 ]
    dest = os . path . join ( dest_path , image_filename )
    image . raise_for_status ( )
    with open ( dest , 'wb' ) as data :
        image . raw . decode_content = True
        shutil . copyfileobj ( image . raw , data )
return image . raw


signature:
def refresh_persistent_maps ( self ) :
    """Get information about persistent maps of the robots .
 : return :"""

implementation:
for robot in self . _robots :
    resp2 = ( requests . get ( urljoin ( self . ENDPOINT , 'users/me/robots/{}/persistent_maps' . format ( robot . serial ) ) , headers = self . _headers ) )
    resp2 . raise_for_status ( )
    self . _persistent_maps . update ( { robot . serial : resp2 . json ( ) } )


signature:
def _message ( self , json ) :
    """Sends message to robot with data from parameter ' json '
 : param json : dict containing data to send
 : return : server response"""

implementation:
cert_path = os . path . join ( os . path . dirname ( __file__ ) , 'cert' , 'neatocloud.com.crt' )
response = requests . post ( self . _url , json = json , verify = cert_path , auth = Auth ( self . serial , self . secret ) , headers = self . _headers )
response . raise_for_status ( )
return response


signature:
def add_edge_lengths ( g ) :
    """Add add the edge lengths as a : any : ` DiGraph < networkx . DiGraph > `
 for the graph .
 Uses the ` ` pos ` ` vertex property to get the location of each
 vertex . These are then used to calculate the length of an edge
 between two vertices .
 Parameters
 g : : any : ` networkx . DiGraph ` , : class : ` numpy . ndarray ` , dict , ` ` None ` ` , etc .
 Any object that networkx can turn into a
 : any : ` DiGraph < networkx . DiGraph > `
 Returns
 : class : ` . QueueNetworkDiGraph `
 Returns the a graph with the ` ` edge _ length ` ` edge property .
 Raises
 TypeError
 Raised when the parameter ` ` g ` ` is not of a type that can be
 made into a : any : ` networkx . DiGraph ` ."""

implementation:
g = _test_graph ( g )
g . new_edge_property ( 'edge_length' )
for e in g . edges ( ) :
    latlon1 = g . vp ( e [ 1 ] , 'pos' )
    latlon2 = g . vp ( e [ 0 ] , 'pos' )
    g . set_ep ( e , 'edge_length' , np . round ( _calculate_distance ( latlon1 , latlon2 ) , 3 ) )
return g


signature:
def _prepare_graph ( g , g_colors , q_cls , q_arg , adjust_graph ) :
    """Prepares a graph for use in : class : ` . QueueNetwork ` .
 This function is called by ` ` _ _ init _ _ ` ` in the
 : class : ` . QueueNetwork ` class . It creates the : class : ` . QueueServer `
 instances that sit on the edges , and sets various edge and node
 properties that are used when drawing the graph .
 Parameters
 g : : any : ` networkx . DiGraph ` , : class : ` numpy . ndarray ` , dict , ` ` None ` ` , etc .
 Any object that networkx can turn into a
 : any : ` DiGraph < networkx . DiGraph > `
 g _ colors : dict
 A dictionary of colors . The specific keys used are
 ` ` vertex _ color ` ` and ` ` vertex _ fill _ color ` ` for vertices that
 do not have any loops . Set : class : ` . QueueNetwork ` for the
 default values passed .
 q _ cls : dict
 A dictionary where the keys are integers that represent an edge
 type , and the values are : class : ` . QueueServer ` classes .
 q _ args : dict
 A dictionary where the keys are integers that represent an edge
 type , and the values are the arguments that are used when
 creating an instance of that : class : ` . QueueServer ` class .
 adjust _ graph : bool
 Specifies whether the graph will be adjusted using
 : func : ` . adjacency2graph ` .
 Returns
 g : : class : ` . QueueNetworkDiGraph `
 queues : list
 A list of : class : ` QueueServers < . QueueServer > ` where
 ` ` queues [ k ] ` ` is the ` ` QueueServer ` ` that sets on the edge with
 edge index ` ` k ` ` .
 Notes
 The graph ` ` g ` ` should have the ` ` edge _ type ` ` edge property map .
 If it does not then an ` ` edge _ type ` ` edge property is
 created and set to 1.
 The following properties are set by each queue : ` ` vertex _ color ` ` ,
 ` ` vertex _ fill _ color ` ` , ` ` vertex _ fill _ color ` ` , ` ` edge _ color ` ` .
 See : class : ` . QueueServer ` for more on setting these values .
 The following properties are assigned as a properties to the graph ;
 their default values for each edge or vertex is shown :
 * ` ` vertex _ pen _ width ` ` : ` ` 1 ` ` ,
 * ` ` vertex _ size ` ` : ` ` 8 ` ` ,
 * ` ` edge _ control _ points ` ` : ` ` [ ] ` `
 * ` ` edge _ marker _ size ` ` : ` ` 8 ` `
 * ` ` edge _ pen _ width ` ` : ` ` 1.25 ` `
 Raises
 TypeError
 Raised when the parameter ` ` g ` ` is not of a type that can be
 made into a : any : ` networkx . DiGraph ` ."""

implementation:
g = _test_graph ( g )
if adjust_graph :
    pos = nx . get_node_attributes ( g , 'pos' )
    ans = nx . to_dict_of_dicts ( g )
    g = adjacency2graph ( ans , adjust = 2 , is_directed = g . is_directed ( ) )
    g = QueueNetworkDiGraph ( g )
    if len ( pos ) > 0 :
        g . set_pos ( pos )
g . new_vertex_property ( 'vertex_color' )
g . new_vertex_property ( 'vertex_fill_color' )
g . new_vertex_property ( 'vertex_pen_width' )
g . new_vertex_property ( 'vertex_size' )
g . new_edge_property ( 'edge_control_points' )
g . new_edge_property ( 'edge_color' )
g . new_edge_property ( 'edge_marker_size' )
g . new_edge_property ( 'edge_pen_width' )
queues = _set_queues ( g , q_cls , q_arg , 'cap' in g . vertex_properties ( ) )
if 'pos' not in g . vertex_properties ( ) :
    g . set_pos ( )
for k , e in enumerate ( g . edges ( ) ) :
    g . set_ep ( e , 'edge_pen_width' , 1.25 )
    g . set_ep ( e , 'edge_marker_size' , 8 )
    if e [ 0 ] == e [ 1 ] :
        g . set_ep ( e , 'edge_color' , queues [ k ] . colors [ 'edge_loop_color' ] )
    else :
        g . set_ep ( e , 'edge_color' , queues [ k ] . colors [ 'edge_color' ] )
for v in g . nodes ( ) :
    g . set_vp ( v , 'vertex_pen_width' , 1 )
    g . set_vp ( v , 'vertex_size' , 8 )
    e = ( v , v )
    if g . is_edge ( e ) :
        g . set_vp ( v , 'vertex_color' , queues [ g . edge_index [ e ] ] . _current_color ( 2 ) )
        g . set_vp ( v , 'vertex_fill_color' , queues [ g . edge_index [ e ] ] . _current_color ( ) )
    else :
        g . set_vp ( v , 'vertex_color' , g_colors [ 'vertex_color' ] )
        g . set_vp ( v , 'vertex_fill_color' , g_colors [ 'vertex_fill_color' ] )
return g , queues


signature:
def desired_destination ( self , network , edge ) :
    """Returns the agents next destination given their current
 location on the network .
 An ` ` Agent ` ` chooses one of the out edges at random . The
 probability that the ` ` Agent ` ` will travel along a specific
 edge is specified in the : class : ` QueueNetwork ' s < . QueueNetwork > `
 transition matrix .
 Parameters
 network : : class : ` . QueueNetwork `
 The : class : ` . QueueNetwork ` where the Agent resides .
 edge : tuple
 A 4 - tuple indicating which edge this agent is located at .
 The first two slots indicate the current edge ' s source and
 target vertices , while the third slot indicates this edges
 ` ` edge _ index ` ` . The last slot indicates the edge type of
 that edge
 Returns
 out : int
 Returns an the edge index corresponding to the agents next
 edge to visit in the network .
 See Also
 : meth : ` . transitions ` : : class : ` QueueNetwork ' s < . QueueNetwork > `
 method that returns the transition probabilities for each
 edge in the graph ."""

implementation:
n = len ( network . out_edges [ edge [ 1 ] ] )
if n <= 1 :
    return network . out_edges [ edge [ 1 ] ] [ 0 ]
u = uniform ( )
pr = network . _route_probs [ edge [ 1 ] ]
k = _choice ( pr , u , n )
# _ choice returns an integer between 0 and n - 1 where the
# probability of k being selected is equal to pr [ k ] .
return network . out_edges [ edge [ 1 ] ] [ k ]


signature:
def desired_destination ( self , network , edge ) :
    """Returns the agents next destination given their current
 location on the network .
 ` ` GreedyAgents ` ` choose their next destination with - in the
 network by picking the adjacent queue with the fewest number of
 : class : ` Agents < . Agent > ` in the queue .
 Parameters
 network : : class : ` . QueueNetwork `
 The : class : ` . QueueNetwork ` where the Agent resides .
 edge : tuple
 A 4 - tuple indicating which edge this agent is located at .
 The first two slots indicate the current edge ' s source and
 target vertices , while the third slot indicates this edges
 ` ` edge _ index ` ` . The last slot indicates the edges edge
 type .
 Returns
 out : int
 Returns an the edge index corresponding to the agents next
 edge to visit in the network ."""

implementation:
adjacent_edges = network . out_edges [ edge [ 1 ] ]
d = _argmin ( [ network . edge2queue [ d ] . number_queued ( ) for d in adjacent_edges ] )
return adjacent_edges [ d ]


signature:
def _calculate_distance ( latlon1 , latlon2 ) :
    """Calculates the distance between two points on earth ."""

implementation:
lat1 , lon1 = latlon1
lat2 , lon2 = latlon2
dlon = lon2 - lon1
dlat = lat2 - lat1
R = 6371
# radius of the earth in kilometers
a = np . sin ( dlat / 2 ) ** 2 + np . cos ( lat1 ) * np . cos ( lat2 ) * ( np . sin ( dlon / 2 ) ) ** 2
c = 2 * np . pi * R * np . arctan2 ( np . sqrt ( a ) , np . sqrt ( 1 - a ) ) / 180
return c


signature:
def graph2dict ( g , return_dict_of_dict = True ) :
    """Takes a graph and returns an adjacency list .
 Parameters
 g : : any : ` networkx . DiGraph ` , : any : ` networkx . Graph ` , etc .
 Any object that networkx can turn into a
 : any : ` DiGraph < networkx . DiGraph > ` .
 return _ dict _ of _ dict : bool ( optional , default : ` ` True ` ` )
 Specifies whether this function will return a dict of dicts
 or a dict of lists .
 Returns
 adj : dict
 An adjacency representation of graph as a dictionary of
 dictionaries , where a key is the vertex index for a vertex
 ` ` v ` ` and the values are : class : ` dicts < . dict > ` with keys for
 the vertex index and values as edge properties .
 Examples
 > > > import queueing _ tool as qt
 > > > import networkx as nx
 > > > adj = { 0 : [ 1 , 2 ] , 1 : [ 0 ] , 2 : [ 0 , 3 ] , 3 : [ 2 ] }
 > > > g = nx . DiGraph ( adj )
 > > > qt . graph2dict ( g , return _ dict _ of _ dict = True )
 . . . # doctest : + NORMALIZE _ WHITESPACE
 {0 : { 1 : { } , 2 : { } } ,
 1 : { 0 : { } } ,
 2 : { 0 : { } , 3 : { } } ,
 3 : { 2 : { } } }
 > > > qt . graph2dict ( g , return _ dict _ of _ dict = False )
 {0 : [ 1 , 2 ] , 1 : [ 0 ] , 2 : [ 0 , 3 ] , 3 : [ 2 ] }"""

implementation:
if not isinstance ( g , nx . DiGraph ) :
    g = QueueNetworkDiGraph ( g )
dict_of_dicts = nx . to_dict_of_dicts ( g )
if return_dict_of_dict :
    return dict_of_dicts
else :
    return { k : list ( val . keys ( ) ) for k , val in dict_of_dicts . items ( ) }


signature:
def _matrix2dict ( matrix , etype = False ) :
    """Takes an adjacency matrix and returns an adjacency list ."""

implementation:
n = len ( matrix )
adj = { k : { } for k in range ( n ) }
for k in range ( n ) :
    for j in range ( n ) :
        if matrix [ k , j ] != 0 :
            adj [ k ] [ j ] = { } if not etype else matrix [ k , j ]
return adj


signature:
def _dict2dict ( adj_dict ) :
    """Takes a dictionary based representation of an adjacency list
 and returns a dict of dicts based representation ."""

implementation:
item = adj_dict . popitem ( )
adj_dict [ item [ 0 ] ] = item [ 1 ]
if not isinstance ( item [ 1 ] , dict ) :
    new_dict = { }
    for key , value in adj_dict . items ( ) :
        new_dict [ key ] = { v : { } for v in value }
    adj_dict = new_dict
return adj_dict


signature:
def _adjacency_adjust ( adjacency , adjust , is_directed ) :
    """Takes an adjacency list and returns a ( possibly ) modified
 adjacency list ."""

implementation:
for v , adj in adjacency . items ( ) :
    for properties in adj . values ( ) :
        if properties . get ( 'edge_type' ) is None :
            properties [ 'edge_type' ] = 1
if is_directed :
    if adjust == 2 :
        null_nodes = set ( )
        for k , adj in adjacency . items ( ) :
            if len ( adj ) == 0 :
                null_nodes . add ( k )
        for k , adj in adjacency . items ( ) :
            for v in adj . keys ( ) :
                if v in null_nodes :
                    adj [ v ] [ 'edge_type' ] = 0
    else :
        for k , adj in adjacency . items ( ) :
            if len ( adj ) == 0 :
                adj [ k ] = { 'edge_type' : 0 }
return adjacency


signature:
def adjacency2graph ( adjacency , edge_type = None , adjust = 1 , ** kwargs ) :
    """Takes an adjacency list , dict , or matrix and returns a graph .
 The purpose of this function is take an adjacency list ( or matrix )
 and return a : class : ` . QueueNetworkDiGraph ` that can be used with a
 : class : ` . QueueNetwork ` instance . The Graph returned has the
 ` ` edge _ type ` ` edge property set for each edge . Note that the graph may
 be altered .
 Parameters
 adjacency : dict or : class : ` ~ numpy . ndarray `
 An adjacency list as either a dict , or an adjacency matrix .
 adjust : int ` ` { 1 , 2 } ` ` ( optional , default : 1)
 Specifies what to do when the graph has terminal vertices
 ( nodes with no out - edges ) . Note that if ` ` adjust ` ` is not 2
 then it is assumed to be 1 . There are two choices :
 * ` ` adjust = 1 ` ` : A loop is added to each terminal node in the
 graph , and their ` ` edge _ type ` ` of that loop is set to 0.
 * ` ` adjust = 2 ` ` : All edges leading to terminal nodes have
 their ` ` edge _ type ` ` set to 0.
 * * kwargs :
 Unused .
 Returns
 out : : any : ` networkx . DiGraph `
 A directed graph with the ` ` edge _ type ` ` edge property .
 Raises
 TypeError
 Is raised if ` ` adjacency ` ` is not a dict or
 : class : ` ~ numpy . ndarray ` .
 Examples
 If terminal nodes are such that all in - edges have edge type ` ` 0 ` `
 then nothing is changed . However , if a node is a terminal node then
 a loop is added with edge type 0.
 > > > import queueing _ tool as qt
 > > > adj = {
 . . . 0 : { 1 : { } } ,
 . . . 1 : { 2 : { } ,
 . . . 3 : { 0 : { } } }
 > > > eTy = { 0 : { 1 : 1 } , 1 : { 2 : 2 , 3 : 4 } , 3 : { 0 : 1 } }
 > > > # A loop will be added to vertex 2
 > > > g = qt . adjacency2graph ( adj , edge _ type = eTy )
 > > > ans = qt . graph2dict ( g )
 > > > sorted ( ans . items ( ) ) # doctest : + NORMALIZE _ WHITESPACE
 [ ( 0 , { 1 : { ' edge _ type ' : 1 } } ) ,
 (1 , { 2 : { ' edge _ type ' : 2 } , 3 : { ' edge _ type ' : 4 } } ) ,
 (2 , { 2 : { ' edge _ type ' : 0 } } ) ,
 (3 , { 0 : { ' edge _ type ' : 1 } } ) ]
 You can use a dict of lists to represent the adjacency list .
 > > > adj = { 0 : [ 1 ] , 1 : [ 2 , 3 ] , 3 : [ 0 ] }
 > > > g = qt . adjacency2graph ( adj , edge _ type = eTy )
 > > > ans = qt . graph2dict ( g )
 > > > sorted ( ans . items ( ) ) # doctest : + NORMALIZE _ WHITESPACE
 [ ( 0 , { 1 : { ' edge _ type ' : 1 } } ) ,
 (1 , { 2 : { ' edge _ type ' : 2 } , 3 : { ' edge _ type ' : 4 } } ) ,
 (2 , { 2 : { ' edge _ type ' : 0 } } ) ,
 (3 , { 0 : { ' edge _ type ' : 1 } } ) ]
 Alternatively , you could have this function adjust the edges that
 lead to terminal vertices by changing their edge type to 0:
 > > > # The graph is unaltered
 > > > g = qt . adjacency2graph ( adj , edge _ type = eTy , adjust = 2)
 > > > ans = qt . graph2dict ( g )
 > > > sorted ( ans . items ( ) ) # doctest : + NORMALIZE _ WHITESPACE
 [ ( 0 , { 1 : { ' edge _ type ' : 1 } } ) ,
 (1 , { 2 : { ' edge _ type ' : 0 } , 3 : { ' edge _ type ' : 4 } } ) ,
 (3 , { 0 : { ' edge _ type ' : 1 } } ) ]"""

implementation:
if isinstance ( adjacency , np . ndarray ) :
    adjacency = _matrix2dict ( adjacency )
elif isinstance ( adjacency , dict ) :
    adjacency = _dict2dict ( adjacency )
else :
    msg = ( "If the adjacency parameter is supplied it must be a " "dict, or a numpy.ndarray." )
    raise TypeError ( msg )
if edge_type is None :
    edge_type = { }
else :
    if isinstance ( edge_type , np . ndarray ) :
        edge_type = _matrix2dict ( edge_type , etype = True )
    elif isinstance ( edge_type , dict ) :
        edge_type = _dict2dict ( edge_type )
for u , ty in edge_type . items ( ) :
    for v , et in ty . items ( ) :
        adjacency [ u ] [ v ] [ 'edge_type' ] = et
g = nx . from_dict_of_dicts ( adjacency , create_using = nx . DiGraph ( ) )
adjacency = nx . to_dict_of_dicts ( g )
adjacency = _adjacency_adjust ( adjacency , adjust , True )
return nx . from_dict_of_dicts ( adjacency , create_using = nx . DiGraph ( ) )


signature:
def get_edge_type ( self , edge_type ) :
    """Returns all edges with the specified edge type .
 Parameters
 edge _ type : int
 An integer specifying what type of edges to return .
 Returns
 out : list of 2 - tuples
 A list of 2 - tuples representing the edges in the graph
 with the specified edge type .
 Examples
 Lets get type 2 edges from the following graph
 > > > import queueing _ tool as qt
 > > > adjacency = {
 . . . 0 : { 1 : { ' edge _ type ' : 2 } } ,
 . . . 1 : { 2 : { ' edge _ type ' : 1 } ,
 . . . 3 : { ' edge _ type ' : 4 } } ,
 . . . 2 : { 0 : { ' edge _ type ' : 2 } } ,
 . . . 3 : { 3 : { ' edge _ type ' : 0 } }
 > > > G = qt . QueueNetworkDiGraph ( adjacency )
 > > > ans = G . get _ edge _ type ( 2)
 > > > ans . sort ( )
 > > > ans
 [ ( 0 , 1 ) , ( 2 , 0 ) ]"""

implementation:
edges = [ ]
for e in self . edges ( ) :
    if self . adj [ e [ 0 ] ] [ e [ 1 ] ] . get ( 'edge_type' ) == edge_type :
        edges . append ( e )
return edges


signature:
def draw_graph ( self , line_kwargs = None , scatter_kwargs = None , ** kwargs ) :
    """Draws the graph .
 Uses matplotlib , specifically
 : class : ` ~ matplotlib . collections . LineCollection ` and
 : meth : ` ~ matplotlib . axes . Axes . scatter ` . Gets the default
 keyword arguments for both methods by calling
 : meth : ` ~ . QueueNetworkDiGraph . lines _ scatter _ args ` first .
 Parameters
 line _ kwargs : dict ( optional , default : ` ` None ` ` )
 Any keyword arguments accepted by
 : class : ` ~ matplotlib . collections . LineCollection `
 scatter _ kwargs : dict ( optional , default : ` ` None ` ` )
 Any keyword arguments accepted by
 : meth : ` ~ matplotlib . axes . Axes . scatter ` .
 bgcolor : list ( optional , keyword only )
 A list with 4 floats representing a RGBA color . Defaults
 to ` ` [ 1 , 1 , 1 , 1 ] ` ` .
 figsize : tuple ( optional , keyword only , default : ` ` ( 7 , 7 ) ` ` )
 The width and height of the figure in inches .
 kwargs :
 Any keyword arguments used by
 : meth : ` ~ matplotlib . figure . Figure . savefig ` .
 Raises
 ImportError :
 If Matplotlib is not installed then an : exc : ` ImportError `
 is raised .
 Notes
 If the ` ` fname ` ` keyword is passed , then the figure is saved
 locally ."""

implementation:
if not HAS_MATPLOTLIB :
    raise ImportError ( "Matplotlib is required to draw the graph." )
fig = plt . figure ( figsize = kwargs . get ( 'figsize' , ( 7 , 7 ) ) )
ax = fig . gca ( )
mpl_kwargs = { 'line_kwargs' : line_kwargs , 'scatter_kwargs' : scatter_kwargs , 'pos' : kwargs . get ( 'pos' ) }
line_kwargs , scatter_kwargs = self . lines_scatter_args ( ** mpl_kwargs )
edge_collection = LineCollection ( ** line_kwargs )
ax . add_collection ( edge_collection )
ax . scatter ( ** scatter_kwargs )
if hasattr ( ax , 'set_facecolor' ) :
    ax . set_facecolor ( kwargs . get ( 'bgcolor' , [ 1 , 1 , 1 , 1 ] ) )
else :
    ax . set_axis_bgcolor ( kwargs . get ( 'bgcolor' , [ 1 , 1 , 1 , 1 ] ) )
ax . get_xaxis ( ) . set_visible ( False )
ax . get_yaxis ( ) . set_visible ( False )
if 'fname' in kwargs : # savefig needs a positional argument for some reason
    new_kwargs = { k : v for k , v in kwargs . items ( ) if k in SAVEFIG_KWARGS }
    fig . savefig ( kwargs [ 'fname' ] , ** new_kwargs )
else :
    plt . ion ( )
    plt . show ( )


signature:
def lines_scatter_args ( self , line_kwargs = None , scatter_kwargs = None , pos = None ) :
    """Returns the arguments used when plotting .
 Takes any keyword arguments for
 : class : ` ~ matplotlib . collections . LineCollection ` and
 : meth : ` ~ matplotlib . axes . Axes . scatter ` and returns two
 dictionaries with all the defaults set .
 Parameters
 line _ kwargs : dict ( optional , default : ` ` None ` ` )
 Any keyword arguments accepted by
 : class : ` ~ matplotlib . collections . LineCollection ` .
 scatter _ kwargs : dict ( optional , default : ` ` None ` ` )
 Any keyword arguments accepted by
 : meth : ` ~ matplotlib . axes . Axes . scatter ` .
 Returns
 tuple
 A 2 - tuple of dicts . The first entry is the keyword
 arguments for
 : class : ` ~ matplotlib . collections . LineCollection ` and the
 second is the keyword args for
 : meth : ` ~ matplotlib . axes . Axes . scatter ` .
 Notes
 If a specific keyword argument is not passed then the defaults
 are used ."""

implementation:
if pos is not None :
    self . set_pos ( pos )
elif self . pos is None :
    self . set_pos ( )
edge_pos = [ 0 for e in self . edges ( ) ]
for e in self . edges ( ) :
    ei = self . edge_index [ e ]
    edge_pos [ ei ] = ( self . pos [ e [ 0 ] ] , self . pos [ e [ 1 ] ] )
line_collecton_kwargs = { 'segments' : edge_pos , 'colors' : self . edge_color , 'linewidths' : ( 1 , ) , 'antialiaseds' : ( 1 , ) , 'linestyle' : 'solid' , 'transOffset' : None , 'cmap' : plt . cm . ocean_r , 'pickradius' : 5 , 'zorder' : 0 , 'facecolors' : None , 'norm' : None , 'offsets' : None , 'offset_position' : 'screen' , 'hatch' : None , }
scatter_kwargs_ = { 'x' : self . pos [ : , 0 ] , 'y' : self . pos [ : , 1 ] , 's' : 50 , 'c' : self . vertex_fill_color , 'alpha' : None , 'norm' : None , 'vmin' : None , 'vmax' : None , 'marker' : 'o' , 'zorder' : 2 , 'cmap' : plt . cm . ocean_r , 'linewidths' : 1 , 'edgecolors' : self . vertex_color , 'facecolors' : None , 'antialiaseds' : None , 'offset_position' : 'screen' , 'hatch' : None , }
line_kwargs = { } if line_kwargs is None else line_kwargs
scatter_kwargs = { } if scatter_kwargs is None else scatter_kwargs
for key , value in line_kwargs . items ( ) :
    if key in line_collecton_kwargs :
        line_collecton_kwargs [ key ] = value
for key , value in scatter_kwargs . items ( ) :
    if key in scatter_kwargs_ :
        scatter_kwargs_ [ key ] = value
return line_collecton_kwargs , scatter_kwargs_


signature:
def poisson_random_measure ( t , rate , rate_max ) :
    """A function that returns the arrival time of the next arrival for
 a Poisson random measure .
 Parameters
 t : float
 The start time from which to simulate the next arrival time .
 rate : function
 The * intensity function * for the measure , where ` ` rate ( t ) ` ` is
 the expected arrival rate at time ` ` t ` ` .
 rate _ max : float
 The maximum value of the ` ` rate ` ` function .
 Returns
 out : float
 The time of the next arrival .
 Notes
 This function returns the time of the next arrival , where the
 distribution of the number of arrivals between times : math : ` t ` and
 : math : ` t + s ` is Poisson with mean
 . . math : :
 \int_{t}^{t+s} dx \, r(x)
 where : math : ` r ( t ) ` is the supplied ` ` rate ` ` function . This function
 can only simulate processes that have bounded intensity functions .
 See chapter 6 of [ 3 ] _ for more on the mathematics behind Poisson
 random measures ; the book ' s publisher , Springer , has that chapter
 available online for free at ( ` pdf ` _ \ ) .
 A Poisson random measure is sometimes called a non - homogeneous
 Poisson process . A Poisson process is a special type of Poisson
 random measure .
 . . _ pdf : http : / / www . springer . com / cda / content / document / cda _ downloaddocument / 9780387878584 - c1 . pdf
 Examples
 Suppose you wanted to model the arrival process as a Poisson
 random measure with rate function : math : ` r ( t ) = 2 + \ sin ( 2 \ pi t ) ` .
 Then you could do so as follows :
 > > > import queueing _ tool as qt
 > > > import numpy as np
 > > > np . random . seed ( 10)
 > > > rate = lambda t : 2 + np . sin ( 2 * np . pi * t )
 > > > arr _ f = lambda t : qt . poisson _ random _ measure ( t , rate , 3)
 > > > arr _ f ( 1 ) # doctest : + ELLIPSIS
 1.491 . . .
 References
 . . [ 3 ] Cinlar , Erhan . * Probability and stochastics * . Graduate Texts in Mathematics . Vol . 261 . Springer , New York , 2011 . : doi : ` 10.1007/978-0-387-87859-1 `"""

implementation:
scale = 1.0 / rate_max
t = t + exponential ( scale )
while rate_max * uniform ( ) > rate ( t ) :
    t = t + exponential ( scale )
return t


signature:
def clear ( self ) :
    """Clears out the queue . Removes all arrivals , departures , and
 queued agents from the : class : ` . QueueServer ` , resets
 ` ` num _ arrivals ` ` , ` ` num _ departures ` ` , ` ` num _ system ` ` , and the clock to
 zero . It also clears any stored ` ` data ` ` and the server is then
 set to inactive ."""

implementation:
self . data = { }
self . _num_arrivals = 0
self . _oArrivals = 0
self . num_departures = 0
self . num_system = 0
self . _num_total = 0
self . _current_t = 0
self . _time = infty
self . _next_ct = 0
self . _active = False
self . queue = collections . deque ( )
inftyAgent = InftyAgent ( )
self . _arrivals = [ inftyAgent ]
self . _departures = [ inftyAgent ]


signature:
def _current_color ( self , which = 0 ) :
    """Returns a color for the queue .
 Parameters
 which : int ( optional , default : ` ` 0 ` ` )
 Specifies the type of color to return .
 Returns
 color : list
 Returns a RGBA color that is represented as a list with 4
 entries where each entry can be any floating point number
 between 0 and 1.
 * If ` ` which ` ` is 1 then it returns the color of the edge
 as if it were a self loop . This is specified in
 ` ` colors [ ' edge _ loop _ color ' ] ` ` .
 * If ` ` which ` ` is 2 then it returns the color of the vertex
 pen color ( defined as color / vertex _ color in
 : meth : ` . QueueNetworkDiGraph . graph _ draw ` ) . This is
 specified in ` ` colors [ ' vertex _ color ' ] ` ` .
 * If ` ` which ` ` is anything else , then it returns the a
 shade of the edge that is proportional to the number of
 agents in the system - - which includes those being
 servered and those waiting to be served . More agents
 correspond to darker edge colors . Uses
 ` ` colors [ ' vertex _ fill _ color ' ] ` ` if the queue sits on a
 loop , and ` ` colors [ ' edge _ color ' ] ` ` otherwise ."""

implementation:
if which == 1 :
    color = self . colors [ 'edge_loop_color' ]
elif which == 2 :
    color = self . colors [ 'vertex_color' ]
else :
    div = self . coloring_sensitivity * self . num_servers + 1.
    tmp = 1. - min ( self . num_system / div , 1 )
    if self . edge [ 0 ] == self . edge [ 1 ] :
        color = [ i * tmp for i in self . colors [ 'vertex_fill_color' ] ]
        color [ 3 ] = 1.0
    else :
        color = [ i * tmp for i in self . colors [ 'edge_color' ] ]
        color [ 3 ] = 1 / 2.
return color


signature:
def delay_service ( self , t = None ) :
    """Adds an extra service time to the next departing
 : class : ` Agent ' s < . Agent > ` service time .
 Parameters
 t : float ( optional )
 Specifies the departing time for the agent scheduled
 to depart next . If ` ` t ` ` is not given , then an additional
 service time is added to the next departing agent ."""

implementation:
if len ( self . _departures ) > 1 :
    agent = heappop ( self . _departures )
    if t is None :
        agent . _time = self . service_f ( agent . _time )
    else :
        agent . _time = t
    heappush ( self . _departures , agent )
    self . _update_time ( )


signature:
def fetch_data ( self , return_header = False ) :
    """Fetches data from the queue .
 Parameters
 return _ header : bool ( optonal , default : ` ` False ` ` )
 Determines whether the column headers are returned .
 Returns
 data : : class : ` ~ numpy . ndarray `
 A six column : class : ` ~ numpy . ndarray ` of all the data . The
 columns are :
 * 1st : The arrival time of an agent .
 * 2nd : The service start time of an agent .
 * 3rd : The departure time of an agent .
 * 4th : The length of the queue upon the agents arrival .
 * 5th : The total number of : class : ` Agents < . Agent > ` in the
 : class : ` . QueueServer ` .
 * 6th : The : class : ` QueueServer ' s < . QueueServer > ` edge index .
 headers : str ( optional )
 A comma seperated string of the column headers . Returns
 ` ` ' arrival , service , departure , num _ queued , num _ total , q _ id ' ` `"""

implementation:
qdata = [ ]
for d in self . data . values ( ) :
    qdata . extend ( d )
dat = np . zeros ( ( len ( qdata ) , 6 ) )
if len ( qdata ) > 0 :
    dat [ : , : 5 ] = np . array ( qdata )
    dat [ : , 5 ] = self . edge [ 2 ]
    dType = [ ( 'a' , float ) , ( 's' , float ) , ( 'd' , float ) , ( 'q' , float ) , ( 'n' , float ) , ( 'id' , float ) ]
    dat = np . array ( [ tuple ( d ) for d in dat ] , dtype = dType )
    dat = np . sort ( dat , order = 'a' )
    dat = np . array ( [ tuple ( d ) for d in dat ] )
if return_header :
    return dat , 'arrival,service,departure,num_queued,num_total,q_id'
return dat


signature:
def next_event ( self ) :
    """Simulates the queue forward one event .
 Use : meth : ` . simulate ` instead .
 Returns
 out : : class : ` . Agent ` ( sometimes )
 If the next event is a departure then the departing agent
 is returned , otherwise nothing is returned .
 See Also
 : meth : ` . simulate ` : Simulates the queue forward ."""

implementation:
if self . _departures [ 0 ] . _time < self . _arrivals [ 0 ] . _time :
    new_depart = heappop ( self . _departures )
    self . _current_t = new_depart . _time
    self . _num_total -= 1
    self . num_system -= 1
    self . num_departures += 1
    if self . collect_data and new_depart . agent_id in self . data :
        self . data [ new_depart . agent_id ] [ - 1 ] [ 2 ] = self . _current_t
    if len ( self . queue ) > 0 :
        agent = self . queue . popleft ( )
        if self . collect_data and agent . agent_id in self . data :
            self . data [ agent . agent_id ] [ - 1 ] [ 1 ] = self . _current_t
        agent . _time = self . service_f ( self . _current_t )
        agent . queue_action ( self , 1 )
        heappush ( self . _departures , agent )
    new_depart . queue_action ( self , 2 )
    self . _update_time ( )
    return new_depart
elif self . _arrivals [ 0 ] . _time < infty :
    arrival = heappop ( self . _arrivals )
    self . _current_t = arrival . _time
    if self . _active :
        self . _add_arrival ( )
    self . num_system += 1
    self . _num_arrivals += 1
    if self . collect_data :
        b = 0 if self . num_system <= self . num_servers else 1
        if arrival . agent_id not in self . data :
            self . data [ arrival . agent_id ] = [ [ arrival . _time , 0 , 0 , len ( self . queue ) + b , self . num_system ] ]
        else :
            self . data [ arrival . agent_id ] . append ( [ arrival . _time , 0 , 0 , len ( self . queue ) + b , self . num_system ] )
    arrival . queue_action ( self , 0 )
    if self . num_system <= self . num_servers :
        if self . collect_data :
            self . data [ arrival . agent_id ] [ - 1 ] [ 1 ] = arrival . _time
        arrival . _time = self . service_f ( arrival . _time )
        arrival . queue_action ( self , 1 )
        heappush ( self . _departures , arrival )
    else :
        self . queue . append ( arrival )
    self . _update_time ( )


signature:
def next_event_description ( self ) :
    """Returns an integer representing whether the next event is
 an arrival , a departure , or nothing .
 Returns
 out : int
 An integer representing whether the next event is an
 arrival or a departure : ` ` 1 ` ` corresponds to an arrival ,
 ` ` 2 ` ` corresponds to a departure , and ` ` 0 ` ` corresponds to
 nothing scheduled to occur ."""

implementation:
if self . _departures [ 0 ] . _time < self . _arrivals [ 0 ] . _time :
    return 2
elif self . _arrivals [ 0 ] . _time < infty :
    return 1
else :
    return 0


signature:
def set_num_servers ( self , n ) :
    """Change the number of servers in the queue to ` ` n ` ` .
 Parameters
 n : int or : const : ` numpy . infty `
 A positive integer ( or ` ` numpy . infty ` ` ) to set the number
 of queues in the system to .
 Raises
 TypeError
 If ` ` n ` ` is not an integer or positive infinity then this
 error is raised .
 ValueError
 If ` ` n ` ` is not positive ."""

implementation:
if not isinstance ( n , numbers . Integral ) and n is not infty :
    the_str = "n must be an integer or infinity.\n{0}"
    raise TypeError ( the_str . format ( str ( self ) ) )
elif n <= 0 :
    the_str = "n must be a positive integer or infinity.\n{0}"
    raise ValueError ( the_str . format ( str ( self ) ) )
else :
    self . num_servers = n


signature:
def simulate ( self , n = 1 , t = None , nA = None , nD = None ) :
    """This method simulates the queue forward for a specified
 amount of simulation time , or for a specific number of
 events .
 Parameters
 n : int ( optional , default : ` ` 1 ` ` )
 The number of events to simulate . If ` ` t ` ` , ` ` nA ` ` , and
 ` ` nD ` ` are not given then this parameter is used .
 t : float ( optional )
 The minimum amount of simulation time to simulate forward .
 nA : int ( optional )
 Simulate until ` ` nA ` ` additional arrivals are observed .
 nD : int ( optional )
 Simulate until ` ` nD ` ` additional departures are observed .
 Examples
 Before any simulations can take place the ` ` QueueServer ` ` must
 be activated :
 > > > import queueing _ tool as qt
 > > > import numpy as np
 > > > rate = lambda t : 2 + 16 * np . sin ( np . pi * t / 8 ) * * 2
 > > > arr = lambda t : qt . poisson _ random _ measure ( t , rate , 18)
 > > > ser = lambda t : t + np . random . gamma ( 4 , 0.1)
 > > > q = qt . QueueServer ( 5 , arrival _ f = arr , service _ f = ser , seed = 54)
 > > > q . set _ active ( )
 To simulate 50000 events do the following :
 > > > q . simulate ( 50000)
 > > > num _ events = q . num _ arrivals [ 0 ] + q . num _ departures
 > > > num _ events
 50000
 To simulate forward 75 time units , do the following :
 > > > t0 = q . time
 > > > q . simulate ( t = 75)
 > > > round ( float ( q . time - t0 ) , 1)
 75.1
 > > > q . num _ arrivals [ 1 ] + q . num _ departures - num _ events
 1597
 To simulate forward until 1000 new departures are observed run :
 > > > nA0 , nD0 = q . num _ arrivals [ 1 ] , q . num _ departures
 > > > q . simulate ( nD = 1000)
 > > > q . num _ departures - nD0 , q . num _ arrivals [ 1 ] - nA0
 (1000 , 983)
 To simulate until 1000 new arrivals are observed run :
 > > > nA0 , nD0 = q . num _ arrivals [ 1 ] , q . num _ departures
 > > > q . simulate ( nA = 1000)
 > > > q . num _ departures - nD0 , q . num _ arrivals [ 1 ] - nA0,
 (987 , 1000)"""

implementation:
if t is None and nD is None and nA is None :
    for dummy in range ( n ) :
        self . next_event ( )
elif t is not None :
    then = self . _current_t + t
    while self . _current_t < then and self . _time < infty :
        self . next_event ( )
elif nD is not None :
    num_departures = self . num_departures + nD
    while self . num_departures < num_departures and self . _time < infty :
        self . next_event ( )
elif nA is not None :
    num_arrivals = self . _oArrivals + nA
    while self . _oArrivals < num_arrivals and self . _time < infty :
        self . next_event ( )


signature:
def _get_queues ( g , queues , edge , edge_type ) :
    """Used to specify edge indices from different types of arguments ."""

implementation:
INT = numbers . Integral
if isinstance ( queues , INT ) :
    queues = [ queues ]
elif queues is None :
    if edge is not None :
        if isinstance ( edge , tuple ) :
            if isinstance ( edge [ 0 ] , INT ) and isinstance ( edge [ 1 ] , INT ) :
                queues = [ g . edge_index [ edge ] ]
        elif isinstance ( edge [ 0 ] , collections . Iterable ) :
            if np . array ( [ len ( e ) == 2 for e in edge ] ) . all ( ) :
                queues = [ g . edge_index [ e ] for e in edge ]
        else :
            queues = [ g . edge_index [ edge ] ]
    elif edge_type is not None :
        if isinstance ( edge_type , collections . Iterable ) :
            edge_type = set ( edge_type )
        else :
            edge_type = set ( [ edge_type ] )
        tmp = [ ]
        for e in g . edges ( ) :
            if g . ep ( e , 'edge_type' ) in edge_type :
                tmp . append ( g . edge_index [ e ] )
        queues = np . array ( tmp , int )
    if queues is None :
        queues = range ( g . number_of_edges ( ) )
return queues


signature:
def animate ( self , out = None , t = None , line_kwargs = None , scatter_kwargs = None , ** kwargs ) :
    """Animates the network as it ' s simulating .
 The animations can be saved to disk or viewed in interactive
 mode . Closing the window ends the animation if viewed in
 interactive mode . This method calls
 : meth : ` ~ matplotlib . axes . scatter ` , and
 : class : ` ~ matplotlib . collections . LineCollection ` , and any
 keyword arguments they accept can be passed to them .
 Parameters
 out : str ( optional )
 The location where the frames for the images will be saved .
 If this parameter is not given , then the animation is shown
 in interactive mode .
 t : float ( optional )
 The amount of simulation time to simulate forward . If
 given , and ` ` out ` ` is given , ` ` t ` ` is used instead of
 line _ kwargs : dict ( optional , default : None )
 Any keyword arguments accepted by
 : class : ` ~ matplotlib . collections . LineCollection ` .
 scatter _ kwargs : dict ( optional , default : None )
 Any keyword arguments accepted by
 : meth : ` ~ matplotlib . axes . Axes . scatter ` .
 bgcolor : list ( optional , keyword only )
 A list with 4 floats representing a RGBA color . The
 default is defined in ` ` self . colors [ ' bgcolor ' ] ` ` .
 figsize : tuple ( optional , keyword only , default : ` ` ( 7 , 7 ) ` ` )
 The width and height of the figure in inches .
 * * kwargs :
 This method calls
 : class : ` ~ matplotlib . animation . FuncAnimation ` and
 optionally : meth : ` . matplotlib . animation . FuncAnimation . save ` .
 Any keyword that can be passed to these functions are
 passed via ` ` kwargs ` ` .
 Notes
 There are several parameters automatically set and passed to
 matplotlib ' s : meth : ` ~ matplotlib . axes . Axes . scatter ` ,
 : class : ` ~ matplotlib . collections . LineCollection ` , and
 : class : ` ~ matplotlib . animation . FuncAnimation ` by default .
 These include :
 * : class : ` ~ matplotlib . animation . FuncAnimation ` : Uses the
 defaults for that function . Saving the animation is done
 by passing the ' filename ' keyword argument to this method .
 This method also accepts any keyword arguments accepted
 by : meth : ` ~ matplotlib . animation . FuncAnimation . save ` .
 * : class : ` ~ matplotlib . collections . LineCollection ` : The default
 arguments are taken from
 : meth : ` . QueueNetworkDiGraph . lines _ scatter _ args ` .
 * : meth : ` ~ matplotlib . axes . Axes . scatter ` : The default
 arguments are taken from
 : meth : ` . QueueNetworkDiGraph . lines _ scatter _ args ` .
 Raises
 QueueingToolError
 Will raise a : exc : ` . QueueingToolError ` if the
 ` ` QueueNetwork ` ` has not been initialized . Call
 : meth : ` . initialize ` before running .
 Examples
 This function works similarly to ` ` QueueNetwork ' s ` `
 : meth : ` . draw ` method .
 > > > import queueing _ tool as qt
 > > > g = qt . generate _ pagerank _ graph ( 100 , seed = 13)
 > > > net = qt . QueueNetwork ( g , seed = 13)
 > > > net . initialize ( )
 > > > net . animate ( figsize = ( 4 , 4 ) ) # doctest : + SKIP
 To stop the animation just close the window . If you want to
 write the animation to disk run something like the following :
 > > > kwargs = {
 . . . ' filename ' : ' test . mp4 ' ,
 . . . ' frames ' : 300,
 . . . ' fps ' : 30,
 . . . ' writer ' : ' mencoder ' ,
 . . . ' figsize ' : ( 4 , 4 ) ,
 . . . ' vertex _ size ' : 15
 > > > net . animate ( * * kwargs ) # doctest : + SKIP"""

implementation:
if not self . _initialized :
    msg = ( "Network has not been initialized. " "Call '.initialize()' first." )
    raise QueueingToolError ( msg )
if not HAS_MATPLOTLIB :
    msg = "Matplotlib is necessary to animate a simulation."
    raise ImportError ( msg )
self . _update_all_colors ( )
kwargs . setdefault ( 'bgcolor' , self . colors [ 'bgcolor' ] )
fig = plt . figure ( figsize = kwargs . get ( 'figsize' , ( 7 , 7 ) ) )
ax = fig . gca ( )
mpl_kwargs = { 'line_kwargs' : line_kwargs , 'scatter_kwargs' : scatter_kwargs , 'pos' : kwargs . get ( 'pos' ) }
line_args , scat_args = self . g . lines_scatter_args ( ** mpl_kwargs )
lines = LineCollection ( ** line_args )
lines = ax . add_collection ( lines )
scatt = ax . scatter ( ** scat_args )
t = np . infty if t is None else t
now = self . _t
def update ( frame_number ) :
    if t is not None :
        if self . _t > now + t :
            return False
    self . _simulate_next_event ( slow = True )
    lines . set_color ( line_args [ 'colors' ] )
    scatt . set_edgecolors ( scat_args [ 'edgecolors' ] )
    scatt . set_facecolor ( scat_args [ 'c' ] )
if hasattr ( ax , 'set_facecolor' ) :
    ax . set_facecolor ( kwargs [ 'bgcolor' ] )
else :
    ax . set_axis_bgcolor ( kwargs [ 'bgcolor' ] )
ax . get_xaxis ( ) . set_visible ( False )
ax . get_yaxis ( ) . set_visible ( False )
animation_args = { 'fargs' : None , 'event_source' : None , 'init_func' : None , 'frames' : None , 'blit' : False , 'interval' : 10 , 'repeat' : None , 'func' : update , 'repeat_delay' : None , 'fig' : fig , 'save_count' : None , }
for key , value in kwargs . items ( ) :
    if key in animation_args :
        animation_args [ key ] = value
animation = FuncAnimation ( ** animation_args )
if 'filename' not in kwargs :
    plt . ioff ( )
    plt . show ( )
else :
    save_args = { 'filename' : None , 'writer' : None , 'fps' : None , 'dpi' : None , 'codec' : None , 'bitrate' : None , 'extra_args' : None , 'metadata' : None , 'extra_anim' : None , 'savefig_kwargs' : None }
    for key , value in kwargs . items ( ) :
        if key in save_args :
            save_args [ key ] = value
    animation . save ( ** save_args )


signature:
def clear ( self ) :
    """Resets the queue to its initial state .
 The attributes ` ` t ` ` , ` ` num _ events ` ` , ` ` num _ agents ` ` are set to
 zero , : meth : ` . reset _ colors ` is called , and the
 : meth : ` . QueueServer . clear ` method is called for each queue in
 the network .
 Notes
 ` ` QueueNetwork ` ` must be re - initialized before any simulations
 can run ."""

implementation:
self . _t = 0
self . num_events = 0
self . num_agents = np . zeros ( self . nE , int )
self . _fancy_heap = PriorityQueue ( )
self . _prev_edge = None
self . _initialized = False
self . reset_colors ( )
for q in self . edge2queue :
    q . clear ( )


signature:
def clear_data ( self , queues = None , edge = None , edge_type = None ) :
    """Clears data from all queues .
 If none of the parameters are given then every queue ' s data is
 cleared .
 Parameters
 queues : int or an iterable of int ( optional )
 The edge index ( or an iterable of edge indices ) identifying
 the : class : ` QueueServer ( s ) < . QueueServer > ` whose data will
 be cleared .
 edge : 2 - tuple of int or * array _ like * ( optional )
 Explicitly specify which queues ' data to clear . Must be
 either :
 * A 2 - tuple of the edge ' s source and target vertex
 indices , or
 * An iterable of 2 - tuples of the edge ' s source and
 target vertex indices .
 edge _ type : int or an iterable of int ( optional )
 A integer , or a collection of integers identifying which
 edge types will have their data cleared ."""

implementation:
queues = _get_queues ( self . g , queues , edge , edge_type )
for k in queues :
    self . edge2queue [ k ] . data = { }


signature:
def copy ( self ) :
    """Returns a deep copy of itself ."""

implementation:
net = QueueNetwork ( None )
net . g = self . g . copy ( )
net . max_agents = copy . deepcopy ( self . max_agents )
net . nV = copy . deepcopy ( self . nV )
net . nE = copy . deepcopy ( self . nE )
net . num_agents = copy . deepcopy ( self . num_agents )
net . num_events = copy . deepcopy ( self . num_events )
net . _t = copy . deepcopy ( self . _t )
net . _initialized = copy . deepcopy ( self . _initialized )
net . _prev_edge = copy . deepcopy ( self . _prev_edge )
net . _blocking = copy . deepcopy ( self . _blocking )
net . colors = copy . deepcopy ( self . colors )
net . out_edges = copy . deepcopy ( self . out_edges )
net . in_edges = copy . deepcopy ( self . in_edges )
net . edge2queue = copy . deepcopy ( self . edge2queue )
net . _route_probs = copy . deepcopy ( self . _route_probs )
if net . _initialized :
    keys = [ q . _key ( ) for q in net . edge2queue if q . _time < np . infty ]
    net . _fancy_heap = PriorityQueue ( keys , net . nE )
return net


signature:
def draw ( self , update_colors = True , line_kwargs = None , scatter_kwargs = None , ** kwargs ) :
    """Draws the network . The coloring of the network corresponds
 to the number of agents at each queue .
 Parameters
 update _ colors : ` ` bool ` ` ( optional , default : ` ` True ` ` ) .
 Specifies whether all the colors are updated .
 line _ kwargs : dict ( optional , default : None )
 Any keyword arguments accepted by
 : class : ` ~ matplotlib . collections . LineCollection `
 scatter _ kwargs : dict ( optional , default : None )
 Any keyword arguments accepted by
 : meth : ` ~ matplotlib . axes . Axes . scatter ` .
 bgcolor : list ( optional , keyword only )
 A list with 4 floats representing a RGBA color . The
 default is defined in ` ` self . colors [ ' bgcolor ' ] ` ` .
 figsize : tuple ( optional , keyword only , default : ` ` ( 7 , 7 ) ` ` )
 The width and height of the canvas in inches .
 * * kwargs
 Any parameters to pass to
 : meth : ` . QueueNetworkDiGraph . draw _ graph ` .
 Notes
 This method relies heavily on
 : meth : ` . QueueNetworkDiGraph . draw _ graph ` . Also , there is a
 parameter that sets the background color of the canvas , which
 is the ` ` bgcolor ` ` parameter .
 Examples
 To draw the current state of the network , call :
 > > > import queueing _ tool as qt
 > > > g = qt . generate _ pagerank _ graph ( 100 , seed = 13)
 > > > net = qt . QueueNetwork ( g , seed = 13)
 > > > net . initialize ( 100)
 > > > net . simulate ( 1200)
 > > > net . draw ( ) # doctest : + SKIP
 If you specify a file name and location , the drawing will be
 saved to disk . For example , to save the drawing to the current
 working directory do the following :
 > > > net . draw ( fname = " state . png " , scatter _ kwargs = { ' s ' : 40 } ) # doctest : + SKIP
 . . figure : : current _ state1 . png
 : align : center
 The shade of each edge depicts how many agents are located at
 the corresponding queue . The shade of each vertex is determined
 by the total number of inbound agents . Although loops are not
 visible by default , the vertex that corresponds to a loop shows
 how many agents are in that loop .
 There are several additional parameters that can be passed - -
 all : meth : ` . QueueNetworkDiGraph . draw _ graph ` parameters are
 valid . For example , to show the edges as dashed lines do the
 following .
 > > > net . draw ( line _ kwargs = { ' linestyle ' : ' dashed ' } ) # doctest : + SKIP"""

implementation:
if not HAS_MATPLOTLIB :
    raise ImportError ( "matplotlib is necessary to draw the network." )
if update_colors :
    self . _update_all_colors ( )
if 'bgcolor' not in kwargs :
    kwargs [ 'bgcolor' ] = self . colors [ 'bgcolor' ]
self . g . draw_graph ( line_kwargs = line_kwargs , scatter_kwargs = scatter_kwargs , ** kwargs )


signature:
def get_agent_data ( self , queues = None , edge = None , edge_type = None , return_header = False ) :
    """Gets data from queues and organizes it by agent .
 If none of the parameters are given then data from every
 : class : ` . QueueServer ` is retrieved .
 Parameters
 queues : int or * array _ like * ( optional )
 The edge index ( or an iterable of edge indices ) identifying
 the : class : ` QueueServer ( s ) < . QueueServer > ` whose data will
 be retrieved .
 edge : 2 - tuple of int or * array _ like * ( optional )
 Explicitly specify which queues to retrieve agent data
 from . Must be either :
 * A 2 - tuple of the edge ' s source and target vertex
 indices , or
 * An iterable of 2 - tuples of the edge ' s source and
 target vertex indices .
 edge _ type : int or an iterable of int ( optional )
 A integer , or a collection of integers identifying which
 edge types to retrieve agent data from .
 return _ header : bool ( optonal , default : False )
 Determines whether the column headers are returned .
 Returns
 dict
 Returns a ` ` dict ` ` where the keys are the
 : class : ` Agent ' s < . Agent > ` ` ` agent _ id ` ` and the values are
 : class : ` ndarrays < ~ numpy . ndarray > ` for that
 : class : ` Agent ' s < . Agent > ` data . The columns of this array
 are as follows :
 * First : The arrival time of an agent .
 * Second : The service start time of an agent .
 * Third : The departure time of an agent .
 * Fourth : The length of the queue upon the agents arrival .
 * Fifth : The total number of : class : ` Agents < . Agent > ` in the
 : class : ` . QueueServer ` .
 * Sixth : the : class : ` QueueServer ' s < . QueueServer > ` id
 ( its edge index ) .
 headers : str ( optional )
 A comma seperated string of the column headers . Returns
 ` ` ' arrival , service , departure , num _ queued , num _ total , q _ id ' ` `"""

implementation:
queues = _get_queues ( self . g , queues , edge , edge_type )
data = { }
for qid in queues :
    for agent_id , dat in self . edge2queue [ qid ] . data . items ( ) :
        datum = np . zeros ( ( len ( dat ) , 6 ) )
        datum [ : , : 5 ] = np . array ( dat )
        datum [ : , 5 ] = qid
        if agent_id in data :
            data [ agent_id ] = np . vstack ( ( data [ agent_id ] , datum ) )
        else :
            data [ agent_id ] = datum
dType = [ ( 'a' , float ) , ( 's' , float ) , ( 'd' , float ) , ( 'q' , float ) , ( 'n' , float ) , ( 'id' , float ) ]
for agent_id , dat in data . items ( ) :
    datum = np . array ( [ tuple ( d ) for d in dat . tolist ( ) ] , dtype = dType )
    datum = np . sort ( datum , order = 'a' )
    data [ agent_id ] = np . array ( [ tuple ( d ) for d in datum ] )
if return_header :
    return data , 'arrival,service,departure,num_queued,num_total,q_id'
return data


signature:
def get_queue_data ( self , queues = None , edge = None , edge_type = None , return_header = False ) :
    """Gets data from all the queues .
 If none of the parameters are given then data from every
 : class : ` . QueueServer ` is retrieved .
 Parameters
 queues : int or an * array _ like * of int , ( optional )
 The edge index ( or an iterable of edge indices ) identifying
 the : class : ` QueueServer ( s ) < . QueueServer > ` whose data will
 be retrieved .
 edge : 2 - tuple of int or * array _ like * ( optional )
 Explicitly specify which queues to retrieve data from . Must
 be either :
 * A 2 - tuple of the edge ' s source and target vertex
 indices , or
 * An iterable of 2 - tuples of the edge ' s source and
 target vertex indices .
 edge _ type : int or an iterable of int ( optional )
 A integer , or a collection of integers identifying which
 edge types to retrieve data from .
 return _ header : bool ( optonal , default : False )
 Determines whether the column headers are returned .
 Returns
 out : : class : ` ~ numpy . ndarray `
 * 1st : The arrival time of an agent .
 * 2nd : The service start time of an agent .
 * 3rd : The departure time of an agent .
 * 4th : The length of the queue upon the agents arrival .
 * 5th : The total number of : class : ` Agents < . Agent > ` in the
 : class : ` . QueueServer ` .
 * 6th : The : class : ` QueueServer ' s < . QueueServer > ` edge index .
 out : str ( optional )
 A comma seperated string of the column headers . Returns
 ` ` ' arrival , service , departure , num _ queued , num _ total , q _ id ' ` ` `
 Examples
 Data is not collected by default . Before simulating , by sure to
 turn it on ( as well as initialize the network ) . The following
 returns data from queues with ` ` edge _ type ` ` 1 or 3:
 > > > import queueing _ tool as qt
 > > > g = qt . generate _ pagerank _ graph ( 100 , seed = 13)
 > > > net = qt . QueueNetwork ( g , seed = 13)
 > > > net . start _ collecting _ data ( )
 > > > net . initialize ( 10)
 > > > net . simulate ( 2000)
 > > > data = net . get _ queue _ data ( edge _ type = ( 1 , 3 ) )
 To get data from an edge connecting two vertices do the
 following :
 > > > data = net . get _ queue _ data ( edge = ( 1 , 50 ) )
 To get data from several edges do the following :
 > > > data = net . get _ queue _ data ( edge = [ ( 1 , 50 ) , ( 10 , 91 ) , ( 99 , 99 ) ] )
 You can specify the edge indices as well :
 > > > data = net . get _ queue _ data ( queues = ( 20 , 14 , 0 , 4 ) )"""

implementation:
queues = _get_queues ( self . g , queues , edge , edge_type )
data = np . zeros ( ( 0 , 6 ) )
for q in queues :
    dat = self . edge2queue [ q ] . fetch_data ( )
    if len ( dat ) > 0 :
        data = np . vstack ( ( data , dat ) )
if return_header :
    return data , 'arrival,service,departure,num_queued,num_total,q_id'
return data


signature:
def initialize ( self , nActive = 1 , queues = None , edges = None , edge_type = None ) :
    """Prepares the ` ` QueueNetwork ` ` for simulation .
 Each : class : ` . QueueServer ` in the network starts inactive ,
 which means they do not accept arrivals from outside the
 network , and they have no agents in their system . This method
 sets queues to active , which then allows agents to arrive from
 outside the network .
 Parameters
 nActive : int ( optional , default : ` ` 1 ` ` )
 The number of queues to set as active . The queues are
 selected randomly .
 queues : int * array _ like * ( optional )
 The edge index ( or an iterable of edge indices ) identifying
 the : class : ` QueueServer ( s ) < . QueueServer > ` to make active by .
 edges : 2 - tuple of int or * array _ like * ( optional )
 Explicitly specify which queues to make active . Must be
 either :
 * A 2 - tuple of the edge ' s source and target vertex
 indices , or
 * An iterable of 2 - tuples of the edge ' s source and
 target vertex indices .
 edge _ type : int or an iterable of int ( optional )
 A integer , or a collection of integers identifying which
 edge types will be set active .
 Raises
 ValueError
 If ` ` queues ` ` , ` ` egdes ` ` , and ` ` edge _ type ` ` are all ` ` None ` `
 and ` ` nActive ` ` is an integer less than 1
 : exc : ` ~ ValueError ` is raised .
 TypeError
 If ` ` queues ` ` , ` ` egdes ` ` , and ` ` edge _ type ` ` are all ` ` None ` `
 and ` ` nActive ` ` is not an integer then a : exc : ` ~ TypeError `
 is raised .
 QueueingToolError
 Raised if all the queues specified are
 : class : ` NullQueues < . NullQueue > ` .
 Notes
 : class : ` NullQueues < . NullQueue > ` cannot be activated , and are
 sifted out if they are specified . More specifically , every edge
 with edge type 0 is sifted out ."""

implementation:
if queues is None and edges is None and edge_type is None :
    if nActive >= 1 and isinstance ( nActive , numbers . Integral ) :
        qs = [ q . edge [ 2 ] for q in self . edge2queue if q . edge [ 3 ] != 0 ]
        n = min ( nActive , len ( qs ) )
        queues = np . random . choice ( qs , size = n , replace = False )
    elif not isinstance ( nActive , numbers . Integral ) :
        msg = "If queues is None, then nActive must be an integer."
        raise TypeError ( msg )
    else :
        msg = ( "If queues is None, then nActive must be a " "positive int." )
        raise ValueError ( msg )
else :
    queues = _get_queues ( self . g , queues , edges , edge_type )
queues = [ e for e in queues if self . edge2queue [ e ] . edge [ 3 ] != 0 ]
if len ( queues ) == 0 :
    raise QueueingToolError ( "There were no queues to initialize." )
if len ( queues ) > self . max_agents :
    queues = queues [ : self . max_agents ]
for ei in queues :
    self . edge2queue [ ei ] . set_active ( )
    self . num_agents [ ei ] = self . edge2queue [ ei ] . _num_total
keys = [ q . _key ( ) for q in self . edge2queue if q . _time < np . infty ]
self . _fancy_heap = PriorityQueue ( keys , self . nE )
self . _initialized = True


signature:
def next_event_description ( self ) :
    """Returns whether the next event is an arrival or a departure
 and the queue the event is accuring at .
 Returns
 des : str
 Indicates whether the next event is an arrival , a
 departure , or nothing ; returns ` ` ' Arrival ' ` ` ,
 ` ` ' Departure ' ` ` , or ` ` ' Nothing ' ` ` .
 edge : int or ` ` None ` `
 The edge index of the edge that this event will occur at .
 If there are no events then ` ` None ` ` is returned ."""

implementation:
if self . _fancy_heap . size == 0 :
    event_type = 'Nothing'
    edge_index = None
else :
    s = [ q . _key ( ) for q in self . edge2queue ]
    s . sort ( )
    e = s [ 0 ] [ 1 ]
    q = self . edge2queue [ e ]
    event_type = 'Arrival' if q . next_event_description ( ) == 1 else 'Departure'
    edge_index = q . edge [ 2 ]
return event_type , edge_index


signature:
def reset_colors ( self ) :
    """Resets all edge and vertex colors to their default values ."""

implementation:
for k , e in enumerate ( self . g . edges ( ) ) :
    self . g . set_ep ( e , 'edge_color' , self . edge2queue [ k ] . colors [ 'edge_color' ] )
for v in self . g . nodes ( ) :
    self . g . set_vp ( v , 'vertex_fill_color' , self . colors [ 'vertex_fill_color' ] )


signature:
def set_transitions ( self , mat ) :
    """Change the routing transitions probabilities for the
 network .
 Parameters
 mat : dict or : class : ` ~ numpy . ndarray `
 A transition routing matrix or transition dictionary . If
 passed a dictionary , the keys are source vertex indices and
 the values are dictionaries with target vertex indicies
 as the keys and the probabilities of routing from the
 source to the target as the values .
 Raises
 ValueError
 A : exc : ` . ValueError ` is raised if : the keys in the dict
 don ' t match with a vertex index in the graph ; or if the
 : class : ` ~ numpy . ndarray ` is passed with the wrong shape ,
 must be ( ` ` num _ vertices ` ` , ` ` num _ vertices ` ` ) ; or the values
 passed are not probabilities ( for each vertex they are
 positive and sum to 1 ) ;
 TypeError
 A : exc : ` . TypeError ` is raised if mat is not a dict or
 : class : ` ~ numpy . ndarray ` .
 Examples
 The default transition matrix is every out edge being equally
 likely :
 > > > import queueing _ tool as qt
 > > > adjacency = {
 . . . 0 : [ 2 ] ,
 . . . 1 : [ 2 , 3 ] ,
 . . . 2 : [ 0 , 1 , 2 , 4 ] ,
 . . . 3 : [ 1 ] ,
 . . . 4 : [ 2 ] ,
 > > > g = qt . adjacency2graph ( adjacency )
 > > > net = qt . QueueNetwork ( g )
 > > > net . transitions ( False ) # doctest : + ELLIPSIS
 . . . # doctest : + NORMALIZE _ WHITESPACE
 {0 : { 2 : 1.0 } ,
 1 : { 2 : 0.5 , 3 : 0.5 } ,
 2 : { 0 : 0.25 , 1 : 0.25 , 2 : 0.25 , 4 : 0.25 } ,
 3 : { 1 : 1.0 } ,
 4 : { 2 : 1.0 } }
 If you want to change only one vertex ' s transition
 probabilities , you can do so with the following :
 > > > net . set _ transitions ( { 1 : { 2 : 0.75 , 3 : 0.25 } } )
 > > > net . transitions ( False ) # doctest : + ELLIPSIS
 . . . # doctest : + NORMALIZE _ WHITESPACE
 {0 : { 2 : 1.0 } ,
 1 : { 2 : 0.75 , 3 : 0.25 } ,
 2 : { 0 : 0.25 , 1 : 0.25 , 2 : 0.25 , 4 : 0.25 } ,
 3 : { 1 : 1.0 } ,
 4 : { 2 : 1.0 } }
 One can generate a transition matrix using
 : func : ` . generate _ transition _ matrix ` . You can change all
 transition probabilities with an : class : ` ~ numpy . ndarray ` :
 > > > mat = qt . generate _ transition _ matrix ( g , seed = 10)
 > > > net . set _ transitions ( mat )
 > > > net . transitions ( False ) # doctest : + ELLIPSIS
 . . . # doctest : + NORMALIZE _ WHITESPACE
 {0 : { 2 : 1.0 } ,
 1 : { 2 : 0.962 . . . , 3 : 0.037 . . . } ,
 2 : { 0 : 0.301 . . . , 1 : 0.353 . . . , 2 : 0.235 . . . , 4 : 0.108 . . . } ,
 3 : { 1 : 1.0 } ,
 4 : { 2 : 1.0 } }
 See Also
 : meth : ` . transitions ` : Return the current routing
 probabilities .
 : func : ` . generate _ transition _ matrix ` : Generate a random routing
 matrix ."""

implementation:
if isinstance ( mat , dict ) :
    for key , value in mat . items ( ) :
        probs = list ( value . values ( ) )
        if key not in self . g . node :
            msg = "One of the keys don't correspond to a vertex."
            raise ValueError ( msg )
        elif len ( self . out_edges [ key ] ) > 0 and not np . isclose ( sum ( probs ) , 1 ) :
            msg = "Sum of transition probabilities at a vertex was not 1."
            raise ValueError ( msg )
        elif ( np . array ( probs ) < 0 ) . any ( ) :
            msg = "Some transition probabilities were negative."
            raise ValueError ( msg )
        for k , e in enumerate ( sorted ( self . g . out_edges ( key ) ) ) :
            self . _route_probs [ key ] [ k ] = value . get ( e [ 1 ] , 0 )
elif isinstance ( mat , np . ndarray ) :
    non_terminal = np . array ( [ self . g . out_degree ( v ) > 0 for v in self . g . nodes ( ) ] )
    if mat . shape != ( self . nV , self . nV ) :
        msg = ( "Matrix is the wrong shape, should " "be {0} x {1}." ) . format ( self . nV , self . nV )
        raise ValueError ( msg )
    elif not np . allclose ( np . sum ( mat [ non_terminal , : ] , axis = 1 ) , 1 ) :
        msg = "Sum of transition probabilities at a vertex was not 1."
        raise ValueError ( msg )
    elif ( mat < 0 ) . any ( ) :
        raise ValueError ( "Some transition probabilities were negative." )
    for k in range ( self . nV ) :
        for j , e in enumerate ( sorted ( self . g . out_edges ( k ) ) ) :
            self . _route_probs [ k ] [ j ] = mat [ k , e [ 1 ] ]
else :
    raise TypeError ( "mat must be a numpy array or a dict." )


signature:
def show_active ( self , ** kwargs ) :
    """Draws the network , highlighting active queues .
 The colored vertices represent vertices that have at least one
 queue on an in - edge that is active . Dark edges represent
 queues that are active , light edges represent queues that are
 inactive .
 Parameters
 * * kwargs
 Any additional parameters to pass to : meth : ` . draw ` , and
 : meth : ` . QueueNetworkDiGraph . draw _ graph ` .
 Notes
 Active queues are : class : ` QueueServers < . QueueServer > ` that
 accept arrivals from outside the network . The colors are
 defined by the class attribute ` ` colors ` ` . The relevant keys
 are ` ` vertex _ active ` ` , ` ` vertex _ inactive ` ` , ` ` edge _ active ` ` ,
 and ` ` edge _ inactive ` ` ."""

implementation:
g = self . g
for v in g . nodes ( ) :
    self . g . set_vp ( v , 'vertex_color' , [ 0 , 0 , 0 , 0.9 ] )
    is_active = False
    my_iter = g . in_edges ( v ) if g . is_directed ( ) else g . out_edges ( v )
    for e in my_iter :
        ei = g . edge_index [ e ]
        if self . edge2queue [ ei ] . _active :
            is_active = True
            break
    if is_active :
        self . g . set_vp ( v , 'vertex_fill_color' , self . colors [ 'vertex_active' ] )
    else :
        self . g . set_vp ( v , 'vertex_fill_color' , self . colors [ 'vertex_inactive' ] )
for e in g . edges ( ) :
    ei = g . edge_index [ e ]
    if self . edge2queue [ ei ] . _active :
        self . g . set_ep ( e , 'edge_color' , self . colors [ 'edge_active' ] )
    else :
        self . g . set_ep ( e , 'edge_color' , self . colors [ 'edge_inactive' ] )
self . draw ( update_colors = False , ** kwargs )
self . _update_all_colors ( )


signature:
def show_type ( self , edge_type , ** kwargs ) :
    """Draws the network , highlighting queues of a certain type .
 The colored vertices represent self loops of type ` ` edge _ type ` ` .
 Dark edges represent queues of type ` ` edge _ type ` ` .
 Parameters
 edge _ type : int
 The type of vertices and edges to be shown .
 * * kwargs
 Any additional parameters to pass to : meth : ` . draw ` , and
 : meth : ` . QueueNetworkDiGraph . draw _ graph `
 Notes
 The colors are defined by the class attribute ` ` colors ` ` . The
 relevant colors are ` ` vertex _ active ` ` , ` ` vertex _ inactive ` ` ,
 ` ` vertex _ highlight ` ` , ` ` edge _ active ` ` , and ` ` edge _ inactive ` ` .
 Examples
 The following code highlights all edges with edge type ` ` 2 ` ` .
 If the edge is a loop then the vertex is highlighted as well .
 In this case all edges with edge type ` ` 2 ` ` happen to be loops .
 > > > import queueing _ tool as qt
 > > > g = qt . generate _ pagerank _ graph ( 100 , seed = 13)
 > > > net = qt . QueueNetwork ( g , seed = 13)
 > > > fname = ' edge _ type _ 2 . png '
 > > > net . show _ type ( 2 , fname = fname ) # doctest : + SKIP
 . . figure : : edge _ type _ 2-1 . png
 : align : center"""

implementation:
for v in self . g . nodes ( ) :
    e = ( v , v )
    if self . g . is_edge ( e ) and self . g . ep ( e , 'edge_type' ) == edge_type :
        ei = self . g . edge_index [ e ]
        self . g . set_vp ( v , 'vertex_fill_color' , self . colors [ 'vertex_highlight' ] )
        self . g . set_vp ( v , 'vertex_color' , self . edge2queue [ ei ] . colors [ 'vertex_color' ] )
    else :
        self . g . set_vp ( v , 'vertex_fill_color' , self . colors [ 'vertex_inactive' ] )
        self . g . set_vp ( v , 'vertex_color' , [ 0 , 0 , 0 , 0.9 ] )
for e in self . g . edges ( ) :
    if self . g . ep ( e , 'edge_type' ) == edge_type :
        self . g . set_ep ( e , 'edge_color' , self . colors [ 'edge_active' ] )
    else :
        self . g . set_ep ( e , 'edge_color' , self . colors [ 'edge_inactive' ] )
self . draw ( update_colors = False , ** kwargs )
self . _update_all_colors ( )


signature:
def simulate ( self , n = 1 , t = None ) :
    """Simulates the network forward .
 Simulates either a specific number of events or for a specified
 amount of simulation time .
 Parameters
 n : int ( optional , default : 1)
 The number of events to simulate . If ` ` t ` ` is not given
 then this parameter is used .
 t : float ( optional )
 The amount of simulation time to simulate forward . If
 given , ` ` t ` ` is used instead of ` ` n ` ` .
 Raises
 QueueingToolError
 Will raise a : exc : ` . QueueingToolError ` if the
 ` ` QueueNetwork ` ` has not been initialized . Call
 : meth : ` . initialize ` before calling this method .
 Examples
 Let ` ` net ` ` denote your instance of a ` ` QueueNetwork ` ` . Before
 you simulate , you need to initialize the network , which allows
 arrivals from outside the network . To initialize with 2 ( random
 chosen ) edges accepting arrivals run :
 > > > import queueing _ tool as qt
 > > > g = qt . generate _ pagerank _ graph ( 100 , seed = 50)
 > > > net = qt . QueueNetwork ( g , seed = 50)
 > > > net . initialize ( 2)
 To simulate the network 50000 events run :
 > > > net . num _ events
 > > > net . simulate ( 50000)
 > > > net . num _ events
 50000
 To simulate the network for at least 75 simulation time units
 run :
 > > > t0 = net . current _ time
 > > > net . simulate ( t = 75)
 > > > t1 = net . current _ time
 > > > t1 - t0 # doctest : + ELLIPSIS
 75 . . ."""

implementation:
if not self . _initialized :
    msg = ( "Network has not been initialized. " "Call '.initialize()' first." )
    raise QueueingToolError ( msg )
if t is None :
    for dummy in range ( n ) :
        self . _simulate_next_event ( slow = False )
else :
    now = self . _t
    while self . _t < now + t :
        self . _simulate_next_event ( slow = False )


signature:
def start_collecting_data ( self , queues = None , edge = None , edge_type = None ) :
    """Tells the queues to collect data on agents ' arrival , service
 start , and departure times .
 If none of the parameters are given then every
 : class : ` . QueueServer ` will start collecting data .
 Parameters
 queues : : any : ` int ` , * array _ like * ( optional )
 The edge index ( or an iterable of edge indices ) identifying
 the : class : ` QueueServer ( s ) < . QueueServer > ` that will start
 collecting data .
 edge : 2 - tuple of int or * array _ like * ( optional )
 Explicitly specify which queues will collect data . Must be
 either :
 * A 2 - tuple of the edge ' s source and target vertex
 indices , or
 * An iterable of 2 - tuples of the edge ' s source and
 target vertex indices .
 edge _ type : int or an iterable of int ( optional )
 A integer , or a collection of integers identifying which
 edge types will be set active ."""

implementation:
queues = _get_queues ( self . g , queues , edge , edge_type )
for k in queues :
    self . edge2queue [ k ] . collect_data = True