Add support for weighted graphs

.github/workflows/python-package.yml

@@ -46,6 +46,7 @@ jobs:
           python -m pip install mypy
           python -m mypy utilities/ --install-types --non-interactive
       - name: Test with pytest
+        timeout-minutes: 45
         run: |
           source env/bin/activate
-          pytest -n auto -m "not serial" tests/ && pytest -m "serial" tests/
+          pytest -v -n auto -m "not serial" tests/ && pytest -v -m "serial" tests/

tests/shared_testing_functions.py

@@ -5,26 +5,31 @@
 from modularitypruning.partition_utilities import num_communities
 
 
-def generate_connected_ER(n, m, directed):
+def generate_random_values(num_values, start_value, end_value):
+    return [uniform(start_value, end_value) for _ in range(num_values)]
+
+
+def generate_connected_ER(n, m, directed, weighted=False, low=0.1, high=10.0):
     G = ig.Graph.Erdos_Renyi(n=n, m=m, directed=directed)
     while not G.is_connected():
         G = ig.Graph.Erdos_Renyi(n=n, m=m, directed=directed)
+    if weighted:
+        G.es['weight'] = generate_random_values(G.ecount(), low, high)
     return G
 
 
-def generate_connected_multilayer_ER(num_nodes_per_layer, m, num_layers, directed):
+def generate_connected_multilayer_ER(num_nodes_per_layer, m, num_layers, directed, weighted=False, low=0.1, high=10.0):
     total_num_nodes = num_nodes_per_layer * num_layers
     G = generate_connected_ER(n=total_num_nodes, m=m, directed=directed)
     layer_membership = [i // num_nodes_per_layer for i in range(total_num_nodes)]
     intralayer_edges = [(e.source, e.target) for e in G.es if layer_membership[e.source] == layer_membership[e.target]]
     interlayer_edges = [(e.source, e.target) for e in G.es if layer_membership[e.source] != layer_membership[e.target]]
-    return (ig.Graph(n=total_num_nodes, edges=intralayer_edges, directed=directed),
-            ig.Graph(n=total_num_nodes, edges=interlayer_edges, directed=directed),
-            layer_membership)
-
-
-def generate_random_values(num_values, start_value, end_value):
-    return [uniform(start_value, end_value) for _ in range(num_values)]
+    G_intralayer = ig.Graph(n=total_num_nodes, edges=intralayer_edges, directed=directed)
+    G_interlayer = ig.Graph(n=total_num_nodes, edges=interlayer_edges, directed=directed)
+    if weighted:
+        G_intralayer.es['weight'] = generate_random_values(G_intralayer.ecount(), low, high)
+        G_interlayer.es['weight'] = generate_random_values(G_interlayer.ecount(), low, high)
+    return G_intralayer, G_interlayer, layer_membership
 
 
 def generate_random_partition(num_nodes, K):
@@ -44,7 +49,7 @@ def generate_random_partitions(num_nodes, num_partitions, K_max):
 
 
 def generate_igraph_famous():
-    """Generate various famous graphs from igraph.
+    """Generate various famous graphs from igraph as (name, graph) pairs.
 
     In particular, we use
         Meredith (n=70, m=140): a counterexample to a conjecture regarding 4-regular 4-connected Hamiltonian graphs
@@ -54,10 +59,11 @@ def generate_igraph_famous():
         Zachary (n=34, m=78): popular network of the social interactions between 34 members of a karate club
     """
 
-    return [ig.Graph.Famous(name) for name in ['meredith', 'nonline', 'thomassen', 'tutte', 'zachary']]
+    return [(name, ig.Graph.Famous(name)) for name in ['meredith', 'nonline', 'thomassen', 'tutte', 'zachary']]
 
 
-def generate_multilayer_intralayer_SBM(copying_probability, p_in, p_out, first_layer_membership, num_layers):
+def generate_connected_multilayer_intralayer_SBM(copying_probability, p_in, p_out, first_layer_membership, num_layers,
+                                                 weighted=False, low=0.5, high=5.0):
     num_nodes_per_layer = len(first_layer_membership)
     community_labels_per_layer = [[0] * num_nodes_per_layer for _ in range(num_layers)]
     community_labels_per_layer[0] = list(first_layer_membership)
@@ -86,8 +92,22 @@ def generate_multilayer_intralayer_SBM(copying_probability, p_in, p_out, first_l
                     if random() < p_out:
                         intralayer_edges.append((u, v))
 
+    # force disconnected intralayer nodes to have at least one edge
+    intralayer_degrees = [0] * len(combined_community_labels)
+    for u, v in intralayer_edges:
+        intralayer_degrees[u] += 1
+        intralayer_degrees[v] += 1
+    for v, degree in enumerate(intralayer_degrees):
+        if degree == 0:
+            same_layer = [u for u in range(len(combined_community_labels))
+                          if layer_membership[u] == layer_membership[v] and u != v]
+            intralayer_edges.append((v, same_layer[randint(0, len(same_layer) - 1)]))
+
     G_intralayer = ig.Graph(n=len(combined_community_labels), edges=intralayer_edges, directed=False)
 
+    if weighted:
+        G_intralayer.es['weight'] = generate_random_values(G_intralayer.ecount(), low, high)
+
     return G_intralayer, layer_membership
 
 

tests/test_champ_coefficients_2D.py

@@ -4,8 +4,8 @@
 from modularitypruning.champ_utilities import partition_coefficients_2D
 from modularitypruning.leiden_utilities import leiden_part_with_membership, repeated_leiden_from_gammas
 
-from .shared_testing_functions import generate_connected_ER, generate_random_values, generate_random_partitions, \
-    generate_igraph_famous
+from .shared_testing_functions import generate_connected_ER, \
+    generate_random_values, generate_random_partitions, generate_igraph_famous
 
 
 class TestCHAMPCoefficients2D(unittest.TestCase):
@@ -24,64 +24,87 @@ def assert_partition_coefficient_correctness(self, G, partitions, coefficients):
             #   = P_hat
             leiden_P_hat = leiden_A_hat - leiden_part.quality(resolution_parameter=1)
 
-            self.assertAlmostEqual(A_hat, leiden_A_hat, places=10)
-            self.assertAlmostEqual(P_hat, leiden_P_hat, places=10)
+            self.assertAlmostEqual(A_hat, leiden_A_hat, places=8)
+            self.assertAlmostEqual(P_hat, leiden_P_hat, places=8)
 
-    def assert_partition_coefficient_correctness_unweighted_ER(self, n=100, m=500, directed=False,
-                                                               num_partitions=10, K_max=5):
-        G = generate_connected_ER(n=n, m=m, directed=directed)
+    def assert_partition_coefficient_correctness_ER(self, n=100, m=500, directed=False,
+                                                    num_partitions=10, K_max=5, weighted=False):
+        G = generate_connected_ER(n=n, m=m, directed=directed, weighted=weighted)
         partitions = generate_random_partitions(num_nodes=n, num_partitions=num_partitions, K_max=K_max)
         coefficients = partition_coefficients_2D(G, partitions)
         self.assert_partition_coefficient_correctness(G, partitions, coefficients)
 
-    def test_partition_coefficient_correctness_undirected_unweighted_varying_n(self):
+    def test_partition_coefficient_correctness_undirected_varying_n(self):
         for n in [50, 100, 250, 500]:
-            self.assert_partition_coefficient_correctness_unweighted_ER(n=n, m=5 * n)
+            for weighted in [False, True]:
+                with self.subTest(n=n, weighted=weighted):
+                    self.assert_partition_coefficient_correctness_ER(n=n, m=5 * n, weighted=weighted)
 
-    def test_partition_coefficient_correctness_undirected_unweighted_varying_m(self):
+    def test_partition_coefficient_correctness_undirected_varying_m(self):
         for m in [200, 500, 1000]:
-            self.assert_partition_coefficient_correctness_unweighted_ER(m=m)
+            for weighted in [False, True]:
+                with self.subTest(m=m, weighted=weighted):
+                    self.assert_partition_coefficient_correctness_ER(m=m, weighted=weighted)
 
-    def test_partition_coefficient_correctness_undirected_unweighted_varying_num_partitions(self):
+    def test_partition_coefficient_correctness_undirected_varying_num_partitions(self):
         for num_partitions in [10, 100, 1000]:
-            self.assert_partition_coefficient_correctness_unweighted_ER(num_partitions=num_partitions)
+            for weighted in [False, True]:
+                with self.subTest(num_partitions=num_partitions, weighted=weighted):
+                    self.assert_partition_coefficient_correctness_ER(num_partitions=num_partitions,
+                                                                     weighted=weighted)
 
-    def test_partition_coefficient_correctness_undirected_unweighted_varying_K_max(self):
+    def test_partition_coefficient_correctness_undirected_varying_K_max(self):
         for K_max in [2, 5, 10, 20]:
-            self.assert_partition_coefficient_correctness_unweighted_ER(num_partitions=100, K_max=K_max)
+            for weighted in [False, True]:
+                with self.subTest(K_max=K_max, weighted=weighted):
+                    self.assert_partition_coefficient_correctness_ER(num_partitions=100, K_max=K_max,
+                                                                     weighted=weighted)
 
-    def test_partition_coefficient_correctness_directed_unweighted_varying_n(self):
+    def test_partition_coefficient_correctness_directed_varying_n(self):
         for n in [50, 100, 250, 500]:
-            self.assert_partition_coefficient_correctness_unweighted_ER(n=n, m=10 * n, directed=True)
+            for weighted in [False, True]:
+                with self.subTest(n=n, weighted=weighted):
+                    self.assert_partition_coefficient_correctness_ER(n=n, m=10 * n, directed=True,
+                                                                     weighted=weighted)
 
-    def test_partition_coefficient_correctness_directed_unweighted_varying_m(self):
+    def test_partition_coefficient_correctness_directed_varying_m(self):
         for m in [400, 1000, 2000]:
-            self.assert_partition_coefficient_correctness_unweighted_ER(m=m, directed=True)
+            for weighted in [False, True]:
+                with self.subTest(m=m, weighted=weighted):
+                    self.assert_partition_coefficient_correctness_ER(m=m, directed=True, weighted=weighted)
 
-    def test_partition_coefficient_correctness_directed_unweighted_varying_num_partitions(self):
+    def test_partition_coefficient_correctness_directed_varying_num_partitions(self):
         for num_partitions in [10, 100, 1000]:
-            self.assert_partition_coefficient_correctness_unweighted_ER(directed=True, num_partitions=num_partitions)
+            for weighted in [False, True]:
+                with self.subTest(num_partitions=num_partitions, weighted=weighted):
+                    self.assert_partition_coefficient_correctness_ER(directed=True, num_partitions=num_partitions,
+                                                                     weighted=weighted)
 
-    def test_partition_coefficient_correctness_directed_unweighted_varying_K_max(self):
+    def test_partition_coefficient_correctness_directed_varying_K_max(self):
         for K_max in [2, 5, 10, 20]:
-            self.assert_partition_coefficient_correctness_unweighted_ER(directed=True, num_partitions=100, K_max=K_max)
+            for weighted in [False, True]:
+                with self.subTest(K_max=K_max, weighted=weighted):
+                    self.assert_partition_coefficient_correctness_ER(directed=True, num_partitions=100, K_max=K_max,
+                                                                     weighted=weighted)
 
     def test_partition_coefficient_correctness_igraph_famous_leiden(self):
         """Test partition coefficient correctness on various famous graphs while obtaining partitions via Leiden.
 
         The correctness is checked for the original undirected and (symmetric) directed variants.
         """
 
-        for G in generate_igraph_famous():
+        for name, G in generate_igraph_famous():
             gammas = generate_random_values(100, start_value=0, end_value=5)
-            partitions = repeated_leiden_from_gammas(G, gammas)
-            coefficients = partition_coefficients_2D(G, partitions)
-            self.assert_partition_coefficient_correctness(G, partitions, coefficients)
+            with self.subTest(graph=name, directed=False):
+                partitions = repeated_leiden_from_gammas(G, gammas)
+                coefficients = partition_coefficients_2D(G, partitions)
+                self.assert_partition_coefficient_correctness(G, partitions, coefficients)
 
             G.to_directed()  # check the directed version of the graph as well
-            partitions = repeated_leiden_from_gammas(G, gammas)
-            coefficients = partition_coefficients_2D(G, partitions)
-            self.assert_partition_coefficient_correctness(G, partitions, coefficients)
+            with self.subTest(graph=name, directed=True):
+                partitions = repeated_leiden_from_gammas(G, gammas)
+                coefficients = partition_coefficients_2D(G, partitions)
+                self.assert_partition_coefficient_correctness(G, partitions, coefficients)
 
 
 if __name__ == "__main__":