Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/networkx/algorithms/centrality/reaching.py: 18%

1"""Functions for computing reaching centrality of a node or a graph.""" 

2 

3import networkx as nx 

4from networkx.utils import pairwise 

5 

6__all__ = ["global_reaching_centrality", "local_reaching_centrality"] 

7 

8 

9def _average_weight(G, path, weight=None): 

10 """Returns the average weight of an edge in a weighted path. 

11 

12 Parameters 

13 ---------- 

14 G : graph 

15 A networkx graph. 

16 

17 path: list 

18 A list of vertices that define the path. 

19 

20 weight : None or string, optional (default=None) 

21 If None, edge weights are ignored. Then the average weight of an edge 

22 is assumed to be the multiplicative inverse of the length of the path. 

23 Otherwise holds the name of the edge attribute used as weight. 

24 """ 

25 path_length = len(path) - 1 

26 if path_length <= 0: 

27 return 0 

28 if weight is None: 

29 return 1 / path_length 

30 total_weight = sum(G.edges[i, j][weight] for i, j in pairwise(path)) 

31 return total_weight / path_length 

32 

33 

34@nx._dispatchable(edge_attrs="weight") 

35def global_reaching_centrality(G, weight=None, normalized=True): 

36 """Returns the global reaching centrality of a directed graph. 

37 

38 The *global reaching centrality* of a weighted directed graph is the 

39 average over all nodes of the difference between the local reaching 

40 centrality of the node and the greatest local reaching centrality of 

41 any node in the graph [1]_. For more information on the local 

42 reaching centrality, see :func:`local_reaching_centrality`. 

43 Informally, the local reaching centrality is the proportion of the 

44 graph that is reachable from the neighbors of the node. 

45 

46 Parameters 

47 ---------- 

48 G : DiGraph 

49 A networkx DiGraph. 

50 

51 weight : None or string, optional (default=None) 

52 Attribute to use for edge weights. If ``None``, each edge weight 

53 is assumed to be one. A higher weight implies a stronger 

54 connection between nodes and a *shorter* path length. 

55 

56 normalized : bool, optional (default=True) 

57 Whether to normalize the edge weights by the total sum of edge 

58 weights. 

59 

60 Returns 

61 ------- 

62 h : float 

63 The global reaching centrality of the graph. 

64 

65 Examples 

66 -------- 

67 >>> G = nx.DiGraph() 

68 >>> G.add_edge(1, 2) 

69 >>> G.add_edge(1, 3) 

70 >>> nx.global_reaching_centrality(G) 

71 1.0 

72 >>> G.add_edge(3, 2) 

73 >>> nx.global_reaching_centrality(G) 

74 0.75 

75 

76 See also 

77 -------- 

78 local_reaching_centrality 

79 

80 References 

81 ---------- 

82 .. [1] Mones, Enys, Lilla Vicsek, and Tamás Vicsek. 

83 "Hierarchy Measure for Complex Networks." 

84 *PLoS ONE* 7.3 (2012): e33799. 

85 https://doi.org/10.1371/journal.pone.0033799 

86 """ 

87 if nx.is_negatively_weighted(G, weight=weight): 

88 raise nx.NetworkXError("edge weights must be positive") 

89 total_weight = G.size(weight=weight) 

90 if total_weight <= 0: 

91 raise nx.NetworkXError("Size of G must be positive") 

92 # If provided, weights must be interpreted as connection strength 

93 # (so higher weights are more likely to be chosen). However, the 

94 # shortest path algorithms in NetworkX assume the provided "weight" 

95 # is actually a distance (so edges with higher weight are less 

96 # likely to be chosen). Therefore we need to invert the weights when 

97 # computing shortest paths. 

98 # 

99 # If weight is None, we leave it as-is so that the shortest path 

100 # algorithm can use a faster, unweighted algorithm. 

101 if weight is not None: 

102 

103 def as_distance(u, v, d): 

104 return total_weight / d.get(weight, 1) 

105 

106 shortest_paths = dict(nx.shortest_path(G, weight=as_distance)) 

107 else: 

