rebase + clean

Author: Andrea Tomassilli

networkx/algorithms/approximation/kcutsets.py

@@ -1,6 +1,7 @@
 """Functions involving natural generalizations of the minimum cut problem."""
 
 import itertools
+
 import networkx as nx
 from networkx.utils import not_implemented_for
 
@@ -9,8 +10,9 @@
 
 @not_implemented_for("directed")
 @not_implemented_for("multigraph")
+@nx._dispatchable(edge_attrs="weight")
 def minimum_multiway_cut(G, terminals, weight=None):
-    """Compute an approximated Minimum Multiway Cut and the corresponding cut value.
+    r"""Compute an approximated Minimum Multiway Cut and the corresponding cut value.
 
     Given an undirected graph $G = (V, E)$ and a set of terminals
     $S = \{s_1, s_2, \dots ,s_k\} \subseteq V$, a multiway cut is a set of edges
@@ -40,8 +42,8 @@ def minimum_multiway_cut(G, terminals, weight=None):
         A container with a subset of nodes of G.
 
     weight : string, optional (default = None)
-        If None, every node has weight 1. If a string, use this node
-        attribute as the node weight. A node without this attribute is
+        If None, every edge has weight 1. If a string, use this edge
+        attribute as the edge weight. An edge without this attribute is
         assumed to have weight 1.
 
     Returns
@@ -51,7 +53,27 @@ def minimum_multiway_cut(G, terminals, weight=None):
 
     cutset : set
         Set of edges that, if removed from the graph, disconnects each
-         terminal from all the others.
+        terminal from all the others.
+
+    Examples
+    --------
+    >>> G = nx.path_graph(5)
+    >>> terminals = [0, 4]
+    >>> cut_value, cutset = nx.approximation.minimum_multiway_cut(G, terminals)
+    >>> cut_value
+    1
+    >>> len(cutset)
+    1
+
+    For a weighted graph:
+
+    >>> G = nx.Graph()
+    >>> G.add_weighted_edges_from([(0, 1, 10), (1, 2, 5), (2, 3, 10)])
+    >>> cut_value, cutset = nx.approximation.minimum_multiway_cut(
+    ...     G, [0, 3], weight="weight"
+    ... )
+    >>> cut_value
+    5
 
     Raises
     ------
@@ -73,56 +95,45 @@ def minimum_multiway_cut(G, terminals, weight=None):
         raise nx.NetworkXError("Expected non-empty NetworkX graph!")
     if not nx.is_connected(G):
         raise nx.NetworkXError("Graph not connected.")
-    # only consider the terminals in G
     terminals = set(terminals) & G.nodes()
-    # raise an error if less than two terminal have been provided
     if len(terminals) < 2:
         raise nx.NetworkXError("At least two terminals should be provided.")
 
-    # extract edges weight, and set edges weights with no attribute to 1
-    edges_weights = G.edges(data=weight, default=1)
-    # create a new Graph G2
+    # Build working graph with uniform weight attribute name
     G2 = nx.Graph()
-    G2.add_weighted_edges_from(edges_weights, weight="capacity")
+    G2.add_weighted_edges_from(G.edges(data=weight, default=1), weight="capacity")
 
-    # take a non-existing node of G to be the sink
+    # Create auxiliary sink connected to all terminals with infinite capacity
     sink = next(u for u in range(len(G) + 1) if u not in G)
-    # add edges from the terminals to the sink node with infinite capacity
     G2.add_edges_from([(u, sink) for u in terminals], capacity=float("inf"))
 
+    # Compute minimum isolating cut for each terminal
     all_cuts = []
-    # compute the minimum weight isolating cut for each terminal
     for u in terminals:
-        # remove the edge from u to the sink
         G2.remove_edge(u, sink)
-        # get the cut value and the 2 partitions of nodes
         value_cut, (p1, p2) = nx.minimum_cut(G2, u, sink)
-        # get the edges crossing the cut
         edges_cut = set(nx.edge_boundary(G2, p1, p2))
-        # add to the result
         all_cuts.append((edges_cut, value_cut))
-        # re-add the edge to the sink
         G2.add_edge(u, sink, capacity=float("inf"))
 
-    # discard the heaviest cut and take the union of the rest
+    # Discard heaviest cut and union the rest (avoiding duplicate edges)
     cutset = set()
     cut_value = 0
     for u, v in itertools.chain.from_iterable(
         el[0] for el in sorted(all_cuts, key=lambda x: x[1])[:-1]
     ):
         if (u, v) not in cutset and (v, u) not in cutset:
