Networkx地下管网2D图转3D压力分布可视化及动画实现

需求说明 #

我有一个用Networkx构建的2D有向图，用来代表地下管网——每条边对应一根独立管道，且每条边都带有一个属性：存储长度-压力分布的NumPy数组，数据格式如下：

我需要把所有管道的内部压力分布和Networkx原图在同一图形中可视化，理想输出是3D图形：x、y轴对应Networkx图的布局位置，每条边的z轴展示其内部压力分布；如果能实现压力分布动态变化的动画效果就更好了。

以下是带边属性的网络图示例代码：

import numpy as np
import pandas as pd
import networkx as nx
import matplotlib.pyplot as plt

Sections = np.array([0, 1, 2, 3, 4])
Sources = np.array(["A", "B", "B", "D", "E"])
Targets = np.array(["B", "C", "D", "E", "F"])
Sections_L = np.array([250, 250, 100, 250, 250])
Sections_a = np.full(len(Sections), 1000)
Sections_D = np.array([0.173, 0.173, 0.1, 0.173, 0.173])
Sections_f = np.array([0.047, 0.047, 0.057, 0.047, 0.047])
Sections_U = np.array([0.85, 0.85, 2.55, 0.85, 0.85])
Sections_A = (np.pi * Sections_D ** 2) / 4
Sections_Q0 = Sections_A * Sections_U
Sections_S = np.zeros(len(Sections))
Data = np.column_stack(
    (Sections_L, Sections_D, Sections_A, Sections_a, Sections_f, Sections_U, Sections_Q0, Sections_S, Sources,
     Targets))
PipePD = pd.DataFrame(data=Data, index=Sections,
                      columns=["L", "D", "A", "a", "f", "U", "Q0", "isSource", "sources", "targets"])
G = nx.from_pandas_edgelist(PipePD, source="sources", target="targets", edge_attr=True, create_using=nx.DiGraph())

for edge in G.edges:
    length = int(G.edges[edge]["L"])
    lengthDist = range(0, length, 1)
    dist = np.random.uniform(0,10, len(lengthDist))
    PressureDist = np.column_stack((lengthDist, dist))
    G.edges[edge]["PressureDist"] = PressureDist

静态3D可视化实现 #

核心逻辑 #

1. 获取Networkx图的2D布局坐标，用管道长度作为权重优化布局，让管网结构更贴合实际；
2. 对每条管道，从起点到终点的x/y坐标做线性插值，插值点数量匹配压力分布的采样点数量；
3. 将插值得到的x/y序列与压力值对应，在3D轴上绘制曲线，同时保留原图节点作为位置参考。

实现代码 #

from mpl_toolkits.mplot3d import Axes3D

# 获取图布局，用管道长度做权重保证布局稳定
pos = nx.spring_layout(G, weight="L", seed=42)

# 创建3D画布
fig = plt.figure(figsize=(12,8))
ax = fig.add_subplot(111, projection='3d')

# 绘制z=0平面的节点与标签
for node, (x, y) in pos.items():
    ax.scatter(x, y, 0, color='black', s=100, zorder=10)
    ax.text(x, y, 0, node, fontsize=12, zorder=11)

# 绘制每条管道的压力分布曲线
for edge in G.edges:
    u, v = edge
    x_u, y_u = pos[u]
    x_v, y_v = pos[v]
    # 提取压力分布数据
    pressure_data = G.edges[edge]["PressureDist"]
    length_points = pressure_data[:,0]
    pressure_values = pressure_data[:,1]
    
    # 插值生成管道路径的x/y坐标
    num_points = len(length_points)
    x_coords = np.linspace(x_u, x_v, num_points)
    y_coords = np.linspace(y_u, y_v, num_points)
    
    # 绘制3D压力曲线
    ax.plot3D(x_coords, y_coords, pressure_values, linewidth=2, alpha=0.8)

# 设置坐标轴与标题
ax.set_xlabel('X 布局坐标')
ax.set_ylabel('Y 布局坐标')
ax.set_zlabel('压力值')
ax.set_title('地下管网3D压力分布可视化')

plt.tight_layout()
plt.show()

动态压力分布动画实现 #

核心逻辑 #

1. 基于静态3D可视化框架，用FuncAnimation实现帧更新；
2. 每帧生成随机波动的压力数据，模拟管网压力变化；
3. 清除旧曲线并重新绘制新的压力曲线，保留节点和基础布局不变。

实现代码 #

from matplotlib.animation import FuncAnimation

# 初始化3D画布与布局
fig = plt.figure(figsize=(12,8))
ax = fig.add_subplot(111, projection='3d')
pos = nx.spring_layout(G, weight="L", seed=42)

# 绘制固定的节点与标签
for node, (x, y) in pos.items():
    ax.scatter(x, y, 0, color='black', s=100, zorder=10)
    ax.text(x, y, 0, node, fontsize=12, zorder=11)

# 存储管道曲线的列表
pipe_lines = []

# 初始化函数：绘制初始压力曲线
def init():
    for edge in G.edges:
        u, v = edge
        x_u, y_u = pos[u]
        x_v, y_v = pos[v]
        pressure_data = G.edges[edge]["PressureDist"]
        num_points = len(pressure_data[:,0])
        x_coords = np.linspace(x_u, x_v, num_points)
        y_coords = np.linspace(y_u, y_v, num_points)
        line, = ax.plot3D(x_coords, y_coords, pressure_data[:,1], linewidth=2, alpha=0.8)
        pipe_lines.append(line)
    ax.set_xlabel('X 布局坐标')
    ax.set_ylabel('Y 布局坐标')
    ax.set_zlabel('压力值')
    ax.set_title('地下管网压力分布动态变化')
    return pipe_lines

# 更新函数：每帧更新压力数据并刷新曲线
def update(frame):
    for i, edge in enumerate(G.edges):
        pressure_data = G.edges[edge]["PressureDist"]
        # 生成压力波动值，限制在0-10范围内
        delta = np.random.uniform(-0.5, 0.5, len(pressure_data[:,1]))
        new_pressure = np.clip(pressure_data[:,1] + delta, 0, 10)
        # 更新边属性（可选，需保留变化数据时启用）
        G.edges[edge]["PressureDist"][:,1] = new_pressure
        
        # 更新曲线的3D数据
        u, v = edge
        x_u, y_u = pos[u]
        x_v, y_v = pos[v]
        num_points = len(new_pressure)
        x_coords = np.linspace(x_u, x_v, num_points)
        y_coords = np.linspace(y_u, y_v, num_points)
        pipe_lines[i].set_data_3d(x_coords, y_coords, new_pressure)
    return pipe_lines

# 创建动画：100帧，每帧间隔200ms
ani = FuncAnimation(fig, update, frames=100, init_func=init, interval=200, blit=True)

# 可选：保存动画为MP4文件（需安装ffmpeg）
# ani.save('pipe_pressure_animation.mp4', writer='ffmpeg', fps=5)

plt.tight_layout()
plt.show()

内容的提问来源于stack exchange，提问作者Sirui W