Python案例如何用Networkx做流问题

wen python案例 2026-07-16 245

本文目录导读：

安装和导入NetworkX
最大流问题案例
最小费用流问题案例
实际应用案例
高级功能
可视化辅助函数

我来为你详细介绍如何使用NetworkX解决各种流问题,包括最大流、最小费用流等经典案例。

安装和导入NetworkX

import networkx as nx
import matplotlib.pyplot as plt
import numpy as np
# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

最大流问题案例

案例1：基础网络最大流

# 创建有向图
G = nx.DiGraph()
# 添加边（源点，终点，容量）
edges = [
    ('S', 'A', 10),   # S->A 容量10
    ('S', 'B', 5),    # S->B 容量5
    ('A', 'B', 15),   # A->B 容量15
    ('A', 'C', 10),   # A->C 容量10
    ('A', 'D', 5),    # A->D 容量5
    ('B', 'C', 5),    # B->C 容量5
    ('B', 'D', 10),   # B->D 容量10
    ('C', 'T', 10),   # C->T 容量10
    ('D', 'T', 10)    # D->T 容量10
]
# 添加边的容量属性
for u, v, capacity in edges:
    G.add_edge(u, v, capacity=capacity)
# 计算最大流
flow_value, flow_dict = nx.maximum_flow(G, 'S', 'T')
print(f"最大流值: {flow_value}")
print("\n各边的流量分配:")
for u in flow_dict:
    for v, flow in flow_dict[u].items():
        capacity = G[u][v]['capacity']
        print(f"{u} -> {v}: 流量 = {flow}, 容量 = {capacity}")
# 可视化网络
def draw_flow_network(G, flow_dict):
    pos = nx.spring_layout(G, seed=42)
    # 绘制节点
    nx.draw_networkx_nodes(G, pos, node_color='lightblue', 
                          node_size=500)
    nx.draw_networkx_labels(G, pos)
    # 绘制边
    edge_labels = {}
    for u, v, data in G.edges(data=True):
        capacity = data['capacity']
        flow = flow_dict[u][v]
        edge_labels[(u, v)] = f"{flow}/{capacity}"
        # 根据利用率改变颜色
        utilization = flow / capacity if capacity > 0 else 0
        if utilization > 0.8:
            color = 'red'
        elif utilization > 0.5:
            color = 'orange'
        else:
            color = 'green'
        nx.draw_networkx_edges(G, pos, edgelist=[(u, v)], 
                              edge_color=color, width=2)
    nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels)
    plt.title(f"最大流网络 (总流量 = {sum(flow_dict['S'].values())})")
    plt.show()
draw_flow_network(G, flow_dict)

案例2：多源多汇最大流

def solve_multi_source_sink_flow():
    """解决多源多汇问题"""
    G = nx.DiGraph()
    # 原始网络
    edges = [
        ('A', 'C', 10), ('A', 'D', 8),
        ('B', 'C', 5), ('B', 'D', 12),
        ('C', 'E', 9), ('C', 'F', 7),
        ('D', 'E', 6), ('D', 'F', 8),
        ('E', 'G', 12), ('F', 'G', 10)
    ]
    # 添加超源点和超汇点
    G.add_edge('SuperSource', 'A', capacity=100)
    G.add_edge('SuperSource', 'B', capacity=100)
    G.add_edge('E', 'SuperSink', capacity=100)
    G.add_edge('F', 'SuperSink', capacity=100)
    # 添加原始边
    for u, v, cap in edges:
        G.add_edge(u, v, capacity=cap)
    # 计算最大流
    flow_value, flow_dict = nx.maximum_flow(G, 'SuperSource', 'SuperSink')
    print(f"多源多汇最大流值: {flow_value}")
    # 只显示原始网络的流量
    print("\n原始网络流量分配:")
    for u, v, cap in edges:
        if v in flow_dict.get(u, {}):
            print(f"{u} -> {v}: 流量 = {flow_dict[u][v]}, 容量 = {cap}")
    return flow_value, flow_dict
solve_multi_source_sink_flow()

最小费用流问题案例

案例3：运输问题