-            # add to the cutset
             cutset.add((u, v))
-            # add the weight to the cut cost
             cut_value += G2.edges[u, v]["capacity"]
 
     return cut_value, cutset
 
 
 @not_implemented_for("directed")
 @not_implemented_for("multigraph")
+@nx._dispatchable(edge_attrs="weight")
 def minimum_k_cut(G, k, weight=None):
-    """Compute an approximated Minimum k-Cut and the corresponding cut value.
+    r"""Compute an approximated Minimum k-Cut and the corresponding cut value.
 
     Given an undirected graph $G = (V, E)$ and an integer $k \geq 2$, a $k$-cut is a
     set of edges whose removal leaves at least $k$ connected components.
@@ -180,6 +191,24 @@ def minimum_k_cut(G, k, weight=None):
     If more than $k$ components are created then it's enough to throw back
     some of the removed edges until there are exactly $k$ components.
 
+    Examples
+    --------
+    >>> G = nx.path_graph(5)
+    >>> cut_value, cutset = nx.approximation.minimum_k_cut(G, k=3)
+    >>> cut_value
+    2
+    >>> G.remove_edges_from(cutset)
+    >>> len(list(nx.connected_components(G))) >= 3
+    True
+
+    For a weighted graph:
+
+    >>> G = nx.complete_graph(4)
+    >>> nx.set_edge_attributes(G, values=10, name="weight")
+    >>> cut_value, cutset = nx.approximation.minimum_k_cut(G, k=4, weight="weight")
+    >>> cut_value
+    60
+
     See also
     --------
     networkx.algorithms.flow.minimum_cut
@@ -197,37 +226,28 @@ def minimum_k_cut(G, k, weight=None):
     if not 1 <= k <= len(G):
         raise nx.NetworkXError(f"k should be within 1 and {len(G)}")
 
-    # extract edges weights, and set edges weights with no attribute to 1
-    edges_weights = G.edges(data=weight, default=1)
-    # create a new Graph G2
+    # Build working graph with uniform weight attribute name
     G2 = nx.Graph()
-    G2.add_weighted_edges_from(edges_weights, weight="capacity")
+    G2.add_weighted_edges_from(G.edges(data=weight, default=1), weight="capacity")
 
-    # build a Gomory-Hu tree T from G
+    # Build Gomory-Hu tree and get k-1 lightest edges
     T = nx.gomory_hu_tree(G2)
-    # get the k-1 cheapest edges of the Gomory-Hu tree
     min_weight_edges = sorted(T.edges(data="weight"), key=lambda x: x[2])[: k - 1]
 
-    # compute the cutset, the value is computed after as an edge might appear more than once
+    # Collect boundary edges for each cut induced by removing tree edges
     all_edges_cut = set()
     for u, v, _ in min_weight_edges:
-        # remove (u,v) from the tree
         T.remove_edge(u, v)
-        # get the connected component that contains u
         p1 = nx.node_connected_component(T, u)
-        # add the boundary edges of p1 to the cutset
         all_edges_cut |= set(nx.edge_boundary(G2, p1))
-        # re-add (u,v) to the tree
         T.add_edge(u, v)
 
-    # consider edges only in a direction
+    # Compute final cutset avoiding duplicate edges
     cutset = set()
     cut_value = 0
     for u, v in all_edges_cut:
         if (u, v) not in cutset and (v, u) not in cutset:
-            # add to the cutset
             cutset.add((u, v))
-            # add the weight to the cut cost
             cut_value += G2.edges[u, v]["capacity"]
 
     return cut_value, cutset

networkx/algorithms/approximation/tests/test_kcutsets.py

@@ -1,10 +1,11 @@
 """Unit tests for the :mod:`networkx.algorithms.approximation.kcutsets` module."""
 
 import itertools
+
 import pytest
+
 import networkx as nx
-from networkx.algorithms.approximation import minimum_multiway_cut, minimum_k_cut
-from networkx import minimum_cut_value
+from networkx.algorithms import approximation as approx
 
 
 class TestMinMultiwayCut:
@@ -18,14 +19,14 @@ def test_null_graph(self):
         with pytest.raises(
             nx.NetworkXError, match="Expected non-empty NetworkX graph!"
         ):
