signature:
def neurommsig_topology ( graph : BELGraph , nodes : List [ BaseEntity ] ) -> float :
    """Calculate the node neighbor score for a given list of nodes .
 - Doesn ' t consider self loops
 . . math : :
 \f rac { \ sum _ i ^ n N _ G [ i ] } { n * ( n - 1 ) }"""

implementation:
nodes = list ( nodes )
number_nodes = len ( nodes )
if number_nodes <= 1 : # log . debug ( ' ' )
    return 0.0
unnormalized_sum = sum ( u in graph [ v ] for u , v in itt . product ( nodes , repeat = 2 ) if v in graph and u != v )
return unnormalized_sum / ( number_nodes * ( number_nodes - 1.0 ) )


signature:
def bond_run ( perc_graph_result , seed , ps , convolution_factors_tasks ) :
    """Perform a single run ( realization ) over all microstates and return the
 canonical cluster statistics"""

implementation:
microcanonical_statistics = percolate . hpc . bond_microcanonical_statistics ( seed = seed , ** perc_graph_result )
# initialize statistics array
canonical_statistics = np . empty ( ps . size , dtype = percolate . hpc . canonical_statistics_dtype ( spanning_cluster = SPANNING_CLUSTER , ) )
# loop over all p ' s and convolve canonical statistics
# http : / / docs . scipy . org / doc / numpy / reference / arrays . nditer . html # modifying - array - values
for row , convolution_factors_task in zip ( np . nditer ( canonical_statistics , op_flags = [ 'writeonly' ] ) , convolution_factors_tasks , ) : # load task result
# http : / / jug . readthedocs . org / en / latest / api . html # jug . Task . load
    assert not convolution_factors_task . is_loaded ( )
    convolution_factors_task . load ( )
    # fetch task result
    my_convolution_factors = convolution_factors_task . result
    # convolve to canonical statistics
    row [ ... ] = percolate . hpc . bond_canonical_statistics ( microcanonical_statistics = microcanonical_statistics , convolution_factors = my_convolution_factors , spanning_cluster = SPANNING_CLUSTER , )
    # explicitly unload task to save memory
    # http : / / jug . readthedocs . org / en / latest / api . html # jug . Task . unload
    convolution_factors_task . unload ( )
# initialize canonical averages for reduce
ret = percolate . hpc . bond_initialize_canonical_averages ( canonical_statistics = canonical_statistics , spanning_cluster = SPANNING_CLUSTER , )
return ret


signature:
def bond_task ( perc_graph_result , seeds , ps , convolution_factors_tasks_iterator ) :
    """Perform a number of runs
 The number of runs is the number of seeds
 convolution _ factors _ tasks _ iterator needs to be an iterator
 We shield the convolution factors tasks from jug value / result mechanism
 by supplying an iterator to the list of tasks for lazy evaluation
 http : / / github . com / luispedro / jug / blob / 43f0d80a78f418fd3aa2b8705eaf7c4a5175fff7 / jug / task . py # L100
 http : / / github . com / luispedro / jug / blob / 43f0d80a78f418fd3aa2b8705eaf7c4a5175fff7 / jug / task . py # L455"""

implementation:
# restore the list of convolution factors tasks
convolution_factors_tasks = list ( convolution_factors_tasks_iterator )
return reduce ( percolate . hpc . bond_reduce , map ( bond_run , itertools . repeat ( perc_graph_result ) , seeds , itertools . repeat ( ps ) , itertools . repeat ( convolution_factors_tasks ) , ) )


signature:
def get_leaves_by_type ( graph , func = None , prune_threshold = 1 ) :
    """Returns an iterable over all nodes in graph ( in - place ) with only a connection to one node . Useful for gene and
 RNA . Allows for optional filter by function type .
 : param pybel . BELGraph graph : A BEL graph
 : param func : If set , filters by the node ' s function from : mod : ` pybel . constants ` like
 : data : ` pybel . constants . GENE ` , : data : ` pybel . constants . RNA ` , : data : ` pybel . constants . PROTEIN ` , or
 : data : ` pybel . constants . BIOPROCESS `
 : type func : str
 : param prune _ threshold : Removes nodes with less than or equal to this number of connections . Defaults to : code : ` 1 `
 : type prune _ threshold : int
 : return : An iterable over nodes with only a connection to one node
 : rtype : iter [ tuple ]"""

implementation:
for node , data in graph . nodes ( data = True ) :
    if func and func != data . get ( FUNCTION ) :
        continue
    if graph . in_degree ( node ) + graph . out_degree ( node ) <= prune_threshold :
        yield node


signature:
def get_peripheral_successor_edges ( graph : BELGraph , subgraph : BELGraph ) -> EdgeIterator :
    """Get the set of possible successor edges peripheral to the sub - graph .
 The source nodes in this iterable are all inside the sub - graph , while the targets are outside ."""

implementation:
for u in subgraph :
    for _ , v , k in graph . out_edges ( u , keys = True ) :
        if v not in subgraph :
            yield u , v , k


signature:
def get_peripheral_predecessor_edges ( graph : BELGraph , subgraph : BELGraph ) -> EdgeIterator :
    """Get the set of possible predecessor edges peripheral to the sub - graph .
 The target nodes in this iterable are all inside the sub - graph , while the sources are outside ."""

implementation:
for v in subgraph :
    for u , _ , k in graph . in_edges ( v , keys = True ) :
        if u not in subgraph :
            yield u , v , k


signature:
def count_sources ( edge_iter : EdgeIterator ) -> Counter :
    """Count the source nodes in an edge iterator with keys and data .
 : return : A counter of source nodes in the iterable"""

implementation: return Counter ( u for u , _ , _ in edge_iter )


signature:
def count_targets ( edge_iter : EdgeIterator ) -> Counter :
    """Count the target nodes in an edge iterator with keys and data .
 : return : A counter of target nodes in the iterable"""

implementation: return Counter ( v for _ , v , _ in edge_iter )


signature:
def get_subgraph_edges ( graph : BELGraph , annotation : str , value : str , source_filter = None , target_filter = None , ) :
    """Gets all edges from a given subgraph whose source and target nodes pass all of the given filters
 : param pybel . BELGraph graph : A BEL graph
 : param str annotation : The annotation to search
 : param str value : The annotation value to search by
 : param source _ filter : Optional filter for source nodes ( graph , node ) - > bool
 : param target _ filter : Optional filter for target nodes ( graph , node ) - > bool
 : return : An iterable of ( source node , target node , key , data ) for all edges that match the annotation / value and
 node filters
 : rtype : iter [ tuple ]"""

implementation:
if source_filter is None :
    source_filter = keep_node_permissive
if target_filter is None :
    target_filter = keep_node_permissive
for u , v , k , data in graph . edges ( keys = True , data = True ) :
    if not edge_has_annotation ( data , annotation ) :
        continue
    if data [ ANNOTATIONS ] [ annotation ] == value and source_filter ( graph , u ) and target_filter ( graph , v ) :
        yield u , v , k , data


signature:
def get_subgraph_peripheral_nodes ( graph : BELGraph , subgraph : Iterable [ BaseEntity ] , node_predicates : NodePredicates = None , edge_predicates : EdgePredicates = None , ) :
    """Get a summary dictionary of all peripheral nodes to a given sub - graph .
 : return : A dictionary of { external node : { ' successor ' : { internal node : list of ( key , dict ) } ,
 ' predecessor ' : { internal node : list of ( key , dict ) } } }
 : rtype : dict
 For example , it might be useful to quantify the number of predecessors and successors :
 > > > from pybel . struct . filters import exclude _ pathology _ filter
 > > > value = ' Blood vessel dilation subgraph '
 > > > sg = get _ subgraph _ by _ annotation _ value ( graph , annotation = ' Subgraph ' , value = value )
 > > > p = get _ subgraph _ peripheral _ nodes ( graph , sg , node _ predicates = exclude _ pathology _ filter )
 > > > for node in sorted ( p , key = lambda n : len ( set ( p [ n ] [ ' successor ' ] ) | set ( p [ n ] [ ' predecessor ' ] ) ) , reverse = True ) :
 > > > if 1 = = len ( p [ value ] [ node ] [ ' successor ' ] ) or 1 = = len ( p [ value ] [ node ] [ ' predecessor ' ] ) :
 > > > continue
 > > > print ( node ,
 > > > len ( p [ node ] [ ' successor ' ] ) ,
 > > > len ( p [ node ] [ ' predecessor ' ] ) ,
 > > > len ( set ( p [ node ] [ ' successor ' ] ) | set ( p [ node ] [ ' predecessor ' ] ) ) )"""

implementation:
node_filter = concatenate_node_predicates ( node_predicates = node_predicates )
edge_filter = and_edge_predicates ( edge_predicates = edge_predicates )
result = defaultdict ( lambda : defaultdict ( lambda : defaultdict ( list ) ) )
for u , v , k , d in get_peripheral_successor_edges ( graph , subgraph ) :
    if not node_filter ( graph , v ) or not node_filter ( graph , u ) or not edge_filter ( graph , u , v , k ) :
        continue
    result [ v ] [ 'predecessor' ] [ u ] . append ( ( k , d ) )
for u , v , k , d in get_peripheral_predecessor_edges ( graph , subgraph ) :
    if not node_filter ( graph , v ) or not node_filter ( graph , u ) or not edge_filter ( graph , u , v , k ) :
        continue
    result [ u ] [ 'successor' ] [ v ] . append ( ( k , d ) )
return result


signature:
def expand_periphery ( universe : BELGraph , graph : BELGraph , node_predicates : NodePredicates = None , edge_predicates : EdgePredicates = None , threshold : int = 2 , ) -> None :
    """Iterates over all possible edges , peripheral to a given subgraph , that could be added from the given graph .
 Edges could be added if they go to nodes that are involved in relationships that occur with more than the
 threshold ( default 2 ) number of nodes in the subgraph .
 : param universe : The universe of BEL knowledge
 : param graph : The ( sub ) graph to expand
 : param threshold : Minimum frequency of betweenness occurrence to add a gap node
 A reasonable edge filter to use is : func : ` pybel _ tools . filters . keep _ causal _ edges ` because this function can allow
 for huge expansions if there happen to be hub nodes ."""

implementation:
nd = get_subgraph_peripheral_nodes ( universe , graph , node_predicates = node_predicates , edge_predicates = edge_predicates )
for node , dd in nd . items ( ) :
    pred_d = dd [ 'predecessor' ]
    succ_d = dd [ 'successor' ]
    in_subgraph_connections = set ( pred_d ) | set ( succ_d )
    if threshold > len ( in_subgraph_connections ) :
        continue
    graph . add_node ( node , attr_dict = universe [ node ] )
    for u , edges in pred_d . items ( ) :
        for k , d in edges :
            safe_add_edge ( graph , u , node , k , d )
    for v , edges in succ_d . items ( ) :
        for k , d in edges :
            safe_add_edge ( graph , node , v , k , d )


signature:
def enrich_complexes ( graph : BELGraph ) -> None :
    """Add all of the members of the complex abundances to the graph ."""

implementation:
nodes = list ( get_nodes_by_function ( graph , COMPLEX ) )
for u in nodes :
    for v in u . members :
        graph . add_has_component ( u , v )


signature:
def enrich_composites ( graph : BELGraph ) :
    """Adds all of the members of the composite abundances to the graph ."""

implementation:
nodes = list ( get_nodes_by_function ( graph , COMPOSITE ) )
for u in nodes :
    for v in u . members :
        graph . add_has_component ( u , v )


signature:
def enrich_reactions ( graph : BELGraph ) :
    """Adds all of the reactants and products of reactions to the graph ."""

implementation:
nodes = list ( get_nodes_by_function ( graph , REACTION ) )
for u in nodes :
    for v in u . reactants :
        graph . add_has_reactant ( u , v )
    for v in u . products :
        graph . add_has_product ( u , v )


signature:
def enrich_variants ( graph : BELGraph , func : Union [ None , str , Iterable [ str ] ] = None ) :
    """Add the reference nodes for all variants of the given function .
 : param graph : The target BEL graph to enrich
 : param func : The function by which the subject of each triple is filtered . Defaults to the set of protein , rna ,
 mirna , and gene ."""

implementation:
if func is None :
    func = { PROTEIN , RNA , MIRNA , GENE }
