""" 工艺预测模型 — 灰箱（Gray-box）架构 设计思路： 物理结构来自 Arrhenius 酸洗动力学，参数取自公开文献实验值， 而非理论推导。每个模型内置校准系数 K_cal（初始=1.0）， 投产后可通过 calibrate() 方法用实测结果回归更新， 使模型随数据积累逐步收敛到真实工况。 关键文献依据： [1] 碳钢 HCl 酸洗活化能：Ea ≈ 40~50 kJ/mol（实验测定均值取 45 kJ/mol） 来源：Hydrochloric Acid Pickling Process Optimization in Metal Wire, IJSSST Vol-16 No-5; IspatGuru Pickling of Hot Rolled Strip [2] H⁺ 浓度动力学阶次：1.0~2.0阶（取保守值 1.2） 来源：Optimizing pickling process for 30Cr13 steel, ScienceDirect 2025; neural network MPC studies [3] 温度效应校验：速率每升温 6~8°C 翻倍（Ea≈45 kJ/mol 时对应约 7°C） [4] 欠酸洗风险判别特征：strip thickness, speed, conc, temp 来源：Prediction of under pickling defects on steel strip surface, arXiv:1207.0911 [5] 速度优化：Nelder-Mead simplex 已在实际 1450mm 酸洗线验证 来源：Zhu et al., Advances in Mechanical Engineering, 2016 """ import math import json import os from typing import List, Dict, Any, Optional, Tuple # ── 校准系数持久化路径 ──────────────────────────────────────────────────────── _CAL_FILE = os.path.join(os.path.dirname(__file__), "cal_coeffs.json") def _load_cal() -> Dict[str, float]: try: with open(_CAL_FILE) as f: return json.load(f) except Exception: return {} def _save_cal(d: Dict[str, float]): with open(_CAL_FILE, "w") as f: json.dump(d, f, indent=2) # ───────────────────────────────────────────────────────────────────────────── # 1. 酸洗速度模型（Gray-box） # ───────────────────────────────────────────────────────────────────────────── class AcidSpeedModel: """ 基于文献实测参数的 Arrhenius 灰箱模型。 与上一版本的关键差异： - Ea/R: 3000 K → 5413 K（45 kJ/mol 实验值，文献[1]） - 浓度指数: 0.6 → 1.2（H⁺ 二阶动力学，文献[2]） - 增加氧化铁皮结构修正（FeO/Fe₃O₄双层模型，文献[4]） - 内置 K_cal 校准系数，支持投产后在线标定 """ # 文献实验值（碳钢 HCl 连续酸洗） TANK_LENGTH = 18.0 # m，单槽有效长度（按设备规格书） NUM_TANKS = 6 # K0 由实际工况反推： # 目标：75°C、180 g/L 正常酸液条件下，最大速度约 120~130 m/min（PI≥95%） # 推导：t_total = 6×18/(125/60)=51.8s，k=ln(20)/51.8=0.058 s⁻¹ # k=K0×SCALE_RATE_FACTOR → K0=0.058/0.765≈0.075 s⁻¹ K0 = 0.075 # 指前因子 s⁻¹，由设备规格反推标定 EA_R = 5413.0 # Ea/R (K)，Ea=45 kJ/mol / R=8.314（文献实验值[1]） T_REF = 348.15 # 参考温度 75°C (K) C_REF = 180.0 # 参考游离酸浓度 g/L N_CONC = 1.2 # 浓度动力学阶次（文献[2] 取保守值） V_MIN = 20.0 V_MAX = 180.0 CAL_KEY = "acid_speed_kcal" # 氧化铁皮结构系数（FeO 快速溶解 + Fe₃O₄ 慢速溶解，文献[4]） # 热轧碳钢铁皮组成约：FeO 70%，Fe₃O₄ 20%，Fe₂O₃ 10% # FeO 溶速约为 Fe₃O₄ 的 4 倍；有效速率取加权平均 SCALE_RATE_FACTOR = 0.70 * 1.0 + 0.20 * 0.25 + 0.10 * 0.15 # ≈ 0.765 def __init__( self, thickness: float, # mm width: float, # mm steel_grade: str, acid_conc_list: List[float], # 各槽游离酸 g/L acid_temp_list: List[float], # 各槽温度 °C scale_weight: float = 8.5, # g/m²，氧化铁皮重量 target_pi: float = 95.0, ): if len(acid_conc_list) != self.NUM_TANKS: raise ValueError(f"acid_conc_list 需要 {self.NUM_TANKS} 个元素") if len(acid_temp_list) != self.NUM_TANKS: raise ValueError(f"acid_temp_list 需要 {self.NUM_TANKS} 个元素") self.thickness = thickness self.width = width self.steel_grade = steel_grade self.acid_conc_list = acid_conc_list self.acid_temp_list = acid_temp_list self.scale_weight = scale_weight self.target_pi = target_pi self.K_cal = _load_cal().get(self.CAL_KEY, 1.0) def _k_i(self, conc: float, temp_c: float) -> float: """单槽有效酸洗速率常数（含文献参数 + 铁皮结构修正）""" T_k = temp_c + 273.15 arrhenius = math.exp(-self.EA_R * (1.0 / T_k - 1.0 / self.T_REF)) conc_factor = max(conc / self.C_REF, 0.01) ** self.N_CONC # 铁皮越厚，有效接触面积越低（正比于 1/scale_weight^0.3 经验修正） scale_corr = (8.5 / max(self.scale_weight, 1.0)) ** 0.3 return self.K0 * arrhenius * conc_factor * self.SCALE_RATE_FACTOR * scale_corr * self.K_cal def _compute_pi(self, v_mpm: float) -> Tuple[float, List[float], List[float]]: v_mps = v_mpm / 60.0 pi_prev = 0.0 pi_per_tank, rt_per_tank = [], [] for i in range(self.NUM_TANKS): t_i = self.TANK_LENGTH / v_mps k_i = self._k_i(self.acid_conc_list[i], self.acid_temp_list[i]) # 精确解析解：dPI/dt = k*(1-PI/100) → PI_new = 100-(100-PI_old)*exp(-k*t) # 避免 Euler 一阶近似在 k*t 较大时的严重失真 pi_prev = 100.0 - (100.0 - pi_prev) * math.exp(-k_i * t_i) pi_per_tank.append(round(pi_prev, 2)) rt_per_tank.append(round(t_i, 1)) return pi_prev, pi_per_tank, rt_per_tank def calculate(self) -> Dict[str, Any]: # Nelder-Mead 单维退化为二分搜索（文献[5]验证有效） pi_at_min, _, _ = self._compute_pi(self.V_MIN) if pi_at_min < self.target_pi: pi, pp, rt = self._compute_pi(self.V_MIN) return { "max_speed": self.V_MIN, "pi_per_tank": pp, "residence_time_per_tank": rt, "total_pi": round(pi, 2), "under_pickling_risk": self._risk_level(self.V_MIN, pi), "warning": "酸液条件不足，即使最低速下酸洗指数仍低于目标，请检查酸浓度和温度", "K_cal": self.K_cal, } lo, hi, best_v = self.V_MIN, self.V_MAX, self.V_MIN while hi - lo >= 0.5: mid = (lo + hi) / 2.0 pi_mid, _, _ = self._compute_pi(mid) if pi_mid >= self.target_pi: best_v = mid; lo = mid + 0.5 else: hi = mid - 0.5 best_v = math.floor(best_v) total_pi, pi_per_tank, rt_per_tank = self._compute_pi(best_v) return { "max_speed": best_v, "pi_per_tank": pi_per_tank, "residence_time_per_tank": rt_per_tank, "total_pi": round(total_pi, 2), "under_pickling_risk": self._risk_level(best_v, total_pi), "warning": None, "K_cal": self.K_cal, } def _risk_level(self, speed: float, pi: float) -> str: """ 欠酸洗风险评估（文献[4] decision-tree 特征阈值） 输入：speed(m/min), pi(%)，结合厚度、浓度综合判断 """ avg_conc = sum(self.acid_conc_list) / len(self.acid_conc_list) avg_temp = sum(self.acid_temp_list) / len(self.acid_temp_list) # 文献给出的欠酸洗高风险条件组合 risk_score = 0 if pi < 85: risk_score += 3 elif pi < 92: risk_score += 1 if speed > 140: risk_score += 2 if avg_conc < 120: risk_score += 2 if avg_temp < 68: risk_score += 2 if self.thickness > 4.0: risk_score += 1 if risk_score >= 5: return "HIGH" elif risk_score >= 2: return "MEDIUM" else: return "LOW" def calibrate(self, actual_max_speed: float, actual_quality_ok: bool) -> float: """ 投产后标定接口： 传入某卷的实际最大可用速度（操作员确认质量合格时的速度）， 用简单比例更新 K_cal，使模型逐步向真实工况收敛。 actual_max_speed: 实际测得质量合格的最高速度 (m/min) actual_quality_ok: True=该速度下质量合格，False=出现欠酸洗 """ predicted = self.calculate()["max_speed"] if not actual_quality_ok: # 预测速度偏高，缩减 K_cal adjustment = 0.95 else: ratio = actual_max_speed / max(predicted, 1.0) # 平滑更新，避免单次样本过拟合 adjustment = 1.0 + 0.3 * (ratio - 1.0) adjustment = max(0.7, min(1.3, adjustment)) self.K_cal = round(self.K_cal * adjustment, 4) cal = _load_cal() cal[self.CAL_KEY] = self.K_cal _save_cal(cal) return self.K_cal # ───────────────────────────────────────────────────────────────────────────── # 2. 张力设定模型 # ───────────────────────────────────────────────────────────────────────────── class TensionModel: """ 张力模型：基于截面积×屈服强度，区间比例系数参考酸洗线工程手册。 每个区段独立校准系数 zone_kcal[zone]，互不干扰。 """ # 各区基准比例系数（酸洗线工程实践均值） ZONE_RATIOS = { "inlet": 1.00, "s1_roller": 0.85, "acid_entry": 0.78, "acid1": 0.72, "acid2": 0.68, "acid3": 0.68, "rinse": 0.70, "leveler": 0.76, "s2_roller": 0.88, "outlet": 1.00, } ZONE_NAMES_CN = { "inlet": "入口张力辊", "s1_roller": "S1夹送辊", "acid_entry": "酸洗入口辊", "acid1": "1#酸槽", "acid2": "2#酸槽", "acid3": "3#酸槽", "rinse": "漂洗段辊", "leveler": "拉矫机", "s2_roller": "S2夹送辊", "outlet": "出口张力辊", } @staticmethod def _zone_cal_key(zone: str) -> str: return f"tension_zone_{zone}" def __init__( self, thickness: float, width: float, yield_strength: float, tension_coef: float = 0.25, ): self.thickness = thickness self.width = width self.yield_strength = yield_strength self.tension_coef = tension_coef cal = _load_cal() # 每个区段独立加载自己的校准系数，默认 1.0 self.zone_kcal: Dict[str, float] = { z: cal.get(self._zone_cal_key(z), 1.0) for z in self.ZONE_RATIOS } def calculate(self) -> Dict[str, Any]: cross_section = self.thickness * self.width # mm² # T_max 是理论基准值（不含区段校准，区段校准在各 zone 内单独乘） t_base_kn = (self.tension_coef * self.yield_strength * cross_section / 1000.0) # kN zones = {} for zone, ratio in self.ZONE_RATIOS.items(): k = self.zone_kcal.get(zone, 1.0) zones[zone] = { "tension_kN": round(t_base_kn * ratio * k, 2), "ratio": ratio, "k_cal": k, "name_cn": self.ZONE_NAMES_CN[zone], } density = 7850.0 mass_per_m = density * (self.thickness / 1000.0) * (self.width / 1000.0) accel_kn = round(mass_per_m * (30.0 / 60.0) / 1000.0, 3) t_max_kn = round(t_base_kn * self.zone_kcal.get("inlet", 1.0), 2) return { "T_max": t_max_kn, "T_base": round(t_base_kn, 2), "cross_section_mm2": round(cross_section, 1), "zones": zones, "weld_speed_limit": 60.0, "weld_tension_kN": round(t_max_kn * 0.60, 2), "accel_tension": accel_kn, "zone_kcal": self.zone_kcal, } def calibrate(self, zone: str, measured_kn: float) -> Dict[str, float]: """仅更新指定区段的校准系数，其他区段不变""" if zone not in self.ZONE_RATIOS: raise ValueError(f"未知区段: {zone}") t_base = (self.tension_coef * self.yield_strength * self.thickness * self.width / 1000.0) predicted = t_base * self.ZONE_RATIOS[zone] * self.zone_kcal[zone] ratio = measured_kn / max(predicted, 0.1) # 平滑更新，步长 40%，范围限制在 [0.5, 2.0] adjustment = 1.0 + 0.4 * (ratio - 1.0) adjustment = max(0.5, min(2.0, adjustment)) new_k = round(self.zone_kcal[zone] * adjustment, 4) self.zone_kcal[zone] = new_k cal = _load_cal() cal[self._zone_cal_key(zone)] = new_k _save_cal(cal) return self.zone_kcal # ───────────────────────────────────────────────────────────────────────────── # 3. 质量预测模型 # ───────────────────────────────────────────────────────────────────────────── class QualityPredictionModel: """ 欠酸洗风险 + 质量等级预测。 v2 变化： - 使用与 AcidSpeedModel 一致的文献参数（Ea/R=5413, n=1.2） - 欠酸洗风险特征阈值参考 arXiv:1207.0911 的 decision-tree 结论 - 增加铁离子浓度（FeCl₂）对酸洗能力的抑制修正 - 支持投产后用实际质量等级校准评分阈值 """ EA_R = 5413.0 T_REF = 348.15 C_REF = 180.0 N_CONC = 1.2 CAL_KEY = "quality_kcal" def __init__( self, thickness: float, avg_speed: float, acid_conc_avg: float, # 游离酸均值 g/L acid_temp_avg: float, # 温度均值 °C scale_weight: float = 8.5, fe_conc_avg: float = 60.0, # FeCl₂ 浓度 g/L（铁离子抑制效应） ): self.thickness = thickness self.avg_speed = avg_speed self.acid_conc_avg = acid_conc_avg self.acid_temp_avg = acid_temp_avg self.scale_weight = scale_weight self.fe_conc_avg = fe_conc_avg self.K_cal = _load_cal().get(self.CAL_KEY, 1.0) def _pickling_index_score(self) -> float: T_k = self.acid_temp_avg + 273.15 arrhenius = math.exp(-self.EA_R * (1.0 / T_k - 1.0 / self.T_REF)) conc_factor = max(self.acid_conc_avg / self.C_REF, 0.01) ** self.N_CONC # 铁离子抑制：FeCl₂ > 80 g/L 时显著降低酸洗速率（文献经验） fe_inhibition = 1.0 - max(0.0, (self.fe_conc_avg - 80.0) / 200.0) * 0.35 scale_corr = (8.5 / max(self.scale_weight, 1.0)) ** 0.3 exposure = (1.20 * arrhenius * conc_factor * fe_inhibition * scale_corr * 18.0 * 6) / (self.avg_speed / 60.0) pi_score = 100.0 * (1.0 - math.exp(-exposure / 10.0)) return min(max(pi_score * self.K_cal, 0.0), 100.0) def _surface_score(self, pi_score: float) -> float: # 最优速度区间 80-140 m/min（文献[4] 欠酸洗风险判别边界） if self.avg_speed < 60: speed_score = 80.0 elif self.avg_speed <= 140: speed_score = 80.0 + 15.0 * (self.avg_speed - 60) / 80.0 else: over = (self.avg_speed - 140) / 40.0 speed_score = 95.0 - 30.0 * over return min(max(pi_score * 0.65 + speed_score * 0.35, 0.0), 100.0) def _grade(self, pi: float, surface: float) -> str: c = (pi + surface) / 2.0 if c >= 90: return "A1" if c >= 80: return "A2" if c >= 70: return "B1" if c >= 60: return "B2" return "C" def _recommendations(self, pi: float, surface: float) -> List[str]: recs = [] if self.fe_conc_avg > 80: recs.append(f"铁离子浓度偏高（{self.fe_conc_avg:.0f} g/L），酸洗能力受抑制，建议加速换酸或补充新酸") if pi < 80: recs.append("酸洗指数偏低，建议提高酸液浓度至 180 g/L 以上，或将温度升至 80°C") if pi < 65: recs.append(f"欠酸洗风险高，建议将线速降至 {max(self.avg_speed*0.75, 20):.0f} m/min 以下") if self.acid_temp_avg < 70: recs.append(f"酸液温度偏低（{self.acid_temp_avg:.1f}°C），建议升温至 75~85°C") if self.acid_conc_avg < 120: recs.append(f"游离酸浓度偏低（{self.acid_conc_avg:.0f} g/L），建议补充新酸至 150 g/L") if self.avg_speed > 150: recs.append(f"线速过高（{self.avg_speed:.0f} m/min），欠酸洗风险，建议不超过 140 m/min") if self.scale_weight > 12.0: recs.append(f"氧化铁皮偏重（{self.scale_weight:.1f} g/m²），建议检查加热炉气氛控制") if not recs: recs.append("工艺参数在正常范围内，当前设定可继续保持") return recs def calculate(self) -> Dict[str, Any]: pi = round(self._pickling_index_score(), 1) surface = round(self._surface_score(pi), 1) return { "pi_score": pi, "surface_score": surface, "overall_grade": self._grade(pi, surface), "recommendations": self._recommendations(pi, surface), "K_cal": self.K_cal, } def calibrate(self, actual_grade: str) -> float: """传入实际质检等级，更新评分校准系数""" grade_map = {"A1": 95, "A2": 85, "B1": 75, "B2": 65, "C": 50} actual_score = grade_map.get(actual_grade, 75) result = self.calculate() predicted_score = (result["pi_score"] + result["surface_score"]) / 2.0 ratio = actual_score / max(predicted_score, 1.0) adjustment = 1.0 + 0.3 * (ratio - 1.0) adjustment = max(0.7, min(1.3, adjustment)) self.K_cal = round(self.K_cal * adjustment, 4) cal = _load_cal() cal[self.CAL_KEY] = self.K_cal _save_cal(cal) return self.K_cal # ───────────────────────────────────────────────────────────────────────────── # 4. 消耗预测模型 # ───────────────────────────────────────────────────────────────────────────── class AcidConsumptionModel: """ 单卷资源消耗预测。 单位消耗定额取自浙江企鹅1250mm规格书； 酸耗额外引入铁离子浓度修正（FeCl₂ 越高酸液越快失效，换酸频率越高）。 """ ACID_WITH_REGEN = 2.0 # kg/t ACID_WITHOUT_REGEN = 35.0 # kg/t STEAM_UNIT = 39.8 # kg/t POWER_UNIT = 14.0 # kWh/t COOLING_UNIT = 1.21 # m³/t def __init__( self, thickness: float, width: float, coil_weight_kg: float, has_regen_station: bool = True, fe_conc_avg: float = 60.0, # FeCl₂ g/L，影响换酸频率 ): self.thickness = thickness self.width = width self.coil_weight_kg = coil_weight_kg self.has_regen_station = has_regen_station self.fe_conc_avg = fe_conc_avg def calculate(self) -> Dict[str, Any]: weight_t = self.coil_weight_kg / 1000.0 acid_base = self.ACID_WITH_REGEN if self.has_regen_station else self.ACID_WITHOUT_REGEN # 铁离子修正：FeCl₂ > 100 g/L 时酸液利用率下降，有效酸耗上升 fe_factor = 1.0 + max(0.0, (self.fe_conc_avg - 100.0) / 100.0) * 0.4 acid_unit = round(acid_base * fe_factor, 3) return { "coil_weight_t": round(weight_t, 3), "has_regen_station": self.has_regen_station, "acid_consumption_kg": round(acid_unit * weight_t, 2), "acid_unit_kg_per_t": acid_unit, "steam_consumption_kg": round(self.STEAM_UNIT * weight_t, 2), "steam_unit_kg_per_t": self.STEAM_UNIT, "power_consumption_kwh": round(self.POWER_UNIT * weight_t, 2), "power_unit_kwh_per_t": self.POWER_UNIT, "cooling_water_m3": round(self.COOLING_UNIT * weight_t, 3), "cooling_water_unit_m3_per_t": self.COOLING_UNIT, "fe_conc_factor": round(fe_factor, 3), }