signature:
def yaml_get_data ( filename ) :
    """Get data from . yml file"""

implementation:
with open ( filename , 'rb' ) as fd :
    yaml_data = yaml . load ( fd )
    return yaml_data
return False


signature:
def yaml_write_data ( yaml_data , filename ) :
    """Write data into a . yml file"""

implementation:
with open ( filename , 'w' ) as fd :
    yaml . dump ( yaml_data , fd , default_flow_style = False )
    return True
return False


signature:
def fit ( self , X , y = None ) :
    '''If scale _ by _ median , find : attr : ` median _ ` ; otherwise , do nothing .
 Parameters
 X : array
 The raw pairwise distances .'''

implementation:
X = check_array ( X )
if self . scale_by_median :
    self . median_ = np . median ( X [ np . triu_indices_from ( X , k = 1 ) ] , overwrite_input = True )
elif hasattr ( self , 'median_' ) :
    del self . median_
return self


signature:
def transform ( self , X ) :
    '''Turns distances into RBF values .
 Parameters
 X : array
 The raw pairwise distances .
 Returns
 X _ rbf : array of same shape as X
 The distances in X passed through the RBF kernel .'''

implementation:
X = check_array ( X )
X_rbf = np . empty_like ( X ) if self . copy else X
X_in = X
if not self . squared :
    np . power ( X_in , 2 , out = X_rbf )
    X_in = X_rbf
if self . scale_by_median :
    scale = self . median_ if self . squared else self . median_ ** 2
    gamma = self . gamma * scale
else :
    gamma = self . gamma
np . multiply ( X_in , - gamma , out = X_rbf )
np . exp ( X_rbf , out = X_rbf )
return X_rbf


signature:
def fit ( self , X , y = None ) :
    '''Learn the linear transformation to clipped eigenvalues .
 Note that if min _ eig isn ' t zero and any of the original eigenvalues
 were exactly zero , this will leave those eigenvalues as zero .
 Parameters
 X : array , shape [ n , n ]
 The * symmetric * input similarities . If X is asymmetric , it will be
 treated as if it were symmetric based on its lower - triangular part .'''

implementation:
n = X . shape [ 0 ]
if X . shape != ( n , n ) :
    raise TypeError ( "Input must be a square matrix." )
# TODO : only get negative eigs somehow ?
memory = get_memory ( self . memory )
vals , vecs = memory . cache ( scipy . linalg . eigh , ignore = [ 'overwrite_a' ] ) ( X , overwrite_a = not self . copy )
vals = vals . reshape ( - 1 , 1 )
if self . min_eig == 0 :
    inner = vals > self . min_eig
else :
    with np . errstate ( divide = 'ignore' ) :
        inner = np . where ( vals >= self . min_eig , 1 , np . where ( vals == 0 , 0 , self . min_eig / vals ) )
self . clip_ = np . dot ( vecs , inner * vecs . T )
return self


signature:
def fit ( self , X , y = None ) :
    '''Learn the linear transformation to flipped eigenvalues .
 Parameters
 X : array , shape [ n , n ]
 The * symmetric * input similarities . If X is asymmetric , it will be
 treated as if it were symmetric based on its lower - triangular part .'''

implementation:
n = X . shape [ 0 ]
if X . shape != ( n , n ) :
    raise TypeError ( "Input must be a square matrix." )
# TODO : only get negative eigs somehow ?
memory = get_memory ( self . memory )
vals , vecs = memory . cache ( scipy . linalg . eigh , ignore = [ 'overwrite_a' ] ) ( X , overwrite_a = not self . copy )
vals = vals [ : , None ]
self . flip_ = np . dot ( vecs , np . sign ( vals ) * vecs . T )
return self


signature:
def transform ( self , X ) :
    '''Transforms X according to the linear transformation corresponding to
 flipping the input eigenvalues .
 Parameters
 X : array , shape [ n _ test , n ]
 The test similarities to training points .
 Returns
 Xt : array , shape [ n _ test , n ]
 The transformed test similarites to training points .'''

implementation:
n = self . flip_ . shape [ 0 ]
if X . ndim != 2 or X . shape [ 1 ] != n :
    msg = "X should have {} columns, the number of samples at fit time"
    raise TypeError ( msg . format ( self . flip_ . shape [ 0 ] ) )
return np . dot ( X , self . flip_ )


signature:
def fit_transform ( self , X , y = None ) :
    '''Flips the negative eigenvalues of X .
 Parameters
 X : array , shape [ n , n ]
 The * symmetric * input similarities . If X is asymmetric , it will be
 treated as if it were symmetric based on its lower - triangular part .
 Returns
 Xt : array , shape [ n , n ]
 The transformed training similarities .'''

implementation:
n = X . shape [ 0 ]
if X . shape != ( n , n ) :
    raise TypeError ( "Input must be a square matrix." )
memory = get_memory ( self . memory )
discard_X = not self . copy and self . negatives_likely
vals , vecs = memory . cache ( scipy . linalg . eigh , ignore = [ 'overwrite_a' ] ) ( X , overwrite_a = discard_X )
vals = vals [ : , None ]
self . clip_ = np . dot ( vecs , np . sign ( vals ) * vecs . T )
if discard_X or vals [ 0 , 0 ] < 0 :
    del X
    np . abs ( vals , out = vals )
    X = np . dot ( vecs , vals * vecs . T )
    del vals , vecs
    # should be symmetric , but make sure because floats
    X = Symmetrize ( copy = False ) . fit_transform ( X )
return X


signature:
def fit ( self , X , y = None ) :
    '''Learn the transformation to shifted eigenvalues . Only depends
 on the input dimension .
 Parameters
 X : array , shape [ n , n ]
 The * symmetric * input similarities .'''

implementation:
n = X . shape [ 0 ]
if X . shape != ( n , n ) :
    raise TypeError ( "Input must be a square matrix." )
self . train_ = X
memory = get_memory ( self . memory )
lo , = memory . cache ( scipy . linalg . eigvalsh ) ( X , eigvals = ( 0 , 0 ) )
self . shift_ = max ( self . min_eig - lo , 0 )
return self


signature:
def transform ( self , X ) :
    '''Transforms X according to the linear transformation corresponding to
 shifting the input eigenvalues to all be at least ` ` self . min _ eig ` ` .
 Parameters
 X : array , shape [ n _ test , n ]
 The test similarities to training points .
 Returns
 Xt : array , shape [ n _ test , n ]
 The transformed test similarites to training points . Only different
 from X if X is the training data .'''

implementation:
n = self . train_ . shape [ 0 ]
if X . ndim != 2 or X . shape [ 1 ] != n :
    msg = "X should have {} columns, the number of samples at fit time"
    raise TypeError ( msg . format ( n ) )
if self . copy :
    X = X . copy ( )
if self . shift_ != 0 and X is self . train_ or ( X . shape == self . train_ . shape and np . allclose ( X , self . train_ ) ) :
    X [ xrange ( n ) , xrange ( n ) ] += self . shift_
return X


signature:
def fit ( self , X , y = None ) :
    '''Picks the elements of the basis to use for the given data .
 Only depends on the dimension of X . If it ' s more convenient , you can
 pass a single integer for X , which is the dimension to use .
 Parameters
 X : an integer , a : class : ` Features ` instance , or a list of bag features
 The input data , or just its dimension , since only the dimension is
 needed here .'''

implementation:
if is_integer ( X ) :
    dim = X
else :
    X = as_features ( X )
    dim = X . dim
M = self . smoothness
# figure out the smooth - enough elements of our basis
inds = np . mgrid [ ( slice ( M + 1 ) , ) * dim ] . reshape ( dim , ( M + 1 ) ** dim ) . T
self . inds_ = inds [ ( inds ** 2 ) . sum ( axis = 1 ) <= M ** 2 ]
return self


signature:
def transform ( self , X ) :
    '''Transform a list of bag features into its projection series
 representation .
 Parameters
 X : : class : ` skl _ groups . features . Features ` or list of bag feature arrays
 New data to transform . The data should all lie in [ 0 , 1 ] ;
 use : class : ` skl _ groups . preprocessing . BagMinMaxScaler ` if not .
 Returns
 X _ new : integer array , shape ` ` [ len ( X ) , dim _ ] ` `
 X transformed into the new space .'''

implementation:
self . _check_fitted ( )
M = self . smoothness
dim = self . dim_
inds = self . inds_
do_check = self . do_bounds_check
X = as_features ( X )
if X . dim != dim :
    msg = "model fit for dimension {} but got dim {}"
    raise ValueError ( msg . format ( dim , X . dim ) )
Xt = np . empty ( ( len ( X ) , self . inds_ . shape [ 0 ] ) )
Xt . fill ( np . nan )
if self . basis == 'cosine' : # TODO : put this in a C extension ?
    coefs = ( np . pi * np . arange ( M + 1 ) ) [ ... , : ]
    for i , bag in enumerate ( X ) :
        if do_check :
            if np . min ( bag ) < 0 or np . max ( bag ) > 1 :
                raise ValueError ( "Bag {} not in [0, 1]" . format ( i ) )
        # apply each phi func to each dataset point : n x dim x M
        phi = coefs * bag [ ... , np . newaxis ]
        np . cos ( phi , out = phi )
        phi [ : , : , 1 : ] *= np . sqrt ( 2 )
        # B is the evaluation of each tensor - prodded basis func
        # at each point : n x inds . shape [ 0]
        B = reduce ( op . mul , ( phi [ : , i , inds [ : , i ] ] for i in xrange ( dim ) ) )
        Xt [ i , : ] = np . mean ( B , axis = 0 )
else :
    raise ValueError ( "unknown basis '{}'" . format ( self . basis ) )
return Xt


signature:
def get_version ( self ) :
    """Get distribution version .
 This method is enhanced compared to original distutils implementation .
 If the version string is set to a special value then instead of using
 the actual value the real version is obtained by querying versiontools .
 If versiontools package is not installed then the version is obtained
 from the standard section of the ` ` PKG - INFO ` ` file . This file is
 automatically created by any source distribution . This method is less
 useful as it cannot take advantage of version control information that
 is automatically loaded by versiontools . It has the advantage of not
 requiring versiontools installation and that it does not depend on
 ` ` setup _ requires ` ` feature of ` ` setuptools ` ` ."""

implementation:
if ( self . name is not None and self . version is not None and self . version . startswith ( ":versiontools:" ) ) :
    return ( self . __get_live_version ( ) or self . __get_frozen_version ( ) or self . __fail_to_get_any_version ( ) )
else :
    return self . __base . get_version ( self )


signature:
def __get_live_version ( self ) :
    """Get a live version string using versiontools"""

implementation:
try :
    import versiontools
except ImportError :
    return None
else :
    return str ( versiontools . Version . from_expression ( self . name ) )


signature:
def fit ( self , X , y = None , ** params ) :
    '''Fit the transformer on the stacked points .
 Parameters
 X : : class : ` Features ` or list of arrays of shape ` ` [ n _ samples [ i ] , n _ features ] ` `
 Training set . If a Features object , it will be stacked .
 any other keyword argument :
 Passed on as keyword arguments to the transformer ' s ` ` fit ( ) ` ` .'''

implementation:
X = as_features ( X , stack = True )
self . transformer . fit ( X . stacked_features , y , ** params )
return self


signature:
def transform ( self , X , ** params ) :
    '''Transform the stacked points .
 Parameters
 X : : class : ` Features ` or list of bag feature arrays
 New data to transform .
 any other keyword argument :
 Passed on as keyword arguments to the transformer ' s ` ` transform ( ) ` ` .
 Returns
 X _ new : : class : ` Features `
 Transformed features .'''

implementation:
X = as_features ( X , stack = True )
X_new = self . transformer . transform ( X . stacked_features , ** params )
return self . _gather_outputs ( X , X_new )


signature:
def fit_transform ( self , X , y = None , ** params ) :
    '''Fit and transform the stacked points .
 Parameters
 X : : class : ` Features ` or list of bag feature arrays
 Data to train on and transform .
 any other keyword argument :
 Passed on as keyword arguments to the transformer ' s ` ` transform ( ) ` ` .
 Returns
 X _ new : : class : ` Features `
 Transformed features .'''

implementation:
X = as_features ( X , stack = True )
X_new = self . transformer . fit_transform ( X . stacked_features , y , ** params )
return self . _gather_outputs ( X , X_new )


signature:
def inverse_transform ( self , X , ** params ) :
    '''Transform data back to its original space , i . e . , return an input
 X _ original whose transform would ( maybe approximately ) be X .
 Parameters
 X : : class : ` Features ` or list of bag feature arrays
 Data to train on and transform .
 any other keyword argument :
 Passed on as keyword arguments to the transformer ' s
 ` ` inverse _ transform ( ) ` ` .
 Returns
 X _ original : : class : ` Features `'''

implementation:
X = as_features ( X , stack = True )
Xo = self . transformer . inverse_transform ( X . stacked_features , ** params )
return self . _gather_outputs ( X , Xo )


