python的先进制造技术工业场景模拟第八篇:使用Networkx搭建简易车间拓扑图,节点代表设备,连线代表物料输送路径。
周四上午,车间工艺办公室。
"新来的工艺员小王,对着一张 Excel 表格发呆,"生产主管老李指着屏幕说,"我让他画一张车间设备布局图,标注物料从仓库到各机床的流转路径。他画了三天,还是画不明白。"
我凑过去看了一眼那张 Excel——A 列是设备名称,B 列是上游设备,C 列是下游设备,D 列是距离,E 列是运输时间。几百行,密密麻麻。
"这不是表格能表达的东西,"我说,"物料流转是'谁连着谁'的关系——这是图(Graph),不是表(Table)。"
老李点头:"对。我们车间有 12 台加工中心、4 台机器人、2 条传送带、1 个立体仓库。物料从仓库出来,经过传送带送到机器人上料位,机器人装夹到加工中心,加工完再经另一条传送带到检测工位……这中间路径怎么走、哪条路径堵了、哪台设备是瓶颈,我想看一张图就明白。"
"那你需要的不是 CAD 图纸,"我打开 VS Code,"你需要一个'车间拓扑图'——节点是设备,连线是物料输送路径,连线上标注距离或运输时间。这样你一眼就能看出:物料从 A 到 B 有几种走法、哪条路径最短、如果某台设备坏了,哪些路径会断。"
我敲了几行代码:
import networkx as nx
import matplotlib.pyplot as plt
# 创建有向图
G = nx.DiGraph()
# 添加设备节点
machines = ["仓库", "传送带1", "机器人A", "加工中心1", "加工中心2", "检测台"]
G.add_nodes_from(machines)
# 添加物料路径(边)
edges = [
("仓库", "传送带1", {"distance": 15, "time": 30}),
("传送带1", "机器人A", {"distance": 5, "time": 10}),
("机器人A", "加工中心1", {"distance": 3, "time": 8}),
("机器人A", "加工中心2", {"distance": 3, "time": 8}),
("加工中心1", "检测台", {"distance": 20, "time": 40}),
("加工中心2", "检测台", {"distance": 20, "time": 40}),
]
G.add_edges_from(edges)
# 绘制拓扑图
pos = nx.spring_layout(G, seed=42)
nx.draw(G, pos, with_labels=True, node_color="lightblue",
node_size=2000, font_size=10, arrowsize=20)
edge_labels = {(u, v): f"{d['distance']}m" for u, v, d in G.edges(data=True)}
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels)
plt.title("车间物料流转拓扑图")
plt.show()
"这只是核心逻辑,"我说,"完整版会用 OOP 封装:一个类管设备节点定义,一个类管路径边定义,一个类管拓扑构建,一个类管路径分析(最短路径、连通性、瓶颈识别),一个类管可视化。数据自包含——用 numpy 合成一份车间拓扑 CSV,读者下载就能跑。"
老李凑近屏幕:"所以你这东西,就是把 Excel 里的'谁连着谁'变成一张图——节点是设备,线是输送路径,线上标距离。哪条路最短、哪台设备坏了路就断了,一眼就看到了。"
"对。而且还能做路径分析,"我补充,"比如从仓库到检测台,有几条路可以走?哪条最短?如果传送带1坏了,物料还能不能到加工中心?用 networkx 的 shortest_path 和 has_path 一行代码就搞定。车间物流规划的第一步,就是先把设备连接关系画成一张看得懂的拓扑图。"
一、实际应用场景(真实痛点)
场景设定:中小型机加工车间拥有多台加工中心、机器人、传送带、立体仓库等硬件设备,物料在各设备之间按工艺路线流转。车间管理者需要了解设备之间的连接关系、物料输送路径和潜在的物流瓶颈。现场常见痛点:设备连接关系用 Excel 表格记录,无法直观展示"谁连着谁";新增设备或调整工艺路线时,需要手动更新多张表格;无法快速判断某条输送路径是否可达、是否存在单点故障。
现场原话(叙事化):
"我不是不想管物流,"老李说,"设备之间的连接关系全在系统里,但我打开就是一张表——A 连 B,B 连 C,C 连 D……这有什么用?我需要知道的是:从仓库到检测台有几种走法、哪条路径最短、如果传送带 1 坏了,物料还能不能到加工中心。这些系统给不了我。"
核心矛盾:"表格化的设备连接关系"与"管理者需要的可视化拓扑与路径分析"之间的表达断层。需要一个"车间拓扑图构建与路径分析程序",用
"networkx" 构建图模型,用
"matplotlib" 绘制拓扑图,用
"numpy" 计算路径权重,实现从设备连接表到车间物流拓扑洞察的转化。
二、痛点分析(映射到滨州职业学院《先进制造技术》课程模型)
《先进制造技术》模块 本篇痛点对应
先进制造技术基础:先进制造系统的系统构成 系统拓扑建模:将物理设备抽象为节点、物料路径抽象为边,构建制造系统的图论模型。
柔性制造系统FMS与先进生产管理:FMS 的物料输送系统 物流路径规划:基于拓扑图分析物料输送路径,识别瓶颈和冗余。
智能制造与数字孪生:车间数字孪生的空间拓扑 拓扑可视化:用图论方法表达设备间的物理/逻辑连接,支撑数字孪生中的物流仿真。
先进制造新模式:可重构制造系统 拓扑可重构:设备增减时,拓扑图能快速更新,反映新的物流路径。
一句话总结:我们需要构建一个"车间拓扑图构建与路径分析程序",用
"networkx" 创建设备节点和物料路径边,用
"matplotlib" 绘制拓扑图,用
"numpy" 计算路径权重,实现从连接关系到物流拓扑的完整工程链路。
三、核心逻辑讲解(大白话)
3.1 问题本质:把车间设备想象成"地铁线路图"
把车间拓扑图想象成"一张地铁线路图":
* 设备 = 地铁站:每台加工中心、每个机器人、每条传送带都是一个"站"。
* 物料路径 = 地铁线路:连接两个站的轨道,上面标注了距离(站间距)和运输时间(运行时间)。
* 拓扑图 = 地铁网络图:把所有站和线路画在一张图上,你一眼就能看出从起点到终点怎么换乘、哪条线路最短。
* 路径分析 = 导航 App:输入"从仓库到检测台",程序自动算出最短路径——就像高德地图告诉你"坐 1 号线转 2 号线最快"。
工业应用:
* networkx 建图:
"G = nx.DiGraph()" 创建有向图,
"add_nodes_from()" 添加设备节点,
"add_edges_from()" 添加带权重的边(距离、时间)。
* 最短路径:
"nx.shortest_path(G, source="仓库", target="检测台", weight="distance")" 一行代码算出最短路径。
* 连通性检查:
"nx.has_path(G, "仓库", "加工中心2")" 判断两个设备之间是否有可达路径。
* 瓶颈识别:找出"如果去掉这条边,图就不连通"的桥接边——对应车间里"如果传送带坏了,整条线就断了"的关键路径。
3.2 业务逻辑 → 代码映射
定义车间拓扑数据模型
│
▼ NodeManager (networkx)
设备节点管理:
节点类型:仓库、传送带、机器人、加工中心、检测台、AGV
节点属性:名称、类型、坐标(x,y)、加工能力
│
▼ EdgeManager (networkx)
路径边管理:
边属性:距离(m)、运输时间(s)、最大流量(件/h)、当前负载
│
▼ TopologyBuilder (networkx)
拓扑构建:
创建有向图 nx.DiGraph()
添加节点和边
设置边权重
│
▼ PathAnalyzer (networkx/numpy)
路径分析:
最短路径(Dijkstra 算法)
连通性检查
关键路径/桥接边识别
│
▼ TopologyVisualizer (matplotlib/networkx)
可视化:
1. 车间拓扑图(节点+边+标签)
2. 最短路径高亮图
3. 节点类型着色图
│
▼ SyntheticDataGenerator (numpy)
合成数据生成:
用 numpy 生成设备坐标和连接关系
模拟不同车间的拓扑结构
确保数据自包含、可复现
3.3 为什么用 networkx 而不是手动画流程图?
* 问题:12 台设备、20 条路径,手动在 CAD 或 PPT 里画线——每次调整都要重画,而且算不出最短路径。
* 处理策略:
"networkx" 用代码定义拓扑,调整设备只需改一行数据,路径分析自动更新。
* 工程合理性:车间布局经常调整(新设备进场、旧设备淘汰),图模型可以版本化管理,随时重新计算。
3.4 分析前后对比
维度 Excel 表格 Networkx 拓扑图
可读性 行列表格,看不出连接关系 一张图,节点和连线一目了然
路径查询 手动追踪,容易遗漏
"shortest_path" 自动计算
连通性检查 无法判断
"has_path" 一行搞定
瓶颈识别 靠经验猜 桥接边算法自动识别
可维护性 改布局要改多张表 改数据自动重绘
四、OOP 代码实现
4.1 项目结构
workshop_topology_analyzer/
├── workshop_topology_analyzer/
│ ├── __init__.py
│ ├── node_manager.py # 设备节点管理
│ ├── edge_manager.py # 路径边管理
│ ├── topology_builder.py # 拓扑构建
│ ├── path_analyzer.py # 路径分析
│ ├── visualizer.py # 拓扑可视化
│ └── synthetic_data.py # 合成数据生成
├── tests/
│ ├── __init__.py
│ └── test_topology_analyzer.py # 单元测试
├── results/
│ ├── topology_basic.png # 基础拓扑图
│ ├── shortest_path.png # 最短路径高亮图
│ ├── topology_by_type.png # 按节点类型着色图
│ ├── topology_data.csv # 拓扑数据
│ └── topology_report.txt # 拓扑分析报告
└── run_topology.py # 主程序入口
4.2 核心源码
<details>
<summary></summary>
"""设备节点管理器"""
import networkx as nx
from typing import Dict, List, Optional, Tuple
class NodeManager:
"""
车间设备节点管理器
管理拓扑图中的设备节点及其属性。
"""
# 节点类型常量
TYPE_WAREHOUSE = "仓库"
TYPE_CONVEYOR = "传送带"
TYPE_ROBOT = "机器人"
TYPE_MACHINE = "加工中心"
TYPE_INSPECT = "检测台"
TYPE_AGV = "AGV"
VALID_TYPES = {
TYPE_WAREHOUSE, TYPE_CONVEYOR, TYPE_ROBOT,
TYPE_MACHINE, TYPE_INSPECT, TYPE_AGV,
}
def __init__(self):
self._nodes: Dict[str, Dict] = {}
def add_node(self,
name: str,
node_type: str,
pos: Optional[Tuple[float, float]] = None,
capacity: Optional[float] = None,
description: str = "") -> None:
"""
添加设备节点
Parameters
----------
name : str
设备名称(唯一标识)
node_type : str
设备类型
pos : tuple, optional
(x, y) 坐标
capacity : float, optional
加工能力或吞吐量
description : str
描述
"""
if node_type not in self.VALID_TYPES:
raise ValueError(f"无效的节点类型: {node_type}")
self._nodes[name] = {
"name": name,
"type": node_type,
"pos": pos or (0.0, 0.0),
"capacity": capacity,
"description": description,
}
def add_nodes_from_list(self, node_list: List[Dict]) -> None:
"""从列表批量添加节点"""
for node in node_list:
self.add_node(**node)
def get_node(self, name: str) -> Optional[Dict]:
return self._nodes.get(name)
def get_all_nodes(self) -> Dict[str, Dict]:
return self._nodes.copy()
def get_nodes_by_type(self, node_type: str) -> List[str]:
"""获取指定类型的所有节点名称"""
return [
name for name, attr in self._nodes.items()
if attr["type"] == node_type
]
def node_count(self) -> int:
return len(self._nodes)
def node_types_summary(self) -> Dict[str, int]:
"""统计各类型节点数量"""
summary = {}
for attr in self._nodes.values():
t = attr["type"]
summary[t] = summary.get(t, 0) + 1
return summary
</details>
<details>
<summary></summary>
"""路径边管理器"""
from typing import Dict, List, Optional, Tuple
class EdgeManager:
"""
车间物料路径边管理器
管理拓扑图中设备之间的连接边及其权重属性。
"""
def __init__(self):
self._edges: List[Tuple[str, str, Dict]] = []
def add_edge(self,
source: str,
target: str,
distance: float = 0.0,
transport_time: float = 0.0,
max_flow: float = 0.0,
description: str = "") -> None:
"""
添加物料路径边
Parameters
----------
source : str
源设备名称
target : str
目标设备名称
distance : float
距离(米)
transport_time : float
运输时间(秒)
max_flow : float
最大流量(件/小时)
description : str
描述
"""
self._edges.append((
source, target,
{
"distance": distance,
"transport_time": transport_time,
"max_flow": max_flow,
"description": description,
}
))
def add_edges_from_list(self, edge_list: List[Dict]) -> None:
"""从列表批量添加边"""
for edge in edge_list:
self.add_edge(**edge)
def get_all_edges(self) -> List[Tuple[str, str, Dict]]:
return self._edges.copy()
def edge_count(self) -> int:
return len(self._edges)
def get_edges_from(self, source: str) -> List[Tuple[str, str, Dict]]:
"""获取从指定节点出发的所有边"""
return [(s, t, d) for s, t, d in self._edges if s == source]
def get_edges_to(self, target: str) -> List[Tuple[str, str, Dict]]:
"""获取指向指定节点的所有边"""
return [(s, t, d) for s, t, d in self._edges if t == target]
</details>
<details>
<summary></summary>
"""拓扑构建器"""
import networkx as nx
from typing import Dict, List, Optional, Tuple
from .node_manager import NodeManager
from .edge_manager import EdgeManager
class TopologyBuilder:
"""
车间拓扑构建器
将节点和边组合成 networkx 图对象。
"""
def __init__(self,
node_manager: Optional[NodeManager] = None,
edge_manager: Optional[EdgeManager] = None):
self.node_manager = node_manager or NodeManager()
self.edge_manager = edge_manager or EdgeManager()
self._graph: Optional[nx.DiGraph] = None
def build(self) -> nx.DiGraph:
"""
构建拓扑图
Returns
-------
nx.DiGraph
"""
G = nx.DiGraph()
# 添加节点
for name, attrs in self.node_manager.get_all_nodes().items():
G.add_node(
name,
type=attrs["type"],
pos=attrs["pos"],
capacity=attrs["capacity"],
description=attrs["description"],
)
# 添加边
for source, target, attrs in self.edge_manager.get_all_edges():
G.add_edge(source, target, **attrs)
self._graph = G
return G
def get_graph(self) -> Optional[nx.DiGraph]:
return self._graph
def is_connected(self) -> bool:
"""判断图是否弱连通"""
if self._graph is None:
return False
return nx.is_weakly_connected(self._graph)
def get_weakly_connected_components(self) -> List:
"""获取弱连通分量"""
if self._graph is None:
return []
return list(nx.weakly_connected_components(self._graph))
def get_node_degree(self) -> Dict[str, int]:
"""获取每个节点的入度+出度"""
if self._graph is None:
return {}
return dict(self._graph.degree())
</details>
<details>
<summary></summary>
"""路径分析器"""
import networkx as nx
import numpy as np
from typing import Dict, List, Optional, Tuple
class PathAnalyzer:
"""
车间物流路径分析器
基于 networkx 图进行最短路径、连通性、瓶颈分析。
"""
def __init__(self, graph: nx.DiGraph):
self.graph = graph
def shortest_path(self,
source: str,
target: str,
weight: str = "distance") -> Optional[List[str]]:
"""
计算最短路径
Parameters
----------
source : str
起点设备
target : str
终点设备
weight : str
边权重属性名
Returns
-------
list or None
"""
try:
return nx.shortest_path(
self.graph, source=source, target=target, weight=weight
)
except (nx.NetworkXNoPath, nx.NodeNotFound):
return None
def shortest_path_length(self,
source: str,
target: str,
weight: str = "distance") -> Optional[float]:
"""计算最短路径长度"""
try:
return nx.shortest_path_length(
self.graph, source=source, target=target, weight=weight
)
except (nx.NetworkXNoPath, nx.NodeNotFound):
return None
def has_path(self, source: str, target: str) -> bool:
"""判断两节点间是否存在路径"""
try:
return nx.has_path(self.graph, source, target)
except nx.NodeNotFound:
return False
def all_shortest_paths(self,
weight: str = "distance") -> Dict[Tuple[str, str], List]:
"""计算所有节点对之间的最短路径"""
paths = {}
nodes = list(self.graph.nodes())
for i, src in enumerate(nodes):
for tgt in nodes[i + 1:]:
path = self.shortest_path(src, tgt, weight)
if path:
length = self.shortest_path_length(src, tgt, weight)
paths[(src, tgt)] = {
"path": path,
"length": length,
}
return paths
def find_bridges(self) -> List[Tuple[str, str]]:
"""
识别桥接边(去掉后图不再连通的边)
Returns
-------
list of (source, target)
"""
# 将 DiGraph 转为无向图找桥
undirected = self.graph.to_undirected()
return list(nx.bridges(undirected))
def find_articulation_points(self) -> List[str]:
"""识别关节点(去掉后图不再连通的节点)"""
undirected = self.graph.to_undirected()
return list(nx.articulation_points(undirected))
def get_node_centrality(self) -> Dict[str, float]:
"""计算节点中介中心性(衡量节点在路径中的重要性)"""
undirected = self.graph.to_undirected()
return nx.betweenness_centrality(undirected)
def analyze_bottlenecks(self) -> Dict:
"""
综合瓶颈分析
Returns
-------
dict
"""
bridges = self.find_bridges()
articulation_pts = self.find_articulation_points()
centrality = self.get_node_centrality()
# 按中心性排序
sorted_centrality = sorted(
centrality.items(), key=lambda x: x[1], reverse=True
)
return {
"bridges": bridges,
"articulation_points": articulation_pts,
"top_central_nodes": sorted_centrality[:5],
"centrality": centrality,
}
</details>
<details>
<summary></summary>
"""拓扑可视化器"""
import networkx as nx
import matplotlib.pyplot as plt
from pathlib import Path
from typing import Dict, List, Optional, Tuple
plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
class TopologyVisualizer:
"""车间拓扑可视化"""
# 节点类型颜色映射
TYPE_COLORS = {
"仓库": "#95A5A6",
"传送带": "#3498DB",
"机器人": "#E74C3C",
"加工中心": "#2ECC71",
"检测台": "#F39C12",
"AGV": "#9B59B6",
}
def __init__(self, results_dir: str = "results"):
self.results_dir = Path(results_dir)
self.results_dir.mkdir(exist_ok=True)
def plot_topology(self,
graph: nx.DiGraph,
highlight_path: Optional[List[str]] = None,
title: str = "车间设备拓扑图") -> None:
"""
绘制拓扑图
Parameters
----------
graph : nx.DiGraph
拓扑图
highlight_path : list, optional
要高亮的路径节点列表
title : str
图标题
"""
fig, ax = plt.subplots(figsize=(14, 10))
# 获取节点位置
pos = nx.get_node_attributes(graph, "pos")
if not pos or all(v == (0, 0) for v in pos.values()):
pos = nx.spring_layout(graph, seed=42, k=2, iterations=50)
# 节点颜色
node_colors = []
for node in graph.nodes():
node_type = graph.nodes[node].get("type", "未知")
node_colors.append(self.TYPE_COLORS.get(node_type, "#BDC3C7"))
# 绘制节点
nx.draw_networkx_nodes(
graph, pos, ax=ax,
node_color=node_colors,
node_size=2500,
alpha=0.9,
)
# 绘制边
nx.draw_networkx_edges(
graph, pos, ax=ax,
edge_color="#7F8C8D",
arrowsize=20,
arrowstyle="->",
width=1.5,
alpha=0.7,
)
# 高亮路径
if highlight_path and len(highlight_path) > 1:
path_edges = list(zip(highlight_path[:-1], highlight_path[1:]))
nx.draw_networkx_edges(
graph, pos, ax=ax,
edgelist=path_edges,
edge_color="#E74C3C",
arrowsize=25,
arrowstyle="->",
width=3.0,
alpha=1.0,
)
# 节点标签
nx.draw_networkx_labels(
graph, pos, ax=ax,
font_size=9,
font_weight="bold",
font_color="white",
)
# 边标签(距离)
edge_labels = {
(u, v): f"{d.get('distance', 0):.0f}m"
for u, v, d in graph.edges(data=True)
}
nx.draw_networkx_edge_labels(
graph, pos, edge_labels=edge_labels, ax=ax,
font_size=8,
)
# 图例
legend_elements = [
plt.Line2D([0], [0], marker='o', color='w',
markerfacecolor=color, markersize=12, label=ntype)
for ntype, color in self.TYPE_COLORS.items()
]
ax.legend(handles=legend_elements, loc="upper left",
bbox_to_anchor=(1.02, 1), borderaxespad=0)
ax.set_title(title, fontsize=14, fontweight="bold")
ax.axis("off")
plt.tight_layout()
plt.savefig(self.results_dir / "topology_basic.png",
dpi=150, bbox_inches="tight")
plt.close()
def plot_topology_by_type(self,
graph: nx.DiGraph,
title: str = "车间拓扑图(按类型着色)") -> None:
"""绘制按节点类型着色的拓扑图"""
self.plot_topology(graph, highlight_path=None, title=title)
# 覆盖保存
plt.figure(figsize=(14, 10))
pos = nx.get_node_attributes(graph, "pos")
if not pos or all(v == (0, 0) for v in pos.values()):
pos = nx.spring_layout(graph, seed=42, k=2, iterations=50)
node_colors = []
for node in graph.nodes():
node_type = graph.nodes[node].get("type", "未知")
node_colors.append(self.TYPE_COLORS.get(node_type, "#BDC3C7"))
nx.draw_networkx_nodes(
graph, pos, node_color=node_colors,
node_size=2500, alpha=0.9,
)
nx.draw_networkx_edges(
graph, pos, edge_color="#7F8C8D",
arrowsize=20, arrowstyle="->", width=1.5, alpha=0.7,
)
nx.draw_networkx_labels(graph, pos, font_size=9,
font_weight="bold", font_color="white")
nx.draw_networkx_edge_labels(
graph, pos,
edge_labels={(u, v): f"{d.get('distance', 0):.0f}m"
for u, v, d in graph.edges(data=True)},
font_size=8,
)
legend_elements = [
plt.Line2D([0], [0], marker='o', color='w',
markerfacecolor=color, markersize=12, label=ntype)
for ntype, color in self.TYPE_COLORS.items()
]
plt.legend(handles=legend_elements, loc="upper left",
bbox_to_anchor=(1.02, 1))
plt.title(title, fontsize=14, fontweight="bold")
plt.axis("off")
plt.tight_layout()
plt.savefig(self.results_dir / "topology_by_type.png",
dpi=150, bbox_inches="tight")
plt.close()
def plot_shortest_path(self,
graph: nx.DiGraph,
path: List[str],
title: str = "最短路径高亮") -> None:
"""绘制最短路径高亮图"""
self.plot_topology(graph, highlight_path=path, title=title)
plt.figure(figsize=(14, 10))
pos = nx.get_node_attributes(graph, "pos")
if not pos or all(v == (0, 0) for v in pos.values()):
po
利用AI解决实际问题,如果你觉得这个工具好用,欢迎关注长安牧笛!
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐


所有评论(0)