signature:
def constraint_matrices ( model , array_type = 'dense' , include_vars = False , zero_tol = 1e-6 ) :
    """Create a matrix representation of the problem .
 This is used for alternative solution approaches that do not use optlang .
 The function will construct the equality matrix , inequality matrix and
 bounds for the complete problem .
 Notes
 To accomodate non - zero equalities the problem will add the variable
 " const _ one " which is a variable that equals one .
 Arguments
 model : cobra . Model
 The model from which to obtain the LP problem .
 array _ type : string
 The type of array to construct . if ' dense ' , return a standard
 numpy . array , ' dok ' , or ' lil ' will construct a sparse array using
 scipy of the corresponding type and ' DataFrame ' will give a
 pandas ` DataFrame ` with metabolite indices and reaction columns .
 zero _ tol : float
 The zero tolerance used to judge whether two bounds are the same .
 Returns
 collections . namedtuple
 A named tuple consisting of 6 matrices and 2 vectors :
 - " equalities " is a matrix S such that S * vars = b . It includes a row
 for each constraint and one column for each variable .
 - " b " the right side of the equality equation such that S * vars = b .
 - " inequalities " is a matrix M such that lb < = M * vars < = ub .
 It contains a row for each inequality and as many columns as
 variables .
 - " bounds " is a compound matrix [ lb ub ] containing the lower and
 upper bounds for the inequality constraints in M .
 - " variable _ fixed " is a boolean vector indicating whether the variable
 at that index is fixed ( lower bound = = upper _ bound ) and
 is thus bounded by an equality constraint .
 - " variable _ bounds " is a compound matrix [ lb ub ] containing the
 lower and upper bounds for all variables ."""

implementation:
if array_type not in ( 'DataFrame' , 'dense' ) and not dok_matrix :
    raise ValueError ( 'Sparse matrices require scipy' )
array_builder = { 'dense' : np . array , 'dok' : dok_matrix , 'lil' : lil_matrix , 'DataFrame' : pd . DataFrame , } [ array_type ]
Problem = namedtuple ( "Problem" , [ "equalities" , "b" , "inequalities" , "bounds" , "variable_fixed" , "variable_bounds" ] )
equality_rows = [ ]
inequality_rows = [ ]
inequality_bounds = [ ]
b = [ ]
for const in model . constraints :
    lb = - np . inf if const . lb is None else const . lb
    ub = np . inf if const . ub is None else const . ub
    equality = ( ub - lb ) < zero_tol
    coefs = const . get_linear_coefficients ( model . variables )
    coefs = [ coefs [ v ] for v in model . variables ]
    if equality :
        b . append ( lb if abs ( lb ) > zero_tol else 0.0 )
        equality_rows . append ( coefs )
    else :
        inequality_rows . append ( coefs )
        inequality_bounds . append ( [ lb , ub ] )
var_bounds = np . array ( [ [ v . lb , v . ub ] for v in model . variables ] )
fixed = var_bounds [ : , 1 ] - var_bounds [ : , 0 ] < zero_tol
results = Problem ( equalities = array_builder ( equality_rows ) , b = np . array ( b ) , inequalities = array_builder ( inequality_rows ) , bounds = array_builder ( inequality_bounds ) , variable_fixed = np . array ( fixed ) , variable_bounds = array_builder ( var_bounds ) )
return results


signature:
def room ( model , solution = None , linear = False , delta = 0.03 , epsilon = 1E-03 ) :
    """Compute a single solution based on regulatory on / off minimization ( ROOM ) .
 Compute a new flux distribution that minimizes the number of active
 reactions needed to accommodate a previous reference solution .
 Regulatory on / off minimization ( ROOM ) is generally used to assess the
 impact of knock - outs . Thus the typical usage is to provide a wildtype flux
 distribution as reference and a model in knock - out state .
 Parameters
 model : cobra . Model
 The model state to compute a ROOM - based solution for .
 solution : cobra . Solution , optional
 A ( wildtype ) reference solution .
 linear : bool , optional
 Whether to use the linear ROOM formulation or not ( default False ) .
 delta : float , optional
 The relative tolerance range ( additive ) ( default 0.03 ) .
 epsilon : float , optional
 The absolute tolerance range ( multiplicative ) ( default 0.001 ) .
 Returns
 cobra . Solution
 A flux distribution with minimal active reaction changes compared to
 the reference .
 See Also
 add _ room : add ROOM constraints and objective"""

implementation:
with model :
    add_room ( model = model , solution = solution , linear = linear , delta = delta , epsilon = epsilon )
    solution = model . optimize ( )
return solution


signature:
def add_room ( model , solution = None , linear = False , delta = 0.03 , epsilon = 1E-03 ) :
    r"""Add constraints and objective for ROOM .
 This function adds variables and constraints for applying regulatory
 on / off minimization ( ROOM ) to the model .
 Parameters
 model : cobra . Model
 The model to add ROOM constraints and objective to .
 solution : cobra . Solution , optional
 A previous solution to use as a reference . If no solution is given ,
 one will be computed using pFBA .
 linear : bool , optional
 Whether to use the linear ROOM formulation or not ( default False ) .
 delta : float , optional
 The relative tolerance range which is additive in nature
 ( default 0.03 ) .
 epsilon : float , optional
 The absolute range of tolerance which is multiplicative
 ( default 0.001 ) .
 Notes
 The formulation used here is the same as stated in the original paper [ 1 ] _ .
 The mathematical expression is given below :
 minimize \ sum _ { i = 1 } ^ m y ^ i
 s . t . Sv = 0
 v _ min < = v < = v _ max
 v _ j = 0
 j ∈ A
 for 1 < = i < = m
 v _ i - y _ i ( v _ { max , i } - w _ i ^ u ) < = w _ i ^ u ( 1)
 v _ i - y _ i ( v _ { min , i } - w _ i ^ l ) < = w _ i ^ l ( 2)
 y _ i ∈ { 0,1 } ( 3)
 w _ i ^ u = w _ i + \ delta | w _ i | + \ epsilon
 w _ i ^ l = w _ i - \ delta | w _ i | - \ epsilon
 So , for the linear version of the ROOM , constraint ( 3 ) is relaxed to
 0 < = y _ i < = 1.
 See Also
 pfba : parsimonious FBA
 References
 . . [ 1 ] Tomer Shlomi , Omer Berkman and Eytan Ruppin , " Regulatory on / off
 minimization of metabolic flux changes after genetic perturbations " ,
 PNAS 2005 102 ( 21 ) 7695-7700 ; doi : 10.1073 / pnas . 0406346102"""

implementation:
if 'room_old_objective' in model . solver . variables :
    raise ValueError ( 'model is already adjusted for ROOM' )
# optimizes if no reference solution is provided
if solution is None :
    solution = pfba ( model )
prob = model . problem
variable = prob . Variable ( "room_old_objective" , ub = solution . objective_value )
constraint = prob . Constraint ( model . solver . objective . expression - variable , ub = 0.0 , lb = 0.0 , name = "room_old_objective_constraint" )
model . objective = prob . Objective ( Zero , direction = "min" , sloppy = True )
vars_and_cons = [ variable , constraint ]
obj_vars = [ ]
for rxn in model . reactions :
    flux = solution . fluxes [ rxn . id ]
    if linear :
        y = prob . Variable ( "y_" + rxn . id , lb = 0 , ub = 1 )
        delta = epsilon = 0.0
    else :
        y = prob . Variable ( "y_" + rxn . id , type = "binary" )
    # upper constraint
    w_u = flux + ( delta * abs ( flux ) ) + epsilon
    upper_const = prob . Constraint ( rxn . flux_expression - y * ( rxn . upper_bound - w_u ) , ub = w_u , name = "room_constraint_upper_" + rxn . id )
    # lower constraint
    w_l = flux - ( delta * abs ( flux ) ) - epsilon
    lower_const = prob . Constraint ( rxn . flux_expression - y * ( rxn . lower_bound - w_l ) , lb = w_l , name = "room_constraint_lower_" + rxn . id )
    vars_and_cons . extend ( [ y , upper_const , lower_const ] )
    obj_vars . append ( y )
model . add_cons_vars ( vars_and_cons )
model . objective . set_linear_coefficients ( { v : 1.0 for v in obj_vars } )


signature:
def sample ( model , n , method = "optgp" , thinning = 100 , processes = 1 , seed = None ) :
    """Sample valid flux distributions from a cobra model .
 The function samples valid flux distributions from a cobra model .
 Currently we support two methods :
 1 . ' optgp ' ( default ) which uses the OptGPSampler that supports parallel
 sampling [ 1 ] _ . Requires large numbers of samples to be performant
 ( n < 1000 ) . For smaller samples ' achr ' might be better suited .
 or
 2 . ' achr ' which uses artificial centering hit - and - run . This is a single
 process method with good convergence [ 2 ] _ .
 Parameters
 model : cobra . Model
 The model from which to sample flux distributions .
 n : int
 The number of samples to obtain . When using ' optgp ' this must be a
 multiple of ` processes ` , otherwise a larger number of samples will be
 returned .
 method : str , optional
 The sampling algorithm to use .
 thinning : int , optional
 The thinning factor of the generated sampling chain . A thinning of 10
 means samples are returned every 10 steps . Defaults to 100 which in
 benchmarks gives approximately uncorrelated samples . If set to one
 will return all iterates .
 processes : int , optional
 Only used for ' optgp ' . The number of processes used to generate
 samples .
 seed : int > 0 , optional
 The random number seed to be used . Initialized to current time stamp
 if None .
 Returns
 pandas . DataFrame
 The generated flux samples . Each row corresponds to a sample of the
 fluxes and the columns are the reactions .
 Notes
 The samplers have a correction method to ensure equality feasibility for
 long - running chains , however this will only work for homogeneous models ,
 meaning models with no non - zero fixed variables or constraints (
 right - hand side of the equalities are zero ) .
 References
 . . [ 1 ] Megchelenbrink W , Huynen M , Marchiori E ( 2014)
 optGpSampler : An Improved Tool for Uniformly Sampling the Solution - Space
 of Genome - Scale Metabolic Networks .
 PLoS ONE 9(2 ) : e86587.
 . . [ 2 ] Direction Choice for Accelerated Convergence in Hit - and - Run Sampling
 David E . Kaufman Robert L . Smith
 Operations Research 199846:1 , 84-95"""

implementation:
if method == "optgp" :
    sampler = OptGPSampler ( model , processes , thinning = thinning , seed = seed )
elif method == "achr" :
    sampler = ACHRSampler ( model , thinning = thinning , seed = seed )
else :
    raise ValueError ( "method must be 'optgp' or 'achr'!" )
return pandas . DataFrame ( columns = [ rxn . id for rxn in model . reactions ] , data = sampler . sample ( n ) )


