signature:
def export_namespaces ( graph , namespaces , directory = None , cacheable = False ) :
    """Thinly wraps : func : ` export _ namespace ` for an iterable of namespaces .
 : param pybel . BELGraph graph : A BEL graph
 : param iter [ str ] namespaces : An iterable of strings for the namespaces to process
 : param str directory : The path to the directory where to output the namespaces . Defaults to the current working
 directory returned by : func : ` os . getcwd `
 : param bool cacheable : Should the namespaces be cacheable ? Defaults to ` ` False ` ` because , in general , this operation
 will probably be used for evil , and users won ' t want to reload their entire cache after each
 iteration of curation ."""

implementation:
directory = os . getcwd ( ) if directory is None else directory
# avoid making multiple calls to os . getcwd later
for namespace in namespaces :
    export_namespace ( graph , namespace , directory = directory , cacheable = cacheable )


signature:
def lint_file ( in_file , out_file = None ) :
    """Helps remove extraneous whitespace from the lines of a file
 : param file in _ file : A readable file or file - like
 : param file out _ file : A writable file or file - like"""

implementation:
for line in in_file :
    print ( line . strip ( ) , file = out_file )


signature:
def lint_directory ( source , target ) :
    """Adds a linted version of each document in the source directory to the target directory
 : param str source : Path to directory to lint
 : param str target : Path to directory to output"""

implementation:
for path in os . listdir ( source ) :
    if not path . endswith ( '.bel' ) :
        continue
    log . info ( 'linting: %s' , path )
    with open ( os . path . join ( source , path ) ) as i , open ( os . path . join ( target , path ) , 'w' ) as o :
        lint_file ( i , o )


signature:
def make_pubmed_abstract_group ( pmids : Iterable [ Union [ str , int ] ] ) -> Iterable [ str ] :
    """Build a skeleton for the citations ' statements .
 : param pmids : A list of PubMed identifiers
 : return : An iterator over the lines of the citation section"""

implementation:
for pmid in set ( pmids ) :
    yield ''
    res = requests . get ( title_url_fmt . format ( pmid ) )
    title = res . content . decode ( 'utf-8' ) . strip ( )
    yield 'SET Citation = {{"{}", "{}"}}' . format ( title , pmid )
    res = requests . get ( abstract_url_fmt . format ( pmid ) )
    abstract = res . content . decode ( 'utf-8' ) . strip ( )
    yield 'SET Evidence = "{}"' . format ( abstract )
    yield '\nUNSET Evidence\nUNSET Citation'


signature:
def get_entrez_gene_data ( entrez_ids : Iterable [ Union [ str , int ] ] ) :
    """Get gene info from Entrez ."""

implementation:
url = PUBMED_GENE_QUERY_URL . format ( ',' . join ( str ( x ) . strip ( ) for x in entrez_ids ) )
response = requests . get ( url )
tree = ElementTree . fromstring ( response . content )
return { element . attrib [ 'uid' ] : { 'summary' : _sanitize ( element . find ( 'Summary' ) . text ) , 'description' : element . find ( 'Description' ) . text } for element in tree . findall ( './DocumentSummarySet/DocumentSummary' ) }


signature:
def make_pubmed_gene_group ( entrez_ids : Iterable [ Union [ str , int ] ] ) -> Iterable [ str ] :
    """Builds a skeleton for gene summaries
 : param entrez _ ids : A list of Entrez Gene identifiers to query the PubMed service
 : return : An iterator over statement lines for NCBI Entrez Gene summaries"""

implementation:
url = PUBMED_GENE_QUERY_URL . format ( ',' . join ( str ( x ) . strip ( ) for x in entrez_ids ) )
response = requests . get ( url )
tree = ElementTree . fromstring ( response . content )
for x in tree . findall ( './DocumentSummarySet/DocumentSummary' ) :
    yield '\n# {}' . format ( x . find ( 'Description' ) . text )
    yield 'SET Citation = {{"Other", "PubMed Gene", "{}"}}' . format ( x . attrib [ 'uid' ] )
    yield 'SET Evidence = "{}"' . format ( x . find ( 'Summary' ) . text . strip ( ) . replace ( '\n' , '' ) )
    yield '\nUNSET Evidence\nUNSET Citation'


signature:
def write_boilerplate ( name : str , version : Optional [ str ] = None , description : Optional [ str ] = None , authors : Optional [ str ] = None , contact : Optional [ str ] = None , copyright : Optional [ str ] = None , licenses : Optional [ str ] = None , disclaimer : Optional [ str ] = None , namespace_url : Optional [ Mapping [ str , str ] ] = None , namespace_patterns : Optional [ Mapping [ str , str ] ] = None , annotation_url : Optional [ Mapping [ str , str ] ] = None , annotation_patterns : Optional [ Mapping [ str , str ] ] = None , annotation_list : Optional [ Mapping [ str , Set [ str ] ] ] = None , pmids : Optional [ Iterable [ Union [ str , int ] ] ] = None , entrez_ids : Optional [ Iterable [ Union [ str , int ] ] ] = None , file : Optional [ TextIO ] = None , ) -> None :
    """Write a boilerplate BEL document , with standard document metadata , definitions .
 : param name : The unique name for this BEL document
 : param contact : The email address of the maintainer
 : param description : A description of the contents of this document
 : param authors : The authors of this document
 : param version : The version . Defaults to current date in format ` ` YYYYMMDD ` ` .
 : param copyright : Copyright information about this document
 : param licenses : The license applied to this document
 : param disclaimer : The disclaimer for this document
 : param namespace _ url : an optional dictionary of { str name : str URL } of namespaces
 : param namespace _ patterns : An optional dictionary of { str name : str regex } namespaces
 : param annotation _ url : An optional dictionary of { str name : str URL } of annotations
 : param annotation _ patterns : An optional dictionary of { str name : str regex } of regex annotations
 : param annotation _ list : An optional dictionary of { str name : set of names } of list annotations
 : param pmids : A list of PubMed identifiers to auto - populate with citation and abstract
 : param entrez _ ids : A list of Entrez identifiers to autopopulate the gene summary as evidence
 : param file : A writable file or file - like . If None , defaults to : data : ` sys . stdout `"""

implementation:
lines = make_knowledge_header ( name = name , version = version or '1.0.0' , description = description , authors = authors , contact = contact , copyright = copyright , licenses = licenses , disclaimer = disclaimer , namespace_url = namespace_url , namespace_patterns = namespace_patterns , annotation_url = annotation_url , annotation_patterns = annotation_patterns , annotation_list = annotation_list , )
for line in lines :
    print ( line , file = file )
if pmids is not None :
    for line in make_pubmed_abstract_group ( pmids ) :
        print ( line , file = file )
if entrez_ids is not None :
    for line in make_pubmed_gene_group ( entrez_ids ) :
        print ( line , file = file )


signature:
def get_subgraph_by_node_filter ( graph : BELGraph , node_predicates : NodePredicates ) -> BELGraph :
    """Induce a sub - graph on the nodes that pass the given predicate ( s ) ."""

implementation: return get_subgraph_by_induction ( graph , filter_nodes ( graph , node_predicates ) )


signature:
def get_subgraph_by_node_search ( graph : BELGraph , query : Strings ) -> BELGraph :
    """Get a sub - graph induced over all nodes matching the query string .
 : param graph : A BEL Graph
 : param query : A query string or iterable of query strings for node names
 Thinly wraps : func : ` search _ node _ names ` and : func : ` get _ subgraph _ by _ induction ` ."""

implementation:
nodes = search_node_names ( graph , query )
return get_subgraph_by_induction ( graph , nodes )


signature:
def get_largest_component ( graph : BELGraph ) -> BELGraph :
    """Get the giant component of a graph ."""

implementation:
biggest_component_nodes = max ( nx . weakly_connected_components ( graph ) , key = len )
return subgraph ( graph , biggest_component_nodes )


signature:
def random_by_nodes ( graph : BELGraph , percentage : Optional [ float ] = None ) -> BELGraph :
    """Get a random graph by inducing over a percentage of the original nodes .
 : param graph : A BEL graph
 : param percentage : The percentage of edges to keep"""

implementation:
percentage = percentage or 0.9
assert 0 < percentage <= 1
nodes = graph . nodes ( )
n = int ( len ( nodes ) * percentage )
subnodes = random . sample ( nodes , n )
result = graph . subgraph ( subnodes )
update_node_helper ( graph , result )
return result


signature:
def random_by_edges ( graph : BELGraph , percentage : Optional [ float ] = None ) -> BELGraph :
    """Get a random graph by keeping a certain percentage of original edges .
 : param graph : A BEL graph
 : param percentage : What percentage of eges to take"""

implementation:
percentage = percentage or 0.9
assert 0 < percentage <= 1
edges = graph . edges ( keys = True )
n = int ( graph . number_of_edges ( ) * percentage )
subedges = random . sample ( edges , n )
rv = graph . fresh_copy ( )
for u , v , k in subedges :
    safe_add_edge ( rv , u , v , k , graph [ u ] [ v ] [ k ] )
update_node_helper ( graph , rv )
return rv


signature:
def shuffle_node_data ( graph : BELGraph , key : str , percentage : Optional [ float ] = None ) -> BELGraph :
    """Shuffle the node ' s data .
 Useful for permutation testing .
 : param graph : A BEL graph
 : param key : The node data dictionary key
 : param percentage : What percentage of possible swaps to make"""

implementation:
percentage = percentage or 0.3
assert 0 < percentage <= 1
n = graph . number_of_nodes ( )
swaps = int ( percentage * n * ( n - 1 ) / 2 )
result : BELGraph = graph . copy ( )
for _ in range ( swaps ) :
    s , t = random . sample ( result . node , 2 )
    result . nodes [ s ] [ key ] , result . nodes [ t ] [ key ] = result . nodes [ t ] [ key ] , result . nodes [ s ] [ key ]
return result


signature:
def shuffle_relations ( graph : BELGraph , percentage : Optional [ str ] = None ) -> BELGraph :
    """Shuffle the relations .
 Useful for permutation testing .
 : param graph : A BEL graph
 : param percentage : What percentage of possible swaps to make"""

implementation:
percentage = percentage or 0.3
assert 0 < percentage <= 1
n = graph . number_of_edges ( )
swaps = int ( percentage * n * ( n - 1 ) / 2 )
result : BELGraph = graph . copy ( )
edges = result . edges ( keys = True )
for _ in range ( swaps ) :
    ( s1 , t1 , k1 ) , ( s2 , t2 , k2 ) = random . sample ( edges , 2 )
    result [ s1 ] [ t1 ] [ k1 ] , result [ s2 ] [ t2 ] [ k2 ] = result [ s2 ] [ t2 ] [ k2 ] , result [ s1 ] [ t1 ] [ k1 ]
return result


signature:
def is_edge_consistent ( graph , u , v ) :
    """Check if all edges between two nodes have the same relation .
 : param pybel . BELGraph graph : A BEL Graph
 : param tuple u : The source BEL node
 : param tuple v : The target BEL node
 : return : If all edges from the source to target node have the same relation
 : rtype : bool"""

implementation:
if not graph . has_edge ( u , v ) :
    raise ValueError ( '{} does not contain an edge ({}, {})' . format ( graph , u , v ) )
return 0 == len ( set ( d [ RELATION ] for d in graph . edge [ u ] [ v ] . values ( ) ) )


signature:
def all_edges_consistent ( graph ) :
    """Return if all edges are consistent in a graph . Wraps : func : ` pybel _ tools . utils . is _ edge _ consistent ` .
 : param pybel . BELGraph graph : A BEL graph
 : return : Are all edges consistent
 : rtype : bool"""

implementation: return all ( is_edge_consistent ( graph , u , v ) for u , v in graph . edges ( ) )


