networkx-flow-networks

NetworkX 流动网络

掌握图论和网络流算法，用于对液压管路系统、多泵配置和复杂流量分配网络进行建模。

安装

pip install networkx numpy matplotlib

图基础

NetworkX 提供了用于创建和分析有向图／无向图的强大工具：

import networkx as nx
import numpy as np

# Create a directed graph for flow networks
G = nx.DiGraph()

# Add nodes (junctions, pumps, tanks)
G.add_node('Tank', type='source', elevation=30)
G.add_node('J1', type='junction', elevation=0)
G.add_node('J2', type='junction', elevation=0)
G.add_node('Outlet', type='sink', elevation=0)

# Add edges (pipes) with attributes
G.add_edge('Tank', 'J1', diameter=0.15, length=100, roughness=0.00015)
G.add_edge('J1', 'J2', diameter=0.10, length=50, roughness=0.00015)
G.add_edge('J2', 'Outlet', diameter=0.10, length=75, roughness=0.00015)

工程应用

管道网络表示

将复杂管道系统建模为有向图，其中：

• 节点 表示接头、储罐、泵、阀门
• 边 表示具有流量容量的管道
• 边属性 存储直径、长度、粗糙度、阻力

def create_piping_network():
    """Create a directed graph representing a piping network"""
    G = nx.DiGraph()

    # Add nodes with elevation data
    nodes = {
        'Reservoir': {'type': 'source', 'elevation': 50, 'pressure': 0},
        'Pump1': {'type': 'pump', 'elevation': 0},
        'J1': {'type': 'junction', 'elevation': 0},
        'J2': {'type': 'junction', 'elevation': 0},
        'Tank': {'type': 'sink', 'elevation': 20, 'demand': 100}
    }

    for node, attrs in nodes.items():
        G.add_node(node, **attrs)

    # Add pipes with hydraulic properties
    pipes = [
        ('Reservoir', 'Pump1', {'D': 0.2, 'L': 50, 'k': 0.00015}),
        ('Pump1', 'J1', {'D': 0.15, 'L': 100, 'k': 0.00015}),
        ('J1', 'J2', {'D': 0.10, 'L': 80, 'k': 0.00015}),
        ('J2', 'Tank', {'D': 0.10, 'L': 60, 'k': 0.00015})
    ]

    for u, v, attrs in pipes:
        G.add_edge(u, v, **attrs)

    return G

并联系统中的流量分配

分析具有并行路径的网络中的流量分配：

def analyze_parallel_flow(G, source, sink):
    """Analyze flow distribution in parallel pipe systems"""
    # Find all simple paths from source to sink
    all_paths = list(nx.all_simple_paths(G, source, sink))

    print(f"Number of flow paths: {len(all_paths)}")
    for i, path in enumerate(all_paths, 1):
        print(f"Path {i}: {' -> '.join(path)}")

        # Calculate total resistance for each path
        total_length = sum(G[u][v].get('L', 0) for u, v in zip(path[:-1], path[1:]))
        print(f"  Total length: {total_length} m")

    return all_paths

# Calculate resistance for Darcy-Weisbach
def pipe_resistance(D, L, f=0.02):
    """Calculate pipe resistance coefficient K = f*L/D"""
    return f * L / D

# Add resistance to edges
G = create_piping_network()
for u, v, data in G.edges(data=True):
    data['K'] = pipe_resistance(data['D'], data['L'])

路径分析

查找关键路径、最短路径，并识别瓶颈：

def path_analysis(G, source, sink):
    """Perform comprehensive path analysis"""

    # Shortest path by number of pipes
    shortest_path = nx.shortest_path(G, source, sink)
    print(f"Shortest path (by hops): {' -> '.join(shortest_path)}")

    # Shortest path by total length
    for u, v, data in G.edges(data=True):
        data['weight'] = data['L']

    shortest_length_path = nx.shortest_path(G, source, sink, weight='weight')
    total_length = nx.shortest_path_length(G, source, sink, weight='weight')
    print(f"Shortest path (by length): {' -> '.join(shortest_length_path)}")
    print(f"Total length: {total_length} m")

    # Find bottlenecks (minimum diameter in path)
    min_diameter = min(G[u][v]['D'] for u, v in zip(shortest_path[:-1], shortest_path[1:]))
    print(f"Bottleneck diameter: {min_diameter} m")

    return shortest_path, shortest_length_path