-            minimum_multiway_cut(G, G.nodes())
+            approx.minimum_multiway_cut(G, G.nodes())
 
     def test_undirected_non_connected(self):
         """Test an undirected disconnected graph."""
         G = nx.path_graph(10)
         G.remove_edge(3, 4)
         with pytest.raises(nx.NetworkXError, match="Graph not connected."):
-            minimum_multiway_cut(G, G.nodes())
+            approx.minimum_multiway_cut(G, G.nodes())
 
     def test_invalid_terminals(self):
         """Test empty terminals."""
@@ -34,12 +35,12 @@ def test_invalid_terminals(self):
         with pytest.raises(
             nx.NetworkXError, match="At least two terminals should be provided."
         ):
-            minimum_multiway_cut(G, [])
+            approx.minimum_multiway_cut(G, [])
 
     def test_path_graph_unweighted(self):
         """Test min multiway cut for a path graph."""
         G = nx.path_graph(2)
-        cut_value, cutset = minimum_multiway_cut(G, [0, 1])
+        cut_value, cutset = approx.minimum_multiway_cut(G, [0, 1])
         assert cut_value == 1
         G.remove_edges_from(cutset)
         assert len(list(nx.connected_components(G))) == 2
@@ -50,18 +51,50 @@ def test_path_graph_weighted(self):
         G.add_weighted_edges_from(
             [(0, 1, 10), (1, 2, 10), (2, 3, 5)], weight="capacity"
         )
-        cut_value, cutset = minimum_multiway_cut(G, [0, 3], weight="capacity")
+        cut_value, cutset = approx.minimum_multiway_cut(G, [0, 3], weight="capacity")
         assert cut_value == 5
 
     def test_complete_graph(self):
         """Test min multiway cut for a complete graph."""
         G = nx.complete_graph(5)
-        cut_value, cutset = minimum_multiway_cut(G, G.nodes())
+        cut_value, cutset = approx.minimum_multiway_cut(G, G.nodes())
         assert cut_value == 10
         # remove the edges
         G.remove_edges_from(cutset)
         assert set(G.edges()) == set()
 