signature:
def rewire_targets ( graph , rewiring_probability ) :
    """Rewire a graph ' s edges ' target nodes .
 - For BEL graphs , assumes edge consistency ( all edges between two given nodes are have the same relation )
 - Doesn ' t make self - edges
 : param pybel . BELGraph graph : A BEL graph
 : param float rewiring _ probability : The probability of rewiring ( between 0 and 1)
 : return : A rewired BEL graph"""

implementation:
if not all_edges_consistent ( graph ) :
    raise ValueError ( '{} is not consistent' . format ( graph ) )
result = graph . copy ( )
nodes = result . nodes ( )
for u , v in result . edges ( ) :
    if random . random ( ) < rewiring_probability :
        continue
    w = random . choice ( nodes )
    while w == u or result . has_edge ( u , w ) :
        w = random . choice ( nodes )
    result . add_edge ( w , v )
    result . remove_edge ( u , v )
return result


signature:
def self_edge_filter ( _ : BELGraph , source : BaseEntity , target : BaseEntity , __ : str ) -> bool :
    """Check if the source and target nodes are the same ."""

implementation: return source == target


signature:
def has_protein_modification_increases_activity ( graph : BELGraph , source : BaseEntity , target : BaseEntity , key : str , ) -> bool :
    """Check if pmod of source causes activity of target ."""

implementation:
edge_data = graph [ source ] [ target ] [ key ]
return has_protein_modification ( graph , source ) and part_has_modifier ( edge_data , OBJECT , ACTIVITY )


signature:
def has_degradation_increases_activity ( data : Dict ) -> bool :
    """Check if the degradation of source causes activity of target ."""

implementation: return part_has_modifier ( data , SUBJECT , DEGRADATION ) and part_has_modifier ( data , OBJECT , ACTIVITY )


signature:
def has_translocation_increases_activity ( data : Dict ) -> bool :
    """Check if the translocation of source causes activity of target ."""

implementation: return part_has_modifier ( data , SUBJECT , TRANSLOCATION ) and part_has_modifier ( data , OBJECT , ACTIVITY )


signature:
def complex_has_member ( graph : BELGraph , complex_node : ComplexAbundance , member_node : BaseEntity ) -> bool :
    """Does the given complex contain the member ?"""

implementation:
return any ( # TODO can ' t you look in the members of the complex object ( if it ' s enumerated )
v == member_node for _ , v , data in graph . out_edges ( complex_node , data = True ) if data [ RELATION ] == HAS_COMPONENT )


signature:
def complex_increases_activity ( graph : BELGraph , u : BaseEntity , v : BaseEntity , key : str ) -> bool :
    """Return if the formation of a complex with u increases the activity of v ."""

implementation: return ( isinstance ( u , ( ComplexAbundance , NamedComplexAbundance ) ) and complex_has_member ( graph , u , v ) and part_has_modifier ( graph [ u ] [ v ] [ key ] , OBJECT , ACTIVITY ) )


signature:
def find_activations ( graph : BELGraph ) :
    """Find edges that are A - A , meaning that some conditions in the edge best describe the interaction ."""

implementation:
for u , v , key , data in graph . edges ( keys = True , data = True ) :
    if u != v :
        continue
    bel = graph . edge_to_bel ( u , v , data )
    line = data . get ( LINE )
    if line is None :
        continue
    # this was inferred , so need to investigate another way
    elif has_protein_modification_increases_activity ( graph , u , v , key ) :
        print ( line , '- pmod changes -' , bel )
        find_related ( graph , v , data )
    elif has_degradation_increases_activity ( data ) :
        print ( line , '- degradation changes -' , bel )
        find_related ( graph , v , data )
    elif has_translocation_increases_activity ( data ) :
        print ( line , '- translocation changes -' , bel )
        find_related ( graph , v , data )
    elif complex_increases_activity ( graph , u , v , key ) :
        print ( line , '- complex changes - ' , bel )
        find_related ( graph , v , data )
    elif has_same_subject_object ( graph , u , v , key ) :
        print ( line , '- same sub/obj -' , bel )
    else :
        print ( line , '- *** - ' , bel )


implementation:
a , b = itt . tee ( iterable )
next ( b , None )
return zip ( a , b )


signature:
def rank_path ( graph , path , edge_ranking = None ) :
    """Takes in a path ( a list of nodes in the graph ) and calculates a score
 : param pybel . BELGraph graph : A BEL graph
 : param list [ tuple ] path : A list of nodes in the path ( includes terminal nodes )
 : param dict edge _ ranking : A dictionary of { relationship : score }
 : return : The score for the edge
 : rtype : int"""

implementation:
edge_ranking = default_edge_ranking if edge_ranking is None else edge_ranking
return sum ( max ( edge_ranking [ d [ RELATION ] ] for d in graph . edge [ u ] [ v ] . values ( ) ) for u , v in pairwise ( path ) )


signature:
def find_root_in_path ( graph , path_nodes ) :
    """Find the ' root ' of the path - > The node with the lowest out degree , if multiple :
 root is the one with the highest out degree among those with lowest out degree
 : param pybel . BELGraph graph : A BEL Graph
 : param list [ tuple ] path _ nodes : A list of nodes in their order in a path
 : return : A pair of the graph : graph of the path and the root node
 : rtype : tuple [ pybel . BELGraph , tuple ]"""

implementation:
path_graph = graph . subgraph ( path_nodes )
# node _ in _ degree _ tuple : list of tuples with ( node , in _ degree _ of _ node ) in ascending order
node_in_degree_tuple = sorted ( [ ( n , d ) for n , d in path_graph . in_degree ( ) . items ( ) ] , key = itemgetter ( 1 ) )
# node _ out _ degree _ tuple : ordered list of tuples with ( node , in _ degree _ of _ node ) in descending order
node_out_degree_tuple = sorted ( [ ( n , d ) for n , d in path_graph . out_degree ( ) . items ( ) ] , key = itemgetter ( 1 ) , reverse = True )
# In case all have the same in degree it needs to be reference before
tied_root_index = 0
# Get index where the min in _ degree stops ( in case they are duplicates )
for i in range ( 0 , ( len ( node_in_degree_tuple ) - 1 ) ) :
    if node_in_degree_tuple [ i ] [ 1 ] < node_in_degree_tuple [ i + 1 ] [ 1 ] :
        tied_root_index = i
        break
# If there are multiple nodes with minimum in _ degree take the one with max out degree
# ( in case multiple have the same out degree pick one random )
if tied_root_index != 0 :
    root_tuple = max ( node_out_degree_tuple [ : tied_root_index ] , key = itemgetter ( 1 ) )
else :
    root_tuple = node_in_degree_tuple [ 0 ]
return path_graph , root_tuple [ 0 ]


signature:
def summarize_edge_filter ( graph : BELGraph , edge_predicates : EdgePredicates ) -> None :
    """Print a summary of the number of edges passing a given set of filters ."""

implementation:
passed = count_passed_edge_filter ( graph , edge_predicates )
print ( '{}/{} edges passed {}' . format ( passed , graph . number_of_edges ( ) , ( ', ' . join ( edge_filter . __name__ for edge_filter in edge_predicates ) if isinstance ( edge_predicates , Iterable ) else edge_predicates . __name__ ) ) )


signature:
def node_has_namespace ( node : BaseEntity , namespace : str ) -> bool :
    """Pass for nodes that have the given namespace ."""

implementation:
ns = node . get ( NAMESPACE )
return ns is not None and ns == namespace


signature:
def node_has_namespaces ( node : BaseEntity , namespaces : Set [ str ] ) -> bool :
    """Pass for nodes that have one of the given namespaces ."""

implementation:
ns = node . get ( NAMESPACE )
return ns is not None and ns in namespaces


signature:
def build_source_namespace_filter ( namespaces : Strings ) -> EdgePredicate :
    """Pass for edges whose source nodes have the given namespace or one of the given namespaces .
 : param namespaces : The namespace or namespaces to filter by"""

implementation:
if isinstance ( namespaces , str ) :
    def source_namespace_filter ( _ , u : BaseEntity , __ , ___ ) -> bool :
        return node_has_namespace ( u , namespaces )
elif isinstance ( namespaces , Iterable ) :
    namespaces = set ( namespaces )
    def source_namespace_filter ( _ , u : BaseEntity , __ , ___ ) -> bool :
        return node_has_namespaces ( u , namespaces )
else :
    raise TypeError
return source_namespace_filter


signature:
def build_target_namespace_filter ( namespaces : Strings ) -> EdgePredicate :
    """Only passes for edges whose target nodes have the given namespace or one of the given namespaces
 : param namespaces : The namespace or namespaces to filter by"""

implementation:
if isinstance ( namespaces , str ) :
    def target_namespace_filter ( _ , __ , v : BaseEntity , ___ ) -> bool :
        return node_has_namespace ( v , namespaces )
elif isinstance ( namespaces , Iterable ) :
    namespaces = set ( namespaces )
    def target_namespace_filter ( _ , __ , v : BaseEntity , ___ ) -> bool :
        return node_has_namespaces ( v , namespaces )
else :
    raise TypeError
return target_namespace_filter


signature:
def search_node_namespace_names ( graph , query , namespace ) :
    """Search for nodes with the given namespace ( s ) and whose names containing a given string ( s ) .
 : param pybel . BELGraph graph : A BEL graph
 : param query : The search query
 : type query : str or iter [ str ]
 : param namespace : The namespace ( s ) to filter
 : type namespace : str or iter [ str ]
 : return : An iterator over nodes whose names match the search query
 : rtype : iter"""

implementation:
node_predicates = [ namespace_inclusion_builder ( namespace ) , build_node_name_search ( query ) ]
return filter_nodes ( graph , node_predicates )


signature:
def get_cutoff ( value : float , cutoff : Optional [ float ] = None ) -> int :
    """Assign if a value is greater than or less than a cutoff ."""

implementation:
cutoff = cutoff if cutoff is not None else 0
if value > cutoff :
    return 1
if value < ( - 1 * cutoff ) :
    return - 1
return 0


signature:
def calculate_concordance_helper ( graph : BELGraph , key : str , cutoff : Optional [ float ] = None , ) -> Tuple [ int , int , int , int ] :
    """Help calculate network - wide concordance
 Assumes data already annotated with given key
 : param graph : A BEL graph
 : param key : The node data dictionary key storing the logFC
 : param cutoff : The optional logFC cutoff for significance"""

implementation:
scores = defaultdict ( int )
for u , v , k , d in graph . edges ( keys = True , data = True ) :
    c = edge_concords ( graph , u , v , k , d , key , cutoff = cutoff )
    scores [ c ] += 1
return ( scores [ Concordance . correct ] , scores [ Concordance . incorrect ] , scores [ Concordance . ambiguous ] , scores [ Concordance . unassigned ] , )


signature:
def calculate_concordance ( graph : BELGraph , key : str , cutoff : Optional [ float ] = None , use_ambiguous : bool = False ) -> float :
    """Calculates network - wide concordance .
 Assumes data already annotated with given key
 : param graph : A BEL graph
 : param key : The node data dictionary key storing the logFC
 : param cutoff : The optional logFC cutoff for significance
 : param use _ ambiguous : Compare to ambiguous edges as well"""

implementation:
correct , incorrect , ambiguous , _ = calculate_concordance_helper ( graph , key , cutoff = cutoff )
try :
    return correct / ( correct + incorrect + ( ambiguous if use_ambiguous else 0 ) )
except ZeroDivisionError :
    return - 1.0


signature:
def one_sided ( value : float , distribution : List [ float ] ) -> float :
    """Calculate the one - sided probability of getting a value more extreme than the distribution ."""

implementation:
assert distribution
return sum ( value < element for element in distribution ) / len ( distribution )


