企业官网建设流程全解析

周四下午，数控车间。

"主轴又报警了，"操作工老刘拍了拍 VMC850 的防护门，"连续干到第四个钟头，主轴温度窜到 58°C，系统直接降速保护。前面三件尺寸还行，到第五件，孔径全超差了 0.02mm。"

我蹲在机床旁边看冷却液循环箱。

"冷却液温度多少？"我问。

"系统显示 32°C，"老刘指了指面板，"就一个数，还是当前瞬时值。我怀疑是冷却液越干越热，热变形把主轴和丝杠都带起来了，但系统不给我历史曲线，只给当前值。"

"你们有记录过冷却液温度随时间的变化吗？"

"有，"老刘从 PLC 里导了一份 CSV，"每分钟采一次，连续加工 8 小时，480 个数据点。但我拿 Excel 画了下，就是一条抖动的线，看不出规律。我想知道的是：开机后第几分钟开始升温明显？几小时到稳态？稳态温度是多少？超过 35°C 后尺寸是不是就开始飘了？"

"所以你要的不是'当前温度'，是'温度随连续加工时长的变化模型'？"

"对，"老刘点头，"我需要把 480 个点按加工时长拟合出一条升温曲线，算出热平衡时间，标出超过工艺阈值的时段，再跟加工件数的尺寸偏差对上。下次排产，我就知道连续干几小时必须停机冷却，或者提前开制冷机。"

"明白了，"我打开 VS Code，"你需要一个程序：读取数控冷却液温度时序数据，用指数饱和模型拟合升温过程，算热平衡时间和稳态温度，用 scipy 做曲线拟合和置信区间，用 pandas 做滑动窗口统计，用 matplotlib 画时序曲线+拟合曲线+阈值带，用 scikit-learn 做异常时段聚类。数据自包含——用 numpy 合成 480 点带噪声的升温数据，模拟'前 3 小时快速升温、之后趋近稳态'的真实工况，下载就能跑。"

我敲了几行代码：

import pandas as pd

import numpy as np

from scipy.optimize import curve_fit

df = pd.read_csv("coolant_temp.csv")

t = df["minutes"].values

temp = df["coolant_temp_c"].values

# 指数饱和模型: T(t) = T0 + (T_inf - T0)*(1 - exp(-t/tau))

def sat_model(t, t0, t_inf, tau):

return t0 + (t_inf - t0) * (1 - np.exp(-t / tau))

popt, _ = curve_fit(sat_model, t, temp, p0=[28, 38, 120])

t0, t_inf, tau = popt

steady_time = 3 * tau # 工程上 3τ 视为稳态

"这只是核心逻辑，"我说，"完整版用 OOP 封装：一个类管时序加载（解析分钟级时间戳），一个类管数据清洗（去毛刺、滑动平均），一个类管升温拟合（指数饱和模型 + 线性段检测），一个类管阈值分析（超温时段标记 + 热平衡时间计算），一个类管可视化（原始曲线+拟合曲线+阈值带+分段着色）。输出热平衡报告，直接给排产建议。"

老刘凑近屏幕："所以你这东西，就是告诉我：开机 0 分钟 28°C，约 180 分钟到 37.5°C 附近稳住，超过 35°C 的时段从第 95 分钟开始。那我排产就按 90 分钟一轮，中间强制循环冷却 10 分钟。"

"对。而且还能反推，"我补充，"用 scipy 拟合出 τ 后，能算不同开机功率下的热平衡时间；用 scikit-learn 把温度曲线分段聚类，区分'升温段/稳态段/异常跳变段'。数控热误差补偿不是'等报警再降速'，是'用升温曲线预判热平衡，把尺寸稳定性前置到排产阶段'。"

一、实际应用场景（真实痛点）

场景设定：数控加工中心长时间连续运行时，切削液（冷却液）因主轴箱散热、液压系统发热、循环泵做功而持续升温。冷却液温度升高会直接带动主轴轴承、滚珠丝杠热伸长，导致加工尺寸漂移。设备 HMI 仅显示当前冷却液温度瞬时值，不提供历史趋势、热平衡时间、超温时段统计，工程师只能靠经验判断"该停机了"。

现场原话（叙事化）：

