Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.9/dist-packages/networkx/algorithms/flow/utils.py: 30%
77 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-10-20 07:00 +0000
1"""
2Utility classes and functions for network flow algorithms.
3"""
5from collections import deque
7import networkx as nx
9__all__ = [
10 "CurrentEdge",
11 "Level",
12 "GlobalRelabelThreshold",
13 "build_residual_network",
14 "detect_unboundedness",
15 "build_flow_dict",
16]
19class CurrentEdge:
20 """Mechanism for iterating over out-edges incident to a node in a circular
21 manner. StopIteration exception is raised when wraparound occurs.
22 """
24 __slots__ = ("_edges", "_it", "_curr")
26 def __init__(self, edges):
27 self._edges = edges
28 if self._edges:
29 self._rewind()
31 def get(self):
32 return self._curr
34 def move_to_next(self):
35 try:
36 self._curr = next(self._it)
37 except StopIteration:
38 self._rewind()
39 raise
41 def _rewind(self):
42 self._it = iter(self._edges.items())
43 self._curr = next(self._it)
46class Level:
47 """Active and inactive nodes in a level."""
49 __slots__ = ("active", "inactive")
51 def __init__(self):
52 self.active = set()
53 self.inactive = set()
56class GlobalRelabelThreshold:
57 """Measurement of work before the global relabeling heuristic should be
58 applied.
59 """
61 def __init__(self, n, m, freq):
62 self._threshold = (n + m) / freq if freq else float("inf")
63 self._work = 0
65 def add_work(self, work):
66 self._work += work
68 def is_reached(self):
69 return self._work >= self._threshold
71 def clear_work(self):
72 self._work = 0
75@nx._dispatch(edge_attrs={"capacity": float("inf")})
76def build_residual_network(G, capacity):
77 """Build a residual network and initialize a zero flow.
79 The residual network :samp:`R` from an input graph :samp:`G` has the
80 same nodes as :samp:`G`. :samp:`R` is a DiGraph that contains a pair
81 of edges :samp:`(u, v)` and :samp:`(v, u)` iff :samp:`(u, v)` is not a
82 self-loop, and at least one of :samp:`(u, v)` and :samp:`(v, u)` exists
83 in :samp:`G`.
85 For each edge :samp:`(u, v)` in :samp:`R`, :samp:`R[u][v]['capacity']`
86 is equal to the capacity of :samp:`(u, v)` in :samp:`G` if it exists
87 in :samp:`G` or zero otherwise. If the capacity is infinite,
88 :samp:`R[u][v]['capacity']` will have a high arbitrary finite value
89 that does not affect the solution of the problem. This value is stored in
90 :samp:`R.graph['inf']`. For each edge :samp:`(u, v)` in :samp:`R`,
91 :samp:`R[u][v]['flow']` represents the flow function of :samp:`(u, v)` and
92 satisfies :samp:`R[u][v]['flow'] == -R[v][u]['flow']`.
94 The flow value, defined as the total flow into :samp:`t`, the sink, is
95 stored in :samp:`R.graph['flow_value']`. If :samp:`cutoff` is not
96 specified, reachability to :samp:`t` using only edges :samp:`(u, v)` such
97 that :samp:`R[u][v]['flow'] < R[u][v]['capacity']` induces a minimum
98 :samp:`s`-:samp:`t` cut.
100 """
101 if G.is_multigraph():
102 raise nx.NetworkXError("MultiGraph and MultiDiGraph not supported (yet).")
104 R = nx.DiGraph()
105 R.add_nodes_from(G)
107 inf = float("inf")
108 # Extract edges with positive capacities. Self loops excluded.
109 edge_list = [
110 (u, v, attr)
111 for u, v, attr in G.edges(data=True)
112 if u != v and attr.get(capacity, inf) > 0
113 ]
114 # Simulate infinity with three times the sum of the finite edge capacities
115 # or any positive value if the sum is zero. This allows the
116 # infinite-capacity edges to be distinguished for unboundedness detection
117 # and directly participate in residual capacity calculation. If the maximum
118 # flow is finite, these edges cannot appear in the minimum cut and thus
119 # guarantee correctness. Since the residual capacity of an
120 # infinite-capacity edge is always at least 2/3 of inf, while that of an
121 # finite-capacity edge is at most 1/3 of inf, if an operation moves more
122 # than 1/3 of inf units of flow to t, there must be an infinite-capacity
123 # s-t path in G.
124 inf = (
125 3
126 * sum(
127 attr[capacity]
128 for u, v, attr in edge_list
129 if capacity in attr and attr[capacity] != inf
130 )
131 or 1
132 )
133 if G.is_directed():
134 for u, v, attr in edge_list:
135 r = min(attr.get(capacity, inf), inf)
136 if not R.has_edge(u, v):
137 # Both (u, v) and (v, u) must be present in the residual
138 # network.
139 R.add_edge(u, v, capacity=r)
140 R.add_edge(v, u, capacity=0)
141 else:
142 # The edge (u, v) was added when (v, u) was visited.
143 R[u][v]["capacity"] = r
144 else:
145 for u, v, attr in edge_list:
146 # Add a pair of edges with equal residual capacities.
147 r = min(attr.get(capacity, inf), inf)
148 R.add_edge(u, v, capacity=r)
149 R.add_edge(v, u, capacity=r)
151 # Record the value simulating infinity.
152 R.graph["inf"] = inf
154 return R
157@nx._dispatch(
158 graphs="R",
159 preserve_edge_attrs={"R": {"capacity": float("inf")}},
160 preserve_graph_attrs=True,
161)
162def detect_unboundedness(R, s, t):
163 """Detect an infinite-capacity s-t path in R."""
164 q = deque([s])
165 seen = {s}
166 inf = R.graph["inf"]
167 while q:
168 u = q.popleft()
169 for v, attr in R[u].items():
170 if attr["capacity"] == inf and v not in seen:
171 if v == t:
172 raise nx.NetworkXUnbounded(
173 "Infinite capacity path, flow unbounded above."
174 )
175 seen.add(v)
176 q.append(v)
179@nx._dispatch(graphs={"G": 0, "R": 1}, preserve_edge_attrs={"R": {"flow": None}})
180def build_flow_dict(G, R):
181 """Build a flow dictionary from a residual network."""
182 flow_dict = {}
183 for u in G:
184 flow_dict[u] = {v: 0 for v in G[u]}
185 flow_dict[u].update(
186 (v, attr["flow"]) for v, attr in R[u].items() if attr["flow"] > 0
187 )
188 return flow_dict