signature:
def calculate_concordance_probability ( graph : BELGraph , key : str , cutoff : Optional [ float ] = None , permutations : Optional [ int ] = None , percentage : Optional [ float ] = None , use_ambiguous : bool = False , permute_type : str = 'shuffle_node_data' , ) -> Tuple [ float , List [ float ] , float ] :
    """Calculates a graph ' s concordance as well as its statistical probability .
 : param graph : A BEL graph
 : param str key : The node data dictionary key storing the logFC
 : param float cutoff : The optional logFC cutoff for significance
 : param int permutations : The number of random permutations to test . Defaults to 500
 : param float percentage : The percentage of the graph ' s edges to maintain . Defaults to 0.9
 : param bool use _ ambiguous : Compare to ambiguous edges as well
 : returns : A triple of the concordance score , the null distribution , and the p - value ."""

implementation:
if permute_type == 'random_by_edges' :
    permute_func = partial ( random_by_edges , percentage = percentage )
elif permute_type == 'shuffle_node_data' :
    permute_func = partial ( shuffle_node_data , key = key , percentage = percentage )
elif permute_type == 'shuffle_relations' :
    permute_func = partial ( shuffle_relations , percentage = percentage )
else :
    raise ValueError ( 'Invalid permute_type: {}' . format ( permute_type ) )
graph : BELGraph = graph . copy ( )
collapse_to_genes ( graph )
collapse_all_variants ( graph )
score = calculate_concordance ( graph , key , cutoff = cutoff )
distribution = [ ]
for _ in range ( permutations or 500 ) :
    permuted_graph = permute_func ( graph )
    permuted_graph_scores = calculate_concordance ( permuted_graph , key , cutoff = cutoff , use_ambiguous = use_ambiguous )
    distribution . append ( permuted_graph_scores )
return score , distribution , one_sided ( score , distribution )


signature:
def calculate_concordance_by_annotation ( graph , annotation , key , cutoff = None ) :
    """Returns the concordance scores for each stratified graph based on the given annotation
 : param pybel . BELGraph graph : A BEL graph
 : param str annotation : The annotation to group by .
 : param str key : The node data dictionary key storing the logFC
 : param float cutoff : The optional logFC cutoff for significance
 : rtype : dict [ str , tuple ]"""

implementation: return { value : calculate_concordance ( subgraph , key , cutoff = cutoff ) for value , subgraph in get_subgraphs_by_annotation ( graph , annotation ) . items ( ) }


signature:
def calculate_concordance_probability_by_annotation ( graph , annotation , key , cutoff = None , permutations = None , percentage = None , use_ambiguous = False ) :
    """Returns the results of concordance analysis on each subgraph , stratified by the given annotation .
 : param pybel . BELGraph graph : A BEL graph
 : param str annotation : The annotation to group by .
 : param str key : The node data dictionary key storing the logFC
 : param float cutoff : The optional logFC cutoff for significance
 : param int permutations : The number of random permutations to test . Defaults to 500
 : param float percentage : The percentage of the graph ' s edges to maintain . Defaults to 0.9
 : param bool use _ ambiguous : Compare to ambiguous edges as well
 : rtype : dict [ str , tuple ]"""

implementation:
result = [ ( value , calculate_concordance_probability ( subgraph , key , cutoff = cutoff , permutations = permutations , percentage = percentage , use_ambiguous = use_ambiguous , ) ) for value , subgraph in get_subgraphs_by_annotation ( graph , annotation ) . items ( ) ]
return dict ( result )


signature:
def _get_drug_target_interactions ( manager : Optional [ 'bio2bel_drugbank.manager' ] = None ) -> Mapping [ str , List [ str ] ] :
    """Get a mapping from drugs to their list of gene ."""

implementation:
if manager is None :
    import bio2bel_drugbank
    manager = bio2bel_drugbank . Manager ( )
if not manager . is_populated ( ) :
    manager . populate ( )
return manager . get_drug_to_hgnc_symbols ( )


signature:
def multi_run_epicom ( graphs : Iterable [ BELGraph ] , path : Union [ None , str , TextIO ] ) -> None :
    """Run EpiCom analysis on many graphs ."""

implementation:
if isinstance ( path , str ) :
    with open ( path , 'w' ) as file :
        _multi_run_helper_file_wrapper ( graphs , file )
else :
    _multi_run_helper_file_wrapper ( graphs , path )


signature:
def main ( ) :
    """Convert the Alzheimer ' s and Parkinson ' s disease NeuroMMSig excel sheets to BEL ."""

implementation:
logging . basicConfig ( level = logging . INFO )
log . setLevel ( logging . INFO )
bms_base = get_bms_base ( )
neurommsig_base = get_neurommsig_base ( )
neurommsig_excel_dir = os . path . join ( neurommsig_base , 'resources' , 'excels' , 'neurommsig' )
nift_values = get_nift_values ( )
log . info ( 'Starting Alzheimers' )
ad_path = os . path . join ( neurommsig_excel_dir , 'alzheimers' , 'alzheimers.xlsx' )
ad_df = preprocess ( ad_path )
with open ( os . path . join ( bms_base , 'aetionomy' , 'alzheimers' , 'neurommsigdb_ad.bel' ) , 'w' ) as ad_file :
    write_neurommsig_bel ( ad_file , ad_df , mesh_alzheimer , nift_values )
log . info ( 'Starting Parkinsons' )
pd_path = os . path . join ( neurommsig_excel_dir , 'parkinsons' , 'parkinsons.xlsx' )
pd_df = preprocess ( pd_path )
with open ( os . path . join ( bms_base , 'aetionomy' , 'parkinsons' , 'neurommsigdb_pd.bel' ) , 'w' ) as pd_file :
    write_neurommsig_bel ( pd_file , pd_df , mesh_parkinson , nift_values )


signature:
def remove_inconsistent_edges ( graph : BELGraph ) -> None :
    """Remove all edges between node pairs with inconsistent edges .
 This is the all - or - nothing approach . It would be better to do more careful investigation of the evidences during
 curation ."""

implementation:
for u , v in get_inconsistent_edges ( graph ) :
    edges = [ ( u , v , k ) for k in graph [ u ] [ v ] ]
    graph . remove_edges_from ( edges )


signature:
def get_walks_exhaustive ( graph , node , length ) :
    """Gets all walks under a given length starting at a given node
 : param networkx . Graph graph : A graph
 : param node : Starting node
 : param int length : The length of walks to get
 : return : A list of paths
 : rtype : list [ tuple ]"""

implementation:
if 0 == length :
    return ( node , ) ,
return tuple ( ( node , key ) + path for neighbor in graph . edge [ node ] for path in get_walks_exhaustive ( graph , neighbor , length - 1 ) if node not in path for key in graph . edge [ node ] [ neighbor ] )


signature:
def match_simple_metapath ( graph , node , simple_metapath ) :
    """Matches a simple metapath starting at the given node
 : param pybel . BELGraph graph : A BEL graph
 : param tuple node : A BEL node
 : param list [ str ] simple _ metapath : A list of BEL Functions
 : return : An iterable over paths from the node matching the metapath
 : rtype : iter [ tuple ]"""

implementation:
if 0 == len ( simple_metapath ) :
    yield node ,
else :
    for neighbor in graph . edges [ node ] :
        if graph . nodes [ neighbor ] [ FUNCTION ] == simple_metapath [ 0 ] :
            for path in match_simple_metapath ( graph , neighbor , simple_metapath [ 1 : ] ) :
                if node not in path :
                    yield ( node , ) + path


signature:
def build_database ( manager : pybel . Manager , annotation_url : Optional [ str ] = None ) -> None :
    """Build a database of scores for NeuroMMSig annotated graphs .
 1 . Get all networks that use the Subgraph annotation
 2 . run on each"""

implementation:
annotation_url = annotation_url or NEUROMMSIG_DEFAULT_URL
annotation = manager . get_namespace_by_url ( annotation_url )
if annotation is None :
    raise RuntimeError ( 'no graphs in database with given annotation' )
networks = get_networks_using_annotation ( manager , annotation )
dtis = ...
for network in networks :
    graph = network . as_bel ( )
    scores = epicom_on_graph ( graph , dtis )
    for ( drug_name , subgraph_name ) , score in scores . items ( ) :
        drug_model = get_drug_model ( manager , drug_name )
        subgraph_model = manager . get_annotation_entry ( annotation_url , subgraph_name )
        score_model = Score ( network = network , annotation = subgraph_model , drug = drug_model , score = score )
        manager . session . add ( score_model )
t = time . time ( )
logger . info ( 'committing scores' )
manager . session . commit ( )
logger . info ( 'committed scores in %.2f seconds' , time . time ( ) - t )


signature:
def calculate_average_scores_on_graph ( graph : BELGraph , key : Optional [ str ] = None , tag : Optional [ str ] = None , default_score : Optional [ float ] = None , runs : Optional [ int ] = None , use_tqdm : bool = False , ) :
    """Calculate the scores over all biological processes in the sub - graph .
 As an implementation , it simply computes the sub - graphs then calls : func : ` calculate _ average _ scores _ on _ subgraphs ` as
 described in that function ' s documentation .
 : param graph : A BEL graph with heats already on the nodes
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 : param tag : The key for the nodes ' data dictionaries where the scores will be put . Defaults to ' score '
 : param default _ score : The initial score for all nodes . This number can go up or down .
 : param runs : The number of times to run the heat diffusion workflow . Defaults to 100.
 : param use _ tqdm : Should there be a progress bar for runners ?
 : return : A dictionary of { pybel node tuple : results tuple }
 : rtype : dict [ tuple , tuple ]
 Suggested usage with : mod : ` pandas ` :
 > > > import pandas as pd
 > > > from pybel _ tools . analysis . heat import calculate _ average _ scores _ on _ graph
 > > > graph = . . . # load graph and data
 > > > scores = calculate _ average _ scores _ on _ graph ( graph )
 > > > pd . DataFrame . from _ items ( scores . items ( ) , orient = ' index ' , columns = RESULT _ LABELS )"""

implementation:
subgraphs = generate_bioprocess_mechanisms ( graph , key = key )
scores = calculate_average_scores_on_subgraphs ( subgraphs , key = key , tag = tag , default_score = default_score , runs = runs , use_tqdm = use_tqdm )
return scores


signature:
def calculate_average_scores_on_subgraphs ( subgraphs : Mapping [ H , BELGraph ] , key : Optional [ str ] = None , tag : Optional [ str ] = None , default_score : Optional [ float ] = None , runs : Optional [ int ] = None , use_tqdm : bool = False , tqdm_kwargs : Optional [ Mapping [ str , Any ] ] = None , ) -> Mapping [ H , Tuple [ float , float , float , float , int , int ] ] :
    """Calculate the scores over precomputed candidate mechanisms .
 : param subgraphs : A dictionary of keys to their corresponding subgraphs
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 : param tag : The key for the nodes ' data dictionaries where the scores will be put . Defaults to ' score '
 : param default _ score : The initial score for all nodes . This number can go up or down .
 : param runs : The number of times to run the heat diffusion workflow . Defaults to 100.
 : param use _ tqdm : Should there be a progress bar for runners ?
 : return : A dictionary of keys to results tuples
 Example Usage :
 > > > import pandas as pd
 > > > from pybel _ tools . generation import generate _ bioprocess _ mechanisms
 > > > from pybel _ tools . analysis . heat import calculate _ average _ scores _ on _ subgraphs
 > > > # load graph and data
 > > > graph = . . .
 > > > candidate _ mechanisms = generate _ bioprocess _ mechanisms ( graph )
 > > > scores = calculate _ average _ scores _ on _ subgraphs ( candidate _ mechanisms )
 > > > pd . DataFrame . from _ items ( scores . items ( ) , orient = ' index ' , columns = RESULT _ LABELS )"""

implementation:
results = { }
log . info ( 'calculating results for %d candidate mechanisms using %d permutations' , len ( subgraphs ) , runs )
it = subgraphs . items ( )
if use_tqdm :
    _tqdm_kwargs = dict ( total = len ( subgraphs ) , desc = 'Candidate mechanisms' )
    if tqdm_kwargs :
        _tqdm_kwargs . update ( tqdm_kwargs )
    it = tqdm ( it , ** _tqdm_kwargs )