"我不是不知道冷却液会热，"老刘说，"但'会热'和'热到什么程度、什么时候稳、什么时候开始影响尺寸'是两回事。面板就给我一个数：32°C。可我连续干 8 小时，前面 32°C，后面 39°C，平均一下 35°C，看着都在工艺范围内，但第五件就是超差了。后来我把 480 个采样点拉出来看，发现 35°C 是个分水岭，过了这条线尺寸偏差明显变大。可系统从来不告诉我'第几分钟过 35°C'。每次都是出废品了才知道。"

核心矛盾："单点瞬时温度显示"与"连续加工时长维度的升温建模与热平衡预判"之间的分析断层。需要一个"数控冷却液温度时序趋势分析程序"，用

"pandas" 处理分钟级时序，用

"scipy" 做指数饱和拟合，用

"numpy" 算滑动统计，用

"matplotlib" 绘制升温曲线+阈值带，用

"scikit-learn" 做时段聚类，输出可指导排产的热平衡报告。

二、痛点分析（映射到滨州职业学院《先进制造技术》课程模型）

《先进制造技术》模块 本篇痛点对应

数控加工与CAD/CAM技术：加工精度与热误差控制 冷却液温升 → 主轴/丝杠热变形 → 尺寸漂移

先进制造技术基础：制造过程状态监控 分钟级温度时序采集与趋势建模

智能制造与数字孪生：机床热特性数字孪生 升温曲线映射物理热平衡过程

先进制造新模式：精益生产（减少废品） 按热平衡时间排产，前置防超差

一句话总结：我们需要构建一个"数控冷却液温度-连续加工时长趋势分析程序"，用

"scipy.optimize.curve_fit" 拟合指数饱和升温模型，用

"pandas" 做时序滑动统计，用

"matplotlib" 绘制阈值带与分段曲线，用

"scikit-learn" 做时段聚类，实现从"看瞬时温度"到"预判热平衡、指导排产"的转化。

三、核心逻辑讲解（大白话）

3.1 问题本质：把冷却液想象成"一锅慢慢热起来的汤"

把数控冷却液箱想象成"灶上的一锅汤"：

* 开机时刻 = 刚开火：汤 28°C，跟车间环境温度差不多。

* 连续加工 = 小火一直炖：主轴在旁边发热，循环泵也在搅，热量只进不出（制冷机功率有限）。

* 前段升温快：刚开始温差大，散热慢，温度蹭蹭涨。

* 后段变慢：温度越高，跟环境的散热也越快，升得就慢了。

* 最后稳住 = 热平衡：进来的热量 = 散出去的热量，温度不再涨，比如稳定在 37.5°C。

* 工艺阈值 = 汤不能太烫：规定 ≤35°C，超过就可能影响加工精度。

* 热平衡时间 = 这锅汤多久不再变烫：工程上用 3τ（τ 是时间常数）来估算。

工业应用：

* scipy.optimize.curve_fit：用

"T(t)=T0+(T∞-T0)(1-e^(-t/τ))" 拟合，一行拿到 T0、T∞、τ 三个参数。

* pandas rolling：

"df["temp"].rolling(window=15).mean()" 做 15 分钟滑动平均，去抖动。

* numpy where：标记超过 35°C 的时间点。

3.2 业务逻辑 → 代码映射

定义冷却液温度时序数据模型

│

▼ CoolantDataLoader (pandas)

导入时序 CSV：

pd.read_csv("coolant_temp.csv")

解析：分钟序号、时间戳、冷却液温度、主轴负载率

│

▼ DataSmoother (pandas/numpy)

数据清洗：

去毛刺（差分跳变检测）

15分钟滑动平均

重采样对齐

│

▼ ThermalFitModel (scipy)

升温拟合：

指数饱和模型 curve_fit

输出 T0 / T∞ / τ

计算热平衡时间 = 3τ

线性段检测（前段斜率）

│

▼ ThresholdAnalyzer (numpy)

阈值分析：

标记超 35°C 时段

统计超温时长、首次超温时刻

分段：升温段 / 稳态段 / 超温段

│

▼ CoolantVisualizer (matplotlib)

可视化：

1. 原始+平滑+拟合曲线叠加

2. 阈值带（绿色安全区/红色超温区）

3. 分段着色图

4. 残差分布图