nodes = list ( get_nodes_by_function ( graph , func ) )
for u in nodes :
    parent = u . get_parent ( )
    if parent is None :
        continue
    if parent not in graph :
        graph . add_has_variant ( parent , u )


signature:
def enrich_unqualified ( graph : BELGraph ) :
    """Enrich the sub - graph with the unqualified edges from the graph .
 The reason you might want to do this is you induce a sub - graph from the original graph based on an annotation
 filter , but the unqualified edges that don ' t have annotations that most likely connect elements within your graph
 are not included .
 . . seealso : :
 This function thinly wraps the successive application of the following functions :
 - : func : ` enrich _ complexes `
 - : func : ` enrich _ composites `
 - : func : ` enrich _ reactions `
 - : func : ` enrich _ variants `
 Equivalent to :
 > > > enrich _ complexes ( graph )
 > > > enrich _ composites ( graph )
 > > > enrich _ reactions ( graph )
 > > > enrich _ variants ( graph )"""

implementation:
enrich_complexes ( graph )
enrich_composites ( graph )
enrich_reactions ( graph )
enrich_variants ( graph )


signature:
def expand_internal ( universe : BELGraph , graph : BELGraph , edge_predicates : EdgePredicates = None ) -> None :
    """Edges between entities in the sub - graph that pass the given filters .
 : param universe : The full graph
 : param graph : A sub - graph to find the upstream information
 : param edge _ predicates : Optional list of edge filter functions ( graph , node , node , key , data ) - > bool"""

implementation:
edge_filter = and_edge_predicates ( edge_predicates )
for u , v in itt . product ( graph , repeat = 2 ) :
    if graph . has_edge ( u , v ) or not universe . has_edge ( u , v ) :
        continue
    rs = defaultdict ( list )
    for key , data in universe [ u ] [ v ] . items ( ) :
        if not edge_filter ( universe , u , v , key ) :
            continue
        rs [ data [ RELATION ] ] . append ( ( key , data ) )
    if 1 == len ( rs ) :
        relation = list ( rs ) [ 0 ]
        for key , data in rs [ relation ] :
            graph . add_edge ( u , v , key = key , ** data )
    else :
        log . debug ( 'Multiple relationship types found between %s and %s' , u , v )


signature:
def expand_internal_causal ( universe : BELGraph , graph : BELGraph ) -> None :
    """Add causal edges between entities in the sub - graph .
 Is an extremely thin wrapper around : func : ` expand _ internal ` .
 : param universe : A BEL graph representing the universe of all knowledge
 : param graph : The target BEL graph to enrich with causal relations between contained nodes
 Equivalent to :
 > > > from pybel _ tools . mutation import expand _ internal
 > > > from pybel . struct . filters . edge _ predicates import is _ causal _ relation
 > > > expand _ internal ( universe , graph , edge _ predicates = is _ causal _ relation )"""

implementation: expand_internal ( universe , graph , edge_predicates = is_causal_relation )


signature:
def get_namespaces_with_incorrect_names ( graph : BELGraph ) -> Set [ str ] :
    """Return the set of all namespaces with incorrect names in the graph ."""

implementation: return { exc . namespace for _ , exc , _ in graph . warnings if isinstance ( exc , ( MissingNamespaceNameWarning , MissingNamespaceRegexWarning ) ) }


signature:
def get_undefined_namespaces ( graph : BELGraph ) -> Set [ str ] :
    """Get all namespaces that are used in the BEL graph aren ' t actually defined ."""

implementation: return { exc . namespace for _ , exc , _ in graph . warnings if isinstance ( exc , UndefinedNamespaceWarning ) }


signature:
def get_incorrect_names_by_namespace ( graph : BELGraph , namespace : str ) -> Set [ str ] :
    """Return the set of all incorrect names from the given namespace in the graph .
 : return : The set of all incorrect names from the given namespace in the graph"""

implementation: return { exc . name for _ , exc , _ in graph . warnings if isinstance ( exc , ( MissingNamespaceNameWarning , MissingNamespaceRegexWarning ) ) and exc . namespace == namespace }


signature:
def get_undefined_namespace_names ( graph : BELGraph , namespace : str ) -> Set [ str ] :
    """Get the names from a namespace that wasn ' t actually defined .
 : return : The set of all names from the undefined namespace"""

implementation: return { exc . name for _ , exc , _ in graph . warnings if isinstance ( exc , UndefinedNamespaceWarning ) and exc . namespace == namespace }


signature:
def get_incorrect_names ( graph : BELGraph ) -> Mapping [ str , Set [ str ] ] :
    """Return the dict of the sets of all incorrect names from the given namespace in the graph .
 : return : The set of all incorrect names from the given namespace in the graph"""

implementation: return { namespace : get_incorrect_names_by_namespace ( graph , namespace ) for namespace in get_namespaces ( graph ) }


signature:
def get_undefined_annotations ( graph : BELGraph ) -> Set [ str ] :
    """Get all annotations that aren ' t actually defined .
 : return : The set of all undefined annotations"""

implementation: return { exc . annotation for _ , exc , _ in graph . warnings if isinstance ( exc , UndefinedAnnotationWarning ) }


signature:
def calculate_incorrect_name_dict ( graph : BELGraph ) -> Mapping [ str , str ] :
    """Group all of the incorrect identifiers in a dict of { namespace : list of erroneous names } .
 : return : A dictionary of { namespace : list of erroneous names }"""

implementation:
missing = defaultdict ( list )
for _ , e , ctx in graph . warnings :
    if not isinstance ( e , ( MissingNamespaceNameWarning , MissingNamespaceRegexWarning ) ) :
        continue
    missing [ e . namespace ] . append ( e . name )
return dict ( missing )


signature:
def group_errors ( graph : BELGraph ) -> Mapping [ str , List [ int ] ] :
    """Group the errors together for analysis of the most frequent error .
 : return : A dictionary of { error string : list of line numbers }"""

implementation:
warning_summary = defaultdict ( list )
for _ , exc , _ in graph . warnings :
    warning_summary [ str ( exc ) ] . append ( exc . line_number )
return dict ( warning_summary )


signature:
def get_most_common_errors ( graph : BELGraph , n : Optional [ int ] = 20 ) :
    """Get the ( n ) most common errors in a graph ."""

implementation: return count_dict_values ( group_errors ( graph ) ) . most_common ( n )


signature:
def get_names_including_errors_by_namespace ( graph : BELGraph , namespace : str ) -> Set [ str ] :
    """Takes the names from the graph in a given namespace ( : func : ` pybel . struct . summary . get _ names _ by _ namespace ` ) and
 the erroneous names from the same namespace ( : func : ` get _ incorrect _ names _ by _ namespace ` ) and returns them together
 as a unioned set
 : return : The set of all correct and incorrect names from the given namespace in the graph"""

implementation: return get_names_by_namespace ( graph , namespace ) | get_incorrect_names_by_namespace ( graph , namespace )


signature:
def get_names_including_errors ( graph : BELGraph ) -> Mapping [ str , Set [ str ] ] :
    """Takes the names from the graph in a given namespace and the erroneous names from the same namespace and returns
 them together as a unioned set
 : return : The dict of the sets of all correct and incorrect names from the given namespace in the graph"""

implementation: return { namespace : get_names_including_errors_by_namespace ( graph , namespace ) for namespace in get_namespaces ( graph ) }


signature:
def pairwise ( iterable : Iterable [ X ] ) -> Iterable [ Tuple [ X , X ] ] :
    """Iterate over pairs in list s - > ( s0 , s1 ) , ( s1 , s2 ) , ( s2 , s3 ) , . . ."""

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


signature:
def count_defaultdict ( dict_of_lists : Mapping [ X , List [ Y ] ] ) -> Mapping [ X , typing . Counter [ Y ] ] :
    """Count the number of elements in each value of the dictionary ."""

implementation: return { k : Counter ( v ) for k , v in dict_of_lists . items ( ) }


signature:
def count_dict_values ( dict_of_counters : Mapping [ X , Sized ] ) -> typing . Counter [ X ] :
    """Count the number of elements in each value ( can be list , Counter , etc ) .
 : param dict _ of _ counters : A dictionary of things whose lengths can be measured ( lists , Counters , dicts )
 : return : A Counter with the same keys as the input but the count of the length of the values list / tuple / set / Counter"""

implementation: return Counter ( { k : len ( v ) for k , v in dict_of_counters . items ( ) } )


signature:
def set_percentage ( x : Iterable [ X ] , y : Iterable [ X ] ) -> float :
    """What percentage of x is contained within y ?
 : param set x : A set
 : param set y : Another set
 : return : The percentage of x contained within y"""

implementation:
a , b = set ( x ) , set ( y )
if not a :
    return 0.0
return len ( a & b ) / len ( a )


signature:
def tanimoto_set_similarity ( x : Iterable [ X ] , y : Iterable [ X ] ) -> float :
    """Calculate the tanimoto set similarity ."""

implementation:
a , b = set ( x ) , set ( y )
union = a | b
if not union :
    return 0.0
return len ( a & b ) / len ( union )


signature:
def min_tanimoto_set_similarity ( x : Iterable [ X ] , y : Iterable [ X ] ) -> float :
    """Calculate the tanimoto set similarity using the minimum size .
 : param set x : A set
 : param set y : Another set
 : return : The similarity between"""

implementation:
a , b = set ( x ) , set ( y )
if not a or not b :
    return 0.0
return len ( a & b ) / min ( len ( a ) , len ( b ) )


signature:
def calculate_single_tanimoto_set_distances ( target : Iterable [ X ] , dict_of_sets : Mapping [ Y , Set [ X ] ] ) -> Mapping [ Y , float ] :
    """Return a dictionary of distances keyed by the keys in the given dict .
 Distances are calculated based on pairwise tanimoto similarity of the sets contained
 : param set target : A set
 : param dict _ of _ sets : A dict of { x : set of y }
 : type dict _ of _ sets : dict
 : return : A similarity dicationary based on the set overlap ( tanimoto ) score between the target set and the sets in
 dos
 : rtype : dict"""

implementation:
target_set = set ( target )
return { k : tanimoto_set_similarity ( target_set , s ) for k , s in dict_of_sets . items ( ) }


signature:
def calculate_tanimoto_set_distances ( dict_of_sets : Mapping [ X , Set ] ) -> Mapping [ X , Mapping [ X , float ] ] :
    """Return a distance matrix keyed by the keys in the given dict .
 Distances are calculated based on pairwise tanimoto similarity of the sets contained .
 : param dict _ of _ sets : A dict of { x : set of y }
 : return : A similarity matrix based on the set overlap ( tanimoto ) score between each x as a dict of dicts"""

implementation:
result : Dict [ X , Dict [ X , float ] ] = defaultdict ( dict )
for x , y in itt . combinations ( dict_of_sets , 2 ) :
    result [ x ] [ y ] = result [ y ] [ x ] = tanimoto_set_similarity ( dict_of_sets [ x ] , dict_of_sets [ y ] )
for x in dict_of_sets :
    result [ x ] [ x ] = 1.0
return dict ( result )


signature:
def calculate_global_tanimoto_set_distances ( dict_of_sets : Mapping [ X , Set ] ) -> Mapping [ X , Mapping [ X , float ] ] :
    r"""Calculate an alternative distance matrix based on the following equation .
 . . math : : distance ( A , B ) = 1 - \ | A \ cup B \ | / \ | \ cup _ { s \ in S } s \ |
 : param dict _ of _ sets : A dict of { x : set of y }
 : return : A similarity matrix based on the alternative tanimoto distance as a dict of dicts"""

implementation:
universe = set ( itt . chain . from_iterable ( dict_of_sets . values ( ) ) )
universe_size = len ( universe )
result : Dict [ X , Dict [ X , float ] ] = defaultdict ( dict )
for x , y in itt . combinations ( dict_of_sets , 2 ) :
    result [ x ] [ y ] = result [ y ] [ x ] = 1.0 - len ( dict_of_sets [ x ] | dict_of_sets [ y ] ) / universe_size
for x in dict_of_sets :
    result [ x ] [ x ] = 1.0 - len ( x ) / universe_size
return dict ( result )


signature:
def barh ( d , plt , title = None ) :
    """A convenience function for plotting a horizontal bar plot from a Counter"""

implementation:
labels = sorted ( d , key = d . get )
index = range ( len ( labels ) )
plt . yticks ( index , labels )
plt . barh ( index , [ d [ v ] for v in labels ] )
if title is not None :
    plt . title ( title )


