周二上午，工艺准备室。
"这个叶轮的工艺路线又断了，"CAM 工程师老赵指着电脑屏幕，"零件模型在 UG 里，工艺方案写在本地的 Word 文档里，刀路文件在编程员的电脑上，机床程序拷在 U 盘里——四个东西散落在四个地方，中间靠人肉传。上次换了个新人，拿着旧版刀路去上机，结果刀具和工艺方案不匹配，撞刀了，主轴轴承干废一个，八千多。"
我看了眼他桌面上的文件：
叶轮_v3.prt
叶轮_工艺方案_v2.docx
叶轮_刀路_A3.nc
叶轮_机床程序_v1.nc
"这些文件之间有没有关联记录？"我问。
"没有，"老赵摇头，"全靠文件名和人的记忆。系统里不记录'这个刀路是用哪个工艺方案生成的'、'这个机床程序对应哪个版本的零件模型'。每次换人接手，或者工艺变更，就得翻聊天记录找——'上次用的是哪个版本？' '刀路谁编的？' '程序拷到哪台机床了？'"
"所以你需要的不是文件管理，是关系管理？"
"对，"老赵眼睛一亮，"我需要知道的是：从零件模型到最终机床程序，中间经过了哪些节点、每个节点产出了什么、节点之间是怎么连的。比如——零件模型 v3 → 工艺方案 v2 → 刀路 A3 → 机床程序 v1。这条链路要能画出来，一眼看清。而且如果中间某个节点变了——比如工艺方案升到 v3——下游哪些东西需要跟着重新生成，要能追溯。"
"明白了，"我打开 VS Code，"你需要一个程序：用节点表示 CAD 模型、工艺方案、刀路文件、机床程序，用有向边表示'生成关系'。零件模型 → 工艺方案（基于），工艺方案 → 刀路（生成），刀路 → 机床程序（后处理）。用 NetworkX 构建这个有向图，节点颜色区分类型，边标签标注关系，输出一张 CAD/CAM 工艺数据流图。"
我敲了几行代码：
import networkx as nx
G = nx.DiGraph()
G.add_node("叶轮_v3", type="CAD模型")
G.add_node("工艺方案_v2", type="工艺方案")
G.add_node("刀路_A3", type="刀路文件")
G.add_node("机床程序_v1", type="机床程序")
G.add_edge("叶轮_v3", "工艺方案_v2", relation="基于")
G.add_edge("工艺方案_v2", "刀路_A3", relation="生成")
G.add_edge("刀路_A3", "机床程序_v1", relation="后处理")
"这只是核心逻辑，"我说，"完整版会用 OOP 封装：一个类管节点定义（CAD/工艺/刀路/程序四种类型），一个类管边关系（基于/生成/后处理/验证），一个类管数据加载（从 CSV 或 JSON 读取工艺链路数据），一个类管图分析（追溯上游/下游、找孤立节点、检测循环依赖），一个类管可视化（节点按类型着色、边标签标注关系、子图高亮）。数据自包含——用 numpy 合成一份含 15+ 节点的 CAD/CAM 工艺数据流，模拟多零件、多版本的复杂链路，读者下载就能跑。"
老赵凑近屏幕："所以你这东西，就是把工艺准备过程中散落的文件和它们之间的关系，画成一张图——哪个模型生成了哪个刀路、哪个程序对应哪个工艺方案，一眼就看出来了。换版本的时候，顺着边往下游找，就知道哪些要重新生成。"
"对。而且还能做影响分析，"我补充，"比如用 NetworkX 的 ancestors() 和 descendants() 函数，输入一个节点，自动找出它的所有上游依赖和下游产物。工艺方案改了，descendants 告诉你哪些刀路和程序必须重做。CAD/CAM 的工艺数据管理不是'文件名带版本号'，是'用图结构管理数据血缘，让每一次变更都可追溯'。"
一、实际应用场景（真实痛点）
场景设定：CAD/CAM 工艺准备流程中，从零件三维模型到最终机床 G 代码，经历多个阶段：CAD 建模 → 工艺方案制定 → CAM 刀路编程 → 后处理生成机床程序。每个阶段产出一种数据文件，文件之间存在着明确的"生成"或"基于"关系。实际车间中，这些文件分散存储，关系仅靠工程师个人记忆或文件名约定维护。当工艺变更（如刀具更换、模型修订）时，无法快速追溯"哪些下游文件需要重新生成"，导致使用过期文件上机的风险。
现场原话（叙事化）：
"我不是不想管版本，"老赵说，"每个文件我都加了版本号——v1、v2、v3。但版本号管的是单个文件，管不了文件之间的关系。叶轮_v3.prt 改了，我怎么知道哪些刀路是基于旧版模型编的？工艺方案从 v1 改到 v3，哪些机床程序还是用 v1 方案生成的？系统里没有'血缘关系图'——我只看到一堆带版本号的文件，不知道谁依赖谁。"
核心矛盾："分散存储的工艺文件"与"工程师需要的上下游关系追溯"之间的管理断层。需要一个"CAD/CAM 工艺数据流图程序"，用
"networkx" 构建有向图管理节点关系，用
"matplotlib" 可视化工艺链路，用
"pandas" 管理节点属性表，用
"numpy" 计算节点中心性指标，输出结构化的数据流图和变更影响报告。
二、痛点分析（映射到滨州职业学院《先进制造技术》课程模型）
《先进制造技术》模块 本篇痛点对应
数控加工与CAD/CAM技术：CAD/CAM 集成与数据流转 工艺数据流建模：从三维模型到 G 代码的全链路关系管理。
先进制造技术基础：先进制造信息流 数据血缘追溯：工艺变更时的下游影响分析。
智能制造与数字孪生：工艺数字孪生 图结构可视化：用节点-边模型映射物理工艺准备流程。
先进制造新模式：协同设计与并行工程 依赖关系管理：多角色协作时的数据一致性保障。
一句话总结：我们需要构建一个"CAD/CAM 工艺数据流图程序"，用
"networkx" 构建有向图管理 CAD 模型→工艺方案→刀路→机床程序的生成关系，用
"matplotlib" 绘制可视化数据流，用
"pandas" 管理节点属性，用
"numpy" 计算节点中心性，实现从文件堆到可追溯工艺链路的转化。
三、核心逻辑讲解（大白话）
3.1 问题本质：把工艺数据想象成"家谱树"
把 CAD/CAM 的工艺文件关系想象成"一个大家族的家谱"：
* CAD 模型 = 老祖宗：所有东西都从它派生出来。
* 工艺方案 = 第二代：基于 CAD 模型制定，告诉后面的人"怎么加工这个零件"。
* 刀路文件 = 第三代：工艺方案说"用球头刀走曲面"，刀路就是具体的走刀路径。
* 机床程序 = 第四代：刀路经过后处理，变成机床能读的 G 代码。
* 有向边 = "生父子关系"：CAD 模型 → 工艺方案（基于），工艺方案 → 刀路（生成），刀路 → 机床程序（后处理）。
* 变更影响 = "一人改，全家动"：老祖宗改了（模型升版），所有后代都可能要跟着改——用 NetworkX 的
"descendants()" 一键查出"哪些后代受影响"。
工业应用：
* networkx.DiGraph：
"G = nx.DiGraph()" 创建一个有向图，节点是文件，边是"生成关系"。
* 节点类型着色：
"node_type" 属性决定颜色——CAD 模型蓝色、工艺方案橙色、刀路文件绿色、机床程序红色。
* 中心性分析：
"nx.betweenness_centrality()" 找出"最关键的中间节点"——哪个文件被引用最多，它就是工艺链路的瓶颈。
3.2 业务逻辑 → 代码映射
定义 CAD/CAM 工艺数据模型
│
▼ NodeDefinition (枚举)
节点类型：
CAD_MODEL = "CAD模型"
PROCESS_PLAN = "工艺方案"
TOOLPATH = "刀路文件"
MACHINE_PROGRAM = "机床程序"
│
▼ EdgeDefinition
边关系：
BASED_ON = "基于" # CAD → 工艺
GENERATES = "生成" # 工艺 → 刀路
POST_PROCESSES = "后处理" # 刀路 → 程序
VERIFIES = "验证" # 程序 → 仿真
│
▼ ProcessDataLoader (pandas)
加载节点和边数据：
nodes.csv: node_id, node_type, name, version, owner
edges.csv: source, target, relation
│
▼ ProcessGraphBuilder (networkx)
构建有向图：
nx.DiGraph()
add_node() with attributes
add_edge() with relation label
│
▼ GraphAnalyzer (networkx + numpy)
图分析：
ancestors(node) → 上游追溯
descendants(node) → 下游影响
betweenness_centrality() → 关键节点
is_directed_acyclic() → 检测循环依赖
│
▼ ProcessVisualizer (matplotlib + networkx)
可视化：
1. 工艺数据流有向图（节点着色、边标签）
2. 节点类型饼图
3. 关键路径高亮子图
4. 变更影响范围图
│
▼ SyntheticDataGenerator (numpy)
合成数据生成：
模拟 4 种零件 × 多版本的工艺链路
3.3 为什么用图结构而不是文件夹？
* 问题：文件夹只能表达"谁和谁放在一起"，不能表达"谁依赖谁"。
* 处理策略：有向图天然表达"生成关系"，
"ancestors()" 和
"descendants()" 自动追溯上下游。
* 工程合理性：复杂零件（叶轮、叶片）的工艺链路可能有数十个节点、上百条边，图结构是唯一能完整表达多对多依赖的方式。
3.4 分析前后对比
维度 文件夹 + 文件名版本号 图结构数据流
关系表达 无 有向边明确标注
变更追溯 翻聊天记录问人
"descendants()" 一键查下游
孤立文件检测 不知道
"nx.isolates()" 自动发现
循环依赖 不可能发现
"nx.find_cycle()" 自动检测
关键瓶颈 凭经验猜 中心性指标量化
四、OOP 代码实现
4.1 项目结构
cam_process_graph/
├── cam_process_graph/
│ ├── __init__.py
│ ├── node_definition.py # 节点类型定义
│ ├── edge_definition.py # 边关系定义
│ ├── data_loader.py # 数据加载
│ ├── graph_builder.py # 图构建
│ ├── graph_analyzer.py # 图分析
│ ├── visualizer.py # 可视化
│ └── synthetic_data.py # 合成数据生成
├── tests/
│ ├── __init__.py
│ └── test_process_graph.py # 单元测试
├── results/
│ ├── process_dataflow.png # 工艺数据流图
│ ├── node_type_pie.png # 节点类型分布
│ ├── critical_path.png # 关键路径高亮
│ ├── impact_analysis.png # 变更影响范围
│ ├── graph_report.txt # 文本分析报告
│ └── graph_export.gexf # 图数据导出
└── run_process_graph.py # 主程序入口
4.2 核心源码
<details>
<summary></summary>
"""CAD/CAM 工艺节点类型定义"""
from enum import Enum
from typing import Dict
class NodeType(Enum):
"""工艺节点类型"""
CAD_MODEL = "CAD模型"
PROCESS_PLAN = "工艺方案"
TOOLPATH = "刀路文件"
MACHINE_PROGRAM = "机床程序"
SIMULATION = "仿真验证"
@classmethod
def get_color_map(cls) -> Dict[str, str]:
"""节点类型到颜色的映射"""
return {
cls.CAD_MODEL.value: "#3498DB", # 蓝
cls.PROCESS_PLAN.value: "#E67E22", # 橙
cls.TOOLPATH.value: "#2ECC71", # 绿
cls.MACHINE_PROGRAM.value: "#E74C3C", # 红
cls.SIMULATION.value: "#9B59B6", # 紫
}
@classmethod
def get_short_label(cls) -> Dict[str, str]:
"""短标签"""
return {
cls.CAD_MODEL.value: "CAD",
cls.PROCESS_PLAN.value: "CAPP",
cls.TOOLPATH.value: "CAM",
cls.MACHINE_PROGRAM.value: "NC",
cls.SIMULATION.value: "SIM",
}
</details>
<details>
<summary></summary>
"""CAD/CAM 工艺边关系定义"""
from enum import Enum
from typing import Dict
class EdgeRelation(Enum):
"""边关系类型"""
BASED_ON = "基于"
GENERATES = "生成"
POST_PROCESSES = "后处理"
VERIFIES = "验证"
SUPERSEDES = "替代"
@classmethod
def get_valid_edges(cls) -> Dict[str, str]:
"""
合法的边关系（source_type → target_type: relation）
规则：
CAD模型 → 工艺方案: 基于
工艺方案 → 刀路文件: 生成
刀路文件 → 机床程序: 后处理
机床程序 → 仿真验证: 验证
"""
return {
("CAD模型", "工艺方案"): cls.BASED_ON.value,
("工艺方案", "刀路文件"): cls.GENERATES.value,
("刀路文件", "机床程序"): cls.POST_PROCESSES.value,
("机床程序", "仿真验证"): cls.VERIFIES.value,
}
</details>
<details>
<summary></summary>
"""工艺数据加载器"""
import pandas as pd
from pathlib import Path
from typing import Optional, Tuple
class ProcessDataLoader:
"""
CAD/CAM 工艺数据加载器
支持 CSV 格式的节点和边数据。
"""
def __init__(self, encoding: str = "utf-8"):
self.encoding = encoding
def load_nodes(self, filepath: str) -> pd.DataFrame:
"""
加载节点数据
CSV 格式：node_id, node_type, name, version, owner
"""
path = Path(filepath)
if not path.exists():
raise FileNotFoundError(f"节点文件不存在: {path}")
df = pd.read_csv(path, encoding=self.encoding)
return df
def load_edges(self, filepath: str) -> pd.DataFrame:
"""
加载边数据
CSV 格式：source, target, relation
"""
path = Path(filepath)
if not path.exists():
raise FileNotFoundError(f"边文件不存在: {path}")
df = pd.read_csv(path, encoding=self.encoding)
return df
def load_from_dicts(self,
nodes: list,
edges: list) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""从字典列表加载"""
nodes_df = pd.DataFrame(nodes)
edges_df = pd.DataFrame(edges)
return nodes_df, edges_df
</details>
<details>
<summary></summary>
"""工艺数据流图构建器"""
import networkx as nx
import pandas as pd
from typing import Dict, List, Optional
from cam_process_graph.node_definition import NodeType
from cam_process_graph.edge_definition import EdgeRelation
class ProcessGraphBuilder:
"""
CAD/CAM 工艺数据流有向图构建器
"""
def __init__(self):
self.graph = nx.DiGraph()
def build_from_dataframes(self,
nodes_df: pd.DataFrame,
edges_df: pd.DataFrame) -> nx.DiGraph:
"""
从 DataFrame 构建图
Parameters
----------
nodes_df : pd.DataFrame
节点数据，含 node_id, node_type, name 等列
edges_df : pd.DataFrame
边数据，含 source, target, relation 列
Returns
-------
nx.DiGraph
"""
self.graph.clear()
# 添加节点
for _, row in nodes_df.iterrows():
node_attrs = row.to_dict()
node_id = node_attrs.pop("node_id", None)
if node_id is not None:
self.graph.add_node(node_id, **node_attrs)
# 添加边
for _, row in edges_df.iterrows():
source = row.get("source")
target = row.get("target")
relation = row.get("relation", "")
if source and target:
self.graph.add_edge(source, target, relation=relation)
return self.graph
def add_node(self, node_id: str, **attrs) -> None:
"""添加单个节点"""
self.graph.add_node(node_id, **attrs)
def add_edge(self, source: str, target: str, relation: str = "") -> None:
"""添加单条边"""
self.graph.add_edge(source, target, relation=relation)
def get_graph(self) -> nx.DiGraph:
return self.graph
def validate_edges(self) -> List[str]:
"""
验证边的合法性
Returns
-------
list
不合法边的描述列表
"""
invalid = []
valid_edges = EdgeRelation.get_valid_edges()
for u, v, data in self.graph.edges(data=True):
u_type = self.graph.nodes[u].get("node_type", "")
v_type = self.graph.nodes[v].get("node_type", "")
relation = data.get("relation", "")
key = (u_type, v_type)
if key in valid_edges:
expected_relation = valid_edges[key]
if relation != expected_relation:
invalid.append(
f"{u} → {v}: 关系应为'{expected_relation}'，实际为'{relation}'"
)
else:
invalid.append(
f"{u} → {v}: 类型组合 '{u_type} → {v_type}' 不在合法边列表中"
)
return invalid
</details>
<details>
<summary></summary>
"""工艺数据流图分析器"""
import networkx as nx
import numpy as np
from typing import Dict, List, Set, Optional
class GraphAnalyzer:
"""
CAD/CAM 工艺数据流图分析器
提供追溯、影响分析、关键节点识别等功能。
"""
def __init__(self, graph: nx.DiGraph):
self.graph = graph
def get_upstream(self, node_id: str) -> Set[str]:
"""获取所有上游节点（祖先）"""
if node_id not in self.graph:
return set()
return nx.ancestors(self.graph, node_id)
def get_downstream(self, node_id: str) -> Set[str]:
"""获取所有下游节点（后代）"""
if node_id not in self.graph:
return set()
return nx.descendants(self.graph, node_id)
def get_direct_upstream(self, node_id: str) -> List[str]:
"""获取直接上游（前驱）"""
if node_id not in self.graph:
return []
return list(self.graph.predecessors(node_id))
def get_direct_downstream(self, node_id: str) -> List[str]:
"""获取直接下游（后继）"""
if node_id not in self.graph:
return []
return list(self.graph.successors(node_id))
def find_isolates(self) -> List[str]:
"""查找孤立节点"""
return list(nx.isolates(self.graph))
def detect_cycles(self) -> List:
"""检测循环依赖"""
try:
cycles = list(nx.find_cycle(self.graph, orientation="original"))
return cycles
except nx.NetworkXNoCycle:
return []
def is_dag(self) -> bool:
"""检查是否为有向无环图"""
return nx.is_directed_acyclic_graph(self.graph)
def betweenness_centrality(self) -> Dict[str, float]:
"""计算介数中心性"""
return nx.betweenness_centrality(self.graph)
def degree_centrality(self) -> Dict[str, float]:
"""计算度中心性"""
return nx.degree_centrality(self.graph)
def get_critical_nodes(self, top_k: int = 5) -> List[tuple]:
"""获取最关键的前 K 个节点"""
bc = self.betweenness_centrality()
sorted_nodes = sorted(bc.items(), key=lambda x: x[1], reverse=True)
return sorted_nodes[:top_k]
def impact_analysis(self, changed_node: str) -> Dict:
"""
变更影响分析
Parameters
----------
changed_node : str
发生变更的节点 ID
Returns
-------
dict
影响分析结果
"""
downstream = self.get_downstream(changed_node)
direct_downstream = self.get_direct_downstream(changed_node)
# 按类型统计下游
type_counts = {}
for node in downstream:
node_type = self.graph.nodes[node].get("node_type", "未知")
type_counts[node_type] = type_counts.get(node_type, 0) + 1
return {
"changed_node": changed_node,
"total_downstream": len(downstream),
"direct_downstream": len(direct_downstream),
"downstream_nodes": list(downstream),
"direct_downstream_nodes": direct_downstream,
"type_distribution": type_counts,
}
def get_paths_between(self, source: str, target: str) -> List[List[str]]:
"""获取两个节点之间的所有路径"""
if source not in self.graph or target not in self.graph:
return []
try:
paths = list(nx.all_simple_paths(self.graph, source, target))
return paths
except nx.NetworkXNoPath:
return []
</details>
<details>
<summary></summary>
"""工艺数据流可视化"""
import numpy as np
import matplotlib.pyplot as plt
import networkx as nx
from pathlib import Path
from typing import Optional, List
plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
from cam_process_graph.node_definition import NodeType
class ProcessVisualizer:
"""CAD/CAM 工艺数据流可视化"""
def __init__(self, results_dir: str = "results"):
self.results_dir = Path(results_dir)
self.results_dir.mkdir(exist_ok=True)
self.color_map = NodeType.get_color_map()
def plot_dataflow(self,
graph: nx.DiGraph,
title: str = "CAD/CAM 工艺数据流图",
layout: str = "hierarchical",
highlight_nodes: Optional[List[str]] = None) -> None:
"""
绘制工艺数据流有向图
Parameters
----------
graph : nx.DiGraph
工艺数据流图
title : str
标题
layout : str
布局方式 ("hierarchical", "spring", "circular")
highlight_nodes : list, optional
需要高亮的节点列表
"""
fig, ax = plt.subplots(figsize=(18, 12))
# 布局
if layout == "hierarchical":
# 按节点类型分层
layers = {}
for i, ntype in enumerate(NodeType):
nodes_of_type = [
n for n, d in graph.nodes(data=True)
if d.get("node_type") == ntype.value
]
for j, node in enumerate(nodes_of_type):
layers[node] = (i, -j)
if len(layers) < len(graph.nodes):
pos = nx.spring_layout(graph, k=3, iterations=50, seed=42)
else:
pos = layers
elif layout == "spring":
pos = nx.spring_layout(graph, k=3, iterations=50, seed=42)
else:
pos = nx.circular_layout(graph)
# 节点颜色
node_colors = []
for node in graph.nodes():
node_type = graph.nodes[node].get("node_type", "")
color = self.color_map.get(node_type, "#BDC3C7")
if highlight_nodes and node in highlight_nodes:
color = "#F1C40F" # 高亮为黄色
node_colors.append(color)
# 节点大小（基于度中心性）
degrees = dict(graph.degree())
max_degree = max(degrees.values()) if degrees else 1
node_sizes = [500 + 1500 * (degrees.get(n, 0) / max_degree)
for n in graph.nodes()]
# 绘制节点
nx.draw_networkx_nodes(
graph, pos, ax=ax,
node_color=node_colors,
node_size=node_sizes,
alpha=0.9,
edgecolors="white",
linewidths=1.5,
)
# 绘制边
nx.draw_networkx_edges(
graph, pos, ax=ax,
edge_color="#7F8C8D",
width=1.5,
arrows=True,
arrowsize=15,
arrowstyle="-|>",
connectionstyle="arc3,rad=0.15",
)
# 边标签
edge_labels = nx.get_edge_attributes(graph, "relation")
nx.draw_networkx_edge_labels(
graph, pos, ax=ax,
edge_labels=edge_labels,
font_size=8,
font_color="#2C3E50",
)
# 节点标签
labels = {}
for node in graph.nodes():
name = graph.nodes[node].get("name", node)
version = graph.nodes[node].get("version", "")
if version:
labels[node] = f"{name}\n(v{version})"
else:
labels[node] = name
nx.draw_networkx_labels(
graph, pos, ax=ax,
labels=labels,
font_size=7,
font_weight="bold",
font_color="white",
)
# 图例
from matplotlib.patches import Patch
legend_elements = [
Patch(facecolor=color, label=ntype)
for ntype, color in self.color_map.items()
]
legend_elements.append(Patch(facecolor="#F1C40F", label="高亮/选中"))
ax.legend(handles=legend_elements, loc="upper left", fontsize=9)
ax.set_title(title, fontsize=16, fontweight="bold", pad=20)
ax.axis("off")
plt.tight_layout()
plt.savefig(self.results_dir / "process_dataflow.png",
dpi=150, bbox_inches="tight")
plt.close()
def plot_node_type_pie(self, graph: nx.DiGraph) -> None:
"""绘制节点类型分布饼图"""
type_counts = {}
for _, data in graph.nodes(data=True):
ntype = data.get("node_type", "未知")
type_counts[ntype] = type_counts.get(ntype, 0) + 1
fig, ax = plt.subplots(figsize=(8, 8))
labels = list(type_counts.keys())
sizes = list(type_counts.values())
colors = [self.color_map.get(l, "#BDC3C7") for l in labels]
ax.pie(sizes, labels=labels, colors=colors, autopct="%1.1f%%",
shadow=False, startangle=90, textprops={"fontsize": 10})
ax.set_title("节点类型分布", fontsize=14, fontweight="bold")
plt.tight_layout()
plt.savefig(self.results_dir / "node_type_pie.png",
dpi=150, bbox_inches="tight")
plt.close()
def plot_impact_analysis(self,
graph: nx.DiGraph,
impact_result: dict) -> None:
"""绘制变更影响范围图"""
changed_node = impact_result["changed_node"]
downstream = set(impact_result["downstream_nodes"])
fig, ax = plt.subplots(figsize=(18, 12))
pos = nx.spring_layout(graph, k=3, iterations=50, seed=42)
# 节点颜色
node_colors = []
for
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