signature:
def fit ( self , X , y = None ) :
    """Compute the minimum and maximum to be used for later scaling .
 Parameters
 X : array - like , shape [ n _ samples , n _ features ]
 The data used to compute the per - feature minimum and maximum
 used for later scaling along the features axis ."""

implementation:
X = check_array ( X , copy = self . copy , dtype = [ np . float64 , np . float32 , np . float16 , np . float128 ] )
feature_range = self . feature_range
if feature_range [ 0 ] >= feature_range [ 1 ] :
    raise ValueError ( "Minimum of desired feature range must be smaller" " than maximum. Got %s." % str ( feature_range ) )
if self . fit_feature_range is not None :
    fit_feature_range = self . fit_feature_range
    if fit_feature_range [ 0 ] >= fit_feature_range [ 1 ] :
        raise ValueError ( "Minimum of desired (fit) feature range must " "be smaller than maximum. Got %s." % str ( feature_range ) )
    if ( fit_feature_range [ 0 ] < feature_range [ 0 ] or fit_feature_range [ 1 ] > feature_range [ 1 ] ) :
        raise ValueError ( "fit_feature_range must be a subset of " "feature_range. Got %s, fit %s." % ( str ( feature_range ) , str ( fit_feature_range ) ) )
    feature_range = fit_feature_range
data_min = np . min ( X , axis = 0 )
data_range = np . max ( X , axis = 0 ) - data_min
# Do not scale constant features
data_range [ data_range == 0.0 ] = 1.0
self . scale_ = ( feature_range [ 1 ] - feature_range [ 0 ] ) / data_range
self . min_ = feature_range [ 0 ] - data_min * self . scale_
self . data_range = data_range
self . data_min = data_min
return self


signature:
def transform ( self , X ) :
    """Scaling features of X according to feature _ range .
 Parameters
 X : array - like with shape [ n _ samples , n _ features ]
 Input data that will be transformed ."""

implementation:
X = check_array ( X , copy = self . copy )
X *= self . scale_
X += self . min_
if self . truncate :
    np . maximum ( self . feature_range [ 0 ] , X , out = X )
    np . minimum ( self . feature_range [ 1 ] , X , out = X )
return X


signature:
def inverse_transform ( self , X ) :
    """Undo the scaling of X according to feature _ range .
 Note that if truncate is true , any truncated points will not
 be restored exactly .
 Parameters
 X : array - like with shape [ n _ samples , n _ features ]
 Input data that will be transformed ."""

implementation:
X = check_array ( X , copy = self . copy )
X -= self . min_
X /= self . scale_
return X


signature:
def fit ( self , X , y = None ) :
    '''Choose the codewords based on a training set .
 Parameters
 X : : class : ` skl _ groups . features . Features ` or list of arrays of shape ` ` [ n _ samples [ i ] , n _ features ] ` `
 Training set . If a Features object , it will be stacked .'''

implementation:
self . kmeans_fit_ = copy ( self . kmeans )
X = as_features ( X , stack = True )
self . kmeans_fit_ . fit ( X . stacked_features )
return self


signature:
def transform ( self , X ) :
    '''Transform a list of bag features into its bag - of - words representation .
 Parameters
 X : : class : ` skl _ groups . features . Features ` or list of bag feature arrays
 New data to transform .
 Returns
 X _ new : integer array , shape [ len ( X ) , kmeans . n _ clusters ]
 X transformed into the new space .'''

implementation:
self . _check_fitted ( )
X = as_features ( X , stack = True )
assignments = self . kmeans_fit_ . predict ( X . stacked_features )
return self . _group_assignments ( X , assignments )


signature:
def fit_transform ( self , X ) :
    '''Compute clustering and transform a list of bag features into its
 bag - of - words representation . Like calling fit ( X ) and then transform ( X ) ,
 but more efficient .
 Parameters
 X : : class : ` skl _ groups . features . Features ` or list of bag feature arrays
 New data to transform .
 Returns
 X _ new : integer array , shape [ len ( X ) , kmeans . n _ clusters ]
 X transformed into the new space .'''

implementation:
X = as_features ( X , stack = True )
self . kmeans_fit_ = copy ( self . kmeans )
assignments = self . kmeans_fit_ . fit_predict ( X . stacked_features )
return self . _group_assignments ( X , assignments )


signature:
def is_categorical_type ( ary ) :
    "Checks whether the array is either integral or boolean ."

implementation:
ary = np . asanyarray ( ary )
return is_integer_type ( ary ) or ary . dtype . kind == 'b'


signature:
def as_integer_type ( ary ) :
    '''Returns argument as an integer array , converting floats if convertable .
 Raises ValueError if it ' s a float array with nonintegral values .'''

implementation:
ary = np . asanyarray ( ary )
if is_integer_type ( ary ) :
    return ary
rounded = np . rint ( ary )
if np . any ( rounded != ary ) :
    raise ValueError ( "argument array must contain only integers" )
return rounded . astype ( int )


signature:
def show_progress ( name , ** kwargs ) :
    '''Sets up a : class : ` ProgressBarHandler ` to handle progess logs for
 a given module .
 Parameters
 name : string
 The module name of the progress logger to use . For example ,
 : class : ` skl _ groups . divergences . KNNDivergenceEstimator `
 uses ` ` ' skl _ groups . divergences . knn . progress ' ` ` .
 * : anything
 Other keyword arguments are passed to the : class : ` ProgressBarHandler ` .'''

implementation:
logger = logging . getLogger ( name )
logger . setLevel ( logging . INFO )
logger . addHandler ( ProgressBarHandler ( ** kwargs ) )


signature:
def start ( self , total ) :
    '''Signal the start of the process .
 Parameters
 total : int
 The total number of steps in the process , or None if unknown .'''

implementation: self . logger . info ( json . dumps ( [ 'START' , self . name , total ] ) )


signature:
def _build_indices ( X , flann_args ) :
    "Builds FLANN indices for each bag ."

implementation:
# TODO : should probably multithread this
logger . info ( "Building indices..." )
indices = [ None ] * len ( X )
for i , bag in enumerate ( plog ( X , name = "index building" ) ) :
    indices [ i ] = idx = FLANNIndex ( ** flann_args )
    idx . build_index ( bag )
return indices


signature:
def _get_rhos ( X , indices , Ks , max_K , save_all_Ks , min_dist ) :
    "Gets within - bag distances for each bag ."

implementation:
logger . info ( "Getting within-bag distances..." )
if max_K >= X . n_pts . min ( ) :
    msg = "asked for K = {}, but there's a bag with only {} points"
    raise ValueError ( msg . format ( max_K , X . n_pts . min ( ) ) )
# need to throw away the closest neighbor , which will always be self
# thus K = 1 corresponds to column 1 in the result array
which_Ks = slice ( 1 , None ) if save_all_Ks else Ks
indices = plog ( indices , name = "within-bag distances" )
rhos = [ None ] * len ( X )
for i , ( idx , bag ) in enumerate ( zip ( indices , X ) ) :
    r = np . sqrt ( idx . nn_index ( bag , max_K + 1 ) [ 1 ] [ : , which_Ks ] )
    np . maximum ( min_dist , r , out = r )
    rhos [ i ] = r
return rhos


signature:
def linear ( Ks , dim , num_q , rhos , nus ) :
    r'''Estimates the linear inner product \ int p q between two distributions ,
 based on kNN distances .'''

implementation: return _get_linear ( Ks , dim ) ( num_q , rhos , nus )


signature:
def alpha_div ( alphas , Ks , dim , num_q , rhos , nus ) :
    r'''Estimate the alpha divergence between distributions :
 \ int p ^ \ alpha q ^ ( 1 - \ alpha )
 based on kNN distances .
 Used in Renyi , Hellinger , Bhattacharyya , Tsallis divergences .
 Enforces that estimates are > = 0.
 Returns divergence estimates with shape ( num _ alphas , num _ Ks ) .'''

implementation: return _get_alpha_div ( alphas , Ks , dim ) ( num_q , rhos , nus )


signature:
def jensen_shannon_core ( Ks , dim , num_q , rhos , nus ) :
    r'''Estimates
 1/2 mean _ X ( d * log radius of largest ball in X + Y around X _ i
 with no more than M / ( n + m - 1 ) weight
 where X points have weight 1 / ( 2 n - 1)
 and Y points have weight n / ( m ( 2 n - 1 ) )
 - digamma ( # of neighbors in that ball ) )
 This is the core pairwise component of the estimator of Jensen - Shannon
 divergence based on the Hino - Murata weighted information estimator . See
 the docstring for jensen _ shannon for an explanation .'''

implementation:
ns = np . array ( [ rhos . shape [ 0 ] , num_q ] )
return _get_jensen_shannon_core ( Ks , dim , ns ) [ 0 ] ( num_q , rhos , nus )


signature:
def bhattacharyya ( Ks , dim , required , clamp = True , to_self = False ) :
    r'''Estimate the Bhattacharyya coefficient between distributions , based on kNN
 distances : \ int \ sqrt { p q }
 If clamp ( the default ) , enforces 0 < = BC < = 1.
 Returns an array of shape ( num _ Ks , ) .'''

implementation:
est = required
if clamp :
    est = np . minimum ( est , 1 )
    # BC < = 1
return est


signature:
def hellinger ( Ks , dim , required , clamp = True , to_self = False ) :
    r'''Estimate the Hellinger distance between distributions , based on kNN
 distances : \ sqrt { 1 - \ int \ sqrt { p q } }
 Always enforces 0 < = H , to be able to sqrt ; if clamp , also enforces
 H < = 1.
 Returns a vector : one element for each K .'''

implementation:
bc = required
est = 1 - bc
np . maximum ( est , 0 , out = est )
if clamp :
    np . minimum ( est , 1 , out = est )
np . sqrt ( est , out = est )
return est


signature:
def renyi ( alphas , Ks , dim , required , min_val = np . spacing ( 1 ) , clamp = True , to_self = False ) :
    r'''Estimate the Renyi - alpha divergence between distributions , based on kNN
 distances : 1 / ( \ alpha - 1 ) \ log \ int p ^ alpha q ^ ( 1 - \ alpha )
 If the inner integral is less than min _ val ( default ` ` np . spacing ( 1 ) ` ` ) ,
 uses the log of min _ val instead .
 If clamp ( the default ) , enforces that the estimates are nonnegative by
 replacing any negative estimates with 0.
 Returns an array of shape ( num _ alphas , num _ Ks ) .'''

implementation:
alphas = np . reshape ( alphas , ( - 1 , 1 ) )
est = required
est = np . maximum ( est , min_val )
# TODO : can we modify in - place ?
np . log ( est , out = est )
est /= alphas - 1
if clamp :
    np . maximum ( est , 0 , out = est )
return est


signature:
def tsallis ( alphas , Ks , dim , required , clamp = True , to_self = False ) :
    r'''Estimate the Tsallis - alpha divergence between distributions , based on kNN
 distances : ( \ int p ^ alpha q ^ ( 1 - \ alpha ) - 1 ) / ( \ alpha - 1)
 If clamp ( the default ) , enforces the estimate is nonnegative .
 Returns an array of shape ( num _ alphas , num _ Ks ) .'''

implementation:
alphas = np . reshape ( alphas , ( - 1 , 1 ) )
alpha_est = required
est = alpha_est - 1
est /= alphas - 1
if clamp :
    np . maximum ( est , 0 , out = est )
return est


signature:
def l2 ( Ks , dim , X_rhos , Y_rhos , required , clamp = True , to_self = False ) :
    r'''Estimates the L2 distance between distributions , via
 \ int ( p - q ) ^ 2 = \ int p ^ 2 - \ int p q - \ int q p + \ int q ^ 2.
 \ int pq and \ int qp are estimated with the linear function ( in both
 directions ) , while \ int p ^ 2 and \ int q ^ 2 are estimated via the quadratic
 function below .
 Always clamps negative estimates of l2 ^ 2 to 0 , because otherwise the sqrt
 would break .'''

implementation:
n_X = len ( X_rhos )
n_Y = len ( Y_rhos )
linears = required
assert linears . shape == ( 1 , Ks . size , n_X , n_Y , 2 )
X_quadratics = np . empty ( ( Ks . size , n_X ) , dtype = np . float32 )
for i , rho in enumerate ( X_rhos ) :
    X_quadratics [ : , i ] = quadratic ( Ks , dim , rho )
Y_quadratics = np . empty ( ( Ks . size , n_Y ) , dtype = np . float32 )
for j , rho in enumerate ( Y_rhos ) :
    Y_quadratics [ : , j ] = quadratic ( Ks , dim , rho )
est = - linears . sum ( axis = 4 )
est += X_quadratics [ None , : , : , None ]
est += Y_quadratics [ None , : , None , : ]
np . maximum ( est , 0 , out = est )
np . sqrt ( est , out = est )
# diagonal is of course known to be zero
if to_self :
    est [ : , : , xrange ( n_X ) , xrange ( n_Y ) ] = 0