for node , subgraph in it :
    number_first_neighbors = subgraph . in_degree ( node )
    number_first_neighbors = 0 if isinstance ( number_first_neighbors , dict ) else number_first_neighbors
    mechanism_size = subgraph . number_of_nodes ( )
    runners = workflow ( subgraph , node , key = key , tag = tag , default_score = default_score , runs = runs )
    scores = [ runner . get_final_score ( ) for runner in runners ]
    if 0 == len ( scores ) :
        results [ node ] = ( None , None , None , None , number_first_neighbors , mechanism_size , )
        continue
    scores = np . array ( scores )
    average_score = np . average ( scores )
    score_std = np . std ( scores )
    med_score = np . median ( scores )
    chi_2_stat , norm_p = stats . normaltest ( scores )
    results [ node ] = ( average_score , score_std , norm_p , med_score , number_first_neighbors , mechanism_size , )
return results


signature:
def workflow ( graph : BELGraph , node : BaseEntity , key : Optional [ str ] = None , tag : Optional [ str ] = None , default_score : Optional [ float ] = None , runs : Optional [ int ] = None , minimum_nodes : int = 1 , ) -> List [ 'Runner' ] :
    """Generate candidate mechanisms and run the heat diffusion workflow .
 : param graph : A BEL graph
 : param node : The BEL node that is the focus of this analysis
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 : param tag : The key for the nodes ' data dictionaries where the scores will be put . Defaults to ' score '
 : param default _ score : The initial score for all nodes . This number can go up or down .
 : param runs : The number of times to run the heat diffusion workflow . Defaults to 100.
 : param minimum _ nodes : The minimum number of nodes a sub - graph needs to try running heat diffusion
 : return : A list of runners"""

implementation:
subgraph = generate_mechanism ( graph , node , key = key )
if subgraph . number_of_nodes ( ) <= minimum_nodes :
    return [ ]
runners = multirun ( subgraph , node , key = key , tag = tag , default_score = default_score , runs = runs )
return list ( runners )


signature:
def multirun ( graph : BELGraph , node : BaseEntity , key : Optional [ str ] = None , tag : Optional [ str ] = None , default_score : Optional [ float ] = None , runs : Optional [ int ] = None , use_tqdm : bool = False , ) -> Iterable [ 'Runner' ] :
    """Run the heat diffusion workflow multiple times , each time yielding a : class : ` Runner ` object upon completion .
 : param graph : A BEL graph
 : param node : The BEL node that is the focus of this analysis
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 : param tag : The key for the nodes ' data dictionaries where the scores will be put . Defaults to ' score '
 : param default _ score : The initial score for all nodes . This number can go up or down .
 : param runs : The number of times to run the heat diffusion workflow . Defaults to 100.
 : param use _ tqdm : Should there be a progress bar for runners ?
 : return : An iterable over the runners after each iteration"""

implementation:
if runs is None :
    runs = 100
it = range ( runs )
if use_tqdm :
    it = tqdm ( it , total = runs )
for i in it :
    try :
        runner = Runner ( graph , node , key = key , tag = tag , default_score = default_score )
        runner . run ( )
        yield runner
    except Exception :
        log . debug ( 'Run %s failed for %s' , i , node )


signature:
def workflow_aggregate ( graph : BELGraph , node : BaseEntity , key : Optional [ str ] = None , tag : Optional [ str ] = None , default_score : Optional [ float ] = None , runs : Optional [ int ] = None , aggregator : Optional [ Callable [ [ Iterable [ float ] ] , float ] ] = None , ) -> Optional [ float ] :
    """Get the average score over multiple runs .
 This function is very simple , and can be copied to do more interesting statistics over the : class : ` Runner `
 instances . To iterate over the runners themselves , see : func : ` workflow `
 : param graph : A BEL graph
 : param node : The BEL node that is the focus of this analysis
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 : param tag : The key for the nodes ' data dictionaries where the scores will be put . Defaults to ' score '
 : param default _ score : The initial score for all nodes . This number can go up or down .
 : param runs : The number of times to run the heat diffusion workflow . Defaults to 100.
 : param aggregator : A function that aggregates a list of scores . Defaults to : func : ` numpy . average ` .
 Could also use : : func : ` numpy . mean ` , : func : ` numpy . median ` , : func : ` numpy . min ` , : func : ` numpy . max `
 : return : The average score for the target node"""

implementation:
runners = workflow ( graph , node , key = key , tag = tag , default_score = default_score , runs = runs )
scores = [ runner . get_final_score ( ) for runner in runners ]
if not scores :
    log . warning ( 'Unable to run the heat diffusion workflow for %s' , node )
    return
if aggregator is None :
    return np . average ( scores )
return aggregator ( scores )


signature:
def workflow_all ( graph : BELGraph , key : Optional [ str ] = None , tag : Optional [ str ] = None , default_score : Optional [ float ] = None , runs : Optional [ int ] = None , ) -> Mapping [ BaseEntity , List [ Runner ] ] :
    """Run the heat diffusion workflow and get runners for every possible candidate mechanism
 1 . Get all biological processes
 2 . Get candidate mechanism induced two level back from each biological process
 3 . Heat diffusion workflow for each candidate mechanism for multiple runs
 4 . Return all runner results
 : param graph : A BEL graph
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 : param tag : The key for the nodes ' data dictionaries where the scores will be put . Defaults to ' score '
 : param default _ score : The initial score for all nodes . This number can go up or down .
 : param runs : The number of times to run the heat diffusion workflow . Defaults to 100.
 : return : A dictionary of { node : list of runners }"""

implementation:
results = { }
for node in get_nodes_by_function ( graph , BIOPROCESS ) :
    results [ node ] = workflow ( graph , node , key = key , tag = tag , default_score = default_score , runs = runs )
return results


signature:
def workflow_all_aggregate ( graph : BELGraph , key : Optional [ str ] = None , tag : Optional [ str ] = None , default_score : Optional [ float ] = None , runs : Optional [ int ] = None , aggregator : Optional [ Callable [ [ Iterable [ float ] ] , float ] ] = None , ) :
    """Run the heat diffusion workflow to get average score for every possible candidate mechanism .
 1 . Get all biological processes
 2 . Get candidate mechanism induced two level back from each biological process
 3 . Heat diffusion workflow on each candidate mechanism for multiple runs
 4 . Report average scores for each candidate mechanism
 : param graph : A BEL graph
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 : param tag : The key for the nodes ' data dictionaries where the scores will be put . Defaults to ' score '
 : param default _ score : The initial score for all nodes . This number can go up or down .
 : param runs : The number of times to run the heat diffusion workflow . Defaults to 100.
 : param aggregator : A function that aggregates a list of scores . Defaults to : func : ` numpy . average ` .
 Could also use : : func : ` numpy . mean ` , : func : ` numpy . median ` , : func : ` numpy . min ` , : func : ` numpy . max `
 : return : A dictionary of { node : upstream causal subgraph }"""

implementation:
results = { }
bioprocess_nodes = list ( get_nodes_by_function ( graph , BIOPROCESS ) )
for bioprocess_node in tqdm ( bioprocess_nodes ) :
    subgraph = generate_mechanism ( graph , bioprocess_node , key = key )
    try :
        results [ bioprocess_node ] = workflow_aggregate ( graph = subgraph , node = bioprocess_node , key = key , tag = tag , default_score = default_score , runs = runs , aggregator = aggregator )
    except Exception :
        log . exception ( 'could not run on %' , bioprocess_node )
return results


signature:
def calculate_average_score_by_annotation ( graph : BELGraph , annotation : str , key : Optional [ str ] = None , runs : Optional [ int ] = None , use_tqdm : bool = False , ) -> Mapping [ str , float ] :
    """For each sub - graph induced over the edges matching the annotation , calculate the average score
 for all of the contained biological processes
 Assumes you haven ' t done anything yet
 1 . Generates biological process upstream candidate mechanistic sub - graphs with
 : func : ` generate _ bioprocess _ mechanisms `
 2 . Calculates scores for each sub - graph with : func : ` calculate _ average _ scores _ on _ sub - graphs `
 3 . Overlays data with pbt . integration . overlay _ data
 4 . Calculates averages with pbt . selection . group _ nodes . average _ node _ annotation
 : param graph : A BEL graph
 : param annotation : A BEL annotation
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 : param runs : The number of times to run the heat diffusion workflow . Defaults to 100.
 : param use _ tqdm : Should there be a progress bar for runners ?
 : return : A dictionary from { str annotation value : tuple scores }
 Example Usage :
 > > > import pybel
 > > > from pybel _ tools . integration import overlay _ data
 > > > from pybel _ tools . analysis . heat import calculate _ average _ score _ by _ annotation
 > > > graph = pybel . from _ path ( . . . )
 > > > scores = calculate _ average _ score _ by _ annotation ( graph , ' subgraph ' )"""

implementation:
candidate_mechanisms = generate_bioprocess_mechanisms ( graph , key = key )
# : { bp tuple : list of scores }
scores : Mapping [ BaseEntity , Tuple ] = calculate_average_scores_on_subgraphs ( subgraphs = candidate_mechanisms , key = key , runs = runs , use_tqdm = use_tqdm , )
subgraph_bp : Mapping [ str , List [ BaseEntity ] ] = defaultdict ( list )
subgraphs : Mapping [ str , BELGraph ] = get_subgraphs_by_annotation ( graph , annotation )
for annotation_value , subgraph in subgraphs . items ( ) :
    subgraph_bp [ annotation_value ] . extend ( get_nodes_by_function ( subgraph , BIOPROCESS ) )
# : Pick the average by slicing with 0 . Refer to : func : ` calculate _ average _ score _ on _ subgraphs `
return { annotation_value : np . average ( scores [ bp ] [ 0 ] for bp in bps ) for annotation_value , bps in subgraph_bp . items ( ) }


signature:
def iter_leaves ( self ) -> Iterable [ BaseEntity ] :
    """Return an iterable over all nodes that are leaves .
 A node is a leaf if either :
 - it doesn ' t have any predecessors , OR
 - all of its predecessors have a score in their data dictionaries"""

implementation:
for node in self . graph :
    if self . tag in self . graph . nodes [ node ] :
        continue
    if not any ( self . tag not in self . graph . nodes [ p ] for p in self . graph . predecessors ( node ) ) :
        yield node


signature:
def in_out_ratio ( self , node : BaseEntity ) -> float :
    """Calculate the ratio of in - degree / out - degree of a node ."""

implementation: return self . graph . in_degree ( node ) / float ( self . graph . out_degree ( node ) )


signature:
def unscored_nodes_iter ( self ) -> BaseEntity :
    """Iterate over all nodes without a score ."""

implementation:
for node , data in self . graph . nodes ( data = True ) :
    if self . tag not in data :
        yield node


signature:
def get_random_edge ( self ) :
    """This function should be run when there are no leaves , but there are still unscored nodes . It will introduce
 a probabilistic element to the algorithm , where some edges are disregarded randomly to eventually get a score
 for the network . This means that the score can be averaged over many runs for a given graph , and a better
 data structure will have to be later developed that doesn ' t destroy the graph ( instead , annotates which edges
 have been disregarded , later )
 1 . get all un - scored
 2 . rank by in - degree
 3 . weighted probability over all in - edges where lower in - degree means higher probability
 4 . pick randomly which edge
 : return : A random in - edge to the lowest in / out degree ratio node . This is a 3 - tuple of ( node , node , key )
 : rtype : tuple"""

implementation:
nodes = [ ( n , self . in_out_ratio ( n ) ) for n in self . unscored_nodes_iter ( ) if n != self . target_node ]
node , deg = min ( nodes , key = itemgetter ( 1 ) )
log . log ( 5 , 'checking %s (in/out ratio: %.3f)' , node , deg )
possible_edges = self . graph . in_edges ( node , keys = True )
log . log ( 5 , 'possible edges: %s' , possible_edges )
edge_to_remove = random . choice ( possible_edges )
log . log ( 5 , 'chose: %s' , edge_to_remove )
return edge_to_remove


