def test_sznajd_model(self):
g = nx.complete_graph(100)
model = sm.SznajdModel(g)
config = mc.Configuration()
config.add_model_parameter("percentage_infected", 0.2)
model.set_initial_status(config)
iterations = model.iteration_bunch(10)
self.assertEqual(len(iterations), 10)
iterations = model.iteration_bunch(10, node_status=False)
self.assertEqual(len(iterations), 10)
g = nx.complete_graph(100)
g = g.to_directed()
model = sm.SznajdModel(g)
config = mc.Configuration()
config.add_model_parameter("percentage_infected", 0.2)
model.set_initial_status(config)
iterations = model.iteration_bunch(10)
self.assertEqual(len(iterations), 10)
iterations = model.iteration_bunch(10, node_status=False)
self.assertEqual(len(iterations), 10)
python类complete_graph()的实例源码
def test_cognitive_model(self):
g = nx.complete_graph(100)
model = cm.CognitiveOpDynModel(g)
config = mc.Configuration()
config.add_model_parameter("I", 0.15)
config.add_model_parameter("B_range_min", 0)
config.add_model_parameter("B_range_max", 1)
config.add_model_parameter("T_range_min", 0)
config.add_model_parameter("T_range_max", 1)
config.add_model_parameter("R_fraction_negative", 1.0 / 3)
config.add_model_parameter("R_fraction_neutral", 1.0 / 3)
config.add_model_parameter("R_fraction_positive", 1.0 / 3)
model.set_initial_status(config)
iterations = model.iteration_bunch(10)
self.assertEqual(len(iterations), 10)
iterations = model.iteration_bunch(10, node_status=False)
self.assertEqual(len(iterations), 10)
def test_kertesz_model_predefined_blocked(self):
g = nx.complete_graph(100)
model = ks.KerteszThresholdModel(g)
config = mc.Configuration()
config.add_model_parameter('adopter_rate', 0.4)
predefined_blocked = [0, 1, 2, 3, 4, 5]
config.add_model_initial_configuration("Blocked", predefined_blocked)
config.add_model_parameter('percentage_infected', 0.1)
threshold = 0.2
for i in g.nodes():
config.add_node_configuration("threshold", i, threshold)
model.set_initial_status(config)
iteration = model.iteration()
blocked = [x for x, v in future.utils.iteritems(iteration['status']) if v == -1]
self.assertEqual(blocked, predefined_blocked)
def induced_graph(partition, graph) :
"""Produce the graph where nodes are the communities
there is a link of weight w between communities if the sum of the weights of the links between their elements is w
Parameters
----------
partition : dict
a dictionary where keys are graph nodes and values the part the node belongs to
graph : networkx.Graph
the initial graph
Returns
-------
g : networkx.Graph
a networkx graph where nodes are the parts
Examples
--------
>>> n = 5
>>> g = nx.complete_graph(2*n)
>>> part = dict([])
>>> for node in g.nodes() :
>>> part[node] = node % 2
>>> ind = induced_graph(part, g)
>>> goal = nx.Graph()
>>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)])
>>> nx.is_isomorphic(int, goal)
True
"""
ret = nx.Graph()
ret.add_nodes_from(partition.values())
for node1, node2, datas in graph.edges_iter(data = True) :
weight = datas.get("weight", 1)
com1 = partition[node1]
com2 = partition[node2]
w_prec = ret.get_edge_data(com1, com2, {"weight":0}).get("weight", 1)
ret.add_edge(com1, com2, weight = w_prec + weight)
return ret
def induced_graph(partition, graph) :
"""Produce the graph where nodes are the communities
there is a link of weight w between communities if the sum of the weights of the links between their elements is w
Parameters
----------
partition : dict
a dictionary where keys are graph nodes and values the part the node belongs to
graph : networkx.Graph
the initial graph
Returns
-------
g : networkx.Graph
a networkx graph where nodes are the parts
Examples
--------
>>> n = 5
>>> g = nx.complete_graph(2*n)
>>> part = dict([])
>>> for node in g.nodes() :
>>> part[node] = node % 2
>>> ind = induced_graph(part, g)
>>> goal = nx.Graph()
>>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)])
>>> nx.is_isomorphic(int, goal)
True
"""
ret = nx.Graph()
ret.add_nodes_from(partition.values())
for node1, node2, datas in graph.edges_iter(data = True) :
weight = datas.get("weight", 1)
com1 = partition[node1]
com2 = partition[node2]
w_prec = ret.get_edge_data(com1, com2, {"weight":0}).get("weight", 1)
ret.add_edge(com1, com2, weight = w_prec + weight)
return ret
def induced_graph(partition, graph) :
"""Produce the graph where nodes are the communities
there is a link of weight w between communities if the sum of the weights of the links between their elements is w
Parameters
----------
partition : dict
a dictionary where keys are graph nodes and values the part the node belongs to
graph : networkx.Graph
the initial graph
Returns
-------
g : networkx.Graph
a networkx graph where nodes are the parts
Examples
--------
>>> n = 5
>>> g = nx.complete_graph(2*n)
>>> part = dict([])
>>> for node in g.nodes() :
>>> part[node] = node % 2
>>> ind = induced_graph(part, g)
>>> goal = nx.Graph()
>>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)])
>>> nx.is_isomorphic(int, goal)
True
"""
ret = nx.Graph()
ret.add_nodes_from(partition.values())
for node1, node2, datas in graph.edges_iter(data = True) :
weight = datas.get("weight", 1)
com1 = partition[node1]
com2 = partition[node2]
w_prec = ret.get_edge_data(com1, com2, {"weight":0}).get("weight", 1)
ret.add_edge(com1, com2, weight = w_prec + weight)
return ret
def Link_Prediction(index, G, ebunch=None):
#G = nx.complete_graph(5)
if index == "RA":
Rank_List = resource_allocation_index(G, ebunch)
if index == "AA":
Rank_List = adamic_adar_index(G, ebunch)
if index == "CN":
Rank_List = common_neighbor_index(G, ebunch)
if index == "LP":
Rank_List = local_path_index(G, ebunch)
if index == "SD":
Rank_List = structure_dependent_index(G, ebunch)
return Rank_List
def test_voter_model(self):
g = nx.complete_graph(100)
model = vm.VoterModel(g)
config = mc.Configuration()
config.add_model_parameter("percentage_infected", 0.2)
model.set_initial_status(config)
iterations = model.iteration_bunch(10)
self.assertEqual(len(iterations), 10)
iterations = model.iteration_bunch(10, node_status=False)
self.assertEqual(len(iterations), 10)
def test_majorityrule_model(self):
g = nx.complete_graph(100)
model = mrm.MajorityRuleModel(g)
config = mc.Configuration()
config.add_model_parameter("q", 3)
config.add_model_parameter("percentage_infected", 0.2)
model.set_initial_status(config)
iterations = model.iteration_bunch(10)
self.assertEqual(len(iterations), 10)
iterations = model.iteration_bunch(10, node_status=False)
self.assertEqual(len(iterations), 10)
def test_qvoter_model(self):
g = nx.complete_graph(100)
model = qvm.QVoterModel(g)
config = mc.Configuration()
config.add_model_parameter("q", 5)
config.add_model_parameter("percentage_infected", 0.6)
model.set_initial_status(config)
iterations = model.iteration_bunch(10)
self.assertEqual(len(iterations), 10)
iterations = model.iteration_bunch(10, node_status=False)
self.assertEqual(len(iterations), 10)
def get_couling_derivate_matrix(self, h, twist_number, s):
if type(twist_number) == InPhase:
G = networkx.complete_graph(self.n)
d = np.zeros((self.n, self.n))
for ir in range(self.n):
ir_neigh = G.neighbors(ir)
d[ir, ir_neigh] = 1
d[ir, :] = d[ir, :] / np.sum(d[ir, :])
return d
else:
raise Exception('Topology not compatible with state')
def induced_graph(partition, graph, weight="weight"):
"""Produce the graph where nodes are the communities
there is a link of weight w between communities if the sum of the weights
of the links between their elements is w
Parameters
----------
partition : dict
a dictionary where keys are graph nodes and values the part the node
belongs to
graph : networkx.Graph
the initial graph
weight : str, optional
the key in graph to use as weight. Default to 'weight'
Returns
-------
g : networkx.Graph
a networkx graph where nodes are the parts
Examples
--------
>>> n = 5
>>> g = nx.complete_graph(2*n)
>>> part = dict([])
>>> for node in g.nodes() :
>>> part[node] = node % 2
>>> ind = induced_graph(part, g)
>>> goal = nx.Graph()
>>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)]) # NOQA
>>> nx.is_isomorphic(int, goal)
True
"""
ret = nx.Graph()
ret.add_nodes_from(partition.values())
for node1, node2, datas in graph.edges_iter(data=True):
edge_weight = datas.get(weight, 1)
com1 = partition[node1]
com2 = partition[node2]
w_prec = ret.get_edge_data(com1, com2, {weight: 0}).get(weight, 1)
ret.add_edge(com1, com2, attr_dict={weight: w_prec + edge_weight})
return ret
def induced_graph(partition, graph, weight="weight"):
"""Produce the graph where nodes are the communities
there is a link of weight w between communities if the sum of the weights
of the links between their elements is w
Parameters
----------
partition : dict
a dictionary where keys are graph nodes and values the part the node
belongs to
graph : networkx.Graph
the initial graph
weight : str, optional
the key in graph to use as weight. Default to 'weight'
Returns
-------
g : networkx.Graph
a networkx graph where nodes are the parts
Examples
--------
>>> n = 5
>>> g = nx.complete_graph(2*n)
>>> part = dict([])
>>> for node in g.nodes() :
>>> part[node] = node % 2
>>> ind = induced_graph(part, g)
>>> goal = nx.Graph()
>>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)]) # NOQA
>>> nx.is_isomorphic(int, goal)
True
"""
ret = nx.Graph()
ret.add_nodes_from(partition.values())
for node1, node2, datas in graph.edges_iter(data=True):
edge_weight = datas.get(weight, 1)
com1 = partition[node1]
com2 = partition[node2]
w_prec = ret.get_edge_data(com1, com2, {weight: 0}).get(weight, 1)
ret.add_edge(com1, com2, attr_dict={weight: w_prec + edge_weight})
return ret
generate_random_dag.py 文件源码
项目：breaking_cycles_in_noisy_hierarchies
作者: zhenv5
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def gnm_random_graph(n, m, seed=None, directed=True):
"""Return the random graph G_{n,m}.
Produces a graph picked randomly out of the set of all graphs
with n nodes and m edges.
Parameters
----------
n : int
The number of nodes.
m : int
The number of edges.
seed : int, optional
Seed for random number generator (default=None).
directed : bool, optional (default=False)
If True return a directed graph
"""
if directed:
G=nx.DiGraph()
g = nx.DiGraph()
else:
G=nx.Graph()
g = nx.Graph()
G.add_nodes_from(range(n))
G.name="gnm_random_graph(%s,%s)"%(n,m)
if seed is not None:
random.seed(seed)
if n==1:
return G
max_edges=n*(n-1)
if not directed:
max_edges/=2.0
if m>=max_edges:
return nx.complete_graph(n,create_using=G)
nlist=G.nodes()
edge_count=0
while edge_count < m:
# generate random edge,u,v
u = random.choice(nlist)
v = random.choice(nlist)
if u>=v or G.has_edge(u,v):
continue
else:
G.add_edge(u,v)
edge_count = edge_count+1
permutation = np.random.permutation(n)
#print permutation
new_edges = []
for e in G.edges():
u,v = e
new_edges.append((permutation[u],permutation[v]))
g.add_edges_from(new_edges)
print("is_directed_acyclic_graph: %s" % nx.is_directed_acyclic_graph(g))
return g
def induced_graph(partition, graph, weight="weight"):
"""Produce the graph where nodes are the communities
there is a link of weight w between communities if the sum of the weights
of the links between their elements is w
Parameters
----------
partition : dict
a dictionary where keys are graph nodes and values the part the node
belongs to
graph : networkx.Graph
the initial graph
weight : str, optional
the key in graph to use as weight. Default to 'weight'
Returns
-------
g : networkx.Graph
a networkx graph where nodes are the parts
Examples
--------
>>> n = 5
>>> g = nx.complete_graph(2*n)
>>> part = dict([])
>>> for node in g.nodes() :
>>> part[node] = node % 2
>>> ind = induced_graph(part, g)
>>> goal = nx.Graph()
>>> goal.add_weighted_edges_from([(0,1,n*n),(0,0,n*(n-1)/2), (1, 1, n*(n-1)/2)]) # NOQA
>>> nx.is_isomorphic(int, goal)
True
"""
ret = nx.Graph()
ret.add_nodes_from(partition.values())
for node1, node2, datas in graph.edges_iter(data=True):
edge_weight = datas.get(weight, 1)
com1 = partition[node1]
com2 = partition[node2]
w_prec = ret.get_edge_data(com1, com2, {weight: 0}).get(weight, 1)
ret.add_edge(com1, com2, attr_dict={weight: w_prec + edge_weight})
return ret
def test_config(self):
g = nx.erdos_renyi_graph(99, 0.1)
model = th.ThresholdModel(g)
config = mc.Configuration()
config.add_model_parameter('percentage_infected', 0.1)
config.add_model_initial_configuration("Infected", [1, 2, 3])
config.add_node_set_configuration("partial", {1: 1, 2: 2})
try:
model.set_initial_status(config)
except:
pass
config.add_edge_set_configuration("partial", {e: 1 for e in list(g.edges)[:10]})
try:
model.set_initial_status(config)
except:
pass
config.add_node_set_configuration("partial", {n: 1 for n in g.nodes})
config.add_edge_set_configuration("partial", {e: 1 for e in g.edges})
model.set_initial_status(config)
g = nx.complete_graph(100)
model = mrm.MajorityRuleModel(g)
config = mc.Configuration()
config.add_model_parameter("percentage_infected", 0.2)
try:
model.set_initial_status(config)
except:
pass
g = nx.erdos_renyi_graph(1000, 0.1)
model = ids.IndependentCascadesModel(g)
config = mc.Configuration()
config.add_model_parameter('percentage_infected', 0.1)
try:
model.set_initial_status(config)
except:
pass
g = nx.erdos_renyi_graph(1000, 0.1)
model = th.ThresholdModel(g)
config = mc.Configuration()
config.add_model_parameter('percentage_infected', 0.1)
try:
model.set_initial_status(config)
except:
pass