网络优化

优化流量分配并识别关键组件：

def find_critical_pipes(G):
    """Identify critical pipes whose removal would disconnect the network"""
    critical_edges = list(nx.bridges(G.to_undirected()))

    print("Critical pipes (bridges):")
    for u, v in critical_edges:
        data = G[u][v] if G.has_edge(u, v) else G[v][u]
        print(f"  {u} <-> {v}: D={data['D']} m, L={data['L']} m")

    return critical_edges

def redundancy_analysis(G, source, sink):
    """Analyze network redundancy"""
    # Node connectivity (minimum nodes to remove to disconnect)
    node_connectivity = nx.node_connectivity(G, source, sink)
    print(f"Node connectivity: {node_connectivity}")

    # Edge connectivity (minimum edges to remove to disconnect)
    edge_connectivity = nx.edge_connectivity(G, source, sink)
    print(f"Edge connectivity: {edge_connectivity}")

    if edge_connectivity > 1:
        print("Network has redundant paths")
    else:
        print("Network has no redundancy - single point of failure")

最大流／最小割

应用最大流算法进行容量分析：

def max_flow_analysis(G, source, sink):
    """Calculate maximum flow capacity using Ford-Fulkerson"""

    # Set capacity based on pipe diameter (Q ∝ D^2 for velocity limit)
    for u, v, data in G.edges(data=True):
        # Capacity proportional to cross-sectional area
        data['capacity'] = np.pi * (data['D']/2)**2 * 2.0  # Assuming 2 m/s max velocity

    # Calculate max flow
    flow_value, flow_dict = nx.maximum_flow(G, source, sink, capacity='capacity')

    print(f"Maximum flow capacity: {flow_value:.4f} m³/s")
    print("\nFlow distribution:")
    for u in flow_dict:
        for v, flow in flow_dict[u].items():
            if flow > 0:
                capacity = G[u][v]['capacity']
                utilization = flow / capacity * 100
                print(f"  {u} -> {v}: {flow:.4f} m³/s ({utilization:.1f}% capacity)")

    # Find minimum cut
    cut_value, partition = nx.minimum_cut(G, source, sink, capacity='capacity')
    reachable, non_reachable = partition

    print(f"\nMinimum cut value: {cut_value:.4f} m³/s")
    print(f"Cut set (bottleneck pipes):")
    for u, v, data in G.edges(data=True):
        if u in reachable and v in non_reachable:
            print(f"  {u} -> {v}: D={data['D']} m, capacity={data['capacity']:.4f} m³/s")

    return flow_value, flow_dict

液压网络建模

液压网络分析的完整工作流：

def hydraulic_network_model(G, source, sink, fluid_props):
    """Complete hydraulic network analysis"""

    print("=== Network Topology ===")
    print(f"Nodes: {G.number_of_nodes()}")
    print(f"Pipes: {G.number_of_edges()}")
    print(f"Average degree: {sum(dict(G.degree()).values()) / G.number_of_nodes():.2f}")

    # Check connectivity
    if not nx.is_connected(G.to_undirected()):
        print("WARNING: Network is not fully connected!")
        components = list(nx.connected_components(G.to_undirected()))
        print(f"Number of separate components: {len(components)}")

    print("\n=== Path Analysis ===")
    paths = analyze_parallel_flow(G, source, sink)

    print("\n=== Critical Components ===")
    find_critical_pipes(G)

    print("\n=== Redundancy Analysis ===")
    redundancy_analysis(G, source, sink)

    print("\n=== Max Flow Analysis ===")
    max_flow_analysis(G, source, sink)

    # Calculate pressure drops (simplified)
    print("\n=== Pressure Drop Analysis ===")
    for u, v, data in G.edges(data=True):
        rho = fluid_props['density']
        mu = fluid_props['viscosity']
        f = 0.02  # Friction factor (simplified)

        # Estimate velocity based on diameter
        V = 2.0  # m/s assumed
        dP = f * (data['L'] / data['D']) * (rho * V**2 / 2)
        data['dP'] = dP

        print(f"{u} -> {v}: ΔP = {dP/1000:.2f} kPa")

