shortest_path_centrality.py

"""
最短路径与中间中心性模块
Shortest Path and Betweenness Centrality Module
"""

from collections import deque, defaultdict
from graph_storage import SocialNetwork


def floyd_warshall_all_pairs_shortest_path(network):
    """
    使用Floyd-Warshall算法计算所有节点对之间的最短路径
    Calculate all-pairs shortest paths using Floyd-Warshall algorithm
    
    Args:
        network (SocialNetwork): 社交网络对象 Social network object
        
    Returns:
        list: 二维数组，dist[i][j]表示从节点i到节点j的最短距离
              2D array where dist[i][j] represents the shortest distance from node i to node j
    """
    # 初始化距离矩阵 Initialize distance matrix
    INF = float('inf')
    dist = [[INF for _ in range(network.n)] for _ in range(network.n)]
    
    # 设置初始距离 Set initial distances
    for i in range(network.n):
        dist[i][i] = 0  # 自己到自己的距离为0 Distance from node to itself is 0
    
    # 添加直接相连的边 Add directly connected edges
    for i in range(network.n):
        neighbors = network.get_following(i)
        for j in neighbors:
            dist[i][j] = 1  # 直接相连的距离为1 Distance for directly connected nodes is 1
    
    # Floyd-Warshall算法 Floyd-Warshall algorithm
    for k in range(network.n):
        for i in range(network.n):
            for j in range(network.n):
                if dist[i][k] != INF and dist[k][j] != INF:
                    dist[i][j] = min(dist[i][j], dist[i][k] + dist[k][j])
    
    return dist


def compute_betweenness_centrality(network):
    """
    计算每个节点的介数中心性（Betweenness Centrality）
    Compute betweenness centrality for each node
    
    Args:
        network (SocialNetwork): 社交网络对象 Social network object
        
    Returns:
        dict: 每个节点的介数中心性值 Dictionary of betweenness centrality values for each node
    """
    betweenness = {i: 0.0 for i in range(network.n)}
    
    # 对每个节点计算它在多少最短路径上 For each node, calculate how many shortest paths it lies on
    for s in range(network.n):
        # 使用BFS计算从s出发的最短路径 Use BFS to calculate shortest paths from s
        stack = []
        predecessors = {i: [] for i in range(network.n)}
        shortest_paths_num = {i: 0 for i in range(network.n)}
        distance = {i: -1 for i in range(network.n)}
        
        queue = deque([s])
        distance[s] = 0
        shortest_paths_num[s] = 1
        
        while queue:
            v = queue.popleft()
            stack.append(v)
            
            # 遍历v的邻居节点 Iterate through neighbors of v
            for w in network.get_following(v):
                if distance[w] == -1:  # w未被访问过 w hasn't been visited
                    queue.append(w)
                    distance[w] = distance[v] + 1
                
                if distance[w] == distance[v] + 1:  # w在最短路径上 w is on shortest path
                    shortest_paths_num[w] += shortest_paths_num[v]
                    predecessors[w].append(v)
        
        # 计算依赖度 Calculate dependencies
        delta = {i: 0.0 for i in range(network.n)}
        
        # 从栈顶开始计算 Calculate from top of stack
        while stack:
            w = stack.pop()
            for v in predecessors[w]:
                delta[v] += (shortest_paths_num[v] / shortest_paths_num[w]) * (1 + delta[w])
            
            if w != s:
                betweenness[w] += delta[w]
    
    # 标准化介数中心性 Normalize betweenness centrality
    for node in betweenness:
        betweenness[node] /= 2  # 因为图是有向的，但我们在计算无向图的中心性 Because the graph is directed, but we're calculating undirected graph centrality
    
    return betweenness


class ShortestPathNetwork(SocialNetwork):
    """
    具有最短路径和中间中心性计算功能的社交网络类
    Social Network class with shortest path and betweenness centrality computation functionality
    """
    
    def floyd_warshall_all_pairs_shortest_path(self):
        """
        使用Floyd-Warshall算法计算所有节点对之间的最短路径
        Calculate all-pairs shortest paths using Floyd-Warshall algorithm
        
        Returns:
            list: 二维数组，dist[i][j]表示从节点i到节点j的最短距离
                  2D array where dist[i][j] represents the shortest distance from node i to node j
        """
        return floyd_warshall_all_pairs_shortest_path(self)
    
    def compute_betweenness_centrality(self):
        """
        计算每个节点的介数中心性（Betweenness Centrality）
        Compute betweenness centrality for each node
        
        Returns:
            dict: 每个节点的介数中心性值 Dictionary of betweenness centrality values for each node
        """
        return compute_betweenness_centrality(self)


# 测试示例
if __name__ == "__main__":
    print("最短路径与中间中心性 - 示例运行")
    print("Shortest Path and Betweenness Centrality - Example Run")
    
    # 创建一个最短路径网络 Create a shortest path network
    network = ShortestPathNetwork(6, directed=True, storage_type="adj_list")
    
    # 添加一些边 Add some edges
    edges = [(0, 1), (1, 2), (2, 3), (0, 4), (4, 5), (5, 3), (1, 3)]
    for u, v in edges:
        network.add_edge(u, v)
    
    print(f"添加了 {len(edges)} 条边")
    print(f"Added {len(edges)} edges")
    
    # 计算所有节点对之间的最短路径 Calculate all-pairs shortest paths
    shortest_paths = network.floyd_warshall_all_pairs_shortest_path()
    print(f"所有节点对之间的最短路径矩阵 (前3x3):")
    print(f"All-pairs shortest path matrix (first 3x3):")
    for i in range(min(3, network.n)):
        row = []
        for j in range(min(3, network.n)):
            if shortest_paths[i][j] == float('inf'):
                row.append("INF")
            else:
                row.append(shortest_paths[i][j])
        print(f"  {row}")
    
    # 计算介数中心性 Calculate betweenness centrality
    betweenness = network.compute_betweenness_centrality()
    print(f"\n各节点的介数中心性:")
    print(f"Betweenness centrality for each node:")
    for node, centrality in sorted(betweenness.items()):
        print(f"  节点 {node}: {centrality:.4f}")
        print(f"  Node {node}: {centrality:.4f}")