return est [ : , : , : , : , None ]


signature:
def quadratic ( Ks , dim , rhos , required = None ) :
    r'''Estimates \ int p ^ 2 based on kNN distances .
 In here because it ' s used in the l2 distance , above .
 Returns array of shape ( num _ Ks , ) .'''

implementation:
# Estimated with alpha = 1 , beta = 0:
# B _ { k , d , 1,0 } is the same as B _ { k , d , 0,1 } in linear ( )
# and the full estimator is
# B / ( n - 1 ) * mean ( rho ^ - dim )
N = rhos . shape [ 0 ]
Ks = np . asarray ( Ks )
Bs = ( Ks - 1 ) / np . pi ** ( dim / 2 ) * gamma ( dim / 2 + 1 )
# shape ( num _ Ks , )
est = Bs / ( N - 1 ) * np . mean ( rhos ** ( - dim ) , axis = 0 )
return est


signature:
def jensen_shannon ( Ks , dim , X_rhos , Y_rhos , required , clamp = True , to_self = False ) :
    r'''Estimate the difference between the Shannon entropy of an equally - weighted
 mixture between X and Y and the mixture of the Shannon entropies :
 JS ( X , Y ) = H [ ( X + Y ) / 2 ] - ( H [ X ] + H [ Y ] ) / 2
 We use a special case of the Hino - Murata weighted information estimator with
 a fixed M = n \ alpha , about equivalent to the K - nearest - neighbor approach
 used for the other estimators :
 Hideitsu Hino and Noboru Murata ( 2013 ) .
 Information estimators for weighted observations . Neural Networks .
 http : / / linkinghub . elsevier . com / retrieve / pii / S0893608013001676
 The estimator for JS ( X , Y ) is :
 log volume of the unit ball - log M + log ( n + m - 1 ) + digamma ( M )
 + 1/2 mean _ X ( d * log radius of largest ball in X + Y around X _ i
 with no more than M / ( n + m - 1 ) weight
 where X points have weight 1 / ( 2 n - 1)
 and Y points have weight n / ( m ( 2 n - 1 ) )
 - digamma ( # of neighbors in that ball ) )
 + 1/2 mean _ Y ( d * log radius of largest ball in X + Y around Y _ i
 with no more than M / ( n + m - 1 ) weight
 where X points have weight m / ( n ( 2 m - 1 ) )
 and Y points have weight 1 / ( 2 m - 1)
 - digamma ( # of neighbors in that ball ) )
 - 1/2 ( log volume of the unit ball - log M + log ( n - 1 ) + digamma ( M ) )
 - 1/2 mean _ X ( d * log radius of the largest ball in X around X _ i
 with no more than M / ( n - 1 ) weight
 where X points have weight 1 / ( n - 1 ) )
 - digamma ( # of neighbors in that ball ) )
 - 1/2 ( log volume of the unit ball - log M + log ( m - 1 ) + digamma ( M ) )
 - 1/2 mean _ Y ( d * log radius of the largest ball in Y around Y _ i
 with no more than M / ( n - 1 ) weight
 where X points have weight 1 / ( m - 1 ) )
 - digamma ( # of neighbors in that ball ) )
 log ( n + m - 1 ) + digamma ( M )
 + 1/2 mean _ X ( d * log radius of largest ball in X + Y around X _ i
 with no more than M / ( n + m - 1 ) weight
 where X points have weight 1 / ( 2 n - 1)
 and Y points have weight n / ( m ( 2 n - 1 ) )
 - digamma ( # of neighbors in that ball ) )
 + 1/2 mean _ Y ( d * log radius of largest ball in X + Y around Y _ i
 with no more than M / ( n + m - 1 ) weight
 where X points have weight m / ( n ( 2 m - 1 ) )
 and Y points have weight 1 / ( 2 m - 1)
 - digamma ( # of neighbors in that ball ) )
 - 1/2 [ log ( n - 1 ) + mean _ X ( d * log rho _ M ( X _ i ) ) ]
 - 1/2 [ log ( m - 1 ) + mean _ Y ( d * log rho _ M ( Y _ i ) ) ]'''

implementation:
X_ns = np . array ( [ rho . shape [ 0 ] for rho in X_rhos ] )
Y_ns = np . array ( [ rho . shape [ 0 ] for rho in Y_rhos ] )
n_X = X_ns . size
n_Y = Y_ns . size
# cores [ 0 , k , i , j , 0 ] is mean _ X ( d * . . . - psi ( . . . ) ) for X [ i ] , Y [ j ] , M = Ks [ k ]
# cores [ 0 , k , i , j , 1 ] is mean _ Y ( d * . . . - psi ( . . . ) ) for X [ i ] , Y [ j ] , M = Ks [ k ]
cores = required
assert cores . shape == ( 1 , Ks . size , n_X , n_Y , 2 )
# X _ bits [ k , i ] is log ( n - 1 ) + mean _ X ( d * log rho _ M ( X _ i ) ) for X [ i ] , M = Ks [ k ]
X_bits = np . empty ( ( Ks . size , n_X ) , dtype = np . float32 )
for i , rho in enumerate ( X_rhos ) :
    X_bits [ : , i ] = dim * np . mean ( np . log ( rho ) , axis = 0 )
X_bits += np . log ( X_ns - 1 ) [ np . newaxis , : ]
# Y _ bits [ k , j ] is log ( n - 1 ) + mean _ Y ( d * log rho _ M ( Y _ i ) ) for Y [ j ] , M = Ks [ k ]
Y_bits = np . empty ( ( Ks . size , n_Y ) , dtype = np . float32 )
for j , rho in enumerate ( Y_rhos ) :
    Y_bits [ : , j ] = dim * np . mean ( np . log ( rho ) , axis = 0 )
Y_bits += np . log ( Y_ns - 1 ) [ np . newaxis , : ]
est = cores . sum ( axis = 4 )
est -= X_bits . reshape ( 1 , Ks . size , n_X , 1 )
est -= Y_bits . reshape ( 1 , Ks . size , 1 , n_Y )
est /= 2
est += np . log ( - 1 + X_ns [ None , None , : , None ] + Y_ns [ None , None , None , : ] )
est += psi ( Ks ) [ None , : , None , None ]
# diagonal is zero
if to_self :
    est [ : , : , xrange ( n_X ) , xrange ( n_Y ) ] = 0
if clamp : # know that 0 < = JS < = ln ( 2)
    np . maximum ( 0 , est , out = est )
    np . minimum ( np . log ( 2 ) , est , out = est )
return est [ : , : , : , : , None ]


signature:
def topological_sort ( deps ) :
    '''Topologically sort a DAG , represented by a dict of child = > set of parents .
 The dependency dict is destroyed during operation .
 Uses the Kahn algorithm : http : / / en . wikipedia . org / wiki / Topological _ sorting
 Not a particularly good implementation , but we ' re just running it on tiny
 graphs .'''

implementation:
order = [ ]
available = set ( )
def _move_available ( ) :
    to_delete = [ ]
    for n , parents in iteritems ( deps ) :
        if not parents :
            available . add ( n )
            to_delete . append ( n )
    for n in to_delete :
        del deps [ n ]
_move_available ( )
while available :
    n = available . pop ( )
    order . append ( n )
    for parents in itervalues ( deps ) :
        parents . discard ( n )
    _move_available ( )
if available :
    raise ValueError ( "dependency cycle found" )
return order


signature:
def _parse_specs ( specs , Ks ) :
    '''Set up the different functions we need to call .
 Returns :
 - a dict mapping base estimator functions to _ FuncInfo objects .
 If the function needs _ alpha , then the alphas attribute is an array
 of alpha values and pos is a corresponding array of indices .
 Otherwise , alphas is None and pos is a list containing a single index .
 Indices are > = 0 if they correspond to something in a spec ,
 and negative if they ' re just used for a meta estimator but not
 directly requested .
 - an OrderedDict mapping functions to _ MetaFuncInfo objects .
 alphas and pos are like for _ FuncInfo ; deps is a list of indices
 which should be passed to the estimator . Note that these might be
 other meta functions ; this list is guaranteed to be in an order
 such that all dependencies are resolved before calling that function .
 If no such order is possible , raise ValueError .
 - the number of meta - only results
 # TODO : update doctests for _ parse _ specs
 > > > _ parse _ specs ( [ ' renyi : . 8 ' , ' hellinger ' , ' renyi : . 9 ' ] )
 ( { < function alpha _ div at 0x10954f848 > :
 _ FuncInfo ( alphas = [ 0.8 , 0.5 , 0.9 ] , pos = [ - 1 , - 2 , - 3 ] ) } ,
 OrderedDict ( [
 ( < function hellinger at 0x10954fc80 > ,
 _ MetaFuncInfo ( alphas = None , pos = [ 1 ] , deps = [ array ( - 2 ) ] ) ) ,
 ( < function renyi at 0x10954fcf8 > ,
 _ MetaFuncInfo ( alphas = [ 0.8 , 0.9 ] , pos = [ 0 , 2 ] , deps = [ - 1 , - 3 ] ) )
 > > > _ parse _ specs ( [ ' renyi : . 8 ' , ' hellinger ' , ' renyi : . 9 ' , ' l2 ' ] )
 ( { < function alpha _ div at 0x10954f848 > :
 _ FuncInfo ( alphas = [ 0.8 , 0.5 , 0.9 ] , pos = [ - 1 , - 2 , - 3 ] ) ,
 < function linear at 0x10954f758 > : _ FuncInfo ( alphas = None , pos = [ - 4 ] )
 } , OrderedDict ( [
 ( < function hellinger at 0x10954fc80 > ,
 _ MetaFuncInfo ( alphas = None , pos = [ 1 ] , deps = [ array ( - 2 ) ] ) ) ,
 ( < function l2 at 0x10954fde8 > ,
 _ MetaFuncInfo ( alphas = None , pos = [ 3 ] , deps = [ - 4 ] ) ) ,
 ( < function renyi at 0x10954fcf8 > ,
 _ MetaFuncInfo ( alphas = [ 0.8 , 0.9 ] , pos = [ 0 , 2 ] , deps = [ - 1 , - 3 ] ) )
 > > > _ parse _ specs ( [ ' renyi : . 8 ' , ' hellinger ' , ' renyi : . 9 ' , ' l2 ' , ' linear ' ] )
 ( { < function alpha _ div at 0x10954f848 > :
 _ FuncInfo ( alphas = [ 0.8 , 0.5 , 0.9 ] , pos = [ - 1 , - 2 , - 3 ] ) ,
 < function linear at 0x10954f758 > : _ FuncInfo ( alphas = None , pos = [ 4 ] )
 } , OrderedDict ( [
 ( < function hellinger at 0x10954fc80 > ,
 _ MetaFuncInfo ( alphas = None , pos = [ 1 ] , deps = [ array ( - 2 ) ] ) ) ,
 ( < function l2 at 0x10954fde8 > ,
 _ MetaFuncInfo ( alphas = None , pos = [ 3 ] , deps = [ 4 ] ) ) ,
 ( < function renyi at 0x10954fcf8 > ,
 _ MetaFuncInfo ( alphas = [ 0.8 , 0.9 ] , pos = [ 0 , 2 ] , deps = [ - 1 , - 3 ] ) )'''

implementation:
funcs = { }
metas = { }
meta_deps = defaultdict ( set )
def add_func ( func , alpha = None , pos = None ) :
    needs_alpha = getattr ( func , 'needs_alpha' , False )
    is_meta = hasattr ( func , 'needs_results' )
    d = metas if is_meta else funcs
    if func not in d :
        if needs_alpha :
            args = { 'alphas' : [ alpha ] , 'pos' : [ pos ] }
        else :
            args = { 'alphas' : None , 'pos' : [ pos ] }
        if not is_meta :
            d [ func ] = _FuncInfo ( ** args )
        else :
            d [ func ] = _MetaFuncInfo ( deps = [ ] , ** args )
            for req in func . needs_results :
                if callable ( req . alpha ) :
                    req_alpha = req . alpha ( alpha )
                else :
                    req_alpha = req . alpha
                add_func ( req . func , alpha = req_alpha )
                meta_deps [ func ] . add ( req . func )
                meta_deps [ req . func ]
    # make sure required func is in there
    else : # already have an entry for the func
    # need to give it this pos , if it ' s not None
    # and also make sure that the alpha is present
        info = d [ func ]
        if not needs_alpha :
            if pos is not None :
                if info . pos != [ None ] :
                    msg = "{} passed more than once"
                    raise ValueError ( msg . format ( func_name ) )
                info . pos [ 0 ] = pos
        else : # needs alpha
            try :
                idx = info . alphas . index ( alpha )
            except ValueError : # this is a new alpha value we haven ' t seen yet
                info . alphas . append ( alpha )
                info . pos . append ( pos )
                if is_meta :
                    for req in func . needs_results :
                        if callable ( req . alpha ) :
                            req_alpha = req . alpha ( alpha )
                        else :
                            req_alpha = req . alpha
                        add_func ( req . func , alpha = req_alpha )
            else : # repeated alpha value
                if pos is not None :
                    if info . pos [ idx ] is not None :
                        msg = "{} with alpha {} passed more than once"
                        raise ValueError ( msg . format ( func_name , alpha ) )
                    info . pos [ idx ] = pos