signature:
def barv ( d , plt , title = None , rotation = 'vertical' ) :
    """A convenience function for plotting a vertical bar plot from a Counter"""

implementation:
labels = sorted ( d , key = d . get , reverse = True )
index = range ( len ( labels ) )
plt . xticks ( index , labels , rotation = rotation )
plt . bar ( index , [ d [ v ] for v in labels ] )
if title is not None :
    plt . title ( title )


signature:
def safe_add_edge ( graph , u , v , key , attr_dict , ** attr ) :
    """Adds an edge while preserving negative keys , and paying no respect to positive ones
 : param pybel . BELGraph graph : A BEL Graph
 : param tuple u : The source BEL node
 : param tuple v : The target BEL node
 : param int key : The edge key . If less than zero , corresponds to an unqualified edge , else is disregarded
 : param dict attr _ dict : The edge data dictionary
 : param dict attr : Edge data to assign via keyword arguments"""

implementation:
if key < 0 :
    graph . add_edge ( u , v , key = key , attr_dict = attr_dict , ** attr )
else :
    graph . add_edge ( u , v , attr_dict = attr_dict , ** attr )


signature:
def prepare_c3 ( data : Union [ List [ Tuple [ str , int ] ] , Mapping [ str , int ] ] , y_axis_label : str = 'y' , x_axis_label : str = 'x' , ) -> str :
    """Prepares C3 JSON for making a bar chart from a Counter
 : param data : A dictionary of { str : int } to display as bar chart
 : param y _ axis _ label : The Y axis label
 : param x _ axis _ label : X axis internal label . Should be left as default ' x ' )
 : return : A JSON dictionary for making a C3 bar chart"""

implementation:
if not isinstance ( data , list ) :
    data = sorted ( data . items ( ) , key = itemgetter ( 1 ) , reverse = True )
try :
    labels , values = zip ( * data )
except ValueError :
    log . info ( f'no values found for {x_axis_label}, {y_axis_label}' )
    labels , values = [ ] , [ ]
return json . dumps ( [ [ x_axis_label ] + list ( labels ) , [ y_axis_label ] + list ( values ) , ] )


signature:
def prepare_c3_time_series ( data : List [ Tuple [ int , int ] ] , y_axis_label : str = 'y' , x_axis_label : str = 'x' ) -> str :
    """Prepare C3 JSON string dump for a time series .
 : param data : A list of tuples [ ( year , count ) ]
 : param y _ axis _ label : The Y axis label
 : param x _ axis _ label : X axis internal label . Should be left as default ' x ' )"""

implementation:
years , counter = zip ( * data )
years = [ datetime . date ( year , 1 , 1 ) . isoformat ( ) for year in years ]
return json . dumps ( [ [ x_axis_label ] + list ( years ) , [ y_axis_label ] + list ( counter ) ] )


signature:
def calculate_betweenness_centality ( graph : BELGraph , number_samples : int = CENTRALITY_SAMPLES ) -> Counter :
    """Calculate the betweenness centrality over nodes in the graph .
 Tries to do it with a certain number of samples , but then tries a complete approach if it fails ."""

implementation:
try :
    res = nx . betweenness_centrality ( graph , k = number_samples )
except Exception :
    res = nx . betweenness_centrality ( graph )
return Counter ( res )


signature:
def get_circulations ( elements : T ) -> Iterable [ T ] :
    """Iterate over all possible circulations of an ordered collection ( tuple or list ) .
 Example :
 > > > list ( get _ circulations ( [ 1 , 2 , 3 ] ) )
 [ [ 1 , 2 , 3 ] , [ 2 , 3 , 1 ] , [ 3 , 1 , 2 ] ]"""

implementation:
for i in range ( len ( elements ) ) :
    yield elements [ i : ] + elements [ : i ]


signature:
def canonical_circulation ( elements : T , key : Optional [ Callable [ [ T ] , bool ] ] = None ) -> T :
    """Get get a canonical representation of the ordered collection by finding its minimum circulation with the
 given sort key"""

implementation: return min ( get_circulations ( elements ) , key = key )


signature:
def pair_has_contradiction ( graph : BELGraph , u : BaseEntity , v : BaseEntity ) -> bool :
    """Check if a pair of nodes has any contradictions in their causal relationships .
 Assumes both nodes are in the graph ."""

implementation:
relations = { data [ RELATION ] for data in graph [ u ] [ v ] . values ( ) }
return relation_set_has_contradictions ( relations )


signature:
def relation_set_has_contradictions ( relations : Set [ str ] ) -> bool :
    """Return if the set of BEL relations contains a contradiction ."""

implementation:
has_increases = any ( relation in CAUSAL_INCREASE_RELATIONS for relation in relations )
has_decreases = any ( relation in CAUSAL_DECREASE_RELATIONS for relation in relations )
has_cnc = any ( relation == CAUSES_NO_CHANGE for relation in relations )
return 1 < sum ( [ has_cnc , has_decreases , has_increases ] )


signature:
def percolation_graph ( graph , spanning_cluster = True ) :
    """Prepare the ( internal ) percolation graph from a given graph
 Helper function to prepare the given graph for spanning cluster detection
 ( if required ) .
 Basically it strips off the auxiliary nodes and edges again .
 It also returns fundamental graph quantitities ( number of nodes and edges ) .
 Parameters
 graph
 spanning _ cluster
 Returns
 ret : tuple"""

implementation:
ret = dict ( )
ret [ 'graph' ] = graph
ret [ 'spanning_cluster' ] = bool ( spanning_cluster )
# initialize percolation graph
if spanning_cluster :
    spanning_auxiliary_node_attributes = nx . get_node_attributes ( graph , 'span' )
    ret [ 'auxiliary_node_attributes' ] = spanning_auxiliary_node_attributes
    auxiliary_nodes = spanning_auxiliary_node_attributes . keys ( )
    if not list ( auxiliary_nodes ) :
        raise ValueError ( 'Spanning cluster is to be detected, but no auxiliary nodes ' 'given.' )
    spanning_sides = list ( set ( spanning_auxiliary_node_attributes . values ( ) ) )
    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.' )
    ret [ 'spanning_sides' ] = spanning_sides
    ret [ 'auxiliary_edge_attributes' ] = nx . get_edge_attributes ( graph , 'span' )
# get subgraph on which percolation is to take place ( strip off the
# auxiliary nodes )
if spanning_cluster :
    perc_graph = graph . subgraph ( [ node for node in graph . nodes_iter ( ) if 'span' not in graph . node [ node ] ] )
else :
    perc_graph = graph
ret [ 'perc_graph' ] = perc_graph
# number of nodes N
ret [ 'num_nodes' ] = nx . number_of_nodes ( perc_graph )
# number of edges M
ret [ 'num_edges' ] = nx . number_of_edges ( perc_graph )
return ret