signature:
def fastcc ( model , flux_threshold = 1.0 , zero_cutoff = None ) :
    r"""Check consistency of a metabolic network using FASTCC [ 1 ] _ .
 FASTCC ( Fast Consistency Check ) is an algorithm for rapid and
 efficient consistency check in metabolic networks . FASTCC is
 a pure LP implementation and is low on computation resource
 demand . FASTCC also circumvents the problem associated with
 reversible reactions for the purpose . Given a global model ,
 it will generate a consistent global model i . e . , remove
 blocked reactions . For more details on FASTCC , please
 check [ 1 ] _ .
 Parameters
 model : cobra . Model
 The constraint - based model to operate on .
 flux _ threshold : float , optional ( default 1.0)
 The flux threshold to consider .
 zero _ cutoff : float , optional
 The cutoff to consider for zero flux ( default model . tolerance ) .
 Returns
 cobra . Model
 The consistent constraint - based model .
 Notes
 The LP used for FASTCC is like so :
 maximize : \ sum _ { i \ in J } z _ i
 s . t . : z _ i \ in [ 0 , \ varepsilon ] \ forall i \ in J , z _ i \ in \ mathbb { R } _ +
 v _ i \ ge z _ i \ forall i \ in J
 Sv = 0 v \ in B
 References
 . . [ 1 ] Vlassis N , Pacheco MP , Sauter T ( 2014)
 Fast Reconstruction of Compact Context - Specific Metabolic Network
 Models .
 PLoS Comput Biol 10(1 ) : e1003424 . doi : 10.1371 / journal . pcbi . 1003424"""

implementation:
zero_cutoff = normalize_cutoff ( model , zero_cutoff )
with model :
    obj_vars = [ ]
    vars_and_cons = [ ]
    prob = model . problem
    for rxn in model . reactions :
        var = prob . Variable ( "auxiliary_{}" . format ( rxn . id ) , lb = 0.0 , ub = flux_threshold )
        const = prob . Constraint ( rxn . forward_variable + rxn . reverse_variable - var , name = "constraint_{}" . format ( rxn . id ) , lb = 0.0 )
        vars_and_cons . extend ( [ var , const ] )
        obj_vars . append ( var )
    model . add_cons_vars ( vars_and_cons )
    model . objective = prob . Objective ( Zero , sloppy = True , direction = "max" )
    model . objective . set_linear_coefficients ( { v : 1.0 for v in obj_vars } )
    sol = model . optimize ( )
rxns_to_remove = sol . fluxes [ sol . fluxes . abs ( ) < zero_cutoff ] . index
consistent_model = model . copy ( )
consistent_model . remove_reactions ( rxns_to_remove , remove_orphans = True )
return consistent_model


signature:
def sample ( self , n , fluxes = True ) :
    """Generate a set of samples .
 This is the basic sampling function for all hit - and - run samplers .
 Parameters
 n : int
 The number of samples that are generated at once .
 fluxes : boolean
 Whether to return fluxes or the internal solver variables . If set
 to False will return a variable for each forward and backward flux
 as well as all additional variables you might have defined in the
 model .
 Returns
 numpy . matrix
 Returns a matrix with ` n ` rows , each containing a flux sample .
 Notes
 Performance of this function linearly depends on the number
 of reactions in your model and the thinning factor ."""

