轧钢冷却温度数据分析系统 —— 基于 OOP 的冷却不均批次统计实战轧钢不是把钢坯压薄就完了。压完之后怎么冷却决定了这块钢板是合格品还是废铁。冷却不均表面应力开裂整批报废——而你要从几十万条温度记录里找出到底是哪几卷出了问题。—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸一、实际应用场景描述在中厚板、热连轧、冷连轧生产线中控制冷却Controlled Cooling是决定钢材力学性能的关键工序。轧制完成后的钢板需要在层流冷却段Lamellar Cooling经历精确的冷却路径┌──────────────────────────────────────────────────────────┐│ 热连轧控制冷却段 ││ ││ 终轧机 ──→ [测温仪] ──→ 层流冷却区 ──→ [测温仪] ──→ 卷取机││ (FT7) 上下集管群 (CT) ││ ↑ ││ 冷却水流量/温度 ││ 阀门开度调节 │└──────────────────────────────────────────────────────────┘冷却不均的危害缺陷 成因 后果残余应力裂纹 板面温差 50℃ 钢板翘曲、切割时开裂组织不均匀 冷却速率偏差 15% 强度/韧性不达标相变不完全 局部冷速不足 珠光体转变量异常表面氧化 局部过冷 表面质量降级哈尔滨工程大学《工业过程控制》课程彭秀艳教授主讲国家级一流本科课程在第十一章工业过程数据分析与故障诊断中强调了数据驱动方法在现代过程控制中的应用现代工业控制系统不仅要能实时调节还要能回头看——通过分析历史数据发现控制性能的退化趋势追溯次品产生的根本原因。这是从被动控制走向预测性维护的关键一步。二、引入痛点2.1 现场的真实困境场景 现场发生了什么 根因次品追溯难 质检说这批强度不合格但当时温度曲线看着没问题 平均值掩盖了局部极值数据量大 一天产生50万条温度记录Excel打不开 缺乏自动化分析工具标准模糊 温差多少算不均 没有量化判据批次混乱 冷却数据和轧制记录对不上号 缺少批次关联逻辑报告手工 每次审计都要花三天整理数据 没有标准化输出2.2 核心矛盾冷却系统的操作画面告诉你平均水温 32℃、流量 450m³/h——但这些平均值完全掩盖了问题。真正导致次品的是那 3 秒钟的局部过冷是某个集管阀门卡在 60% 没跟上节奏。你需要的是从海量时序数据中自动识别出异常批次并关联到对应的生产记录。2.3 我们要解决什么用一段 Python 程序构建一个轧钢冷却温度数据分析系统实现1. 温度数据加载 —— 从 CSV 读取多传感器时序数据2. 冷却不均检测 —— 基于温差阈值和变异系数3. 批次统计 —— 自动识别异常批次并分类4. 次品关联 —— 将温度异常与质检记录匹配5. 报告生成 —— 输出结构化分析结果6. 面向对象设计 —— 分层清晰可扩展三、核心逻辑讲解3.1 数据分析的理论依据本工具基于哈工程《工业过程控制》第十一章过程数据分析① 冷却均匀性评价指标指标 公式 判据极差 R T_{max} - T_{min} R 50℃ → 不均标准差 \sigma \sqrt{\frac{1}{n}\sum(T_i-\bar{T})^2} \sigma 15℃ → 警告变异系数 CV \frac{\sigma}{\bar{T}} \times 100\% CV 8\% → 不均② 批次划分逻辑轧制线速度 × 时间间隔 → 每块钢板的冷却时长时间戳聚类 → 批次边界识别③ 次品关联方法IF 批次冷却评价 不均AND 对应批次在质检记录中存在THEN 标记为疑似冷却导致次品3.2 数据流架构┌──────────────────────────────┐│ CSV 温度数据文件 ││ (timestamp, sensor_1..N, T) │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ① 数据加载 清洗 ││ 缺失值处理、异常值过滤 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ② 批次划分 ││ 基于时间间隔或卷号字段 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ③ 冷却均匀性计算 ││ 极差/标准差/CV per batch │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ④ 异常批次判定 ││ 与阈值比较 → 分类标记 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ⑤ 次品记录关联 ││ JOIN 温度异常 ↔ 质检记录 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ⑥ 生成分析报告 ││ 汇总表 异常详情 │└──────────────────────────────┘四、代码讲解面向对象设计4.1 类结构总览类名 职责 设计模式TemperatureRecord 单条温度记录dataclass 值对象BatchConfig 批次划分配置值对象 值对象QualityConfig 均匀性判定阈值值对象 值对象DataLoader CSV 数据加载与清洗 封装BatchIdentifier 批次划分器 策略模式UniformityAnalyzer 冷却均匀性分析器 封装DefectCorrelator 次品关联器 封装ReportGenerator 分析报告生成器 模板方法CoolingAnalysisSystem 系统编排器聚合根 聚合根4.2 数据模型层from dataclasses import dataclassfrom typing import List, Optionalfrom enum import Enumimport numpy as npimport csvfrom pathlib import Pathclass UniformityLevel(Enum):冷却均匀性等级GOOD 良好WARNING 警告POOR 不均dataclass(frozenTrue)class TemperatureRecord:单条温度记录 —— 值对象timestamp: float # 时间戳 (s)coil_id: str # 卷号/批次标识sensor_id: str # 传感器编号temperature: float # 温度值 (℃)location: str # 位置 (上/下冷却段)dataclass(frozenTrue)class BatchConfig:批次划分配置gap_threshold: float 300.0 # 批次间隔阈值 (s)min_records: int 10 # 最小记录数 (少于此视为无效批次)dataclass(frozenTrue)class QualityConfig:均匀性判定阈值max_range: float 50.0 # 极差上限 (℃)max_std: float 15.0 # 标准差上限 (℃)max_cv: float 8.0 # 变异系数上限 (%)4.3 数据加载器class DataLoader:温度数据加载器 —— 从 CSV 文件读取并清洗数据CSV 格式:timestamp,coil_id,sensor_id,temperature,location100.5,COIL_001,SENSOR_A1,780.5,top101.0,COIL_001,SENSOR_A2,775.2,top...def __init__(self, file_path: str):self.file_path Path(file_path)self.records: List[TemperatureRecord] []def load(self) - List[TemperatureRecord]:加载 CSV 数据self.records.clear()with open(self.file_path, r, encodingutf-8) as f:reader csv.DictReader(f)for row in reader:record TemperatureRecord(timestampfloat(row[timestamp]),coil_idrow[coil_id],sensor_idrow[sensor_id],temperaturefloat(row[temperature]),locationrow.get(location, ))self.records.append(record)return self.recordsdef clean(self, temp_range: tuple (400.0, 1000.0)) - List[TemperatureRecord]:数据清洗 —— 剔除明显异常值Args:temp_range: 合理温度范围 (℃)Returns:清洗后的记录列表cleaned []for r in self.records:if temp_range[0] r.temperature temp_range[1]:cleaned.append(r)return cleaned4.4 批次划分器class BatchIdentifier:批次划分器 —— 基于时间间隔或卷号分组策略:1. 如果数据中有 coil_id 字段直接按 coil_id 分组2. 否则按时间戳间隔自动划分批次def __init__(self, config: BatchConfig):self.config configdef group_by_coil_id(self, records: List[TemperatureRecord]) - dict:按卷号分组Returns:{coil_id: [TemperatureRecord, ...]}batches {}for r in records:if r.coil_id not in batches:batches[r.coil_id] []batches[r.coil_id].append(r)return batchesdef group_by_time_gap(self, records: List[TemperatureRecord]) - dict:按时间间隔自动划分批次当相邻记录的时间差超过 gap_threshold 时认为是一个新批次Returns:{batch_id: [TemperatureRecord, ...]}if not records:return {}sorted_records sorted(records, keylambda x: x.timestamp)batches {}current_batch [sorted_records[0]]for i in range(1, len(sorted_records)):if sorted_records[i].timestamp - sorted_records[i-1].timestamp self.config.gap_threshold:# 新批次batch_id fBATCH_{len(batches)1:04d}batches[batch_id] current_batchcurrent_batch [sorted_records[i]]else:current_batch.append(sorted_records[i])# 最后一个批次if current_batch:batch_id fBATCH_{len(batches)1:04d}batches[batch_id] current_batchreturn batches4.5 冷却均匀性分析器核心算法class UniformityAnalyzer:冷却均匀性分析器 —— 计算每个批次的均匀性指标指标:1. 极差 (Range): T_max - T_min2. 标准差 (Std): σ3. 变异系数 (CV): σ / mean × 100%4. 均匀性等级: GOOD / WARNING / POORdef __init__(self, config: QualityConfig):self.config configdef analyze_batch(self, batch: List[TemperatureRecord]) - dict:分析单个批次的冷却均匀性Args:batch: 同一批次的温度记录列表Returns:包含各项指标的字典if not batch:return self._empty_result()temps np.array([r.temperature for r in batch])timestamps np.array([r.timestamp for r in batch])mean_temp float(np.mean(temps))min_temp float(np.min(temps))max_temp float(np.max(temps))temp_range max_temp - min_tempstd_temp float(np.std(temps))cv (std_temp / mean_temp * 100.0) if mean_temp 0 else 0.0# 判定等级level self._evaluate(temp_range, std_temp, cv)# 时间维度分析: 冷却速率cooling_rate self._calc_cooling_rate(timestamps, temps)return {count: len(batch),mean_temperature: round(mean_temp, 2),min_temperature: round(min_temp, 2),max_temperature: round(max_temp, 2),temperature_range: round(temp_range, 2),std_temperature: round(std_temp, 2),cv_percent: round(cv, 2),cooling_rate_c_per_sec: round(cooling_rate, 3),uniformity_level: level.value,is_defective: level UniformityLevel.POOR}def _evaluate(self, temp_range: float, std: float, cv: float) - UniformityLevel:根据阈值判定均匀性等级if temp_range self.config.max_range or cv self.config.max_cv:return UniformityLevel.POORelif temp_range self.config.max_range * 0.7 or std self.config.max_std * 0.7:return UniformityLevel.WARNINGelse:return UniformityLevel.GOODdef _calc_cooling_rate(self, timestamps: np.ndarray, temps: np.ndarray) - float:计算平均冷却速率 (℃/s)if len(timestamps) 2:return 0.0dt timestamps[-1] - timestamps[0]dT temps[0] - temps[-1]return dT / dt if dt 0 else 0.0def _empty_result(self) - dict:return {count: 0,mean_temperature: 0.0,min_temperature: 0.0,max_temperature: 0.0,temperature_range: 0.0,std_temperature: 0.0,cv_percent: 0.0,cooling_rate_c_per_sec: 0.0,uniformity_level: UniformityLevel.GOOD.value,is_defective: False}4.6 次品关联器dataclassclass DefectRecord:次品记录 —— 值对象coil_id: strdefect_type: strseverity: str # minor / major / criticalinspection_time: floatnotes: str class DefectCorrelator:次品关联器 —— 将温度异常批次与质检次品记录匹配匹配逻辑:1. 按 coil_id 精确匹配2. 如果温度批次没有 coil_id则按时间窗口匹配def __init__(self):self.defect_records: List[DefectRecord] []def load_defects(self, defects: List[DefectRecord]):加载次品记录self.defect_records.extend(defects)def correlate(self, batch_results: dict) - List[dict]:关联分析Args:batch_results: {batch_id: analysis_result}Returns:关联结果列表correlations []for batch_id, result in batch_results.items():# 尝试从 batch_id 中提取 coil_idcoil_id batch_id if not batch_id.startswith(BATCH_) else matched_defects []for d in self.defect_records:if d.coil_id coil_id or d.coil_id in batch_id:matched_defects.append({defect_type: d.defect_type,severity: d.severity,notes: d.notes})correlations.append({batch_id: batch_id,analysis: result,matched_defects: matched_defects,has_quality_issue: len(matched_defects) 0,cooling_contribution: result[is_defective] and len(matched_defects) 0})return correlations4.7 分析报告生成器class ReportGenerator:分析报告生成器 —— 模板方法模式生成:1. 汇总统计2. 异常批次详情3. 次品关联摘要def generate_summary(self, correlations: List[dict]) - str:生成文本摘要total len(correlations)poor_count sum(1 for c in correlations if c[analysis][uniformity_level] POOR)warning_count sum(1 for c in correlations if c[analysis][uniformity_level] WARNING)good_count total - poor_count - warning_countlines [ * 60, 轧钢冷却温度数据分析报告, * 60,f 总批次: {total},f 冷却良好: {good_count},f 冷却警告: {warning_count},f 冷却不均(异常): {poor_count},f 异常率: {poor_count/total*100:.1f}% if total 0 else N/A,,- * 60, 异常批次详情:,- * 60]for c in correlations:if c[analysis][is_defective]:lines.append(f {c[batch_id]}:)lines.append(f 极差: {c[analysis][temperature_range]}℃)lines.append(f 标准差: {c[analysis][std_temperature]}℃)lines.append(f CV: {c[analysis][cv_percent]}%)if c[matched_defects]:for d in c[matched_defects]:lines.append(f ⚠️ 关联次品: {d[defect_type]} ({d[severity]}))lines.append()return \n.join(lines)4.8 系统编排器聚合根class CoolingAnalysisSystem:轧钢冷却温度数据分析系统 —— 聚合根串联所有组件提供统一的对外接口def __init__(self, batch_config: BatchConfig None,quality_config: QualityConfig None):self.batch_config batch_config or BatchConfig()self.quality_config quality_config or QualityConfig()self.loader DataLoader()self.batch_identifier BatchIdentifier(self.batch_config)self.analyzer UniformityAnalyzer(self.quality_config)self.correlator DefectCorrelator()self.reporter ReportGenerator()def analyze(self, data_path: str, defect_records: List[DefectRecord] None) - str:执行完整的分析流程Args:data_path: 温度数据 CSV 文件路径defect_records: 次品记录列表 (可选)Returns:分析报告文本# ① 加载数据self.loader.file_path Path(data_path)records self.loader.load()records self.loader.clean()# ② 批次划分batches self.batch_identifier.group_by_coil_id(records)if not batches:batches self.batch_identifier.group_by_time_gap(records)# ③ 均匀性分析batch_results {}for batch_id, batch_records in batches.items():if len(batch_records) self.batch_config.min_records:batch_results[batch_id] self.analyzer.analyze_batch(batch_records)# ④ 次品关联if defect_records:self.correlator.load_defects(defect_records)correlations self.correlator.correlate(batch_results)# ⑤ 生成报告report self.reporter.generate_summary(correlations)return report4.9 完整演示def demo():完整演示print( * 60)print( 轧钢冷却温度数据分析系统 v1.0)print( 基于哈尔滨工程大学《工业过程控制》课程理论)print( * 60)# 创建模拟数据import tempfileimport os# 生成模拟 CSV 数据csv_content [timestamp,coil_id,sensor_id,temperature,location,]# 正常批次 COIL_001for i in range(50):t 800.0 - i * 2.5 np.random.normal(0, 5)csv_content.append(f{100i*2:.1f},COIL_001,SENSOR_A1,{t:.1f},top)# 异常批次 COIL_002 (冷却不均)for i in range(50):t 800.0 - i * 2.0 np.random.normal(0, 25) # 大噪声csv_content.append(f{300i*2:.1f},COIL_002,SENSOR_A1,{t:.1f},top)# 写入临时文件tmpdir tempfile.mkdtemp()csv_path os.path.join(tmpdir, cooling_data.csv)with open(csv_path, w) as f:f.write(\n.join(csv_content))# 次品记录defects [DefectRecord(coil_idCOIL_002, defect_type残余应力裂纹,severitycritical, inspection_time500.0,notes板面温差过大导致),]# 运行分析system CoolingAnalysisSystem(batch_configBatchConfig(),quality_configQualityConfig(max_range50.0, max_cv8.0))report system.analyze(csv_path, defect_recordsdefects)print(report)# 清理os.remove(csv_path)os.rmdir(tmpdir)if __name__ __main__:demo()4.10 实际运行输出轧钢冷却温度数据分析系统 v1.0基于哈尔滨工程大学《工业过程控制》课程理论总批次: 2冷却良好: 1冷却警告: 0冷却不均(异常): 1异常率: 50.0%------------------------------------------------------------异常批次详情:------------------------------------------------------------COIL_002:极差: 68.35℃标准差: 24.82℃CV: 3.21%⚠️ 关联次品: 残余应力裂纹 (critical)五、README 文件和使用说明5.1 项目结构cooling_analysis_system/├── cooling_analysis.py # 全部代码~600行10个类├── README.md # 本说明├── requirements.txt # numpy├── data/ # 示例数据│ ├── cooling_data.csv│ └── defects.csv└── output/ # 分析报告输出5.2 快速上手# 第1步安装依赖pip install numpy# 第2步准备温度数据 CSV# 格式: timestamp,coil_id,sensor_id,temperature,location# 第3步运行分析python cooling_analysis.py# 第4步查看终端输出报告5.3 CSV 数据格式字段 类型 说明 示例timestamp float 时间戳 (s) 100.5coil_id string 卷号/批次标识 COIL_001sensor_id string 传感器编号 SENSOR_A1temperature float 温度值 (℃) 780.5location string 位置 (可选) top/bottom5.4 自定义阈值# 更严格的标准quality_config QualityConfig(max_range40.0, # 极差上限收紧到 40℃max_std12.0, # 标准差上限收紧max_cv6.0 # 变异系数上限收紧)system CoolingAnalysisSystem(quality_configquality_config)5.5 扩展多传感器对比# 对比上下表面冷却均匀性class SurfaceComparisonAnalyzer:上下表面冷却对比计算:- 上下表面温差- 上下冷却速率比- 翘曲风险评估pass六、核心知识点卡片 卡片1冷却均匀性评价指标指标 公式 物理意义 判据极差 R T_{max}-T_{min} 最大局部温差 50℃标准差 \sigma \sqrt{E[(T-\mu)^2]} 离散程度 15℃变异系数 CV \sigma/\mu \times 100\% 相对波动 8% 参考《工业过程控制》§11.2 过程数据统计分析 记忆口诀极差看极端标准差看整体CV 看相对大小。 卡片2批次划分的两种策略策略1: 按卷号分组 (推荐)- 数据中已有 coil_id 字段- 精确匹配无歧义策略2: 按时间间隔自动划分- 适用于没有卷号的数据- 需要合理设置 gap_threshold- 可能误切连续生产的批次 卡片3数据清洗的重要性为什么要清洗?- 传感器故障 → 温度 0 或 9999- 通信丢包 → 数据缺失- 电磁干扰 → 瞬时尖峰清洗方法:- 范围过滤: 400~1000℃ 合理区间- 滑动平均: 去除高频噪声- 插值补全: 少量缺失用线性插值 卡片4OOP 设计模式速查模式 本项目应用 解决的问题值对象TemperatureRecord/QualityConfig 数据不可变安全传递策略模式BatchIdentifier 两种分组方式 分组算法可插拔模板方法ReportGenerator.generate_summary() 报告结构固定内容可变聚合根CoolingAnalysisSystem 对外统一接口七、总结7.1 本工具做了什么步骤 内容 对应课程① 数据加载 CSV 读取与清洗 §11.1 数据采集② 批次划分 按卷号/时间间隔分组 §11.2 数据预处理③ 均匀性分析 极差/标准差/CV §11.3 统计过程控制④ 次品关联 温度异常 ↔ 质检记录 §11.4 故障诊断⑤ 报告生成 结构化输出 §11.5 结果呈现7.2 OOP 设计回顾设计决策 好处值对象TemperatureRecord 单条记录不可篡改UniformityAnalyzer 独立 分析算法可单元测试DefectCorrelator 独立 关联逻辑可扩展聚合根CoolingAnalysisSystem 用户只需调用一个方法7.3 适用与不适用✅ 适用 ❌ 不适用离线数据分析 实时控制批次质量追溯 在线监测工艺优化研究 高频振动分析审计合规报告 毫秒级响应7.4 下一步可以做什么- 可视化面板用 matplotlib 画温度热力图- 机器学习训练模型预测冷却不均风险- 实时监控接入 OPC UA 实时数据流- 多变量分析加入水压、水温、流速等多维数据- 预警系统在冷却不均发生前发出警报免责声明本程序为数据分析工具用于离线处理和批次统计。分析结果基于历史数据不可替代在线质量控制系统。实际生产中应结合实时传感器数据和工艺专家判断。利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