signature:
def sample_states ( graph , spanning_cluster = True , model = 'bond' , copy_result = True ) :
    '''Generate successive sample states of the 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 .
 Parameters
 graph : networkx . Graph
 The substrate graph on which percolation is to take place
 spanning _ cluster : bool , optional
 Whether to detect a spanning cluster or not .
 Defaults to ` ` True ` ` .
 model : str , optional
 The percolation model ( either ` ` ' bond ' ` ` or ` ` ' site ' ` ` ) .
 Defaults to ` ` ' bond ' ` ` .
 . . note : : Other models than ` ` ' bond ' ` ` are not supported yet .
 copy _ result : bool , optional
 Whether to return a copy or a reference to the result dictionary .
 Defaults to ` ` True ` ` .
 Yields
 ret : dict
 Cluster statistics
 ret [ ' n ' ] : int
 Number of occupied bonds
 ret [ ' N ' ] : int
 Total number of sites
 ret [ ' M ' ] : int
 Total number of bonds
 ret [ ' has _ spanning _ cluster ' ] : 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 ` model ` does not equal ` ` ' bond ' ` ` .
 ValueError
 If ` spanning _ cluster ` is ` ` True ` ` , but ` graph ` does not contain any
 auxiliary nodes to detect spanning clusters .
 See also
 microcanonical _ averages : Evolves multiple sample states in parallel
 Notes
 Iterating through this generator is a single run of the Newman - Ziff
 algorithm . [ 2 ] _
 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 [ 2 ] _ , 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 ` . [ 3 ] _
 The primed sum : math : ` \ sum ' ` signifies that the largest cluster is
 excluded from the sum . [ 4 ] _
 References
 . . [ 2 ] 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 > ` _ .
 . . [ 3 ] Stauffer , D . & Aharony , A . Introduction to Percolation Theory ( Taylor &
 Francis , London , 1994 ) , second edn .
 . . [ 4 ] 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:
if model != 'bond' :
    raise ValueError ( 'Only bond percolation supported.' )
if spanning_cluster :
    auxiliary_node_attributes = nx . get_node_attributes ( graph , 'span' )
    auxiliary_nodes = auxiliary_node_attributes . keys ( )
    if not list ( auxiliary_nodes ) :
        raise ValueError ( 'Spanning cluster is to be detected, but no auxiliary nodes ' 'given.' )
    spanning_sides = list ( set ( auxiliary_node_attributes . values ( ) ) )
    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.' )
    auxiliary_edge_attributes = nx . get_edge_attributes ( graph , 'span' )
# get subgraph on which percolation is to take place ( strip off the
# auxiliary nodes )
if spanning_cluster :
    perc_graph = graph . subgraph ( [ node for node in graph . nodes_iter ( ) if 'span' not in graph . node [ node ] ] )
else :
    perc_graph = graph
# get a list of edges for easy access in later iterations
perc_edges = perc_graph . edges ( )
# number of nodes N
num_nodes = nx . number_of_nodes ( perc_graph )
# number of edges M
num_edges = nx . number_of_edges ( perc_graph )
# initial iteration : no edges added yet ( n = = 0)
ret = dict ( )
ret [ 'n' ] = 0
ret [ 'N' ] = num_nodes
ret [ 'M' ] = num_edges
ret [ 'max_cluster_size' ] = 1
ret [ 'moments' ] = np . ones ( 5 ) * ( num_nodes - 1 )
if spanning_cluster :
    ret [ 'has_spanning_cluster' ] = False
if copy_result :
    yield copy . deepcopy ( ret )
else :
    yield ret
# permute edges
perm_edges = np . random . permutation ( num_edges )
# 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' ] = n + 1
    # draw new edge from permutation
    edge_index = perm_edges [ n ]
    edge = perc_edges [ edge_index ]
    ret [ 'edge' ] = edge
    # 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 )
        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 )
            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 )
                ret [ 'max_cluster_size' ] = weight
    if copy_result :
        yield copy . deepcopy ( ret )
    else :
        yield ret


signature:
def single_run_arrays ( spanning_cluster = True , ** kwargs ) :
    r'''Generate statistics for a single run
 This is a stand - alone helper function to evolve a single sample state
 ( realization ) and return the cluster statistics .
 Parameters
 spanning _ cluster : bool , optional
 Whether to detect a spanning cluster or not .
 Defaults to ` ` True ` ` .
 kwargs : keyword arguments
 Piped through to : func : ` sample _ states `
 Returns
 ret : dict
 Cluster statistics
 ret [ ' N ' ] : int
 Total number of sites
 ret [ ' M ' ] : int
 Total number of bonds
 ret [ ' max _ cluster _ size ' ] : 1 - D : py : class : ` numpy . ndarray ` of int , size ` ` ret [ ' M ' ] + 1 ` `
 Array of the sizes of the largest cluster ( absolute number of sites ) at
 the respective occupation number .
 ret [ ' has _ spanning _ cluster ' ] : 1 - D : py : class : ` numpy . ndarray ` of bool , size ` ` ret [ ' M ' ] + 1 ` `
 Array of booleans for each occupation number .
 The respective entry is ` ` True ` ` if there is a spanning cluster ,
 ` ` False ` ` otherwise .
 Only exists if ` spanning _ cluster ` argument is set to ` ` True ` ` .
 ret [ ' moments ' ] : 2 - D : py : class : ` numpy . ndarray ` of int
 Array of shape ` ` ( 5 , ret [ ' M ' ] + 1 ) ` ` .
 The ` ` ( k , m ) ` ` - th entry is the ` ` k ` ` - th raw moment of the ( absolute )
 cluster size distribution , with ` ` k ` ` ranging from ` ` 0 ` ` to ` ` 4 ` ` , at
 occupation number ` ` m ` ` .
 See Also
 sample _ states'''

implementation:
# initial iteration
# we do not need a copy of the result dictionary since we copy the values
# anyway
kwargs [ 'copy_result' ] = False
ret = dict ( )
for n , state in enumerate ( sample_states ( spanning_cluster = spanning_cluster , ** kwargs ) ) : # merge cluster statistics
    if 'N' in ret :
        assert ret [ 'N' ] == state [ 'N' ]
    else :
        ret [ 'N' ] = state [ 'N' ]
    if 'M' in ret :
        assert ret [ 'M' ] == state [ 'M' ]
    else :
        ret [ 'M' ] = state [ 'M' ]
        number_of_states = state [ 'M' ] + 1
        max_cluster_size = np . empty ( number_of_states )
        if spanning_cluster :
            has_spanning_cluster = np . empty ( number_of_states , dtype = np . bool )
        moments = np . empty ( ( 5 , number_of_states ) )
    max_cluster_size [ n ] = state [ 'max_cluster_size' ]
    for k in range ( 5 ) :
        moments [ k , n ] = state [ 'moments' ] [ k ]
    if spanning_cluster :
        has_spanning_cluster [ n ] = state [ 'has_spanning_cluster' ]
ret [ 'max_cluster_size' ] = max_cluster_size
ret [ 'moments' ] = moments
if spanning_cluster :
    ret [ 'has_spanning_cluster' ] = has_spanning_cluster
return ret


signature:
def _microcanonical_average_spanning_cluster ( has_spanning_cluster , alpha ) :
    r'''Compute the average number of runs that have a spanning cluster
 Helper function for : func : ` microcanonical _ averages `
 Parameters
 has _ spanning _ cluster : 1 - D : py : class : ` numpy . ndarray ` of bool
 Each entry is the ` ` has _ spanning _ cluster ` ` field of the output of
 : func : ` sample _ states ` :
 An entry is ` ` True ` ` if there is a spanning cluster in that respective
 run , and ` ` False ` ` otherwise .
 alpha : float
 Significance level .
 Returns
 ret : dict
 Spanning cluster statistics
 ret [ ' spanning _ cluster ' ] : float
 The average relative number ( Binomial proportion ) of runs that have a
 spanning cluster .
 This is the Bayesian point estimate of the posterior mean , with a
 uniform prior .
 ret [ ' spanning _ cluster _ ci ' ] : 1 - D : py : class : ` numpy . ndarray ` of float , size 2
 The lower and upper bounds of the Binomial proportion confidence
 interval with uniform prior .
 See Also
 sample _ states : spanning cluster detection
 microcanonical _ averages : spanning cluster statistics
 Notes
 Averages and confidence intervals for Binomial proportions
 As Cameron [ 8 ] _ puts it , the normal approximation to the confidence
 interval for a Binomial proportion : math : ` p ` " suffers a * systematic *
 decline in performance ( . . . ) towards extreme values of : math : ` p ` near
 : math : ` 0 ` and : math : ` 1 ` , generating binomial [ confidence intervals ]
 with effective coverage far below the desired level . " ( see also
 References [ 6 ] _ and [ 7 ] _ ) .
 A different approach to quantifying uncertainty is Bayesian inference .
 [5 ] _
 For : math : ` n ` independent Bernoulli trails with common success
 probability : math : ` p ` , the * likelihood * to have : math : ` k ` successes
 given : math : ` p ` is the binomial distribution
 . . math : :
 P ( k | p ) = \ binom { n } { k } p ^ k ( 1 - p ) ^ { n - k } \ equiv B ( a , b ) ,
 where : math : ` B ( a , b ) ` is the * Beta distribution * with parameters
 : math : ` a = k + 1 ` and : math : ` b = n - k + 1 ` .
 Assuming a uniform prior : math : ` P ( p ) = 1 ` , the * posterior * is [ 5 ] _
 . . math : :
 P ( p | k ) = P ( k | p ) = B ( a , b ) .
 A point estimate is the posterior mean
 . . math : :
 \ bar { p } = \ frac { k + 1 } { n + 2}
 with the : math : ` 1 - \ alpha ` credible interval : math : ` ( p _ l , p _ u ) ` given
 by
 . . math : :
 \ int _ 0 ^ { p _ l } dp B ( a , b ) = \ int _ { p _ u } ^ 1 dp B ( a , b ) = \ frac { \ alpha } { 2 } .
 References
 . . [ 5 ] Wasserman , L . All of Statistics ( Springer New York , 2004 ) ,
 ` doi : 10.1007/978-0-387-21736-9 < http : / / dx . doi . org / 10.1007/978-0-387-21736-9 > ` _ .
 . . [ 6 ] DasGupta , A . , Cai , T . T . & Brown , L . D . Interval Estimation for a
 Binomial Proportion . Statistical Science 16 , 101-133 ( 2001 ) .
 ` doi : 10.1214 / ss / 1009213286 < http : / / dx . doi . org / 10.1214 / ss / 1009213286 > ` _ .
 . . [ 7 ] Agresti , A . & Coull , B . A . Approximate is Better than " Exact " for
 Interval Estimation of Binomial Proportions . The American Statistician
 52 , 119-126 ( 1998 ) ,
 ` doi : 10.2307/2685469 < http : / / dx . doi . org / 10.2307/2685469 > ` _ .
 . . [ 8 ] Cameron , E . On the Estimation of Confidence Intervals for Binomial
 Population Proportions in Astronomy : The Simplicity and Superiority of
 the Bayesian Approach . Publications of the Astronomical Society of
 Australia 28 , 128-139 ( 2011 ) ,
 ` doi : 10.1071 / as10046 < http : / / dx . doi . org / 10.1071 / as10046 > ` _ .'''

implementation:
ret = dict ( )
runs = has_spanning_cluster . size
# Bayesian posterior mean for Binomial proportion ( uniform prior )
k = has_spanning_cluster . sum ( dtype = np . float )
ret [ 'spanning_cluster' ] = ( ( k + 1 ) / ( runs + 2 ) )
# Bayesian credible interval for Binomial proportion ( uniform
# prior )
ret [ 'spanning_cluster_ci' ] = scipy . stats . beta . ppf ( [ alpha / 2 , 1 - alpha / 2 ] , k + 1 , runs - k + 1 )
return ret


signature:
def _microcanonical_average_max_cluster_size ( max_cluster_size , alpha ) :
    """Compute the average size of the largest cluster
 Helper function for : func : ` microcanonical _ averages `
 Parameters
 max _ cluster _ size : 1 - D : py : class : ` numpy . ndarray ` of int
 Each entry is the ` ` max _ cluster _ size ` ` field of the output of
 : func : ` sample _ states ` :
 The size of the largest cluster ( absolute number of sites ) .
 alpha : float
 Significance level .
 Returns
 ret : dict
 Largest cluster statistics
 ret [ ' max _ cluster _ size ' ] : float
 Average size of the largest cluster ( absolute number of sites )
 ret [ ' max _ cluster _ size _ ci ' ] : 1 - D : py : class : ` numpy . ndarray ` of float , size 2
 Lower and upper bounds of the normal confidence interval of the average
 size of the largest cluster ( absolute number of sites )
 See Also
 sample _ states : largest cluster detection
 microcanonical _ averages : largest cluster statistics"""

implementation:
ret = dict ( )
runs = max_cluster_size . size
sqrt_n = np . sqrt ( runs )
max_cluster_size_sample_mean = max_cluster_size . mean ( )
ret [ 'max_cluster_size' ] = max_cluster_size_sample_mean
max_cluster_size_sample_std = max_cluster_size . std ( ddof = 1 )
if max_cluster_size_sample_std :
    old_settings = np . seterr ( all = 'raise' )
    ret [ 'max_cluster_size_ci' ] = scipy . stats . t . interval ( 1 - alpha , df = runs - 1 , loc = max_cluster_size_sample_mean , scale = max_cluster_size_sample_std / sqrt_n )
    np . seterr ( ** old_settings )
else :
    ret [ 'max_cluster_size_ci' ] = ( max_cluster_size_sample_mean * np . ones ( 2 ) )
return ret


signature:
def _microcanonical_average_moments ( moments , alpha ) :
    """Compute the average moments of the cluster size distributions
 Helper function for : func : ` microcanonical _ averages `
 Parameters
 moments : 2 - D : py : class : ` numpy . ndarray ` of int
 ` ` moments . shape [ 1 ] = = 5 ` .
 Each array ` ` moments [ r , : ] ` ` is the ` ` moments ` ` field of the output of
 : func : ` sample _ states ` :
 The ` ` k ` ` - th entry is the ` ` k ` ` - th raw moment of the ( absolute ) cluster
 size distribution .
 alpha : float
 Significance level .
 Returns
 ret : dict
 Moment statistics
 ret [ ' moments ' ] : 1 - D : py : class : ` numpy . ndarray ` of float , size 5
 The ` ` k ` ` - th entry is the average ` ` k ` ` - th raw moment of the ( absolute )
 cluster size distribution , with ` ` k ` ` ranging from ` ` 0 ` ` to ` ` 4 ` ` .
 ret [ ' moments _ ci ' ] : 2 - D : py : class : ` numpy . ndarray ` of float , shape ( 5,2)
 ` ` ret [ ' moments _ ci ' ] [ k ] ` ` are the lower and upper bounds of the normal
 confidence interval of the average ` ` k ` ` - th raw moment of the
 ( absolute ) cluster size distribution , with ` ` k ` ` ranging from ` ` 0 ` ` to
 See Also
 sample _ states : computation of moments
 microcanonical _ averages : moment statistics"""

implementation:
ret = dict ( )
runs = moments . shape [ 0 ]
sqrt_n = np . sqrt ( runs )
moments_sample_mean = moments . mean ( axis = 0 )
ret [ 'moments' ] = moments_sample_mean
moments_sample_std = moments . std ( axis = 0 , ddof = 1 )
ret [ 'moments_ci' ] = np . empty ( ( 5 , 2 ) )
for k in range ( 5 ) :
    if moments_sample_std [ k ] :
        old_settings = np . seterr ( all = 'raise' )
        ret [ 'moments_ci' ] [ k ] = scipy . stats . t . interval ( 1 - alpha , df = runs - 1 , loc = moments_sample_mean [ k ] , scale = moments_sample_std [ k ] / sqrt_n )
        np . seterr ( ** old_settings )
    else :
        ret [ 'moments_ci' ] [ k ] = ( moments_sample_mean [ k ] * np . ones ( 2 ) )
return ret


signature:
def microcanonical_averages ( graph , runs = 40 , spanning_cluster = True , model = 'bond' , alpha = alpha_1sigma , copy_result = True ) :
    r'''Generate successive microcanonical percolation ensemble averages
 This is a : ref : ` generator function < python : tut - generators > ` to successively
 add one edge at a time from the graph to the percolation model for a number
 of independent runs in parallel .
 At each iteration , it calculates and returns the averaged cluster
 statistics .
 Parameters
 graph : networkx . Graph
 The substrate graph on which percolation is to take place
 runs : int , optional
 Number of independent runs .
 Defaults to ` ` 40 ` ` .
 spanning _ cluster : bool , optional
 Defaults to ` ` True ` ` .
 model : str , optional
 The percolation model ( either ` ` ' bond ' ` ` or ` ` ' site ' ` ` ) .
 Defaults to ` ` ' bond ' ` ` .
 . . note : : Other models than ` ` ' bond ' ` ` are not supported yet .
 alpha : float , optional
 Significance level .
 Defaults to 1 sigma of the normal distribution .
 ` ` 1 - alpha ` ` is the confidence level .
 copy _ result : bool , optional
 Whether to return a copy or a reference to the result dictionary .
 Defaults to ` ` True ` ` .
 Yields
 ret : dict
 Cluster statistics
 ret [ ' n ' ] : int
 Number of occupied bonds
 ret [ ' N ' ] : int
 Total number of sites
 ret [ ' M ' ] : int
 Total number of bonds
 ret [ ' spanning _ cluster ' ] : float
 The average number ( Binomial proportion ) of runs that have a spanning
 cluster .
 This is the Bayesian point estimate of the posterior mean , with a
 uniform prior .
 Only exists if ` spanning _ cluster ` is set to ` ` True ` ` .
 ret [ ' spanning _ cluster _ ci ' ] : 1 - D : py : class : ` numpy . ndarray ` of float , size 2
 The lower and upper bounds of the Binomial proportion confidence
 interval with uniform prior .
 Only exists if ` spanning _ cluster ` is set to ` ` True ` ` .
 ret [ ' max _ cluster _ size ' ] : float
 Average size of the largest cluster ( absolute number of sites )
 ret [ ' max _ cluster _ size _ ci ' ] : 1 - D : py : class : ` numpy . ndarray ` of float , size 2
 Lower and upper bounds of the normal confidence interval of the average
 size of the largest cluster ( absolute number of sites )
 ret [ ' moments ' ] : 1 - D : py : class : ` numpy . ndarray ` of float , size 5
 The ` ` k ` ` - th entry is the average ` ` k ` ` - th raw moment of the ( absolute )
 cluster size distribution , with ` ` k ` ` ranging from ` ` 0 ` ` to ` ` 4 ` ` .
 ret [ ' moments _ ci ' ] : 2 - D : py : class : ` numpy . ndarray ` of float , shape ( 5,2)
 ` ` ret [ ' moments _ ci ' ] [ k ] ` ` are the lower and upper bounds of the normal
 confidence interval of the average ` ` k ` ` - th raw moment of the
 ( absolute ) cluster size distribution , with ` ` k ` ` ranging from ` ` 0 ` ` to
 Raises
 ValueError
 If ` runs ` is not a positive integer
 ValueError
 If ` alpha ` is not a float in the interval ( 0 , 1)
 See also
 sample _ states
 percolate . percolate . _ microcanonical _ average _ spanning _ cluster
 percolate . percolate . _ microcanonical _ average _ max _ cluster _ size
 Notes
 Iterating through this generator corresponds to several parallel runs of
 the Newman - Ziff algorithm .
 Each iteration yields a microcanonical percolation ensemble for the number
 : math : ` n ` of occupied bonds . [ 9 ] _
 The first iteration yields the trivial microcanonical percolation ensemble
 with : math : ` n = 0 ` occupied bonds .
 Spanning cluster
 . . seealso : : : py : func : ` sample _ states `
 Raw moments of the cluster size distribution
 . . seealso : : : py : func : ` sample _ states `
 References
 . . [ 9 ] 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 > ` _ .'''

implementation:
try :
    runs = int ( runs )
except :
    raise ValueError ( "runs needs to be a positive integer" )
if runs <= 0 :
    raise ValueError ( "runs needs to be a positive integer" )
try :
    alpha = float ( alpha )
except :
    raise ValueError ( "alpha needs to be a float in the interval (0, 1)" )
if alpha <= 0.0 or alpha >= 1.0 :
    raise ValueError ( "alpha needs to be a float in the interval (0, 1)" )
# initial iteration
# we do not need a copy of the result dictionary since we copy the values
# anyway
run_iterators = [ sample_states ( graph , spanning_cluster = spanning_cluster , model = model , copy_result = False ) for _ in range ( runs ) ]
ret = dict ( )
for microcanonical_ensemble in zip ( * run_iterators ) : # merge cluster statistics
    ret [ 'n' ] = microcanonical_ensemble [ 0 ] [ 'n' ]
    ret [ 'N' ] = microcanonical_ensemble [ 0 ] [ 'N' ]
    ret [ 'M' ] = microcanonical_ensemble [ 0 ] [ 'M' ]
    max_cluster_size = np . empty ( runs )
    moments = np . empty ( ( runs , 5 ) )
    if spanning_cluster :
        has_spanning_cluster = np . empty ( runs )
    for r , state in enumerate ( microcanonical_ensemble ) :
        assert state [ 'n' ] == ret [ 'n' ]
        assert state [ 'N' ] == ret [ 'N' ]
        assert state [ 'M' ] == ret [ 'M' ]
        max_cluster_size [ r ] = state [ 'max_cluster_size' ]
        moments [ r ] = state [ 'moments' ]
        if spanning_cluster :
            has_spanning_cluster [ r ] = state [ 'has_spanning_cluster' ]
    ret . update ( _microcanonical_average_max_cluster_size ( max_cluster_size , alpha ) )
    ret . update ( _microcanonical_average_moments ( moments , alpha ) )
    if spanning_cluster :
        ret . update ( _microcanonical_average_spanning_cluster ( has_spanning_cluster , alpha ) )
    if copy_result :
        yield copy . deepcopy ( ret )
    else :
        yield ret


signature:
def spanning_1d_chain ( length ) :
    """Generate a linear chain with auxiliary nodes for spanning cluster detection
 Parameters
 length : int
 Number of nodes in the chain , excluding the auxiliary nodes .
 Returns
 networkx . Graph
 A linear chain graph with auxiliary nodes for spanning cluster detection
 See Also
 sample _ states : spanning cluster detection"""

implementation:
ret = nx . grid_graph ( dim = [ int ( length + 2 ) ] )
ret . node [ 0 ] [ 'span' ] = 0
ret [ 0 ] [ 1 ] [ 'span' ] = 0
ret . node [ length + 1 ] [ 'span' ] = 1
ret [ length ] [ length + 1 ] [ 'span' ] = 1
return ret


signature:
def spanning_2d_grid ( length ) :
    """Generate a square lattice with auxiliary nodes for spanning detection
 Parameters
 length : int
 Number of nodes in one dimension , excluding the auxiliary nodes .
 Returns
 networkx . Graph
 A square lattice graph with auxiliary nodes for spanning cluster
 detection
 See Also
 sample _ states : spanning cluster detection"""

implementation:
ret = nx . grid_2d_graph ( length + 2 , length )
for i in range ( length ) : # side 0
    ret . node [ ( 0 , i ) ] [ 'span' ] = 0
    ret [ ( 0 , i ) ] [ ( 1 , i ) ] [ 'span' ] = 0
    # side 1
    ret . node [ ( length + 1 , i ) ] [ 'span' ] = 1
    ret [ ( length + 1 , i ) ] [ ( length , i ) ] [ 'span' ] = 1
return ret


signature:
def microcanonical_averages_arrays ( microcanonical_averages ) :
    """Compile microcanonical averages over all iteration steps into single arrays
 Helper function to aggregate the microcanonical averages over all iteration
 steps into single arrays for further processing
 Parameters
 microcanonical _ averages : iterable
 Typically , this is the : func : ` microcanonical _ averages ` generator
 Returns
 ret : dict
 Aggregated cluster statistics
 ret [ ' N ' ] : int
 Total number of sites
 ret [ ' M ' ] : int
 Total number of bonds
 ret [ ' spanning _ cluster ' ] : 1 - D : py : class : ` numpy . ndarray ` of float
 The percolation probability :
 The normalized average number of runs that have a spanning cluster .
 ret [ ' spanning _ cluster _ ci ' ] : 2 - D : py : class : ` numpy . ndarray ` of float , size 2
 The lower and upper bounds of the percolation probability .
 ret [ ' max _ cluster _ size ' ] : 1 - D : py : class : ` numpy . ndarray ` of float
 The percolation strength :
 Average relative size of the largest cluster
 ret [ ' max _ cluster _ size _ ci ' ] : 2 - D : py : class : ` numpy . ndarray ` of float
 Lower and upper bounds of the normal confidence interval of the
 percolation strength .
 ret [ ' moments ' ] : 2 - D : py : class : ` numpy . ndarray ` of float , shape ( 5 , M + 1)
 Average raw moments of the ( relative ) cluster size distribution .
 ret [ ' moments _ ci ' ] : 3 - D : py : class : ` numpy . ndarray ` of float , shape ( 5 , M + 1 , 2)
 Lower and upper bounds of the normal confidence interval of the raw
 moments of the ( relative ) cluster size distribution .
 See Also
 microcanonical _ averages"""

implementation:
ret = dict ( )
for n , microcanonical_average in enumerate ( microcanonical_averages ) :
    assert n == microcanonical_average [ 'n' ]
    if n == 0 :
        num_edges = microcanonical_average [ 'M' ]
        num_sites = microcanonical_average [ 'N' ]
        spanning_cluster = ( 'spanning_cluster' in microcanonical_average )
        ret [ 'max_cluster_size' ] = np . empty ( num_edges + 1 )
        ret [ 'max_cluster_size_ci' ] = np . empty ( ( num_edges + 1 , 2 ) )
        if spanning_cluster :
            ret [ 'spanning_cluster' ] = np . empty ( num_edges + 1 )
            ret [ 'spanning_cluster_ci' ] = np . empty ( ( num_edges + 1 , 2 ) )
        ret [ 'moments' ] = np . empty ( ( 5 , num_edges + 1 ) )
        ret [ 'moments_ci' ] = np . empty ( ( 5 , num_edges + 1 , 2 ) )
    ret [ 'max_cluster_size' ] [ n ] = microcanonical_average [ 'max_cluster_size' ]
    ret [ 'max_cluster_size_ci' ] [ n ] = ( microcanonical_average [ 'max_cluster_size_ci' ] )
    if spanning_cluster :
        ret [ 'spanning_cluster' ] [ n ] = ( microcanonical_average [ 'spanning_cluster' ] )
        ret [ 'spanning_cluster_ci' ] [ n ] = ( microcanonical_average [ 'spanning_cluster_ci' ] )
    ret [ 'moments' ] [ : , n ] = microcanonical_average [ 'moments' ]
    ret [ 'moments_ci' ] [ : , n ] = microcanonical_average [ 'moments_ci' ]
# normalize by number of sites
for key in ret :
    if 'spanning_cluster' in key :
        continue
    ret [ key ] /= num_sites
ret [ 'M' ] = num_edges
ret [ 'N' ] = num_sites
return ret


signature:
def _binomial_pmf ( n , p ) :
    """Compute the binomial PMF according to Newman and Ziff
 Helper function for : func : ` canonical _ averages `
 See Also
 canonical _ averages
 Notes
 See Newman & Ziff , Equation ( 10 ) [ 10 ] _
 References
 . . [ 10 ] 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 > ` _ ."""

implementation:
n = int ( n )
ret = np . empty ( n + 1 )
nmax = int ( np . round ( p * n ) )
ret [ nmax ] = 1.0
old_settings = np . seterr ( under = 'ignore' )
# seterr to known value
for i in range ( nmax + 1 , n + 1 ) :
    ret [ i ] = ret [ i - 1 ] * ( n - i + 1.0 ) / i * p / ( 1.0 - p )
for i in range ( nmax - 1 , - 1 , - 1 ) :
    ret [ i ] = ret [ i + 1 ] * ( i + 1.0 ) / ( n - i ) * ( 1.0 - p ) / p
np . seterr ( ** old_settings )
# reset to default
return ret / ret . sum ( )


signature:
def canonical_averages ( ps , microcanonical_averages_arrays ) :
    """Compute the canonical cluster statistics from microcanonical statistics
 This is according to Newman and Ziff , Equation ( 2 ) .
 Note that we also simply average the bounds of the confidence intervals
 according to this formula .
 Parameters
 ps : iterable of float
 Each entry is a probability for which to form the canonical ensemble
 and compute the weighted statistics from the microcanonical statistics
 microcanonical _ averages _ arrays
 Typically the output of : func : ` microcanonical _ averages _ arrays `
 Returns
 ret : dict
 Canonical ensemble cluster statistics
 ret [ ' ps ' ] : iterable of float
 The parameter ` ps `
 ret [ ' N ' ] : int
 Total number of sites
 ret [ ' M ' ] : int
 Total number of bonds
 ret [ ' spanning _ cluster ' ] : 1 - D : py : class : ` numpy . ndarray ` of float
 The percolation probability :
 The normalized average number of runs that have a spanning cluster .
 ret [ ' spanning _ cluster _ ci ' ] : 2 - D : py : class : ` numpy . ndarray ` of float , size 2
 The lower and upper bounds of the percolation probability .
 ret [ ' max _ cluster _ size ' ] : 1 - D : py : class : ` numpy . ndarray ` of float
 The percolation strength :
 Average relative size of the largest cluster
 ret [ ' max _ cluster _ size _ ci ' ] : 2 - D : py : class : ` numpy . ndarray ` of float
 Lower and upper bounds of the normal confidence interval of the
 percolation strength .
 ret [ ' moments ' ] : 2 - D : py : class : ` numpy . ndarray ` of float , shape ( 5 , M + 1)
 Average raw moments of the ( relative ) cluster size distribution .
 ret [ ' moments _ ci ' ] : 3 - D : py : class : ` numpy . ndarray ` of float , shape ( 5 , M + 1 , 2)
 Lower and upper bounds of the normal confidence interval of the raw
 moments of the ( relative ) cluster size distribution .
 See Also
 microcanonical _ averages
 microcanonical _ averages _ arrays"""

implementation:
num_sites = microcanonical_averages_arrays [ 'N' ]
num_edges = microcanonical_averages_arrays [ 'M' ]
spanning_cluster = ( 'spanning_cluster' in microcanonical_averages_arrays )
ret = dict ( )
ret [ 'ps' ] = ps
ret [ 'N' ] = num_sites
ret [ 'M' ] = num_edges
ret [ 'max_cluster_size' ] = np . empty ( ps . size )
ret [ 'max_cluster_size_ci' ] = np . empty ( ( ps . size , 2 ) )
if spanning_cluster :
    ret [ 'spanning_cluster' ] = np . empty ( ps . size )
    ret [ 'spanning_cluster_ci' ] = np . empty ( ( ps . size , 2 ) )
ret [ 'moments' ] = np . empty ( ( 5 , ps . size ) )
ret [ 'moments_ci' ] = np . empty ( ( 5 , ps . size , 2 ) )
for p_index , p in enumerate ( ps ) :
    binomials = _binomial_pmf ( n = num_edges , p = p )
    for key , value in microcanonical_averages_arrays . items ( ) :
        if len ( key ) <= 1 :
            continue
        if key in [ 'max_cluster_size' , 'spanning_cluster' ] :
            ret [ key ] [ p_index ] = np . sum ( binomials * value )
        elif key in [ 'max_cluster_size_ci' , 'spanning_cluster_ci' ] :
            ret [ key ] [ p_index ] = np . sum ( np . tile ( binomials , ( 2 , 1 ) ) . T * value , axis = 0 )
        elif key == 'moments' :
            ret [ key ] [ : , p_index ] = np . sum ( np . tile ( binomials , ( 5 , 1 ) ) * value , axis = 1 )
        elif key == 'moments_ci' :
            ret [ key ] [ : , p_index ] = np . sum ( np . rollaxis ( np . tile ( binomials , ( 5 , 2 , 1 ) ) , 2 , 1 ) * value , axis = 1 )
        else :
            raise NotImplementedError ( '{}-dimensional array' . format ( value . ndim ) )
return ret


signature:
def statistics ( graph , ps , spanning_cluster = True , model = 'bond' , alpha = alpha_1sigma , runs = 40 ) :
    """Helper function to compute percolation statistics
 See Also
 canonical _ averages
 microcanonical _ averages
 sample _ states"""

implementation:
my_microcanonical_averages = microcanonical_averages ( graph = graph , runs = runs , spanning_cluster = spanning_cluster , model = model , alpha = alpha )
my_microcanonical_averages_arrays = microcanonical_averages_arrays ( my_microcanonical_averages )
return canonical_averages ( ps , my_microcanonical_averages_arrays )


signature:
def to_html ( graph : BELGraph ) -> str :
    """Render the graph as an HTML string .
 Common usage may involve writing to a file like :
 > > > from pybel . examples import sialic _ acid _ graph
 > > > with open ( ' html _ output . html ' , ' w ' ) as file :
 . . . print ( to _ html ( sialic _ acid _ graph ) , file = file )"""

implementation:
context = get_network_summary_dict ( graph )
summary_dict = graph . summary_dict ( )
citation_years = context [ 'citation_years' ]
function_count = context [ 'function_count' ]
relation_count = context [ 'relation_count' ]
error_count = context [ 'error_count' ]
transformations_count = context [ 'modifications_count' ]
hub_data = context [ 'hub_data' ]
disease_data = context [ 'disease_data' ]
authors_count = context [ 'authors_count' ]
variants_count = context [ 'variants_count' ]
namespaces_count = context [ 'namespaces_count' ]
confidence_count = context [ 'confidence_count' ]
confidence_data = [ ( label , confidence_count . get ( label , 0 ) ) for label in ( 'None' , 'Very Low' , 'Low' , 'Medium' , 'High' , 'Very High' ) ]
template = environment . get_template ( 'index.html' )
return template . render ( graph = graph , # Node Charts
chart_1_data = prepare_c3 ( function_count , 'Node Count' ) , chart_6_data = prepare_c3 ( namespaces_count , 'Node Count' ) , chart_5_data = prepare_c3 ( variants_count , 'Node Count' ) , number_variants = sum ( variants_count . values ( ) ) , number_namespaces = len ( namespaces_count ) , # Edge Charts
chart_2_data = prepare_c3 ( relation_count , 'Edge Count' ) , chart_4_data = prepare_c3 ( transformations_count , 'Edge Count' ) if transformations_count else None , number_transformations = sum ( transformations_count . values ( ) ) , # Error Charts
chart_3_data = prepare_c3 ( error_count , 'Error Type' ) if error_count else None , # Topology Charts
chart_7_data = prepare_c3 ( hub_data , 'Degree' ) , chart_9_data = prepare_c3 ( disease_data , 'Degree' ) if disease_data else None , # Bibliometrics Charts
chart_authors_count = prepare_c3 ( authors_count , 'Edges Contributed' ) , chart_10_data = prepare_c3_time_series ( citation_years , 'Number of Articles' ) if citation_years else None , chart_confidence_count = prepare_c3 ( confidence_data , 'Edge Count' ) , summary_dict = summary_dict , # Everything else : )
** context )


signature:
def get_network_summary_dict ( graph : BELGraph ) -> Mapping :
    """Create a summary dictionary ."""

implementation:
return dict ( # Counters
function_count = count_functions ( graph ) , modifications_count = get_modifications_count ( graph ) , relation_count = count_relations ( graph ) , authors_count = count_authors ( graph ) . most_common ( 15 ) , variants_count = count_variants ( graph ) , namespaces_count = count_namespaces ( graph ) , hub_data = { ( node . name or node . identifier if NAME in node or IDENTIFIER in node else str ( node ) ) : degree for node , degree in get_top_hubs ( graph , n = 15 ) } , disease_data = { ( node . name or node . identifier if NAME in node or IDENTIFIER in node else str ( node ) ) : count for node , count in get_top_pathologies ( graph , n = 15 ) } , # BioGrammar
regulatory_pairs = [ get_pair_tuple ( u , v ) for u , v in get_regulatory_pairs ( graph ) ] , unstable_pairs = list ( itt . chain ( ( get_pair_tuple ( u , v ) + ( 'Chaotic' , ) for u , v , in get_chaotic_pairs ( graph ) ) , ( get_pair_tuple ( u , v ) + ( 'Dampened' , ) for u , v , in get_dampened_pairs ( graph ) ) , ) ) , contradictory_pairs = [ get_pair_tuple ( u , v ) + ( relation , ) for u , v , relation in get_contradiction_summary ( graph ) ] , contradictory_triplets = list ( itt . chain ( ( get_triplet_tuple ( a , b , c ) + ( 'Separate' , ) for a , b , c in get_separate_unstable_correlation_triples ( graph ) ) , ( get_triplet_tuple ( a , b , c ) + ( 'Mutual' , ) for a , b , c in get_mutually_unstable_correlation_triples ( graph ) ) , ( get_triplet_tuple ( a , b , c ) + ( 'Jens' , ) for a , b , c in get_jens_unstable ( graph ) ) , ( get_triplet_tuple ( a , b , c ) + ( 'Increase Mismatch' , ) for a , b , c in get_increase_mismatch_triplets ( graph ) ) , ( get_triplet_tuple ( a , b , c ) + ( 'Decrease Mismatch' , ) for a , b , c in get_decrease_mismatch_triplets ( graph ) ) , ) ) , unstable_triplets = list ( itt . chain ( ( get_triplet_tuple ( a , b , c ) + ( 'Chaotic' , ) for a , b , c in get_chaotic_triplets ( graph ) ) , ( get_triplet_tuple ( a , b , c ) + ( 'Dampened' , ) for a , b , c in get_dampened_triplets ( graph ) ) , ) ) , causal_pathologies = sorted ( { get_pair_tuple ( u , v ) + ( graph [ u ] [ v ] [ k ] [ RELATION ] , ) for u , v , k in filter_edges ( graph , has_pathology_causal ) } ) , # Misc .
undefined_namespaces = get_undefined_namespaces ( graph ) , undefined_annotations = get_undefined_annotations ( graph ) , namespaces_with_incorrect_names = get_namespaces_with_incorrect_names ( graph ) , unused_namespaces = get_unused_namespaces ( graph ) , unused_annotations = get_unused_annotations ( graph ) , unused_list_annotation_values = get_unused_list_annotation_values ( graph ) , naked_names = get_naked_names ( graph ) , error_count = count_error_types ( graph ) , # Errors
error_groups = get_most_common_errors ( graph ) , syntax_errors = get_syntax_errors ( graph ) , # Bibliometrics
citation_years = get_citation_years ( graph ) , confidence_count = count_confidences ( graph ) , )


signature:
def get_pair_tuple ( a : BaseEntity , b : BaseEntity ) -> Tuple [ str , str , str , str ] :
    """Get the pair as a tuple of BEL / hashes ."""

implementation: return a . as_bel ( ) , a . sha512 , b . as_bel ( ) , b . sha512


signature:
def get_triplet_tuple ( a : BaseEntity , b : BaseEntity , c : BaseEntity ) -> Tuple [ str , str , str , str , str , str ] :
    """Get the triple as a tuple of BEL / hashes ."""

implementation: return a . as_bel ( ) , a . sha512 , b . as_bel ( ) , b . sha512 , c . as_bel ( ) , c . sha512


signature:
def rank_causalr_hypothesis ( graph , node_to_regulation , regulator_node ) :
    """Test the regulator hypothesis of the given node on the input data using the algorithm .
 Note : this method returns both + / - signed hypotheses evaluated
 Algorithm :
 1 . Calculate the shortest path between the regulator node and each node in observed _ regulation
 2 . Calculate the concordance of the causal network and the observed regulation when there is path
 between target node and regulator node
 : param networkx . DiGraph graph : A causal graph
 : param dict node _ to _ regulation : Nodes to score ( 1 , - 1,0)
 : return Dictionaries with hypothesis results ( keys : score , correct , incorrect , ambiguous )
 : rtype : dict"""

implementation:
upregulation_hypothesis = { 'correct' : 0 , 'incorrect' : 0 , 'ambiguous' : 0 }
downregulation_hypothesis = { 'correct' : 0 , 'incorrect' : 0 , 'ambiguous' : 0 }
targets = [ node for node in node_to_regulation if node != regulator_node ]
predicted_regulations = run_cna ( graph , regulator_node , targets )
# + signed hypothesis
for _ , target_node , predicted_regulation in predicted_regulations :
    if ( predicted_regulation is Effect . inhibition or predicted_regulation is Effect . activation ) and ( predicted_regulation . value == node_to_regulation [ target_node ] ) :
        upregulation_hypothesis [ 'correct' ] += 1
        downregulation_hypothesis [ 'incorrect' ] += 1
    elif predicted_regulation is Effect . ambiguous :
        upregulation_hypothesis [ 'ambiguous' ] += 1
        downregulation_hypothesis [ 'ambiguous' ] += 1
    elif predicted_regulation is Effect . no_effect :
        continue
    else :
        downregulation_hypothesis [ 'correct' ] += 1
        upregulation_hypothesis [ 'incorrect' ] += 1
upregulation_hypothesis [ 'score' ] = upregulation_hypothesis [ 'correct' ] - upregulation_hypothesis [ 'incorrect' ]
downregulation_hypothesis [ 'score' ] = downregulation_hypothesis [ 'correct' ] - downregulation_hypothesis [ 'incorrect' ]
return upregulation_hypothesis , downregulation_hypothesis


signature:
def run_cna ( graph , root , targets , relationship_dict = None ) :
    """Returns the effect from the root to the target nodes represented as { - 1,1}
 : param pybel . BELGraph graph : A BEL graph
 : param BaseEntity root : The root node
 : param iter targets : The targets nodes
 : param dict relationship _ dict : dictionary with relationship effects
 : return list [ tuple ] :"""

implementation:
causal_effects = [ ]
relationship_dict = causal_effect_dict if relationship_dict is None else relationship_dict
for target in targets :
    try :
        shortest_paths = nx . all_shortest_paths ( graph , source = root , target = target )
        effects_in_path = set ( )
        for shortest_path in shortest_paths :
            effects_in_path . add ( get_path_effect ( graph , shortest_path , relationship_dict ) )
        if len ( effects_in_path ) == 1 :
            causal_effects . append ( ( root , target , next ( iter ( effects_in_path ) ) ) )
        # Append the only predicted effect
        elif Effect . activation in effects_in_path and Effect . inhibition in effects_in_path :
            causal_effects . append ( ( root , target , Effect . ambiguous ) )
        elif Effect . activation in effects_in_path and Effect . inhibition not in effects_in_path :
            causal_effects . append ( ( root , target , Effect . activation ) )
        elif Effect . inhibition in effects_in_path and Effect . activation not in effects_in_path :
            causal_effects . append ( ( root , target , Effect . inhibition ) )
        else :
            log . warning ( 'Exception in set: {}.' . format ( effects_in_path ) )
    except nx . NetworkXNoPath :
        log . warning ( 'No shortest path between: {} and {}.' . format ( root , target ) )
return causal_effects


signature:
def get_path_effect ( graph , path , relationship_dict ) :
    """Calculate the final effect of the root node to the sink node in the path .
 : param pybel . BELGraph graph : A BEL graph
 : param list path : Path from root to sink node
 : param dict relationship _ dict : dictionary with relationship effects
 : rtype : Effect"""

implementation:
causal_effect = [ ]
for predecessor , successor in pairwise ( path ) :
    if pair_has_contradiction ( graph , predecessor , successor ) :
        return Effect . ambiguous
    edges = graph . get_edge_data ( predecessor , successor )
    edge_key , edge_relation , _ = rank_edges ( edges )
    relation = graph [ predecessor ] [ successor ] [ edge_key ] [ RELATION ]
    # Returns Effect . no _ effect if there is a non causal edge in path
    if relation not in relationship_dict or relationship_dict [ relation ] == 0 :
        return Effect . no_effect
    causal_effect . append ( relationship_dict [ relation ] )
final_effect = reduce ( lambda x , y : x * y , causal_effect )
return Effect . activation if final_effect == 1 else Effect . inhibition


signature:
def rank_edges ( edges , edge_ranking = None ) :
    """Return the highest ranked edge from a multiedge .
 : param dict edges : dictionary with all edges between two nodes
 : param dict edge _ ranking : A dictionary of { relationship : score }
 : return : Highest ranked edge
 : rtype : tuple : ( edge id , relation , score given ranking )"""

implementation:
edge_ranking = default_edge_ranking if edge_ranking is None else edge_ranking
edges_scores = [ ( edge_id , edge_data [ RELATION ] , edge_ranking [ edge_data [ RELATION ] ] ) for edge_id , edge_data in edges . items ( ) ]
return max ( edges_scores , key = itemgetter ( 2 ) )


signature:
def group_nodes_by_annotation ( graph : BELGraph , annotation : str = 'Subgraph' ) -> Mapping [ str , Set [ BaseEntity ] ] :
    """Group the nodes occurring in edges by the given annotation ."""

implementation:
result = defaultdict ( set )
for u , v , d in graph . edges ( data = True ) :
    if not edge_has_annotation ( d , annotation ) :
        continue
    result [ d [ ANNOTATIONS ] [ annotation ] ] . add ( u )
    result [ d [ ANNOTATIONS ] [ annotation ] ] . add ( v )
return dict ( result )


signature:
def group_nodes_by_annotation_filtered ( graph : BELGraph , node_predicates : NodePredicates = None , annotation : str = 'Subgraph' , ) -> Mapping [ str , Set [ BaseEntity ] ] :
    """Group the nodes occurring in edges by the given annotation , with a node filter applied .
 : param graph : A BEL graph
 : param node _ predicates : A predicate or list of predicates ( graph , node ) - > bool
 : param annotation : The annotation to use for grouping
 : return : A dictionary of { annotation value : set of nodes }"""

implementation:
node_filter = concatenate_node_predicates ( node_predicates )
return { key : { node for node in nodes if node_filter ( graph , node ) } for key , nodes in group_nodes_by_annotation ( graph , annotation ) . items ( ) }


signature:
def get_mapped_nodes ( graph : BELGraph , namespace : str , names : Iterable [ str ] ) -> Mapping [ BaseEntity , Set [ BaseEntity ] ] :
    """Return a dict with keys : nodes that match the namespace and in names and values other nodes ( complexes , variants , orthologous . . . ) or this node .
 : param graph : A BEL graph
 : param namespace : The namespace to search
 : param names : List or set of values from which we want to map nodes from
 : return : Main node to variants / groups ."""

implementation:
parent_to_variants = defaultdict ( set )
names = set ( names )
for u , v , d in graph . edges ( data = True ) :
    if d [ RELATION ] in { HAS_MEMBER , HAS_COMPONENT } and v . get ( NAMESPACE ) == namespace and v . get ( NAME ) in names :
        parent_to_variants [ v ] . add ( u )
    elif d [ RELATION ] == HAS_VARIANT and u . get ( NAMESPACE ) == namespace and u . get ( NAME ) in names :
        parent_to_variants [ u ] . add ( v )
    elif d [ RELATION ] == ORTHOLOGOUS and u . get ( NAMESPACE ) == namespace and u . get ( NAME ) in names :
        parent_to_variants [ u ] . add ( v )
return dict ( parent_to_variants )


signature:
def bel_to_spia_matrices ( graph : BELGraph ) -> Mapping [ str , pd . DataFrame ] :
    """Create an excel sheet ready to be used in SPIA software .
 : param graph : BELGraph
 : return : dictionary with matrices"""

implementation:
index_nodes = get_matrix_index ( graph )
spia_matrices = build_spia_matrices ( index_nodes )
for u , v , edge_data in graph . edges ( data = True ) : # Both nodes are CentralDogma abundances
    if isinstance ( u , CentralDogma ) and isinstance ( v , CentralDogma ) : # Update matrix dict
        update_spia_matrices ( spia_matrices , u , v , edge_data )
    # Subject is CentralDogmaAbundance and node is ListAbundance
    elif isinstance ( u , CentralDogma ) and isinstance ( v , ListAbundance ) : # Add a relationship from subject to each of the members in the object
        for node in v . members : # Skip if the member is not in CentralDogma
            if not isinstance ( node , CentralDogma ) :
                continue
            update_spia_matrices ( spia_matrices , u , node , edge_data )
    # Subject is ListAbundance and node is CentralDogmaAbundance
    elif isinstance ( u , ListAbundance ) and isinstance ( v , CentralDogma ) : # Add a relationship from each of the members of the subject to the object
        for node in u . members : # Skip if the member is not in CentralDogma
            if not isinstance ( node , CentralDogma ) :
                continue
            update_spia_matrices ( spia_matrices , node , v , edge_data )
    # Both nodes are ListAbundance
    elif isinstance ( u , ListAbundance ) and isinstance ( v , ListAbundance ) :
        for sub_member , obj_member in product ( u . members , v . members ) : # Update matrix if both are CentralDogma
            if isinstance ( sub_member , CentralDogma ) and isinstance ( obj_member , CentralDogma ) :
                update_spia_matrices ( spia_matrices , sub_member , obj_member , edge_data )
    # else Not valid edge
return spia_matrices


signature:
def get_matrix_index ( graph : BELGraph ) -> Set [ str ] :
    """Return set of HGNC names from Proteins / Rnas / Genes / miRNA , nodes that can be used by SPIA ."""

implementation:
# TODO : Using HGNC Symbols for now
return { node . name for node in graph if isinstance ( node , CentralDogma ) and node . namespace . upper ( ) == 'HGNC' }


signature:
def build_spia_matrices ( nodes : Set [ str ] ) -> Dict [ str , pd . DataFrame ] :
    """Build an adjacency matrix for each KEGG relationship and return in a dictionary .
 : param nodes : A set of HGNC gene symbols
 : return : Dictionary of adjacency matrix for each relationship"""

implementation:
nodes = list ( sorted ( nodes ) )
# Create sheets of the excel in the given order
matrices = OrderedDict ( )
for relation in KEGG_RELATIONS :
    matrices [ relation ] = pd . DataFrame ( 0 , index = nodes , columns = nodes )
return matrices


signature:
def update_spia_matrices ( spia_matrices : Dict [ str , pd . DataFrame ] , u : CentralDogma , v : CentralDogma , edge_data : EdgeData , ) -> None :
    """Populate the adjacency matrix ."""

implementation:
if u . namespace . upper ( ) != 'HGNC' or v . namespace . upper ( ) != 'HGNC' :
    return
u_name = u . name
v_name = v . name
relation = edge_data [ RELATION ]
if relation in CAUSAL_INCREASE_RELATIONS : # If it has pmod check which one and add it to the corresponding matrix
    if v . variants and any ( isinstance ( variant , ProteinModification ) for variant in v . variants ) :
        for variant in v . variants :
            if not isinstance ( variant , ProteinModification ) :
                continue
            if variant [ IDENTIFIER ] [ NAME ] == "Ub" :
                spia_matrices [ "activation_ubiquination" ] [ u_name ] [ v_name ] = 1
            elif variant [ IDENTIFIER ] [ NAME ] == "Ph" :
                spia_matrices [ "activation_phosphorylation" ] [ u_name ] [ v_name ] = 1
    elif isinstance ( v , ( Gene , Rna ) ) : # Normal increase , add activation
        spia_matrices [ 'expression' ] [ u_name ] [ v_name ] = 1
    else :
        spia_matrices [ 'activation' ] [ u_name ] [ v_name ] = 1
elif relation in CAUSAL_DECREASE_RELATIONS : # If it has pmod check which one and add it to the corresponding matrix
    if v . variants and any ( isinstance ( variant , ProteinModification ) for variant in v . variants ) :
        for variant in v . variants :
            if not isinstance ( variant , ProteinModification ) :
                continue
            if variant [ IDENTIFIER ] [ NAME ] == "Ub" :
                spia_matrices [ 'inhibition_ubiquination' ] [ u_name ] [ v_name ] = 1
            elif variant [ IDENTIFIER ] [ NAME ] == "Ph" :
                spia_matrices [ "inhibition_phosphorylation" ] [ u_name ] [ v_name ] = 1
    elif isinstance ( v , ( Gene , Rna ) ) : # Normal decrease , check which matrix
        spia_matrices [ "repression" ] [ u_name ] [ v_name ] = 1
    else :
        spia_matrices [ "inhibition" ] [ u_name ] [ v_name ] = 1
elif relation == ASSOCIATION :
    spia_matrices [ "binding_association" ] [ u_name ] [ v_name ] = 1


signature:
def spia_matrices_to_excel ( spia_matrices : Mapping [ str , pd . DataFrame ] , path : str ) -> None :
    """Export a SPIA data dictionary into an Excel sheet at the given path .
 . . note : :
 # The R import should add the values :
 # [ " nodes " ] from the columns
 # [ " title " ] from the name of the file
 # [ " NumberOfReactions " ] set to " 0" """