signature:
def remove_random_edge ( self ) :
    """Remove a random in - edge from the node with the lowest in / out degree ratio ."""

implementation:
u , v , k = self . get_random_edge ( )
log . log ( 5 , 'removing %s, %s (%s)' , u , v , k )
self . graph . remove_edge ( u , v , k )


signature:
def remove_random_edge_until_has_leaves ( self ) -> None :
    """Remove random edges until there is at least one leaf node ."""

implementation:
while True :
    leaves = set ( self . iter_leaves ( ) )
    if leaves :
        return
    self . remove_random_edge ( )


signature:
def score_leaves ( self ) -> Set [ BaseEntity ] :
    """Calculate the score for all leaves .
 : return : The set of leaf nodes that were scored"""

implementation:
leaves = set ( self . iter_leaves ( ) )
if not leaves :
    log . warning ( 'no leaves.' )
    return set ( )
for leaf in leaves :
    self . graph . nodes [ leaf ] [ self . tag ] = self . calculate_score ( leaf )
    log . log ( 5 , 'chomping %s' , leaf )
return leaves


signature:
def run_with_graph_transformation ( self ) -> Iterable [ BELGraph ] :
    """Calculate scores for all leaves until there are none , removes edges until there are , and repeats until
 all nodes have been scored . Also , yields the current graph at every step so you can make a cool animation
 of how the graph changes throughout the course of the algorithm
 : return : An iterable of BEL graphs"""

implementation:
yield self . get_remaining_graph ( )
while not self . done_chomping ( ) :
    while not list ( self . iter_leaves ( ) ) :
        self . remove_random_edge ( )
        yield self . get_remaining_graph ( )
    self . score_leaves ( )
    yield self . get_remaining_graph ( )


signature:
def done_chomping ( self ) -> bool :
    """Determines if the algorithm is complete by checking if the target node of this analysis has been scored
 yet . Because the algorithm removes edges when it gets stuck until it is un - stuck , it is always guaranteed to
 finish .
 : return : Is the algorithm done running ?"""

implementation: return self . tag in self . graph . nodes [ self . target_node ]


signature:
def get_final_score ( self ) -> float :
    """Return the final score for the target node .
 : return : The final score for the target node"""

implementation:
if not self . done_chomping ( ) :
    raise ValueError ( 'algorithm has not yet completed' )
return self . graph . nodes [ self . target_node ] [ self . tag ]


signature:
def calculate_score ( self , node : BaseEntity ) -> float :
    """Calculate the new score of the given node ."""

implementation:
score = ( self . graph . nodes [ node ] [ self . tag ] if self . tag in self . graph . nodes [ node ] else self . default_score )
for predecessor , _ , d in self . graph . in_edges ( node , data = True ) :
    if d [ RELATION ] in CAUSAL_INCREASE_RELATIONS :
        score += self . graph . nodes [ predecessor ] [ self . tag ]
    elif d [ RELATION ] in CAUSAL_DECREASE_RELATIONS :
        score -= self . graph . nodes [ predecessor ] [ self . tag ]
return score


signature:
def microcanonical_statistics_dtype ( spanning_cluster = True ) :
    """Return the numpy structured array data type for sample states
 Helper function
 Parameters
 spanning _ cluster : bool , optional
 Whether to detect a spanning cluster or not .
 Defaults to ` ` True ` ` .
 Returns
 ret : list of pairs of str
 A list of tuples of field names and data types to be used as ` ` dtype ` `
 argument in numpy ndarray constructors
 See Also
 http : / / docs . scipy . org / doc / numpy / user / basics . rec . html
 canonical _ statistics _ dtype"""

implementation:
fields = list ( )
fields . extend ( [ ( 'n' , 'uint32' ) , ( 'edge' , 'uint32' ) , ] )
if spanning_cluster :
    fields . extend ( [ ( 'has_spanning_cluster' , 'bool' ) , ] )
fields . extend ( [ ( 'max_cluster_size' , 'uint32' ) , ( 'moments' , '(5,)uint64' ) , ] )
return _ndarray_dtype ( fields )


signature:
def bond_sample_states ( perc_graph , num_nodes , num_edges , seed , spanning_cluster = True , auxiliary_node_attributes = None , auxiliary_edge_attributes = None , spanning_sides = None , ** kwargs ) :
    '''Generate successive sample states of the bond percolation model
 This is a : ref : ` generator function < python : tut - generators > ` to successively
 add one edge at a time from the graph to the percolation model .
 At each iteration , it calculates and returns the cluster statistics .
 CAUTION : it returns a reference to the internal array , not a copy .
 Parameters
 perc _ graph : networkx . Graph
 The substrate graph on which percolation is to take place
 num _ nodes : int
 Number ` ` N ` ` of sites in the graph
 num _ edges : int
 Number ` ` M ` ` of bonds in the graph
 seed : { None , int , array _ like }
 Random seed initializing the pseudo - random number generator .
 Piped through to ` numpy . random . RandomState ` constructor .
 spanning _ cluster : bool , optional
 Whether to detect a spanning cluster or not .
 Defaults to ` ` True ` ` .
 auxiliary _ node _ attributes : optional
 Return value of ` ` networkx . get _ node _ attributes ( graph , ' span ' ) ` `
 auxiliary _ edge _ attributes : optional
 Return value of ` ` networkx . get _ edge _ attributes ( graph , ' span ' ) ` `
 spanning _ sides : list , optional
 List of keys ( attribute values ) of the two sides of the auxiliary
 nodes .
 Return value of ` ` list ( set ( auxiliary _ node _ attributes . values ( ) ) ) ` `
 Yields
 ret : ndarray
 Structured array with dtype ` ` dtype = [ ( ' has _ spanning _ cluster ' , ' bool ' ) ,
 ( ' max _ cluster _ size ' , ' uint32 ' ) , ( ' moments ' , ' int64 ' , 5 ) ] ` `
 ret [ ' n ' ] : ndarray of int
 The number of bonds added at the particular iteration
 ret [ ' edge ' ] : ndarray of int
 The index of the edge added at the particular iteration
 Note that in the first step , when ` ` ret [ ' n ' ] = = 0 ` ` , this value is
 undefined !
 ret [ ' has _ spanning _ cluster ' ] : ndarray of bool
 ` ` True ` ` if there is a spanning cluster , ` ` False ` ` otherwise .
 Only exists if ` spanning _ cluster ` argument is set to ` ` True ` ` .
 ret [ ' max _ cluster _ size ' ] : int
 Size of the largest cluster ( absolute number of sites )
 ret [ ' moments ' ] : 1 - D : py : class : ` numpy . ndarray ` of int
 Array of size ` ` 5 ` ` .
 The ` ` k ` ` - th entry is the ` ` k ` ` - th raw moment of the ( absolute ) cluster
 size distribution , with ` ` k ` ` ranging from ` ` 0 ` ` to ` ` 4 ` ` .
 Raises
 ValueError
 If ` spanning _ cluster ` is ` ` True ` ` , but ` graph ` does not contain any
 auxiliary nodes to detect spanning clusters .
 See also
 numpy . random . RandomState
 microcanonical _ statistics _ dtype
 Notes
 Iterating through this generator is a single run of the Newman - Ziff
 algorithm . [ 12 ] _
 The first iteration yields the trivial state with : math : ` n = 0 ` occupied
 bonds .
 Spanning cluster
 In order to detect a spanning cluster , ` graph ` needs to contain
 auxiliary nodes and edges , cf . Reference [ 12 ] _ , Figure 6.
 The auxiliary nodes and edges have the ` ` ' span ' ` ` ` attribute
 < http : / / networkx . github . io / documentation / latest / tutorial / tutorial . html # node - attributes > ` _ .
 The value is either ` ` 0 ` ` or ` ` 1 ` ` , distinguishing the two sides of the
 graph to span .
 Raw moments of the cluster size distribution
 The : math : ` k ` - th raw moment of the ( absolute ) cluster size distribution
 is : math : ` \ sum _ s ' s ^ k N _ s ` , where : math : ` s ` is the cluster size and
 : math : ` N _ s ` is the number of clusters of size : math : ` s ` . [ 13 ] _
 The primed sum : math : ` \ sum ' ` signifies that the largest cluster is
 excluded from the sum . [ 14 ] _
 References
 . . [ 12 ] Newman , M . E . J . & Ziff , R . M . Fast monte carlo algorithm for site
 or bond percolation . Physical Review E 64 , 016706 + ( 2001 ) ,
 ` doi : 10.1103 / physreve . 64.016706 < http : / / dx . doi . org / 10.1103 / physreve . 64.016706 > ` _ .
 . . [ 13 ] Stauffer , D . & Aharony , A . Introduction to Percolation Theory ( Taylor &
 Francis , London , 1994 ) , second edn .
 . . [ 14 ] Binder , K . & Heermann , D . W . Monte Carlo Simulation in Statistical
 Physics ( Springer , Berlin , Heidelberg , 2010 ) ,
 ` doi : 10.1007/978-3-642-03163-2 < http : / / dx . doi . org / 10.1007/978-3-642-03163-2 > ` _ .'''

implementation:
# construct random number generator
rng = np . random . RandomState ( seed = seed )
if spanning_cluster :
    if len ( spanning_sides ) != 2 :
        raise ValueError ( 'Spanning cluster is to be detected, but auxiliary nodes ' 'of less or more than 2 types (sides) given.' )
# get a list of edges for easy access in later iterations
perc_edges = perc_graph . edges ( )
perm_edges = rng . permutation ( num_edges )
# initial iteration : no edges added yet ( n = = 0)
ret = np . empty ( 1 , dtype = microcanonical_statistics_dtype ( spanning_cluster ) )
ret [ 'n' ] = 0
ret [ 'max_cluster_size' ] = 1
ret [ 'moments' ] = np . ones ( 5 , dtype = 'uint64' ) * ( num_nodes - 1 )
if spanning_cluster :
    ret [ 'has_spanning_cluster' ] = False
# yield cluster statistics for n = = 0
yield ret
# set up disjoint set ( union - find ) data structure
ds = nx . utils . union_find . UnionFind ( )
if spanning_cluster :
    ds_spanning = nx . utils . union_find . UnionFind ( )
    # merge all auxiliary nodes for each side
    side_roots = dict ( )
    for side in spanning_sides :
        nodes = [ node for ( node , node_side ) in auxiliary_node_attributes . items ( ) if node_side is side ]
        ds_spanning . union ( * nodes )
        side_roots [ side ] = ds_spanning [ nodes [ 0 ] ]
    for ( edge , edge_side ) in auxiliary_edge_attributes . items ( ) :
        ds_spanning . union ( side_roots [ edge_side ] , * edge )
    side_roots = [ ds_spanning [ side_root ] for side_root in side_roots . values ( ) ]
