企业官网建设流程全解析

周二下午，机器人工作站调试现场。

"又急停了，"现场电气老赵拍了拍安全门，"示教器上就显示'安全链路断开'，可我得花半小时翻电气图，才能知道这个安全门连的是哪台机器人、哪个料台、中间过了几个继电器。新来的实习生更懵，站里有 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解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！