implementation:
writer = pd . ExcelWriter ( path , engine = 'xlsxwriter' )
for relation , df in spia_matrices . items ( ) :
    df . to_excel ( writer , sheet_name = relation , index = False )
# Save excel
writer . save ( )


signature:
def spia_matrices_to_tsvs ( spia_matrices : Mapping [ str , pd . DataFrame ] , directory : str ) -> None :
    """Export a SPIA data dictionary into a directory as several TSV documents ."""

implementation:
os . makedirs ( directory , exist_ok = True )
for relation , df in spia_matrices . items ( ) :
    df . to_csv ( os . path . join ( directory , f'{relation}.tsv' ) , index = True )


signature:
def main ( graph : BELGraph , xlsx : str , tsvs : str ) :
    """Export the graph to a SPIA Excel sheet ."""

implementation:
if not xlsx and not tsvs :
    click . secho ( 'Specify at least one option --xlsx or --tsvs' , fg = 'red' )
    sys . exit ( 1 )
spia_matrices = bel_to_spia_matrices ( graph )
if xlsx :
    spia_matrices_to_excel ( spia_matrices , xlsx )
if tsvs :
    spia_matrices_to_tsvs ( spia_matrices , tsvs )


signature:
def overlay_data ( graph : BELGraph , data : Mapping [ BaseEntity , Any ] , label : Optional [ str ] = None , overwrite : bool = False , ) -> None :
    """Overlays tabular data on the network
 : param graph : A BEL Graph
 : param data : A dictionary of { tuple node : data for that node }
 : param label : The annotation label to put in the node dictionary
 : param overwrite : Should old annotations be overwritten ?"""