│

▼ SegmentClusterer (scikit-learn)

时段聚类：

KMeans 按温度斜率分段

区分升温段/稳态段/异常跳变段

│

▼ SyntheticDataGenerator (numpy)

合成数据生成：

480点（8小时×60分钟）

指数升温+高斯噪声+局部跳变

3.3 为什么用指数饱和模型而不是简单线性？

* 问题：线性拟合会预测"永远涨下去"，实际物理上会饱和。

* 处理策略：指数饱和模型符合热力学一阶系统响应，有物理意义。

* 工程合理性：τ 可直接用于排产计算，T∞ 是设备热设计上限参考。

3.4 分析前后对比

维度 HMI 瞬时显示 本程序分析

数据维度 当前 1 个值 480 点全量时序

趋势 无 指数饱和拟合曲线

热平衡时间 无 3τ 定量计算

超温时段 无 首次超温时刻+累计时长

排产指导 凭经验停机 按热平衡曲线定加工轮次

四、OOP 代码实现

4.1 项目结构

cnc_coolant_temp/

├── cnc_coolant_temp/

│ ├── __init__.py

│ ├── data_loader.py # 时序数据加载

│ ├── data_smoother.py # 数据平滑与清洗

│ ├── thermal_fit.py # 升温拟合模型

│ ├── threshold_analyzer.py # 阈值分析

│ ├── visualizer.py # 可视化

│ ├── segment_clusterer.py # 时段聚类

│ └── synthetic_data.py # 合成数据生成

├── tests/

│ ├── __init__.py

│ └── test_coolant_temp.py # 单元测试

├── results/

│ ├── temp_fit_curve.png # 原始+拟合+阈值带

│ ├── segment_colored.png # 分段着色图

│ ├── residual_plot.png # 残差分布图

│ ├── cluster_segments.png # 聚类分段图

│ ├── thermal_report.txt # 热平衡分析报告

│ └── coolant_temp.csv # 合成时序数据

└── run_coolant_analysis.py # 主程序入口

4.2 核心源码

<details>

<summary></summary>

"""数控冷却液温度时序数据加载器"""

import pandas as pd

from pathlib import Path

from typing import Optional

class CoolantDataLoader:

"""

冷却液温度时序加载器

支持分钟级 CSV 时序数据。

"""

def __init__(self, filepath: str = "coolant_temp.csv",

encoding: str = "utf-8"):

self.filepath = Path(filepath)

self.encoding = encoding

self._raw_df: Optional[pd.DataFrame] = None

def load(self) -> pd.DataFrame:

if not self.filepath.exists():

raise FileNotFoundError(f"文件不存在: {self.filepath}")

self._raw_df = pd.read_csv(self.filepath, encoding=self.encoding)

col_aliases = {

"minutes": ["分钟", "minutes", "min", "t_min"],

"timestamp": ["时间", "timestamp", "time"],

"coolant_temp_c": ["冷却液温度", "temp", "coolant_temp", "temperature_c"],

"spindle_load": ["主轴负载", "load", "spindle_load_pct", "load_pct"],

}

rename_map = {}

for target, aliases in col_aliases.items():

if target not in self._raw_df.columns:

for alias in aliases:

if alias in self._raw_df.columns:

rename_map[alias] = target

break

if rename_map:

self._raw_df = self._raw_df.rename(columns=rename_map)

# 数值化

if "coolant_temp_c" in self._raw_df.columns:

self._raw_df["coolant_temp_c"] = pd.to_numeric(

self._raw_df["coolant_temp_c"], errors="coerce"

)

if "minutes" in self._raw_df.columns:

self._raw_df["minutes"] = pd.to_numeric(

self._raw_df["minutes"], errors="coerce"

)

# 按分钟排序

if "minutes" in self._raw_df.columns:

self._raw_df = self._raw_df.sort_values("minutes").reset_index(drop=True)

return self._raw_df.copy()

def get_raw_data(self) -> Optional[pd.DataFrame]:

return self._raw_df.copy() if self._raw_df is not None else None

</details>

<details>

<summary></summary>

"""数据平滑与清洗"""

import numpy as np

import pandas as pd

from typing import Dict, Optional

class DataSmoother:

"""

冷却液温度数据清洗器

去毛刺 + 滑动平均。

"""

