网硕互联帮助中心周三夜班，机加车间。
"这批件干到 180 件，刀又崩了，"刀具管理员老周把几根硬质合金刀摆桌上，"系统里记的是'换刀事件'，可我想知道：同牌号的刀，干这个材料到底能撑多少件？是 150 件就到头，还是能到 220？现在全靠老师傅拍脑袋，换早了浪费刀，换晚了出废品。"
我打开 MES 的导出表。
"这表里有什么？"老周问。
"每把刀一个编号，加工零件时记开始件号、结束件号、加工材料、切削参数、是否崩刃，"我指着 CSV，"一把刀一行，或者按寿命事件一行。可系统不聚合'同型号刀平均能加工几件'，也不告诉你是后刀面磨损先到，还是崩刃先到。"
"我就想问一句，"老周说，"WC-Co 细晶的、超细晶的、涂层 PVD 的，各统计一下平均寿命件数，画个分布，再告诉我哪批刀寿命离散大，哪批稳。最好能按加工材料分开看，钢件和不锈钢不能混着算。"
"所以你要的不是换刀记录表，是'同型号刀具耐用度统计 + 寿命分布 + 离散度评估 + 异常刀识别'？"
"对，"老周点头，"比如同是 PVD 涂层刀干 45 钢，平均 195 件，但有一把只干了 80 件就崩了，我想把它挑出来，看是不是那台机床主轴跳动了。"
"明白了，"我开 VS Code，"用 pandas 读刀具加工记录，numpy 算寿命件数，scipy 做分布检验和置信区间，matplotlib 画寿命箱线图+威布尔图，scikit-learn 聚类找异常刀，networkx 把'刀具型号→加工材料→机床'建关系网看瓶颈。数据自包含——合成 5 类硬质合金刀×多把，模拟正常磨损与少量早损，下载就能跑。"
我敲了段原型：
import pandas as pd
life = df.groupby("tool_id").apply(
lambda g: g["end_part"].max() – g["start_part"].min() + 1)
"完整版用 OOP 封好，"我说，"一个类管刀具记录加载，一个类算单刀寿命，一个类做型号级统计，一个类做威布尔分析，一个类找异常刀，一个类画关系网，一个类出图。输出每种刀平均件数、寿命分布、早损清单、威布尔斜率。"
老周凑近看："那我以后看报告：PVD 涂层干 45 钢，均值 195 件，威布尔斜率 8.2，说明寿命挺集中；有一把 82 件崩刃，标红，关联机床 M3，就知道去查 M3 主轴了。"
"对，"我接话，"刀具耐用度不是标称值，是统计出来的分布。数字孪生里刀是消耗品模型，得先有这张统计底表，才能做换刀预测。"
一、实际应用场景（真实痛点）
场景设定：多型号硬质合金刀具（不同牌号/涂层）加工不同材料零件，MES 记录每把刀的启用件号、停用件号、加工材料、机床号、停用原因（正常磨损/崩刃/崩角/达到寿命）。现场只记录事件，不提供同型号刀具寿命聚合统计，也不区分材料工况，导致换刀策略全凭经验。
现场原话（叙事化）：
"我不是舍不得换刀，"老周说，"是没数。同是 PVD 涂层刀，有的干 210 件，有的 160 件就崩，系统里就是两条换刀记录，没人帮我算均值。有次我按 150 件换，结果后面刀都还能用，一年白扔几百根；另一次按 220 件换，出了批振纹废品。后来我把每把刀寿命手记在本子上，可本子跟 MES 对不上，也分不了材料。"
核心矛盾："换刀事件流水"与"同型号刀具耐用度统计 + 寿命分布 + 威布尔评估 + 早损异常识别 + 工况归因"之间的断层。需要一个"硬质合金刀具耐用度统计与评估程序"，用
"pandas" 聚合、
"numpy" 算寿命、
"scipy" 做分布拟合与检验、
"matplotlib" 画箱线图/威布尔图、
"scikit-learn" 聚类异常刀、
"networkx" 建工况关系网。
二、痛点分析（映射到滨州职业学院《先进制造技术》课程模型）
《先进制造技术》模块 本篇痛点对应
数控加工与CAD/CAM技术：刀具材料与切削参数优化 硬质合金牌号/涂层耐用度量化统计
先进制造技术基础：制造过程资源消耗管理 刀具作为关键消耗资源寿命建模
FMS与先进生产管理：单元级刀具管理与换刀策略 按型号聚合寿命，支撑换刀阈值设定
智能制造与数字孪生：刀具数字孪生 寿命分布→可预测换刀模型底座
先进制造新模式：成本与质量协同（零缺陷） 早损识别避免批量振纹废品
一句话总结：我们需要构建一个"硬质合金刀具耐用度统计与评估程序"，用
"pandas" 按型号聚合，
"scipy" 做威布尔拟合，
"scikit-learn" 找早损异常，
"networkx" 关联工况，实现从"换刀流水账"到"可统计、可预测、可归因的刀具寿命模型"。
三、核心逻辑讲解（大白话）
3.1 问题本质：把刀具想成"灯泡寿命"
把每把刀想成一个灯泡：
* 刀具型号 = 灯泡品牌（细晶 WC-Co / 超细晶 / PVD涂层…）
* 加工件数 = 亮了多少小时
* 停用原因 = 怎么灭的（自然暗下去=正常磨损，炸了=崩刃）
* 同型号多把刀 = 同一品牌一堆灯泡，有的亮久点有的短点
* 平均件数 = 品牌标称寿命
* 威布尔斜率 = 这批灯泡"齐不齐心"：斜率大（>5）说明寿命很集中，斜率小说明忽长忽短
* 早损刀 = 同批里早早炸掉的那个灯泡，得拎出来看是不是装错机床了
* 工况关系网 = 哪种刀在哪种材料哪台机床上最费
工业应用：
*
"pandas.groupby(["tool_grade","material"])" 聚合寿命件数
*
"scipy.stats.weibull_min" 拟合寿命分布，取形状参数 k
*
"scikit-learn" 用寿命+斜率偏差做早损聚类
*
"networkx" 建
"刀具型号—材料—机床" 三层图，找高消耗路径
3.2 业务逻辑 → 代码映射
定义刀具加工记录模型
│
▼ ToolLogLoader (pandas)
导入 CSV：
tool_id, tool_grade, coating, material, machine,
start_part, end_part, stop_reason
校验件号连续、去重
│
▼ ToolLifeCalculator (numpy/pandas)
单刀寿命：
life_parts = end_part – start_part + 1
按刀聚合，标记早损
│
▼ GradeStatistics (pandas/numpy)
型号级统计：
按 (牌号,涂层,材料) 分组
均值/中位数/标准差/P10/P90/置信区间
│
▼ WeibullAnalyzer (scipy)
寿命分布：
weibull_min.fit -> 形状k, 尺度η
k>5 集中, k<2 离散大
计算 B10 寿命(10%失效件数)
│
▼ AnomalyDetector (scikit-learn)
早损识别：
特征=[寿命, 偏离同组均值比例]
IsolationForest / KMeans 找异常刀
│
▼ ToolConditionGraph (networkx)
工况关系网：
刀具型号-材料-机床 三层图
边权=平均消耗件数倒数(越费权重越大)
│
▼ ToolVisualizer (matplotlib)
可视化：
1. 各型号寿命箱线图
2. 威布尔概率图
3. 早损刀散点标记
4. 工况关系网图
5. 寿命直方图+标称线
│
▼ SyntheticToolGenerator (numpy)
合成数据：
5类硬质合金刀，每类多把
干45钢/不锈钢/铸铁，含正常磨损+少量早损
3.3 为什么用威布尔而不是正态分布？
* 问题：寿命数据有下限 0、右偏，正态会算出负寿命，且对早损不敏感。
* 处理策略：威布尔分布是可靠性工程标准模型，形状参数直接反映失效模式。
* 工程合理性：跟刀具厂寿命样本报告口径一致，B10 寿命可直接用于换刀策略。
3.4 分析前后对比
维度 MES换刀记录 本程序
同型号均值 手算 自动分组聚合
寿命分布 无 威布尔拟合 + B10
早损识别 靠肉眼 聚类自动标红
工况归因 无 networkx 关系网
换刀阈值 拍脑袋 P90 + B10 双参考
四、OOP 代码实现
4.1 项目结构
tool_life_eval/
├── tool_life_eval/
│ ├── __init__.py
│ ├── log_loader.py # 刀具记录加载
│ ├── life_calculator.py # 单刀寿命计算
│ ├── grade_statistics.py # 型号级统计
│ ├── weibull_analyzer.py # 威布尔分析
│ ├── anomaly_detector.py # 早损异常识别
│ ├── condition_graph.py # 工况关系网
│ ├── visualizer.py # 可视化
│ └── synthetic_data.py # 合成数据
├── tests/
│ ├── __init__.py
│ └── test_tool_life.py
├── results/
│ ├── box_life_by_grade.png # 箱线图
│ ├── weibull_probplot.png # 威布尔概率图
│ ├── anomaly_scatter.png # 早损散点
│ ├── condition_graph.png # 工况关系网
│ ├── life_hist.png # 寿命直方图
│ ├── grade_life_stats.csv # 型号统计表
│ ├── early_failure_list.csv # 早损清单
│ └── weibull_report.txt
└── run_tool_life.py
4.2 核心源码
<details>
<summary></summary>
"""刀具加工记录加载器"""
import pandas as pd
from pathlib import Path
from typing import Optional
class ToolLogLoader:
"""加载刀具加工流水记录"""
STOP_REASONS = ("normal_wear", "chipping", "edge_break", "reach_life")
def __init__(self, filepath: str = "tool_logs.csv",
encoding: str = "utf-8"):
self.filepath = Path(filepath)
self.encoding = encoding
self._raw: Optional[pd.DataFrame] = None
def load(self) -> pd.DataFrame:
if not self.filepath.exists():
raise FileNotFoundError(f"文件不存在: {self.filepath}")
self._raw = pd.read_csv(self.filepath, encoding=self.encoding)
rename = {}
for tgt, al in {
"tool_id": ["tool_id", "刀号", "id"],
"tool_grade": ["tool_grade", "牌号", "grade"],
"coating": ["coating", "涂层", "coat"],
"material": ["material", "加工材料", "mat"],
"machine": ["machine", "机床", "mc"],
"start_part": ["start_part", "起始件", "sp"],
"end_part": ["end_part", "结束件", "ep"],
"stop_reason": ["stop_reason", "停用原因", "reason"],
}.items():
if tgt not in self._raw.columns:
for a in al:
if a in self._raw.columns:
rename[a] = tgt
break
self._raw = self._raw.rename(columns=rename)
req = ["tool_id", "tool_grade", "coating", "material",
"machine", "start_part", "end_part", "stop_reason"]
miss = [c for c in req if c not in self._raw.columns]
if miss:
raise ValueError(f"缺少必要列: {miss}")
for c in ["start_part", "end_part"]:
self._raw[c] = pd.to_numeric(self._raw[c], errors="coerce")
self._raw = self._raw.dropna(subset=["start_part", "end_part"])
self._raw = self._raw[(self._raw["end_part"] >= self._raw["start_part"])]
self._raw = self._raw.sort_values(["tool_grade", "tool_id"]).reset_index(drop=True)
return self._raw.copy()
</details>
<details>
<summary></summary>
"""单刀寿命计算"""
import numpy as np
import pandas as pd
from typing import Optional
class ToolLifeCalculator:
"""计算每把刀可加工零件数"""
def __init__(self):
pass
def per_tool(self, df: pd.DataFrame) -> pd.DataFrame:
rows = []
for tool_id, g in df.groupby("tool_id"):
head = g.iloc[0]
# 支持一把刀分段加工时取全局起止
sp = int(g["start_part"].min())
ep = int(g["end_part"].max())
life = ep – sp + 1
rows.append({
"tool_id": tool_id,
"tool_grade": head["tool_grade"],
"coating": head["coating"],
"material": head["material"],
"machine": head["machine"],
"start_part": sp,
"end_part": ep,
"life_parts": int(life),
"stop_reason": g["stop_reason"].mode().iloc[0],
})
out = pd.DataFrame(rows)
return out.sort_values("tool_id").reset_index(drop=True)
def tag_early_failure(self, life_df: pd.DataFrame,
group_keys=("tool_grade", "coating", "material"),
k_sigma: float = 2.0) -> pd.DataFrame:
"""按同工况组均值-2σ标记早损"""
df = life_df.copy()
stats = df.groupby(list(group_keys))["life_parts"].agg(["mean", "std"]).reset_index()
stats["std"] = stats["std"].fillna(0.0)
df = df.merge(stats, on=list(group_keys), how="left")
df["is_early"] = (df["life_parts"] < df["mean"] – k_sigma * df["std"]) | \\
(df["stop_reason"].isin(["chipping", "edge_break"]))
df["dev_from_mean"] = (df["life_parts"] – df["mean"]).round(2)
for c in ["mean", "std"]:
df[c] = df[c].round(2)
return df
</details>
<details>
<summary></summary>
"""型号级耐用度统计"""
import numpy as np
import pandas as pd
from scipy import stats
from typing import Optional
class GradeStatistics:
"""按 (牌号,涂层,材料) 聚合统计"""
def __init__(self, conf: float = 0.95):
self.conf = conf
def summarize(self, life_df: pd.DataFrame) -> pd.DataFrame:
keys = ["tool_grade", "coating", "material"]
g = life_df.groupby(keys)["life_parts"]
n = g.count()
mean = g.mean()
med = g.median()
std = g.std(ddof=1).fillna(0.0)
p10 = g.quantile(0.10)
p90 = g.quantile(0.90)
# 均值置信区间
rows = []
for key_tuple in mean.index:
sub = life_df[(life_df["tool_grade"] == key_tuple[0]) &
(life_df["coating"] == key_tuple[1]) &
(life_df["material"] == key_tuple[2])]["life_parts"].values
if len(sub) > 1:
ci = stats.t.interval(self.conf, len(sub) – 1,
loc=sub.mean(), scale=stats.sem(sub))
ci_lo, ci_hi = round(ci[0], 1), round(ci[1], 1)
else:
ci_lo = ci_hi = round(float(sub.mean()), 1)
rows.append({
"tool_grade": key_tuple[0],
"coating": key_tuple[1],
"material": key_tuple[2],
"n_tools": int(n[key_tuple]),
"mean_life": round(mean[key_tuple], 1),
"median_life": round(med[key_tuple], 1),
"std_life": round(std[key_tuple], 1),
"p10_life": round(p10[key_tuple], 1),
"p90_life": round(p90[key_tuple], 1),
"ci95_lo": ci_lo,
"ci95_hi": ci_hi,
"early_fail_cnt": int(((life_df[
(life_df["tool_grade"] == key_tuple[0]) &
(life_df["coating"] == key_tuple[1]) &
(life_df["material"] == key_tuple[2])
]["is_early"] == True).sum()),
})
return pd.DataFrame(rows).sort_values("mean_life", ascending=False).reset_index(drop=True)
</details>
<details>
<summary></summary>
"""威布尔寿命分析 (scipy)"""
import numpy as np
import pandas as pd
from scipy import stats
from typing import Dict, Optional
class WeibullAnalyzer:
"""对每组寿命做威布尔最小分布拟合"""
def __init__(self):
pass
def fit_group(self, life_values: np.ndarray) -> Dict:
x = np.asarray(life_values, dtype=float)
x = x[x > 0]
if len(x) < 3:
return {"k": np.nan, "eta": np.nan, "b10": np.nan}
# weibull_min(c, loc, scale), c即形状k
c, loc, scale = stats.weibull_min.fit(x, floc=0)
k = float(c)
eta = float(scale)
b10 = float(eta * np.log(1 / 0.9) ** (1 / k)) if k > 0 else np.nan
return {"k": round(k, 3), "eta": round(eta, 1), "b10": round(b10, 1)}
def fit_all_groups(self, life_df: pd.DataFrame) -> pd.DataFrame:
keys = ["tool_grade", "coating", "material"]
rows = []
for key_tuple, g in life_df.groupby(keys):
res = self.fit_group(g["life_parts"].values)
rows.append({
"tool_grade": key_tuple[0],
"coating": key_tuple[1],
"material": key_tuple[2],
"weibull_k": res["k"],
"weibull_eta": res["eta"],
"b10_life": res["b10"],
"dispersion": "集中" if (res["k"] and res["k"] > 5)
else ("离散" if (res["k"] and res["k"] < 2) else "中等"),
})
return pd.DataFrame(rows).sort_values("weibull_k", ascending=False).reset_index(drop=True)
def probplot_data(self, life_values: np.ndarray):
"""返回威布尔概率图坐标"""
x = np.sort(np.asarray(life_values, dtype=float))
x = x[x > 0]
n = len(x)
if n < 2:
return x, np.array([])
f = (np.arange(1, n + 1) – 0.3) / (n + 0.4)
y = np.log(-np.log(1 – f))
return x, y
</details>
<details>
<summary></summary>
"""早损异常刀识别 (scikit-learn)"""
import numpy as np
import pandas as pd
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from typing import Optional
class AnomalyDetector:
"""基于同工况偏离度聚类找早损刀"""
def __init__(self, n_clusters: int = 2, random_state: int = 42):
self.n_clusters = n_clusters
self.random_state = random_state
self.scaler = StandardScaler()
def detect(self, tagged_df: pd.DataFrame) -> pd.DataFrame:
df = tagged_df.copy()
# 特征：寿命 + 偏离均值比例
df["dev_ratio"] = df["dev_from_mean"] / (df["mean"] + 1e-9)
X = df[["life_parts", "dev_ratio"]].values
Xs = self.scaler.fit_transform(X)
k = min(self.n_clusters, len(df))
km = KMeans(n_clusters=k, random_state=self.random_state, n_init=10)
df["cluster"] = km.fit_predict(Xs)
# 寿命低的簇标记为异常簇
cmean = df.groupby("cluster")["life_parts"].mean()
low_cluster = cmean.idxmin()
df["cluster_is_early"] = df["cluster"] == low_cluster
# 综合标记：规则早损 OR 聚类异常簇
df["final_early_flag"] = df["is_early"] | df["cluster_is_early"]
return df.sort_values("life_parts").reset_index(drop=True)
def summary(self, df: pd.DataFrame) -> pd.DataFrame:
sub = df[df["final_early_flag"]]
return sub[["tool_id", "tool_grade", "coating", "material",
"machine", "life_parts", "mean", "stop_reason",
"dev_from_mean"]].copy()
</details>
<details>
<summary></summary>
"""工况关系网 (networkx)"""
import networkx as nx
import pandas as pd
from typing import Optional
class ToolConditionGraph:
"""刀具型号-材料-机床 三层关系网"""
def __init__(self):
self.G = nx.Graph()
def build(self, life_df: pd.DataFrame, stats_df: pd.DataFrame) -> nx.Graph:
self.G.clear()
# 节点
for _, r in stats_df.iterrows():
gid = f"{r['tool_grade']}|{r['coating']}"
self.G.add_node(gid, ntype="grade",
mean_life=r["mean_life"])
self.G.add_node(r["material"], ntype="material")
for m in life_df["machine"].unique():
self.G.add_node(m, ntype="machine")
# 边：grade-material, 权=1/均值寿命(越费越大)
for _, r in stats_df.iterrows():
gid = f"{r['tool_grade']}|{r['coating']}"
w = round(1.0 / (r["mean_life"] + 1e-9), 5)
self.G.add_edge(gid, r["material"], weight=w,
mean_life=r["mean_life"])
# 机床级平均寿命
mc = life_df.groupby(["machine", "tool_grade", "coating"])["life_parts"].mean().reset_index()
for _, r in mc.iterrows():
gid = f"{r['tool_grade']}|{r['coating']}"
self.G.add_edge(gid, r["machine"],
weight=round(1.0 / (r["life_parts"] + 1e-9), 5),
mean_life=round(r["life_parts"], 1))
return self.G
def bottleneck_paths(self) -> pd.DataFrame:
"""找平均寿命最低的 grade-material-machine 组合"""
rows = []
for u, v, d in self.G.edges(data=True):
if "mean_life" in d:
rows.append({"node_a": u, "node_b": v,
"mean_life": d["mean_life"],
"weight": d["weight"]})
df = pd.DataFrame(rows)
if df.empty:
return df
return df.sort_values("mean_life").reset_index(drop=True)
</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
plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
class ToolVisualizer:
def __init__(self, results_dir: str = "results"):
self.results_dir = Path(results_dir)
self.results_dir.mkdir(exist_ok=True)
def box_life(self, life_df, stats_df):
fig, ax = plt.subplots(figsize=(14, 6))
labels = stats_df.apply(lambda r: f"{r['tool_grade']}\\n{r['coating']}\\n{r['material']}", axis=1)
data = []
for _, r in stats_df.iterrows():
sub = life_df[(life_df["tool_grade"] == r["tool_grade"]) &
(life_df["coating"] == r["coating"]) &
(life_df["material"] == r["material"])]["life_parts"].values
data.append(sub)
ax.boxplot(data, labels=labels, showmeans=True)
ax.set_ylabel("可加工件数", fontsize=12)
ax.set_title("各型号刀具寿命箱线图", fontsize=14, fontweight="bold")
plt.xticks(rotation=30, ha="right", fontsize=8)
ax.grid(axis="y", alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir / "box_life_by_grade.png",
dpi=150, bbox_inches="tight")
plt.close()
def weibull_probplot(self, life_df, weibull_df):
fig, ax = plt.subplots(figsize=(12, 7))
for _, w in weibull_df.iterrows():
sub = life_df[(life_df["tool_grade"] == w["tool_grade"]) &
(life_df["coating"] == w["coating"]) &
(life_df["material"] == w["material"])]["life_parts"].values
x, y = WeibullProbHelper().probplot_data(sub)
if len(x) < 2:
continue
ax.plot(x, y, ".-", label=f"{w['tool_grade']}|{w['coating']}|{w['material']} k={w['weibull_k']}")
ax.set_xscale("log")
ax.set_xlabel("寿命件数 (log)", fontsize=12)
ax.set_ylabel("ln(-ln(1-F))", fontsize=12)
ax.set_title("威布尔概率图", fontsize=14, fontweight="bold")
ax.legend(fontsize=7)
ax.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir / "weibull_probplot.png",
dpi=150, bbox_inches="tight")
plt.close()
def anomaly_scatter(self, det_df):
fig, ax = plt.subplots(figsize=(12, 6))
normal = det_df[~det_df["final_early_flag"]]
early = det_df[det_df["final_early_flag"]]
ax.scatter(normal["mean"], normal["life_parts"], c="#3498DB",
label="正常刀", s=30, alpha=0.7)
ax.scatter(early["mean"], early["life_parts"], c="#E74C3C",
label="早损刀", s=60, edgecolor="k")
# 参考线 y=mean
allx = det_df["mean"].values
ax.plot(allx, allx, "g–", alpha=0.5, label="同组均值线")
for _, r in early.iterrows():
ax.annotate(r["tool_id"], (r["mean"], r["life_parts"]),
fontsize=7, color="red")
ax.set_xlabel("同工况组平均寿命(件)", fontsize=12)
ax.set_ylabel("本刀寿命(件)", fontsize=12)
ax.set_title("早损刀具识别散点图", fontsize=14, fontweight="bold")
ax.legend()
ax.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir / "anomaly_scatter.png",
dpi=150, bbox_inches="tight")
plt.close()
def condition_graph(self, G):
fig, ax = plt.subplots(figsize=(14, 9))
pos = nx.spring_layout(G, seed=42, k=0.6)
ncolors = []
for n, d in G.nodes(data=True):
t = d.get("ntype", "grade")
ncolors.append({"grade": "#E74C3C", "material": "#2ECC71",
"machine": "#3498DB"}.get(t, "#999"))
sizes = [300 + (G.nodes[n].get("mean_life", 50) * 3 if G.nodes[n].get("ntype")=="grade" else 300) for n in G.nodes()]
nx.draw_networkx_nodes(G, pos, node_color=ncolors,
node_size=sizes, ax=ax, alpha=0.9)
edges = [(u, v) for u, v, d in G.edges(data=True)]
ws = [G.edges[u, v]["weight"] * 8000 for u, v in edges]
nx.draw_networkx_edg
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
评论前必须登录！
注册