implementation:
if label is None :
    label = WEIGHT
for node , value in data . items ( ) :
    if node not in graph :
        log . debug ( '%s not in graph' , node )
        continue
    if label in graph . nodes [ node ] and not overwrite :
        log . debug ( '%s already on %s' , label , node )
        continue
    graph . nodes [ node ] [ label ] = value


signature:
def overlay_type_data ( graph : BELGraph , data : Mapping [ str , float ] , func : str , namespace : str , label : Optional [ str ] = None , overwrite : bool = False , impute : Optional [ float ] = None , ) -> None :
    """Overlay tabular data on the network for data that comes from an data set with identifiers that lack
 namespaces .
 For example , if you want to overlay differential gene expression data from a table , that table
 probably has HGNC identifiers , but no specific annotations that they are in the HGNC namespace or
 that the entities to which they refer are RNA .
 : param graph : A BEL Graph
 : param dict data : A dictionary of { name : data }
 : param func : The function of the keys in the data dictionary
 : param namespace : The namespace of the keys in the data dictionary
 : param label : The annotation label to put in the node dictionary
 : param overwrite : Should old annotations be overwritten ?
 : param impute : The value to use for missing data"""

implementation:
new_data = { node : data . get ( node [ NAME ] , impute ) for node in filter_nodes ( graph , function_namespace_inclusion_builder ( func , namespace ) ) }
overlay_data ( graph , new_data , label = label , overwrite = overwrite )