# get first node
max_cluster_root = next ( perc_graph . nodes_iter ( ) )
# loop over all edges ( n = = 1 . . M )
for n in range ( num_edges ) :
    ret [ 'n' ] += 1
    # draw new edge from permutation
    edge_index = perm_edges [ n ]
    edge = perc_edges [ edge_index ]
    ret [ 'edge' ] = edge_index
    # find roots and weights
    roots = [ ds [ node ] for node in edge ]
    weights = [ ds . weights [ root ] for root in roots ]
    if roots [ 0 ] is not roots [ 1 ] : # not same cluster : union !
        ds . union ( * roots )
        if spanning_cluster :
            ds_spanning . union ( * roots )
            ret [ 'has_spanning_cluster' ] = ( ds_spanning [ side_roots [ 0 ] ] == ds_spanning [ side_roots [ 1 ] ] )
        # find new root and weight
        root = ds [ edge [ 0 ] ]
        weight = ds . weights [ root ]
        # moments and maximum cluster size
        # deduct the previous sub - maximum clusters from moments
        for i in [ 0 , 1 ] :
            if roots [ i ] is max_cluster_root :
                continue
            ret [ 'moments' ] -= weights [ i ] ** np . arange ( 5 , dtype = 'uint64' )
        if max_cluster_root in roots : # merged with maximum cluster
            max_cluster_root = root
            ret [ 'max_cluster_size' ] = weight
        else : # merged previously sub - maximum clusters
            if ret [ 'max_cluster_size' ] >= weight : # previously largest cluster remains largest cluster
            # add merged cluster to moments
                ret [ 'moments' ] += weight ** np . arange ( 5 , dtype = 'uint64' )
            else : # merged cluster overtook previously largest cluster
            # add previously largest cluster to moments
                max_cluster_root = root
                ret [ 'moments' ] += ret [ 'max_cluster_size' ] ** np . arange ( 5 , dtype = 'uint64' )
                ret [ 'max_cluster_size' ] = weight
    yield ret


signature:
def bond_microcanonical_statistics ( perc_graph , num_nodes , num_edges , seed , spanning_cluster = True , auxiliary_node_attributes = None , auxiliary_edge_attributes = None , spanning_sides = None , ** kwargs ) :
    """Evolve a single run over all microstates ( bond occupation numbers )
 Return the cluster statistics for each microstate
 Parameters
 perc _ graph : networkx . Graph
 The substrate graph on which percolation is to take place
 num _ nodes : int
 Number ` ` N ` ` of sites in the graph
 num _ edges : int
 Number ` ` M ` ` of bonds in the graph
 seed : { None , int , array _ like }
 Random seed initializing the pseudo - random number generator .
 Piped through to ` numpy . random . RandomState ` constructor .
 spanning _ cluster : bool , optional
 Whether to detect a spanning cluster or not .
 Defaults to ` ` True ` ` .
 auxiliary _ node _ attributes : optional
 Value of ` ` networkx . get _ node _ attributes ( graph , ' span ' ) ` `
 auxiliary _ edge _ attributes : optional
 Value of ` ` networkx . get _ edge _ attributes ( graph , ' span ' ) ` `
 spanning _ sides : list , optional
 List of keys ( attribute values ) of the two sides of the auxiliary
 nodes .
 Return value of ` ` list ( set ( auxiliary _ node _ attributes . values ( ) ) ) ` `
 Returns
 ret : ndarray of size ` ` num _ edges + 1 ` `
 Structured array with dtype ` ` dtype = [ ( ' has _ spanning _ cluster ' , ' bool ' ) ,
 ( ' max _ cluster _ size ' , ' uint32 ' ) , ( ' moments ' , ' uint64 ' , 5 ) ] ` `
 ret [ ' n ' ] : ndarray of int
 The number of bonds added at the particular iteration
 ret [ ' edge ' ] : ndarray of int
 The index of the edge added at the particular iteration .
 Note that ` ` ret [ ' edge ' ] [ 0 ] ` ` is undefined !
 ret [ ' has _ spanning _ cluster ' ] : ndarray of bool
 ` ` True ` ` if there is a spanning cluster , ` ` False ` ` otherwise .
 Only exists if ` spanning _ cluster ` argument is set to ` ` True ` ` .
 ret [ ' max _ cluster _ size ' ] : int
 Size of the largest cluster ( absolute number of sites )
 ret [ ' moments ' ] : 2 - D : py : class : ` numpy . ndarray ` of int
 Array of shape ` ` ( num _ edges + 1 , 5 ) ` ` .
 The ` ` k ` ` - th entry is the ` ` k ` ` - th raw moment of the ( absolute ) cluster
 size distribution , with ` ` k ` ` ranging from ` ` 0 ` ` to ` ` 4 ` ` .
 See also
 bond _ sample _ states
 microcanonical _ statistics _ dtype
 numpy . random . RandomState"""

implementation:
# initialize generator
sample_states = bond_sample_states ( perc_graph = perc_graph , num_nodes = num_nodes , num_edges = num_edges , seed = seed , spanning_cluster = spanning_cluster , auxiliary_node_attributes = auxiliary_node_attributes , auxiliary_edge_attributes = auxiliary_edge_attributes , spanning_sides = spanning_sides , )
# get cluster statistics over all microstates
return np . fromiter ( sample_states , dtype = microcanonical_statistics_dtype ( spanning_cluster ) , count = num_edges + 1 )


signature:
def canonical_statistics_dtype ( spanning_cluster = True ) :
    """The NumPy Structured Array type for canonical statistics
 Helper function
 Parameters
 spanning _ cluster : bool , optional
 Whether to detect a spanning cluster or not .
 Defaults to ` ` True ` ` .
 Returns
 ret : list of pairs of str
 A list of tuples of field names and data types to be used as ` ` dtype ` `
 argument in numpy ndarray constructors
 See Also
 http : / / docs . scipy . org / doc / numpy / user / basics . rec . html
 microcanoncial _ statistics _ dtype
 canonical _ averages _ dtype"""

implementation:
fields = list ( )
if spanning_cluster :
    fields . extend ( [ ( 'percolation_probability' , 'float64' ) , ] )
fields . extend ( [ ( 'max_cluster_size' , 'float64' ) , ( 'moments' , '(5,)float64' ) , ] )
return _ndarray_dtype ( fields )


signature:
def bond_canonical_statistics ( microcanonical_statistics , convolution_factors , ** kwargs ) :
    """canonical cluster statistics for a single run and a single probability
 Parameters
 microcanonical _ statistics : ndarray
 Return value of ` bond _ microcanonical _ statistics `
 convolution _ factors : 1 - D array _ like
 The coefficients of the convolution for the given probabilty ` ` p ` `
 and for each occupation number ` ` n ` ` .
 Returns
 ret : ndarray of size ` ` 1 ` `
 Structured array with dtype as returned by
 ` canonical _ statistics _ dtype `
 ret [ ' percolation _ probability ' ] : ndarray of float
 The " percolation probability " of this run at the value of ` ` p ` ` .
 Only exists if ` microcanonical _ statistics ` argument has the
 ` ` has _ spanning _ cluster ` ` field .
 ret [ ' max _ cluster _ size ' ] : ndarray of int
 Weighted size of the largest cluster ( absolute number of sites )
 ret [ ' moments ' ] : 1 - D : py : class : ` numpy . ndarray ` of float
 Array of size ` ` 5 ` ` .
 The ` ` k ` ` - th entry is the weighted ` ` k ` ` - th raw moment of the
 ( absolute ) cluster size distribution , with ` ` k ` ` ranging from ` ` 0 ` ` to
 See Also
 bond _ microcanonical _ statistics
 canonical _ statistics _ dtype"""

implementation:
# initialize return array
spanning_cluster = ( 'has_spanning_cluster' in microcanonical_statistics . dtype . names )
ret = np . empty ( 1 , dtype = canonical_statistics_dtype ( spanning_cluster ) )
# compute percolation probability
if spanning_cluster :
    ret [ 'percolation_probability' ] = np . sum ( convolution_factors * microcanonical_statistics [ 'has_spanning_cluster' ] )
# convolve maximum cluster size
ret [ 'max_cluster_size' ] = np . sum ( convolution_factors * microcanonical_statistics [ 'max_cluster_size' ] )
# convolve moments
ret [ 'moments' ] = np . sum ( convolution_factors [ : , np . newaxis ] * microcanonical_statistics [ 'moments' ] , axis = 0 , )
# return convolved cluster statistics
return ret


signature:
def canonical_averages_dtype ( spanning_cluster = True ) :
    """The NumPy Structured Array type for canonical averages over several
 runs
 Helper function
 Parameters
 spanning _ cluster : bool , optional
 Whether to detect a spanning cluster or not .
 Defaults to ` ` True ` ` .
 Returns
 ret : list of pairs of str
 A list of tuples of field names and data types to be used as ` ` dtype ` `
 argument in numpy ndarray constructors
 See Also
 http : / / docs . scipy . org / doc / numpy / user / basics . rec . html
 canonical _ statistics _ dtype
 finalized _ canonical _ averages _ dtype"""

implementation:
fields = list ( )
fields . extend ( [ ( 'number_of_runs' , 'uint32' ) , ] )
if spanning_cluster :
    fields . extend ( [ ( 'percolation_probability_mean' , 'float64' ) , ( 'percolation_probability_m2' , 'float64' ) , ] )
fields . extend ( [ ( 'max_cluster_size_mean' , 'float64' ) , ( 'max_cluster_size_m2' , 'float64' ) , ( 'moments_mean' , '(5,)float64' ) , ( 'moments_m2' , '(5,)float64' ) , ] )
return _ndarray_dtype ( fields )


signature:
def bond_initialize_canonical_averages ( canonical_statistics , ** kwargs ) :
    """Initialize the canonical averages from a single - run cluster statistics
 Parameters
 canonical _ statistics : 1 - D structured ndarray
 Typically contains the canonical statistics for a range of values
 of the occupation probability ` ` p ` ` .
 The dtype is the result of ` canonical _ statistics _ dtype ` .
 Returns
 ret : structured ndarray
 The dype is the result of ` canonical _ averages _ dtype ` .
 ret [ ' number _ of _ runs ' ] : 1 - D ndarray of int
 Equals ` ` 1 ` ` ( initial run ) .
 ret [ ' percolation _ probability _ mean ' ] : 1 - D array of float
 Equals ` ` canonical _ statistics [ ' percolation _ probability ' ] ` `
 ( if ` ` percolation _ probability ` ` is present )
 ret [ ' percolation _ probability _ m2 ' ] : 1 - D array of float
 Each entry is ` ` 0.0 ` `
 ret [ ' max _ cluster _ size _ mean ' ] : 1 - D array of float
 Equals ` ` canonical _ statistics [ ' max _ cluster _ size ' ] ` `
 ret [ ' max _ cluster _ size _ m2 ' ] : 1 - D array of float
 Each entry is ` ` 0.0 ` `
 ret [ ' moments _ mean ' ] : 2 - D array of float
 Equals ` ` canonical _ statistics [ ' moments ' ] ` `
 ret [ ' moments _ m2 ' ] : 2 - D array of float
 Each entry is ` ` 0.0 ` `
 See Also
 canonical _ averages _ dtype
 bond _ canonical _ statistics"""

implementation:
# initialize return array
spanning_cluster = ( 'percolation_probability' in canonical_statistics . dtype . names )
# array should have the same size as the input array
ret = np . empty_like ( canonical_statistics , dtype = canonical_averages_dtype ( spanning_cluster = spanning_cluster ) , )
ret [ 'number_of_runs' ] = 1
# initialize percolation probability mean and sum of squared differences
if spanning_cluster :
    ret [ 'percolation_probability_mean' ] = ( canonical_statistics [ 'percolation_probability' ] )
    ret [ 'percolation_probability_m2' ] = 0.0
# initialize maximum cluster size mean and sum of squared differences
ret [ 'max_cluster_size_mean' ] = ( canonical_statistics [ 'max_cluster_size' ] )
ret [ 'max_cluster_size_m2' ] = 0.0
# initialize moments means and sums of squared differences
ret [ 'moments_mean' ] = canonical_statistics [ 'moments' ]
ret [ 'moments_m2' ] = 0.0
return ret


signature:
def bond_reduce ( row_a , row_b ) :
    """Reduce the canonical averages over several runs
 This is a " true " reducer .
 It is associative and commutative .
 This is a wrapper around ` simoa . stats . online _ variance ` .
 Parameters
 row _ a , row _ b : structured ndarrays
 Output of this function , or initial input from
 ` bond _ initialize _ canonical _ averages `
 Returns
 ret : structured ndarray
 Array is of dtype as returned by ` canonical _ averages _ dtype `
 See Also
 bond _ initialize _ canonical _ averages
 canonical _ averages _ dtype
 simoa . stats . online _ variance"""