def solve_transportation_problem():
    """解决运输问题（最小费用流）"""
    G = nx.DiGraph()
    # 工厂（供应节点）
    factories = {
        'F1': 40,  # 供应量
        'F2': 50   # 供应量
    }
    # 仓库（需求节点）
    warehouses = {
        'W1': 30,  # 需求量
        'W2': 35,  # 需求量
        'W3': 25   # 需求量
    }
    # 运输成本（工厂->仓库）
    costs = {
        ('F1', 'W1'): 3,
        ('F1', 'W2'): 5,
        ('F1', 'W3'): 7,
        ('F2', 'W1'): 4,
        ('F2', 'W2'): 6,
        ('F2', 'W3'): 8
    }
    # 添加超源点和超汇点
    G.add_edge('Source', 'F1', capacity=factories['F1'], weight=0)
    G.add_edge('Source', 'F2', capacity=factories['F2'], weight=0)
    G.add_edge('W1', 'Sink', capacity=warehouses['W1'], weight=0)
    G.add_edge('W2', 'Sink', capacity=warehouses['W2'], weight=0)
    G.add_edge('W3', 'Sink', capacity=warehouses['W3'], weight=0)
    # 添加运输边
    for (f, w), cost in costs.items():
        min_capacity = min(factories[f], warehouses[w])
        G.add_edge(f, w, capacity=min_capacity, weight=cost)
    # 计算最小费用流
    flow_cost, flow_dict = nx.network_simplex(G)
    # 计算实际流量
    total_flow = sum(flow_dict['Source'].values())
    print(f"最小运输费用: {flow_cost}")
    print(f"总运输量: {total_flow}")
    print("\n运输方案:")
    for (f, w), cost in costs.items():
        flow = flow_dict.get(f, {}).get(w, 0)
        if flow > 0:
            print(f"{f} -> {w}: {flow}单位, 成本={cost}/单位, 总成本={flow*cost}")
    return flow_cost, flow_dict
solve_transportation_problem()

实际应用案例

案例4：通信网络带宽分配

def solve_bandwidth_allocation():
    """通信网络带宽分配问题"""
    G = nx.DiGraph()
    # 网络拓扑（路由器之间的可用带宽）
    links = [
        ('R1', 'R2', 100),   # R1->R2 100Mbps
        ('R1', 'R3', 50),    # R1->R3 50Mbps
        ('R2', 'R3', 80),    # R2->R3 80Mbps
        ('R2', 'R4', 70),    # R2->R4 70Mbps
        ('R3', 'R4', 60),    # R3->R4 60Mbps
        ('R3', 'R5', 90),    # R3->R5 90Mbps
        ('R4', 'R5', 50)     # R4->R5 50Mbps
    ]
    # 添加边（带宽作为容量）
    for u, v, bandwidth in links:
        G.add_edge(u, v, capacity=bandwidth)
        G.add_edge(v, u, capacity=bandwidth)  # 双向链路
    # 假设服务器在R1，客户端在R5
    # 需要分配带宽给多个应用
    apps = ['视频流', '文件传输', '远程桌面']
    demands = {app: {'min': 20, 'max': 40} for app in apps}
    # 计算最大有效带宽
    flow_value, flow_dict = nx.maximum_flow(G, 'R1', 'R5')
    print(f"R1到R5的最大可用带宽: {flow_value} Mbps")
    print("\n带宽分配建议:")
    total_bandwidth = flow_value
    total_demand_min = sum(d['min'] for d in demands.values())
    total_demand_max = sum(d['max'] for d in demands.values())
    for app, demand in demands.items():
        if total_demand_max <= total_bandwidth:
            allocated = demand['max']
        elif total_demand_min <= total_bandwidth:
            # 按比例分配
            proportion = demand['min'] / total_demand_min
            allocated = min(demand['max'], 
                         demand['min'] + proportion * (total_bandwidth - total_demand_min))
        else:
            allocated = demand['min'] * (total_bandwidth / total_demand_min)
        print(f"  {app}: {allocated:.1f} Mbps")
    return flow_value, flow_dict
solve_bandwidth_allocation()

案例5：排水系统网络

def solve_drainage_system():
    """城市排水系统容量分析"""
    G = nx.DiGraph()
    # 排水管道（节点，节点，最大容量m³/s）
    pipes = [
        ('入口1', '分流1', 50),
        ('入口2', '分流1', 30),
        ('入口3', '分流2', 40),
        ('分流1', '分流2', 20),
        ('分流1', '管道1', 60),
        ('分流2', '管道1', 30),
        ('分流2', '管道2', 50),
        ('管道1', '处理厂1', 70),
        ('管道2', '处理厂2', 40),
        ('处理厂1', '出口', 80),
        ('处理厂2', '出口', 60)
    ]
    for u, v, capacity in pipes:
        G.add_edge(u, v, capacity=capacity)
    # 计算从入口到出口的最大排水能力
    flow_value, flow_dict = nx.maximum_flow(G, '入口1', '出口')
    print("排水系统容量分析:")
    print(f"最大排水能力: {flow_value} m³/s")
    # 识别瓶颈
    print("\n瓶颈分析（高利用率管道）:")
    for u, v, data in G.edges(data=True):
        if v in flow_dict.get(u, {}):
            utilization = flow_dict[u][v] / data['capacity']
            if utilization > 0.8:
                print(f"  {u} -> {v}: 利用率 {utilization:.1%} [瓶颈]")
            elif utilization > 0.5:
                print(f"  {u} -> {v}: 利用率 {utilization:.1%}")
    return flow_value, flow_dict
