### comparison env/lib/python3.9/site-packages/networkx/algorithms/flow/tests/test_maxflow_large_graph.py @ 0:4f3585e2f14bdraftdefaulttip

author shellac Mon, 22 Mar 2021 18:12:50 +0000
comparison
equal inserted replaced
-1:000000000000 0:4f3585e2f14b
1 """Maximum flow algorithms test suite on large graphs.
2 """
3
4 import os
5 import pytest
6
7 import networkx as nx
8 from networkx.algorithms.flow import build_flow_dict, build_residual_network
9 from networkx.algorithms.flow import boykov_kolmogorov
10 from networkx.algorithms.flow import dinitz
11 from networkx.algorithms.flow import edmonds_karp
12 from networkx.algorithms.flow import preflow_push
13 from networkx.algorithms.flow import shortest_augmenting_path
14 from networkx.testing import almost_equal
15
16 flow_funcs = [
17 boykov_kolmogorov,
18 dinitz,
19 edmonds_karp,
20 preflow_push,
21 shortest_augmenting_path,
22 ]
23
24
25 def gen_pyramid(N):
26 # This graph admits a flow of value 1 for which every arc is at
27 # capacity (except the arcs incident to the sink which have
28 # infinite capacity).
29 G = nx.DiGraph()
30
31 for i in range(N - 1):
32 cap = 1.0 / (i + 2)
33 for j in range(i + 1):
34 G.add_edge((i, j), (i + 1, j), capacity=cap)
35 cap = 1.0 / (i + 1) - cap
36 G.add_edge((i, j), (i + 1, j + 1), capacity=cap)
37 cap = 1.0 / (i + 2) - cap
38
39 for j in range(N):
40 G.add_edge((N - 1, j), "t")
41
42 return G
43
44
46 dirname = os.path.dirname(__file__)
47 path = os.path.join(dirname, name + ".gpickle.bz2")
49
50
51 def validate_flows(G, s, t, soln_value, R, flow_func):
52 flow_value = R.graph["flow_value"]
53 flow_dict = build_flow_dict(G, R)
54 errmsg = f"Assertion failed in function: {flow_func.__name__}"
55 assert soln_value == flow_value, errmsg
56 assert set(G) == set(flow_dict), errmsg
57 for u in G:
58 assert set(G[u]) == set(flow_dict[u]), errmsg
59 excess = {u: 0 for u in flow_dict}
60 for u in flow_dict:
61 for v, flow in flow_dict[u].items():
62 assert flow <= G[u][v].get("capacity", float("inf")), errmsg
63 assert flow >= 0, errmsg
64 excess[u] -= flow
65 excess[v] += flow
66 for u, exc in excess.items():
67 if u == s:
68 assert exc == -soln_value, errmsg
69 elif u == t:
70 assert exc == soln_value, errmsg
71 else:
72 assert exc == 0, errmsg
73
74
75 class TestMaxflowLargeGraph:
76 def test_complete_graph(self):
77 N = 50
78 G = nx.complete_graph(N)
79 nx.set_edge_attributes(G, 5, "capacity")
80 R = build_residual_network(G, "capacity")
81 kwargs = dict(residual=R)
82
83 for flow_func in flow_funcs:
84 kwargs["flow_func"] = flow_func
85 errmsg = f"Assertion failed in function: {flow_func.__name__}"
86 flow_value = nx.maximum_flow_value(G, 1, 2, **kwargs)
87 assert flow_value == 5 * (N - 1), errmsg
88
89 def test_pyramid(self):
90 N = 10
91 # N = 100 # this gives a graph with 5051 nodes
92 G = gen_pyramid(N)
93 R = build_residual_network(G, "capacity")
94 kwargs = dict(residual=R)
95
96 for flow_func in flow_funcs:
97 kwargs["flow_func"] = flow_func
98 errmsg = f"Assertion failed in function: {flow_func.__name__}"
99 flow_value = nx.maximum_flow_value(G, (0, 0), "t", **kwargs)
100 assert almost_equal(flow_value, 1.0), errmsg
101
102 def test_gl1(self):
104 s = 1
105 t = len(G)
106 R = build_residual_network(G, "capacity")
107 kwargs = dict(residual=R)
108
109 # do one flow_func to save time
110 flow_func = flow_funcs
111 validate_flows(G, s, t, 156545, flow_func(G, s, t, **kwargs), flow_func)
112
113 # for flow_func in flow_funcs:
114 # validate_flows(G, s, t, 156545, flow_func(G, s, t, **kwargs),
115 # flow_func)
116
117 @pytest.mark.slow
118 def test_gw1(self):
120 s = 1
121 t = len(G)
122 R = build_residual_network(G, "capacity")
123 kwargs = dict(residual=R)
124
125 for flow_func in flow_funcs:
126 validate_flows(G, s, t, 1202018, flow_func(G, s, t, **kwargs), flow_func)
127
128 def test_wlm3(self):
130 s = 1
131 t = len(G)
132 R = build_residual_network(G, "capacity")
133 kwargs = dict(residual=R)
134
135 # do one flow_func to save time
136 flow_func = flow_funcs
137 validate_flows(G, s, t, 11875108, flow_func(G, s, t, **kwargs), flow_func)
138
139 # for flow_func in flow_funcs:
140 # validate_flows(G, s, t, 11875108, flow_func(G, s, t, **kwargs),
141 # flow_func)
142
143 def test_preflow_push_global_relabel(self):