signature:
def load_differential_gene_expression ( path : str , gene_symbol_column : str = 'Gene.symbol' , logfc_column : str = 'logFC' , aggregator : Optional [ Callable [ [ List [ float ] ] , float ] ] = None , ) -> Mapping [ str , float ] :
    """Load and pre - process a differential gene expression data .
 : param path : The path to the CSV
 : param gene _ symbol _ column : The header of the gene symbol column in the data frame
 : param logfc _ column : The header of the log - fold - change column in the data frame
 : param aggregator : A function that aggregates a list of differential gene expression values . Defaults to
 : func : ` numpy . median ` . Could also use : : func : ` numpy . mean ` , : func : ` numpy . average ` ,
 : func : ` numpy . min ` , or : func : ` numpy . max `
 : return : A dictionary of { gene symbol : log fold change }"""

implementation:
if aggregator is None :
    aggregator = np . median
# Load the data frame
df = pd . read_csv ( path )
# Check the columns exist in the data frame
assert gene_symbol_column in df . columns
assert logfc_column in df . columns
# throw away columns that don ' t have gene symbols - these represent control sequences
df = df . loc [ df [ gene_symbol_column ] . notnull ( ) , [ gene_symbol_column , logfc_column ] ]
values = defaultdict ( list )
for _ , gene_symbol , log_fold_change in df . itertuples ( ) :
    values [ gene_symbol ] . append ( log_fold_change )