implementation:
samples = np . zeros ( ( n , self . warmup . shape [ 1 ] ) )
for i in range ( 1 , self . thinning * n + 1 ) :
    self . __single_iteration ( )
    if i % self . thinning == 0 :
        samples [ i // self . thinning - 1 , ] = self . prev
if fluxes :
    names = [ r . id for r in self . model . reactions ]
    return pandas . DataFrame ( samples [ : , self . fwd_idx ] - samples [ : , self . rev_idx ] , columns = names )
else :
    names = [ v . name for v in self . model . variables ]
    return pandas . DataFrame ( samples , columns = names )


signature:
def optimizely ( parser , token ) :
    """Optimizely template tag .
 Renders Javascript code to set - up A / B testing . You must supply
 your Optimizely account number in the ` ` OPTIMIZELY _ ACCOUNT _ NUMBER ` `
 setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return OptimizelyNode ( )


signature:
def google_analytics ( parser , token ) :
    """Google Analytics tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your website property ID ( as a string ) in the
 ` ` GOOGLE _ ANALYTICS _ PROPERTY _ ID ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return GoogleAnalyticsNode ( )


signature:
def google_analytics_js ( parser , token ) :
    """Google Analytics tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your website property ID ( as a string ) in the
 ` ` GOOGLE _ ANALYTICS _ JS _ PROPERTY _ ID ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return GoogleAnalyticsJsNode ( )


signature:
def rating_mailru ( parser , token ) :
    """Rating @ Mail . ru counter template tag .
 Renders Javascript code to track page visits . You must supply
 your website counter ID ( as a string ) in the
 ` ` RATING _ MAILRU _ COUNTER _ ID ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return RatingMailruNode ( )


signature:
def clicky ( parser , token ) :
    """Clicky tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your Clicky Site ID ( as a string ) in the ` ` CLICKY _ SITE _ ID ` `
 setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return ClickyNode ( )


signature:
def chartbeat_top ( parser , token ) :
    """Top Chartbeat template tag .
 Render the top Javascript code for Chartbeat ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return ChartbeatTopNode ( )


signature:
def chartbeat_bottom ( parser , token ) :
    """Bottom Chartbeat template tag .
 Render the bottom Javascript code for Chartbeat . You must supply
 your Chartbeat User ID ( as a string ) in the ` ` CHARTBEAT _ USER _ ID ` `
 setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return ChartbeatBottomNode ( )


signature:
def woopra ( parser , token ) :
    """Woopra tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your Woopra domain in the ` ` WOOPRA _ DOMAIN ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return WoopraNode ( )


signature:
def spring_metrics ( parser , token ) :
    """Spring Metrics tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your Spring Metrics Tracking ID in the
 ` ` SPRING _ METRICS _ TRACKING _ ID ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return SpringMetricsNode ( )


signature:
def kiss_insights ( parser , token ) :
    """KISSinsights set - up template tag .
 Renders Javascript code to set - up surveys . You must supply
 your account number and site code in the
 ` ` KISS _ INSIGHTS _ ACCOUNT _ NUMBER ` ` and ` ` KISS _ INSIGHTS _ SITE _ CODE ` `
 settings ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return KissInsightsNode ( )


signature:
def matomo ( parser , token ) :
    """Matomo tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your Matomo domain ( plus optional URI path ) , and tracked site ID
 in the ` ` MATOMO _ DOMAIN _ PATH ` ` and the ` ` MATOMO _ SITE _ ID ` ` setting .
 Custom variables can be passed in the ` ` matomo _ vars ` ` context
 variable . It is an iterable of custom variables as tuples like :
 ` ` ( index , name , value [ , scope ] ) ` ` where scope may be ` ` ' page ' ` `
 ( default ) or ` ` ' visit ' ` ` . Index should be an integer and the
 other parameters should be strings ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return MatomoNode ( )


signature:
def snapengage ( parser , token ) :
    """SnapEngage set - up template tag .
 Renders Javascript code to set - up SnapEngage chat . You must supply
 your widget ID in the ` ` SNAPENGAGE _ WIDGET _ ID ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return SnapEngageNode ( )


signature:
def performable ( parser , token ) :
    """Performable template tag .
 Renders Javascript code to set - up Performable tracking . You must
 supply your Performable API key in the ` ` PERFORMABLE _ API _ KEY ` `
 setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return PerformableNode ( )


signature:
def _timestamp ( when ) :
    """Python 2 compatibility for ` datetime . timestamp ( ) ` ."""

implementation: return ( time . mktime ( when . timetuple ( ) ) if sys . version_info < ( 3 , ) else when . timestamp ( ) )


signature:
def _hashable_bytes ( data ) :
    """Coerce strings to hashable bytes ."""

implementation:
if isinstance ( data , bytes ) :
    return data
elif isinstance ( data , str ) :
    return data . encode ( 'ascii' )
    # Fail on anything non - ASCII .
else :
    raise TypeError ( data )


signature:
def intercom_user_hash ( data ) :
    """Return a SHA - 256 HMAC ` user _ hash ` as expected by Intercom , if configured .
 Return None if the ` INTERCOM _ HMAC _ SECRET _ KEY ` setting is not configured ."""

implementation:
if getattr ( settings , 'INTERCOM_HMAC_SECRET_KEY' , None ) :
    return hmac . new ( key = _hashable_bytes ( settings . INTERCOM_HMAC_SECRET_KEY ) , msg = _hashable_bytes ( data ) , digestmod = hashlib . sha256 , ) . hexdigest ( )
else :
    return None


signature:
def intercom ( parser , token ) :
    """Intercom . io template tag .
 Renders Javascript code to intercom . io testing . You must supply
 your APP ID account number in the ` ` INTERCOM _ APP _ ID ` `
 setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return IntercomNode ( )


signature:
def uservoice ( parser , token ) :
    """UserVoice tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your UserVoice Widget Key in the ` ` USERVOICE _ WIDGET _ KEY ` `
 setting or the ` ` uservoice _ widget _ key ` ` template context variable ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return UserVoiceNode ( )


signature:
def kiss_metrics ( parser , token ) :
    """KISSinsights tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your KISSmetrics API key in the ` ` KISS _ METRICS _ API _ KEY ` `
 setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return KissMetricsNode ( )


signature:
def piwik ( parser , token ) :
    """Piwik tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your Piwik domain ( plus optional URI path ) , and tracked site ID
 in the ` ` PIWIK _ DOMAIN _ PATH ` ` and the ` ` PIWIK _ SITE _ ID ` ` setting .
 Custom variables can be passed in the ` ` piwik _ vars ` ` context
 variable . It is an iterable of custom variables as tuples like :
 ` ` ( index , name , value [ , scope ] ) ` ` where scope may be ` ` ' page ' ` `
 ( default ) or ` ` ' visit ' ` ` . Index should be an integer and the
 other parameters should be strings ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return PiwikNode ( )


signature:
def get_required_setting ( setting , value_re , invalid_msg ) :
    """Return a constant from ` ` django . conf . settings ` ` . The ` setting `
 argument is the constant name , the ` value _ re ` argument is a regular
 expression used to validate the setting value and the ` invalid _ msg `
 argument is used as exception message if the value is not valid ."""

implementation:
try :
    value = getattr ( settings , setting )
except AttributeError :
    raise AnalyticalException ( "%s setting: not found" % setting )
if not value :
    raise AnalyticalException ( "%s setting is not set" % setting )
value = str ( value )
if not value_re . search ( value ) :
    raise AnalyticalException ( "%s setting: %s: '%s'" % ( setting , invalid_msg , value ) )
return value


signature:
def get_user_from_context ( context ) :
    """Get the user instance from the template context , if possible .
 If the context does not contain a ` request ` or ` user ` attribute ,
 ` None ` is returned ."""

implementation:
try :
    return context [ 'user' ]
except KeyError :
    pass
try :
    request = context [ 'request' ]
    return request . user
except ( KeyError , AttributeError ) :
    pass
return None


signature:
def get_identity ( context , prefix = None , identity_func = None , user = None ) :
    """Get the identity of a logged in user from a template context .
 The ` prefix ` argument is used to provide different identities to
 different analytics services . The ` identity _ func ` argument is a
 function that returns the identity of the user ; by default the
 identity is the username ."""

implementation:
if prefix is not None :
    try :
        return context [ '%s_identity' % prefix ]
    except KeyError :
        pass
try :
    return context [ 'analytical_identity' ]
except KeyError :
    pass
if getattr ( settings , 'ANALYTICAL_AUTO_IDENTIFY' , True ) :
    try :
        if user is None :
            user = get_user_from_context ( context )
        if get_user_is_authenticated ( user ) :
            if identity_func is not None :
                return identity_func ( user )
            else :
                return user . get_username ( )
    except ( KeyError , AttributeError ) :
        pass
return None


signature:
def get_domain ( context , prefix ) :
    """Return the domain used for the tracking code . Each service may be
 configured with its own domain ( called ` < name > _ domain ` ) , or a
 django - analytical - wide domain may be set ( using ` analytical _ domain ` .
 If no explicit domain is found in either the context or the
 settings , try to get the domain from the contrib sites framework ."""

implementation:
domain = context . get ( '%s_domain' % prefix )
if domain is None :
    domain = context . get ( 'analytical_domain' )
if domain is None :
    domain = getattr ( settings , '%s_DOMAIN' % prefix . upper ( ) , None )
if domain is None :
    domain = getattr ( settings , 'ANALYTICAL_DOMAIN' , None )
if domain is None :
    if 'django.contrib.sites' in settings . INSTALLED_APPS :
        from django . contrib . sites . models import Site
        try :
            domain = Site . objects . get_current ( ) . domain
        except ( ImproperlyConfigured , Site . DoesNotExist ) :
            pass
return domain


signature:
def is_internal_ip ( context , prefix = None ) :
    """Return whether the visitor is coming from an internal IP address ,
 based on information from the template context .
 The prefix is used to allow different analytics services to have
 different notions of internal addresses ."""

implementation:
try :
    request = context [ 'request' ]
    remote_ip = request . META . get ( 'HTTP_X_FORWARDED_FOR' , '' )
    if not remote_ip :
        remote_ip = request . META . get ( 'REMOTE_ADDR' , '' )
    if not remote_ip :
        return False
    internal_ips = None
    if prefix is not None :
        internal_ips = getattr ( settings , '%s_INTERNAL_IPS' % prefix , None )
    if internal_ips is None :
        internal_ips = getattr ( settings , 'ANALYTICAL_INTERNAL_IPS' , None )
    if internal_ips is None :
        internal_ips = getattr ( settings , 'INTERNAL_IPS' , None )
    return remote_ip in ( internal_ips or [ ] )
except ( KeyError , AttributeError ) :
    return False


signature:
def mixpanel ( parser , token ) :
    """Mixpanel tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your Mixpanel token in the ` ` MIXPANEL _ API _ TOKEN ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return MixpanelNode ( )


signature:
def gosquared ( parser , token ) :
    """GoSquared tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your GoSquared site token in the ` ` GOSQUARED _ SITE _ TOKEN ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return GoSquaredNode ( )


signature:
def olark ( parser , token ) :
    """Olark set - up template tag .
 Renders Javascript code to set - up Olark chat . You must supply
 your site ID in the ` ` OLARK _ SITE _ ID ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return OlarkNode ( )


signature:
def clickmap ( parser , token ) :
    """Clickmap tracker template tag .
 Renders Javascript code to track page visits . You must supply
 your clickmap tracker ID ( as a string ) in the ` ` CLICKMAP _ TRACKER _ ID ` `
 setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return ClickmapNode ( )


signature:
def gauges ( parser , token ) :
    """Gaug . es template tag .
 Renders Javascript code to gaug . es testing . You must supply
 your Site ID account number in the ` ` GAUGES _ SITE _ ID ` `
 setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return GaugesNode ( )


signature:
def crazy_egg ( parser , token ) :
    """Crazy Egg tracking template tag .
 Renders Javascript code to track page clicks . You must supply
 your Crazy Egg account number ( as a string ) in the
 ` ` CRAZY _ EGG _ ACCOUNT _ NUMBER ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return CrazyEggNode ( )


signature:
def yandex_metrica ( parser , token ) :
    """Yandex . Metrica counter template tag .
 Renders Javascript code to track page visits . You must supply
 your website counter ID ( as a string ) in the
 ` ` YANDEX _ METRICA _ COUNTER _ ID ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return YandexMetricaNode ( )


signature:
def hubspot ( parser , token ) :
    """HubSpot tracking template tag .
 Renders Javascript code to track page visits . You must supply
 your portal ID ( as a string ) in the ` ` HUBSPOT _ PORTAL _ ID ` ` setting ."""

implementation:
bits = token . split_contents ( )
if len ( bits ) > 1 :
    raise TemplateSyntaxError ( "'%s' takes no arguments" % bits [ 0 ] )
return HubSpotNode ( )


signature:
def status_printer ( ) :
    """Manage the printing and in - place updating of a line of characters
 . . note : :
 If the string is longer than a line , then in - place updating may not
 work ( it will print a new line at each refresh ) ."""

implementation:
last_len = [ 0 ]
def p ( s ) :
    s = next ( spinner ) + ' ' + s
    len_s = len ( s )
    output = '\r' + s + ( ' ' * max ( last_len [ 0 ] - len_s , 0 ) )
    sys . stdout . write ( output )
    sys . stdout . flush ( )
    last_len [ 0 ] = len_s
return p


signature:
def get_or_guess_paths_to_mutate ( paths_to_mutate ) :
    """: type paths _ to _ mutate : str or None
 : rtype : str"""

implementation:
if paths_to_mutate is None : # Guess path with code
    this_dir = os . getcwd ( ) . split ( os . sep ) [ - 1 ]
    if isdir ( 'lib' ) :
        return 'lib'
    elif isdir ( 'src' ) :
        return 'src'
    elif isdir ( this_dir ) :
        return this_dir
    elif isdir ( this_dir . replace ( '-' , '_' ) ) :
        return this_dir . replace ( '-' , '_' )
    elif isdir ( this_dir . replace ( ' ' , '_' ) ) :
        return this_dir . replace ( ' ' , '_' )
    elif isdir ( this_dir . replace ( '-' , '' ) ) :
        return this_dir . replace ( '-' , '' )
    elif isdir ( this_dir . replace ( ' ' , '' ) ) :
        return this_dir . replace ( ' ' , '' )
    else :
        raise FileNotFoundError ( 'Could not figure out where the code to mutate is. ' 'Please specify it on the command line using --paths-to-mutate, ' 'or by adding "paths_to_mutate=code_dir" in setup.cfg to the [mutmut] section.' )
else :
    return paths_to_mutate


signature:
def do_apply ( mutation_pk , dict_synonyms , backup ) :
    """Apply a specified mutant to the source code
 : param mutation _ pk : mutmut cache primary key of the mutant to apply
 : type mutation _ pk : str
 : param dict _ synonyms : list of synonym keywords for a python dictionary
 : type dict _ synonyms : list [ str ]
 : param backup : if : obj : ` True ` create a backup of the source file
 before applying the mutation
 : type backup : bool"""

implementation:
filename , mutation_id = filename_and_mutation_id_from_pk ( int ( mutation_pk ) )
update_line_numbers ( filename )
context = Context ( mutation_id = mutation_id , filename = filename , dict_synonyms = dict_synonyms , )
mutate_file ( backup = backup , context = context , )
if context . number_of_performed_mutations == 0 :
    raise RuntimeError ( 'No mutations performed.' )


signature:
def climain ( command , argument , argument2 , paths_to_mutate , backup , runner , tests_dir , test_time_multiplier , test_time_base , swallow_output , use_coverage , dict_synonyms , cache_only , version , suspicious_policy , untested_policy , pre_mutation , post_mutation , use_patch_file ) :
    """commands : \n
 run [ mutation id ] \n
 Runs mutmut . You probably want to start with just trying this . If you supply a mutation ID mutmut will check just this mutant . \n
 results \n
 Print the results . \n
 apply [ mutation id ] \n
 Apply a mutation on disk . \n
 show [ mutation id ] \n
 Show a mutation diff . \n
 junitxml \n
 Show a mutation diff with junitxml format ."""

implementation:
if test_time_base is None : # click sets the default = 0.0 to None
    test_time_base = 0.0
if test_time_multiplier is None : # click sets the default = 0.0 to None
    test_time_multiplier = 0.0
sys . exit ( main ( command , argument , argument2 , paths_to_mutate , backup , runner , tests_dir , test_time_multiplier , test_time_base , swallow_output , use_coverage , dict_synonyms , cache_only , version , suspicious_policy , untested_policy , pre_mutation , post_mutation , use_patch_file ) )


signature:
def main ( command , argument , argument2 , paths_to_mutate , backup , runner , tests_dir , test_time_multiplier , test_time_base , swallow_output , use_coverage , dict_synonyms , cache_only , version , suspicious_policy , untested_policy , pre_mutation , post_mutation , use_patch_file ) :
    """return exit code , after performing an mutation test run .
 : return : the exit code from executing the mutation tests
 : rtype : int"""

implementation:
if version :
    print ( "mutmut version %s" % __version__ )
    return 0
if use_coverage and use_patch_file :
    raise click . BadArgumentUsage ( "You can't combine --use-coverage and --use-patch" )
valid_commands = [ 'run' , 'results' , 'apply' , 'show' , 'junitxml' ]
if command not in valid_commands :
    raise click . BadArgumentUsage ( '%s is not a valid command, must be one of %s' % ( command , ', ' . join ( valid_commands ) ) )
if command == 'results' and argument :
    raise click . BadArgumentUsage ( 'The %s command takes no arguments' % command )
dict_synonyms = [ x . strip ( ) for x in dict_synonyms . split ( ',' ) ]
if command in ( 'show' , 'diff' ) :
    if not argument :
        print_result_cache ( )
        return 0
    if argument == 'all' :
        print_result_cache ( show_diffs = True , dict_synonyms = dict_synonyms , print_only_filename = argument2 )
        return 0
    print ( get_unified_diff ( argument , dict_synonyms ) )
    return 0
if use_coverage and not exists ( '.coverage' ) :
    raise FileNotFoundError ( 'No .coverage file found. You must generate a coverage file to use this feature.' )
if command == 'results' :
    print_result_cache ( )
    return 0
if command == 'junitxml' :
    print_result_cache_junitxml ( dict_synonyms , suspicious_policy , untested_policy )
    return 0
if command == 'apply' :
    do_apply ( argument , dict_synonyms , backup )
    return 0
paths_to_mutate = get_or_guess_paths_to_mutate ( paths_to_mutate )
if not isinstance ( paths_to_mutate , ( list , tuple ) ) :
    paths_to_mutate = [ x . strip ( ) for x in paths_to_mutate . split ( ',' ) ]
if not paths_to_mutate :
    raise click . BadOptionUsage ( '--paths-to-mutate' , 'You must specify a list of paths to mutate. Either as a command line argument, or by setting paths_to_mutate under the section [mutmut] in setup.cfg' )
tests_dirs = [ ]
for p in tests_dir . split ( ':' ) :
    tests_dirs . extend ( glob ( p , recursive = True ) )
for p in paths_to_mutate :
    for pt in tests_dir . split ( ':' ) :
        tests_dirs . extend ( glob ( p + '/**/' + pt , recursive = True ) )
del tests_dir
os . environ [ 'PYTHONDONTWRITEBYTECODE' ] = '1'
# stop python from creating . pyc files
using_testmon = '--testmon' in runner
print ( """
- Mutation testing starting -

These are the steps:
1. A full test suite run will be made to make sure we
   can run the tests successfully and we know how long
   it takes (to detect infinite loops for example)
2. Mutants will be generated and checked

Results are stored in .mutmut-cache.
Print found mutants with `mutmut results`.

Legend for output:
🎉 Killed mutants.   The goal is for everything to end up in this bucket.
⏰ Timeout.          Test suite took 10 times as long as the baseline so were killed.
🤔 Suspicious.       Tests took a long time, but not long enough to be fatal.
🙁 Survived.         This means your tests needs to be expanded.
""" )
baseline_time_elapsed = time_test_suite ( swallow_output = not swallow_output , test_command = runner , using_testmon = using_testmon )
if using_testmon :
    copy ( '.testmondata' , '.testmondata-initial' )
# if we ' re running in a mode with externally whitelisted lines
if use_coverage or use_patch_file :
    covered_lines_by_filename = { }
    if use_coverage :
        coverage_data = read_coverage_data ( )
    else :
        assert use_patch_file
        covered_lines_by_filename = read_patch_data ( use_patch_file )
        coverage_data = None
    def _exclude ( context ) :
        try :
            covered_lines = covered_lines_by_filename [ context . filename ]
        except KeyError :
            if coverage_data is not None :
                covered_lines = coverage_data . lines ( os . path . abspath ( context . filename ) )
                covered_lines_by_filename [ context . filename ] = covered_lines
            else :
                covered_lines = None
        if covered_lines is None :
            return True
        current_line = context . current_line_index + 1
        if current_line not in covered_lines :
            return True
        return False
else :
    def _exclude ( context ) :
        del context
        return False
if command != 'run' :
    raise click . BadArgumentUsage ( "Invalid command %s" % command )
mutations_by_file = { }
if argument is None :
    for path in paths_to_mutate :
        for filename in python_source_files ( path , tests_dirs ) :
            update_line_numbers ( filename )
            add_mutations_by_file ( mutations_by_file , filename , _exclude , dict_synonyms )
else :
    filename , mutation_id = filename_and_mutation_id_from_pk ( int ( argument ) )
    mutations_by_file [ filename ] = [ mutation_id ]
total = sum ( len ( mutations ) for mutations in mutations_by_file . values ( ) )
print ( )
print ( '2. Checking mutants' )
config = Config ( swallow_output = not swallow_output , test_command = runner , exclude_callback = _exclude , baseline_time_elapsed = baseline_time_elapsed , backup = backup , dict_synonyms = dict_synonyms , total = total , using_testmon = using_testmon , cache_only = cache_only , tests_dirs = tests_dirs , hash_of_tests = hash_of_tests ( tests_dirs ) , test_time_multiplier = test_time_multiplier , test_time_base = test_time_base , pre_mutation = pre_mutation , post_mutation = post_mutation , )
try :
    run_mutation_tests ( config = config , mutations_by_file = mutations_by_file )
except Exception as e :
    traceback . print_exc ( )
    return compute_exit_code ( config , e )
else :
    return compute_exit_code ( config )
finally :
    print ( )


signature:
def run_mutation ( config , filename , mutation_id ) :
    """: type config : Config
 : type filename : str
 : type mutation _ id : MutationID
 : return : ( computed or cached ) status of the tested mutant
 : rtype : str"""

implementation:
context = Context ( mutation_id = mutation_id , filename = filename , exclude = config . exclude_callback , dict_synonyms = config . dict_synonyms , config = config , )
cached_status = cached_mutation_status ( filename , mutation_id , config . hash_of_tests )
if cached_status == BAD_SURVIVED :
    config . surviving_mutants += 1
elif cached_status == BAD_TIMEOUT :
    config . surviving_mutants_timeout += 1
elif cached_status == OK_KILLED :
    config . killed_mutants += 1
elif cached_status == OK_SUSPICIOUS :
    config . suspicious_mutants += 1
else :
    assert cached_status == UNTESTED , cached_status
config . print_progress ( )
if cached_status != UNTESTED :
    return cached_status
if config . pre_mutation :
    result = subprocess . check_output ( config . pre_mutation , shell = True ) . decode ( ) . strip ( )
    if result :
        print ( result )
try :
    number_of_mutations_performed = mutate_file ( backup = True , context = context )
    assert number_of_mutations_performed
    start = time ( )
    try :
        survived = tests_pass ( config )
    except TimeoutError :
        context . config . surviving_mutants_timeout += 1
        return BAD_TIMEOUT
    time_elapsed = time ( ) - start
    if time_elapsed > config . test_time_base + ( config . baseline_time_elapsed * config . test_time_multipler ) :
        config . suspicious_mutants += 1
        return OK_SUSPICIOUS
    if survived :
        context . config . surviving_mutants += 1
        return BAD_SURVIVED
    else :
        context . config . killed_mutants += 1
        return OK_KILLED
finally :
    move ( filename + '.bak' , filename )
    if config . post_mutation :
        result = subprocess . check_output ( config . post_mutation , shell = True ) . decode ( ) . strip ( )
        if result :
            print ( result )


signature:
def read_coverage_data ( ) :
    """: rtype : CoverageData or None"""

implementation:
print ( 'Using coverage data from .coverage file' )
# noinspection PyPackageRequirements , PyUnresolvedReferences
from coverage import Coverage
cov = Coverage ( '.coverage' )
cov . load ( )
return cov . get_data ( )


signature:
def add_mutations_by_file ( mutations_by_file , filename , exclude , dict_synonyms ) :
    """: type mutations _ by _ file : dict [ str , list [ MutationID ] ]
 : type filename : str
 : type exclude : Callable [ [ Context ] , bool ]
 : type dict _ synonyms : list [ str ]"""

implementation:
with open ( filename ) as f :
    source = f . read ( )
context = Context ( source = source , filename = filename , exclude = exclude , dict_synonyms = dict_synonyms , )
try :
    mutations_by_file [ filename ] = list_mutations ( context )
    register_mutants ( mutations_by_file )
except Exception as e :
    raise RuntimeError ( 'Failed while creating mutations for %s, for line "%s"' % ( context . filename , context . current_source_line ) , e )


signature:
def python_source_files ( path , tests_dirs ) :
    """Attempt to guess where the python source files to mutate are and yield
 their paths
 : param path : path to a python source file or package directory
 : type path : str
 : param tests _ dirs : list of directory paths containing test files
 ( we do not want to mutate these ! )
 : type tests _ dirs : list [ str ]
 : return : generator listing the paths to the python source files to mutate
 : rtype : Generator [ str , None , None ]"""

implementation:
if isdir ( path ) :
    for root , dirs , files in os . walk ( path ) :
        dirs [ : ] = [ d for d in dirs if os . path . join ( root , d ) not in tests_dirs ]
        for filename in files :
            if filename . endswith ( '.py' ) :
                yield os . path . join ( root , filename )
else :
    yield path


signature:
def compute_exit_code ( config , exception = None ) :
    """Compute an exit code for mutmut mutation testing
 The following exit codes are available for mutmut :
 * 0 if all mutants were killed ( OK _ KILLED )
 * 1 if a fatal error occurred
 * 2 if one or more mutants survived ( BAD _ SURVIVED )
 * 4 if one or more mutants timed out ( BAD _ TIMEOUT )
 * 8 if one or more mutants caused tests to take twice as long ( OK _ SUSPICIOUS )
 Exit codes 1 to 8 will be bit - ORed so that it is possible to know what
 different mutant statuses occurred during mutation testing .
 : param exception :
 : type exception : Exception
 : param config :
 : type config : Config
 : return : integer noting the exit code of the mutation tests .
 : rtype : int"""

implementation:
code = 0
if exception is not None :
    code = code | 1
if config . surviving_mutants > 0 :
    code = code | 2
if config . surviving_mutants_timeout > 0 :
    code = code | 4
if config . suspicious_mutants > 0 :
    code = code | 8
return code


signature:
def argument_mutation ( children , context , ** _ ) :
    """: type context : Context"""

implementation:
if len ( context . stack ) >= 3 and context . stack [ - 3 ] . type in ( 'power' , 'atom_expr' ) :
    stack_pos_of_power_node = - 3
elif len ( context . stack ) >= 4 and context . stack [ - 4 ] . type in ( 'power' , 'atom_expr' ) :
    stack_pos_of_power_node = - 4
else :
    return
power_node = context . stack [ stack_pos_of_power_node ]
if power_node . children [ 0 ] . type == 'name' and power_node . children [ 0 ] . value in context . dict_synonyms :
    c = children [ 0 ]
    if c . type == 'name' :
        children = children [ : ]
        children [ 0 ] = Name ( c . value + 'XX' , start_pos = c . start_pos , prefix = c . prefix )
        return children


signature:
def mutate ( context ) :
    """: type context : Context
 : return : tuple : mutated source code , number of mutations performed
 : rtype : tuple [ str , int ]"""

implementation:
try :
    result = parse ( context . source , error_recovery = False )
except Exception :
    print ( 'Failed to parse %s. Internal error from parso follows.' % context . filename )
    print ( '----------------------------------' )
    raise
mutate_list_of_nodes ( result , context = context )
mutated_source = result . get_code ( ) . replace ( ' not not ' , ' ' )
if context . remove_newline_at_end :
    assert mutated_source [ - 1 ] == '\n'
    mutated_source = mutated_source [ : - 1 ]
if context . number_of_performed_mutations : # If we said we mutated the code , check that it has actually changed
    assert context . source != mutated_source
context . mutated_source = mutated_source
return mutated_source , context . number_of_performed_mutations


signature:
def mutate_node ( node , context ) :
    """: type context : Context"""

implementation:
context . stack . append ( node )
try :
    if node . type in ( 'tfpdef' , 'import_from' , 'import_name' ) :
        return
    if node . start_pos [ 0 ] - 1 != context . current_line_index :
        context . current_line_index = node . start_pos [ 0 ] - 1
        context . index = 0
    # indexes are unique per line , so start over here !
    if hasattr ( node , 'children' ) :
        mutate_list_of_nodes ( node , context = context )
        # this is just an optimization to stop early
        if context . number_of_performed_mutations and context . mutation_id != ALL :
            return
    mutation = mutations_by_type . get ( node . type )
    if mutation is None :
        return
    for key , value in sorted ( mutation . items ( ) ) :
        old = getattr ( node , key )
        if context . exclude_line ( ) :
            continue
        new = evaluate ( value , context = context , node = node , value = getattr ( node , 'value' , None ) , children = getattr ( node , 'children' , None ) , )
        assert not callable ( new )
        if new is not None and new != old :
            if context . should_mutate ( ) :
                context . number_of_performed_mutations += 1
                context . performed_mutation_ids . append ( context . mutation_id_of_current_index )
                setattr ( node , key , new )
            context . index += 1
        # this is just an optimization to stop early
        if context . number_of_performed_mutations and context . mutation_id != ALL :
            return
finally :
    context . stack . pop ( )


signature:
def mutate_list_of_nodes ( node , context ) :
    """: type context : Context"""

implementation:
return_annotation_started = False
for child_node in node . children :
    if child_node . type == 'operator' and child_node . value == '->' :
        return_annotation_started = True
    if return_annotation_started and child_node . type == 'operator' and child_node . value == ':' :
        return_annotation_started = False
    if return_annotation_started :
        continue
    mutate_node ( child_node , context = context )
    # this is just an optimization to stop early
    if context . number_of_performed_mutations and context . mutation_id != ALL :
        return


signature:
def mutate_file ( backup , context ) :
    """: type backup : bool
 : type context : Context"""

implementation:
with open ( context . filename ) as f :
    code = f . read ( )
context . source = code
if backup :
    with open ( context . filename + '.bak' , 'w' ) as f :
        f . write ( code )
result , number_of_mutations_performed = mutate ( context )
with open ( context . filename , 'w' ) as f :
    f . write ( result )
return number_of_mutations_performed


signature:
def connect ( self , timeout_sec = TIMEOUT_SEC ) :
    """Connect to the device . If not connected within the specified timeout
 then an exception is thrown ."""

implementation:
self . _central_manager . connectPeripheral_options_ ( self . _peripheral , None )
if not self . _connected . wait ( timeout_sec ) :
    raise RuntimeError ( 'Failed to connect to device within timeout period!' )


signature:
def disconnect ( self , timeout_sec = TIMEOUT_SEC ) :
    """Disconnect from the device . If not disconnected within the specified
 timeout then an exception is thrown ."""

implementation:
# Remove all the services , characteristics , and descriptors from the
# lists of those items . Do this before disconnecting because they wont ' t
# be accessible afterwards .
for service in self . list_services ( ) :
    for char in service . list_characteristics ( ) :
        for desc in char . list_descriptors ( ) :
            descriptor_list ( ) . remove ( desc )
        characteristic_list ( ) . remove ( char )
    service_list ( ) . remove ( service )
# Now disconnect .
self . _central_manager . cancelPeripheralConnection_ ( self . _peripheral )
if not self . _disconnected . wait ( timeout_sec ) :
    raise RuntimeError ( 'Failed to disconnect to device within timeout period!' )


signature:
def _update_advertised ( self , advertised ) :
    """Called when advertisement data is received ."""

implementation:
# Advertisement data was received , pull out advertised service UUIDs and
# name from advertisement data .
if 'kCBAdvDataServiceUUIDs' in advertised :
    self . _advertised = self . _advertised + map ( cbuuid_to_uuid , advertised [ 'kCBAdvDataServiceUUIDs' ] )


signature:
def _characteristics_discovered ( self , service ) :
    """Called when GATT characteristics have been discovered ."""

implementation:
# Characteristics for the specified service were discovered . Update
# set of discovered services and signal when all have been discovered .
self . _discovered_services . add ( service )
if self . _discovered_services >= set ( self . _peripheral . services ( ) ) : # Found all the services characteristics , finally time to fire the
# service discovery complete event .
    self . _discovered . set ( )


signature:
def _characteristic_changed ( self , characteristic ) :
    """Called when the specified characteristic has changed its value ."""

implementation:
# Called when a characteristic is changed . Get the on _ changed handler
# for this characteristic ( if it exists ) and call it .
on_changed = self . _char_on_changed . get ( characteristic , None )
if on_changed is not None :
    on_changed ( characteristic . value ( ) . bytes ( ) . tobytes ( ) )
# Also tell the characteristic that it has a new value .
# First get the service that is associated with this characteristic .
char = characteristic_list ( ) . get ( characteristic )
if char is not None :
    char . _value_read . set ( )


signature:
def _descriptor_changed ( self , descriptor ) :
    """Called when the specified descriptor has changed its value ."""

implementation:
# Tell the descriptor it has a new value to read .
desc = descriptor_list ( ) . get ( descriptor )
if desc is not None :
    desc . _value_read . set ( )


signature:
def discover ( self , service_uuids , char_uuids , timeout_sec = TIMEOUT_SEC ) :
    """Wait up to timeout _ sec for the specified services and characteristics
 to be discovered on the device . If the timeout is exceeded without
 discovering the services and characteristics then an exception is thrown ."""

implementation:
# Since OSX tells us when all services and characteristics are discovered
# this function can just wait for that full service discovery .
if not self . _discovered . wait ( timeout_sec ) :
    raise RuntimeError ( 'Failed to discover device services within timeout period!' )


signature:
def rssi ( self , timeout_sec = TIMEOUT_SEC ) :
    """Return the RSSI signal strength in decibels ."""

implementation:
# Kick off query to get RSSI , then wait for it to return asyncronously
# when the _ rssi _ changed ( ) function is called .
self . _rssi_read . clear ( )
self . _peripheral . readRSSI ( )
if not self . _rssi_read . wait ( timeout_sec ) :
    raise RuntimeError ( 'Exceeded timeout waiting for RSSI value!' )
return self . _rssi


signature:
def read ( self , timeout_sec = None ) :
    """Block until data is available to read from the UART . Will return a
 string of data that has been received . Timeout _ sec specifies how many
 seconds to wait for data to be available and will block forever if None
 ( the default ) . If the timeout is exceeded and no data is found then
 None is returned ."""

implementation:
try :
    return self . _queue . get ( timeout = timeout_sec )
except queue . Empty : # Timeout exceeded , return None to signify no data received .
    return None


signature:
def list_characteristics ( self ) :
    """Return list of GATT characteristics that have been discovered for this
 service ."""

implementation:
paths = self . _props . Get ( _SERVICE_INTERFACE , 'Characteristics' )
return map ( BluezGattCharacteristic , get_provider ( ) . _get_objects_by_path ( paths ) )


signature:
def start_notify ( self , on_change ) :
    """Enable notification of changes for this characteristic on the
 specified on _ change callback . on _ change should be a function that takes
 one parameter which is the value ( as a string of bytes ) of the changed
 characteristic value ."""

implementation:
# Setup a closure to be the first step in handling the on change callback .
# This closure will verify the characteristic is changed and pull out the
# new value to pass to the user ' s on change callback .
def characteristic_changed ( iface , changed_props , invalidated_props ) : # Check that this change is for a GATT characteristic and it has a
# new value .
    if iface != _CHARACTERISTIC_INTERFACE :
        return
    if 'Value' not in changed_props :
        return
    # Send the new value to the on _ change callback .
    on_change ( '' . join ( map ( chr , changed_props [ 'Value' ] ) ) )
# Hook up the property changed signal to call the closure above .
self . _props . connect_to_signal ( 'PropertiesChanged' , characteristic_changed )
# Enable notifications for changes on the characteristic .
self . _characteristic . StartNotify ( )


signature:
def list_descriptors ( self ) :
    """Return list of GATT descriptors that have been discovered for this
 characteristic ."""

implementation:
paths = self . _props . Get ( _CHARACTERISTIC_INTERFACE , 'Descriptors' )
return map ( BluezGattDescriptor , get_provider ( ) . _get_objects_by_path ( paths ) )


signature:
def _state_changed ( self , state ) :
    """Called when the power state changes ."""

implementation:
logger . debug ( 'Adapter state change: {0}' . format ( state ) )
# Handle when powered on .
if state == 5 :
    self . _powered_off . clear ( )
    self . _powered_on . set ( )
# Handle when powered off .
elif state == 4 :
    self . _powered_on . clear ( )
    self . _powered_off . set ( )


signature:
def start_scan ( self , timeout_sec = TIMEOUT_SEC ) :
    """Start scanning for BLE devices ."""

implementation:
get_provider ( ) . _central_manager . scanForPeripheralsWithServices_options_ ( None , None )
self . _is_scanning = True


signature:
def stop_scan ( self , timeout_sec = TIMEOUT_SEC ) :
    """Stop scanning for BLE devices ."""

implementation:
get_provider ( ) . _central_manager . stopScan ( )
self . _is_scanning = False


signature:
def power_on ( self , timeout_sec = TIMEOUT_SEC ) :
    """Power on Bluetooth ."""

implementation:
# Turn on bluetooth and wait for powered on event to be set .
self . _powered_on . clear ( )
IOBluetoothPreferenceSetControllerPowerState ( 1 )
if not self . _powered_on . wait ( timeout_sec ) :
    raise RuntimeError ( 'Exceeded timeout waiting for adapter to power on!' )


signature:
def power_off ( self , timeout_sec = TIMEOUT_SEC ) :
    """Power off Bluetooth ."""

implementation:
# Turn off bluetooth .
self . _powered_off . clear ( )
IOBluetoothPreferenceSetControllerPowerState ( 0 )
if not self . _powered_off . wait ( timeout_sec ) :
    raise RuntimeError ( 'Exceeded timeout waiting for adapter to power off!' )


signature:
def find_device ( cls , timeout_sec = TIMEOUT_SEC ) :
    """Find the first available device that supports this service and return
 it , or None if no device is found . Will wait for up to timeout _ sec
 seconds to find the device ."""

implementation: return get_provider ( ) . find_device ( service_uuids = cls . ADVERTISED , timeout_sec = timeout_sec )


signature:
def discover ( cls , device , timeout_sec = TIMEOUT_SEC ) :
    """Wait until the specified device has discovered the expected services
 and characteristics for this service . Should be called once before other
 calls are made on the service . Returns true if the service has been
 discovered in the specified timeout , or false if not discovered ."""

implementation: device . discover ( cls . SERVICES , cls . CHARACTERISTICS , timeout_sec )


signature:
def find_service ( self , uuid ) :
    """Return the first child service found that has the specified
 UUID . Will return None if no service that matches is found ."""

implementation:
for service in self . list_services ( ) :
    if service . uuid == uuid :
        return service
return None


signature:
def connect ( self , timeout_sec = TIMEOUT_SEC ) :
    """Connect to the device . If not connected within the specified timeout
 then an exception is thrown ."""

implementation:
self . _connected . clear ( )
self . _device . Connect ( )
if not self . _connected . wait ( timeout_sec ) :
    raise RuntimeError ( 'Exceeded timeout waiting to connect to device!' )


signature:
def disconnect ( self , timeout_sec = TIMEOUT_SEC ) :
    """Disconnect from the device . If not disconnected within the specified
 timeout then an exception is thrown ."""

implementation:
self . _disconnected . clear ( )
self . _device . Disconnect ( )
if not self . _disconnected . wait ( timeout_sec ) :
    raise RuntimeError ( 'Exceeded timeout waiting to disconnect from device!' )


signature:
def list_services ( self ) :
    """Return a list of GattService objects that have been discovered for
 this device ."""

implementation: return map ( BluezGattService , get_provider ( ) . _get_objects ( _SERVICE_INTERFACE , self . _device . object_path ) )


signature:
def discover ( self , service_uuids , char_uuids , timeout_sec = TIMEOUT_SEC ) :
    """Wait up to timeout _ sec for the specified services and characteristics
 to be discovered on the device . If the timeout is exceeded without
 discovering the services and characteristics then an exception is thrown ."""

implementation:
# Turn expected values into a counter of each UUID for fast comparison .
expected_services = set ( service_uuids )
expected_chars = set ( char_uuids )
# Loop trying to find the expected services for the device .
start = time . time ( )
while True : # Find actual services discovered for the device .
    actual_services = set ( self . advertised )
    # Find actual characteristics discovered for the device .
    chars = map ( BluezGattCharacteristic , get_provider ( ) . _get_objects ( _CHARACTERISTIC_INTERFACE , self . _device . object_path ) )
    actual_chars = set ( map ( lambda x : x . uuid , chars ) )
    # Compare actual discovered UUIDs with expected and return true if at
    # least the expected UUIDs are available .
    if actual_services >= expected_services and actual_chars >= expected_chars : # Found at least the expected services !
        return True
    # Couldn ' t find the devices so check if timeout has expired and try again .
    if time . time ( ) - start >= timeout_sec :
        return False
    time . sleep ( 1 )


signature:
def advertised ( self ) :
    """Return a list of UUIDs for services that are advertised by this
 device ."""

implementation:
uuids = [ ]
# Get UUIDs property but wrap it in a try / except to catch if the property
# doesn ' t exist as it is optional .
try :
    uuids = self . _props . Get ( _INTERFACE , 'UUIDs' )
except dbus . exceptions . DBusException as ex : # Ignore error if device has no UUIDs property ( i . e . might not be
# a BLE device ) .
    if ex . get_dbus_name ( ) != 'org.freedesktop.DBus.Error.InvalidArgs' :
        raise ex
return [ uuid . UUID ( str ( x ) ) for x in uuids ]


signature:
def find_characteristic ( self , uuid ) :
    """Return the first child characteristic found that has the specified
 UUID . Will return None if no characteristic that matches is found ."""

implementation:
for char in self . list_characteristics ( ) :
    if char . uuid == uuid :
        return char
return None


signature:
def find_descriptor ( self , uuid ) :
    """Return the first child descriptor found that has the specified
 UUID . Will return None if no descriptor that matches is found ."""

implementation:
for desc in self . list_descriptors ( ) :
    if desc . uuid == uuid :
        return desc
return None


signature:
def read_value ( self , timeout_sec = TIMEOUT_SEC ) :
    """Read the value of this characteristic ."""

implementation:
# Kick off a query to read the value of the characteristic , then wait
# for the result to return asyncronously .
self . _value_read . clear ( )
self . _device . _peripheral . readValueForCharacteristic_ ( self . _characteristic )
if not self . _value_read . wait ( timeout_sec ) :
    raise RuntimeError ( 'Exceeded timeout waiting to read characteristic value!' )
return self . _characteristic . value ( )


signature:
def write_value ( self , value , write_type = 0 ) :
    """Write the specified value to this characteristic ."""

implementation:
data = NSData . dataWithBytes_length_ ( value , len ( value ) )
self . _device . _peripheral . writeValue_forCharacteristic_type_ ( data , self . _characteristic , write_type )


signature:
def start_notify ( self , on_change ) :
    """Enable notification of changes for this characteristic on the
 specified on _ change callback . on _ change should be a function that takes
 one parameter which is the value ( as a string of bytes ) of the changed
 characteristic value ."""

implementation:
# Tell the device what callback to use for changes to this characteristic .
self . _device . _notify_characteristic ( self . _characteristic , on_change )
# Turn on notifications of characteristic changes .
self . _device . _peripheral . setNotifyValue_forCharacteristic_ ( True , self . _characteristic )


signature:
def read_value ( self ) :
    """Read the value of this descriptor ."""

implementation:
pass
# Kick off a query to read the value of the descriptor , then wait
# for the result to return asyncronously .
self . _value_read . clear ( )
self . _device . _peripheral . readValueForDescriptor ( self . _descriptor )
if not self . _value_read . wait ( timeout_sec ) :
    raise RuntimeError ( 'Exceeded timeout waiting to read characteristic value!' )
return self . _value


signature:
def start_scan ( self , timeout_sec = TIMEOUT_SEC ) :
    """Start scanning for BLE devices with this adapter ."""

implementation:
self . _scan_started . clear ( )
self . _adapter . StartDiscovery ( )
if not self . _scan_started . wait ( timeout_sec ) :
    raise RuntimeError ( 'Exceeded timeout waiting for adapter to start scanning!' )


signature:
def stop_scan ( self , timeout_sec = TIMEOUT_SEC ) :
    """Stop scanning for BLE devices with this adapter ."""

implementation:
self . _scan_stopped . clear ( )
self . _adapter . StopDiscovery ( )
if not self . _scan_stopped . wait ( timeout_sec ) :
    raise RuntimeError ( 'Exceeded timeout waiting for adapter to stop scanning!' )


signature:
def centralManagerDidUpdateState_ ( self , manager ) :
    """Called when the BLE adapter is powered on and ready to scan / connect
 to devices ."""

implementation:
logger . debug ( 'centralManagerDidUpdateState called' )
# Notify adapter about changed central state .
get_provider ( ) . _adapter . _state_changed ( manager . state ( ) )


signature:
def centralManager_didDiscoverPeripheral_advertisementData_RSSI_ ( self , manager , peripheral , data , rssi ) :
    """Called when the BLE adapter found a device while scanning , or has
 new advertisement data for a device ."""

implementation:
logger . debug ( 'centralManager_didDiscoverPeripheral_advertisementData_RSSI called' )
# Log name of device found while scanning .
# logger . debug ( ' Saw device advertised with name : { 0 } ' . format ( peripheral . name ( ) ) )
# Make sure the device is added to the list of devices and then update
# its advertisement state .
device = device_list ( ) . get ( peripheral )
if device is None :
    device = device_list ( ) . add ( peripheral , CoreBluetoothDevice ( peripheral ) )
device . _update_advertised ( data )


signature:
def centralManager_didConnectPeripheral_ ( self , manager , peripheral ) :
    """Called when a device is connected ."""

implementation:
logger . debug ( 'centralManager_didConnectPeripheral called' )
# Setup peripheral delegate and kick off service discovery . For now just
# assume all services need to be discovered .
peripheral . setDelegate_ ( self )
peripheral . discoverServices_ ( None )
# Fire connected event for device .
device = device_list ( ) . get ( peripheral )
if device is not None :
    device . _set_connected ( )


signature:
def centralManager_didDisconnectPeripheral_error_ ( self , manager , peripheral , error ) :
    """Called when a device is disconnected ."""

implementation:
logger . debug ( 'centralManager_didDisconnectPeripheral called' )
# Get the device and remove it from the device list , then fire its
# disconnected event .
device = device_list ( ) . get ( peripheral )
if device is not None : # Fire disconnected event and remove device from device list .
    device . _set_disconnected ( )
    device_list ( ) . remove ( peripheral )


signature:
def peripheral_didDiscoverServices_ ( self , peripheral , services ) :
    """Called when services are discovered for a device ."""

implementation:
logger . debug ( 'peripheral_didDiscoverServices called' )
# Make sure the discovered services are added to the list of known
# services , and kick off characteristic discovery for each one .
# NOTE : For some reason the services parameter is never set to a good
# value , instead you must query peripheral . services ( ) to enumerate the
# discovered services .
for service in peripheral . services ( ) :
    if service_list ( ) . get ( service ) is None :
        service_list ( ) . add ( service , CoreBluetoothGattService ( service ) )
    # Kick off characteristic discovery for this service . Just discover
    # all characteristics for now .
    peripheral . discoverCharacteristics_forService_ ( None , service )


signature:
def peripheral_didDiscoverCharacteristicsForService_error_ ( self , peripheral , service , error ) :
    """Called when characteristics are discovered for a service ."""

implementation:
logger . debug ( 'peripheral_didDiscoverCharacteristicsForService_error called' )
# Stop if there was some kind of error .
if error is not None :
    return
# Make sure the discovered characteristics are added to the list of known
# characteristics , and kick off descriptor discovery for each char .
for char in service . characteristics ( ) : # Add to list of known characteristics .
    if characteristic_list ( ) . get ( char ) is None :
        characteristic_list ( ) . add ( char , CoreBluetoothGattCharacteristic ( char ) )
    # Start descriptor discovery .
    peripheral . discoverDescriptorsForCharacteristic_ ( char )
# Notify the device about the discovered characteristics .
device = device_list ( ) . get ( peripheral )
if device is not None :
    device . _characteristics_discovered ( service )


signature:
def peripheral_didDiscoverDescriptorsForCharacteristic_error_ ( self , peripheral , characteristic , error ) :
    """Called when characteristics are discovered for a service ."""

implementation:
logger . debug ( 'peripheral_didDiscoverDescriptorsForCharacteristic_error called' )
# Stop if there was some kind of error .
if error is not None :
    return
# Make sure the discovered descriptors are added to the list of known
# descriptors .
for desc in characteristic . descriptors ( ) : # Add to list of known descriptors .
    if descriptor_list ( ) . get ( desc ) is None :
        descriptor_list ( ) . add ( desc , CoreBluetoothGattDescriptor ( desc ) )


signature:
def peripheral_didUpdateValueForCharacteristic_error_ ( self , peripheral , characteristic , error ) :
    """Called when characteristic value was read or updated ."""

implementation:
logger . debug ( 'peripheral_didUpdateValueForCharacteristic_error called' )
# Stop if there was some kind of error .
if error is not None :
    return
# Notify the device about the updated characteristic value .
device = device_list ( ) . get ( peripheral )
if device is not None :
    device . _characteristic_changed ( characteristic )


signature:
def peripheral_didUpdateValueForDescriptor_error_ ( self , peripheral , descriptor , error ) :
    """Called when descriptor value was read or updated ."""

implementation:
logger . debug ( 'peripheral_didUpdateValueForDescriptor_error called' )
# Stop if there was some kind of error .
if error is not None :
    return
# Notify the device about the updated descriptor value .
device = device_list ( ) . get ( peripheral )
if device is not None :
    device . _descriptor_changed ( descriptor )


signature:
def peripheral_didReadRSSI_error_ ( self , peripheral , rssi , error ) :
    """Called when a new RSSI value for the peripheral is available ."""

implementation:
logger . debug ( 'peripheral_didReadRSSI_error called' )
# Note this appears to be completely undocumented at the time of this
# writing . Can see more details at :
# http : / / stackoverflow . com / questions / 25952218 / ios - 8 - corebluetooth - deprecated - rssi - methods
# Stop if there was some kind of error .
if error is not None :
    return
# Notify the device about the updated RSSI value .
device = device_list ( ) . get ( peripheral )
if device is not None :
    device . _rssi_changed ( rssi )


signature:
def initialize ( self ) :
    """Initialize the BLE provider . Must be called once before any other
 calls are made to the provider ."""

implementation:
# Setup the central manager and its delegate .
self . _central_manager = CBCentralManager . alloc ( )
self . _central_manager . initWithDelegate_queue_options_ ( self . _central_delegate , None , None )


signature:
def run_mainloop_with ( self , target ) :
    """Start the OS ' s main loop to process asyncronous BLE events and then
 run the specified target function in a background thread . Target
 function should be a function that takes no parameters and optionally
 return an integer response code . When the target function stops
 executing or returns with value then the main loop will be stopped and
 the program will exit with the returned code .
 Note that an OS main loop is required to process asyncronous BLE events
 and this function is provided as a convenience for writing simple tools
 and scripts that don ' t need to be full - blown GUI applications . If you
 are writing a GUI application that has a main loop ( a GTK glib main loop
 on Linux , or a Cocoa main loop on OSX ) then you don ' t need to call this
 function ."""

implementation:
# Create background thread to run user code .
self . _user_thread = threading . Thread ( target = self . _user_thread_main , args = ( target , ) )
self . _user_thread . daemon = True
self . _user_thread . start ( )
# Run main loop . This call will never return !
try :
    AppHelper . runConsoleEventLoop ( installInterrupt = True )
except KeyboardInterrupt :
    AppHelper . stopEventLoop ( )
    sys . exit ( 0 )


signature:
def _user_thread_main ( self , target ) :
    """Main entry point for the thread that will run user ' s code ."""

implementation:
try : # Run user ' s code .
    return_code = target ( )
    # Assume good result ( 0 return code ) if none is returned .
    if return_code is None :
        return_code = 0
    # Call exit on the main thread when user code has finished .
    AppHelper . callAfter ( lambda : sys . exit ( return_code ) )
except Exception as ex : # Something went wrong . Raise the exception on the main thread to exit .
    AppHelper . callAfter ( self . _raise_error , sys . exc_info ( ) )


signature:
def clear_cached_data ( self ) :
    """Clear the internal bluetooth device cache . This is useful if a device
 changes its state like name and it can ' t be detected with the new state
 anymore . WARNING : This will delete some files underneath the running user ' s
 ~ / Library / Preferences / folder !
 See this Stackoverflow question for information on what the function does :
 http : / / stackoverflow . com / questions / 20553957 / how - can - i - clear - the - corebluetooth - cache - on - macos"""

implementation:
# Turn off bluetooth .
if self . _adapter . is_powered :
    self . _adapter . power_off ( )
# Delete cache files and suppress any stdout / err output .
with open ( os . devnull , 'w' ) as devnull :
    subprocess . call ( 'rm ~/Library/Preferences/com.apple.Bluetooth.plist' , shell = True , stdout = devnull , stderr = subprocess . STDOUT )
    subprocess . call ( 'rm ~/Library/Preferences/ByHost/com.apple.Bluetooth.*.plist' , shell = True , stdout = devnull , stderr = subprocess . STDOUT )


signature:
def disconnect_devices ( self , service_uuids ) :
    """Disconnect any connected devices that have any of the specified
 service UUIDs ."""

implementation:
# Get list of connected devices with specified services .
cbuuids = map ( uuid_to_cbuuid , service_uuids )
for device in self . _central_manager . retrieveConnectedPeripheralsWithServices_ ( cbuuids ) :
    self . _central_manager . cancelPeripheralConnection_ ( device )


signature:
def initialize ( self ) :
    """Initialize bluez DBus communication . Must be called before any other
 calls are made !"""

implementation:
# Ensure GLib ' s threading is initialized to support python threads , and
# make a default mainloop that all DBus objects will inherit . These
# commands MUST execute before any other DBus commands !
GObject . threads_init ( )
dbus . mainloop . glib . threads_init ( )
# Set the default main loop , this also MUST happen before other DBus calls .
self . _mainloop = dbus . mainloop . glib . DBusGMainLoop ( set_as_default = True )
# Get the main DBus system bus and root bluez object .
self . _bus = dbus . SystemBus ( )
self . _bluez = dbus . Interface ( self . _bus . get_object ( 'org.bluez' , '/' ) , 'org.freedesktop.DBus.ObjectManager' )


implementation:
# Spin up a background thread to run the target code .
self . _user_thread = threading . Thread ( target = self . _user_thread_main , args = ( target , ) )
self . _user_thread . daemon = True
# Don ' t let the user thread block exit .
self . _user_thread . start ( )
# Spin up a GLib main loop in the main thread to process async BLE events .
self . _gobject_mainloop = GObject . MainLoop ( )
try :
    self . _gobject_mainloop . run ( )
    # Doesn ' t return until the mainloop ends .
except KeyboardInterrupt :
    self . _gobject_mainloop . quit ( )
    sys . exit ( 0 )
# Main loop finished . Check if an exception occured and throw it ,
# otherwise return the status code from the user code .
if self . _exception is not None : # Rethrow exception with its original stack trace following advice from :
# http : / / nedbatchelder . com / blog / 200711 / rethrowing _ exceptions _ in _ python . html
    raise_ ( self . _exception [ 1 ] , None , self . _exception [ 2 ] )
else :
    sys . exit ( self . _return_code )


implementation:
try : # Wait for GLib main loop to start running before starting user code .
    while True :
        if self . _gobject_mainloop is not None and self . _gobject_mainloop . is_running ( ) : # Main loop is running , we should be ready to make bluez DBus calls .
            break
        # Main loop isn ' t running yet , give time back to other threads .
        time . sleep ( 0 )
    # Run user ' s code .
    self . _return_code = target ( )
    # Assume good result ( 0 return code ) if none is returned .
    if self . _return_code is None :
        self . _return_code = 0
    # Signal the main loop to exit .
    self . _gobject_mainloop . quit ( )
except Exception as ex : # Something went wrong . Raise the exception on the main thread to
# exit .
    self . _exception = sys . exc_info ( )
    self . _gobject_mainloop . quit ( )


signature:
def clear_cached_data ( self ) :
    """Clear any internally cached BLE device data . Necessary in some cases
 to prevent issues with stale device data getting cached by the OS ."""

implementation:
# Go through and remove any device that isn ' t currently connected .
for device in self . list_devices ( ) : # Skip any connected device .
    if device . is_connected :
        continue
    # Remove this device . First get the adapter associated with the device .
    adapter = dbus . Interface ( self . _bus . get_object ( 'org.bluez' , device . _adapter ) , _ADAPTER_INTERFACE )
    # Now call RemoveDevice on the adapter to remove the device from
    # bluez ' s DBus hierarchy .
    adapter . RemoveDevice ( device . _device . object_path )


signature:
def disconnect_devices ( self , service_uuids = [ ] ) :
    """Disconnect any connected devices that have the specified list of
 service UUIDs . The default is an empty list which means all devices
 are disconnected ."""

implementation:
service_uuids = set ( service_uuids )
for device in self . list_devices ( ) : # Skip devices that aren ' t connected .
    if not device . is_connected :
        continue
    device_uuids = set ( map ( lambda x : x . uuid , device . list_services ( ) ) )
    if device_uuids >= service_uuids : # Found a device that has at least the requested services , now
    # disconnect from it .
        device . disconnect ( )


signature:
def _get_objects ( self , interface , parent_path = '/org/bluez' ) :
    """Return a list of all bluez DBus objects that implement the requested
 interface name and are under the specified path . The default is to
 search devices under the root of all bluez objects ."""

implementation:
# Iterate through all the objects in bluez ' s DBus hierarchy and return
# any that implement the requested interface under the specified path .
parent_path = parent_path . lower ( )
objects = [ ]
for opath , interfaces in iteritems ( self . _bluez . GetManagedObjects ( ) ) :
    if interface in interfaces . keys ( ) and opath . lower ( ) . startswith ( parent_path ) :
        objects . append ( self . _bus . get_object ( 'org.bluez' , opath ) )
return objects


signature:
def _get_objects_by_path ( self , paths ) :
    """Return a list of all bluez DBus objects from the provided list of paths ."""

implementation: return map ( lambda x : self . _bus . get_object ( 'org.bluez' , x ) , paths )


signature:
def _print_tree ( self ) :
    """Print tree of all bluez objects , useful for debugging ."""

implementation:
# This is based on the bluez sample code get - managed - objects . py .
objects = self . _bluez . GetManagedObjects ( )
for path in objects . keys ( ) :
    print ( "[ %s ]" % ( path ) )
    interfaces = objects [ path ]
    for interface in interfaces . keys ( ) :
        if interface in [ "org.freedesktop.DBus.Introspectable" , "org.freedesktop.DBus.Properties" ] :
            continue
        print ( "    %s" % ( interface ) )
        properties = interfaces [ interface ]
        for key in properties . keys ( ) :
            print ( "      %s = %s" % ( key , properties [ key ] ) )


signature:
def find_devices ( self , service_uuids = [ ] , name = None ) :
    """Return devices that advertise the specified service UUIDs and / or have
 the specified name . Service _ uuids should be a list of Python uuid . UUID
 objects and is optional . Name is a string device name to look for and is
 also optional . Will not block , instead it returns immediately with a
 list of found devices ( which might be empty ) ."""

implementation:
# Convert service UUID list to counter for quicker comparison .
expected = set ( service_uuids )
# Grab all the devices .
devices = self . list_devices ( )
# Filter to just the devices that have the requested service UUID / name .
found = [ ]
for device in devices :
    if name is not None :
        if device . name == name : # Check if the name matches and add the device .
            found . append ( device )
    else : # Check if the advertised UUIDs have at least the expected UUIDs .
        actual = set ( device . advertised )
        if actual >= expected :
            found . append ( device )
return found


signature:
def find_device ( self , service_uuids = [ ] , name = None , timeout_sec = TIMEOUT_SEC ) :
    """Return the first device that advertises the specified service UUIDs or
 has the specified name . Will wait up to timeout _ sec seconds for the device
 to be found , and if the timeout is zero then it will not wait at all and
 immediately return a result . When no device is found a value of None is
 returned ."""

implementation:
start = time . time ( )
while True : # Call find _ devices and grab the first result if any are found .
    found = self . find_devices ( service_uuids , name )
    if len ( found ) > 0 :
        return found [ 0 ]
    # No device was found . Check if the timeout is exceeded and wait to
    # try again .
    if time . time ( ) - start >= timeout_sec : # Failed to find a device within the timeout .
        return None
    time . sleep ( 1 )


signature:
def get_all ( self , cbobjects ) :
    """Retrieve a list of metadata objects associated with the specified
 list of CoreBluetooth objects . If an object cannot be found then an
 exception is thrown ."""

implementation:
try :
    with self . _lock :
        return [ self . _metadata [ x ] for x in cbobjects ]
except KeyError : # Note that if this error gets thrown then the assumption that OSX
# will pass back to callbacks the exact CoreBluetooth objects that
# were used previously is broken ! ( i . e . the CoreBluetooth objects
# are not stateless )
    raise RuntimeError ( 'Failed to find expected metadata for CoreBluetooth object!' )


signature:
def add ( self , cbobject , metadata ) :
    """Add the specified CoreBluetooth item with the associated metadata if
 it doesn ' t already exist . Returns the newly created or preexisting
 metadata item ."""

implementation:
with self . _lock :
    if cbobject not in self . _metadata :
        self . _metadata [ cbobject ] = metadata
    return self . _metadata [ cbobject ]


signature:
def remove ( self , cbobject ) :
    """Remove any metadata associated with the provided CoreBluetooth object ."""

implementation:
with self . _lock :
    if cbobject in self . _metadata :
        del self . _metadata [ cbobject ]


signature:
def cbuuid_to_uuid ( cbuuid ) :
    """Convert Objective - C CBUUID type to native Python UUID type ."""

implementation:
data = cbuuid . data ( ) . bytes ( )
template = '{:0>8}-0000-1000-8000-00805f9b34fb' if len ( data ) <= 4 else '{:0>32}'
value = template . format ( hexlify ( data . tobytes ( ) [ : 16 ] ) . decode ( 'ascii' ) )
return uuid . UUID ( hex = value )


signature:
def set_color ( self , r , g , b ) :
    """Set the red , green , blue color of the bulb ."""

implementation:
# See more details on the bulb ' s protocol from this guide :
# https : / / learn . adafruit . com / reverse - engineering - a - bluetooth - low - energy - light - bulb / overview
command = '\x58\x01\x03\x01\xFF\x00{0}{1}{2}' . format ( chr ( r & 0xFF ) , chr ( g & 0xFF ) , chr ( b & 0xFF ) )
self . _color . write_value ( command )


signature:
def get_provider ( ) :
    """Return an instance of the BLE provider for the current platform ."""

implementation:
global _provider
# Set the provider based on the current platform .
if _provider is None :
    if sys . platform . startswith ( 'linux' ) : # Linux platform
        from . bluez_dbus . provider import BluezProvider
        _provider = BluezProvider ( )
    elif sys . platform == 'darwin' : # Mac OSX platform
        from . corebluetooth . provider import CoreBluetoothProvider
        _provider = CoreBluetoothProvider ( )
    else : # Unsupported platform
        raise RuntimeError ( 'Sorry the {0} platform is not supported by the BLE library!' . format ( sys . platform ) )
return _provider


signature:
def toBigInt ( byteArray ) :
    """Convert the byte array to a BigInteger"""

implementation:
array = byteArray [ : : - 1 ]
# reverse array
out = 0
for key , value in enumerate ( array ) :
    decoded = struct . unpack ( "B" , bytes ( [ value ] ) ) [ 0 ]
    out = out | decoded << key * 8
return out


signature:
def encryptPassword ( self , login , passwd ) :
    """Encrypt credentials using the google publickey , with the
 RSA algorithm"""

implementation:
# structure of the binary key :
# | modulus _ length | modulus | exponent _ length | exponent |
# modulus _ length and exponent _ length are uint32
binaryKey = b64decode ( config . GOOGLE_PUBKEY )
# modulus
i = utils . readInt ( binaryKey , 0 )
modulus = utils . toBigInt ( binaryKey [ 4 : ] [ 0 : i ] )
# exponent
j = utils . readInt ( binaryKey , i + 4 )
exponent = utils . toBigInt ( binaryKey [ i + 8 : ] [ 0 : j ] )
# calculate SHA1 of the pub key
digest = hashes . Hash ( hashes . SHA1 ( ) , backend = default_backend ( ) )
digest . update ( binaryKey )
h = b'\x00' + digest . finalize ( ) [ 0 : 4 ]
# generate a public key
der_data = encode_dss_signature ( modulus , exponent )
publicKey = load_der_public_key ( der_data , backend = default_backend ( ) )
# encrypt email and password using pubkey
to_be_encrypted = login . encode ( ) + b'\x00' + passwd . encode ( )
ciphertext = publicKey . encrypt ( to_be_encrypted , padding . OAEP ( mgf = padding . MGF1 ( algorithm = hashes . SHA1 ( ) ) , algorithm = hashes . SHA1 ( ) , label = None ) )
return urlsafe_b64encode ( h + ciphertext )


signature:
def getHeaders ( self , upload_fields = False ) :
    """Return the default set of request headers , which
 can later be expanded , based on the request type"""

implementation:
if upload_fields :
    headers = self . deviceBuilder . getDeviceUploadHeaders ( )
else :
    headers = self . deviceBuilder . getBaseHeaders ( )
if self . gsfId is not None :
    headers [ "X-DFE-Device-Id" ] = "{0:x}" . format ( self . gsfId )
if self . authSubToken is not None :
    headers [ "Authorization" ] = "GoogleLogin auth=%s" % self . authSubToken
if self . device_config_token is not None :
    headers [ "X-DFE-Device-Config-Token" ] = self . device_config_token
if self . deviceCheckinConsistencyToken is not None :
    headers [ "X-DFE-Device-Checkin-Consistency-Token" ] = self . deviceCheckinConsistencyToken
if self . dfeCookie is not None :
    headers [ "X-DFE-Cookie" ] = self . dfeCookie
return headers


signature:
def uploadDeviceConfig ( self ) :
    """Upload the device configuration of the fake device
 selected in the _ _ init _ _ methodi to the google account ."""

implementation:
upload = googleplay_pb2 . UploadDeviceConfigRequest ( )
upload . deviceConfiguration . CopyFrom ( self . deviceBuilder . getDeviceConfig ( ) )
headers = self . getHeaders ( upload_fields = True )
stringRequest = upload . SerializeToString ( )
response = requests . post ( UPLOAD_URL , data = stringRequest , headers = headers , verify = ssl_verify , timeout = 60 , proxies = self . proxies_config )
response = googleplay_pb2 . ResponseWrapper . FromString ( response . content )
try :
    if response . payload . HasField ( 'uploadDeviceConfigResponse' ) :
        self . device_config_token = response . payload . uploadDeviceConfigResponse
        self . device_config_token = self . device_config_token . uploadDeviceConfigToken
except ValueError :
    pass


signature:
def login ( self , email = None , password = None , gsfId = None , authSubToken = None ) :
    """Login to your Google Account .
 For first time login you should provide :
 * email
 * password
 For the following logins you need to provide :
 * gsfId
 * authSubToken"""

implementation:
if email is not None and password is not None : # First time setup , where we obtain an ac2dm token and
# upload device information
    encryptedPass = self . encryptPassword ( email , password ) . decode ( 'utf-8' )
    # AC2DM token
    params = self . deviceBuilder . getLoginParams ( email , encryptedPass )
    params [ 'service' ] = 'ac2dm'
    params [ 'add_account' ] = '1'
    params [ 'callerPkg' ] = 'com.google.android.gms'
    headers = self . deviceBuilder . getAuthHeaders ( self . gsfId )
    headers [ 'app' ] = 'com.google.android.gsm'
    response = requests . post ( AUTH_URL , data = params , verify = ssl_verify , proxies = self . proxies_config )
    data = response . text . split ( )
    params = { }
    for d in data :
        if "=" not in d :
            continue
        k , v = d . split ( "=" , 1 )
        params [ k . strip ( ) . lower ( ) ] = v . strip ( )
    if "auth" in params :
        ac2dmToken = params [ "auth" ]
    elif "error" in params :
        if "NeedsBrowser" in params [ "error" ] :
            raise SecurityCheckError ( "Security check is needed, try to visit " "https://accounts.google.com/b/0/DisplayUnlockCaptcha " "to unlock, or setup an app-specific password" )
        raise LoginError ( "server says: " + params [ "error" ] )
    else :
        raise LoginError ( "Auth token not found." )
    self . gsfId = self . checkin ( email , ac2dmToken )
    self . getAuthSubToken ( email , encryptedPass )
    self . uploadDeviceConfig ( )
elif gsfId is not None and authSubToken is not None : # no need to initialize API
    self . gsfId = gsfId
    self . setAuthSubToken ( authSubToken )
    # check if token is valid with a simple search
    self . search ( 'drv' )
else :
    raise LoginError ( 'Either (email,pass) or (gsfId, authSubToken) is needed' )


signature:
def search ( self , query ) :
    """Search the play store for an app .
 nb _ result ( int ) : is the maximum number of result to be returned
 offset ( int ) : is used to take result starting from an index ."""

implementation:
if self . authSubToken is None :
    raise LoginError ( "You need to login before executing any request" )
path = SEARCH_URL + "?c=3&q={}" . format ( requests . utils . quote ( query ) )
# FIXME : not sure if this toc call should be here
self . toc ( )
data = self . executeRequestApi2 ( path )
if utils . hasPrefetch ( data ) :
    response = data . preFetch [ 0 ] . response
else :
    response = data
resIterator = response . payload . listResponse . doc
return list ( map ( utils . parseProtobufObj , resIterator ) )


signature:
def details ( self , packageName ) :
    """Get app details from a package name .
 packageName is the app unique ID ( usually starting with ' com . ' ) ."""

implementation:
path = DETAILS_URL + "?doc={}" . format ( requests . utils . quote ( packageName ) )
data = self . executeRequestApi2 ( path )
return utils . parseProtobufObj ( data . payload . detailsResponse . docV2 )


signature:
def bulkDetails ( self , packageNames ) :
    """Get several apps details from a list of package names .
 This is much more efficient than calling N times details ( ) since it
 requires only one request . If an item is not found it returns an empty object
 instead of throwing a RequestError ( ' Item not found ' ) like the details ( ) function
 Args :
 packageNames ( list ) : a list of app IDs ( usually starting with ' com . ' ) .
 Returns :
 a list of dictionaries containing docv2 data , or None
 if the app doesn ' t exist"""

implementation:
params = { 'au' : '1' }
req = googleplay_pb2 . BulkDetailsRequest ( )
req . docid . extend ( packageNames )
data = req . SerializeToString ( )
message = self . executeRequestApi2 ( BULK_URL , post_data = data . decode ( "utf-8" ) , content_type = CONTENT_TYPE_PROTO , params = params )
response = message . payload . bulkDetailsResponse
return [ None if not utils . hasDoc ( entry ) else utils . parseProtobufObj ( entry . doc ) for entry in response . entry ]


signature:
def browse ( self , cat = None , subCat = None ) :
    """Browse categories . If neither cat nor subcat are specified ,
 return a list of categories , otherwise it return a list of apps
 using cat ( category ID ) and subCat ( subcategory ID ) as filters ."""

implementation:
path = BROWSE_URL + "?c=3"
if cat is not None :
    path += "&cat={}" . format ( requests . utils . quote ( cat ) )
if subCat is not None :
    path += "&ctr={}" . format ( requests . utils . quote ( subCat ) )
data = self . executeRequestApi2 ( path )
return utils . parseProtobufObj ( data . payload . browseResponse )


signature:
def list ( self , cat , ctr = None , nb_results = None , offset = None ) :
    """List all possible subcategories for a specific category . If
 also a subcategory is provided , list apps from this category .
 Args :
 cat ( str ) : category id
 ctr ( str ) : subcategory id
 nb _ results ( int ) : if a subcategory is specified , limit number
 of results to this number
 offset ( int ) : if a subcategory is specified , start counting from this
 result
 Returns :
 A list of categories . If subcategory is specified , a list of apps in this
 category ."""

implementation:
path = LIST_URL + "?c=3&cat={}" . format ( requests . utils . quote ( cat ) )
if ctr is not None :
    path += "&ctr={}" . format ( requests . utils . quote ( ctr ) )
if nb_results is not None :
    path += "&n={}" . format ( requests . utils . quote ( str ( nb_results ) ) )
if offset is not None :
    path += "&o={}" . format ( requests . utils . quote ( str ( offset ) ) )
data = self . executeRequestApi2 ( path )
clusters = [ ]
docs = [ ]
if ctr is None : # list subcategories
    for pf in data . preFetch :
        for cluster in pf . response . payload . listResponse . doc :
            clusters . extend ( cluster . child )
    return [ c . docid for c in clusters ]
else :
    apps = [ ]
    for d in data . payload . listResponse . doc : # categories
        for c in d . child : # sub - category
            for a in c . child : # app
                apps . append ( utils . parseProtobufObj ( a ) )
    return apps


signature:
def reviews ( self , packageName , filterByDevice = False , sort = 2 , nb_results = None , offset = None ) :
    """Browse reviews for an application
 Args :
 packageName ( str ) : app unique ID .
 filterByDevice ( bool ) : filter results for current device
 sort ( int ) : sorting criteria ( values are unknown )
 nb _ results ( int ) : max number of reviews to return
 offset ( int ) : return reviews starting from an offset value
 Returns :
 dict object containing all the protobuf data returned from
 the api"""

implementation:
# TODO : select the number of reviews to return
path = REVIEWS_URL + "?doc={}&sort={}" . format ( requests . utils . quote ( packageName ) , sort )
if nb_results is not None :
    path += "&n={}" . format ( nb_results )
if offset is not None :
    path += "&o={}" . format ( offset )
if filterByDevice :
    path += "&dfil=1"
data = self . executeRequestApi2 ( path )
output = [ ]
for review in data . payload . reviewResponse . getResponse . review :
    output . append ( utils . parseProtobufObj ( review ) )
return output