def __init__(self, window: int = 15, spike_threshold: float = 3.0):

"""

Parameters

----------

window : int

滑动平均窗口（分钟）

spike_threshold : float

毛刺检测标准差倍数

"""

self.window = window

self.spike_threshold = spike_threshold

def remove_spikes(self,

df: pd.DataFrame,

temp_col: str = "coolant_temp_c") -> pd.DataFrame:

"""

去毛刺：基于一阶差分检测跳变点

Returns

-------

pd.DataFrame

"""

result = df.copy()

diff = result[temp_col].diff().abs()

diff_mean = diff.mean()

diff_std = diff.std()

if diff_std > 0:

spike_mask = diff > (diff_mean + self.spike_threshold * diff_std)

# 毛刺用前后均值填充

result.loc[spike_mask, temp_col] = result[temp_col].interpolate(

limit_direction="both"

)

result["is_spike"] = spike_mask.astype(int)

else:

result["is_spike"] = 0

return result

def smooth(self,

df: pd.DataFrame,

temp_col: str = "coolant_temp_c") -> pd.DataFrame:

"""

滑动平均平滑

Returns

-------

pd.DataFrame

新增 smoothed_temp 列

"""

result = df.copy()

result["smoothed_temp"] = result[temp_col].rolling(

window=self.window, min_periods=1, center=True

).mean()

return result

def prepare(self,

df: pd.DataFrame,

temp_col: str = "coolant_temp_c") -> pd.DataFrame:

"""完整预处理流程"""

result = self.remove_spikes(df, temp_col)

result = self.smooth(result, temp_col)

return result

</details>

<details>

<summary></summary>

"""升温拟合模型"""

import numpy as np

import pandas as pd

from scipy.optimize import curve_fit

from typing import Dict, Optional, Callable

class ThermalFitModel:

"""

冷却液升温指数饱和拟合模型

T(t) = T0 + (T_inf - T0) * (1 - exp(-t / tau))

"""

def __init__(self):

self.params_ = None # [T0, T_inf, tau]

self.cov_ = None

@staticmethod

def _sat_model(t: np.ndarray, t0: float, t_inf: float, tau: float) -> np.ndarray:

"""指数饱和模型"""

t = np.asarray(t, dtype=float)

# 防止 tau<=0 导致发散

tau = max(tau, 1e-3)

return t0 + (t_inf - t0) * (1.0 - np.exp(-t / tau))

def fit(self,

minutes: np.ndarray,

temps: np.ndarray,

p0: Optional[tuple] = None) -> Dict:

"""

拟合升温曲线

Parameters

----------

minutes : np.ndarray

加工分钟数

temps : np.ndarray

对应温度

Returns

-------

dict

"""

minutes = np.asarray(minutes, dtype=float)

temps = np.asarray(temps, dtype=float)

mask = ~(np.isnan(minutes) | np.isnan(temps))

minutes = minutes[mask]

temps = temps[mask]

if p0 is None:

t0_guess = temps[0]

t_inf_guess = temps.max() + 1.0

