周二下午机器人工作站调试现场。又急停了现场电气老赵拍了拍安全门示教器上就显示安全链路断开可我得花半小时翻电气图才能知道这个安全门连的是哪台机器人、哪个料台、中间过了几个继电器。新来的实习生更懵站里有 2 台机器人、4 套工装、3 个相机、5 道安全门、6 个料台全靠线槽里那把线认关系。我打开运维文档。这图里有什么老赵问。现在是几张 PDF 电气图 一张布局草图我指着屏幕机器人、工装夹具、视觉相机、安全门、料台都是散的。想看相机拍完→给哪个机器人信号→机器人取哪个料台得人脑拼。我就想要一张关系图老赵说像流程图那样节点是设备连线是信号/物料/安全联锁关系。最好能标出哪个节点是枢纽哪条链路断了会连带急停排故障不用翻三本图册。所以你要的不是电气原理图是工作站拓扑图 关键路径识别 联锁影响域分析对老赵点头比如安全门 S2 断开影响哪几台机器人、哪几个料台能不能一键算出来。还有相机 C1 挂了是不是只影响机器人 R1 的来料定位。明白了我开 VS Code用 networkx 把工作站建成有向无向混合图机器人、工装、相机、安全门、料台都是节点边分三类——物料流、信号流、安全联锁。用节点中心度找枢纽用子图遍历算安全门影响域用 matplotlib 画拓扑图用 pandas 管设备台账用 scipy 做度分布统计用 scikit-learn 对节点角色聚类。数据自包含——合成一个双机器人上下料工作站下载就能跑。我敲了段原型import networkx as nxG nx.Graph()G.add_node(R1, typerobot)G.add_node(S2, typesafety_door)G.add_edge(S2, R1, relationsafety_interlock)完整版用 OOP 封好我说一个类管设备台账加载一个类建拓扑一个类算中心度一个类算安全影响域一个类画图一个类做节点角色聚类。输出拓扑图、枢纽排名、安全门影响清单。老赵凑近看那我以后 S2 一报警程序直接列出影响 R1、R2、料台 T3/T4、工装 F2因为 S2 在它们的安全子图里。不用再翻图了。对我接话而且中心度一排R1 是整站枢纽所有料台和相机都绕它走。数字孪生不是先搞炫酷 3D是把设备关系先变成一张可计算的图。一、实际应用场景真实痛点场景设定工业机器人工作站上下料/焊接/检测含多台机器人、工装夹具、视觉定位相机、安全光栅/安全门、来料料台。现场文档以电气原理图 布局图分离存在信号链、物料链、安全联锁链没有统一视图。排故时一个安全门触发急停工程师需人工反推影响范围耗时且易漏判。现场原话叙事化我不是不会看电气图老赵说是图太碎了。安全门在一张图机器人 IO 在另一张相机触发线又在第三张。有次 S2 门被撞开R1 R2 都停了料台也锁了可新人按图找了 40 分钟。后来我把节点关系手画在一张 A3 纸上才发现 S2 是挂了两个机器人安全回路的公共节点。可这纸一换人就容易丢系统里从没存在过这张关系网。核心矛盾设备台账 分散电气图与可计算的工作站拓扑图 中心度排名 安全联锁影响域自动推导之间的断层。需要一个机器人工作站拓扑建模与影响域分析程序用networkx 建图pandas 管台账numpy 算矩阵matplotlib 画图scipy 做度分布检验scikit-learn 做节点角色聚类。二、痛点分析映射到滨州职业学院《先进制造技术》课程模型《先进制造技术》模块 本篇痛点对应工业机器人技术基础工作站组成与 IO 联锁逻辑 机器人/工装/相机/安全门/料台拓扑建模智能制造与数字孪生产线数字孪生 用 networkx 构建可计算的逻辑孪生体先进制造技术基础制造系统组成与信息流 物料流/信号流/安全流三类边分离FMS与先进生产管理单元级调度与联锁管理 安全门影响域推导辅助排故与排产先进制造新模式可视化运维 枢纽节点识别运维优先级排序一句话总结我们需要构建一个机器人工作站拓扑建模程序用networkx 把物理设备抽象成图pandas 管设备属性scipy 分析度分布scikit-learn 聚类节点角色实现从翻三本图册到一张可计算拓扑图的转化。三、核心逻辑讲解大白话3.1 问题本质把工作站想象成小区关系网把工作站想成一个小区* 机器人 小区里的主楼干活的大家都跟它连。* 工装 主楼里的工位隔间机器人得夹着它干活。* 相机 门禁摄像头拍来料位置给机器人发对准了信号。* 安全门 单元门禁一开就全楼断电急停。* 料台 楼前货架放毛坯和成品。* 边 关系有送货关系物料流、有喊话关系信号流、有门禁联锁关系安全流。* 中心度 谁最忙跟它连的设备越多越是不能随便动。* 安全影响域 门禁一开哪些房间断电从安全门节点做图遍历能走到谁就影响谁。工业应用*networkx.Graph 存无向物料/信号关系MultiDiGraph 可区分方向本例用带属性Graph 更直观。* 节点属性type 区分五类设备边属性relation 区分material/signal/safety。*nx.degree_centrality 排枢纽nx.node_connected_component 算安全子图。3.2 业务逻辑 → 代码映射定义工作站设备台账模型│▼ StationTopoLoader (pandas)导入 CSVnode_id, node_type(robot/fixture/camera/safety_door/table), x, y, desc导入边表src, dst, relation│▼ TopoBuilder (networkx)建图节点按类型加属性边按 relation 加属性生成混合关系图│▼ CentralityAnalyzer (networkx/numpy)枢纽分析degree / betweenness / closeness按设备类型分组排名│▼ SafetyImpactAnalyzer (networkx)安全影响域从 safety_door 节点出发仅沿 safety 边遍历连通分量输出受影响节点清单│▼ RoleClusterer (scikit-learn)节点角色聚类用节点特征向量(度/类型编码/中心度)做 KMeans区分 枢纽型/执行型/感知型/防护型│▼ TopoVisualizer (matplotlib networkx)可视化1. 工作站拓扑图颜色类型,大小中心度2. 安全联锁子图高亮3. 中心度柱状图4. 度分布直方图(scipy)│▼ SyntheticWorkcellGenerator (numpy)合成数据双机器人上下料站2机器人4工装3相机5安全门6料台含物料/信号/安全三类边3.3 为什么用图而不是表格* 问题Excel 表里写R1-料台T1看不出传递链更算不出门开了影响谁。* 处理策略图结构天然支持连通分量、路径、中心度是逻辑孪生的数学底座。* 工程合理性跟后续数字孪生平台对接时图可导出为节点表边表直接喂给孪生引擎。3.4 分析前后对比维度 电气图台账 本程序关系视图 分散多图 单张可计算拓扑枢纽识别 凭经验 degree/betweenness 量化安全影响域 人工反推 子图自动枚举节点角色 无 KMeans 聚类归类排故耗时 30~40min 秒级输出清单四、OOP 代码实现4.1 项目结构robot_cell_topo/├── robot_cell_topo/│ ├── __init__.py│ ├── topo_loader.py # 设备台账边表加载│ ├── topo_builder.py # networkx 建图│ ├── centrality_analyzer.py # 中心度分析│ ├── safety_impact.py # 安全影响域│ ├── role_clusterer.py # 节点角色聚类│ ├── visualizer.py # 可视化│ └── synthetic_workcell.py # 合成工作站├── tests/│ ├── __init__.py│ └── test_topo.py├── results/│ ├── workcell_topo.png # 全站拓扑图│ ├── safety_subgraph_S2.png # S2安全子图高亮│ ├── centrality_bar.png # 中心度柱状图│ ├── degree_hist.png # 度分布直方图│ ├── topo_nodes.csv # 节点属性│ ├── centrality_rank.csv # 中心度排名│ └── safety_impact_report.txt└── run_workcell_topo.py4.2 核心源码detailssummary/summary工作站设备台账与边表加载器import pandas as pdfrom pathlib import Pathfrom typing import Optional, Tupleclass StationTopoLoader:加载节点表与边表NODE_TYPES (robot, fixture, camera, safety_door, table)def __init__(self, node_path: str nodes.csv,edge_path: str edges.csv,encoding: str utf-8):self.node_path Path(node_path)self.edge_path Path(edge_path)self.encoding encodingself._nodes: Optional[pd.DataFrame] Noneself._edges: Optional[pd.DataFrame] Nonedef load(self) - Tuple[pd.DataFrame, pd.DataFrame]:if not self.node_path.exists() or not self.edge_path.exists():raise FileNotFoundError(nodes.csv 或 edges.csv 不存在)self._nodes pd.read_csv(self.node_path, encodingself.encoding)self._edges pd.read_csv(self.edge_path, encodingself.encoding)# 列名兼容nrename {}for tgt, al in {node_id: [node_id, id, 设备号],node_type: [node_type, type, 设备类型],x: [x, pos_x, X],y: [y, pos_y, Y],desc: [desc, 描述, name],}.items():if tgt not in self._nodes.columns:for a in al:if a in self._nodes.columns:nrename[a] tgtbreakself._nodes self._nodes.rename(columnsnrename)erename {}for tgt, al in {src: [src, source, 起点],dst: [dst, target, 终点],relation: [relation, rel, 关系],}.items():if tgt not in self._edges.columns:for a in al:if a in self._edges.columns:erename[a] tgtbreakself._edges self._edges.rename(columnserename)bad set(self._nodes[node_type]) - set(self.NODE_TYPES)if bad:raise ValueError(f未知节点类型: {bad})valid_rel {material, signal, safety}bad_rel set(self._edges[relation]) - valid_relif bad_rel:raise ValueError(f未知关系类型: {bad_rel})return self._nodes.copy(), self._edges.copy()def get_nodes(self) - Optional[pd.DataFrame]:return self._nodes.copy() if self._nodes is not None else None/detailsdetailssummary/summarynetworkx 拓扑构建器import networkx as nximport pandas as pdfrom typing import Optionalclass TopoBuilder:工作站拓扑图构建TYPE_COLOR {robot: #E74C3C,fixture: #3498DB,camera: #2ECC71,safety_door: #F39C12,table: #9B59B6,}def __init__(self):self.G nx.Graph()def build(self, nodes: pd.DataFrame, edges: pd.DataFrame) - nx.Graph:self.G.clear()for _, r in nodes.iterrows():self.G.add_node(r[node_id],node_typer[node_type],xfloat(r.get(x, 0)),yfloat(r.get(y, 0)),descstr(r.get(desc, )),)for _, r in edges.iterrows():if r[src] in self.G and r[dst] in self.G:# 同类型关系可多重这里用属性区分self.G.add_edge(r[src], r[dst],relationr[relation])return self.Gdef subgraph_by_relation(self, relation: str) - nx.Graph:sub nx.Graph()sub.add_nodes_from(self.G.nodes(dataTrue))for u, v, d in self.G.edges(dataTrue):if d[relation] relation:sub.add_edge(u, v, **d)return subdef color_map(self, node_list) - list:return [self.TYPE_COLOR.get(self.G.nodes[n][node_type], #999)for n in node_list]/detailsdetailssummary/summary中心度分析import numpy as npimport pandas as pdimport networkx as nxfrom typing import Dict, Optionalclass CentralityAnalyzer:节点枢纽度分析def __init__(self):passdef analyze(self, G: nx.Graph) - pd.DataFrame:deg nx.degree_centrality(G)bet nx.betweenness_centrality(G)clo nx.closeness_centrality(G)rows []for n, data in G.nodes(dataTrue):rows.append({node_id: n,node_type: data[node_type],degree: G.degree(n),degree_centrality: round(deg[n], 4),betweenness: round(bet[n], 4),closeness: round(clo[n], 4),})df pd.DataFrame(rows)return df.sort_values(degree_centrality, ascendingFalse).reset_index(dropTrue)def top_hubs(self, df: pd.DataFrame, top_n: int 5) - pd.DataFrame:return df.head(top_n).copy()def by_type(self, df: pd.DataFrame) - pd.DataFrame:g df.groupby(node_type).agg(n_nodes(node_id, count),mean_degree(degree, mean),max_degree_centrality(degree_centrality, max),).reset_index()for c in [mean_degree, max_degree_centrality]:g[c] g[c].round(4)return g.sort_values(max_degree_centrality, ascendingFalse).reset_index(dropTrue)/detailsdetailssummary/summary安全联锁影响域分析import networkx as nximport pandas as pdfrom typing import Dict, List, Optionalclass SafetyImpactAnalyzer:从安全门节点推导影响域def __init__(self):passdef _safety_subgraph(self, G: nx.Graph) - nx.Graph:sg nx.Graph()for u, v, d in G.edges(dataTrue):if d[relation] safety:sg.add_edge(u, v)for n in G.nodes():if G.nodes[n][node_type] safety_door:sg.add_node(n, **G.nodes[n])return sgdef impact_of(self, G: nx.Graph, door_id: str) - Dict:计算某安全门断开后的影响域Returns-------dict: {door_id, affected_nodes, affected_by_type, subgraph}if door_id not in G.nodes:raise KeyError(f节点不存在: {door_id})if G.nodes[door_id][node_type] ! safety_door:raise ValueError(f{door_id} 不是安全门)sg self._safety_subgraph(G)if door_id not in sg:return {door_id: door_id, affected_nodes: [],affected_by_type: {}, subgraph: nx.Graph()}comp nx.node_connected_component(sg, door_id)# 去掉自身affected [n for n in comp if n ! door_id]by_type {}for n in affected:t G.nodes[n][node_type]by_type[t] by_type.get(t, 0) 1sub sg.subgraph(comp).copy()return {door_id: door_id,affected_nodes: affected,affected_by_type: by_type,subgraph: sub,}def all_doors_impact(self, G: nx.Graph) - pd.DataFrame:doors [n for n, d in G.nodes(dataTrue)if d[node_type] safety_door]rows []for d in doors:res self.impact_of(G, d)rows.append({door_id: d,affected_count: len(res[affected_nodes]),affected_robots: res[affected_by_type].get(robot, 0),affected_tables: res[affected_by_type].get(table, 0),affected_fixtures: res[affected_by_type].get(fixture, 0),affected_nodes: ,.join(res[affected_nodes]),})return pd.DataFrame(rows).sort_values(affected_count,ascendingFalse).reset_index(dropTrue)/detailsdetailssummary/summary节点角色聚类import numpy as npimport pandas as pdfrom sklearn.cluster import KMeansfrom typing import Optionalclass RoleClusterer:基于中心度特征对节点聚类TYPE_CODE {robot: 0, fixture: 1, camera: 2,safety_door: 3, table: 4,}def __init__(self, n_clusters: int 4, random_state: int 42):self.n_clusters n_clustersself.random_state random_stateself.model Nonedef fit_predict(self, centrality_df: pd.DataFrame,G: nx.Graph) - pd.DataFrame:df centrality_df.copy()df[type_code] df[node_type].map(self.TYPE_CODE)X df[[degree, degree_centrality,betweenness, closeness, type_code]].valuesk min(self.n_clusters, len(df))self.model KMeans(n_clustersk, random_stateself.random_state, n_init10)labels self.model.fit_predict(X)df[cluster] labels# 按簇平均中心度排序给语义cmean df.groupby(cluster)[degree_centrality].mean().sort_values(ascendingFalse)name_map {cid: nm for cid, nm in zip(cmean.index,[枢纽型, 执行型, 感知型, 防护型, 辅助型][:k])}df[role_label] df[cluster].map(name_map)return dfdef describe(self, df: pd.DataFrame) - pd.DataFrame:desc df.groupby([cluster, role_label]).agg(n_nodes(node_id, count),mean_degree(degree, mean),mean_betweenness(betweenness, mean),).reset_index()for c in [mean_degree, mean_betweenness]:desc[c] desc[c].round(4)return desc/detailsdetailssummary/summary可视化import numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport networkx as nxfrom pathlib import Pathfrom scipy import statsplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass TopoVisualizer:工作站拓扑可视化def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)def plot_topo(self, G, centrality_df, highlight_subgraphNone):fig, ax plt.subplots(figsize(13, 9))pos {n: (G.nodes[n][x], G.nodes[n][y]) for n in G.nodes()}deg_c dict(zip(centrality_df[node_id],centrality_df[degree_centrality]))node_sizes [600 deg_c.get(n, 0) * 4000 for n in G.nodes()]builder TopoBuilder()node_colors builder.color_map(list(G.nodes()))# 边按关系分色edge_colors []for u, v, d in G.edges(dataTrue):edge_colors.append({material: #3498DB,signal: #2ECC71,safety: #E74C3C,}[d[relation]])if highlight_subgraph is not None:hs_nodes set(highlight_subgraph.nodes())node_colors [#E74C3C if n in hs_nodes else cfor n, c in zip(G.nodes(), node_colors)]nx.draw_networkx_nodes(G, pos, node_sizenode_sizes,node_colornode_colors, axax, alpha0.9)nx.draw_networkx_edges(G, pos, edge_coloredge_colors,width1.5, alpha0.5, axax)nx.draw_networkx_labels(G, pos, font_size8, axax)ax.set_title(机器人工作站拓扑图\n(大小中心度, 红安全链, 绿信号, 蓝物料),fontsize13, fontweightbold)ax.axis(off)plt.tight_layout()plt.savefig(self.results_dir / workcell_topo.png,dpi150, bbox_inchestight)plt.close()def plot_safety_subgraph(self, G, sub, door_id):fig, ax plt.subplots(figsize(10, 8))pos {n: (G.nodes[n][x], G.nodes[n][y]) for n in sub.nodes()if x in G.nodes[n]}nx.draw_networkx_nodes(sub, pos, node_size900,node_color#E74C3C, axax, alpha0.85)nx.draw_networkx_edges(sub, pos, edge_color#E74C3C,width2.5, axax)nx.draw_networkx_labels(sub, pos, font_size9, axax)ax.set_title(f安全门 {door_id} 联锁影响域,fontsize13, fontweightbold)ax.axis(off)plt.tight_layout()plt.savefig(self.results_dir / fsafety_subgraph_{door_id}.png,dpi150, bbox_inchestight)plt.close()def plot_centrality_bar(self, centrality_df):fig, ax plt.subplots(figsize(12, 6))s centrality_df.head(12)colors {robot: #E74C3C, fixture: #3498DB,camera: #2ECC71, safety_door: #F39C12,table: #9B59B6}bar_colors [colors[t] for t in s[node_type]]ax.bar(s[node_id], s[degree_centrality], colorbar_colors)ax.set_xlabel(节点, fontsize12)ax.set_ylabel(度中心度, fontsize12)ax.set_title(节点中心度排名 Top12, fontsize14, fontweightbold)ax.grid(axisy, alpha0.3)plt.tight_layout()plt.savefig(self.results_dir / centrality_bar.png,dpi150, bbox_inchestight)plt.close()def plot_degree_hist(self, G):fig, ax plt.subplots(figsize(10, 6))degrees [d for _, d in G.degree()]ax.hist(degrees, binsmax(3, max(degrees)), color#3498DB,alpha0.8, edgecolorwhite)mean_d np.mean(degrees)ax.axvline(mean_d, color#E74C3C, ls--,labelf均值 {mean_d:.2f})ax.set_xlabel(节点度, fontsize12)ax.set_ylabel(节点数, fontsize12)ax.set_title(工作站节点度分布直方图, fontsize14, fontweightbold)ax.legend()ax.grid(axisy, alpha0.3)plt.tight_layout()plt.savefig(self.results_dir / degree_hist.png,dpi150, bbox_inchestight)plt.close()/detailsdetailssummary/summary合成双机器人工作站数据import numpy as npimport pandas as pdfrom pathlib import Pathfrom typing import Optional, Tupleclass SyntheticWorkcellGenerator:双机器人上下料工作站2机器人 4工装 3相机 5安全门 6料台def __init__(self, rng: Optional[np.random.RandomState] None):self.rng rng or np.random.RandomState(42)def generate(self,output_nodes: str nodes.csv,output_edges: str edges.csv) - Tuple[pd.DataFrame, pd.DataFrame]:nodes []# 机器人robots [R1, R2]for i, r in enumerate(robots):nodes.append({node_id: r, node_type: robot,x: 5.0 i * 4, y: 5.0,desc: f六轴机器人{i1}})# 工装fixtures [fF{j1} for j in range(4)]for j, f in enumerate(fixtures):nodes.append({node_id: f, node_type: fixture,x: 4.0 j % 2 * 6, y: 3.0 (j // 2) * 4,desc: f工装{j1}})# 相机cameras [fC{k1} for k in range(3)]for k, c in enumerate(cameras):nodes.append({node_id: c, node_type: camera,x: 2.0 k * 5, y: 8.0,desc: f视觉相机{k1}})# 安全门doors [fS{d1} for d in range(5)]for d, s in enumerate(doors):nodes.append({node_id: s, node_type: safety_door,x: 0.5 d * 3, y: 0.5,desc: f安全门{d1}})# 料台利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