-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathutilities_antichain.py
121 lines (99 loc) · 3.15 KB
/
utilities_antichain.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
# -*- coding: utf-8 -*-
"""
Created on Wed Jun 20 12:52:55 2018
@author: Vaiva & Tim
"""
from collections import defaultdict
import itertools
import networkx as nx
import numpy as np
import scipy.special as ss
from scipy.stats import hypergeom
import math
import random
import matplotlib.pyplot as plt
import os
import copy
def overlappiness(graph,nodelist,neighbours ="successors"):
"""
Overlappiness is the amount of overlap between neighours of nodes in a nodelist.
Works for predecessors/successors
Implemented for antichains
"""
nset = set()
if neighbours == "successors":
for n in nodelist:
nset.update(set(graph.successors(n)))
try:
return max([graph.out_degree(n) for n in nodelist])/len(nset)
except:
return 0
else:
for n in nodelist:
nset.update(set(graph.predecessors(n)))
try:
return max([graph.in_degree(n) for n in nodelist])/len(nset)
except:
return 0
def tr(DAG, output=False):
"""
Transitive reduction,courtesy of JC
"""
# for printing progress
E = DAG.number_of_edges()
i = 0
print_limit = 10
print_counter = print_limit
edges = list(DAG.edges())
#########################
for edge in edges:
# check edge is necessary for causal structure
[a, b] = edge
DAG.remove_edge(a, b)
if not nx.has_path(DAG, a, b):
DAG.add_edge(a, b)
if output:
i += 1
pc = (i/float(E))*100
if pc > print_counter:
print ('Finished %s percent' % int(math.floor(pc)))
print_counter += print_limit
return DAG
def is_antichain(graph,nodelist):
"""
Tests whether a list of nodes in a graph is an antichain.
(TSE tidy version)
Note this works for any type of graph but only make sense for directed or directed acyclic graphs.
For a simple graph, the maximal antichains are just the components.
Parameters
----------
graph = networkx graph
nodelist = list of nodes suggested as an antichain
Returns
-------
Bool:
True if the nodelist is an antichain in graph
False if the nodelist is not an antichain in graph
"""
for n,m in itertools.combinations(nodelist,2):
if nx.has_path(graph,m,n) or nx.has_path(graph,n,m):
return False
return True
def is_weakly_connected(graph,source_nodes,target_nodes):
"""
Tests whether a list of source nodes in a graph have a path in either direction between a list of target nodes.
Parameters
----------
graph = networkx graph
source_nodes = list of source nodes for paths
target_nodes = list of target nodes for paths
Returns
-------
Bool:
True if there is a path from at least one source node to at least one target node or the other way round
False otherwise
"""
for s,t in itertools.product(source_nodes,target_nodes):
if nx.has_path(graph,s,t) or nx.has_path(graph,t,s):
return True
return False