# add functions for each spec
for i , spec in enumerate ( specs ) :
    func_name , alpha = ( spec . split ( ':' , 1 ) + [ None ] ) [ : 2 ]
    if alpha is not None :
        alpha = float ( alpha )
    try :
        func = func_mapping [ func_name ]
    except KeyError :
        msg = "'{}' is not a known function type"
        raise ValueError ( msg . format ( func_name ) )
    needs_alpha = getattr ( func , 'needs_alpha' , False )
    if needs_alpha and alpha is None :
        msg = "{} needs alpha but not passed in spec '{}'"
        raise ValueError ( msg . format ( func_name , spec ) )
    elif not needs_alpha and alpha is not None :
        msg = "{} doesn't need alpha but is passed in spec '{}'"
        raise ValueError ( msg . format ( func_name , spec ) )
    add_func ( func , alpha , i )
# number things that are dependencies only
meta_counter = itertools . count ( - 1 , step = - 1 )
for info in itertools . chain ( itervalues ( funcs ) , itervalues ( metas ) ) :
    for i , pos in enumerate ( info . pos ) :
        if pos is None :
            info . pos [ i ] = next ( meta_counter )
# fill in the dependencies for metas
for func , info in iteritems ( metas ) :
    deps = info . deps
    assert deps == [ ]
    for req in func . needs_results :
        f = req . func
        req_info = ( metas if hasattr ( f , 'needs_results' ) else funcs ) [ f ]
        if req . alpha is not None :
            if callable ( req . alpha ) :
                req_alpha = req . alpha ( info . alphas )
            else :
                req_alpha = req . alpha
            find_alpha = np . vectorize ( req_info . alphas . index , otypes = [ int ] )
            pos = np . asarray ( req_info . pos ) [ find_alpha ( req_alpha ) ]
            if np . isscalar ( pos ) :
                deps . append ( pos [ ( ) ] )
            else :
                deps . extend ( pos )
        else :
            pos , = req_info . pos
            deps . append ( pos )
# topological sort of metas
meta_order = topological_sort ( meta_deps )
metas_ordered = OrderedDict ( ( f , metas [ f ] ) for f in meta_order if hasattr ( f , 'needs_results' ) )
return funcs , metas_ordered , - next ( meta_counter ) - 1


signature:
def _get_Ks ( self ) :
    "Ks as an array and type - checked ."

implementation:
Ks = as_integer_type ( self . Ks )
if Ks . ndim != 1 :
    raise TypeError ( "Ks should be 1-dim, got shape {}" . format ( Ks . shape ) )
if Ks . min ( ) < 1 :
    raise ValueError ( "Ks should be positive; got {}" . format ( Ks . min ( ) ) )
return Ks


signature:
def _flann_args ( self , X = None ) :
    "The dictionary of arguments to give to FLANN ."

implementation:
args = { 'cores' : self . _n_jobs }
if self . flann_algorithm == 'auto' :
    if X is None or X . dim > 5 :
        args [ 'algorithm' ] = 'linear'
    else :
        args [ 'algorithm' ] = 'kdtree_single'
else :
    args [ 'algorithm' ] = self . flann_algorithm
if self . flann_args :
    args . update ( self . flann_args )
# check that arguments are correct
try :
    FLANNParameters ( ) . update ( args )
except AttributeError as e :
    msg = "flann_args contains an invalid argument:\n  {}"
    raise TypeError ( msg . format ( e ) )
return args


signature:
def fit ( self , X , y = None , get_rhos = False ) :
    '''Sets up for divergence estimation " from " new data " to " X .
 Builds FLANN indices for each bag , and maybe gets within - bag distances .
 Parameters
 X : list of arrays or : class : ` skl _ groups . features . Features `
 The bags to search " to " .
 get _ rhos : boolean , optional , default False
 Compute within - bag distances : attr : ` rhos _ ` . These are only needed
 for some divergence functions or if do _ sym is passed , and they ' ll
 be computed ( and saved ) during : meth : ` transform ` if they ' re not
 computed here .
 If you ' re using Jensen - Shannon divergence , a higher max _ K may
 be needed once it sees the number of points in the transformed bags ,
 so the computation here might be wasted .'''

implementation:
self . features_ = X = as_features ( X , stack = True , bare = True )
# if we ' re using a function that needs to pick its K vals itself ,
# then we need to set max _ K here . when we transform ( ) , might have to
# re - do this : |
Ks = self . _get_Ks ( )
_ , _ , _ , max_K , save_all_Ks , _ = _choose_funcs ( self . div_funcs , Ks , X . dim , X . n_pts , None , self . version )
if max_K >= X . n_pts . min ( ) :
    msg = "asked for K = {}, but there's a bag with only {} points"
    raise ValueError ( msg . format ( max_K , X . n_pts . min ( ) ) )
memory = self . memory
if isinstance ( memory , string_types ) :
    memory = Memory ( cachedir = memory , verbose = 0 )
self . indices_ = id = memory . cache ( _build_indices ) ( X , self . _flann_args ( ) )
if get_rhos :
    self . rhos_ = _get_rhos ( X , id , Ks , max_K , save_all_Ks , self . min_dist )
elif hasattr ( self , 'rhos_' ) :
    del self . rhos_
return self


signature:
def transform ( self , X ) :
    r'''Computes the divergences from X to : attr : ` features _ ` .
 Parameters
 X : list of bag feature arrays or : class : ` skl _ groups . features . Features `
 The bags to search " from " .
 Returns
 divs : array of shape ` ` [ len ( div _ funcs ) , len ( Ks ) , len ( X ) , len ( features _ ) ] + ( [ 2 ] if do _ sym else [ ] ) ` `
 The divergences from X to : attr : ` features _ ` .
 ` ` divs [ d , k , i , j ] ` ` is the ` ` div _ funcs [ d ] ` ` divergence
 from ` ` X [ i ] ` ` to ` ` fetaures _ [ j ] ` ` using a K of ` ` Ks [ k ] ` ` .
 If ` ` do _ sym ` ` , ` ` divs [ d , k , i , j , 0 ] ` ` is
 : math : ` D _ { d , k } ( X _ i \ | \ texttt { features _ } _ j ) ` and
 ` ` divs [ d , k , i , j , 1 ] ` ` is : math : ` D _ { d , k } ( \ texttt { features _ } _ j \ | X _ i ) ` .'''

implementation:
X = as_features ( X , stack = True , bare = True )
Y = self . features_
Ks = np . asarray ( self . Ks )
if X . dim != Y . dim :
    msg = "incompatible dimensions: fit with {}, transform with {}"
    raise ValueError ( msg . format ( Y . dim , X . dim ) )
memory = self . memory
if isinstance ( memory , string_types ) :
    memory = Memory ( cachedir = memory , verbose = 0 )
# ignore Y _ indices to avoid slow pickling of them
# NOTE : if the indices are approximate , then might not get the same
# results !
est = memory . cache ( _est_divs , ignore = [ 'n_jobs' , 'Y_indices' , 'Y_rhos' ] )
output , self . rhos_ = est ( X , Y , self . indices_ , getattr ( self , 'rhos_' , None ) , self . div_funcs , Ks , self . do_sym , self . clamp , self . version , self . min_dist , self . _flann_args ( ) , self . _n_jobs )
return output


signature:
def as_features ( X , stack = False , bare = False ) :
    '''Returns a version of X as a : class : ` Features ` object .
 Parameters
 stack : boolean , default False
 Make a stacked version of X . Note that if X is a features object ,
 this will stack it in - place , since that ' s usually what you want .
 ( If not , just use the : class : ` Features ` constructor instead . )
 bare : boolean , default False
 Return a bare version of X ( no metadata ) .
 Returns
 feats : : class : ` Features `
 A version of X . If X is already a : class : ` Features ` object , the original
 X may be returned , depending on the arguments .'''

implementation:
if isinstance ( X , Features ) :
    if stack :
        X . make_stacked ( )
    return X . bare ( ) if bare else X
return Features ( X , stack = stack , bare = bare )


signature:
def make_stacked ( self ) :
    "If unstacked , convert to stacked . If stacked , do nothing ."

implementation:
if self . stacked :
    return
self . _boundaries = bounds = np . r_ [ 0 , np . cumsum ( self . n_pts ) ]
self . stacked_features = stacked = np . vstack ( self . features )
self . features = np . array ( [ stacked [ bounds [ i - 1 ] : bounds [ i ] ] for i in xrange ( 1 , len ( bounds ) ) ] , dtype = object )
self . stacked = True


signature:
def copy ( self , stack = False , copy_meta = False , memo = None ) :
    '''Copies the Feature object . Makes a copy of the features array .
 Parameters
 stack : boolean , optional , default False
 Whether to stack the copy if this one is unstacked .
 copy _ meta : boolean , optional , default False
 Also copy the metadata . If False , metadata in both points to the
 same object .'''

implementation:
if self . stacked :
    fs = deepcopy ( self . stacked_features , memo )
    n_pts = self . n_pts . copy ( )
elif stack :
    fs = np . vstack ( self . features )
    n_pts = self . n_pts . copy ( )
else :
    fs = deepcopy ( self . features , memo )
    n_pts = None
meta = deepcopy ( self . meta , memo ) if copy_meta else self . meta
return Features ( fs , n_pts , copy = False , ** meta )


signature:
def bare ( self ) :
    "Make a Features object with no metadata ; points to the same features ."

implementation:
if not self . meta :
    return self
elif self . stacked :
    return Features ( self . stacked_features , self . n_pts , copy = False )
else :
    return Features ( self . features , copy = False )


signature:
def kl ( Ks , dim , num_q , rhos , nus , clamp = True ) :
    r'''Estimate the KL divergence between distributions :
 \ int p ( x ) \ log ( p ( x ) / q ( x ) )
 using the kNN - based estimator ( 5 ) of
 Qing Wang , Sanjeev R Kulkarni , and Sergio Verdu ( 2009 ) .
 Divergence Estimation for Multidimensional Densities Via
 k - Nearest - Neighbor Distances .
 IEEE Transactions on Information Theory .
 http : / / www . ee . princeton . edu / ~ verdu / reprints / WanKulVer . May2009 . pdf
 which is :
 d * 1 / n \ sum \ log ( nu _ k ( i ) / rho _ k ( i ) ) + log ( m / ( n - 1 ) )
 If clamp , enforces KL > = 0.
 Returns an array of shape ( num _ Ks , ) .'''

implementation:
est = dim * np . mean ( np . log ( nus ) - np . log ( rhos ) , axis = 0 )
est += np . log ( num_q / ( rhos . shape [ 0 ] - 1 ) )
if clamp :
    np . maximum ( est , 0 , out = est )
return est


signature:
def fit ( self , X , y = None ) :
    '''Specify the data to which kernel values should be computed .
 Parameters
 X : list of arrays or : class : ` skl _ groups . features . Features `
 The bags to compute " to " .'''

implementation:
self . features_ = as_features ( X , stack = True , bare = True )
# TODO : could precompute things like squared norms if kernel = = " rbf " .
# Probably should add support to sklearn instead of hacking it here .
return self


signature:
def transform ( self , X ) :
    '''Compute kernels from X to : attr : ` features _ ` .
 Parameters
 X : list of arrays or : class : ` skl _ groups . features . Features `
 The bags to compute " from " . Must have same dimension as
 : attr : ` features _ ` .
 Returns
 K : array of shape ` ` [ len ( X ) , len ( features _ ) ] ` `
 The kernel evaluations from X to : attr : ` features _ ` .'''

implementation:
X = as_features ( X , stack = True , bare = True )
Y = self . features_
if X . dim != Y . dim :
    raise ValueError ( "MMK transform got dimension {} but had {} at fit" . format ( X . dim , Y . dim ) )
pointwise = pairwise_kernels ( X . stacked_features , Y . stacked_features , metric = self . kernel , filter_params = True , ** self . _get_kernel_params ( ) )
# TODO : is there a way to do this without a Python loop ?
K = np . empty ( ( len ( X ) , len ( Y ) ) )
for i in range ( len ( X ) ) :
    for j in range ( len ( Y ) ) :
        K [ i , j ] = pointwise [ X . _boundaries [ i ] : X . _boundaries [ i + 1 ] , Y . _boundaries [ j ] : Y . _boundaries [ j + 1 ] ] . mean ( )
return K


signature:
def transform ( self , X ) :
    '''Transform a list of bag features into a matrix of its mean features .
 Parameters
 X : : class : ` skl _ groups . features . Features ` or list of bag feature arrays
 Data to transform .
 Returns
 X _ new : array , shape ` ` [ len ( X ) , X . dim ] ` `
 X transformed into its means .'''