# Example usage
G = create_piping_network()
fluid_props = {'density': 1000, 'viscosity': 0.001}  # Water at 20°C
hydraulic_network_model(G, 'Reservoir', 'Tank', fluid_props)

多泵系统建模

对串联或并联的多泵系统进行建模：

def create_multipump_network():
    """Create network with parallel and series pumps"""
    G = nx.DiGraph()

    # Source tank
    G.add_node('Source', type='tank', elevation=0)

    # Parallel pump configuration
    G.add_node('Pump1A', type='pump', head=50, flow_rated=0.05)
    G.add_node('Pump1B', type='pump', head=50, flow_rated=0.05)
    G.add_node('Junction1', type='junction')

    # Series pump configuration
    G.add_node('Pump2', type='pump', head=30, flow_rated=0.08)
    G.add_node('Junction2', type='junction')

    # Destination
    G.add_node('Destination', type='tank', elevation=60)

    # Connect parallel pumps
    G.add_edge('Source', 'Pump1A', D=0.15, L=10)
    G.add_edge('Source', 'Pump1B', D=0.15, L=10)
    G.add_edge('Pump1A', 'Junction1', D=0.15, L=20)
    G.add_edge('Pump1B', 'Junction1', D=0.15, L=20)

    # Connect to series pump
    G.add_edge('Junction1', 'Pump2', D=0.15, L=50)
    G.add_edge('Pump2', 'Junction2', D=0.15, L=30)
    G.add_edge('Junction2', 'Destination', D=0.15, L=100)

    return G

def analyze_pump_configuration(G):
    """Analyze pump system configuration"""
    pumps = [n for n, d in G.nodes(data=True) if d.get('type') == 'pump']

    print(f"Total pumps in system: {len(pumps)}")
    print(f"Pump nodes: {pumps}")

    # Find pumps in series (on same path)
    # Find pumps in parallel (different paths between same nodes)

    for pump in pumps:
        in_degree = G.in_degree(pump)
        out_degree = G.out_degree(pump)
        print(f"\n{pump}:")
        print(f"  Rated head: {G.nodes[pump].get('head', 'N/A')} m")
        print(f"  Rated flow: {G.nodes[pump].get('flow_rated', 'N/A')} m³/s")
        print(f"  Connections: {in_degree} in, {out_degree} out")

最佳实践

1. 进行流量分析前，务必验证网络连通性
2. 所有边属性均使用一致的单位（建议使用 SI 单位）
3. 包含高程数据，以便分析重力驱动的流动
4. 使用有向边对止回阀进行建模
5. 对于同一接头之间的并联管道，可考虑使用 MultiDiGraph
6. 将结果存储为节点／边属性，以便可视化
7. 验证接头处的质量守恒

常见模式

# Pattern 1: Add multiple parallel pipes
G.add_edge('A', 'B', key='pipe1', D=0.1, L=100)
G.add_edge('A', 'B', key='pipe2', D=0.15, L=100)  # Use MultiDiGraph

# Pattern 2: Iterate over all pipes
for u, v, data in G.edges(data=True):
    process_pipe(u, v, data)

# Pattern 3: Find all junctions (nodes with degree > 2)
junctions = [n for n in G.nodes() if G.degree(n) > 2]

# Pattern 4: Export to GraphML for other tools
nx.write_graphml(G, 'network.graphml')

# Pattern 5: Visualize network layout
pos = nx.spring_layout(G)
# Or use hierarchical layout for better piping visualization
pos = nx.nx_agraph.graphviz_layout(G, prog='dot')

另请参阅

• reference.md - 液压系统的图算法
• examples.py - 完整的可运行示例
• NetworkX 文档，了解高级算法