def run(output_path, graph_type, force,
seed, num_nodes, edge_prob, solution_path):
any_op_file_exists = (P.exists(output_path) or P.exists(solution_path))
if any_op_file_exists and not force:
print('Cannot overwrite without --force', file=sys.stderr)
sys.exit(-1)
g = None
if graph_type == 'erdos':
g = nx.erdos_renyi_graph(num_nodes, edge_prob,
seed=seed, directed=True)
else:
print('Unknown graph type: ', graph_type, file=sys.stderr)
sys.exit(-1)
A = np.zeros((num_nodes, num_nodes), dtype='float')
# All edges are given uniformly random weights.
for u, v, d in g.edges(data=True):
d['act_prob'] = R.random()
A[u, v] = d['act_prob']
nx.write_edgelist(g, output_path)
np.savetxt(solution_path, A, delimiter=',')
python类erdos_renyi_graph()的实例源码
def test_multi(self):
# Network topology
g = nx.erdos_renyi_graph(1000, 0.1)
# Model selection
model1 = sir.SIRModel(g)
# Model Configuration
config = mc.Configuration()
config.add_model_parameter('beta', 0.001)
config.add_model_parameter('gamma', 0.01)
config.add_model_parameter("percentage_infected", 0.05)
model1.set_initial_status(config)
# Simulation multiple execution
trends = multi_runs(model1, execution_number=10, iteration_number=100, infection_sets=None, nprocesses=4)
self.assertIsNotNone(trends)
def test_multi_initial_set(self):
# Network topology
g = nx.erdos_renyi_graph(1000, 0.1)
# Model selection
model1 = sir.SIRModel(g)
# Model Configuration
config = mc.Configuration()
config.add_model_parameter('beta', 0.001)
config.add_model_parameter('gamma', 0.01)
model1.set_initial_status(config)
# Simulation multiple execution
infection_sets = [(1, 2, 3, 4, 5), (3, 23, 22, 54, 2), (98, 2, 12, 26, 3), (4, 6, 9)]
trends = multi_runs(model1, execution_number=4, iteration_number=100, infection_sets=infection_sets,
nprocesses=4)
self.assertIsNotNone(trends)
def test_visualize_dynamic(self):
dg = dn.DynGraph()
for t in past.builtins.xrange(0, 4):
g = nx.erdos_renyi_graph(200, 0.05)
dg.add_interactions_from(g.edges(), t)
model = dsi.DynSIModel(dg)
config = mc.Configuration()
config.add_model_parameter('beta', 0.1)
config.add_model_parameter("percentage_infected", 0.1)
model.set_initial_status(config)
iterations = model.execute_snapshots()
trends = model.build_trends(iterations)
# Visualization
viz = DiffusionPrevalence(model, trends)
viz.plot("prevd.pdf")
os.remove("prevd.pdf")
def test_DynSI(self):
dg = dn.DynGraph()
for t in past.builtins.xrange(0, 3):
g = nx.erdos_renyi_graph(200, 0.05)
dg.add_interactions_from(g.edges(), t)
model = si.DynSIModel(dg)
config = mc.Configuration()
config.add_model_parameter('beta', 0.1)
config.add_model_parameter("percentage_infected", 0.1)
model.set_initial_status(config)
iterations = model.execute_snapshots()
self.assertEqual(len(iterations), 3)
iterations = model.execute_iterations()
trends = model.build_trends(iterations)
self.assertEqual(len(trends[0]['trends']['status_delta'][1]),
len([x for x in dg.stream_interactions() if x[2] == "+"]))
def test_DynSIR(self):
dg = dn.DynGraph()
for t in past.builtins.xrange(0, 3):
g = nx.erdos_renyi_graph(200, 0.05)
dg.add_interactions_from(g.edges(), t)
model = sir.DynSIRModel(dg)
config = mc.Configuration()
config.add_model_parameter('beta', 0.1)
config.add_model_parameter('gamma', 0.1)
config.add_model_parameter("percentage_infected", 0.1)
model.set_initial_status(config)
iterations = model.execute_snapshots()
self.assertEqual(len(iterations), 3)
iterations = model.execute_iterations()
trends = model.build_trends(iterations)
self.assertEqual(len(trends[0]['trends']['status_delta'][1]),
len([x for x in dg.stream_interactions() if x[2] == "+"]))
def test_DynProfile(self):
dg = dn.DynGraph()
for t in past.builtins.xrange(0, 3):
g = nx.erdos_renyi_graph(200, 0.05)
dg.add_interactions_from(g.edges(), t)
model = pro.DynProfileModel(dg)
config = mc.Configuration()
config.add_model_parameter("percentage_infected", 0.1)
config.add_model_parameter("blocked", 0.1)
config.add_model_parameter("adopter_rate", 0.001)
profile = 0.1
for i in g.nodes():
config.add_node_configuration("profile", i, profile)
model.set_initial_status(config)
model.set_initial_status(config)
iterations = model.execute_snapshots()
self.assertEqual(len(iterations), 3)
def test_DynProfileThreshold(self):
dg = dn.DynGraph()
for t in past.builtins.xrange(0, 3):
g = nx.erdos_renyi_graph(200, 0.05)
dg.add_interactions_from(g.edges(), t)
model = prTr.DynProfileThresholdModel(dg)
config = mc.Configuration()
config.add_model_parameter("percentage_infected", 0.1)
config.add_model_parameter("blocked", 0.1)
config.add_model_parameter("adopter_rate", 0.001)
threshold = 0.2
profile = 0.1
for i in g.nodes():
config.add_node_configuration("threshold", i, threshold)
config.add_node_configuration("profile", i, profile)
model.set_initial_status(config)
model.set_initial_status(config)
iterations = model.execute_snapshots()
self.assertEqual(len(iterations), 3)
def test_seis_model(self):
g = nx.erdos_renyi_graph(1000, 0.1)
model = seis.SEISModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.5)
config.add_model_parameter('lambda', 0.2)
config.add_model_parameter('alpha', 0.05)
config.add_model_parameter("percentage_infected", 0.1)
model.set_initial_status(config)
iterations = model.iteration_bunch(10)
self.assertEqual(len(iterations), 10)
g = g.to_directed()
model = seis.SEISModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.5)
config.add_model_parameter('lambda', 0.8)
config.add_model_parameter('alpha', 0.5)
config.add_model_parameter("percentage_infected", 0.1)
model.set_initial_status(config)
iterations = model.iteration_bunch(10, node_status=False)
self.assertEqual(len(iterations), 10)
def setUp( self ):
'''Set up the experimental parameters and experiment.'''
# single experiment
self._params = dict(pInfect = 0.1,
pInfected = 0.01,
pRecover = 0.05)
self._network = networkx.erdos_renyi_graph(1000, 0.005)
# lab run
self._lab = epyc.Lab()
self._lab['pInfect'] = [ 0.1, 0.2, 0.3 ]
self._lab['pInfected'] = [ 0.01 ]
self._lab['pRecover'] = [ 0.05, 0.1, 1 ]
# model
self._model = SIS()
# maximum time needed as disease may be endemic
self._maxTime = 2000
def setUp( self ):
'''Set up the experimental parameters and experiment.'''
# single experiment
self._params = dict(pInfect = 0.1,
pInfected = 0.01,
pRemove = 0.05)
self._network = networkx.erdos_renyi_graph(1000, 0.005)
# lab run
self._lab = epyc.Lab()
self._lab['pInfect'] = [ 0.1, 0.2, 0.3 ]
self._lab['pInfected'] = [ 0.01 ]
self._lab['pRecover'] = [ 0.05, 0.1, 1 ]
# model
self._model = SIR()
# no maximum time needed
self._maxTime = None
def test_returns_plausible_results(self):
g = nx.erdos_renyi_graph(100, 0.3)
g_5 = graphpca.reduce_graph_efficiently(g, 5)
self.assertEqual(len(g_5), 5)
self.assertEqual(len(g_5[0]), 100)
for i in range(5):
max_val = max(abs(g_5[i]))
self.assertGreater(max_val, 0.01)
def test_ok_if_multiple_zero_eigens(self):
g = nx.erdos_renyi_graph(100, 0.3)
node = next(g.nodes_iter())
for neighbor in g.neighbors(node):
g.remove_edge(node, neighbor)
g_5 = graphpca.reduce_graph_efficiently(g, 5)
self.assertEqual(len(g_5), 5)
self.assertEqual(len(g_5[0]), 100)
for i in range(5):
max_val = max(abs(g_5[i]))
self.assertGreater(max_val, 0.01)
def test_similar_output_to_naive_small(self):
G = nx.erdos_renyi_graph(10, 0.5)
G2 = graphpca.reduce_graph_efficiently(G, 2)
G2n = graphpca.reduce_graph_naively(G, 2)
self.assertTrue(np.allclose(G2, G2n, rtol=1e-04, atol=1e-06),
'Regular result:\n{}\nNaive result:\n{}\n'.format(G2, G2n))
def test_similar_output_to_naive_big(self):
G = nx.erdos_renyi_graph(1001, 0.02)
G2 = graphpca.reduce_graph_efficiently(G, 2)
G2n = graphpca.reduce_graph_naively(G, 2)
self.assertTrue(np.allclose(G2, G2n, rtol=1e-03, atol=1e-05),
'Regular result:\n{}\nNaive result:\n{}\n'.format(G2, G2n))
def test_add_supernode_similar_output_to_naive_small(self):
G = nx.erdos_renyi_graph(10, 0.5)
G2 = graphpca.reduce_graph_efficiently(G, 2, add_supernode=True)
G2n = graphpca.reduce_graph_naively(G, 2)
self.assertTrue(np.allclose(G2, G2n, rtol=1e-02, atol=1e-06),
'Regular result:\n{}\nNaive result:\n{}\n'.format(G2, G2n))
def test_initialization():
"""
Tests initialization of plot object.
"""
n_nodes = 10
G = nx.erdos_renyi_graph(n=n_nodes, p=0.3) # noqa
b = BasePlot(graph=G)
assert len(b.nodes) == len(G.nodes())
def partition_at_level(dendrogram, level) :
"""Return the partition of the nodes at the given level
A dendrogram is a tree and each level is a partition of the graph nodes.
Level 0 is the first partition, which contains the smallest communities, and the best is len(dendrogram) - 1.
The higher the level is, the bigger are the communities
Parameters
----------
dendrogram : list of dict
a list of partitions, ie dictionnaries where keys of the i+1 are the values of the i.
level : int
the level which belongs to [0..len(dendrogram)-1]
Returns
-------
partition : dictionnary
A dictionary where keys are the nodes and the values are the set it belongs to
Raises
------
KeyError
If the dendrogram is not well formed or the level is too high
See Also
--------
best_partition which directly combines partition_at_level and generate_dendrogram to obtain the partition of highest modularity
Examples
--------
>>> G=nx.erdos_renyi_graph(100, 0.01)
>>> dendo = generate_dendrogram(G)
>>> for level in range(len(dendo) - 1) :
>>> print "partition at level", level, "is", partition_at_level(dendo, level)
"""
partition = dendrogram[0].copy()
for index in range(1, level + 1) :
for node, community in partition.items() :
partition[node] = dendrogram[index][community]
return partition
def partition_at_level(dendrogram, level) :
"""Return the partition of the nodes at the given level
A dendrogram is a tree and each level is a partition of the graph nodes.
Level 0 is the first partition, which contains the smallest communities, and the best is len(dendrogram) - 1.
The higher the level is, the bigger are the communities
Parameters
----------
dendrogram : list of dict
a list of partitions, ie dictionnaries where keys of the i+1 are the values of the i.
level : int
the level which belongs to [0..len(dendrogram)-1]
Returns
-------
partition : dictionnary
A dictionary where keys are the nodes and the values are the set it belongs to
Raises
------
KeyError
If the dendrogram is not well formed or the level is too high
See Also
--------
best_partition which directly combines partition_at_level and generate_dendrogram to obtain the partition of highest modularity
Examples
--------
>>> G=nx.erdos_renyi_graph(100, 0.01)
>>> dendo = generate_dendrogram(G)
>>> for level in range(len(dendo) - 1) :
>>> print "partition at level", level, "is", partition_at_level(dendo, level)
"""
partition = dendrogram[0].copy()
for index in range(1, level + 1) :
for node, community in partition.items() :
partition[node] = dendrogram[index][community]
return partition
def partition_at_level(dendrogram, level) :
"""Return the partition of the nodes at the given level
A dendrogram is a tree and each level is a partition of the graph nodes.
Level 0 is the first partition, which contains the smallest communities, and the best is len(dendrogram) - 1.
The higher the level is, the bigger are the communities
Parameters
----------
dendrogram : list of dict
a list of partitions, ie dictionnaries where keys of the i+1 are the values of the i.
level : int
the level which belongs to [0..len(dendrogram)-1]
Returns
-------
partition : dictionnary
A dictionary where keys are the nodes and the values are the set it belongs to
Raises
------
KeyError
If the dendrogram is not well formed or the level is too high
See Also
--------
best_partition which directly combines partition_at_level and generate_dendrogram to obtain the partition of highest modularity
Examples
--------
>>> G=nx.erdos_renyi_graph(100, 0.01)
>>> dendo = generate_dendrogram(G)
>>> for level in range(len(dendo) - 1) :
>>> print "partition at level", level, "is", partition_at_level(dendo, level)
"""
partition = dendrogram[0].copy()
for index in range(1, level + 1) :
for node, community in partition.items() :
partition[node] = dendrogram[index][community]
return partition
def test_visualize(self):
g = nx.erdos_renyi_graph(1000, 0.1)
model = sir.SIRModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.001)
config.add_model_parameter('gamma', 0.01)
config.add_model_parameter("percentage_infected", 0.05)
model.set_initial_status(config)
iterations = model.iteration_bunch(200)
trends = model.build_trends(iterations)
# Visualization
viz = DiffusionTrend(model, trends)
viz.plot("diffusion.pdf")
os.remove("diffusion.pdf")
def test_visualize_prevalence(self):
g = nx.erdos_renyi_graph(1000, 0.1)
model = sir.SIRModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.001)
config.add_model_parameter('gamma', 0.01)
config.add_model_parameter("percentage_infected", 0.05)
model.set_initial_status(config)
iterations = model.iteration_bunch(200)
trends = model.build_trends(iterations)
# Visualization
viz = DiffusionPrevalence(model, trends)
viz.plot("prev.pdf")
os.remove("prev.pdf")
def test_trend_comparison(self):
# Network topology
g = nx.erdos_renyi_graph(1000, 0.1)
# Model selection
model = sir.SIRModel(g)
# Model Configuration
cfg = mc.Configuration()
cfg.add_model_parameter('beta', 0.001)
cfg.add_model_parameter('gamma', 0.02)
cfg.add_model_parameter("percentage_infected", 0.01)
model.set_initial_status(cfg)
iterations = model.iteration_bunch(200)
trends = model.build_trends(iterations)
model1 = si.SIModel(g)
cfg = mc.Configuration()
cfg.add_model_parameter('beta', 0.001)
cfg.add_model_parameter("percentage_infected", 0.01)
model1.set_initial_status(cfg)
iterations = model1.iteration_bunch(200)
trends1 = model1.build_trends(iterations)
viz = DiffusionTrendComparison([model, model1], [trends, trends1])
viz.plot("trend_comparison.pdf")
os.remove("trend_comparison.pdf")
def test_visualize(self):
g = nx.erdos_renyi_graph(1000, 0.1)
model = sir.SIRModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.001)
config.add_model_parameter('gamma', 0.01)
config.add_model_parameter("percentage_infected", 0.05)
model.set_initial_status(config)
iterations = model.iteration_bunch(200)
trends = model.build_trends(iterations)
viz = DiffusionTrend(model, trends)
p = viz.plot()
self.assertIsInstance(p, Figure)
def test_visualize_prevalence(self):
g = nx.erdos_renyi_graph(1000, 0.1)
model = sir.SIRModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.001)
config.add_model_parameter('gamma', 0.01)
config.add_model_parameter("percentage_infected", 0.05)
model.set_initial_status(config)
iterations = model.iteration_bunch(200)
trends = model.build_trends(iterations)
viz = DiffusionPrevalence(model, trends)
p = viz.plot()
self.assertIsInstance(p, Figure)
def test_multi(self):
vm = MultiPlot()
g = nx.erdos_renyi_graph(1000, 0.1)
model = sir.SIRModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.001)
config.add_model_parameter('gamma', 0.01)
config.add_model_parameter("percentage_infected", 0.05)
model.set_initial_status(config)
iterations = model.iteration_bunch(200)
trends = model.build_trends(iterations)
viz = DiffusionTrend(model, trends)
p = viz.plot()
vm.add_plot(p)
g = nx.erdos_renyi_graph(1000, 0.1)
model = sir.SIRModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.001)
config.add_model_parameter('gamma', 0.01)
config.add_model_parameter("percentage_infected", 0.05)
model.set_initial_status(config)
iterations = model.iteration_bunch(200)
trends = model.build_trends(iterations)
viz = DiffusionPrevalence(model, trends)
p1 = viz.plot()
vm.add_plot(p1)
m = vm.plot()
self.assertIsInstance(m, Column)
def test_si_model(self):
g = nx.erdos_renyi_graph(1000, 0.1)
model = si.SIModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.5)
config.add_model_parameter("percentage_infected", 0.1)
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_sir_model(self):
g = nx.erdos_renyi_graph(1000, 0.1)
model = sir.SIRModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.5)
config.add_model_parameter('gamma', 0.2)
config.add_model_parameter("percentage_infected", 0.1)
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_swir_model(self):
g = nx.erdos_renyi_graph(1000, 0.1)
model = swir.SWIRModel(g)
config = mc.Configuration()
config.add_model_parameter('kappa', 0.5)
config.add_model_parameter('mu', 0.2)
config.add_model_parameter('nu', 0.05)
config.add_model_parameter("percentage_infected", 0.1)
model.set_initial_status(config)
iterations = model.iteration_bunch(10)
self.assertEqual(len(iterations), 10)
def test_sis_model(self):
g = nx.erdos_renyi_graph(1000, 0.1)
model = sis.SISModel(g)
config = mc.Configuration()
config.add_model_parameter('beta', 0.5)
config.add_model_parameter('lambda', 0.2)
config.add_model_parameter("percentage_infected", 0.1)
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)
