diff --git a/_nx_parallel/__init__.py b/_nx_parallel/__init__.py index a9fc7062..124aad01 100644 --- a/_nx_parallel/__init__.py +++ b/_nx_parallel/__init__.py @@ -241,6 +241,13 @@ def get_info(): 'get_chunks : str, function (default = "chunks")': "A function that takes in a list of weakly connected components as input and returns an iterable `component_chunks`. The default chunking is done by slicing the list of weakly connected components into `n_jobs` number of chunks." }, }, + "percolation_centrality": { + "url": "https://github.com/networkx/nx-parallel/blob/main/nx_parallel/algorithms/centrality/percolation.py#L16", + "additional_docs": "The parallel computation is implemented by dividing the source nodes into chunks and computing their percolation centrality contributions concurrently.", + "additional_parameters": { + 'get_chunks : str, function (default = "chunks")': "A function that takes in a list of all the nodes as input and returns an iterable `node_chunks`. The default chunking is done by slicing the `nodes` into `n_jobs` number of chunks." + }, + }, "preferential_attachment": { "url": "https://github.com/networkx/nx-parallel/blob/main/nx_parallel/algorithms/link_prediction.py#L133", "additional_docs": "The edge pairs are chunked into `pairs_chunks` and then the preferential attachment for all `pairs_chunks` is computed in parallel over `n_jobs` number of CPU cores.", diff --git a/benchmarks/benchmarks/bench_centrality.py b/benchmarks/benchmarks/bench_centrality.py index b281dd9e..4760f2ac 100644 --- a/benchmarks/benchmarks/bench_centrality.py +++ b/benchmarks/benchmarks/bench_centrality.py @@ -34,3 +34,15 @@ def setup(self, backend, num_nodes, edge_prob): def time_harmonic_centrality(self, backend, num_nodes, edge_prob): _ = nx.harmonic_centrality(self.G, backend=backend) + + +class Percolation(Benchmark): + params = [backends, num_nodes, edge_prob] + param_names = ["backend", "num_nodes", "edge_prob"] + + def setup(self, backend, num_nodes, edge_prob): + self.G = get_cached_gnp_random_graph(num_nodes, edge_prob) + self.states = {node: (i % 9 + 1) / 10 for i, node in enumerate(self.G)} + + def time_percolation_centrality(self, backend, num_nodes, edge_prob): + _ = nx.percolation_centrality(self.G, states=self.states, backend=backend) diff --git a/nx_parallel/algorithms/centrality/__init__.py b/nx_parallel/algorithms/centrality/__init__.py index d4536580..af3c04c6 100644 --- a/nx_parallel/algorithms/centrality/__init__.py +++ b/nx_parallel/algorithms/centrality/__init__.py @@ -1,2 +1,3 @@ from .betweenness import * from .harmonic import * +from .percolation import * diff --git a/nx_parallel/algorithms/centrality/percolation.py b/nx_parallel/algorithms/centrality/percolation.py new file mode 100644 index 00000000..3dc706d4 --- /dev/null +++ b/nx_parallel/algorithms/centrality/percolation.py @@ -0,0 +1,104 @@ +from joblib import Parallel, delayed +import networkx as nx +from networkx.algorithms.centrality.betweenness import ( + _single_source_dijkstra_path_basic, + _single_source_shortest_path_basic, +) +from networkx.algorithms.centrality.percolation import _accumulate_percolation + +import nx_parallel as nxp + + +__all__ = ["percolation_centrality"] + + +@nxp._configure_if_nx_active() +def percolation_centrality( + G, + attribute="percolation", + states=None, + weight=None, + get_chunks="chunks", +): + """The parallel computation is implemented by dividing the source nodes into + chunks and computing their percolation centrality contributions concurrently. + + networkx.percolation_centrality : + https://networkx.org/documentation/stable/reference/algorithms/generated/networkx.algorithms.centrality.percolation_centrality.html + + Parameters + ---------- + get_chunks : str, function (default = "chunks") + A function that takes in a list of all the nodes as input and returns an + iterable `node_chunks`. The default chunking is done by slicing the + `nodes` into `n_jobs` number of chunks. + """ + if hasattr(G, "graph_object"): + G = G.graph_object + + if not G: + return {} + + nodes = G.nodes + + if states is None: + states = nx.get_node_attributes(G, attribute, default=1) + + p_sigma_x_t = sum(states.values()) + + n_jobs = nxp.get_n_jobs() + + if get_chunks == "chunks": + node_chunks = nxp.create_iterables(G, "node", n_jobs, nodes) + else: + node_chunks = get_chunks(nodes) + + partial_percolation = Parallel()( + delayed(_percolation_centrality_node_subset)( + G, + chunk, + states, + p_sigma_x_t, + weight, + ) + for chunk in node_chunks + ) + + percolation = partial_percolation[0] + for partial in partial_percolation[1:]: + for node in partial: + percolation[node] += partial[node] + + n = len(G) + for node in percolation: + percolation[node] *= 1 / (n - 2) + + return percolation + + +def _percolation_centrality_node_subset( + G, + nodes, + states, + p_sigma_x_t, + weight=None, +): + percolation = dict.fromkeys(G, 0.0) + + for s in nodes: + if weight is None: + S, P, sigma, _ = _single_source_shortest_path_basic(G, s) + else: + S, P, sigma, _ = _single_source_dijkstra_path_basic(G, s, weight) + + percolation = _accumulate_percolation( + percolation, + S, + P, + sigma, + s, + states, + p_sigma_x_t, + ) + + return percolation diff --git a/nx_parallel/interface.py b/nx_parallel/interface.py index 98ce7a1f..e3db57d9 100644 --- a/nx_parallel/interface.py +++ b/nx_parallel/interface.py @@ -27,6 +27,7 @@ "betweenness_centrality", "edge_betweenness_centrality", "harmonic_centrality", + "percolation_centrality", # Components : attracting "number_attracting_components", # Components : connected diff --git a/timing/new_heatmaps/heatmap_percolation_centrality_timing.png b/timing/new_heatmaps/heatmap_percolation_centrality_timing.png new file mode 100644 index 00000000..0dd39ffb Binary files /dev/null and b/timing/new_heatmaps/heatmap_percolation_centrality_timing.png differ