solve_drainage_system()

高级功能

多重流分析

def multiple_flow_analysis():
    """分析不同源汇组合的流量"""
    G = nx.DiGraph()
    # 构建一个中型网络
    nodes = ['A', 'B', 'C', 'D', 'E', 'F']
    edges = [
        ('A', 'B', 20), ('A', 'C', 15),
        ('B', 'C', 10), ('B', 'D', 25),
        ('C', 'D', 15), ('C', 'E', 20),
        ('D', 'E', 10), ('D', 'F', 20),
        ('E', 'F', 25)
    ]
    for u, v, cap in edges:
        G.add_edge(u, v, capacity=cap)
        G.add_edge(v, u, capacity=cap)  # 双向
    # 分析不同的源汇对
    source_sink_pairs = [('A', 'F'), ('A', 'E'), ('B', 'F'), ('C', 'F')]
    print("不同源汇对的最大流:")
    for source, sink in source_sink_pairs:
        flow_value = nx.maximum_flow_value(G, source, sink)
        print(f"  {source} -> {sink}: {flow_value}")
    # 找到所有节点对之间的最大流
    print("\n容量矩阵（源\\汇）:")
    nodes_sorted = sorted(nodes)
    print(f"{'':>6}", end="")
    for sink in nodes_sorted:
        print(f"{sink:>6}", end="")
    print()
    for source in nodes_sorted:
        print(f"{source:>6}", end="")
        for sink in nodes_sorted:
            if source != sink:
                flow = nx.maximum_flow_value(G, source, sink)
                print(f"{flow:>6}", end="")
            else:
                print(f"{' -':>6}", end="")
        print()
multiple_flow_analysis()

可视化辅助函数

def visualize_flow_with_animation(G, flow_dict, pos=None):
    """动态可视化流量分配"""
    if pos is None:
        pos = nx.spring_layout(G, seed=42)
    fig, axes = plt.subplots(1, 2, figsize=(15, 6))
    # 原始网络
    ax1 = axes[0]
    nx.draw_networkx_nodes(G, pos, ax=ax1, node_color='lightblue', 
                          node_size=500)
    nx.draw_networkx_labels(G, pos, ax=ax1)
    edge_labels_orig = {(u, v): f"cap={d['capacity']}" 
                       for u, v, d in G.edges(data=True)}
    nx.draw_networkx_edges(G, pos, ax=ax1)
    nx.draw_networkx_edge_labels(G, pos, edge_labels_orig, ax=ax1)
    ax1.set_title("原始网络（容量）")
    # 流量网络
    ax2 = axes[1]
    nx.draw_networkx_nodes(G, pos, ax=ax2, node_color='lightgreen', 
                          node_size=500)
    nx.draw_networkx_labels(G, pos, ax=ax2)
    # 根据流量着色
    colors = []
    edge_labels_flow = {}
    for u, v, d in G.edges(data=True):
        flow = flow_dict.get(u, {}).get(v, 0)
        utilization = flow / d['capacity'] if d['capacity'] > 0 else 0
        if utilization > 0.8:
            colors.append('red')
        elif utilization > 0.5:
            colors.append('orange')
        else:
            colors.append('green')
        edge_labels_flow[(u, v)] = f"{flow}/{d['capacity']}"
    nx.draw_networkx_edges(G, pos, ax=ax2, edge_color=colors, width=3)
    nx.draw_networkx_edge_labels(G, pos, edge_labels_flow, ax=ax2)
    ax2.set_title("流量分配（流量/容量）")
    plt.tight_layout()
    plt.show()
# 使用示例
G = nx.DiGraph()
edges = [('S', 'A', 10), ('S', 'B', 5), ('A', 'B', 15), 
         ('A', 'T', 10), ('B', 'T', 10)]
for u, v, cap in edges:
    G.add_edge(u, v, capacity=cap)
flow_value, flow_dict = nx.maximum_flow(G, 'S', 'T')
visualize_flow_with_animation(G, flow_dict)

这些案例展示了NetworkX在处理流问题时的强大功能,包括：