python的先进制造技术工业场景模拟第二十三篇:使用Networkx搭建机器人工作站拓扑，节点包含机器人，工装，相机，安全门，料台。
周二下午，机器人工作站调试现场。
"又急停了，"现场电气老赵拍了拍安全门，"示教器上就显示'安全链路断开'，可我得花半小时翻电气图，才能知道这个安全门连的是哪台机器人、哪个料台、中间过了几个继电器。新来的实习生更懵，站里有 2 台机器人、4 套工装、3 个相机、5 道安全门、6 个料台，全靠线槽里那把线认关系。"
我打开运维文档。
"这图里有什么？"老赵问。
"现在是几张 PDF 电气图 + 一张布局草图，"我指着屏幕，"机器人、工装夹具、视觉相机、安全门、料台，都是散的。想看'相机拍完→给哪个机器人信号→机器人取哪个料台'，得人脑拼。"
"我就想要一张'关系图'，"老赵说，"像流程图那样，节点是设备，连线是信号/物料/安全联锁关系。最好能标出哪个节点是枢纽，哪条链路断了会连带急停，排故障不用翻三本图册。"
"所以你要的不是电气原理图，是'工作站拓扑图 + 关键路径识别 + 联锁影响域分析'？"
"对，"老赵点头，"比如安全门 S2 断开，影响哪几台机器人、哪几个料台，能不能一键算出来。还有相机 C1 挂了，是不是只影响机器人 R1 的来料定位。"
"明白了，"我开 VS Code，"用 networkx 把工作站建成有向+无向混合图：机器人、工装、相机、安全门、料台都是节点，边分三类——物料流、信号流、安全联锁。用节点中心度找枢纽，用子图遍历算安全门影响域，用 matplotlib 画拓扑图，用 pandas 管设备台账，用 scipy 做度分布统计，用 scikit-learn 对节点角色聚类。数据自包含——合成一个双机器人上下料工作站，下载就能跑。"
我敲了段原型：
import networkx as nx
G = nx.Graph()
G.add_node("R1", type="robot")
G.add_node("S2", type="safety_door")
G.add_edge("S2", "R1", relation="safety_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)
导入 CSV：
node_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.py
4.2 核心源码
<details>
<summary></summary>
"""工作站设备台账与边表加载器"""
import pandas as pd
from pathlib import Path
from typing import Optional, Tuple
class 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 = encoding
self._nodes: Optional[pd.DataFrame] = None
self._edges: Optional[pd.DataFrame] = None
def 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, encoding=self.encoding)
self._edges = pd.read_csv(self.edge_path, encoding=self.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] = tgt
break
self._nodes = self._nodes.rename(columns=nrename)
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] = tgt
break
self._edges = self._edges.rename(columns=erename)
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_rel
if 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
</details>
<details>
<summary></summary>
"""networkx 拓扑构建器"""
import networkx as nx
import pandas as pd
from typing import Optional
class 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_type=r["node_type"],
x=float(r.get("x", 0)),
y=float(r.get("y", 0)),
desc=str(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"],
relation=r["relation"])
return self.G
def subgraph_by_relation(self, relation: str) -> nx.Graph:
sub = nx.Graph()
sub.add_nodes_from(self.G.nodes(data=True))
for u, v, d in self.G.edges(data=True):
if d["relation"] == relation:
sub.add_edge(u, v, **d)
return sub
def color_map(self, node_list) -> list:
return [self.TYPE_COLOR.get(self.G.nodes[n]["node_type"], "#999")
for n in node_list]
</details>
<details>
<summary></summary>
"""中心度分析"""
import numpy as np
import pandas as pd
import networkx as nx
from typing import Dict, Optional
class CentralityAnalyzer:
"""节点枢纽度分析"""
def __init__(self):
pass
def 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(data=True):
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", ascending=False).reset_index(drop=True)
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", ascending=False).reset_index(drop=True)
</details>
<details>
<summary></summary>
"""安全联锁影响域分析"""
import networkx as nx
import pandas as pd
from typing import Dict, List, Optional
class SafetyImpactAnalyzer:
"""从安全门节点推导影响域"""
def __init__(self):
pass
def _safety_subgraph(self, G: nx.Graph) -> nx.Graph:
sg = nx.Graph()
for u, v, d in G.edges(data=True):
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 sg
def 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) + 1
sub = 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(data=True)
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",
ascending=False).reset_index(drop=True)
</details>
<details>
<summary></summary>
"""节点角色聚类"""
import numpy as np
import pandas as pd
from sklearn.cluster import KMeans
from typing import Optional
class 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_clusters
self.random_state = random_state
self.model = None
def 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"]].values
k = min(self.n_clusters, len(df))
self.model = KMeans(n_clusters=k, random_state=self.random_state, n_init=10)
labels = self.model.fit_predict(X)
df["cluster"] = labels
# 按簇平均中心度排序给语义
cmean = df.groupby("cluster")["degree_centrality"].mean().sort_values(ascending=False)
name_map = {cid: nm for cid, nm in zip(
cmean.index,
["枢纽型", "执行型", "感知型", "防护型", "辅助型"][:k]
)}
df["role_label"] = df["cluster"].map(name_map)
return df
def 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
</details>
<details>
<summary></summary>
"""可视化"""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import networkx as nx
from pathlib import Path
from scipy import stats
plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
class TopoVisualizer:
"""工作站拓扑可视化"""
def __init__(self, results_dir: str = "results"):
self.results_dir = Path(results_dir)
self.results_dir.mkdir(exist_ok=True)
def plot_topo(self, G, centrality_df, highlight_subgraph=None):
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(data=True):
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 c
for n, c in zip(G.nodes(), node_colors)]
nx.draw_networkx_nodes(G, pos, node_size=node_sizes,
node_color=node_colors, ax=ax, alpha=0.9)
nx.draw_networkx_edges(G, pos, edge_color=edge_colors,
width=1.5, alpha=0.5, ax=ax)
nx.draw_networkx_labels(G, pos, font_size=8, ax=ax)
ax.set_title("机器人工作站拓扑图\n(大小=中心度, 红=安全链, 绿=信号, 蓝=物料)",
fontsize=13, fontweight="bold")
ax.axis("off")
plt.tight_layout()
plt.savefig(self.results_dir / "workcell_topo.png",
dpi=150, bbox_inches="tight")
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_size=900,
node_color="#E74C3C", ax=ax, alpha=0.85)
nx.draw_networkx_edges(sub, pos, edge_color="#E74C3C",
width=2.5, ax=ax)
nx.draw_networkx_labels(sub, pos, font_size=9, ax=ax)
ax.set_title(f"安全门 {door_id} 联锁影响域",
fontsize=13, fontweight="bold")
ax.axis("off")
plt.tight_layout()
plt.savefig(self.results_dir / f"safety_subgraph_{door_id}.png",
dpi=150, bbox_inches="tight")
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"], color=bar_colors)
ax.set_xlabel("节点", fontsize=12)
ax.set_ylabel("度中心度", fontsize=12)
ax.set_title("节点中心度排名 Top12", fontsize=14, fontweight="bold")
ax.grid(axis="y", alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir / "centrality_bar.png",
dpi=150, bbox_inches="tight")
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, bins=max(3, max(degrees)), color="#3498DB",
alpha=0.8, edgecolor="white")
mean_d = np.mean(degrees)
ax.axvline(mean_d, color="#E74C3C", ls="--",
label=f"均值 {mean_d:.2f}")
ax.set_xlabel("节点度", fontsize=12)
ax.set_ylabel("节点数", fontsize=12)
ax.set_title("工作站节点度分布直方图", fontsize=14, fontweight="bold")
ax.legend()
ax.grid(axis="y", alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir / "degree_hist.png",
dpi=150, bbox_inches="tight")
plt.close()
</details>
<details>
<summary></summary>
"""合成双机器人工作站数据"""
import numpy as np
import pandas as pd
from pathlib import Path
from typing import Optional, Tuple
class 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"六轴机器人{i+1}"})
# 工装
fixtures = [f"F{j+1}" 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"工装{j+1}"})
# 相机
cameras = [f"C{k+1}" 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"视觉相机{k+1}"})
# 安全门
doors = [f"S{d+1}" 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"安全门{d+1}"})
# 料台
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
1
0- 0
已为社区贡献31条内容
所有评论(0)