Add parallel percolation centrality

_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.",

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)

nx_parallel/algorithms/centrality/__init__.py

@@ -1,2 +1,3 @@
 from .betweenness import *
 from .harmonic import *
+from .percolation import *

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

nx_parallel/interface.py

@@ -27,6 +27,7 @@
     "betweenness_centrality",
     "edge_betweenness_centrality",
     "harmonic_centrality",
+    "percolation_centrality",
     # Components : attracting
     "number_attracting_components",
     # Components : connected