implementation:
X = as_features ( X )
return np . vstack ( [ np . mean ( bag , axis = 0 ) for bag in X ] )


signature:
def from_shypo ( cls , xml , encoding = 'utf-8' ) :
    """Constructor from xml element * SHYPO *
 : param xml . etree . ElementTree xml : the xml * SHYPO * element
 : param string encoding : encoding of the xml"""

implementation:
score = float ( xml . get ( 'SCORE' ) )
words = [ Word . from_whypo ( w_xml , encoding ) for w_xml in xml . findall ( 'WHYPO' ) if w_xml . get ( 'WORD' ) not in [ '<s>' , '</s>' ] ]
return cls ( words , score )


signature:
def from_whypo ( cls , xml , encoding = 'utf-8' ) :
    """Constructor from xml element * WHYPO *
 : param xml . etree . ElementTree xml : the xml * WHYPO * element
 : param string encoding : encoding of the xml"""

implementation:
word = unicode ( xml . get ( 'WORD' ) , encoding )
confidence = float ( xml . get ( 'CM' ) )
return cls ( word , confidence )


signature:
def run ( self ) :
    """Start listening to the server"""

implementation:
logger . info ( u'Started listening' )
while not self . _stop :
    xml = self . _readxml ( )
    # Exit on invalid XML
    if xml is None :
        break
    # Raw xml only
    if not self . modelize :
        logger . info ( u'Raw xml: %s' % xml )
        self . results . put ( xml )
        continue
    # Model objects + raw xml as fallback
    if xml . tag == 'RECOGOUT' :
        sentence = Sentence . from_shypo ( xml . find ( 'SHYPO' ) , self . encoding )
        logger . info ( u'Modelized recognition: %r' % sentence )
        self . results . put ( sentence )
    else :
        logger . info ( u'Unmodelized xml: %s' % xml )
        self . results . put ( xml )
logger . info ( u'Stopped listening' )


signature:
def connect ( self ) :
    """Connect to the server
 : raise ConnectionError : If socket cannot establish a connection"""

implementation:
try :
    logger . info ( u'Connecting %s:%d' % ( self . host , self . port ) )
    self . sock . connect ( ( self . host , self . port ) )
except socket . error :
    raise ConnectionError ( )
self . state = CONNECTED


signature:
def disconnect ( self ) :
    """Disconnect from the server"""

implementation:
logger . info ( u'Disconnecting' )
self . sock . shutdown ( socket . SHUT_RDWR )
self . sock . close ( )
self . state = DISCONNECTED


signature:
def send ( self , command , timeout = 5 ) :
    """Send a command to the server
 : param string command : command to send"""

implementation:
logger . info ( u'Sending %s' % command )
_ , writable , __ = select . select ( [ ] , [ self . sock ] , [ ] , timeout )
if not writable :
    raise SendTimeoutError ( )
writable [ 0 ] . sendall ( command + '\n' )


signature:
def _readline ( self ) :
    """Read a line from the server . Data is read from the socket until a character ` ` \n ` ` is found
 : return : the read line
 : rtype : string"""

implementation:
line = ''
while 1 :
    readable , _ , __ = select . select ( [ self . sock ] , [ ] , [ ] , 0.5 )
    if self . _stop :
        break
    if not readable :
        continue
    data = readable [ 0 ] . recv ( 1 )
    if data == '\n' :
        break
    line += unicode ( data , self . encoding )
return line


signature:
def _readblock ( self ) :
    """Read a block from the server . Lines are read until a character ` ` . ` ` is found
 : return : the read block
 : rtype : string"""

implementation:
block = ''
while not self . _stop :
    line = self . _readline ( )
    if line == '.' :
        break
    block += line
return block


signature:
def _readxml ( self ) :
    """Read a block and return the result as XML
 : return : block as xml
 : rtype : xml . etree . ElementTree"""

implementation:
block = re . sub ( r'<(/?)s>' , r'<\1s>' , self . _readblock ( ) )
try :
    xml = XML ( block )
except ParseError :
    xml = None
return xml


signature:
def cli ( id ) :
    """Analyse an OpenStreetMap changeset ."""

implementation:
ch = Analyse ( id )
ch . full_analysis ( )
click . echo ( 'Created: %s. Modified: %s. Deleted: %s' % ( ch . create , ch . modify , ch . delete ) )
if ch . is_suspect :
    click . echo ( 'The changeset {} is suspect! Reasons: {}' . format ( id , ', ' . join ( ch . suspicion_reasons ) ) )
else :
    click . echo ( 'The changeset %s is not suspect!' % id )


signature:
def get_user_details ( user_id ) :
    """Get information about number of changesets , blocks and mapping days of a
 user , using both the OSM API and the Mapbox comments APIself ."""

implementation:
reasons = [ ]
try :
    url = OSM_USERS_API . format ( user_id = requests . compat . quote ( user_id ) )
    user_request = requests . get ( url )
    if user_request . status_code == 200 :
        user_data = user_request . content
        xml_data = ET . fromstring ( user_data ) . getchildren ( ) [ 0 ] . getchildren ( )
        changesets = [ i for i in xml_data if i . tag == 'changesets' ] [ 0 ]
        blocks = [ i for i in xml_data if i . tag == 'blocks' ] [ 0 ]
        if int ( changesets . get ( 'count' ) ) <= 5 :
            reasons . append ( 'New mapper' )
        elif int ( changesets . get ( 'count' ) ) <= 30 :
            url = MAPBOX_USERS_API . format ( user_id = requests . compat . quote ( user_id ) )
            user_request = requests . get ( url )
            if user_request . status_code == 200 :
                mapping_days = int ( user_request . json ( ) . get ( 'extra' ) . get ( 'mapping_days' ) )
                if mapping_days <= 5 :
                    reasons . append ( 'New mapper' )
        if int ( blocks . getchildren ( ) [ 0 ] . get ( 'count' ) ) > 1 :
            reasons . append ( 'User has multiple blocks' )
except Exception as e :
    message = 'Could not verify user of the changeset: {}, {}'
    print ( message . format ( user_id , str ( e ) ) )
return reasons


signature:
def changeset_info ( changeset ) :
    """Return a dictionary with id , user , user _ id , bounds , date of creation
 and all the tags of the changeset .
 Args :
 changeset : the XML string of the changeset ."""

implementation:
keys = [ tag . attrib . get ( 'k' ) for tag in changeset . getchildren ( ) ]
keys += [ 'id' , 'user' , 'uid' , 'bbox' , 'created_at' ]
values = [ tag . attrib . get ( 'v' ) for tag in changeset . getchildren ( ) ]
values += [ changeset . get ( 'id' ) , changeset . get ( 'user' ) , changeset . get ( 'uid' ) , get_bounds ( changeset ) , changeset . get ( 'created_at' ) ]
return dict ( zip ( keys , values ) )


signature:
def get_changeset ( changeset ) :
    """Get the changeset using the OSM API and return the content as a XML
 ElementTree .
 Args :
 changeset : the id of the changeset ."""

implementation:
url = 'https://www.openstreetmap.org/api/0.6/changeset/{}/download' . format ( changeset )
return ET . fromstring ( requests . get ( url ) . content )


signature:
def get_metadata ( changeset ) :
    """Get the metadata of a changeset using the OSM API and return it as a XML
 ElementTree .
 Args :
 changeset : the id of the changeset ."""

implementation:
url = 'https://www.openstreetmap.org/api/0.6/changeset/{}' . format ( changeset )
return ET . fromstring ( requests . get ( url ) . content ) . getchildren ( ) [ 0 ]


signature:
def get_bounds ( changeset ) :
    """Get the bounds of the changeset and return it as a Polygon object . If
 the changeset has not coordinates ( case of the changesets that deal only
 with relations ) , it returns an empty Polygon .
 Args :
 changeset : the XML string of the changeset ."""

implementation:
try :
    return Polygon ( [ ( float ( changeset . get ( 'min_lon' ) ) , float ( changeset . get ( 'min_lat' ) ) ) , ( float ( changeset . get ( 'max_lon' ) ) , float ( changeset . get ( 'min_lat' ) ) ) , ( float ( changeset . get ( 'max_lon' ) ) , float ( changeset . get ( 'max_lat' ) ) ) , ( float ( changeset . get ( 'min_lon' ) ) , float ( changeset . get ( 'max_lat' ) ) ) , ( float ( changeset . get ( 'min_lon' ) ) , float ( changeset . get ( 'min_lat' ) ) ) , ] )
except TypeError :
    return Polygon ( )


signature:
def find_words ( text , suspect_words , excluded_words = [ ] ) :
    """Check if a text has some of the suspect words ( or words that starts with
 one of the suspect words ) . You can set some words to be excluded of the
 search , so you can remove false positives like ' important ' be detected when
 you search by ' import ' . It will return True if the number of suspect words
 found is greater than the number of excluded words . Otherwise , it will
 return False .
 Args :
 text ( str ) : a string with the text to be analysed . It will be converted
 to lowercase .
 suspect _ words : a list of strings that you want to check the presence in
 the text .
 excluded _ words : a list of strings to be whitelisted ."""

implementation:
text = text . lower ( )
suspect_found = [ i for i in re . finditer ( make_regex ( suspect_words ) , text ) ]
if len ( excluded_words ) > 0 :
    excluded_found = [ i for i in re . finditer ( make_regex ( excluded_words ) , text ) ]
    if len ( suspect_found ) > len ( excluded_found ) :
        return True
    else :
        return False
else :
    if len ( suspect_found ) > 0 :
        return True
    else :
        return False


signature:
def read_file ( self , changeset_file ) :
    """Download the replication changeset file or read it directly from the
 filesystem ( to test purposes ) ."""

implementation:
if isfile ( changeset_file ) :
    self . filename = changeset_file
else :
    self . path = mkdtemp ( )
    self . filename = join ( self . path , basename ( changeset_file ) )
    download ( changeset_file , self . path )
self . xml = ET . fromstring ( gzip . open ( self . filename ) . read ( ) )
# delete folder created to download the file
if not isfile ( changeset_file ) :
    rmtree ( self . path )


signature:
def get_area ( self , geojson ) :
    """Read the first feature from the geojson and return it as a Polygon
 object ."""

implementation:
geojson = json . load ( open ( geojson , 'r' ) )
self . area = Polygon ( geojson [ 'features' ] [ 0 ] [ 'geometry' ] [ 'coordinates' ] [ 0 ] )


signature:
def filter ( self ) :
    """Filter the changesets that intersects with the geojson geometry ."""

implementation: self . content = [ ch for ch in self . xml . getchildren ( ) if get_bounds ( ch ) . intersects ( self . area ) ]


signature:
def set_fields ( self , changeset ) :
    """Set the fields of this class with the metadata of the analysed
 changeset ."""

implementation:
self . id = int ( changeset . get ( 'id' ) )
self . user = changeset . get ( 'user' )
self . uid = changeset . get ( 'uid' )
self . editor = changeset . get ( 'created_by' , None )
self . review_requested = changeset . get ( 'review_requested' , False )
self . host = changeset . get ( 'host' , 'Not reported' )
self . bbox = changeset . get ( 'bbox' ) . wkt
self . comment = changeset . get ( 'comment' , 'Not reported' )
self . source = changeset . get ( 'source' , 'Not reported' )
self . imagery_used = changeset . get ( 'imagery_used' , 'Not reported' )
self . date = datetime . strptime ( changeset . get ( 'created_at' ) , '%Y-%m-%dT%H:%M:%SZ' )
self . suspicion_reasons = [ ]
self . is_suspect = False
self . powerfull_editor = False


signature:
def label_suspicious ( self , reason ) :
    """Add suspicion reason and set the suspicious flag ."""

implementation:
self . suspicion_reasons . append ( reason )
self . is_suspect = True


signature:
def full_analysis ( self ) :
    """Execute the count and verify _ words methods ."""

implementation:
self . count ( )
self . verify_words ( )
self . verify_user ( )
if self . review_requested == 'yes' :
    self . label_suspicious ( 'Review requested' )


signature:
def verify_user ( self ) :
    """Verify if the changeset was made by a inexperienced mapper ( anyone
 with less than 5 edits ) or by a user that was blocked more than once ."""

implementation:
user_reasons = get_user_details ( self . uid )
[ self . label_suspicious ( reason ) for reason in user_reasons ]


signature:
def verify_words ( self ) :
    """Verify the fields source , imagery _ used and comment of the changeset
 for some suspect words ."""

implementation:
if self . comment :
    if find_words ( self . comment , self . suspect_words , self . excluded_words ) :
        self . label_suspicious ( 'suspect_word' )
if self . source :
    for word in self . illegal_sources :
        if word in self . source . lower ( ) :
            self . label_suspicious ( 'suspect_word' )
            break
if self . imagery_used :
    for word in self . illegal_sources :
        if word in self . imagery_used . lower ( ) :
            self . label_suspicious ( 'suspect_word' )
            break
self . suspicion_reasons = list ( set ( self . suspicion_reasons ) )