108 shortest_paths = dict(nx.shortest_path(G)) 

109 

110 centrality = local_reaching_centrality 

111 # TODO This can be trivially parallelized. 

112 lrc = [ 

113 centrality(G, node, paths=paths, weight=weight, normalized=normalized) 

114 for node, paths in shortest_paths.items() 

115 ] 

116 

117 max_lrc = max(lrc) 

118 return sum(max_lrc - c for c in lrc) / (len(G) - 1) 

119 

120 

121@nx._dispatchable(edge_attrs="weight") 

122def local_reaching_centrality(G, v, paths=None, weight=None, normalized=True): 

123 """Returns the local reaching centrality of a node in a directed 

124 graph. 

125 

126 The *local reaching centrality* of a node in a directed graph is the 

127 proportion of other nodes reachable from that node [1]_. 

128 

129 Parameters 

130 ---------- 

131 G : DiGraph 

132 A NetworkX DiGraph. 

133 

134 v : node 

135 A node in the directed graph `G`. 

136 

137 paths : dictionary (default=None) 

138 If this is not `None` it must be a dictionary representation 

139 of single-source shortest paths, as computed by, for example, 

140 :func:`networkx.shortest_path` with source node `v`. Use this 

141 keyword argument if you intend to invoke this function many 

142 times but don't want the paths to be recomputed each time. 

143 

144 weight : None or string, optional (default=None) 

145 Attribute to use for edge weights. If `None`, each edge weight 

146 is assumed to be one. A higher weight implies a stronger 

147 connection between nodes and a *shorter* path length. 

148 

149 normalized : bool, optional (default=True) 

150 Whether to normalize the edge weights by the total sum of edge 

151 weights. 

152 

153 Returns 

154 ------- 

155 h : float 

156 The local reaching centrality of the node ``v`` in the graph 

157 ``G``. 

158 

159 Examples 

160 -------- 

161 >>> G = nx.DiGraph() 

162 >>> G.add_edges_from([(1, 2), (1, 3)]) 

163 >>> nx.local_reaching_centrality(G, 3) 

164 0.0 

165 >>> G.add_edge(3, 2) 

166 >>> nx.local_reaching_centrality(G, 3) 

167 0.5 

168 

169 See also 

170 -------- 

171 global_reaching_centrality 

172 

173 References 

174 ---------- 

175 .. [1] Mones, Enys, Lilla Vicsek, and Tamás Vicsek. 

176 "Hierarchy Measure for Complex Networks." 

177 *PLoS ONE* 7.3 (2012): e33799. 

178 https://doi.org/10.1371/journal.pone.0033799 

179 """ 

180 # Corner case: graph with single node containing a self-loop 

181 if (total_weight := G.size(weight=weight)) > 0 and len(G) == 1: 

182 raise nx.NetworkXError( 

183 "local_reaching_centrality of a single node with self-loop not well-defined" 

184 ) 

185 if paths is None: 

186 if nx.is_negatively_weighted(G, weight=weight): 

187 raise nx.NetworkXError("edge weights must be positive") 

188 if total_weight <= 0: 

189 raise nx.NetworkXError("Size of G must be positive") 

190 if weight is not None: 

191 # Interpret weights as lengths. 

192 def as_distance(u, v, d): 

193 return total_weight / d.get(weight, 1) 

194 

195 paths = nx.shortest_path(G, source=v, weight=as_distance) 

196 else: 

197 paths = nx.shortest_path(G, source=v) 

198 # If the graph is unweighted, simply return the proportion of nodes 

199 # reachable from the source node ``v``. 

200 if weight is None and G.is_directed(): 

201 return (len(paths) - 1) / (len(G) - 1) 

202 if normalized and weight is not None: 

203 norm = G.size(weight=weight) / G.size() 

204 else: 

205 norm = 1 

206 # TODO This can be trivially parallelized. 

207 avgw = (_average_weight(G, path, weight=weight) for path in paths.values()) 

208 sum_avg_weight = sum(avgw) / norm 

209 return sum_avg_weight / (len(G) - 1)