return { gene_symbol : aggregator ( log_fold_changes ) for gene_symbol , log_fold_changes in values . items ( ) }


signature:
def get_merged_namespace_names ( locations , check_keywords = True ) :
    """Loads many namespaces and combines their names .
 : param iter [ str ] locations : An iterable of URLs or file paths pointing to BEL namespaces .
 : param bool check _ keywords : Should all the keywords be the same ? Defaults to ` ` True ` `
 : return : A dictionary of { names : labels }
 : rtype : dict [ str , str ]
 Example Usage
 > > > from pybel . resources import write _ namespace
 > > > from pybel _ tools . definition _ utils import export _ namespace , get _ merged _ namespace _ names
 > > > graph = . . .
 > > > original _ ns _ url = . . .
 > > > export _ namespace ( graph , ' MBS ' ) # Outputs in current directory to MBS . belns
 > > > value _ dict = get _ merged _ namespace _ names ( [ original _ ns _ url , ' MBS . belns ' ] )
 > > > with open ( ' merged _ namespace . belns ' , ' w ' ) as f :
 > > > . . . write _ namespace ( ' MyBrokenNamespace ' , ' MBS ' , ' Other ' , ' Charles Hoyt ' , ' PyBEL Citation ' , value _ dict , file = f )"""

implementation:
resources = { location : get_bel_resource ( location ) for location in locations }
if check_keywords :
    resource_keywords = set ( config [ 'Namespace' ] [ 'Keyword' ] for config in resources . values ( ) )
    if 1 != len ( resource_keywords ) :
        raise ValueError ( 'Tried merging namespaces with different keywords: {}' . format ( resource_keywords ) )
result = { }
for resource in resources :
    result . update ( resource [ 'Values' ] )
return result


signature:
def merge_namespaces ( input_locations , output_path , namespace_name , namespace_keyword , namespace_domain , author_name , citation_name , namespace_description = None , namespace_species = None , namespace_version = None , namespace_query_url = None , namespace_created = None , author_contact = None , author_copyright = None , citation_description = None , citation_url = None , citation_version = None , citation_date = None , case_sensitive = True , delimiter = '|' , cacheable = True , functions = None , value_prefix = '' , sort_key = None , check_keywords = True ) :
    """Merges namespaces from multiple locations to one .
 : param iter input _ locations : An iterable of URLs or file paths pointing to BEL namespaces .
 : param str output _ path : The path to the file to write the merged namespace
 : param str namespace _ name : The namespace name
 : param str namespace _ keyword : Preferred BEL Keyword , maximum length of 8
 : param str namespace _ domain : One of : : data : ` pybel . constants . NAMESPACE _ DOMAIN _ BIOPROCESS ` ,
 : data : ` pybel . constants . NAMESPACE _ DOMAIN _ CHEMICAL ` ,
 : data : ` pybel . constants . NAMESPACE _ DOMAIN _ GENE ` , or
 : data : ` pybel . constants . NAMESPACE _ DOMAIN _ OTHER `
 : param str author _ name : The namespace ' s authors
 : param str citation _ name : The name of the citation
 : param str namespace _ query _ url : HTTP URL to query for details on namespace values ( must be valid URL )
 : param str namespace _ description : Namespace description
 : param str namespace _ species : Comma - separated list of species taxonomy id ' s
 : param str namespace _ version : Namespace version
 : param str namespace _ created : Namespace public timestamp , ISO 8601 datetime
 : param str author _ contact : Namespace author ' s contact info / email address
 : param str author _ copyright : Namespace ' s copyright / license information
 : param str citation _ description : Citation description
 : param str citation _ url : URL to more citation information
 : param str citation _ version : Citation version
 : param str citation _ date : Citation publish timestamp , ISO 8601 Date
 : param bool case _ sensitive : Should this config file be interpreted as case - sensitive ?
 : param str delimiter : The delimiter between names and labels in this config file
 : param bool cacheable : Should this config file be cached ?
 : param functions : The encoding for the elements in this namespace
 : type functions : iterable of characters
 : param str value _ prefix : a prefix for each name
 : param sort _ key : A function to sort the values with : func : ` sorted `
 : param bool check _ keywords : Should all the keywords be the same ? Defaults to ` ` True ` `"""

implementation:
results = get_merged_namespace_names ( input_locations , check_keywords = check_keywords )
with open ( output_path , 'w' ) as file :
    write_namespace ( namespace_name = namespace_name , namespace_keyword = namespace_keyword , namespace_domain = namespace_domain , author_name = author_name , citation_name = citation_name , values = results , namespace_species = namespace_species , namespace_description = namespace_description , namespace_query_url = namespace_query_url , namespace_version = namespace_version , namespace_created = namespace_created , author_contact = author_contact , author_copyright = author_copyright , citation_description = citation_description , citation_url = citation_url , citation_version = citation_version , citation_date = citation_date , case_sensitive = case_sensitive , delimiter = delimiter , cacheable = cacheable , functions = functions , value_prefix = value_prefix , sort_key = sort_key , file = file )


signature:
def run_rcr ( graph , tag = 'dgxp' ) :
    """Run the reverse causal reasoning algorithm on a graph .
 Steps :
 1 . Get all downstream controlled things into map ( that have at least 4 downstream things )
 2 . calculate population of all things that are downstream controlled
 . . note : : Assumes all nodes have been pre - tagged with data
 : param pybel . BELGraph graph :
 : param str tag : The key for the nodes ' data dictionaries that corresponds to the integer value for its differential
 expression ."""

implementation:
# Step 1 : Calculate the hypothesis subnetworks ( just simple star graphs )
hypotheses = defaultdict ( set )
increases = defaultdict ( set )
decreases = defaultdict ( set )
for u , v , d in graph . edges ( data = True ) :
    hypotheses [ u ] . add ( v )
    if d [ RELATION ] in CAUSAL_INCREASE_RELATIONS :
        increases [ u ] . add ( v )
    elif d [ RELATION ] in CAUSAL_DECREASE_RELATIONS :
        decreases [ u ] . add ( v )
# Step 2 : Calculate the matching of the data points to the causal relationships
# : A dictionary from { tuple controller node : int count of correctly matching observations }
correct = defaultdict ( int )
# : A dictionary from { tuple controller node : int count of incorrectly matching observations }
contra = defaultdict ( int )
# : A dictionary from { tuple controller node : int count of ambiguous observations }
ambiguous = defaultdict ( int )
# : A dictionary from { tuple controller node : int count of missing obvservations }
missing = defaultdict ( int )
for controller , downstream_nodes in hypotheses . items ( ) :
    if len ( downstream_nodes ) < 4 :
        continue
    # need enough data to make reasonable calculations !
    for node in downstream_nodes :
        if node in increases [ controller ] and node in decreases [ controller ] :
            ambiguous [ controller ] += 1
        elif node in increases [ controller ] :
            if graph . node [ node ] [ tag ] == 1 :
                correct [ controller ] += 1
            elif graph . node [ node ] [ tag ] == - 1 :
                contra [ controller ] += 1
        elif node in decreases [ controller ] :
            if graph . node [ node ] [ tag ] == 1 :
                contra [ controller ] += 1
            elif graph . node [ node ] [ tag ] == - 1 :
                correct [ controller ] += 1
        else :
            missing [ controller ] += 1
# Step 3 : Keep only controller nodes who have 4 or more downstream nodes
controllers = { controller for controller , downstream_nodes in hypotheses . items ( ) if 4 <= len ( downstream_nodes ) }
# Step 4 : Calculate concordance scores
concordance_scores = { controller : scipy . stats . beta ( 0.5 , correct [ controller ] , contra [ controller ] ) for controller in controllers }
# Step 5 : Calculate richness scores
# TODO
# Calculate the population as the union of all downstream nodes for all controllers
population = { node for controller in controllers for node in hypotheses [ controller ] }
population_size = len ( population )
# Step 6 : Export
return pandas . DataFrame ( { 'contra' : contra , 'correct' : correct , 'concordance' : concordance_scores } )


signature:
def export_namespace ( graph , namespace , directory = None , cacheable = False ) :
    """Exports all names and missing names from the given namespace to its own BEL Namespace files in the given
 directory .
 Could be useful during quick and dirty curation , where planned namespace building is not a priority .
 : param pybel . BELGraph graph : A BEL graph
 : param str namespace : The namespace to process
 : param str directory : The path to the directory where to output the namespace . Defaults to the current working
 directory returned by : func : ` os . getcwd `
 : param bool cacheable : Should the namespace 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
path = os . path . join ( directory , '{}.belns' . format ( namespace ) )
with open ( path , 'w' ) as file :
    log . info ( 'Outputting to %s' , path )
    right_names = get_names_by_namespace ( graph , namespace )
    log . info ( 'Graph has %d correct names in %s' , len ( right_names ) , namespace )
    wrong_names = get_incorrect_names_by_namespace ( graph , namespace )
    log . info ( 'Graph has %d incorrect names in %s' , len ( right_names ) , namespace )
    undefined_ns_names = get_undefined_namespace_names ( graph , namespace )
    log . info ( 'Graph has %d names in missing namespace %s' , len ( right_names ) , namespace )
    names = ( right_names | wrong_names | undefined_ns_names )
    if 0 == len ( names ) :
        log . warning ( '%s is empty' , namespace )
    write_namespace ( namespace_name = namespace , namespace_keyword = namespace , namespace_domain = 'Other' , author_name = graph . authors , author_contact = graph . contact , citation_name = graph . name , values = names , cacheable = cacheable , file = file )