signature:
def verify_editor ( self ) :
    """Verify if the software used in the changeset is a powerfull _ editor ."""

implementation:
powerful_editors = [ 'josm' , 'level0' , 'merkaartor' , 'qgis' , 'arcgis' , 'upload.py' , 'osmapi' , 'Services_OpenStreetMap' ]
if self . editor is not None :
    for editor in powerful_editors :
        if editor in self . editor . lower ( ) :
            self . powerfull_editor = True
            break
    if 'iD' in self . editor :
        trusted_hosts = [ 'www.openstreetmap.org/id' , 'www.openstreetmap.org/edit' , 'improveosm.org' , 'strava.github.io/iD' , 'preview.ideditor.com/release' , 'preview.ideditor.com/master' , 'hey.mapbox.com/iD-internal' , 'projets.pavie.info/id-indoor' , 'maps.mapcat.com/edit' , 'id.softek.ir' ]
        if self . host . split ( '://' ) [ - 1 ] . strip ( '/' ) not in trusted_hosts :
            self . label_suspicious ( 'Unknown iD instance' )
else :
    self . powerfull_editor = True
    self . label_suspicious ( 'Software editor was not declared' )


signature:
def count ( self ) :
    """Count the number of elements created , modified and deleted by the
 changeset and analyses if it is a possible import , mass modification or
 a mass deletion ."""

implementation:
xml = get_changeset ( self . id )
actions = [ action . tag for action in xml . getchildren ( ) ]
self . create = actions . count ( 'create' )
self . modify = actions . count ( 'modify' )
self . delete = actions . count ( 'delete' )
self . verify_editor ( )
try :
    if ( self . create / len ( actions ) > self . percentage and self . create > self . create_threshold and ( self . powerfull_editor or self . create > self . top_threshold ) ) :
        self . label_suspicious ( 'possible import' )
    elif ( self . modify / len ( actions ) > self . percentage and self . modify > self . modify_threshold ) :
        self . label_suspicious ( 'mass modification' )
    elif ( ( self . delete / len ( actions ) > self . percentage and self . delete > self . delete_threshold ) or self . delete > self . top_threshold ) :
        self . label_suspicious ( 'mass deletion' )
except ZeroDivisionError :
    print ( 'It seems this changeset was redacted' )


signature:
def _unwrap_stream ( uri , timeout , scanner , requests_session ) :
    """Get a stream URI from a playlist URI , ` ` uri ` ` .
 Unwraps nested playlists until something that ' s not a playlist is found or
 the ` ` timeout ` ` is reached ."""

implementation:
original_uri = uri
seen_uris = set ( )
deadline = time . time ( ) + timeout
while time . time ( ) < deadline :
    if uri in seen_uris :
        logger . info ( 'Unwrapping stream from URI (%s) failed: ' 'playlist referenced itself' , uri )
        return None
    else :
        seen_uris . add ( uri )
    logger . debug ( 'Unwrapping stream from URI: %s' , uri )
    try :
        scan_timeout = deadline - time . time ( )
        if scan_timeout < 0 :
            logger . info ( 'Unwrapping stream from URI (%s) failed: ' 'timed out in %sms' , uri , timeout )
            return None
        scan_result = scanner . scan ( uri , timeout = scan_timeout )
    except exceptions . ScannerError as exc :
        logger . debug ( 'GStreamer failed scanning URI (%s): %s' , uri , exc )
        scan_result = None
    if scan_result is not None and not ( scan_result . mime . startswith ( 'text/' ) or scan_result . mime . startswith ( 'application/' ) ) :
        logger . debug ( 'Unwrapped potential %s stream: %s' , scan_result . mime , uri )
        return uri
    download_timeout = deadline - time . time ( )
    if download_timeout < 0 :
        logger . info ( 'Unwrapping stream from URI (%s) failed: timed out in %sms' , uri , timeout )
        return None
    content = http . download ( requests_session , uri , timeout = download_timeout )
    if content is None :
        logger . info ( 'Unwrapping stream from URI (%s) failed: ' 'error downloading URI %s' , original_uri , uri )
        return None
    uris = playlists . parse ( content )
    if not uris :
        logger . debug ( 'Failed parsing URI (%s) as playlist; found potential stream.' , uri )
        return uri
    # TODO Test streams and return first that seems to be playable
    logger . debug ( 'Parsed playlist (%s) and found new URI: %s' , uri , uris [ 0 ] )
    uri = uris [ 0 ]


signature:
def serve ( self , sock , request_handler , error_handler , debug = False , request_timeout = 60 , ssl = None , request_max_size = None , reuse_port = False , loop = None , protocol = HttpProtocol , backlog = 100 , ** kwargs ) :
    """Start asynchronous HTTP Server on an individual process .
 : param request _ handler : Sanic request handler with middleware
 : param error _ handler : Sanic error handler with middleware
 : param debug : enables debug output ( slows server )
 : param request _ timeout : time in seconds
 : param ssl : SSLContext
 : param sock : Socket for the server to accept connections from
 : param request _ max _ size : size in bytes , ` None ` for no limit
 : param reuse _ port : ` True ` for multiple workers
 : param loop : asyncio compatible event loop
 : param protocol : subclass of asyncio protocol class
 : return : Nothing"""

implementation:
if debug :
    loop . set_debug ( debug )
server = partial ( protocol , loop = loop , connections = self . connections , signal = self . signal , request_handler = request_handler , error_handler = error_handler , request_timeout = request_timeout , request_max_size = request_max_size , )
server_coroutine = loop . create_server ( server , host = None , port = None , ssl = ssl , reuse_port = reuse_port , sock = sock , backlog = backlog )
# Instead of pulling time at the end of every request ,
# pull it once per minute
loop . call_soon ( partial ( update_current_time , loop ) )
return server_coroutine


signature:
def spawn ( self , generations ) :
    """Grow this Pantheon by multiplying Gods ."""

implementation:
egg_donors = [ god for god in self . gods . values ( ) if god . chromosomes == 'XX' ]
sperm_donors = [ god for god in self . gods . values ( ) if god . chromosomes == 'XY' ]
for i in range ( generations ) :
    print ( "\nGENERATION %d\n" % ( i + 1 ) )
    gen_xx = [ ]
    gen_xy = [ ]
    for egg_donor in egg_donors :
        sperm_donor = random . choice ( sperm_donors )
        brood = self . breed ( egg_donor , sperm_donor )
        for child in brood :
            if child . divinity > human : # divine offspring join the Pantheon
                self . add_god ( child )
            if child . chromosomes == 'XX' :
                gen_xx . append ( child )
            else :
                gen_xy . append ( child )
    # elder gods leave the breeding pool
    egg_donors = [ ed for ed in egg_donors if ed . generation > ( i - 2 ) ]
    sperm_donors = [ sd for sd in sperm_donors if sd . generation > ( i - 3 ) ]
    # mature offspring join the breeding pool
    egg_donors += gen_xx
    sperm_donors += gen_xy


signature:
def breed ( self , egg_donor , sperm_donor ) :
    """Get it on ."""

implementation:
offspring = [ ]
try :
    num_children = npchoice ( [ 1 , 2 ] , 1 , p = [ 0.8 , 0.2 ] ) [ 0 ]
    # 20 % chance of twins
    for _ in range ( num_children ) :
        child = God ( egg_donor , sperm_donor )
        offspring . append ( child )
        send_birth_announcement ( egg_donor , sperm_donor , child )
except ValueError :
    print ( "Breeding error occurred. Likely the generator ran out of names." )
return offspring


signature:
def get_matches ( word , tokens , limit , offset = 0 ) :
    """Return words from < tokens > that are most closely related to < word > ."""

implementation: return closest ( tokens , word_vec ( word ) , limit , offset )


signature:
def cosine ( vec1 , vec2 ) :
    """Compare vectors . Borrowed from A . Parish ."""

implementation:
if norm ( vec1 ) > 0 and norm ( vec2 ) > 0 :
    return dot ( vec1 , vec2 ) / ( norm ( vec1 ) * norm ( vec2 ) )
else :
    return 0.0


signature:
def closest ( tokens , search_vec , limit , offset = 0 ) :
    """Return the < limit > words from < tokens > whose vectors most closely
 resemble the search _ vec . Skip the first < offset > results ."""

implementation: return sorted ( tokens , key = lambda x : cosine ( search_vec , word_vec ( x ) ) , reverse = True ) [ offset : offset + limit ]


signature:
def tokenize_texts ( ) :
    """Generate a json file for each txt file in the / data / corpora directory ."""

implementation:
text_files = [ fname for fname in os . listdir ( corpora_dir ) if fname . split ( '.' ) [ 1 ] == 'txt' ]
for text_fname in text_files :
    json_fname = text_fname . split ( '.' ) [ 0 ] + '.json'
    if os . path . isfile ( corpora_dir + json_fname ) :
        continue
    # already tokenized
    print ( "Tokenizing " + text_fname )
    text = open ( corpora_dir + text_fname ) . read ( )
    words = nltk . word_tokenize ( text )
    with open ( corpora_dir + json_fname , 'w' ) as outjson :
        json . dump ( words , outjson )


signature:
def make_tokens_dir ( dir_ , sources ) :
    """Create a new directory named < dir _ > . Create a new file within it called
 sources . json . The input < sources > is a list of names of tokenized texts .
 Write < sources > into sources . json ."""

implementation:
os . mkdir ( tokens_dir + dir_ )
for source in sources :
    if not os . path . isfile ( corpora_dir + source ) :
        print ( 'Invalid source: ' + source )
        return
with open ( tokens_dir + dir_ + '/sources.json' , 'w' ) as outjson :
    json . dump ( sources , outjson )


signature:
def make_tokens_list ( dir_ , filters ) :
    """Find sources . json in < dir _ > . It contains a list of tokenized texts . For
 each tokenized text listed in sources . json , read its tokens , filter them ,
 and add them to an aggregated list . Write the aggregated list to disk using
 a filename based on the < filters > given ."""

implementation:
with open ( tokens_dir + dir_ + '/sources.json' , 'r' ) as injson :
    data = json . load ( injson )
    sources = [ corpora_dir + fname for fname in data ]
with open ( 'data/skipwords.txt' , 'r' ) as f :
    skipwords = [ line . rstrip ( ) for line in f ]
tokens_list = [ ]
for fname in sources :
    print ( "Incorporating tokens from " + fname )
    with open ( fname , 'r' ) as injson :
        data = json . load ( injson )
        words = [ w . lower ( ) for w in data if not w == '' ]
        filtered = [ w for w , p in nltk . pos_tag ( words ) if p in filters ]
        sanitized = [ w for w in filtered if not w in skipwords ]
        tokens_list += sanitized
tokens_list = list ( set ( tokens_list ) )
# unique
target = tokens_dir + dir_ + '/' + '-' . join ( filters ) + '.json'
with open ( target , 'w' ) as outjson :
    json . dump ( tokens_list , outjson )


signature:
def set_name_lists ( ethnicity = None ) :
    """Set three globally available lists of names ."""

implementation:
if not ethnicity :
    ethnicity = random . choice ( get_ethnicities ( ) )
print ( "Loading names from: " + ethnicity )
filename = names_dir + ethnicity + '.json'
try :
    with open ( filename , 'r' ) as injson :
        data = json . load ( injson )
except :
    return 'Unable to read from file: ' + filename
else :
    names = [ tuple ( name . split ( ',' ) ) for name in data ]
    random . shuffle ( names )
    global female_names
    female_names = [ name for name , gender , * desc in names if gender == 'girl' ]
    global male_names
    male_names = [ name for name , gender , * desc in names if gender == 'boy' ]
    global nb_names
    nb_names = [ name for name , gender , * desc in names if gender == 'boygirl' ]


signature:
def set_chromosomes ( self , chromosomes = None ) :
    """This model uses the XY sex - determination system . Sex ! = gender .
 Assign either XX or XY randomly with a 50/50 chance of each , unless
 < chromosomes > are passed as an argument ."""

implementation:
if chromosomes and chromosomes in valid_chromosomes :
    self . chromosomes = chromosomes
else :
    self . chromosomes = random . choice ( [ XX , XY ] )


signature:
def set_gender ( self , gender = None ) :
    """This model recognizes that sex chromosomes don ' t always line up with
 gender . Assign M , F , or NB according to the probabilities in p _ gender ."""

implementation:
if gender and gender in genders :
    self . gender = gender
else :
    if not self . chromosomes :
        self . set_chromosomes ( )
    self . gender = npchoice ( genders , 1 , p = p_gender [ self . chromosomes ] ) [ 0 ]


signature:
def set_inherited_traits ( self , egg_donor , sperm_donor ) :
    """Accept either strings or Gods as inputs ."""

implementation:
if type ( egg_donor ) == str :
    self . reproduce_asexually ( egg_donor , sperm_donor )
else :
    self . reproduce_sexually ( egg_donor , sperm_donor )


signature:
def reproduce_asexually ( self , egg_word , sperm_word ) :
    """Produce two gametes , an egg and a sperm , from the input strings .
 Combine them to produce a genome a la sexual reproduction ."""