tau_guess = minutes[len(minutes) // 3]

p0 = (t0_guess, t_inf_guess, tau_guess)

try:

popt, pcov = curve_fit(

self._sat_model, minutes, temps,

p0=p0, maxfev=10000

)

except RuntimeError:

# 拟合失败回退

popt = np.array([temps[0], temps.mean(), 60.0])

pcov = np.full((3, 3), np.nan)

self.params_ = popt

self.cov_ = pcov

t0, t_inf, tau = popt

steady_time = 3.0 * tau # 工程稳态时间

# 预测与残差

fitted = self._sat_model(minutes, *popt)

residuals = temps - fitted

rmse = float(np.sqrt(np.mean(residuals ** 2)))

r2 = 1.0 - np.sum(residuals ** 2) / np.sum((temps - temps.mean()) ** 2)

# 初始升温速率 (°C/min)

initial_rate = (t_inf - t0) / tau

return {

"T0": round(float(t0), 3),

"T_inf": round(float(t_inf), 3),

"tau_min": round(float(tau), 2),

"steady_time_min": round(float(steady_time), 2),

"steady_time_hour": round(float(steady_time / 60), 2),

"initial_rate_c_per_min": round(float(initial_rate), 4),

"r_squared": round(float(r2), 4),

"rmse": round(rmse, 3),

"n_samples": len(minutes),

}

def predict(self, minutes: np.ndarray) -> np.ndarray:

"""用拟合参数预测温度"""

if self.params_ is None:

raise ValueError("请先调用 fit()")

return self._sat_model(np.asarray(minutes, dtype=float), *self.params_)

def time_to_reach(self, target_temp: float) -> Optional[float]:

"""

计算达到目标温度的时间（分钟）

T(t) = T0 + (T_inf-T0)(1-e^{-t/τ}) → 反解 t

"""

if self.params_ is None:

return None

t0, t_inf, tau = self.params_

if target_temp <= t0:

return 0.0

if target_temp >= t_inf:

return None # 永远达不到

ratio = 1.0 - (target_temp - t0) / (t_inf - t0)

if ratio <= 0:

return 0.0

t = -tau * np.log(ratio)

return round(float(t), 2)

</details>

<details>

<summary></summary>

"""阈值分析器"""

import numpy as np

import pandas as pd

from typing import Dict, Optional

class ThresholdAnalyzer:

"""

冷却液温度阈值分析器

标记超温时段，计算热平衡分段。

"""

def __init__(self,

safe_limit: float = 35.0,

warn_limit: float = 33.0):

self.safe_limit = safe_limit

self.warn_limit = warn_limit

def mark_phases(self,

df: pd.DataFrame,

minutes_col: str = "minutes",

temp_col: str = "smoothed_temp") -> pd.DataFrame:

"""

标记温度阶段

Returns

-------

pd.DataFrame

新增 is_warn, is_over, phase 列

"""

result = df.copy()

use_col = temp_col if temp_col in result.columns else "coolant_temp_c"

result["is_warn"] = result[use_col] >= self.warn_limit

result["is_over"] = result[use_col] >= self.safe_limit

def phase_label(row):

if row[use_col] >= self.safe_limit:

return "超温段"

elif row[use_col] >= self.warn_limit:

return "预警段"

else:

return "安全段"

result["phase"] = result.apply(phase_label, axis=1)

return result

def over_temp_summary(self,

df: pd.DataFrame,

minutes_col: str = "minutes",

temp_col: str = "smoothed_temp") -> Dict:

"""

超温统计

Returns

-------

dict

"""

use_col = temp_col if temp_col in df.columns else "coolant_temp_c"

result = df.copy()

over_mask = result[use_col] >= self.safe_limit

warn_mask = result[use_col] >= self.warn_limit

over_df = result[over_mask]

warn_df = result[warn_mask]

first_over_min = (

int(over_df[minutes_col].min()) if not over_df.empty else None

)

first_warn_min = (

int(warn_df[minutes_col].min()) if not warn_df.empty else None

)

# 连续超温段切分

over_segments = []

if not over_df.empty:

seg_start = over_df.iloc[0][minutes_col]

prev = over_df.iloc[0][minutes_col]

for _, row in over_df.iloc[1:].iterrows():

if row[minutes_col] - prev > 2: # 间隔>2min视为新段

over_segments.append((seg_start, prev))

seg_start = row[minutes_col]

prev = row[minutes_col]

over_segments.append((seg_start, prev))

return {

"safe_limit": self.safe_limit,

"warn_limit": self.warn_limit,

"first_warn_min": first_warn_min,

"first_over_min": first_over_min,

"first_over_hour": (

round(first_over_min / 60, 2) if first_over_min else None

),

"over_count": int(over_mask.sum()),

"over_duration_min": int(over_mask.sum()), # 每分钟1点

"warn_count": int(warn_mask.sum()),

"over_segments": over_segments,

"max_temp": round(result[use_col].max(), 2),

"max_temp_min": int(result.loc[result[use_col].idxmax(), minutes_col]),

}

def suggest_schedule(self,

steady_time_min: float,

first_over_min: Optional[int],

cycle_safety_margin: float = 0.85) -> Dict:

"""

生成排产建议

Returns

-------

dict

"""

if first_over_min:

recommend_cycle = int(first_over_min * cycle_safety_margin)

else:

recommend_cycle = int(steady_time_min * 0.7)

return {

"recommend_cycle_min": recommend_cycle,

"recommend_cycle_hour": round(recommend_cycle / 60, 2),

"cooldown_min": 10,

"basis": f"按首次超温时刻 {first_over_min}min 的 "

f"{int(cycle_safety_margin*100)}% 作为加工轮次",

}

</details>

<details>

<summary></summary>

"""冷却液温度可视化"""

import numpy as np

import pandas as pd

import matplotlib.pyplot as plt

from pathlib import Path

from typing import Optional

plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]