implementation:
spanning_cluster = ( 'percolation_probability_mean' in row_a . dtype . names and 'percolation_probability_mean' in row_b . dtype . names and 'percolation_probability_m2' in row_a . dtype . names and 'percolation_probability_m2' in row_b . dtype . names )
# initialize return array
ret = np . empty_like ( row_a )
def _reducer ( key , transpose = False ) :
    mean_key = '{}_mean' . format ( key )
    m2_key = '{}_m2' . format ( key )
    res = simoa . stats . online_variance ( * [ ( row [ 'number_of_runs' ] , row [ mean_key ] . T if transpose else row [ mean_key ] , row [ m2_key ] . T if transpose else row [ m2_key ] , ) for row in [ row_a , row_b ] ] )
    ( ret [ mean_key ] , ret [ m2_key ] , ) = ( res [ 1 ] . T , res [ 2 ] . T , ) if transpose else res [ 1 : ]
if spanning_cluster :
    _reducer ( 'percolation_probability' )
_reducer ( 'max_cluster_size' )
_reducer ( 'moments' , transpose = True )
ret [ 'number_of_runs' ] = row_a [ 'number_of_runs' ] + row_b [ 'number_of_runs' ]
return ret


signature:
def finalized_canonical_averages_dtype ( spanning_cluster = True ) :
    """The NumPy Structured Array type for finalized canonical averages over
 several runs
 Helper function
 Parameters
 spanning _ cluster : bool , optional
 Whether to detect a spanning cluster or not .
 Defaults to ` ` True ` ` .
 Returns
 ret : list of pairs of str
 A list of tuples of field names and data types to be used as ` ` dtype ` `
 argument in numpy ndarray constructors
 See Also
 http : / / docs . scipy . org / doc / numpy / user / basics . rec . html
 canonical _ averages _ dtype"""

implementation:
fields = list ( )
fields . extend ( [ ( 'number_of_runs' , 'uint32' ) , ( 'p' , 'float64' ) , ( 'alpha' , 'float64' ) , ] )
if spanning_cluster :
    fields . extend ( [ ( 'percolation_probability_mean' , 'float64' ) , ( 'percolation_probability_std' , 'float64' ) , ( 'percolation_probability_ci' , '(2,)float64' ) , ] )
fields . extend ( [ ( 'percolation_strength_mean' , 'float64' ) , ( 'percolation_strength_std' , 'float64' ) , ( 'percolation_strength_ci' , '(2,)float64' ) , ( 'moments_mean' , '(5,)float64' ) , ( 'moments_std' , '(5,)float64' ) , ( 'moments_ci' , '(5,2)float64' ) , ] )
return _ndarray_dtype ( fields )


signature:
def finalize_canonical_averages ( number_of_nodes , ps , canonical_averages , alpha , ) :
    """Finalize canonical averages"""

implementation:
spanning_cluster = ( ( 'percolation_probability_mean' in canonical_averages . dtype . names ) and 'percolation_probability_m2' in canonical_averages . dtype . names )
# append values of p as an additional field
ret = np . empty_like ( canonical_averages , dtype = finalized_canonical_averages_dtype ( spanning_cluster = spanning_cluster ) , )
n = canonical_averages [ 'number_of_runs' ]
sqrt_n = np . sqrt ( canonical_averages [ 'number_of_runs' ] )
ret [ 'number_of_runs' ] = n
ret [ 'p' ] = ps
ret [ 'alpha' ] = alpha
def _transform ( original_key , final_key = None , normalize = False , transpose = False , ) :
    if final_key is None :
        final_key = original_key
    keys_mean = [ '{}_mean' . format ( key ) for key in [ original_key , final_key ] ]
    keys_std = [ '{}_m2' . format ( original_key ) , '{}_std' . format ( final_key ) , ]
    key_ci = '{}_ci' . format ( final_key )
    # calculate sample mean
    ret [ keys_mean [ 1 ] ] = canonical_averages [ keys_mean [ 0 ] ]
    if normalize :
        ret [ keys_mean [ 1 ] ] /= number_of_nodes
    # calculate sample standard deviation
    array = canonical_averages [ keys_std [ 0 ] ]
    result = np . sqrt ( ( array . T if transpose else array ) / ( n - 1 ) )
    ret [ keys_std [ 1 ] ] = ( result . T if transpose else result )
    if normalize :
        ret [ keys_std [ 1 ] ] /= number_of_nodes
    # calculate standard normal confidence interval
    array = ret [ keys_std [ 1 ] ]
    scale = ( array . T if transpose else array ) / sqrt_n
    array = ret [ keys_mean [ 1 ] ]
    mean = ( array . T if transpose else array )
    result = scipy . stats . t . interval ( 1 - alpha , df = n - 1 , loc = mean , scale = scale , )
    ( ret [ key_ci ] [ ... , 0 ] , ret [ key_ci ] [ ... , 1 ] ) = ( [ my_array . T for my_array in result ] if transpose else result )
if spanning_cluster :
    _transform ( 'percolation_probability' )
_transform ( 'max_cluster_size' , 'percolation_strength' , normalize = True )
_transform ( 'moments' , normalize = True , transpose = True )
return ret


signature:
def compare ( graph : BELGraph , annotation : str = 'Subgraph' ) -> Mapping [ str , Mapping [ str , float ] ] :
    """Compare generated mechanisms to actual ones .
 1 . Generates candidate mechanisms for each biological process
 2 . Gets sub - graphs for all NeuroMMSig signatures
 3 . Make tanimoto similarity comparison for all sets
 : return : A dictionary table comparing the canonical subgraphs to generated ones"""

implementation:
canonical_mechanisms = get_subgraphs_by_annotation ( graph , annotation )
canonical_nodes = _transform_graph_dict_to_node_dict ( canonical_mechanisms )
candidate_mechanisms = generate_bioprocess_mechanisms ( graph )
candidate_nodes = _transform_graph_dict_to_node_dict ( candidate_mechanisms )
results : Dict [ str , Dict [ str , float ] ] = defaultdict ( dict )
it = itt . product ( canonical_nodes . items ( ) , candidate_nodes . items ( ) )
for ( canonical_name , canonical_graph ) , ( candidate_bp , candidate_graph ) in it :
    tanimoto = tanimoto_set_similarity ( candidate_nodes , canonical_nodes )
    results [ canonical_name ] [ candidate_bp ] = tanimoto
return dict ( results )


signature:
def summarize_node_filter ( graph : BELGraph , node_filters : NodePredicates ) -> None :
    """Print a summary of the number of nodes passing a given set of filters .
 : param graph : A BEL graph
 : param node _ filters : A node filter or list / tuple of node filters"""

implementation:
passed = count_passed_node_filter ( graph , node_filters )
print ( '{}/{} nodes passed' . format ( passed , graph . number_of_nodes ( ) ) )


signature:
def variants_of ( graph : BELGraph , node : Protein , modifications : Optional [ Set [ str ] ] = None , ) -> Set [ Protein ] :
    """Returns all variants of the given node ."""

implementation:
if modifications :
    return _get_filtered_variants_of ( graph , node , modifications )
return { v for u , v , key , data in graph . edges ( keys = True , data = True ) if ( u == node and data [ RELATION ] == HAS_VARIANT and pybel . struct . has_protein_modification ( v ) ) }


signature:
def get_variants_to_controllers ( graph : BELGraph , node : Protein , modifications : Optional [ Set [ str ] ] = None , ) -> Mapping [ Protein , Set [ Protein ] ] :
    """Get a mapping from variants of the given node to all of its upstream controllers ."""

implementation:
rv = defaultdict ( set )
variants = variants_of ( graph , node , modifications )
for controller , variant , data in graph . in_edges ( variants , data = True ) :
    if data [ RELATION ] in CAUSAL_RELATIONS :
        rv [ variant ] . add ( controller )
return rv


signature:
def group_dict_set ( iterator : Iterable [ Tuple [ A , B ] ] ) -> Mapping [ A , Set [ B ] ] :
    """Make a dict that accumulates the values for each key in an iterator of doubles ."""

implementation:
d = defaultdict ( set )
for key , value in iterator :
    d [ key ] . add ( value )
return dict ( d )


signature:
def get_edge_relations ( graph : BELGraph ) -> Mapping [ Tuple [ BaseEntity , BaseEntity ] , Set [ str ] ] :
    """Build a dictionary of { node pair : set of edge types } ."""

implementation: return group_dict_set ( ( ( u , v ) , d [ RELATION ] ) for u , v , d in graph . edges ( data = True ) )


signature:
def count_unique_relations ( graph : BELGraph ) -> Counter :
    """Return a histogram of the different types of relations present in a graph .
 Note : this operation only counts each type of edge once for each pair of nodes"""

implementation: return Counter ( itt . chain . from_iterable ( get_edge_relations ( graph ) . values ( ) ) )


signature:
def get_annotations_containing_keyword ( graph : BELGraph , keyword : str ) -> List [ Mapping [ str , str ] ] :
    """Get annotation / value pairs for values for whom the search string is a substring
 : param graph : A BEL graph
 : param keyword : Search for annotations whose values have this as a substring"""

implementation: return [ { 'annotation' : annotation , 'value' : value } for annotation , value in iter_annotation_value_pairs ( graph ) if keyword . lower ( ) in value . lower ( ) ]


signature:
def count_annotation_values ( graph : BELGraph , annotation : str ) -> Counter :
    """Count in how many edges each annotation appears in a graph
 : param graph : A BEL graph
 : param annotation : The annotation to count
 : return : A Counter from { annotation value : frequency }"""

implementation: return Counter ( iter_annotation_values ( graph , annotation ) )


signature:
def count_annotation_values_filtered ( graph : BELGraph , annotation : str , source_predicate : Optional [ NodePredicate ] = None , target_predicate : Optional [ NodePredicate ] = None , ) -> Counter :
    """Count in how many edges each annotation appears in a graph , but filter out source nodes and target nodes .
 See : func : ` pybel _ tools . utils . keep _ node ` for a basic filter .
 : param graph : A BEL graph
 : param annotation : The annotation to count
 : param source _ predicate : A predicate ( graph , node ) - > bool for keeping source nodes
 : param target _ predicate : A predicate ( graph , node ) - > bool for keeping target nodes
 : return : A Counter from { annotation value : frequency }"""

implementation:
if source_predicate and target_predicate :
    return Counter ( data [ ANNOTATIONS ] [ annotation ] for u , v , data in graph . edges ( data = True ) if edge_has_annotation ( data , annotation ) and source_predicate ( graph , u ) and target_predicate ( graph , v ) )
elif source_predicate :
    return Counter ( data [ ANNOTATIONS ] [ annotation ] for u , v , data in graph . edges ( data = True ) if edge_has_annotation ( data , annotation ) and source_predicate ( graph , u ) )
elif target_predicate :
    return Counter ( data [ ANNOTATIONS ] [ annotation ] for u , v , data in graph . edges ( data = True ) if edge_has_annotation ( data , annotation ) and target_predicate ( graph , u ) )
else :
    return Counter ( data [ ANNOTATIONS ] [ annotation ] for u , v , data in graph . edges ( data = True ) if edge_has_annotation ( data , annotation ) )


signature:
def pair_is_consistent ( graph : BELGraph , u : BaseEntity , v : BaseEntity ) -> Optional [ str ] :
    """Return if the edges between the given nodes are consistent , meaning they all have the same relation .
 : return : If the edges aren ' t consistent , return false , otherwise return the relation type"""

implementation:
relations = { data [ RELATION ] for data in graph [ u ] [ v ] . values ( ) }
if 1 != len ( relations ) :
    return
return list ( relations ) [ 0 ]


