diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index c38cc5e..17d144f 100644 --- a/.github/workflows/python-package.yml +++ b/.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/ diff --git a/tests/shared_testing_functions.py b/tests/shared_testing_functions.py index 81538d5..766a62b 100644 --- a/tests/shared_testing_functions.py +++ b/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 diff --git a/tests/test_champ_coefficients_2D.py b/tests/test_champ_coefficients_2D.py index 9c630ef..e45a176 100644 --- a/tests/test_champ_coefficients_2D.py +++ b/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,47 +24,68 @@ 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. @@ -72,16 +93,18 @@ def test_partition_coefficient_correctness_igraph_famous_leiden(self): 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__": diff --git a/tests/test_champ_coefficients_3D.py b/tests/test_champ_coefficients_3D.py index 723b703..4c33b1e 100644 --- a/tests/test_champ_coefficients_3D.py +++ b/tests/test_champ_coefficients_3D.py @@ -32,9 +32,9 @@ def assert_partition_coefficient_correctness(self, G_intralayer, G_interlayer, l # = C_hat leiden_C_hat = interlayer_part.quality(resolution_parameter=0) - self.assertAlmostEqual(A_hat, leiden_A_hat, places=10) - self.assertAlmostEqual(P_hat, leiden_P_hat, places=10) - self.assertAlmostEqual(C_hat, leiden_C_hat, places=10) + self.assertAlmostEqual(A_hat, leiden_A_hat, places=8) + self.assertAlmostEqual(P_hat, leiden_P_hat, places=8) + self.assertAlmostEqual(C_hat, leiden_C_hat, places=8) # Also test against an alternate, per-layer calculation of P_hat alternate_P_hat = 0 @@ -46,79 +46,108 @@ def assert_partition_coefficient_correctness(self, G_intralayer, G_interlayer, l layer_part = leiden_part_with_membership(layer_subgraph, layer_community_membership) alternate_P_hat += (layer_part.quality(resolution_parameter=0.0) - layer_part.quality(resolution_parameter=1.0)) - self.assertAlmostEqual(alternate_P_hat, P_hat, places=10) + self.assertAlmostEqual(alternate_P_hat, P_hat, places=8) - def assert_partition_coefficient_correctness_unweighted_ER(self, num_nodes_per_layer=100, m=25000, num_layers=10, - directed=False, num_partitions=10, K_max=10): + def assert_partition_coefficient_correctness_ER(self, num_nodes_per_layer=100, m=25000, num_layers=10, + directed=False, num_partitions=10, K_max=10, weighted=False): G_intralayer, G_interlayer, layer_membership = generate_connected_multilayer_ER( - num_nodes_per_layer=num_nodes_per_layer, m=m, num_layers=num_layers, directed=directed) + num_nodes_per_layer=num_nodes_per_layer, m=m, num_layers=num_layers, directed=directed, + weighted=weighted) partitions = generate_random_partitions(num_nodes=G_intralayer.vcount(), num_partitions=num_partitions, K_max=K_max) coefficients = partition_coefficients_3D(G_intralayer, G_interlayer, layer_membership, partitions) self.assert_partition_coefficient_correctness(G_intralayer, G_interlayer, layer_membership, partitions, coefficients) - def test_partition_coefficient_correctness_undirected_unweighted_varying_num_nodes_per_layer(self): + def test_partition_coefficient_correctness_undirected_varying_num_nodes_per_layer(self): for num_nodes_per_layer in [50, 100, 250, 500]: - self.assert_partition_coefficient_correctness_unweighted_ER(num_nodes_per_layer=num_nodes_per_layer, - m=50 * num_nodes_per_layer) + for weighted in [False, True]: + with self.subTest(num_nodes_per_layer=num_nodes_per_layer, weighted=weighted): + self.assert_partition_coefficient_correctness_ER(num_nodes_per_layer=num_nodes_per_layer, + m=50 * num_nodes_per_layer, + weighted=weighted) - def test_partition_coefficient_correctness_undirected_unweighted_varying_m(self): + def test_partition_coefficient_correctness_undirected_varying_m(self): for m in [5000, 10000, 15000, 20000]: - 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_layers(self): + def test_partition_coefficient_correctness_undirected_varying_num_layers(self): for num_layers in [5, 10, 20, 30]: - self.assert_partition_coefficient_correctness_unweighted_ER(num_layers=num_layers) + for weighted in [False, True]: + with self.subTest(num_layers=num_layers, weighted=weighted): + self.assert_partition_coefficient_correctness_ER(num_layers=num_layers, 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 [5, 10, 100, 250]: - 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, 25]: - self.assert_partition_coefficient_correctness_unweighted_ER(K_max=K_max) + for weighted in [False, True]: + with self.subTest(K_max=K_max, weighted=weighted): + self.assert_partition_coefficient_correctness_ER(K_max=K_max, weighted=weighted) - def test_partition_coefficient_correctness_directed_unweighted_varying_num_nodes_per_layer(self): + def test_partition_coefficient_correctness_directed_varying_num_nodes_per_layer(self): for num_nodes_per_layer in [50, 100, 250, 500]: - self.assert_partition_coefficient_correctness_unweighted_ER(num_nodes_per_layer=num_nodes_per_layer, - m=100 * num_nodes_per_layer, directed=True) + for weighted in [False, True]: + with self.subTest(num_nodes_per_layer=num_nodes_per_layer, weighted=weighted): + self.assert_partition_coefficient_correctness_ER(num_nodes_per_layer=num_nodes_per_layer, + m=100 * num_nodes_per_layer, 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 [10000, 20000, 30000, 40000]: - 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_layers(self): + def test_partition_coefficient_correctness_directed_varying_num_layers(self): for num_layers in [5, 10, 20, 30]: - self.assert_partition_coefficient_correctness_unweighted_ER(num_layers=num_layers, directed=True) + for weighted in [False, True]: + with self.subTest(num_layers=num_layers, weighted=weighted): + self.assert_partition_coefficient_correctness_ER(num_layers=num_layers, 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 [5, 10, 100, 250]: - 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, 25]: - self.assert_partition_coefficient_correctness_unweighted_ER(directed=True, 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, K_max=K_max, weighted=weighted) def test_partition_coefficient_correctness_interleaved_directedness(self): """Test partition coefficient correctness when directedness of interlayer and intralayer edges do not match.""" - # Intralayer directed edges, but interlayer undirected ones - G_intralayer, G_interlayer, layer_membership = generate_connected_multilayer_ER(num_nodes_per_layer=100, m=5000, - num_layers=10, directed=False) - G_intralayer.to_directed() - partitions = generate_random_partitions(num_nodes=G_intralayer.vcount(), num_partitions=10, K_max=10) - coefficients = partition_coefficients_3D(G_intralayer, G_interlayer, layer_membership, partitions) - self.assert_partition_coefficient_correctness(G_intralayer, G_interlayer, layer_membership, partitions, - coefficients) - - # Interlayer directed edges, but intralayer undirected ones - G_intralayer, G_interlayer, layer_membership = generate_connected_multilayer_ER(num_nodes_per_layer=100, m=5000, - num_layers=10, directed=False) - G_interlayer.to_directed() - partitions = generate_random_partitions(num_nodes=G_intralayer.vcount(), num_partitions=10, K_max=10) - coefficients = partition_coefficients_3D(G_intralayer, G_interlayer, layer_membership, partitions) - self.assert_partition_coefficient_correctness(G_intralayer, G_interlayer, layer_membership, partitions, - coefficients) + for weighted in [False, True]: + # Intralayer directed edges, but interlayer undirected ones + with self.subTest(intralayer_directed=True, weighted=weighted): + G_intralayer, G_interlayer, layer_membership = generate_connected_multilayer_ER( + num_nodes_per_layer=100, m=5000, num_layers=10, directed=False, weighted=weighted) + G_intralayer.to_directed() + partitions = generate_random_partitions(num_nodes=G_intralayer.vcount(), num_partitions=10, K_max=10) + coefficients = partition_coefficients_3D(G_intralayer, G_interlayer, layer_membership, partitions) + self.assert_partition_coefficient_correctness(G_intralayer, G_interlayer, layer_membership, partitions, + coefficients) + + # Interlayer directed edges, but intralayer undirected ones + with self.subTest(intralayer_directed=False, weighted=weighted): + G_intralayer, G_interlayer, layer_membership = generate_connected_multilayer_ER( + num_nodes_per_layer=100, m=5000, num_layers=10, directed=False, weighted=weighted) + G_interlayer.to_directed() + partitions = generate_random_partitions(num_nodes=G_intralayer.vcount(), num_partitions=10, K_max=10) + coefficients = partition_coefficients_3D(G_intralayer, G_interlayer, layer_membership, partitions) + self.assert_partition_coefficient_correctness(G_intralayer, G_interlayer, layer_membership, partitions, + coefficients) if __name__ == "__main__": diff --git a/tests/test_champ_usage_2D.py b/tests/test_champ_usage_2D.py index 62fffc5..d4ba8f2 100644 --- a/tests/test_champ_usage_2D.py +++ b/tests/test_champ_usage_2D.py @@ -4,8 +4,8 @@ from modularitypruning.champ_utilities import CHAMP_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 TestCHAMP2D(unittest.TestCase): @@ -23,59 +23,82 @@ def assert_best_partitions_match_champ_set(self, G, partitions, champ_ranges, ga # note that this is float comparision to within ~10^{-10} self.assertAlmostEqual(best_partition_quality, champ_quality, places=10, msg=f"CHAMP domain quality does not match best input partition " - f"at gamma {gamma:.2f}") + f"at gamma {gamma:.2f}") break else: self.assertFalse(True, msg=f"gamma {gamma:.2f} was not found within any CHAMP domain") - def assert_champ_correctness_unweighted_ER(self, n=100, m=1000, directed=False, num_partitions=10, num_gammas=100, - K_max=5, gamma_start=0.0, gamma_end=2.0): - G = generate_connected_ER(n=n, m=m, directed=directed) + def assert_champ_correctness_ER(self, n=100, m=1000, directed=False, num_partitions=10, num_gammas=100, + K_max=5, gamma_start=0.0, gamma_end=2.0, 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) gammas = generate_random_values(num_gammas, gamma_start, gamma_end) champ_ranges = CHAMP_2D(G, partitions, gamma_start, gamma_end) self.assert_best_partitions_match_champ_set(G, partitions, champ_ranges, gammas) - def test_champ_correctness_undirected_unweighted_varying_n(self): + def test_champ_correctness_undirected_varying_n(self): for n in [50, 100, 250, 500]: - self.assert_champ_correctness_unweighted_ER(n=n, m=5 * n) + for weighted in [False, True]: + with self.subTest(n=n, weighted=weighted): + self.assert_champ_correctness_ER(n=n, m=5 * n, weighted=weighted) - def test_champ_correctness_undirected_unweighted_varying_m(self): + def test_champ_correctness_undirected_varying_m(self): for m in [200, 500, 1000]: - self.assert_champ_correctness_unweighted_ER(m=m) + for weighted in [False, True]: + with self.subTest(m=m, weighted=weighted): + self.assert_champ_correctness_ER(m=m, weighted=weighted) - def test_champ_correctness_undirected_unweighted_varying_num_partitions(self): + def test_champ_correctness_undirected_varying_num_partitions(self): for num_partitions in [10, 100, 1000]: - self.assert_champ_correctness_unweighted_ER(num_partitions=num_partitions) + for weighted in [False, True]: + with self.subTest(num_partitions=num_partitions, weighted=weighted): + self.assert_champ_correctness_ER(num_partitions=num_partitions, weighted=weighted) - def test_champ_correctness_undirected_unweighted_varying_K_max(self): + def test_champ_correctness_undirected_varying_K_max(self): for K_max in [2, 5, 10, 20]: - self.assert_champ_correctness_unweighted_ER(K_max=K_max) + for weighted in [False, True]: + with self.subTest(K_max=K_max, weighted=weighted): + self.assert_champ_correctness_ER(K_max=K_max, weighted=weighted) - def test_champ_correctness_undirected_unweighted_varying_gamma_range(self): + def test_champ_correctness_undirected_varying_gamma_range(self): for gamma_start, gamma_end in zip([0.0, 1.0, 2.0, 3.0, 4.0], [10.0, 9.0, 8.0, 7.0, 6.0]): - self.assert_champ_correctness_unweighted_ER(gamma_start=gamma_start, gamma_end=gamma_end) + for weighted in [False, True]: + with self.subTest(gamma_start=gamma_start, gamma_end=gamma_end, weighted=weighted): + self.assert_champ_correctness_ER(gamma_start=gamma_start, gamma_end=gamma_end, + weighted=weighted) - def test_champ_correctness_directed_unweighted_varying_n(self): + def test_champ_correctness_directed_varying_n(self): for n in [50, 100, 250, 500]: - self.assert_champ_correctness_unweighted_ER(n=n, m=10 * n, directed=True) + for weighted in [False, True]: + with self.subTest(n=n, weighted=weighted): + self.assert_champ_correctness_ER(n=n, m=10 * n, directed=True, weighted=weighted) - def test_champ_correctness_directed_unweighted_varying_m(self): + def test_champ_correctness_directed_varying_m(self): for m in [400, 1000, 2000]: - self.assert_champ_correctness_unweighted_ER(m=m, directed=True) + for weighted in [False, True]: + with self.subTest(m=m, weighted=weighted): + self.assert_champ_correctness_ER(m=m, directed=True, weighted=weighted) - def test_champ_correctness_directed_unweighted_varying_num_partitions(self): + def test_champ_correctness_directed_varying_num_partitions(self): for num_partitions in [10, 100, 1000]: - self.assert_champ_correctness_unweighted_ER(num_partitions=num_partitions, directed=True) + for weighted in [False, True]: + with self.subTest(num_partitions=num_partitions, weighted=weighted): + self.assert_champ_correctness_ER(num_partitions=num_partitions, directed=True, + weighted=weighted) - def test_champ_correctness_directed_unweighted_varying_K_max(self): + def test_champ_correctness_directed_varying_K_max(self): for K_max in [2, 5, 10, 20]: - self.assert_champ_correctness_unweighted_ER(directed=True, K_max=K_max) + for weighted in [False, True]: + with self.subTest(K_max=K_max, weighted=weighted): + self.assert_champ_correctness_ER(directed=True, K_max=K_max, weighted=weighted) - def test_champ_correctness_directed_unweighted_varying_gamma_range(self): + def test_champ_correctness_directed_varying_gamma_range(self): for gamma_start, gamma_end in zip([0.0, 1.0, 2.0, 3.0, 4.0], [10.0, 9.0, 8.0, 7.0, 6.0]): - self.assert_champ_correctness_unweighted_ER(directed=True, gamma_start=gamma_start, gamma_end=gamma_end) + for weighted in [False, True]: + with self.subTest(gamma_start=gamma_start, gamma_end=gamma_end, weighted=weighted): + self.assert_champ_correctness_ER(directed=True, gamma_start=gamma_start, gamma_end=gamma_end, + weighted=weighted) def test_champ_correctness_igraph_famous_leiden(self): """Test CHAMP correctness on various famous graphs while obtaining partitions via Leiden. @@ -83,16 +106,18 @@ def test_champ_correctness_igraph_famous_leiden(self): The correctness of the CHAMP domains are 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) - champ_ranges = CHAMP_2D(G, partitions, gamma_0=0, gamma_f=5) - self.assert_best_partitions_match_champ_set(G, partitions, champ_ranges, gammas) + with self.subTest(graph=name, directed=False): + partitions = repeated_leiden_from_gammas(G, gammas) + champ_ranges = CHAMP_2D(G, partitions, gamma_0=0, gamma_f=5) + self.assert_best_partitions_match_champ_set(G, partitions, champ_ranges, gammas) G.to_directed() # check the directed version of the graph as well - partitions = repeated_leiden_from_gammas(G, gammas) - champ_ranges = CHAMP_2D(G, partitions, gamma_0=0, gamma_f=5) - self.assert_best_partitions_match_champ_set(G, partitions, champ_ranges, gammas) + with self.subTest(graph=name, directed=True): + partitions = repeated_leiden_from_gammas(G, gammas) + champ_ranges = CHAMP_2D(G, partitions, gamma_0=0, gamma_f=5) + self.assert_best_partitions_match_champ_set(G, partitions, champ_ranges, gammas) if __name__ == "__main__": diff --git a/tests/test_champ_usage_3D.py b/tests/test_champ_usage_3D.py index 2643b81..7e798ff 100644 --- a/tests/test_champ_usage_3D.py +++ b/tests/test_champ_usage_3D.py @@ -5,8 +5,8 @@ from modularitypruning.leiden_utilities import multilayer_leiden_part_with_membership from numpy import mean -from .shared_testing_functions import generate_connected_multilayer_ER, generate_random_values, \ - generate_random_partitions +from .shared_testing_functions import generate_connected_multilayer_ER, \ + generate_random_values, generate_random_partitions def point_is_inside_champ_domain(gamma, omega, domain_vertices): @@ -51,26 +51,28 @@ def assert_best_partitions_match_champ_set(self, G_intralayer, G_interlayer, lay if point_is_inside_champ_domain(gamma, omega, domain_vertices): # check that the best partition quality matches that of the champ domains at this gamma champ_quality = ( - sum(p_intra.quality(resolution_parameter=gamma) - for p_intra in membership_to_intralayer_leiden_partitions[membership]) - + omega * - membership_to_interlayer_leiden_partition[membership].quality(resolution_parameter=0) + sum(p_intra.quality(resolution_parameter=gamma) + for p_intra in membership_to_intralayer_leiden_partitions[membership]) + + omega * + membership_to_interlayer_leiden_partition[membership].quality(resolution_parameter=0) ) # note that this is float comparision to within ~10^{-10} self.assertAlmostEqual(best_partition_quality, champ_quality, places=10, msg=f"CHAMP domain quality does not match best input partition " - f"at gamma {gamma:.2f} omega {omega:.2f}") + f"at gamma {gamma:.2f} omega {omega:.2f}") break else: self.assertFalse(True, msg=f"gamma {gamma:.2f}, omega {omega:.2f} " - f"was not found within any CHAMP domain") + f"was not found within any CHAMP domain") - def assert_champ_correctness_unweighted_ER(self, num_nodes_per_layer=100, m=10000, num_layers=10, directed=False, - num_partitions=10, num_sample_points=100, K_max=10, - gamma_start=0.0, gamma_end=2.0, omega_start=0.0, omega_end=2.0): + def assert_champ_correctness_ER(self, num_nodes_per_layer=100, m=10000, num_layers=10, directed=False, + num_partitions=10, num_sample_points=100, K_max=10, + gamma_start=0.0, gamma_end=2.0, omega_start=0.0, omega_end=2.0, + weighted=False): G_intralayer, G_interlayer, layer_membership = generate_connected_multilayer_ER( - num_nodes_per_layer=num_nodes_per_layer, m=m, num_layers=num_layers, directed=directed) + num_nodes_per_layer=num_nodes_per_layer, m=m, num_layers=num_layers, directed=directed, + weighted=weighted) partitions = generate_random_partitions(num_nodes=G_intralayer.vcount(), num_partitions=num_partitions, K_max=K_max) gammas = generate_random_values(num_sample_points, start_value=gamma_start, end_value=gamma_end) @@ -81,65 +83,100 @@ def assert_champ_correctness_unweighted_ER(self, num_nodes_per_layer=100, m=1000 self.assert_best_partitions_match_champ_set(G_intralayer, G_interlayer, layer_membership, partitions, champ_domains, gammas, omegas) - def test_partition_coefficient_correctness_undirected_unweighted_varying_num_nodes_per_layer(self): + def test_partition_coefficient_correctness_undirected_varying_num_nodes_per_layer(self): for num_nodes_per_layer in [50, 100, 250, 500]: - self.assert_champ_correctness_unweighted_ER(num_nodes_per_layer=num_nodes_per_layer, - m=50 * num_nodes_per_layer) + for weighted in [False, True]: + with self.subTest(num_nodes_per_layer=num_nodes_per_layer, weighted=weighted): + self.assert_champ_correctness_ER(num_nodes_per_layer=num_nodes_per_layer, + m=50 * num_nodes_per_layer, weighted=weighted) - def test_partition_coefficient_correctness_undirected_unweighted_varying_m(self): + def test_partition_coefficient_correctness_undirected_varying_m(self): for m in [5000, 10000, 15000, 20000]: - self.assert_champ_correctness_unweighted_ER(m=m) + for weighted in [False, True]: + with self.subTest(m=m, weighted=weighted): + self.assert_champ_correctness_ER(m=m, weighted=weighted) - def test_partition_coefficient_correctness_undirected_unweighted_varying_num_layers(self): + def test_partition_coefficient_correctness_undirected_varying_num_layers(self): for num_layers in [5, 10, 20, 30]: - self.assert_champ_correctness_unweighted_ER(m=15000, num_layers=num_layers) + for weighted in [False, True]: + with self.subTest(num_layers=num_layers, weighted=weighted): + self.assert_champ_correctness_ER(m=15000, num_layers=num_layers, 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 [5, 10, 100, 250]: - self.assert_champ_correctness_unweighted_ER(num_partitions=num_partitions) + for weighted in [False, True]: + with self.subTest(num_partitions=num_partitions, weighted=weighted): + self.assert_champ_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, 25]: - self.assert_champ_correctness_unweighted_ER(K_max=K_max) + for weighted in [False, True]: + with self.subTest(K_max=K_max, weighted=weighted): + self.assert_champ_correctness_ER(K_max=K_max, weighted=weighted) - def test_partition_coefficient_correctness_undirected_unweighted_varying_gamma_range(self): + def test_partition_coefficient_correctness_undirected_varying_gamma_range(self): for gamma_start, gamma_end in zip([0.0, 1.0, 2.0, 3.0, 4.0], [10.0, 9.0, 8.0, 7.0, 6.0]): - self.assert_champ_correctness_unweighted_ER(gamma_start=gamma_start, gamma_end=gamma_end) + for weighted in [False, True]: + with self.subTest(gamma_start=gamma_start, gamma_end=gamma_end, weighted=weighted): + self.assert_champ_correctness_ER(gamma_start=gamma_start, gamma_end=gamma_end, + weighted=weighted) - def test_partition_coefficient_correctness_undirected_unweighted_varying_omega_range(self): + def test_partition_coefficient_correctness_undirected_varying_omega_range(self): for omega_start, omega_end in zip([0.0, 1.0, 2.0, 3.0, 4.0], [10.0, 9.0, 8.0, 7.0, 6.0]): - self.assert_champ_correctness_unweighted_ER(omega_start=omega_start, omega_end=omega_end) + for weighted in [False, True]: + with self.subTest(omega_start=omega_start, omega_end=omega_end, weighted=weighted): + self.assert_champ_correctness_ER(omega_start=omega_start, omega_end=omega_end, + weighted=weighted) - def test_partition_coefficient_correctness_directed_unweighted_varying_num_nodes_per_layer(self): + def test_partition_coefficient_correctness_directed_varying_num_nodes_per_layer(self): for num_nodes_per_layer in [50, 100, 250, 500]: - self.assert_champ_correctness_unweighted_ER(num_nodes_per_layer=num_nodes_per_layer, - m=100 * num_nodes_per_layer, directed=True) + for weighted in [False, True]: + with self.subTest(num_nodes_per_layer=num_nodes_per_layer, weighted=weighted): + self.assert_champ_correctness_ER(num_nodes_per_layer=num_nodes_per_layer, + m=100 * num_nodes_per_layer, 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 [10000, 20000, 30000, 40000]: - self.assert_champ_correctness_unweighted_ER(m=m, directed=True) + for weighted in [False, True]: + with self.subTest(m=m, weighted=weighted): + self.assert_champ_correctness_ER(m=m, directed=True, weighted=weighted) - def test_partition_coefficient_correctness_directed_unweighted_varying_num_layers(self): + def test_partition_coefficient_correctness_directed_varying_num_layers(self): for num_layers in [5, 10, 20, 30]: - self.assert_champ_correctness_unweighted_ER(m=30000, num_layers=num_layers, directed=True) + for weighted in [False, True]: + with self.subTest(num_layers=num_layers, weighted=weighted): + self.assert_champ_correctness_ER(m=30000, num_layers=num_layers, 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 [5, 10, 100, 250]: - self.assert_champ_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_champ_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, 25]: - self.assert_champ_correctness_unweighted_ER(directed=True, K_max=K_max) + for weighted in [False, True]: + with self.subTest(K_max=K_max, weighted=weighted): + self.assert_champ_correctness_ER(directed=True, K_max=K_max, weighted=weighted) - def test_partition_coefficient_correctness_directed_unweighted_varying_gamma_range(self): + def test_partition_coefficient_correctness_directed_varying_gamma_range(self): for gamma_start, gamma_end in zip([0.0, 1.0, 2.0, 3.0, 4.0], [10.0, 9.0, 8.0, 7.0, 6.0]): - self.assert_champ_correctness_unweighted_ER(directed=True, - gamma_start=gamma_start, gamma_end=gamma_end) + for weighted in [False, True]: + with self.subTest(gamma_start=gamma_start, gamma_end=gamma_end, weighted=weighted): + self.assert_champ_correctness_ER(directed=True, + gamma_start=gamma_start, gamma_end=gamma_end, + weighted=weighted) - def test_partition_coefficient_correctness_directed_unweighted_varying_omega_range(self): + def test_partition_coefficient_correctness_directed_varying_omega_range(self): for omega_start, omega_end in zip([0.0, 1.0, 2.0, 3.0, 4.0], [10.0, 9.0, 8.0, 7.0, 6.0]): - self.assert_champ_correctness_unweighted_ER(directed=True, - omega_start=omega_start, omega_end=omega_end) + for weighted in [False, True]: + with self.subTest(omega_start=omega_start, omega_end=omega_end, weighted=weighted): + self.assert_champ_correctness_ER(directed=True, + omega_start=omega_start, omega_end=omega_end, + weighted=weighted) if __name__ == "__main__": diff --git a/tests/test_monolayer_parameter_estimation.py b/tests/test_monolayer_parameter_estimation.py index 5c67598..ded178e 100644 --- a/tests/test_monolayer_parameter_estimation.py +++ b/tests/test_monolayer_parameter_estimation.py @@ -5,11 +5,11 @@ import igraph as ig from modularitypruning.parameter_estimation import iterative_monolayer_resolution_parameter_estimation from modularitypruning.parameter_estimation_utilities import gamma_estimate -from modularitypruning.partition_utilities import all_degrees +from modularitypruning.partition_utilities import all_degrees, all_strengths from numpy import mean from .shared_testing_functions import assert_almost_equal_or_both_none_or_nan, generate_igraph_famous, \ - generate_random_partition + generate_random_partition, generate_random_values class TestMonolayerParameterEstimation(unittest.TestCase): @@ -19,36 +19,58 @@ def test_newman_synthetic_networks(self): The setup here is the same as in FIG 1 of 'Equivalence between modularity optimization and maximum likelihood methods for community detection', albeit using Leiden for modularity maximization.""" - for q in range(3, 15): - community_sizes = [250] * q # q equally sized groups of 250 nodes - n = sum(community_sizes) - p_in = 16 * n / (q * 250 * 249) # ~16 in-edges per node - p_out = 8 * n / (q * (q - 1) * 250 * 250) # ~8 out-edges per node to each community - pref_matrix = [[p_in if i == j else p_out for j in range(q)] for i in range(q)] - G = ig.Graph.SBM(pref_matrix, community_sizes) + for weighted in [False, True]: + for q in range(3, 15): + with self.subTest(q=q, weighted=weighted): + self.assert_newman_synthetic_convergence(q, weighted=weighted) - # compute "ground truth" gamma estimate + @staticmethod + def generate_monolayer_SBM(q, weighted=False): + community_sizes = [250] * q # q equally sized groups of 250 nodes + n = sum(community_sizes) + p_in = 16 * n / (q * 250 * 249) # ~16 in-edges per node + p_out = 8 * n / (q * (q - 1) * 250 * 250) # ~8 out-edges per node to each community + pref_matrix = [[p_in if i == j else p_out for j in range(q)] for i in range(q)] + G = ig.Graph.SBM(pref_matrix, community_sizes) + + if weighted: + G.es['weight'] = generate_random_values(G.ecount(), 0.5, 5.0) + + return G, p_in, p_out + + def assert_newman_synthetic_convergence(self, q, weighted=False): + G, p_in, p_out = self.generate_monolayer_SBM(q, weighted=weighted) + + # compute "ground truth" gamma estimate + if weighted: + k = mean(all_strengths(G)) + m = sum(G.es['weight']) + mean_weight = m / G.ecount() + true_omega_in = mean_weight * p_in * (2 * m) / (k * k) + true_omega_out = mean_weight * p_out * (2 * m) / (k * k) + else: k = mean(all_degrees(G)) true_omega_in = p_in * (2 * G.ecount()) / (k * k) true_omega_out = p_out * (2 * G.ecount()) / (k * k) - true_gamma = (true_omega_in - true_omega_out) / (log(true_omega_in) - log(true_omega_out)) + true_gamma = (true_omega_in - true_omega_out) / (log(true_omega_in) - log(true_omega_out)) - gamma, _ = iterative_monolayer_resolution_parameter_estimation(G, gamma=1.0) + gamma, _ = iterative_monolayer_resolution_parameter_estimation(G, gamma=1.0) - # check we converged close to the ground truth "correct" value - self.assertLess(abs(true_gamma - gamma), 0.05) + # check we converged close to the ground truth "correct" value + self.assertLess(abs(true_gamma - gamma), 0.05) def test_directed_consistency_igraph_famous(self): """Test gamma estimate consistency on undirected and (symmetric) directed versions of various famous graphs.""" - for G in generate_igraph_famous(): + for name, G in generate_igraph_famous(): random_membership = generate_random_partition(G.vcount(), 5) gamma_undirected = gamma_estimate(G, random_membership) G.to_directed() gamma_directed = gamma_estimate(G, random_membership) - assert_almost_equal_or_both_none_or_nan(self, gamma_undirected, gamma_directed, places=10) + with self.subTest(graph=name): + assert_almost_equal_or_both_none_or_nan(self, gamma_undirected, gamma_directed, places=10) if __name__ == "__main__": diff --git a/tests/test_multiplex_parameter_estimation.py b/tests/test_multiplex_parameter_estimation.py index 84fee43..77cb040 100644 --- a/tests/test_multiplex_parameter_estimation.py +++ b/tests/test_multiplex_parameter_estimation.py @@ -10,13 +10,15 @@ from numpy import mean from .shared_testing_functions import assert_almost_equal_or_both_none_or_nan, generate_random_partition, \ - generate_multilayer_intralayer_SBM + generate_connected_multilayer_intralayer_SBM class TestMultiplexParameterEstimation(unittest.TestCase): - def generate_multiplex_SBM(self, copying_probability, p_in, p_out, first_layer_membership, num_layers): - G_intralayer, layer_membership = generate_multilayer_intralayer_SBM(copying_probability, p_in, p_out, - first_layer_membership, num_layers) + def generate_multiplex_SBM(self, copying_probability, p_in, p_out, first_layer_membership, num_layers, + weighted=False): + G_intralayer, layer_membership = generate_connected_multilayer_intralayer_SBM( + copying_probability, p_in, p_out, first_layer_membership, num_layers, weighted=weighted + ) # connect each node to itself in all other layers num_nodes_per_layer = len(first_layer_membership) @@ -29,11 +31,11 @@ def generate_multiplex_SBM(self, copying_probability, p_in, p_out, first_layer_m return G_intralayer, G_interlayer, layer_membership def assert_multiplex_SBM_correct_convergence(self, first_layer_membership, copying_probability=0.75, num_layers=10, - p_in=0.25, p_out=0.05): + p_in=0.25, p_out=0.05, weighted=False): K = num_communities(first_layer_membership) G_intralayer, G_interlayer, layer_membership = self.generate_multiplex_SBM(copying_probability, p_in, p_out, first_layer_membership, - num_layers) + num_layers, weighted=weighted) # compute ground truth gamma k = mean(all_degrees(G_intralayer)) @@ -57,30 +59,46 @@ def assert_multiplex_SBM_correct_convergence(self, first_layer_membership, copyi self.assertLess(abs(true_omega - omega), 0.2) def test_multiplex_SBM_correct_convergence_varying_copying_probabilty(self): - for eta in [0.25, 0.5, 0.75, 0.9]: - membership = generate_random_partition(num_nodes=200, K=2) - self.assert_multiplex_SBM_correct_convergence(copying_probability=eta, - first_layer_membership=membership) + for weighted in [False, True]: + for eta in [0.25, 0.5, 0.75, 0.9]: + with self.subTest(eta=eta, weighted=weighted): + membership = generate_random_partition(num_nodes=200, K=2) + self.assert_multiplex_SBM_correct_convergence(copying_probability=eta, + first_layer_membership=membership, + weighted=weighted) def test_multiplex_SBM_correct_convergence_varying_p_in(self): - for p_in in [0.5, 0.4, 0.3, 0.2]: - membership = generate_random_partition(num_nodes=200, K=2) - self.assert_multiplex_SBM_correct_convergence(p_in=p_in, p_out=0.025, first_layer_membership=membership) + for weighted in [False, True]: + for p_in in [0.5, 0.4, 0.3, 0.2]: + with self.subTest(p_in=p_in, weighted=weighted): + membership = generate_random_partition(num_nodes=200, K=2) + self.assert_multiplex_SBM_correct_convergence(p_in=p_in, p_out=0.025, + first_layer_membership=membership, + weighted=weighted) def test_multiplex_SBM_correct_convergence_varying_p_out(self): - for p_out in [0.05, 0.04, 0.03, 0.02]: - membership = generate_random_partition(num_nodes=100, K=2) - self.assert_multiplex_SBM_correct_convergence(p_out=p_out, first_layer_membership=membership) + for weighted in [False, True]: + for p_out in [0.05, 0.04, 0.03, 0.02]: + with self.subTest(p_out=p_out, weighted=weighted): + membership = generate_random_partition(num_nodes=200, K=2) + self.assert_multiplex_SBM_correct_convergence(p_out=p_out, first_layer_membership=membership, + weighted=weighted) def test_multiplex_SBM_correct_convergence_varying_num_communities(self): - for K in [2, 3, 4, 5]: - membership = generate_random_partition(num_nodes=300, K=K) - self.assert_multiplex_SBM_correct_convergence(first_layer_membership=membership) + for weighted in [False, True]: + for K in [2, 3, 4, 5]: + with self.subTest(K=K, weighted=weighted): + membership = generate_random_partition(num_nodes=300, K=K) + self.assert_multiplex_SBM_correct_convergence(first_layer_membership=membership, + weighted=weighted) def test_multiplex_SBM_correct_convergence_varying_num_layers(self): - for num_layers in [3, 5, 7, 10]: - membership = generate_random_partition(num_nodes=300, K=2) - self.assert_multiplex_SBM_correct_convergence(first_layer_membership=membership, num_layers=num_layers) + for weighted in [False, True]: + for num_layers in [3, 5, 7, 10]: + with self.subTest(num_layers=num_layers, weighted=weighted): + membership = generate_random_partition(num_nodes=300, K=2) + self.assert_multiplex_SBM_correct_convergence(first_layer_membership=membership, + num_layers=num_layers, weighted=weighted) def test_directed_consistency_multiplex_SBM_leiden(self): """Test parameter estimate consistency on a multiplex SBM when the intralayer edges are directed.""" @@ -94,17 +112,18 @@ def test_directed_consistency_multiplex_SBM_leiden(self): gammas=[0.5, 1.0, 1.5], omegas=[0.5, 1.0, 1.5]) for partition in partitions: - # here, undirected/directed refers to the intralayer edges only - # in Pamfil et al.'s multiplex networks, interlayer edges are taken to be directed - gamma_undirected, omega_undirected = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, + with self.subTest(partition=partition): + # here, undirected/directed refers to the intralayer edges only + # in Pamfil et al.'s multiplex networks, interlayer edges are taken to be directed + gamma_undirected, omega_undirected = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, + partition, model="multiplex") + + G_intralayer.to_directed() + gamma_directed, omega_directed = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, partition, model="multiplex") - G_intralayer.to_directed() - gamma_directed, omega_directed = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, - partition, model="multiplex") - - assert_almost_equal_or_both_none_or_nan(self, gamma_undirected, gamma_directed, places=10) - assert_almost_equal_or_both_none_or_nan(self, omega_undirected, omega_directed, places=10) + assert_almost_equal_or_both_none_or_nan(self, gamma_undirected, gamma_directed, places=10) + assert_almost_equal_or_both_none_or_nan(self, omega_undirected, omega_directed, places=10) if __name__ == "__main__": diff --git a/tests/test_parallel_leiden_performance.py b/tests/test_parallel_leiden_performance.py index d4a3fa6..7adcc77 100644 --- a/tests/test_parallel_leiden_performance.py +++ b/tests/test_parallel_leiden_performance.py @@ -13,7 +13,7 @@ repeated_leiden_from_gammas_omegas, repeated_parallel_leiden_from_gammas_omegas) -from .shared_testing_functions import generate_connected_ER, generate_multilayer_intralayer_SBM +from .shared_testing_functions import generate_connected_ER, generate_connected_multilayer_intralayer_SBM # this set of tests ensures that we achieve >= 75% parallel performance compared to perfect scaling of # single-threaded jobs to multiple cores (with no memory contention). This threshold will be decreased in @@ -115,7 +115,7 @@ def test_larger_singlelayer_graph_few_runs(self): def test_tiny_multilayer_graph_many_runs(self): """Single-threaded equivalent is 10k runs on G(n=50, m=150).""" - G_intralayer, layer_membership = generate_multilayer_intralayer_SBM( + G_intralayer, layer_membership = generate_connected_multilayer_intralayer_SBM( copying_probability=0.9, p_in=0.8, p_out=0.2, first_layer_membership=[0] * 5 + [1] * 5, num_layers=5 ) interlayer_edges = [(10 * layer + v, 10 * layer + v + 10) @@ -130,7 +130,7 @@ def test_tiny_multilayer_graph_many_runs(self): def test_larger_multilayer_graph_few_runs(self): """Single-threaded equivalent is 49 runs on approximately G(n=2500, m=15000).""" - G_intralayer, layer_membership = generate_multilayer_intralayer_SBM( + G_intralayer, layer_membership = generate_connected_multilayer_intralayer_SBM( copying_probability=0.9, p_in=0.15, p_out=0.05, first_layer_membership=[0] * 50 + [1] * 50, num_layers=25 ) interlayer_edges = [(100 * layer + v, 100 * layer + v + 100) diff --git a/tests/test_temporal_multilevel_parameter_estimation.py b/tests/test_temporal_multilevel_parameter_estimation.py index e88f36c..b2867e3 100644 --- a/tests/test_temporal_multilevel_parameter_estimation.py +++ b/tests/test_temporal_multilevel_parameter_estimation.py @@ -10,13 +10,15 @@ from numpy import mean from .shared_testing_functions import assert_almost_equal_or_both_none_or_nan, generate_random_partition, \ - generate_multilayer_intralayer_SBM + generate_connected_multilayer_intralayer_SBM class TestTemporalAndMultilevelParameterEstimation(unittest.TestCase): - def generate_temporal_SBM(self, copying_probability, p_in, p_out, first_layer_membership, num_layers): - G_intralayer, layer_membership = generate_multilayer_intralayer_SBM(copying_probability, p_in, p_out, - first_layer_membership, num_layers) + def generate_temporal_SBM(self, copying_probability, p_in, p_out, first_layer_membership, num_layers, + weighted=False): + G_intralayer, layer_membership = generate_connected_multilayer_intralayer_SBM( + copying_probability, p_in, p_out, first_layer_membership, num_layers, weighted=weighted + ) # connect each node to itself in the next layer num_nodes_per_layer = len(first_layer_membership) @@ -26,12 +28,12 @@ def generate_temporal_SBM(self, copying_probability, p_in, p_out, first_layer_me return G_intralayer, G_interlayer, layer_membership - def assert_temporal_SBM_correct_convergence(self, first_layer_membership, copying_probability=0.75, num_layers=25, - p_in=0.25, p_out=0.05, tolerance=0.1): + def assert_temporal_SBM_correct_convergence(self, first_layer_membership, copying_probability=0.75, num_layers=12, + p_in=0.25, p_out=0.05, tolerance=0.1, weighted=False): K = num_communities(first_layer_membership) G_intralayer, G_interlayer, layer_membership = self.generate_temporal_SBM(copying_probability, p_in, p_out, first_layer_membership, - num_layers) + num_layers, weighted=weighted) # compute ground truth gamma k = mean(all_degrees(G_intralayer)) @@ -60,7 +62,7 @@ def assert_temporal_SBM_correct_convergence(self, first_layer_membership, copyin self.assertLess(abs(true_gamma - gamma), tolerance) self.assertLess(abs(true_omega - omega), tolerance) - def retry_test_across_graph_sizes(self, test_fn, sizes=[50, 100, 250, 500], retries_per_size=3): + def retry_test_across_graph_sizes(self, test_fn, sizes=[250, 500], retries_per_size=3): """ Rerun randomized tests of asymptotic convergence across increasingly larger graph sizes until one succeeds. @@ -72,56 +74,71 @@ def retry_test_across_graph_sizes(self, test_fn, sizes=[50, 100, 250, 500], retr try: test_fn(n) return - except Exception as e: + except (AssertionError, TimeoutError) as e: last_error = e - if not isinstance(e, AssertionError): - # test failed in unusual way (perhaps keyboard interrupt?), raising immediately - raise e + except Exception as e: + # test failed in unusual way (perhaps keyboard interrupt?), raising immediately + raise e assert last_error is not None raise last_error def test_temporal_SBM_correct_convergence_varying_copying_probabilty(self): - for eta in [0.25, 0.5, 0.75, 0.9]: - self.retry_test_across_graph_sizes( - test_fn=lambda n: self.assert_temporal_SBM_correct_convergence( - copying_probability=eta, - first_layer_membership=generate_random_partition(num_nodes=n, K=2) - ) - ) + for weighted in [False, True]: + for eta in [0.25, 0.5, 0.75, 0.9]: + with self.subTest(eta=eta, weighted=weighted): + self.retry_test_across_graph_sizes( + test_fn=lambda n, w=weighted: self.assert_temporal_SBM_correct_convergence( + copying_probability=eta, + first_layer_membership=generate_random_partition(num_nodes=n, K=2), + weighted=w + ) + ) def test_temporal_SBM_correct_convergence_varying_p_in(self): - for p_in in [0.5, 0.4, 0.3, 0.2]: - self.retry_test_across_graph_sizes( - test_fn=lambda n: self.assert_temporal_SBM_correct_convergence( - p_in=p_in, p_out=0.025, - first_layer_membership=generate_random_partition(num_nodes=n, K=2) - ) - ) + for weighted in [False, True]: + for p_in in [0.5, 0.4, 0.3, 0.2]: + with self.subTest(p_in=p_in, weighted=weighted): + self.retry_test_across_graph_sizes( + test_fn=lambda n, w=weighted: self.assert_temporal_SBM_correct_convergence( + p_in=p_in, p_out=0.025, + first_layer_membership=generate_random_partition(num_nodes=n, K=2), + weighted=w + ) + ) def test_temporal_SBM_correct_convergence_varying_p_out(self): - for p_out in [0.05, 0.04, 0.03, 0.02]: - self.retry_test_across_graph_sizes( - test_fn=lambda n: self.assert_temporal_SBM_correct_convergence( - p_out=p_out, first_layer_membership=generate_random_partition(num_nodes=n, K=2) - ) - ) + for weighted in [False, True]: + for p_out in [0.05, 0.04, 0.03, 0.02]: + with self.subTest(p_out=p_out, weighted=weighted): + self.retry_test_across_graph_sizes( + test_fn=lambda n, w=weighted: self.assert_temporal_SBM_correct_convergence( + p_out=p_out, first_layer_membership=generate_random_partition(num_nodes=n, K=2), + weighted=w + ) + ) def test_temporal_SBM_correct_convergence_varying_num_communities(self): - for K in [2, 3, 4, 5]: - self.retry_test_across_graph_sizes( - test_fn=lambda n: self.assert_temporal_SBM_correct_convergence( - first_layer_membership=generate_random_partition(num_nodes=n, K=K) - ) - ) + for weighted in [False, True]: + for K in [2, 3, 4, 5]: + with self.subTest(K=K, weighted=weighted): + self.retry_test_across_graph_sizes( + test_fn=lambda n, K_=K, w=weighted: self.assert_temporal_SBM_correct_convergence( + first_layer_membership=generate_random_partition(num_nodes=n, K=K_), + weighted=w + ) + ) def test_temporal_SBM_correct_convergence_varying_num_layers(self): - for num_layers in [20, 30, 40]: - self.retry_test_across_graph_sizes( - test_fn=lambda n: self.assert_temporal_SBM_correct_convergence( - num_layers=num_layers, - first_layer_membership=generate_random_partition(num_nodes=n, K=2) - ) - ) + for weighted in [False, True]: + for num_layers in [8, 12, 16]: + with self.subTest(num_layers=num_layers, weighted=weighted): + self.retry_test_across_graph_sizes( + test_fn=lambda n, nl=num_layers, w=weighted: self.assert_temporal_SBM_correct_convergence( + num_layers=nl, + first_layer_membership=generate_random_partition(num_nodes=n, K=2), + weighted=w + ) + ) def test_directed_consistency_temporal_SBM_leiden(self): """Test parameter estimate consistency on a temporal SBM when the intralayer edges are directed.""" @@ -134,29 +151,32 @@ def test_directed_consistency_temporal_SBM_leiden(self): partitions = repeated_leiden_from_gammas_omegas(G_intralayer, G_interlayer, layer_membership, gammas=[0.5, 1.0, 1.5], omegas=[0.5, 1.0, 1.5]) - for partition in partitions: - # here, undirected/directed refers to the intralayer edges only - # in Pamfil et al.'s temporal networks, interlayer edges are taken to be directed - gamma_undirected, omega_undirected = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, + for i, partition in enumerate(partitions): + with self.subTest(partition=i): + # here, undirected/directed refers to the intralayer edges only + # in Pamfil et al.'s temporal networks, interlayer edges are taken to be directed + G_intralayer.to_undirected() + gamma_undirected, omega_undirected = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, + partition, model="temporal") + + G_intralayer.to_directed() + gamma_directed, omega_directed = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, partition, model="temporal") - G_intralayer.to_directed() - gamma_directed, omega_directed = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, - partition, model="temporal") + assert_almost_equal_or_both_none_or_nan(self, gamma_undirected, gamma_directed, places=10) + assert_almost_equal_or_both_none_or_nan(self, omega_undirected, omega_directed, places=10) - assert_almost_equal_or_both_none_or_nan(self, gamma_undirected, gamma_directed, places=10) - assert_almost_equal_or_both_none_or_nan(self, omega_undirected, omega_directed, places=10) + # check multilevel parameter estimation as well + G_intralayer.to_undirected() + gamma_undirected, omega_undirected = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, + partition, model="multilevel") - # check multilevel parameter estimation as well - gamma_undirected, omega_undirected = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, + G_intralayer.to_directed() + gamma_directed, omega_directed = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, partition, model="multilevel") - G_intralayer.to_directed() - gamma_directed, omega_directed = gamma_omega_estimate(G_intralayer, G_interlayer, layer_membership, - partition, model="multilevel") - - assert_almost_equal_or_both_none_or_nan(self, gamma_undirected, gamma_directed, places=10) - assert_almost_equal_or_both_none_or_nan(self, omega_undirected, omega_directed, places=10) + assert_almost_equal_or_both_none_or_nan(self, gamma_undirected, gamma_directed, places=10) + assert_almost_equal_or_both_none_or_nan(self, omega_undirected, omega_directed, places=10) if __name__ == "__main__": diff --git a/utilities/champ_utilities.py b/utilities/champ_utilities.py index 37b79d9..a135bef 100644 --- a/utilities/champ_utilities.py +++ b/utilities/champ_utilities.py @@ -13,9 +13,9 @@ from scipy.optimize import linprog, OptimizeWarning from scipy.spatial import HalfspaceIntersection -from .partition_utilities import all_degrees, in_degrees, out_degrees, membership_to_communities, \ - membership_to_layered_communities -from .types import (Domain, FloatArray, GammaRange, IntArray, LayerVector, Membership, +from .partition_utilities import (all_strengths, in_strengths, out_strengths, + membership_to_communities, membership_to_layered_communities) +from .types import (Domain, FloatArray, GammaRange, LayerVector, Membership, Partitions, PartitionCoefficients2D, PartitionCoefficients3D) @@ -246,32 +246,33 @@ def CHAMP_3D( def partition_coefficients_2D_serial(G: "igraph.Graph", partitions: Partitions ) -> PartitionCoefficients2D: - """Computes A_hat and P_hat for partitions of :G: + """Computes A_hat and P_hat for partitions of :G:""" - TODO: support edge weights""" - - all_edges = [(e.source, e.target) for e in G.es] + weights = G.es['weight'] if G.is_weighted() else [1.0] * G.ecount() + total_weight = float(sum(weights)) + all_edges = [(e.source, e.target, w) for e, w in zip(G.es, weights)] # multiply by 2 only if undirected here if G.is_directed(): - A_hats = np.array([sum([membership[u] == membership[v] for u, v in all_edges]) - for membership in partitions]) + A_hats = np.array([sum(w for u, v, w in all_edges if membership[u] == membership[v]) + for membership in partitions], dtype=np.float64) else: - A_hats = np.array([2 * sum([membership[u] == membership[v] for u, v in all_edges]) - for membership in partitions]) + A_hats = np.array([2 * sum(w for u, v, w in all_edges if membership[u] == membership[v]) + for membership in partitions], dtype=np.float64) if G.is_directed(): # directed modularity of Leicht and Newman is actually # (1/m) sum_{ij} [A_{ij} - k_i^{in} * k_j^{out} / m] delta(c_i, c_j) - in_degree = in_degrees(G) - out_degree = out_degrees(G) - P_hats = np.array([sum(sum(in_degree[v] for v in vs) * sum(out_degree[v] for v in vs) + in_strength = in_strengths(G) + out_strength = out_strengths(G) + P_hats = np.array([sum(sum(in_strength[v] for v in vs) * sum(out_strength[v] for v in vs) for vs in membership_to_communities(membership).values()) - for membership in partitions]) / G.ecount() + for membership in partitions]) / total_weight else: - twom = 2 * G.ecount() - degree = all_degrees(G) - P_hats = np.array([sum(sum(degree[v] for v in vs) ** 2 for vs in membership_to_communities(membership).values()) + twom = 2 * total_weight + strength = all_strengths(G) + P_hats = np.array([sum(sum(strength[v] for v in vs) ** 2 + for vs in membership_to_communities(membership).values()) for membership in partitions]) / twom return A_hats, P_hats @@ -306,7 +307,7 @@ def partition_coefficients_2D( def halfspaces_from_coefficients_2D( - A_hats: IntArray, P_hats: FloatArray + A_hats: FloatArray, P_hats: FloatArray ) -> FloatArray: """Converts partitions' coefficients to halfspace normal, offset @@ -318,7 +319,7 @@ def halfspaces_from_coefficients_2D( def halfspaces_from_coefficients_3D( - A_hats: IntArray, P_hats: FloatArray, C_hats: IntArray + A_hats: FloatArray, P_hats: FloatArray, C_hats: FloatArray ) -> FloatArray: """Converts partitions' coefficients to halfspace normal, offset @@ -333,49 +334,49 @@ def partition_coefficients_3D_serial(G_intralayer: "igraph.Graph", G_interlayer: layer_vec: LayerVector, partitions: list[Membership]) -> PartitionCoefficients3D: """Computes A_hat, P_hat, C_hat for partitions of a graph with intralayer edges given in :G_intralayer:, - interlayer edges given in :G_interlayer:, and layer membership :layer_vec: - - TODO: support edge weights""" + interlayer edges given in :G_interlayer:, and layer membership :layer_vec:""" - all_intralayer_edges = [(e.source, e.target) for e in G_intralayer.es] - all_interlayer_edges = [(e.source, e.target) for e in G_interlayer.es] + intra_weights = G_intralayer.es['weight'] if G_intralayer.is_weighted() else [1.0] * G_intralayer.ecount() + inter_weights = G_interlayer.es['weight'] if G_interlayer.is_weighted() else [1.0] * G_interlayer.ecount() + all_intralayer_edges = [(e.source, e.target, w) for e, w in zip(G_intralayer.es, intra_weights)] + all_interlayer_edges = [(e.source, e.target, w) for e, w in zip(G_interlayer.es, inter_weights)] # multiply by 2 only if undirected here if G_intralayer.is_directed(): - A_hats = np.array([sum([membership[u] == membership[v] for u, v in all_intralayer_edges]) - for membership in partitions]) + A_hats = np.array([sum(w for u, v, w in all_intralayer_edges if membership[u] == membership[v]) + for membership in partitions], dtype=np.float64) else: - A_hats = np.array([2 * sum([membership[u] == membership[v] for u, v in all_intralayer_edges]) - for membership in partitions]) + A_hats = np.array([2 * sum(w for u, v, w in all_intralayer_edges if membership[u] == membership[v]) + for membership in partitions], dtype=np.float64) num_layers = max(layer_vec) + 1 - ecount_per_layer = np.zeros(num_layers) - for e in G_intralayer.es: - ecount_per_layer[layer_vec[e.source]] += 1 + weight_per_layer = np.zeros(num_layers) + for e, w in zip(G_intralayer.es, intra_weights): + weight_per_layer[layer_vec[e.source]] += w if G_intralayer.is_directed(): - in_degree = in_degrees(G_intralayer) - out_degree = out_degrees(G_intralayer) + in_strength = in_strengths(G_intralayer) + out_strength = out_strengths(G_intralayer) P_hats = np.array([ - sum(sum(in_degree[v] for v in vs) * sum(out_degree[v] for v in vs) / ecount_per_layer[layer] + sum(sum(in_strength[v] for v in vs) * sum(out_strength[v] for v in vs) / weight_per_layer[layer] for (_, layer), vs in membership_to_layered_communities(membership, layer_vec).items()) for membership in partitions ]) else: - degree = all_degrees(G_intralayer) + strength = all_strengths(G_intralayer) P_hats = np.array([ - sum(sum(degree[v] for v in vs) ** 2 / (2 * ecount_per_layer[layer]) + sum(sum(strength[v] for v in vs) ** 2 / (2 * weight_per_layer[layer]) for (_, layer), vs in membership_to_layered_communities(membership, layer_vec).items()) for membership in partitions ]) # multiply by 2 only if undirected here if G_interlayer.is_directed(): - C_hats = np.array([sum([membership[u] == membership[v] for u, v in all_interlayer_edges]) - for membership in partitions]) + C_hats = np.array([sum(w for u, v, w in all_interlayer_edges if membership[u] == membership[v]) + for membership in partitions], dtype=np.float64) else: - C_hats = np.array([2 * sum([membership[u] == membership[v] for u, v in all_interlayer_edges]) - for membership in partitions]) + C_hats = np.array([2 * sum(w for u, v, w in all_interlayer_edges if membership[u] == membership[v]) + for membership in partitions], dtype=np.float64) return A_hats, P_hats, C_hats @@ -397,9 +398,9 @@ def partition_coefficients_3D( for partition_chunk in partition_chunks]) pool.close() - A_hats: IntArray = np.array([v for A_hats, P_hats, C_hats in results for v in A_hats]) + A_hats: FloatArray = np.array([v for A_hats, P_hats, C_hats in results for v in A_hats]) P_hats: FloatArray = np.array([v for A_hats, P_hats, C_hats in results for v in P_hats]) - C_hats: IntArray = np.array([v for A_hats, P_hats, C_hats in results for v in C_hats]) + C_hats: FloatArray = np.array([v for A_hats, P_hats, C_hats in results for v in C_hats]) assert len(A_hats) == len(P_hats) == len(C_hats) == len(partitions) diff --git a/utilities/leiden_utilities.py b/utilities/leiden_utilities.py index d060884..751863e 100644 --- a/utilities/leiden_utilities.py +++ b/utilities/leiden_utilities.py @@ -60,7 +60,7 @@ def singlelayer_leiden( :return: partition from leidenalg :rtype: tuple[int, ...] or leidenalg.RBConfigurationVertexPartition """ - if 'weight' not in G.es: + if not G.is_weighted(): G.es['weight'] = [1.0] * G.ecount() partition: LeidenPartition = leidenalg.find_partition( @@ -79,6 +79,8 @@ def _wrapped_singlelayer_leiden(args: tuple[ig.Graph, float]) -> Membership: def leiden_part(G: ig.Graph) -> LeidenPartition: + if G.is_weighted(): + return leidenalg.RBConfigurationVertexPartition(G, weights='weight') return leidenalg.RBConfigurationVertexPartition(G) @@ -175,10 +177,10 @@ def multilayer_leiden( :return: partition from leidenalg :rtype: tuple[int, ...] or list[leidenalg.RBConfigurationVertexPartition] """ - if 'weight' not in G_intralayer.es: + if not G_intralayer.is_weighted(): G_intralayer.es['weight'] = [1.0] * G_intralayer.ecount() - if 'weight' not in G_interlayer.es: + if not G_interlayer.is_weighted(): G_interlayer.es['weight'] = [1.0] * G_interlayer.ecount() G_split_layers: list[ig.Graph] = split_intralayer_leiden_graph(G_intralayer, layer_vec) @@ -211,10 +213,10 @@ def _wrapped_multilayer_leiden(args: tuple[ig.Graph, ig.Graph, LayerVector, floa def multilayer_leiden_part( G_intralayer: ig.Graph, G_interlayer: ig.Graph, layer_membership: LayerVector ) -> MultilayerLeidenParts: - if 'weight' not in G_intralayer.es: + if not G_intralayer.is_weighted(): G_intralayer.es['weight'] = [1.0] * G_intralayer.ecount() - if 'weight' not in G_interlayer.es: + if not G_interlayer.is_weighted(): G_interlayer.es['weight'] = [1.0] * G_interlayer.ecount() G_split_layers: list[ig.Graph] = split_intralayer_leiden_graph(G_intralayer, layer_membership) diff --git a/utilities/parameter_estimation.py b/utilities/parameter_estimation.py index 321b324..3c1a8ca 100644 --- a/utilities/parameter_estimation.py +++ b/utilities/parameter_estimation.py @@ -39,7 +39,7 @@ def iterative_monolayer_resolution_parameter_estimation( :rtype: tuple[float, leidenalg.RBConfigurationVertexPartition] """ - if 'weight' not in G.es: + if not G.is_weighted(): G.es['weight'] = [1.0] * G.ecount() m: float = sum(G.es['weight']) @@ -185,10 +185,10 @@ def iterative_multilayer_resolution_parameter_estimation( :rtype: tuple[float, float, list[leidenalg.RBConfigurationVertexPartition]] """ - if 'weight' not in G_intralayer.es: + if not G_intralayer.is_weighted(): G_intralayer.es['weight'] = [1.0] * G_intralayer.ecount() - if 'weight' not in G_interlayer.es: + if not G_interlayer.is_weighted(): G_interlayer.es['weight'] = [1.0] * G_interlayer.ecount() T: int = max(layer_vec) + 1 # layer count diff --git a/utilities/parameter_estimation_utilities.py b/utilities/parameter_estimation_utilities.py index f2b5ee2..1814cd5 100644 --- a/utilities/parameter_estimation_utilities.py +++ b/utilities/parameter_estimation_utilities.py @@ -34,6 +34,9 @@ def estimate_singlelayer_SBM_parameters( :rtype: tuple[float, float] """ + if not G.is_weighted(): + G.es['weight'] = [1.0] * G.ecount() + if m is None: m = sum(G.es['weight']) @@ -85,7 +88,7 @@ def estimate_multilayer_SBM_parameters( :rtype: float, float, float, int """ - if 'weight' not in G_intralayer.es: + if not G_intralayer.is_weighted(): G_intralayer.es['weight'] = [1.0] * G_intralayer.ecount() # TODO: check if these None parameters and potentially caching calculate_persistence helps performance @@ -167,8 +170,8 @@ def gamma_estimate(G: ig.Graph, partition: LeidenPartition | Membership) -> Gamm :rtype: float | None """ - if 'weight' not in G.es: - G.es['weight'] = [1.0] * G.vcount() + if not G.is_weighted(): + G.es['weight'] = [1.0] * G.ecount() if not isinstance(partition, leidenalg.RBConfigurationVertexPartition): partition = leiden_part_with_membership(G, partition) @@ -553,6 +556,8 @@ def prune_to_stable_partitions( Runs our full pruning pipeline on a singlelayer network. Returns the pruned list of stable partitions whose parameter estimates are within the provided ``gamma_start`` and ``gamma_end`` bounds. + Weighted graphs are supported via the ``weight`` edge attribute. Edge weights must be non-negative. + See https://doi.org/10.1038/s41598-022-20142-6 for more details. :param G: graph of interest @@ -575,9 +580,11 @@ def prune_to_stable_partitions( warnings.warn("The pruning pipeline has not been thoroughly tested on disconnected graphs. If you run into " "problems, consider using the largest connected component of your graph.") - if G.is_weighted() and any(x != 1.0 for x in G.es['weight']): - warnings.warn("The pruning pipeline does not fully handle weighted graphs and will proceed as though the input " - "graph is unweighted.") + if G.is_weighted(): + if any(w < 0 for w in G.es['weight']): + raise ValueError("Edge weights must be non-negative.") + if sum(G.es['weight']) <= 0: + raise ValueError("Total edge weight must be positive.") if isinstance(parts, leidenalg.RBConfigurationVertexPartition): # convert to (canonically represented) membership vectors if necessary @@ -613,6 +620,12 @@ def prune_to_multilayer_stable_partitions( parameter estimates are within the provided ``gamma_start``, ``gamma_end``, ``omega_start``, and ``omega_end`` bounds. + Weighted graphs are supported via the ``weight`` edge attribute. Edge weights must be non-negative. + + Interlayer weights are not fully supported by the underlying multilayer SBM theory (uniform interlayer coupling). + As such, they are used in the CHAMP step, but parameter estimation treats interlayer edges as unweighted. The code + will emit a warning if this is the case. + There are three network layer topology models available, all from Pamfil et al. * **"temporal"**: Interlayer edges always connect copies of a node from one layer to the next, often representing @@ -669,10 +682,16 @@ def prune_to_multilayer_stable_partitions( warnings.warn("The pruning pipeline has not been thoroughly tested on disconnected graphs. If you run into " "problems, consider using the largest connected component of your graph.") - if (G_intralayer.is_weighted() and any(x != 1.0 for x in G_intralayer.es['weight'])) or ( - G_interlayer.is_weighted() and any(x != 1.0 for x in G_interlayer.es['weight'])): - warnings.warn("The pruning pipeline does not fully handle weighted graphs and will proceed as though the input " - "graph is unweighted.") + for G in (G_intralayer, G_interlayer): + if G.is_weighted(): + if any(w < 0 for w in G.es['weight']): + raise ValueError("Edge weights must be non-negative.") + if sum(G.es['weight']) <= 0: + raise ValueError("Total edge weight must be positive.") + + if G_interlayer.is_weighted() and any(x != 1.0 for x in G_interlayer.es['weight']): + warnings.warn("Parameter estimation treats interlayer edges as unweighted and will proceed as though all " + "interlayer weights are 1.0.") if isinstance(parts, leidenalg.RBConfigurationVertexPartition): # convert to (canonically represented) membership vectors if necessary diff --git a/utilities/partition_utilities.py b/utilities/partition_utilities.py index 685094b..d74dc21 100644 --- a/utilities/partition_utilities.py +++ b/utilities/partition_utilities.py @@ -28,6 +28,18 @@ def out_degrees(G: "igraph.Graph") -> list[int]: return list(G.outdegree()) +def all_strengths(G: "igraph.Graph") -> list[float]: + return list(G.strength(weights='weight' if G.is_weighted() else None)) + + +def in_strengths(G: "igraph.Graph") -> list[float]: + return list(G.strength(mode='in', weights='weight' if G.is_weighted() else None)) + + +def out_strengths(G: "igraph.Graph") -> list[float]: + return list(G.strength(mode='out', weights='weight' if G.is_weighted() else None)) + + def membership_to_communities(membership: IntSequence) -> dict[int, list[int]]: communities: defaultdict[int, list[int]] = defaultdict(list) for v, c in enumerate(membership): diff --git a/utilities/types.py b/utilities/types.py index 7f6f287..91d8e16 100644 --- a/utilities/types.py +++ b/utilities/types.py @@ -41,8 +41,8 @@ DomainWithEstimates = tuple[DomainVertices, Membership, GammaEstimateValue, float] # (A_hats, P_hats) and (A_hats, P_hats, C_hats) coefficients from CHAMP -PartitionCoefficients2D = tuple[IntArray, FloatArray] -PartitionCoefficients3D = tuple[IntArray, FloatArray, IntArray] +PartitionCoefficients2D = tuple[FloatArray, FloatArray] +PartitionCoefficients3D = tuple[FloatArray, FloatArray, FloatArray] # SBM parameter estimates SinglelayerSBMParameters = tuple[float, float]