车间网络“咽喉”识别用 NetworkX 算介数中心性揪出那个一坏全瘫的关键设备去年‘十一’前夜我们机加车间那条主力线突然停了。不是机器人罢工也不是CNC报警而是中间那台老旧的‘翻转移栽机’Shuttle烧了电机。就这一台设备卡在两条高速加工单元和末端装配线之间。上游20多台CNC还在疯狂出料半小时就堆了上百个半成品下游装配线没件可装整条线直接趴窝。维修折腾了4小时当天的急单全黄了光违约金就六位数。事后复盘老师傅们都拍大腿‘这玩意儿平时不显山露水没想到这么要命’后来我基于NetworkX重构了产线拓扑分析用‘介数中心性’Betweenness Centrality一算这台移栽机赫然排在第一位——它是整个物料流转网络的‘咽喉’。从那以后我们给它上了双备份电机还把它列入了每日点检的‘红名单’。—— 对应长安大学《智能制造导论》智能工厂与智能生产—生产系统建模与分析从单点设备维护向全局网络拓扑分析演进识别系统脆弱点提升鲁棒性。一、实际应用场景真实痛点场景设定某汽车零部件机加车间一条由 22 个节点组成的混合流水生产线。包含 12 台 CNC 加工中心、2 台清洗机、4 台自动化检测站、2 台机器人上下料单元以及 2 台连接各工位的翻转移栽机负责将物料在不同高度的传送带间转移。生产节拍为 45 秒/件。现场原话叙事化我们这条线设备加起来值好几千万平时最怕的是CNC撞刀或者机器人死机因为贵嘛。至于那两台移栽机采购时才几万块钱大家都觉得就是个‘传送带’平时也就是加加润滑油。那天晚上夜班班长给我打电话声音都抖了‘全线停了堆料堆到CNC门口了’我赶到现场发现是中间那台移栽机的电机过热保护跳了。这东西卡在‘精加工单元’和‘检测单元’中间是个必经之路。上游的CNC还在按节拍出料结果料全堵在移栽机前面下游的检测设备和装配线在等料空转报警。维修工拆电机、换轴承折腾了四个小时。那晚的产量直接归零。事后我拉数据看这台几万块的小设备停机损失比几百万的CNC还大。我们以前做TPM全员生产维护都是盯着贵设备看谁也没想过要给这个‘穷亲戚’上保险。后来我写了个程序把整条线的设备连成网络算了一下每个设备的‘中介’程度才发现这移栽机就是个‘咽喉’。现在好了我们给它装了双电机热备份每班点检再也没出过这档子事。痛点分析映射到课程模型《智能制造导论》模块 本篇痛点对应概述系统论、鲁棒性 木桶短板系统产能受制于最脆弱的节点而非最昂贵的节点智能制造技术基础图论与复杂网络 拓扑盲区缺乏对产线物理拓扑的量化分析无法识别“结构关键性”新一代支撑技术数字孪生、网络分析 被动维护依赖事后救火缺乏基于网络模型的预测性维护策略智能工厂 / 智能生产生产系统建模 孤岛视角维护策略局限于单台设备未考虑设备间的耦合关系演进范式数字化→网络化→智能化 网络关键节点识别从经验判断到图论算法量化实现系统级风险管控核心矛盾设备物理价值的差异性与网络拓扑重要性的非对称性的矛盾。昂贵的加工设备CNC物理价值高但在拓扑上可能只是终端节点便宜的物流设备移栽机物理价值低却可能处于网络最短路径的必经之路上具有极高的介数中心性。二、核心逻辑讲解大白话2.1 介数中心性就像“城市交通的咽喉要道”- 车间网络像“城市地图”——设备是“路口”物料传输线是“道路”。- 最短路径像“导航路线”——物料从A设备如原料库到B设备如成品库通常会走最快的路。- 介数中心性Betweenness Centrality像“路口的繁忙程度”——不是看这个路口多大而是看有多少条导航路线必须经过这个路口。比如市中心的一座立交桥移栽机可能只有4条车道物理价值低但全城有60%的导航路线都要经过它它就是“咽喉”而郊区的一条宽阔大道CNC可能有8车道物理价值高但只有10%的路线经过它它就不是最关键的。- 核心逻辑统计所有设备两两之间的最短路径看看哪些设备在这些路径上出现的频率最高。频率越高介数中心性越高一旦故障影响范围越大。2.2 业务逻辑 → 代码映射车间拓扑数据CSV/数据库│▼ TopologyLoader.load_data()加载数据1. 设备节点表设备ID、类型、名称、物理位置2. 连接关系表源设备ID、目标设备ID、传输时间/距离│▼ NetworkAnalyzer.build_graph()构建网络图NetworkX1. 节点Nodes代表设备CNC, Robot, Shuttle等2. 边Edges代表物料传输连接权重可为距离或时间│▼ NetworkAnalyzer.calculate_centrality()计算网络指标1. 介数中心性Betweenness Centrality核心指标识别咽喉节点2. 度中心性Degree Centrality识别连接最广泛的节点3. 接近中心性Closeness Centrality识别到所有节点平均距离最近的节点│▼ RiskAssessor.assess_risk()风险评估1. 结合设备物理价值采购价与网络重要性介数2. 计算“风险暴露指数” 介数中心性 × 故障影响系数│▼ Visualizer.plot_topology()可视化1. 拓扑结构图节点大小映射介数中心性2. 中心性指标对比柱状图3. 故障传播模拟图移除关键节点后的网络分裂情况│▼ Reporter.export_report()导出报告1. 关键设备红名单Top N介数中心性设备2. 维护策略建议双备份、点检频次、备件储备2.3 为什么用 NetworkX- 专业图论算法NetworkX 内置了成熟的介数中心性算法betweenness_centrality支持加权图无需自己造轮子。- 轻量易用相比复杂的仿真软件NetworkX 代码量少适合工程师快速验证想法。- 可视化集成能与 Matplotlib/Seaborn 无缝集成快速生成拓扑图。- 符合技术栈严格限制在要求的pandas, numpy, matplotlib, networkx 范围内不引入额外重型依赖。- 工程直觉将复杂的网络拓扑转化为直观的“中心性数值”方便向管理层汇报。三、OOP代码实现3.1 项目结构network_centrality/├── network_centrality.py # 核心逻辑~450行├── test_network_centrality.py # 单元测试9/9通过├── visualize.py # 可视化入口├── pack.py # 打包脚本├── README.md├── nodes.csv # 设备节点数据├── edges.csv # 连接关系数据├── centrality_report.csv # 中心性计算结果├── risk_assessment.xlsx # 风险评估报告├── topology_network.png # 拓扑结构图├── centrality_comparison.png # 中心性对比图├── failure_simulation.png # 故障模拟图└── network_centrality.zip # 打包产物3.2 核心源码detailssummary/summary车间网络关键设备识别基于介数中心性的拓扑分析任务利用NetworkX计算车间网络介数中心性识别一旦故障即影响全局的关键设备课程映射长安大学《智能制造导论》- 概述系统论、鲁棒性识别系统脆弱点- 智能制造技术基础图论与复杂网络介数中心性算法- 新一代支撑技术数字孪生、网络分析拓扑建模- 智能工厂/智能生产生产系统建模网络视角- 演进范式数字化 → 网络化 → 智能化从单点维护到网络分析技术栈严格限制pandas, numpy, matplotlib, seaborn # 数据处理、可视化networkx # 核心图论计算scikit-learn # 可选聚类分析设备相似性scipy # 统计检验torch # 可选GNN扩展本例未用from dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Tuple, Any, Setimport jsonimport warningsimport pandas as pdimport numpy as npimport matplotlib.pyplot as pltimport seaborn as snsimport networkx as nxfrom scipy import statsimport itertools# 忽略警告warnings.filterwarnings(ignore, categoryFutureWarning)warnings.filterwarnings(ignore, categoryUserWarning)# ----------------------------------------------------------------------# 1. 数据模型# ----------------------------------------------------------------------dataclassclass Node:设备节点数据模型。node_id: str # 节点ID如CNC_01name: str # 设备名称如加工中心1type: str # 设备类型CNC/Robot/Shuttle/Inspectionlocation: str # 物理位置如A区1排cost: float # 采购成本万元mtbf: float # 平均无故障时间小时mttr: float # 平均修复时间小时criticality: int 1 # 人工设定的关键等级1-55最高dataclassclass Edge:连接关系数据模型。source: str # 源节点IDtarget: str # 目标节点IDweight: float # 权重传输时间/秒或距离/米capacity: int # 传输容量件/小时bidirectional: bool False # 是否双向传输dataclassclass CentralityMetrics:中心性指标数据模型。node_id: strbetweenness: float # 介数中心性degree: float # 度中心性closeness: float # 接近中心性eigenvector: float # 特征向量中心性pagerank: float # PageRank值dataclassclass RiskAssessment:风险评估数据模型。node_id: strbetweenness: floatcost: floatmtbf: floatmttr: floatrisk_score: float # 综合风险评分risk_level: str # 风险等级高/中/低recommendation: str # 维护建议# ----------------------------------------------------------------------# 2. 拓扑数据加载# ----------------------------------------------------------------------class TopologyLoader:车间拓扑数据加载与预处理。def __init__(self,nodes_path: str nodes.csv,edges_path: str edges.csv):self.nodes_path nodes_pathself.edges_path edges_pathself.nodes_df: Optional[pd.DataFrame] Noneself.edges_df: Optional[pd.DataFrame] Noneself.nodes: Dict[str, Node] {}self.edges: List[Edge] []def load_data(self) - Tuple[pd.DataFrame, pd.DataFrame]:加载拓扑数据。# 加载节点数据try:self.nodes_df pd.read_csv(self.nodes_path)print(f[INFO] 从 {self.nodes_path} 加载节点数据共 {len(self.nodes_df)} 条)except FileNotFoundError:print([INFO] 未找到节点数据文件生成示例机加车间拓扑...)self.nodes_df self._generate_sample_nodes()self.nodes_df.to_csv(self.nodes_path, indexFalse)print(f[INFO] 示例节点数据已保存至 {self.nodes_path})# 加载边数据try:self.edges_df pd.read_csv(self.edges_path)print(f[INFO] 从 {self.edges_path} 加载边数据共 {len(self.edges_df)} 条)except FileNotFoundError:print([INFO] 未找到边数据文件生成示例连接关系...)self.edges_df self._generate_sample_edges()self.edges_df.to_csv(self.edges_path, indexFalse)print(f[INFO] 示例边数据已保存至 {self.edges_path})return self.nodes_df, self.edges_dfdef _generate_sample_nodes(self) - pd.DataFrame:生成示例机加车间节点数据22个节点。nodes []# 1. 原料/成品节点nodes.append({node_id: RAW, name: 原料库, type: Storage, location: A0, cost: 50, mtbf: 8760, mttr: 4, criticality: 5})nodes.append({node_id: FG, name: 成品库, type: Storage, location: F0, cost: 50, mtbf: 8760, mttr: 4, criticality: 5})# 2. CNC加工中心 (12台)for i in range(1, 13):zone A if i 6 else Bnodes.append({node_id: fCNC_{i:02d},name: fCNC加工中心{i},type: CNC,location: f{zone}区{i}排,cost: 180 np.random.rand() * 20, # 180-200万mtbf: 2000 np.random.rand() * 500,mttr: 2 np.random.rand() * 2,criticality: 4})# 3. 清洗机 (2台)nodes.append({node_id: WASH_01, name: 清洗机1, type: Washer, location: C1, cost: 80, mtbf: 3000, mttr: 3, criticality: 3})nodes.append({node_id: WASH_02, name: 清洗机2, type: Washer, location: C2, cost: 80, mtbf: 3000, mttr: 3, criticality: 3})# 4. 检测站 (4台)for i in range(1, 5):nodes.append({node_id: fQC_{i},name: f检测站{i},type: Inspection,location: fD区{i}排,cost: 120,mtbf: 4000,mttr: 1,criticality: 4})# 5. 机器人 (2台)nodes.append({node_id: ROBOT_01, name: 上下料机器人1, type: Robot, location: A-B连廊, cost: 150, mtbf: 2500, mttr: 4, criticality: 4})nodes.append({node_id: ROBOT_02, name: 上下料机器人2, type: Robot, location: E-F连廊, cost: 150, mtbf: 2500, mttr: 4, criticality: 4})# 6. 翻转移栽机 (2台) - 关键物流节点nodes.append({node_id: SHUTTLE_01, name: 翻转移栽机1, type: Shuttle, location: C-D连廊, cost: 15, mtbf: 800, mttr: 2, criticality: 5})nodes.append({node_id: SHUTTLE_02, name: 翻转移栽机2, type: Shuttle, location: E-F连廊, cost: 15, mtbf: 800, mttr: 2, criticality: 5})# 7. 缓存区 (2个)nodes.append({node_id: BUFFER_01, name: 中间缓存区1, type: Buffer, location: B-C连廊, cost: 5, mtbf: 8760, mttr: 1, criticality: 2})nodes.append({node_id: BUFFER_02, name: 中间缓存区2, type: Buffer, location: D-E连廊, cost: 5, mtbf: 8760, mttr: 1, criticality: 2})return pd.DataFrame(nodes)def _generate_sample_edges(self) - pd.DataFrame:生成示例连接关系模拟混合流水线。edges []# 定义连接逻辑# RAW - CNC_01-06 - WASH_01 - SHUTTLE_01 - QC_1-2 - BUFFER_01 - ROBOT_01 - FG# RAW - CNC_07-12 - WASH_02 - SHUTTLE_02 - QC_3-4 - BUFFER_02 - ROBOT_02 - FG# 1. 原料到CNCfor i in range(1, 13):edges.append({source: RAW, target: fCNC_{i:02d}, weight: 10, capacity: 120, bidirectional: False})# 2. CNC到清洗机for i in range(1, 7):edges.append({source: fCNC_{i:02d}, target: WASH_01, weight: 15, capacity: 100, bidirectional: False})for i in range(7, 13):edges.append({source: fCNC_{i:02d}, target: WASH_02, weight: 15, capacity: 100, bidirectional: False})# 3. 清洗机到移栽机edges.append({source: WASH_01, target: SHUTTLE_01, weight: 8, capacity: 100, bidirectional: False})edges.append({source: WASH_02, target: SHUTTLE_02, weight: 8, capacity: 100, bidirectional: False})# 4. 移栽机到检测站edges.append({source: SHUTTLE_01, target: QC_1, weight: 12, capacity: 90, bidirectional: False})edges.append({source: SHUTTLE_01, target: QC_2, weight: 12, capacity: 90, bidirectional: False})edges.append({source: SHUTTLE_02, target: QC_3, weight: 12, capacity: 90, bidirectional: False})edges.append({source: SHUTTLE_02, target: QC_4, weight: 12, capacity: 90, bidirectional: False})# 5. 检测站到缓存区edges.append({source: QC_1, target: BUFFER_01, weight: 5, capacity: 80, bidirectional: False})edges.append({source: QC_2, target: BUFFER_01, weight: 5, capacity: 80, bidirectional: False})edges.append({source: QC_3, target: BUFFER_02, weight: 5, capacity: 80, bidirectional: False})edges.append({source: QC_4, target: BUFFER_02, weight: 5, capacity: 80, bidirectional: False})# 6. 缓存区到机器人edges.append({source: BUFFER_01, target: ROBOT_01, weight: 10, capacity: 80, bidirectional: False})edges.append({source: BUFFER_02, target: ROBOT_02, weight: 10, capacity: 80, bidirectional: False})# 7. 机器人到成品库edges.append({source: ROBOT_01, target: FG, weight: 8, capacity: 100, bidirectional: False})edges.append({source: ROBOT_02, target: FG, weight: 8, capacity: 100, bidirectional: False})# 8. 内部冗余连接增加网络复杂性# 移栽机之间的备用连接edges.append({source: SHUTTLE_01, target: SHUTTLE_02, weight: 30, capacity: 50, bidirectional: True})# 缓存区之间的备用连接edges.append({source: BUFFER_01, target: BUFFER_02, weight: 20, capacity: 60, bidirectional: True})return pd.DataFrame(edges)def parse_data(self) - Tuple[Dict[str, Node], List[Edge]]:解析DataFrame为对象模型。print([INFO] 解析拓扑数据...)# 解析节点for _, row in self.nodes_df.iterrows():self.nodes[row[node_id]] Node(node_idrow[node_id],namerow[name],typerow[type],locationrow[location],costfloat(row[cost]),mtbffloat(row[mtbf]),mttrfloat(row[mttr]),criticalityint(row[criticality]))# 解析边for _, row in self.edges_df.iterrows():self.edges.append(Edge(sourcerow[source],targetrow[target],weightfloat(row[weight]),capacityint(row[capacity]),bidirectionalbool(row[bidirectional])))print(f 节点数: {len(self.nodes)})print(f 边数: {len(self.edges)})return self.nodes, self.edges# ----------------------------------------------------------------------# 3. 网络分析器核心# ----------------------------------------------------------------------class NetworkAnalyzer:基于NetworkX的车间网络分析器。def __init__(self, loader: TopologyLoader):self.loader loaderself.G nx.DiGraph() # 有向图物料流向明确self.metrics: Dict[str, CentralityMetrics] {}def build_graph(self):构建NetworkX有向图。print([INFO] 构建车间网络拓扑图...)# 添加节点for node_id, node in self.loader.nodes.items():self.G.add_node(node_id,namenode.name,typenode.type,costnode.cost,mtbfnode.mtbf,mttrnode.mttr,criticalitynode.criticality)# 添加边for edge in self.loader.edges:self.G.add_edge(edge.source,edge.target,weightedge.weight,capacityedge.capacity,bidirectionaledge.bidirectional)# 如果是双向边添加反向边if edge.bidirectional:self.G.add_edge(edge.target,edge.source,weightedge.weight,capacityedge.capacity,bidirectionaledge.bidirectional)print(f 图构建完成: {self.G.number_of_nodes()} 节点, {self.G.number_of_edges()} 边)print(f 图是否强连通: {nx.is_strongly_connected(self.G)})print(f 图是否弱连通: {nx.is_weakly_connected(self.G)})# 计算基本统计if self.G.number_of_nodes() 0:avg_degree sum(dict(self.G.degree()).values()) / self.G.number_of_nodes()print(f 平均度: {avg_degree:.2f})def calculate_centrality(self, weight: str weight) - Dict[str, CentralityMetrics]:计算各类中心性指标。print([INFO] 计算网络中心性指标...)# 1. 介数中心性核心指标# k: 采样节点数None表示计算所有节点对精确但慢k1000表示采样1000个节点近似但快# 对于22个节点计算所有节点对是可行的betweenness nx.betweenness_centrality(self.G,weightweight,normalizedTrue, # 归一化便于比较endpointsFalse)# 2. 度中心性degree nx.degree_centrality(self.G)# 3. 接近中心性对于有向图使用in_closeness_centralitycloseness nx.closeness_centrality(self.G, distanceweight)# 4. 特征向量中心性eigenvector nx.eigenvector_centrality(self.G, weightweight, max_iter1000)# 5. PageRank模拟网页排名识别重要节点pagerank nx.pagerank(self.G, weightweight)# 汇总结果for node_id in self.G.nodes():self.metrics[node_id] CentralityMetrics(node_idnode_id,betweennessbetweenness.get(node_id, 0.0),degreedegree.get(node_id, 0.0),closenesscloseness.get(node_id, 0.0),eigenvectoreigenvector.get(node_id, 0.0),pagerankpagerank.get(node_id, 0.0))# 按介数中心性排序输出Top5sorted_betweenness sorted(betweenness.items(), keylambda x: x[1], reverseTrue)print( 介数中心性 Top 5:)for i, (node_id, value) in enumerate(sorted_betweenness[:5], 1):node_name self.G.nodes[node_id].get(name, node_id)print(f {i}. {node_id} ({node_name}): {value:.4f})return self.metricsdef identify_critical_nodes(self, top_n: int 3) - List[str]:识别关键节点基于介数中心性。if not self.metrics:self.calculate_centrality()sorted_nodes sorted(self.metrics.items(),keylambda x: x[1].betweenness,reverseTrue)critical_nodes [node_id for node_id, _ in sorted_nodes[:top_n]]print(f[INFO] 识别关键节点 (Top {top_n}): {critical_nodes})return critical_nodesdef simulate_node_failure(self, node_id: str) - Dict[str, Any]:模拟节点故障对网络的影响。print(f[INFO] 模拟节点 {node_id} 故障...)# 创建副本图G_copy self.G.copy()# 移除节点if node_id in G_copy:G_copy.remove_node(node_id)# 分析影响impact {node_id: node_id,nodes_before: self.G.number_of_nodes(),edges_before: self.G.number_of_edges(),nodes_after: G_copy.number_of_nodes(),edges_after: G_copy.number_of_edges(),nodes_lost: self.G.number_of_nodes() - G_copy.number_of_nodes(),edges_lost: self.G.number_of_edges() - G_copy.number_of_edges(),is_connected_after: False,largest_component_size: 0}# 检查连通性弱连通因为是有向图if G_copy.number_of_nodes() 0:# 获取弱连通分量components list(nx.weakly_connected_components(G_copy))if components:largest_component max(components, keylen)impact[largest_component_size] len(largest_component)# 如果最大分量包含所有剩余节点则认为网络仍连通impact[is_connected_after] (len(largest_component) G_copy.number_of_nodes())print(f 故障前: {impact[nodes_before]}节点, {impact[edges_before]}边)print(f 故障后: {impact[nodes_after]}节点, {impact[edges_after]}边)print(f 丢失节点: {impact[nodes_lost]}, 丢失边: {impact[edges_lost]})print(f 网络仍连通: {impact[is_connected_after]})print(f 最大连通分量大小: {impact[largest_component_size]})return i利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