implementation:
egg = self . generate_gamete ( egg_word )
sperm = self . generate_gamete ( sperm_word )
self . genome = list ( set ( egg + sperm ) )
# Eliminate duplicates
self . generation = 1
self . divinity = god


signature:
def reproduce_sexually ( self , egg_donor , sperm_donor ) :
    """Produce two gametes , an egg and a sperm , from input Gods . Combine
 them to produce a genome a la sexual reproduction . Assign divinity
 according to probabilities in p _ divinity . The more divine the parents ,
 the more divine their offspring ."""

implementation:
egg_word = random . choice ( egg_donor . genome )
egg = self . generate_gamete ( egg_word )
sperm_word = random . choice ( sperm_donor . genome )
sperm = self . generate_gamete ( sperm_word )
self . genome = list ( set ( egg + sperm ) )
# Eliminate duplicates
self . parents = [ egg_donor . name , sperm_donor . name ]
self . generation = max ( egg_donor . generation , sperm_donor . generation ) + 1
sum_ = egg_donor . divinity + sperm_donor . divinity
self . divinity = int ( npchoice ( divinities , 1 , p = p_divinity [ sum_ ] ) [ 0 ] )


signature:
def set_name ( self ) :
    """Pick a random name from the lists loaded with the model . For Gods that
 identify as neither M nor F , the model attempts to retrieve an androgynous
 name . Note : not all of the scraped name lists contain androgynous names ."""

implementation:
if not self . gender :
    self . set_gender ( )
name = ''
if self . gender == female :
    name = names . female_names . pop ( )
elif self . gender == male :
    name = names . male_names . pop ( )
else :
    try :
        name = names . nb_names . pop ( )
    except : # No androgynous names available
        name = names . male_names . pop ( )
self . name = name


signature:
def set_epithet ( self ) :
    """Divine an appropriate epithet for this God . ( See what I did there ? )"""

implementation:
if self . divinity == human :
    obsession = random . choice ( self . genome )
    if self . gender == female :
        self . epithet = 'ordinary woman'
    elif self . gender == male :
        self . epithet = 'ordinary man'
    else :
        self . epithet = 'ordinary human being'
    self . epithet += ' who loves ' + obsession
    return
# Return early . The rest of the function deals with gods .
if self . gender == female :
    title = 'Goddess'
elif self . gender == male :
    title = 'God'
else :
    title = 'Divine Being'
if self . divinity == demi_god :
    title = 'Semi-' + title if self . gender == non_binary else 'Demi-' + title
num_domains = npchoice ( [ 1 , 2 , 3 , 4 ] , 1 , p = [ 0.05 , 0.35 , 0.55 , 0.05 ] ) [ 0 ]
if num_domains == 1 :
    template = '%s of %s'
if num_domains == 2 :
    template = '%s of %s and %s'
elif num_domains == 3 :
    template = '%s of %s, %s, and %s'
    # Oxford comma , the most divine punctuation .
elif num_domains == 4 :
    template = '%s of %s, %s, %s, and %s'
self . domains = [ d . title ( ) for d in random . sample ( self . genome , num_domains ) ]
# Put it all together
self . epithet = template % ( title , * self . domains )


signature:
def generate_gamete ( self , egg_or_sperm_word ) :
    """Extract 23 ' chromosomes ' aka words from ' gene pool ' aka list of tokens
 by searching the list of tokens for words that are related to the given
 egg _ or _ sperm _ word ."""

implementation:
p_rate_of_mutation = [ 0.9 , 0.1 ]
should_use_mutant_pool = ( npchoice ( [ 0 , 1 ] , 1 , p = p_rate_of_mutation ) [ 0 ] == 1 )
if should_use_mutant_pool :
    pool = tokens . secondary_tokens
else :
    pool = tokens . primary_tokens
return get_matches ( egg_or_sperm_word , pool , 23 )


signature:
def print_parents ( self ) :
    """Print parents ' names and epithets ."""

implementation:
if self . gender == female :
    title = 'Daughter'
elif self . gender == male :
    title = 'Son'
else :
    title = 'Child'
p1 = self . parents [ 0 ]
p2 = self . parents [ 1 ]
template = '%s of %s, the %s, and %s, the %s.'
print ( template % ( title , p1 . name , p1 . epithet , p2 . name , p2 . epithet ) )


signature:
def instance ( self , counter = None , pipeline_counter = None ) :
    """Returns all the information regarding a specific stage run
 See the ` Go stage instance documentation ` _ _ for examples .
 . . _ _ : http : / / api . go . cd / current / # get - stage - instance
 Args :
 counter ( int ) : The stage instance to fetch .
 If falsey returns the latest stage instance from : meth : ` history ` .
 pipeline _ counter ( int ) : The pipeline instance for which to fetch
 the stage . If falsey returns the latest pipeline instance .
 Returns :
 Response : : class : ` gocd . api . response . Response ` object"""

implementation:
pipeline_counter = pipeline_counter or self . pipeline_counter
pipeline_instance = None
if not pipeline_counter :
    pipeline_instance = self . server . pipeline ( self . pipeline_name ) . instance ( )
    self . pipeline_counter = int ( pipeline_instance [ 'counter' ] )
if not counter :
    if pipeline_instance is None :
        pipeline_instance = ( self . server . pipeline ( self . pipeline_name ) . instance ( pipeline_counter ) )
    for stages in pipeline_instance [ 'stages' ] :
        if stages [ 'name' ] == self . stage_name :
            return self . instance ( counter = int ( stages [ 'counter' ] ) , pipeline_counter = pipeline_counter )
return self . _get ( '/instance/{pipeline_counter:d}/{counter:d}' . format ( pipeline_counter = pipeline_counter , counter = counter ) )


signature:
def is_json ( self ) :
    """Returns :
 bool : True if ` content _ type ` is ` application / json `"""

implementation: return ( self . content_type . startswith ( 'application/json' ) or re . match ( r'application/vnd.go.cd.v(\d+)\+json' , self . content_type ) )


signature:
def payload ( self ) :
    """Returns :
 ` str ` when not json .
 ` dict ` when json ."""

implementation:
if self . is_json :
    if not self . _body_parsed :
        if hasattr ( self . _body , 'decode' ) :
            body = self . _body . decode ( 'utf-8' )
        else :
            body = self . _body
        self . _body_parsed = json . loads ( body )
    return self . _body_parsed
else :
    return self . _body


signature:
def request ( self , path , data = None , headers = None , method = None ) :
    """Performs a HTTP request to the Go server
 Args :
 path ( str ) : The full path on the Go server to request .
 This includes any query string attributes .
 data ( str , dict , bool , optional ) : If any data is present this
 request will become a POST request .
 headers ( dict , optional ) : Headers to set for this particular
 request
 Raises :
 HTTPError : when the HTTP request fails .
 Returns :
 file like object : The response from a
 : func : ` urllib2 . urlopen ` call"""

implementation:
if isinstance ( data , str ) :
    data = data . encode ( 'utf-8' )
response = urlopen ( self . _request ( path , data = data , headers = headers , method = method ) )
self . _set_session_cookie ( response )
return response


signature:
def add_logged_in_session ( self , response = None ) :
    """Make the request appear to be coming from a browser
 This is to interact with older parts of Go that doesn ' t have a
 proper API call to be made . What will be done :
 1 . If no response passed in a call to ` go / api / pipelines . xml ` is
 made to get a valid session
 2 . ` JSESSIONID ` will be populated from this request
 3 . A request to ` go / pipelines ` will be so the
 ` authenticity _ token ` ( CSRF ) can be extracted . It will then
 silently be injected into ` post _ args ` on any POST calls that
 doesn ' t start with ` go / api ` from this point .
 Args :
 response : a : class : ` Response ` object from a previously successful
 API call . So we won ' t have to query ` go / api / pipelines . xml `
 unnecessarily .
 Raises :
 HTTPError : when the HTTP request fails .
 AuthenticationFailed : when failing to get the ` session _ id `
 or the ` authenticity _ token ` ."""

implementation:
if not response :
    response = self . get ( 'go/api/pipelines.xml' )
self . _set_session_cookie ( response )
if not self . _session_id :
    raise AuthenticationFailed ( 'No session id extracted from request.' )
response = self . get ( 'go/pipelines' )
match = re . search ( r'name="authenticity_token".+?value="([^"]+)' , response . read ( ) . decode ( 'utf-8' ) )
if match :
    self . _authenticity_token = match . group ( 1 )
else :
    raise AuthenticationFailed ( 'Authenticity token not found on page' )


signature:
def stage ( self , pipeline_name , stage_name , pipeline_counter = None ) :
    """Returns an instance of : class : ` Stage `
 Args :
 pipeline _ name ( str ) : Name of the pipeline the stage belongs to
 stage _ name ( str ) : Name of the stage to act on
 pipeline _ counter ( int ) : The pipeline instance the stage is for .
 Returns :
 Stage : an instantiated : class : ` Stage ` ."""

implementation: return Stage ( self , pipeline_name , stage_name , pipeline_counter = pipeline_counter )


implementation:
if not isinstance ( d , dict ) :
    return [ [ d ] ]
returned = [ ]
for key , value in d . items ( ) : # Each key , value is treated as a row .
    nested = flatten ( value )
    for nest in nested :
        current_row = [ key ]
        current_row . extend ( nest )
        returned . append ( current_row )
return returned


signature:
def instance ( self , counter = None ) :
    """Returns all the information regarding a specific pipeline run
 See the ` Go pipeline instance documentation ` _ _ for examples .
 . . _ _ : http : / / api . go . cd / current / # get - pipeline - instance
 Args :
 counter ( int ) : The pipeline instance to fetch .
 If falsey returns the latest pipeline instance from : meth : ` history ` .
 Returns :
 Response : : class : ` gocd . api . response . Response ` object"""

implementation:
if not counter :
    history = self . history ( )
    if not history :
        return history
    else :
        return Response . _from_json ( history [ 'pipelines' ] [ 0 ] )
return self . _get ( '/instance/{counter:d}' . format ( counter = counter ) )


signature:
def schedule ( self , variables = None , secure_variables = None , materials = None , return_new_instance = False , backoff_time = 1.0 ) :
    """Schedule a pipeline run
 Aliased as : meth : ` run ` , : meth : ` schedule ` , and : meth : ` trigger ` .
 Args :
 variables ( dict , optional ) : Variables to set / override
 secure _ variables ( dict , optional ) : Secure variables to set / override
 materials ( dict , optional ) : Material revisions to be used for
 this pipeline run . The exact format for this is a bit iffy ,
 have a look at the official
 ` Go pipeline scheduling documentation ` _ _ or inspect a call
 from triggering manually in the UI .
 return _ new _ instance ( bool ) : Returns a : meth : ` history ` compatible
 response for the newly scheduled instance . This is primarily so
 users easily can get the new instance number . * * Note : * * This is done
 in a very naive way , it just checks that the instance number is
 higher than before the pipeline was triggered .
 backoff _ time ( float ) : How long between each check for
 : arg : ` return _ new _ instance ` .
 . . _ _ : http : / / api . go . cd / current / # scheduling - pipelines
 Returns :
 Response : : class : ` gocd . api . response . Response ` object"""

implementation:
scheduling_args = dict ( variables = variables , secure_variables = secure_variables , material_fingerprint = materials , headers = { "Confirm" : True } , )
scheduling_args = dict ( ( k , v ) for k , v in scheduling_args . items ( ) if v is not None )
# TODO : Replace this with whatever is the official way as soon as gocd # 990 is fixed .
# https : / / github . com / gocd / gocd / issues / 990
if return_new_instance :
    pipelines = self . history ( ) [ 'pipelines' ]
    if len ( pipelines ) == 0 :
        last_run = None
    else :
        last_run = pipelines [ 0 ] [ 'counter' ]
    response = self . _post ( '/schedule' , ok_status = 202 , ** scheduling_args )
    if not response :
        return response
    max_tries = 10
    while max_tries > 0 :
        current = self . instance ( )
        if not last_run and current :
            return current
        elif last_run and current [ 'counter' ] > last_run :
            return current
        else :
            time . sleep ( backoff_time )
            max_tries -= 1
    # I can ' t come up with a scenario in testing where this would happen , but it seems
    # better than returning None .
    return response
else :
    return self . _post ( '/schedule' , ok_status = 202 , ** scheduling_args )


signature:
def artifact ( self , counter , stage , job , stage_counter = 1 ) :
    """Helper to instantiate an : class : ` gocd . api . artifact . Artifact ` object
 Args :
 counter ( int ) : The pipeline counter to get the artifact for
 stage : Stage name
 job : Job name
 stage _ counter : Defaults to 1
 Returns :
 Artifact : : class : ` gocd . api . artifact . Artifact ` object"""

implementation: return Artifact ( self . server , self . name , counter , stage , job , stage_counter )


