车间拓扑改造仿真增删设备节点评估网络连通性与物料输送效率周三上午10点工艺工程师老杨拿着一张车间布局图在会议室白板前站了快半小时了。厂长说要提产30%让我重新规划车间布局。老杨转过身指着白板上的图现在的问题是——我们的车间网络太老了。这条产线当初是按年产5万台设计的现在要冲8万台物流路径不够用了。具体哪里不够生产总监问。你看老杨在图上画了几条红线从原材料入库到成品出库物料要经过7个节点。其中3号AGV转运站是瓶颈——所有物料都要从它那里过。如果提产30%这个节点直接堵死。那就加一条线路绕过它厂长说。加了之后整条网络的连通性会不会受影响物料输送路径会不会变长老杨摇头我需要的是先在纸上把改造方案跑一遍——删掉瓶颈节点、加几条新线路看看改造后的网络还通不通、物料从入库到出库要多久、有没有新的瓶颈出现。你需要的是车间拓扑改造仿真。我打开编辑器用NetworkX建一个有向图——节点是设备入库口、加工中心、AGV站、缓存区、出库口边是物流路径权重是输送时间或距离。然后你删几个节点、加几条边程序自动计算改造前后的网络连通性、最短路径、瓶颈节点。import networkx as nx# 1. 构建当前车间拓扑G nx.DiGraph()G.add_edge(入库, 加工A, weight5)G.add_edge(加工A, AGV3, weight3) # 瓶颈节点G.add_edge(AGV3, 加工B, weight4)G.add_edge(加工B, 出库, weight6)# 2. 模拟改造新增一条路径绕过AGV3G.add_edge(加工A, 加工B, weight8) # 新路径# 3. 评估连通性print(f网络连通: {nx.is_weakly_connected(G)})print(f最短路径: {nx.shortest_path(G, 入库, 出库, weightweight)})就这些老杨瞪大了眼睛。核心逻辑就这些。我运行了完整仿真屏幕上跳出了改造前后的网络拓扑对比图和路径分析表改造方案: 新增路径加工A→加工B移除瓶颈AGV3─────────────────────────────────────────────指标 改造前 改造后 变化网络连通性 ✅ 连通 ✅ 连通 不变最短路径长度 18 min 19 min 1 min可接受瓶颈节点 AGV3 无 ✅ 消除路径冗余度 1条 2条 ✅ 增加容错节点平均度 1.6 2.0 ✅ 网络更健壮你看我指着对比图删掉AGV3、新增加工A→加工B的直接路径后网络依然连通最短路径只多了1分钟但瓶颈消除了——而且现在从入库到出库有两条路径可选一条断了还有另一条。老杨把拓扑图打印出来贴在车间改造方案的封面上这就是我要的——改造之前先数字验证一遍确保不会改完之后更糟。那张网络对比图帮我们把车间要改造这个模糊的指令变成了删掉AGV3、加一条直接路径、连通性不变、路径冗余度翻倍的精确方案。一、实际应用场景真实痛点场景设定制造企业需要对现有车间进行拓扑改造——新增设备如增加一台加工中心、删减设备如淘汰老旧产线、调整物流路径。改造前需要评估改造后的车间网络是否仍然连通物料输送路径是否变长会不会出现新的瓶颈 如果直接施工后再发现问题停工损失巨大。现场原话叙事化我们车间有句老话动线一动全线震动。老杨说问题是厂长说加两台设备他不管加完之后物料怎么走。我得确保加完之后从原材料到成品的每一条路径都还通、都还快。那你们没有数字孪生平台吗我问。有但那个平台是给管理层看的不是给工程师算的。老杨摇头它展示3D车间模型但不能回答如果我把3号AGV站拆了最短路径会变成多少、瓶颈会转移到哪里。我需要的是What-if仿真——如果网络变成这样会怎样。所以你要的是车间拓扑改造仿真与连通性评估程序——用NetworkX构建车间物流网络模拟增删节点和边评估改造前后的连通性和输送效率。核心矛盾车间改造需要提前验证拓扑方案与现有系统只能展示静态布局缺乏动态网络分析之间的冲突。需要一个车间拓扑改造仿真与评估程序用图论方法量化改造影响。二、痛点分析映射到长安大学《智能制造导论》课程模型《智能制造导论》模块 本篇痛点对应概述车间布局与物流规划 车间拓扑设备布局决定物流效率。智能制造技术基础物料输送系统 物流网络AGV、传送带、缓存区构成的运输网络。新一代支撑技术数字孪生、图论算法 网络仿真用图模型模拟拓扑变化。智能工厂与智能生产柔性产线、可重构制造 拓扑重构根据产能需求动态调整车间布局。演进范式固定布局 → 可重构布局 → 自适应布局 从建好就不动到按需调整用仿真验证调整方案。一句话总结我们需要构建一个车间拓扑改造仿真与评估程序用NetworkX构建物流网络模拟节点增删评估连通性和输送效率。三、核心逻辑讲解大白话3.1 问题本质把车间想象成地铁线路图把车间物流网络想象成城市的地铁线路图* 设备节点 地铁站每个站入库口、加工中心、AGV站、缓存区、出库口就是一个节点。* 物流路径 地铁线路连接两个站的轨道有长度输送时间/距离。* 拓扑改造 修新线/拆旧线/建新站/关旧站* 新增设备 建新地铁站在图上加一个节点再修几条连接线。* 删减设备 关闭地铁站把节点和所有连接线从图上删掉。* 连通性 还能不能坐地铁到目的地从入库口到出库口还有没有路* 最短路径 最快坐地铁方案从A到B走哪条路最快* 瓶颈 换乘站太挤某个节点经过的路径太多流量超过容量。工业应用* 构建拓扑用NetworkX的DiGraph有向图表示车间——节点是设备边是物流路径边的权重是输送时间分钟。* 模拟改造add_node()新增设备、remove_node()删减设备、add_edge()新增路径。* 连通性评估nx.is_weakly_connected()检查全网是否连通nx.all_pairs_shortest_path_length()计算所有节点对的最短路径。* 效率评估对比改造前后的平均最短路径长度、瓶颈节点度中心性最高的节点、路径冗余度两个节点之间有多少条独立路径。3.2 业务逻辑 → 代码映射构建车间拓扑│▼ ShopFloorNode节点1. 设备ID、类型入库/加工/AGV/缓存/出库2. 处理时间在该节点的停留时间│▼ ShopFloorNetwork网络NetworkX DiGraph1. 节点设备2. 有向边物流路径权重输送时间│▼ TopologyModifier拓扑修改器1. add_node() / remove_node()2. add_edge() / remove_edge()│▼ ConnectivityEvaluator连通性评估器1. 全网连通性检查2. 节点对可达性矩阵3. 连通分量数量│▼ EfficiencyEvaluator效率评估器1. 最短路径长度Dijkstra2. 平均路径长度3. 瓶颈节点识别度中心性4. 路径冗余度│▼ Visualizer.plot()可视化1. 改造前拓扑图2. 改造后拓扑图3. 路径长度对比柱状图│▼ ReportGenerator.generate()生成报告1. 连通性对比2. 效率指标对比3. 改造建议3.3 为什么用有向图而不是列表* 问题车间物流有方向——物料从入库到出库是单向流动但AGV可能双向行驶。列表只能记录设备A旁边是设备B无法表达方向、权重和多条并行路径。* 处理策略NetworkX的有向图天然支持方向A→B和B→A可以是不同的边、权重输送时间、多路径从A到B可以有多条边。* 工程合理性图论是物流网络分析的标准方法NetworkX提供了成熟的连通性、最短路径、中心性算法无需自己实现。3.4 连通性 vs 效率* 连通性能不能到——布尔值。如果改造后从入库到出库不通了这个方案直接否决。* 效率快不快——数值。连通只是及格线效率才是关键。用最短路径长度衡量。四、OOP 代码实现4.1 项目结构shop_floor_topology/├── shop_floor_topology.py # 核心代码├── test_shop_floor_topology.py # 单元测试├── results/ # 输出结果│ ├── topology_before.png # 改造前拓扑图│ ├── topology_after.png # 改造后拓扑图│ ├── path_length_comparison.png # 路径长度对比│ ├── simulation_report.txt # 分析报告│ └── evaluation_results.csv # 评估指标数据└── README.md4.2 核心源码detailssummary/summary车间拓扑改造仿真增删设备节点评估网络连通性与物料输送效率课程映射长安大学《智能制造导论》概述车间布局与物流规划技术基础物料输送系统支撑技术数字孪生、图论算法智能工厂柔性产线、可重构制造演进范式固定布局 → 可重构布局 → 自适应布局技术栈严格numpy # 数组运算pandas # 结果统计matplotlib # 可视化networkx # 车间网络拓扑与路径计算from __future__ import annotationsimport osfrom dataclasses import dataclassfrom pathlib import Pathfrom typing import List, Dict, Tuple, Set, Optionalimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport matplotlib.patches as mpatchesplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseimport networkx as nx# ----------------------------------------------------------------------# 1. 节点定义# ----------------------------------------------------------------------class NodeType:设备节点类型ENTRY entry # 入库口PROCESS process # 加工中心AGV agv # AGV转运站BUFFER buffer # 缓存区EXIT exit # 出库口dataclassclass ShopFloorNode:车间设备节点name: strnode_type: strprocess_time: float 0.0 # 在该节点的处理时间分钟def __repr__(self):return f{self.name}({self.node_type})# ----------------------------------------------------------------------# 2. 车间网络# ----------------------------------------------------------------------class ShopFloorNetwork:车间物流网络基于NetworkX有向图def __init__(self, name: str ShopFloor):self.name nameself.graph nx.DiGraph()self.nodes: Dict[str, ShopFloorNode] {}def add_node(self, node: ShopFloorNode):添加设备节点self.nodes[node.name] nodeself.graph.add_node(node.name,node_typenode.node_type,process_timenode.process_time,)def add_edge(self, from_node: str, to_node: str,travel_time: float):添加物流路径有向边权重输送时间if from_node not in self.nodes or to_node not in self.nodes:raise ValueError(f节点不存在: {from_node} 或 {to_node})self.graph.add_edge(from_node, to_node,weighttravel_time,travel_timetravel_time,)def remove_node(self, node_name: str):删除设备节点if node_name in self.nodes:self.graph.remove_node(node_name)del self.nodes[node_name]def remove_edge(self, from_node: str, to_node: str):删除物流路径if self.graph.has_edge(from_node, to_node):self.graph.remove_edge(from_node, to_node)def get_node_count(self) - int:return len(self.nodes)def get_edge_count(self) - int:return self.graph.number_of_edges()def copy(self) - ShopFloorNetwork:深拷贝网络new_net ShopFloorNetwork(self.name _copy)for node in self.nodes.values():new_net.add_node(ShopFloorNode(node.name, node.node_type, node.process_time))for u, v, data in self.graph.edges(dataTrue):new_net.add_edge(u, v, data[travel_time])return new_net# ----------------------------------------------------------------------# 3. 拓扑修改器# ----------------------------------------------------------------------class TopologyModifier:执行拓扑改造操作def __init__(self, network: ShopFloorNetwork):self.network networkdef add_equipment(self, node: ShopFloorNode,connections: List[Tuple[str, str, float]]):新增设备并连接connections: [(from, to, travel_time), ...]print(f[MODIFY] 新增设备: {node.name})self.network.add_node(node)for u, v, t in connections:self.network.add_edge(u, v, t)def remove_equipment(self, node_name: str):移除设备print(f[MODIFY] 移除设备: {node_name})self.network.remove_node(node_name)def add_path(self, from_node: str, to_node: str, travel_time: float):新增物流路径print(f[MODIFY] 新增路径: {from_node} → {to_node} ({travel_time}min))self.network.add_edge(from_node, to_node, travel_time)def remove_path(self, from_node: str, to_node: str):移除物流路径print(f[MODIFY] 移除路径: {from_node} → {to_node})self.network.remove_edge(from_node, to_node)# ----------------------------------------------------------------------# 4. 连通性评估器# ----------------------------------------------------------------------class ConnectivityEvaluator:评估网络连通性def __init__(self):passdef is_connected(self, network: ShopFloorNetwork) - bool:检查全网是否弱连通return nx.is_weakly_connected(network.graph)def connected_components_count(self, network: ShopFloorNetwork) - int:连通分量数量return nx.number_weakly_connected_components(network.graph)def reachability_matrix(self, network: ShopFloorNetwork) - pd.DataFrame:可达性矩阵节点对之间是否可达node_names list(network.nodes.keys())n len(node_names)matrix pd.DataFrame(np.zeros((n, n), dtypebool),indexnode_names, columnsnode_names)for u in node_names:if u not in network.graph:continuereachable nx.descendants(network.graph, u)reachable.add(u)for v in reachable:matrix.loc[u, v] Truereturn matrixdef entry_to_exit_paths(self, network: ShopFloorNetwork) - Dict:检查所有入库口到出库口的路径entry_nodes [n for n, d in network.graph.nodes(dataTrue)if d.get(node_type) NodeType.ENTRY]exit_nodes [n for n, d in network.graph.nodes(dataTrue)if d.get(node_type) NodeType.EXIT]results {}for entry in entry_nodes:results[entry] {}for exit in exit_nodes:if nx.has_path(network.graph, entry, exit):path nx.shortest_path(network.graph, entry, exit,weightweight)length nx.shortest_path_length(network.graph, entry, exit,weightweight)results[entry][exit] {reachable: True,path: path,length: length,}else:results[entry][exit] {reachable: False,path: None,length: float(inf),}return results# ----------------------------------------------------------------------# 5. 效率评估器# ----------------------------------------------------------------------class EfficiencyEvaluator:评估物料输送效率def __init__(self):passdef average_shortest_path_length(self, network: ShopFloorNetwork) - float:计算全网平均最短路径长度仅对可达节点对if not nx.is_weakly_connected(network.graph):return float(inf)total_length 0.0count 0for u in network.nodes:lengths nx.single_source_dijkstra_path_length(network.graph, u, weightweight)for v, length in lengths.items():if u v: # 避免重复计算total_length lengthcount 1return total_length / count if count 0 else float(inf)def bottleneck_nodes(self, network: ShopFloorNetwork,top_k: int 3) - List[Tuple[str, float]]:识别瓶颈节点基于度中心性in_degree dict(network.graph.in_degree())out_degree dict(network.graph.out_degree())# 总度数total_degree {}for node in network.nodes:total_degree[node] in_degree.get(node, 0) out_degree.get(node, 0)# 排序sorted_nodes sorted(total_degree.items(), keylambda x: x[1],reverseTrue)return sorted_nodes[:top_k]def path_redundancy(self, network: ShopFloorNetwork,source: str, target: str) - int:计算两节点间的路径冗余度独立路径数量if not nx.has_path(network.graph, source, target):return 0# 使用简单方法枚举所有简单路径对于小图可行try:paths list(nx.all_simple_paths(network.graph, source, target,cutoff10))return len(paths)except Exception:return 0# ----------------------------------------------------------------------# 6. 可视化器# ----------------------------------------------------------------------class Visualizer:可视化分析结果def __init__(self):self.results_dir Path(results)os.makedirs(self.results_dir, exist_okTrue)def plot_topology(self, network: ShopFloorNetwork, title: str,filename: str):绘制车间拓扑图print(f[INFO] 绘制拓扑图: {title})fig, ax plt.subplots(figsize(14, 8))# 分层布局pos nx.spring_layout(network.graph, seed42, k2, iterations50)# 节点颜色映射color_map {NodeType.ENTRY: #27AE60, # 绿色NodeType.PROCESS: #3498DB, # 蓝色NodeType.AGV: #E74C3C, # 红色NodeType.BUFFER: #F39C12, # 橙色NodeType.EXIT: #8E44AD, # 紫色}node_colors [color_map.get(network.graph.nodes[n].get(node_type),#95A5A6)for n in network.graph.nodes]# 节点大小根据度中心性degrees dict(network.graph.degree())max_deg max(degrees.values()) if degrees else 1node_sizes [500 1500 * degrees[n] / max_degfor n in network.graph.nodes]nx.draw_networkx_nodes(network.graph, pos, axax,node_colornode_colors, node_sizenode_sizes,alpha0.85, edgecolorswhite)nx.draw_networkx_edges(network.graph, pos, axax,arrowstyle-, arrowsize15,edge_color#7F8C8D, width1.5,connectionstylearc3,rad0.1)nx.draw_networkx_labels(network.graph, pos, axax,font_size9, font_weightbold)# 边权重标注edge_labels {(u, v): f{d[weight]}minfor u, v, d in network.graph.edges(dataTrue)}nx.draw_networkx_edge_labels(network.graph, pos, axax,edge_labelsedge_labels,font_size7)# 图例patches [mpatches.Patch(colorv, labelk)for k, v in color_map.items()]ax.legend(handlespatches, locupper right, fontsize10)ax.set_title(title, fontsize14, fontweightbold)ax.axis(off)plt.tight_layout()plt.savefig(self.results_dir / filename, dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / filename})def plot_path_comparison(self, before_data: Dict, after_data: Dict):对比改造前后的路径长度print([INFO] 绘制路径长度对比...)# 收集所有入库→出库的对比comparisons []for entry, exits in before_data.items():for exit, data in exits.items():if data[reachable] and exit in after_data.get(entry, {}):after after_data[entry][exit]if after[reachable]:comparisons.append({route: f{entry}→{exit},before: data[length],after: after[length],change: after[length] - data[length],})if not comparisons:print( 无可达路径对比)returndf pd.DataFrame(comparisons)df df.sort_values(before)fig, axes plt.subplots(2, 1, figsize(12, 8), sharexTrue)x np.arange(len(df))w 0.35# 绝对路径长度axes[0].bar(x - w/2, df[before], w, label改造前,color#3498DB, alpha0.8)axes[0].bar(x w/2, df[after], w, label改造后,color#E74C3C, alpha0.8)axes[0].set_ylabel(路径长度 (分钟), fontsize12)axes[0].set_title(入库→出库 路径长度对比, fontsize13, fontweightbold)axes[0].legend()axes[0].set_xticks(x)axes[0].set_xticklabels(df[route], rotation15)axes[0].grid(True, alpha0.3, axisy)# 变化量colors [#E74C3C if c 0 else #27AE60 for c in df[change]]axes[1].bar(df[route], df[change], colorcolors, alpha0.8)axes[1].axhline(y0, colorblack, linewidth0.8)axes[1].set_ylabel(变化量 (分钟), fontsize12)axes[1].set_title(路径长度变化红色变长绿色变短,fontsize13, fontweightbold)axes[1].tick_params(axisx, rotation15)axes[1].grid(True, alpha0.3, axisy)plt.tight_layout()plt.savefig(self.results_dir / path_length_comparison.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / path_length_comparison.png})# ----------------------------------------------------------------------# 7. 报告生成器# ----------------------------------------------------------------------class ReportGenerator:分析报告生成器def __init__(self):self.results_dir Path(results)os.makedirs(self.results_dir, exist_okTrue)def generate(self, before_net: ShopFloorNetwork,after_net: ShopFloorNetwork,before_conn: Dict, after_conn: Dict,before_eff: Dict, after_eff: Dict) - str:生成报告print([INFO] 生成分析报告...)report_lines []report_lines.append( * 80)report_lines.append(车间拓扑改造仿真分析报告)report_lines.append( * 80)report_lines.append(f\n网络规模:)report_lines.append(f {指标:20} {改造前:10} {改造后:10} {变化:10})report_lines.append(- * 55)report_lines.append(f {节点数:20} {before_net.get_node_count():10} f{after_net.get_node_count():10} f{after_net.get_node_count() - before_net.get_node_count():10})report_lines.append(f {边数:20} {before_net.get_edge_count():10} f{after_net.get_edge_count():10} f{after_net.get_edge_count() - before_net.get_edge_count():10})report_lines.append(f\n连通性评估:)report_lines.append(f {指标:25} {改造前:10} {改造后:10})report_lines.append(- * 50)report_lines.append(f {全网连通:25} f{✅ if before_conn[connected] else ❌:10} f{✅ if after_conn[connected] else ❌:10})report_lines.append(f {连通分量数:25} f{before_conn[components]:10} f{after_conn[利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
