summaryrefslogtreecommitdiff
path: root/networkx/algorithms/flow
diff options
context:
space:
mode:
authorJarrod Millman <jarrod.millman@gmail.com>2019-11-19 23:12:09 -0800
committerJarrod Millman <jarrod.millman@gmail.com>2019-11-19 23:36:38 -0800
commita9daddcd6dbd3a9323ed892d8d0c8236deb37e2f (patch)
treeed079a14afbf8c4f1b34510b26cbfee1e3576b6b /networkx/algorithms/flow
parentf34040c8e5c72294c4d9b99dad1ff515a0a06eaf (diff)
downloadnetworkx-a9daddcd6dbd3a9323ed892d8d0c8236deb37e2f.tar.gz
Update style
Diffstat (limited to 'networkx/algorithms/flow')
-rw-r--r--networkx/algorithms/flow/tests/test_gomory_hu.py55
-rw-r--r--networkx/algorithms/flow/tests/test_maxflow_large_graph.py76
2 files changed, 65 insertions, 66 deletions
diff --git a/networkx/algorithms/flow/tests/test_gomory_hu.py b/networkx/algorithms/flow/tests/test_gomory_hu.py
index b82a32f7..0d9ded84 100644
--- a/networkx/algorithms/flow/tests/test_gomory_hu.py
+++ b/networkx/algorithms/flow/tests/test_gomory_hu.py
@@ -18,10 +18,9 @@ flow_funcs = [
class TestGomoryHuTree:
-
def minimum_edge_weight(self, T, u, v):
- path = nx.shortest_path(T, u, v, weight='weight')
- return min((T[u][v]['weight'], (u, v)) for (u, v) in zip(path, path[1:]))
+ path = nx.shortest_path(T, u, v, weight="weight")
+ return min((T[u][v]["weight"], (u, v)) for (u, v) in zip(path, path[1:]))
def compute_cutset(self, G, T_orig, edge):
T = T_orig.copy()
@@ -34,61 +33,56 @@ class TestGomoryHuTree:
def test_default_flow_function_karate_club_graph(self):
G = nx.karate_club_graph()
- nx.set_edge_attributes(G, 1, 'capacity')
+ nx.set_edge_attributes(G, 1, "capacity")
T = nx.gomory_hu_tree(G)
assert nx.is_tree(T)
for u, v in combinations(G, 2):
cut_value, edge = self.minimum_edge_weight(T, u, v)
- assert (nx.minimum_cut_value(G, u, v) ==
- cut_value)
+ assert nx.minimum_cut_value(G, u, v) == cut_value
def test_karate_club_graph(self):
G = nx.karate_club_graph()
- nx.set_edge_attributes(G, 1, 'capacity')
+ nx.set_edge_attributes(G, 1, "capacity")
for flow_func in flow_funcs:
T = nx.gomory_hu_tree(G, flow_func=flow_func)
assert nx.is_tree(T)
for u, v in combinations(G, 2):
cut_value, edge = self.minimum_edge_weight(T, u, v)
- assert (nx.minimum_cut_value(G, u, v) ==
- cut_value)
+ assert nx.minimum_cut_value(G, u, v) == cut_value
def test_davis_southern_women_graph(self):
G = nx.davis_southern_women_graph()
- nx.set_edge_attributes(G, 1, 'capacity')
+ nx.set_edge_attributes(G, 1, "capacity")
for flow_func in flow_funcs:
T = nx.gomory_hu_tree(G, flow_func=flow_func)
assert nx.is_tree(T)
for u, v in combinations(G, 2):
cut_value, edge = self.minimum_edge_weight(T, u, v)
- assert (nx.minimum_cut_value(G, u, v) ==
- cut_value)
+ assert nx.minimum_cut_value(G, u, v) == cut_value
def test_florentine_families_graph(self):
G = nx.florentine_families_graph()
- nx.set_edge_attributes(G, 1, 'capacity')
+ nx.set_edge_attributes(G, 1, "capacity")
for flow_func in flow_funcs:
T = nx.gomory_hu_tree(G, flow_func=flow_func)
assert nx.is_tree(T)
for u, v in combinations(G, 2):
cut_value, edge = self.minimum_edge_weight(T, u, v)
- assert (nx.minimum_cut_value(G, u, v) ==
- cut_value)
+ assert nx.minimum_cut_value(G, u, v) == cut_value
def test_les_miserables_graph_cutset(self):
G = nx.les_miserables_graph()
- nx.set_edge_attributes(G, 1, 'capacity')
+ nx.set_edge_attributes(G, 1, "capacity")
for flow_func in flow_funcs:
T = nx.gomory_hu_tree(G, flow_func=flow_func)
assert nx.is_tree(T)
for u, v in combinations(G, 2):
cut_value, edge = self.minimum_edge_weight(T, u, v)
- assert (nx.minimum_cut_value(G, u, v) ==
- cut_value)
+ assert nx.minimum_cut_value(G, u, v) == cut_value
def test_karate_club_graph_cutset(self):
G = nx.karate_club_graph()
- nx.set_edge_attributes(G, 1, 'capacity')
+ nx.set_edge_attributes(G, 1, "capacity")
T = nx.gomory_hu_tree(G)
assert nx.is_tree(T)
u, v = 0, 33
@@ -99,18 +93,25 @@ class TestGomoryHuTree:
def test_wikipedia_example(self):
# Example from https://en.wikipedia.org/wiki/Gomory%E2%80%93Hu_tree
G = nx.Graph()
- G.add_weighted_edges_from((
- (0, 1, 1), (0, 2, 7), (1, 2, 1),
- (1, 3, 3), (1, 4, 2), (2, 4, 4),
- (3, 4, 1), (3, 5, 6), (4, 5, 2),
- ))
+ G.add_weighted_edges_from(
+ (
+ (0, 1, 1),
+ (0, 2, 7),
+ (1, 2, 1),
+ (1, 3, 3),
+ (1, 4, 2),
+ (2, 4, 4),
+ (3, 4, 1),
+ (3, 5, 6),
+ (4, 5, 2),
+ )
+ )
for flow_func in flow_funcs:
- T = nx.gomory_hu_tree(G, capacity='weight', flow_func=flow_func)
+ T = nx.gomory_hu_tree(G, capacity="weight", flow_func=flow_func)
assert nx.is_tree(T)
for u, v in combinations(G, 2):
cut_value, edge = self.minimum_edge_weight(T, u, v)
- assert (nx.minimum_cut_value(G, u, v, capacity='weight') ==
- cut_value)
+ assert nx.minimum_cut_value(G, u, v, capacity="weight") == cut_value
def test_directed_raises(self):
with pytest.raises(nx.NetworkXNotImplemented):
diff --git a/networkx/algorithms/flow/tests/test_maxflow_large_graph.py b/networkx/algorithms/flow/tests/test_maxflow_large_graph.py
index e152cb50..dceba17b 100644
--- a/networkx/algorithms/flow/tests/test_maxflow_large_graph.py
+++ b/networkx/algorithms/flow/tests/test_maxflow_large_graph.py
@@ -30,29 +30,27 @@ def gen_pyramid(N):
G = nx.DiGraph()
for i in range(N - 1):
- cap = 1. / (i + 2)
+ cap = 1.0 / (i + 2)
for j in range(i + 1):
- G.add_edge((i, j), (i + 1, j),
- capacity=cap)
- cap = 1. / (i + 1) - cap
- G.add_edge((i, j), (i + 1, j + 1),
- capacity=cap)
- cap = 1. / (i + 2) - cap
+ G.add_edge((i, j), (i + 1, j), capacity=cap)
+ cap = 1.0 / (i + 1) - cap
+ G.add_edge((i, j), (i + 1, j + 1), capacity=cap)
+ cap = 1.0 / (i + 2) - cap
for j in range(N):
- G.add_edge((N - 1, j), 't')
+ G.add_edge((N - 1, j), "t")
return G
def read_graph(name):
dirname = os.path.dirname(__file__)
- path = os.path.join(dirname, name + '.gpickle.bz2')
+ path = os.path.join(dirname, name + ".gpickle.bz2")
return nx.read_gpickle(path)
def validate_flows(G, s, t, soln_value, R, flow_func):
- flow_value = R.graph['flow_value']
+ flow_value = R.graph["flow_value"]
flow_dict = build_flow_dict(G, R)
assert soln_value == flow_value, msg.format(flow_func.__name__)
assert set(G) == set(flow_dict), msg.format(flow_func.__name__)
@@ -61,7 +59,9 @@ def validate_flows(G, s, t, soln_value, R, flow_func):
excess = {u: 0 for u in flow_dict}
for u in flow_dict:
for v, flow in flow_dict[u].items():
- assert flow <= G[u][v].get('capacity', float('inf')), msg.format(flow_func.__name__)
+ assert flow <= G[u][v].get("capacity", float("inf")), msg.format(
+ flow_func.__name__
+ )
assert flow >= 0, msg.format(flow_func.__name__)
excess[u] -= flow
excess[v] += flow
@@ -75,16 +75,15 @@ def validate_flows(G, s, t, soln_value, R, flow_func):
class TestMaxflowLargeGraph:
-
def test_complete_graph(self):
N = 50
G = nx.complete_graph(N)
- nx.set_edge_attributes(G, 5, 'capacity')
- R = build_residual_network(G, 'capacity')
+ nx.set_edge_attributes(G, 5, "capacity")
+ R = build_residual_network(G, "capacity")
kwargs = dict(residual=R)
for flow_func in flow_funcs:
- kwargs['flow_func'] = flow_func
+ kwargs["flow_func"] = flow_func
flow_value = nx.maximum_flow_value(G, 1, 2, **kwargs)
assert flow_value == 5 * (N - 1), msg.format(flow_func.__name__)
@@ -92,56 +91,55 @@ class TestMaxflowLargeGraph:
N = 10
# N = 100 # this gives a graph with 5051 nodes
G = gen_pyramid(N)
- R = build_residual_network(G, 'capacity')
+ R = build_residual_network(G, "capacity")
kwargs = dict(residual=R)
for flow_func in flow_funcs:
- kwargs['flow_func'] = flow_func
- flow_value = nx.maximum_flow_value(G, (0, 0), 't', **kwargs)
- assert almost_equal(flow_value, 1.), msg.format(flow_func.__name__)
+ kwargs["flow_func"] = flow_func
+ flow_value = nx.maximum_flow_value(G, (0, 0), "t", **kwargs)
+ assert almost_equal(flow_value, 1.0), msg.format(flow_func.__name__)
def test_gl1(self):
- G = read_graph('gl1')
+ G = read_graph("gl1")
s = 1
t = len(G)
- R = build_residual_network(G, 'capacity')
+ R = build_residual_network(G, "capacity")
kwargs = dict(residual=R)
# do one flow_func to save time
flow_func = flow_funcs[0]
- validate_flows(G, s, t, 156545, flow_func(G, s, t, **kwargs),
- flow_func)
-# for flow_func in flow_funcs:
-# validate_flows(G, s, t, 156545, flow_func(G, s, t, **kwargs),
-# flow_func)
+ validate_flows(G, s, t, 156545, flow_func(G, s, t, **kwargs), flow_func)
+
+ # for flow_func in flow_funcs:
+ # validate_flows(G, s, t, 156545, flow_func(G, s, t, **kwargs),
+ # flow_func)
def test_gw1(self):
- G = read_graph('gw1')
+ G = read_graph("gw1")
s = 1
t = len(G)
- R = build_residual_network(G, 'capacity')
+ R = build_residual_network(G, "capacity")
kwargs = dict(residual=R)
for flow_func in flow_funcs:
- validate_flows(G, s, t, 1202018, flow_func(G, s, t, **kwargs),
- flow_func)
+ validate_flows(G, s, t, 1202018, flow_func(G, s, t, **kwargs), flow_func)
def test_wlm3(self):
- G = read_graph('wlm3')
+ G = read_graph("wlm3")
s = 1
t = len(G)
- R = build_residual_network(G, 'capacity')
+ R = build_residual_network(G, "capacity")
kwargs = dict(residual=R)
# do one flow_func to save time
flow_func = flow_funcs[0]
- validate_flows(G, s, t, 11875108, flow_func(G, s, t, **kwargs),
- flow_func)
-# for flow_func in flow_funcs:
-# validate_flows(G, s, t, 11875108, flow_func(G, s, t, **kwargs),
-# flow_func)
+ validate_flows(G, s, t, 11875108, flow_func(G, s, t, **kwargs), flow_func)
+
+ # for flow_func in flow_funcs:
+ # validate_flows(G, s, t, 11875108, flow_func(G, s, t, **kwargs),
+ # flow_func)
def test_preflow_push_global_relabel(self):
- G = read_graph('gw1')
+ G = read_graph("gw1")
R = preflow_push(G, 1, len(G), global_relabel_freq=50)
- assert R.graph['flow_value'] == 1202018
+ assert R.graph["flow_value"] == 1202018