plt.rcParams["axes.unicode_minus"] = False

class CoolantVisualizer:

"""数控冷却液温度可视化"""

def __init__(self, results_dir: str = "results"):

self.results_dir = Path(results_dir)

self.results_dir.mkdir(exist_ok=True)

def plot_fit_curve(self,

df: pd.DataFrame,

fit_model,

safe_limit: float = 35.0,

warn_limit: float = 33.0,

minutes_col: str = "minutes") -> None:

"""原始+平滑+拟合+阈值带"""

fig, ax = plt.subplots(figsize=(14, 7))

t = df[minutes_col].values

raw = df["coolant_temp_c"].values

smooth = df["smoothed_temp"].values if "smoothed_temp" in df.columns else raw

ax.scatter(t, raw, s=8, color="#BDC3C7", alpha=0.4, label="原始采样")

ax.plot(t, smooth, color="#3498DB", linewidth=1.2, label="滑动平均(15min)")

# 拟合曲线

t_dense = np.linspace(t.min(), t.max(), 500)

fit_vals = fit_model.predict(t_dense)

ax.plot(t_dense, fit_vals, color="#E67E22", linewidth=2.5,

label="指数饱和拟合")

# 阈值带

ax.axhline(y=safe_limit, color="#E74C3C", linestyle="--", linewidth=2,

label=f"安全上限 {safe_limit}°C")

ax.axhline(y=warn_limit, color="#F39C12", linestyle=":", linewidth=1.5,

label=f"预警线 {warn_limit}°C")

ax.fill_between(t, warn_limit, safe_limit, color="#F39C12",

alpha=0.08, label="预警区")

ax.fill_between(t, safe_limit, ax.get_ylim()[1], color="#E74C3C",

alpha=0.06, label="超温区")

# 稳态线

if fit_model.params_ is not None:

t_inf = fit_model.params_[1]

ax.axhline(y=t_inf, color="#2ECC71", linestyle="-.", linewidth=1.5,

label=f"拟合稳态 {t_inf:.1f}°C")

ax.set_xlabel("连续加工时长 (分钟)", fontsize=12)

ax.set_ylabel("冷却液温度 (°C)", fontsize=12)

ax.set_title("冷却液温度随连续加工时长变化曲线",

fontsize=14, fontweight="bold")

ax.legend(fontsize=9, ncol=3)

ax.grid(alpha=0.3)

plt.tight_layout()

plt.savefig(self.results_dir / "temp_fit_curve.png",

dpi=150, bbox_inches="tight")

plt.close()

def plot_segment_colored(self,

df: pd.DataFrame,

minutes_col: str = "minutes",

temp_col: str = "smoothed_temp") -> None:

"""分段着色图"""

use_col = temp_col if temp_col in df.columns else "coolant_temp_c"

fig, ax = plt.subplots(figsize=(14, 6))

colors = {"安全段": "#2ECC71", "预警段": "#F39C12", "超温段": "#E74C3C"}

for phase, color in colors.items():

sub = df[df["phase"] == phase]

if not sub.empty:

ax.scatter(sub[minutes_col], sub[use_col],

s=12, color=color, label=phase, alpha=0.7)

ax.set_xlabel("连续加工时长 (分钟)", fontsize=12)

ax.set_ylabel("冷却液温度 (°C)", fontsize=12)

ax.set_title("温度阶段分段着色", fontsize=14, fontweight="bold")

ax.legend(fontsize=10)

ax.grid(alpha=0.3)

plt.tight_layout()

plt.savefig(self.results_dir / "segment_colored.png",

dpi=150, bbox_inches="tight")

plt.close()

def plot_residual(self,

df: pd.DataFrame,

fit_model,

minutes_col: str = "minutes",

temp_col: str

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！