signature:
def console_output ( self , instance = None ) :
    """Yields the output and metadata from all jobs in the pipeline
 Args :
 instance : The result of a : meth : ` instance ` call , if not supplied
 the latest of the pipeline will be used .
 Yields :
 tuple : ( metadata ( dict ) , output ( str ) ) .
 metadata contains :
 - pipeline
 - pipeline _ counter
 - stage
 - stage _ counter
 - job
 - job _ result"""

implementation:
if instance is None :
    instance = self . instance ( )
for stage in instance [ 'stages' ] :
    for job in stage [ 'jobs' ] :
        if job [ 'result' ] not in self . final_results :
            continue
        artifact = self . artifact ( instance [ 'counter' ] , stage [ 'name' ] , job [ 'name' ] , stage [ 'counter' ] )
        output = artifact . get ( 'cruise-output/console.log' )
        yield ( { 'pipeline' : self . name , 'pipeline_counter' : instance [ 'counter' ] , 'stage' : stage [ 'name' ] , 'stage_counter' : stage [ 'counter' ] , 'job' : job [ 'name' ] , 'job_result' : job [ 'result' ] , } , output . body )


signature:
def stage ( self , name , pipeline_counter = None ) :
    """Helper to instantiate a : class : ` gocd . api . stage . Stage ` object
 Args :
 name : The name of the stage
 pipeline _ counter :
 Returns :"""

implementation: return Stage ( self . server , pipeline_name = self . name , stage_name = name , pipeline_counter = pipeline_counter , )


signature:
def edit ( self , config , etag ) :
    """Update template config for specified template name .
 . . _ _ : https : / / api . go . cd / current / # edit - template - config
 Returns :
 Response : : class : ` gocd . api . response . Response ` object"""

implementation:
data = self . _json_encode ( config )
headers = self . _default_headers ( )
if etag is not None :
    headers [ "If-Match" ] = etag
return self . _request ( self . name , ok_status = None , data = data , headers = headers , method = "PUT" )


signature:
def create ( self , config ) :
    """Create template config for specified template name .
 . . _ _ : https : / / api . go . cd / current / # create - template - config
 Returns :
 Response : : class : ` gocd . api . response . Response ` object"""

implementation:
assert config [ "name" ] == self . name , "Given config is not for this template"
data = self . _json_encode ( config )
headers = self . _default_headers ( )
return self . _request ( "" , ok_status = None , data = data , headers = headers )


signature:
def delete ( self ) :
    """Delete template config for specified template name .
 . . _ _ : https : / / api . go . cd / current / # delete - a - template
 Returns :
 Response : : class : ` gocd . api . response . Response ` object"""

implementation:
headers = self . _default_headers ( )
return self . _request ( self . name , ok_status = None , data = None , headers = headers , method = "DELETE" )


signature:
def pipelines ( self ) :
    """Returns a set of all pipelines from the last response
 Returns :
 set : Response success : all the pipelines available in the response
 Response failure : an empty set"""

implementation:
if not self . response :
    return set ( )
elif self . _pipelines is None and self . response :
    self . _pipelines = set ( )
    for group in self . response . payload :
        for pipeline in group [ 'pipelines' ] :
            self . _pipelines . add ( pipeline [ 'name' ] )
return self . _pipelines


signature:
def get_directory ( self , path_to_directory , timeout = 30 , backoff = 0.4 , max_wait = 4 ) :
    """Gets an artifact directory by its path .
 See the ` Go artifact directory documentation ` _ _ for example responses .
 . . _ _ : http : / / api . go . cd / current / # get - artifact - directory
 . . note : :
 Getting a directory relies on Go creating a zip file of the
 directory in question . Because of this Go will zip the file in
 the background and return a 202 Accepted response . It ' s then up
 to the client to check again later and get the final file .
 To work with normal assumptions this : meth : ` get _ directory ` will
 retry itself up to ` ` timeout ` ` seconds to get a 200 response to
 return . At that point it will then return the response as is , no
 matter whether it ' s still 202 or 200 . The retry is done with an
 exponential backoff with a max value between retries . See the
 ` ` backoff ` ` and ` ` max _ wait ` ` variables .
 If you want to handle the retry logic yourself then use : meth : ` get `
 and add ' . zip ' as a suffix on the directory .
 Args :
 path _ to _ directory ( str ) : The path to the directory to get .
 It can be nested eg ` ` target / dist . zip ` `
 timeout ( int ) : How many seconds we will wait in total for a
 successful response from Go when we ' re receiving 202
 backoff ( float ) : The initial value used for backoff , raises
 exponentially until it reaches ` ` max _ wait ` `
 max _ wait ( int ) : The max time between retries
 Returns :
 Response : : class : ` gocd . api . response . Response ` object
 A successful response is a zip - file ."""

implementation:
response = None
started_at = None
time_elapsed = 0
i = 0
while time_elapsed < timeout :
    response = self . _get ( '{0}.zip' . format ( path_to_directory ) )
    if response :
        break
    else :
        if started_at is None :
            started_at = time . time ( )
        time . sleep ( min ( backoff * ( 2 ** i ) , max_wait ) )
        i += 1
        time_elapsed = time . time ( ) - started_at
return response


signature:
def ask ( self ) :
    """Return the wait time in seconds required to retrieve the
 item currently at the head of the queue .
 Note that there is no guarantee that a call to ` get ( ) ` will
 succeed even if ` ask ( ) ` returns 0 . By the time the calling
 thread reacts , other threads may have caused a different
 item to be at the head of the queue ."""

implementation:
with self . mutex :
    if not len ( self . queue ) :
        raise Empty
    utcnow = dt . datetime . utcnow ( )
    if self . queue [ 0 ] [ 0 ] <= utcnow :
        self . ready . notify ( )
        return 0
    return ( self . queue [ 0 ] [ 0 ] - utcnow ) . total_seconds ( )


signature:
def p_transition ( p ) :
    """transition : START _ KWD KEY NULL _ KWD FLOAT
 transition : KEY KEY NULL _ KWD FLOAT
 transition : KEY END _ KWD NULL _ KWD FLOAT
 transition : START _ KWD KEY KEY FLOAT
 transition : KEY KEY KEY FLOAT
 transition : KEY END _ KWD KEY FLOAT
 transition : START _ KWD KEY NULL _ KWD INTEGER
 transition : KEY KEY NULL _ KWD INTEGER
 transition : KEY END _ KWD NULL _ KWD INTEGER
 transition : START _ KWD KEY KEY INTEGER
 transition : KEY KEY KEY INTEGER
 transition : KEY END _ KWD KEY INTEGER
 transition : START _ KWD KEY NULL _ KWD KEY
 transition : KEY KEY NULL _ KWD KEY
 transition : KEY END _ KWD NULL _ KWD KEY
 transition : START _ KWD KEY KEY KEY
 transition : KEY KEY KEY KEY
 transition : KEY END _ KWD KEY KEY"""

implementation:
p [ 3 ] = None if p [ 3 ] == 'NULL' else p [ 3 ]
if p [ 4 ] == 'error' :
    p [ 0 ] = MarionetteTransition ( p [ 1 ] , p [ 2 ] , p [ 3 ] , 0 , True )
else :
    p [ 0 ] = MarionetteTransition ( p [ 1 ] , p [ 2 ] , p [ 3 ] , p [ 4 ] , False )


signature:
def p_action_blocks ( p ) :
    """action _ blocks : action _ blocks action _ block"""


signature:
def p_action_block ( p ) :
    """action _ block : ACTION _ KWD KEY COLON actions"""


signature:
def p_action ( p ) :
    """action : CLIENT _ KWD KEY DOT KEY LPAREN args RPAREN
 action : SERVER _ KWD KEY DOT KEY LPAREN args RPAREN
 action : CLIENT _ KWD KEY DOT KEY LPAREN args RPAREN IF _ KWD REGEX _ MATCH _ INCOMING _ KWD LPAREN p _ string _ arg RPAREN
 action : SERVER _ KWD KEY DOT KEY LPAREN args RPAREN IF _ KWD REGEX _ MATCH _ INCOMING _ KWD LPAREN p _ string _ arg RPAREN"""


signature:
def config_loader ( app , ** kwargs_config ) :
    """Configuration loader .
 Adds support for loading templates from the Flask application ' s instance
 folder ( ` ` < instance _ folder > / templates ` ` ) ."""

implementation:
# This is the only place customize the Flask application right after
# it has been created , but before all extensions etc are loaded .
local_templates_path = os . path . join ( app . instance_path , 'templates' )
if os . path . exists ( local_templates_path ) : # Let ' s customize the template loader to look into packages
# and application templates folders .
    app . jinja_loader = ChoiceLoader ( [ FileSystemLoader ( local_templates_path ) , app . jinja_loader , ] )
app . jinja_options = dict ( app . jinja_options , cache_size = 1000 , bytecode_cache = BytecodeCache ( app ) )
invenio_config_loader ( app , ** kwargs_config )


signature:
def app_class ( ) :
    """Create Flask application class .
 Invenio - Files - REST needs to patch the Werkzeug form parsing in order to
 support streaming large file uploads . This is done by subclassing the Flask
 application class ."""

implementation:
try :
    pkg_resources . get_distribution ( 'invenio-files-rest' )
    from invenio_files_rest . app import Flask as FlaskBase
except pkg_resources . DistributionNotFound :
    from flask import Flask as FlaskBase
# Add Host header validation via APP _ ALLOWED _ HOSTS configuration variable .
class Request ( TrustedHostsMixin , FlaskBase . request_class ) :
    pass
class Flask ( FlaskBase ) :
    request_class = Request
return Flask


signature:
def init_config ( self , app ) :
    """Initialize configuration .
 : param app : An instance of : class : ` ~ flask . Flask ` ."""

implementation:
config_apps = [ 'APP_' , 'RATELIMIT_' ]
flask_talisman_debug_mode = [ "'unsafe-inline'" ]
for k in dir ( config ) :
    if any ( [ k . startswith ( prefix ) for prefix in config_apps ] ) :
        app . config . setdefault ( k , getattr ( config , k ) )
if app . config [ 'DEBUG' ] :
    app . config . setdefault ( 'APP_DEFAULT_SECURE_HEADERS' , { } )
    headers = app . config [ 'APP_DEFAULT_SECURE_HEADERS' ]
    # ensure ` content _ security _ policy ` is not set to { }
    if headers . get ( 'content_security_policy' ) != { } :
        headers . setdefault ( 'content_security_policy' , { } )
        csp = headers [ 'content_security_policy' ]
        # ensure ` default - src ` is not set to [ ]
        if csp . get ( 'default-src' ) != [ ] :
            csp . setdefault ( 'default-src' , [ ] )
            # add default ` content _ security _ policy ` value when debug
            csp [ 'default-src' ] += flask_talisman_debug_mode


signature:
def remove_leading ( needle , haystack ) :
    """Remove leading needle string ( if exists ) .
 > > > remove _ leading ( ' Test ' , ' TestThisAndThat ' )
 ' ThisAndThat '
 > > > remove _ leading ( ' Test ' , ' ArbitraryName ' )
 ' ArbitraryName '"""

implementation:
if haystack [ : len ( needle ) ] == needle :
    return haystack [ len ( needle ) : ]
return haystack


signature:
def remove_trailing ( needle , haystack ) :
    """Remove trailing needle string ( if exists ) .
 > > > remove _ trailing ( ' Test ' , ' ThisAndThatTest ' )
 ' ThisAndThat '
 > > > remove _ trailing ( ' Test ' , ' ArbitraryName ' )
 ' ArbitraryName '"""

implementation:
if haystack [ - len ( needle ) : ] == needle :
    return haystack [ : - len ( needle ) ]
return haystack


signature:
def camel2word ( string ) :
    """Covert name from CamelCase to " Normal case " .
 > > > camel2word ( ' CamelCase ' )
 ' Camel case '
 > > > camel2word ( ' CaseWithSpec ' )
 ' Case with spec '"""

implementation:
def wordize ( match ) :
    return ' ' + match . group ( 1 ) . lower ( )
return string [ 0 ] + re . sub ( r'([A-Z])' , wordize , string [ 1 : ] )


signature:
def complete_english ( string ) :
    """> > > complete _ english ( ' dont do this ' )
 " don ' t do this "
 > > > complete _ english ( ' doesnt is matched as well ' )
 " doesn ' t is matched as well " """

implementation:
for x , y in [ ( "dont" , "don't" ) , ( "doesnt" , "doesn't" ) , ( "wont" , "won't" ) , ( "wasnt" , "wasn't" ) ] :
    string = string . replace ( x , y )
return string


signature:
def format_seconds ( self , n_seconds ) :
    """Format a time in seconds ."""

implementation:
func = self . ok
if n_seconds >= 60 :
    n_minutes , n_seconds = divmod ( n_seconds , 60 )
    return "%s minutes %s seconds" % ( func ( "%d" % n_minutes ) , func ( "%.3f" % n_seconds ) )
else :
    return "%s seconds" % ( func ( "%.3f" % n_seconds ) )