signature:
def get_contradictory_pairs ( graph : BELGraph ) -> Iterable [ Tuple [ BaseEntity , BaseEntity ] ] :
    """Iterates over contradictory node pairs in the graph based on their causal relationships
 : return : An iterator over ( source , target ) node pairs that have contradictory causal edges"""

implementation:
for u , v in graph . edges ( ) :
    if pair_has_contradiction ( graph , u , v ) :
        yield u , v


signature:
def get_consistent_edges ( graph : BELGraph ) -> Iterable [ Tuple [ BaseEntity , BaseEntity ] ] :
    """Yield pairs of ( source node , target node ) for which all of their edges have the same type of relation .
 : return : An iterator over ( source , target ) node pairs corresponding to edges with many inconsistent relations"""

implementation:
for u , v in graph . edges ( ) :
    if pair_is_consistent ( graph , u , v ) :
        yield u , v


signature:
def infer_missing_two_way_edges ( graph ) :
    """Add edges to the graph when a two way edge exists , and the opposite direction doesn ' t exist .
 Use : two way edges from BEL definition and / or axiomatic inverses of membership relations
 : param pybel . BELGraph graph : A BEL graph"""

implementation:
for u , v , k , d in graph . edges ( data = True , keys = True ) :
    if d [ RELATION ] in TWO_WAY_RELATIONS :
        infer_missing_backwards_edge ( graph , u , v , k )


signature:
def infer_missing_backwards_edge ( graph , u , v , k ) :
    """Add the same edge , but in the opposite direction if not already present .
 : type graph : pybel . BELGraph
 : type u : tuple
 : type v : tuple
 : type k : int"""

implementation:
if u in graph [ v ] :
    for attr_dict in graph [ v ] [ u ] . values ( ) :
        if attr_dict == graph [ u ] [ v ] [ k ] :
            return
graph . add_edge ( v , u , key = k , ** graph [ u ] [ v ] [ k ] )


signature:
def enrich_internal_unqualified_edges ( graph , subgraph ) :
    """Add the missing unqualified edges between entities in the subgraph that are contained within the full graph .
 : param pybel . BELGraph graph : The full BEL graph
 : param pybel . BELGraph subgraph : The query BEL subgraph"""

implementation:
for u , v in itt . combinations ( subgraph , 2 ) :
    if not graph . has_edge ( u , v ) :
        continue
    for k in graph [ u ] [ v ] :
        if k < 0 :
            subgraph . add_edge ( u , v , key = k , ** graph [ u ] [ v ] [ k ] )


signature:
def boilerplate ( name , contact , description , pmids , version , copyright , authors , licenses , disclaimer , output ) :
    """Build a template BEL document with the given PubMed identifiers ."""

implementation:
from . document_utils import write_boilerplate
write_boilerplate ( name = name , version = version , description = description , authors = authors , contact = contact , copyright = copyright , licenses = licenses , disclaimer = disclaimer , pmids = pmids , file = output , )


signature:
def serialize_namespaces ( namespaces , connection : str , path , directory ) :
    """Parse a BEL document then serializes the given namespaces ( errors and all ) to the given directory ."""

implementation:
from . definition_utils import export_namespaces
graph = from_lines ( path , manager = connection )
export_namespaces ( namespaces , graph , directory )


signature:
def get_pmids ( graph : BELGraph , output : TextIO ) :
    """Output PubMed identifiers from a graph to a stream ."""

implementation:
for pmid in get_pubmed_identifiers ( graph ) :
    click . echo ( pmid , file = output )


signature:
def getrowcount ( self , window_name , object_name ) :
    """Get count of rows in table object .
 @ param window _ name : Window name to look for , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to look for , either full name ,
 LDTP ' s name convention , or a Unix glob . Or menu heirarchy
 @ type object _ name : string
 @ return : Number of rows .
 @ rtype : integer"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
return len ( object_handle . AXRows )


signature:
def selectrow ( self , window_name , object_name , row_text , partial_match = False ) :
    """Select row
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ text : Row text to select
 @ type row _ text : string
 @ return : 1 on success .
 @ rtype : integer"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
for cell in object_handle . AXRows :
    if re . match ( row_text , cell . AXChildren [ 0 ] . AXValue ) :
        if not cell . AXSelected :
            object_handle . activate ( )
            cell . AXSelected = True
        else : # Selected
            pass
        return 1
raise LdtpServerException ( u"Unable to select row: %s" % row_text )


signature:
def multiselect ( self , window_name , object_name , row_text_list , partial_match = False ) :
    """Select multiple row
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ text _ list : Row list with matching text to select
 @ type row _ text : string
 @ return : 1 on success .
 @ rtype : integer"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
object_handle . activate ( )
selected = False
try :
    window = self . _get_front_most_window ( )
except ( IndexError , ) :
    window = self . _get_any_window ( )
for row_text in row_text_list :
    selected = False
    for cell in object_handle . AXRows :
        parent_cell = cell
        cell = self . _getfirstmatchingchild ( cell , "(AXTextField|AXStaticText)" )
        if not cell :
            continue
        if re . match ( row_text , cell . AXValue ) :
            selected = True
            if not parent_cell . AXSelected :
                x , y , width , height = self . _getobjectsize ( parent_cell )
                window . clickMouseButtonLeftWithMods ( ( x + width / 2 , y + height / 2 ) , [ '<command_l>' ] )
                # Following selection doesn ' t work
                # parent _ cell . AXSelected = True
                self . wait ( 0.5 )
            else : # Selected
                pass
            break
    if not selected :
        raise LdtpServerException ( u"Unable to select row: %s" % row_text )
if not selected :
    raise LdtpServerException ( u"Unable to select any row" )
return 1


signature:
def selectrowpartialmatch ( self , window_name , object_name , row_text ) :
    """Select row partial match
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ text : Row text to select
 @ type row _ text : string
 @ return : 1 on success .
 @ rtype : integer"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
for cell in object_handle . AXRows :
    if re . search ( row_text , cell . AXChildren [ 0 ] . AXValue ) :
        if not cell . AXSelected :
            object_handle . activate ( )
            cell . AXSelected = True
        else : # Selected
            pass
        return 1
raise LdtpServerException ( u"Unable to select row: %s" % row_text )


signature:
def selectrowindex ( self , window_name , object_name , row_index ) :
    """Select row index
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ index : Row index to select
 @ type row _ index : integer
 @ return : 1 on success .
 @ rtype : integer"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
count = len ( object_handle . AXRows )
if row_index < 0 or row_index > count :
    raise LdtpServerException ( 'Row index out of range: %d' % row_index )
cell = object_handle . AXRows [ row_index ]
if not cell . AXSelected :
    object_handle . activate ( )
    cell . AXSelected = True
else : # Selected
    pass
return 1


signature:
def selectlastrow ( self , window_name , object_name ) :
    """Select last row
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ return : 1 on success .
 @ rtype : integer"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
cell = object_handle . AXRows [ - 1 ]
if not cell . AXSelected :
    object_handle . activate ( )
    cell . AXSelected = True
else : # Selected
    pass
return 1


signature:
def getcellvalue ( self , window_name , object_name , row_index , column = 0 ) :
    """Get cell value
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ index : Row index to get
 @ type row _ index : integer
 @ param column : Column index to get , default value 0
 @ type column : integer
 @ return : cell value on success .
 @ rtype : string"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
count = len ( object_handle . AXRows )
if row_index < 0 or row_index > count :
    raise LdtpServerException ( 'Row index out of range: %d' % row_index )
cell = object_handle . AXRows [ row_index ]
count = len ( cell . AXChildren )
if column < 0 or column > count :
    raise LdtpServerException ( 'Column index out of range: %d' % column )
obj = cell . AXChildren [ column ]
if not re . search ( "AXColumn" , obj . AXRole ) :
    obj = cell . AXChildren [ column ]
return obj . AXValue


signature:
def gettablerowindex ( self , window_name , object_name , row_text ) :
    """Get table row index matching given text
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ text : Row text to select
 @ type row _ text : string
 @ return : row index matching the text on success .
 @ rtype : integer"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
index = 0
for cell in object_handle . AXRows :
    if re . match ( row_text , cell . AXChildren [ 0 ] . AXValue ) :
        return index
    index += 1
raise LdtpServerException ( u"Unable to find row: %s" % row_text )


signature:
def doubleclickrow ( self , window_name , object_name , row_text ) :
    """Double click row matching given text
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ text : Row text to select
 @ type row _ text : string
 @ return : row index matching the text on success .
 @ rtype : integer"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
object_handle . activate ( )
self . wait ( 1 )
for cell in object_handle . AXRows :
    cell = self . _getfirstmatchingchild ( cell , "(AXTextField|AXStaticText)" )
    if not cell :
        continue
    if re . match ( row_text , cell . AXValue ) :
        x , y , width , height = self . _getobjectsize ( cell )
        # Mouse double click on the object
        cell . doubleClickMouse ( ( x + width / 2 , y + height / 2 ) )
        return 1
raise LdtpServerException ( 'Unable to get row text: %s' % row_text )


signature:
def doubleclickrowindex ( self , window_name , object_name , row_index , col_index = 0 ) :
    """Double click row matching given text
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ index : Row index to click
 @ type row _ index : integer
 @ param col _ index : Column index to click
 @ type col _ index : integer
 @ return : row index matching the text on success .
 @ rtype : integer"""

implementation:
object_handle = self . _get_object_handle ( window_name , object_name )
if not object_handle . AXEnabled :
    raise LdtpServerException ( u"Object %s state disabled" % object_name )
count = len ( object_handle . AXRows )
if row_index < 0 or row_index > count :
    raise LdtpServerException ( 'Row index out of range: %d' % row_index )
cell = object_handle . AXRows [ row_index ]
self . _grabfocus ( cell )
x , y , width , height = self . _getobjectsize ( cell )
# Mouse double click on the object
cell . doubleClickMouse ( ( x + width / 2 , y + height / 2 ) )
return 1


signature:
def verifytablecell ( self , window_name , object_name , row_index , column_index , row_text ) :
    """Verify table cell value with given text
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ index : Row index to get
 @ type row _ index : integer
 @ param column _ index : Column index to get , default value 0
 @ type column _ index : integer
 @ param row _ text : Row text to match
 @ type string
 @ return : 1 on success 0 on failure .
 @ rtype : integer"""

implementation:
try :
    value = getcellvalue ( window_name , object_name , row_index , column_index )
    if re . match ( row_text , value ) :
        return 1
except LdtpServerException :
    pass
return 0


signature:
def doesrowexist ( self , window_name , object_name , row_text , partial_match = False ) :
    """Verify table cell value with given text
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ text : Row text to match
 @ type string
 @ param partial _ match : Find partial match strings
 @ type boolean
 @ return : 1 on success 0 on failure .
 @ rtype : integer"""

implementation:
try :
    object_handle = self . _get_object_handle ( window_name , object_name )
    if not object_handle . AXEnabled :
        return 0
    for cell in object_handle . AXRows :
        if not partial_match and re . match ( row_text , cell . AXChildren [ 0 ] . AXValue ) :
            return 1
        elif partial_match and re . search ( row_text , cell . AXChildren [ 0 ] . AXValue ) :
            return 1
except LdtpServerException :
    pass
return 0


signature:
def verifypartialtablecell ( self , window_name , object_name , row_index , column_index , row_text ) :
    """Verify partial table cell value
 @ param window _ name : Window name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type window _ name : string
 @ param object _ name : Object name to type in , either full name ,
 LDTP ' s name convention , or a Unix glob .
 @ type object _ name : string
 @ param row _ index : Row index to get
 @ type row _ index : integer
 @ param column _ index : Column index to get , default value 0
 @ type column _ index : integer
 @ param row _ text : Row text to match
 @ type string
 @ return : 1 on success 0 on failure .
 @ rtype : integer"""

implementation:
try :
    value = getcellvalue ( window_name , object_name , row_index , column_index )
    if re . searchmatch ( row_text , value ) :
        return 1
except LdtpServerException :
    pass
return 0