+    def test_single_terminal(self):
+        """Test that a single terminal raises an error."""
+        G = nx.path_graph(5)
+        with pytest.raises(
+            nx.NetworkXError, match="At least two terminals should be provided."
+        ):
+            approx.minimum_multiway_cut(G, [0])
+
+    def test_terminals_not_in_graph(self):
+        """Test terminals that are not in the graph are ignored."""
+        G = nx.path_graph(5)
+        # terminals 100 and 200 don't exist, only 0 and 4 are valid
+        cut_value, cutset = approx.minimum_multiway_cut(G, [0, 4, 100, 200])
+        assert cut_value == 1
+        G.remove_edges_from(cutset)
+        components = list(nx.connected_components(G))
+        assert len(components) == 2
+
+    def test_directed_graph_raises(self):
+        """Test that directed graphs raise NetworkXNotImplemented."""
+        G = nx.DiGraph()
+        G.add_edges_from([(0, 1), (1, 2), (2, 3)])
+        with pytest.raises(nx.NetworkXNotImplemented):
+            approx.minimum_multiway_cut(G, [0, 3])
+
+    def test_multigraph_raises(self):
+        """Test that multigraphs raise NetworkXNotImplemented."""
+        G = nx.MultiGraph()
+        G.add_edges_from([(0, 1), (1, 2), (2, 3)])
+        with pytest.raises(nx.NetworkXNotImplemented):
+            approx.minimum_multiway_cut(G, [0, 3])
+
     @pytest.mark.parametrize(
         "graph_class",
         [
@@ -82,8 +115,8 @@ def test_compare_min_cut(self, graph_class, s, t):
         """
         G = graph_class()
         nx.set_edge_attributes(G, values=10, name="weight")
-        cut_value, cutset = minimum_multiway_cut(G, {s, t}, weight="weight")
-        assert cut_value == minimum_cut_value(G, s, t, capacity="weight")
+        cut_value, cutset = approx.minimum_multiway_cut(G, {s, t}, weight="weight")
+        assert cut_value == nx.minimum_cut_value(G, s, t, capacity="weight")
 
 
 class TestMinkCut:
@@ -97,28 +130,49 @@ def test_null_graph(self):
         with pytest.raises(
             nx.NetworkXError, match="Expected non-empty NetworkX graph!"
         ):
-            minimum_k_cut(G, 3)
+            approx.minimum_k_cut(G, 3)
 
     def test_undirected_non_connected(self):
         """Test an undirected disconnected graph."""
         G = nx.path_graph(10)
         G.remove_edge(3, 4)
         with pytest.raises(nx.NetworkXError, match="Graph not connected."):
-            minimum_k_cut(G, 3)
+            approx.minimum_k_cut(G, 3)
 
     def test_invalid_k(self):
-        """Test empty terminals."""
+        """Test invalid k values."""
         G = nx.path_graph(10)
         with pytest.raises(nx.NetworkXError, match="k should be within 1 and 10"):
-            minimum_k_cut(G, 0)
+            approx.minimum_k_cut(G, 0)
         with pytest.raises(nx.NetworkXError, match="k should be within 1 and 10"):
-            minimum_k_cut(G, 11)
+            approx.minimum_k_cut(G, 11)
+
+    def test_directed_graph_raises(self):
+        """Test that directed graphs raise NetworkXNotImplemented."""
+        G = nx.DiGraph()
+        G.add_edges_from([(0, 1), (1, 2), (2, 3)])
+        with pytest.raises(nx.NetworkXNotImplemented):
+            approx.minimum_k_cut(G, 2)
+
+    def test_multigraph_raises(self):
+        """Test that multigraphs raise NetworkXNotImplemented."""
+        G = nx.MultiGraph()
+        G.add_edges_from([(0, 1), (1, 2), (2, 3)])
+        with pytest.raises(nx.NetworkXNotImplemented):
+            approx.minimum_k_cut(G, 2)
+
+    def test_k_equals_one(self):
+        """Test k=1 returns empty cutset with zero cut value."""
+        G = nx.complete_graph(5)
+        cut_value, cutset = approx.minimum_k_cut(G, 1)
+        assert cut_value == 0
+        assert len(cutset) == 0
 
     @pytest.mark.parametrize("k,expected", [(1, 0), (2, 1), (3, 2), (4, 3), (5, 4)])
     def test_path_graph(self, k, expected):
         """Test various k for a path graph of 5 nodes."""
         G = nx.path_graph(n=5)
-        cut_value, cutset = minimum_k_cut(G, k, weight="capacity")
+        cut_value, cutset = approx.minimum_k_cut(G, k, weight="capacity")
         assert cut_value == expected
         G.remove_edges_from(cutset)
         assert len(list(nx.connected_components(G))) == k
@@ -127,7 +181,7 @@ def test_path_graph(self, k, expected):
     def test_complete_graph(self, k, expected):
         """Test various k for a complete graph of 5 nodes."""
         G = nx.complete_graph(5)
-        cut_value, cutset = minimum_k_cut(G, k)
+        cut_value, cutset = approx.minimum_k_cut(G, k)
         assert cut_value == expected
         G.remove_edges_from(cutset)
         assert len(list(nx.connected_components(G))) >= k
@@ -147,15 +201,15 @@ def test_complete_graph(self, k, expected):
     def test_connected_components(self, graph_class, k):
         """Test multiple graph types and k."""
         G = graph_class()
-        cut_value, cutset = minimum_k_cut(G, k)
+        cut_value, cutset = approx.minimum_k_cut(G, k)
         G.remove_edges_from(cutset)
         assert len(list(nx.connected_components(G))) >= k
 
     def test_complete_graph_weighted(self):
         """Test min k-cut for a weighted complete graph."""
         G = nx.complete_graph(5)
         nx.set_edge_attributes(G, values=10, name="weight")
-        cut_value, cutset = minimum_k_cut(G, 5, weight="weight")
+        cut_value, cutset = approx.minimum_k_cut(G, 5, weight="weight")
         assert cut_value == 100
         # remove the edges
         G.remove_edges_from(cutset)
@@ -167,7 +221,7 @@ def test_path_graph_weighted(self):
         G.add_weighted_edges_from(
             [(0, 1, 10), (1, 2, 10), (2, 3, 5)], weight="capacity"
         )
-        cut_value, cutset = minimum_k_cut(G, 3, weight="capacity")
+        cut_value, cutset = approx.minimum_k_cut(G, 3, weight="capacity")
         assert cut_value == 15
         G.remove_edges_from(cutset)
         assert len(list(nx.connected_components(G))) == 3