import pandas as pd import numpy as np # 尝试导入torch，如果失败则标记为不可用 try: import torch except ImportError: torch = None class ParameterAdjustmentAdvisor: """ 挤出机参数调节建议器 根据实时米重与标准米重的偏差，给出螺杆转速和流程主速的调整建议 """ def __init__(self): # 初始化默认参数关系系数 # 这些系数表示米重偏差1%时，参数需要调整的百分比 # 可以根据实际生产数据进行优化 self.default_coefficients = { 'screw_speed': 0.1, # 米重偏差1%，螺杆转速调整0.3%（降低调整幅度） 'process_main_speed': -0.1 # 米重偏差1%，流程主速调整-0.2%（降低调整幅度） } # 默认参数上下限（可根据实际设备情况调整） self.default_limits = { 'screw_speed': {'min': 30, 'max': 500}, 'process_main_speed': {'min': 0, 'max': 200} } # 最大调整幅度限制（百分比） self.max_adjustment_percentage = { 'screw_speed': 15.0, # 螺杆转速单次最大调整15% 'process_main_speed': 10.0 # 流程主速单次最大调整10% } def calculate_adjustment(self, real_time_weight, standard_weight, upper_limit, lower_limit, current_screw_speed, current_process_speed, current_screw_temperature=None, current_rear_barrel_temperature=None, current_front_barrel_temperature=None, current_head_temperature=None, coefficients=None, limits=None): """ 计算参数调整建议 :param real_time_weight: 实时米重 (Kg/m) :param standard_weight: 标准米重 (Kg/m) :param upper_limit: 米重上限 (Kg/m) :param lower_limit: 米重下限 (Kg/m) :param current_screw_speed: 当前螺杆转速 (rpm) :param current_process_speed: 当前流程主速 (m/min) :param current_screw_temperature: 当前螺杆温度 (°C) :param current_rear_barrel_temperature: 当前后机筒温度 (°C) :param current_front_barrel_temperature: 当前前机筒温度 (°C) :param current_head_temperature: 当前机头温度 (°C) :param coefficients: 自定义参数关系系数 (可选) :param limits: 自定义参数上下限 (可选) :return: 调整建议字典 """ # 使用默认系数或自定义系数 coeffs = coefficients if coefficients else self.default_coefficients # 使用默认上下限或自定义上下限 param_limits = limits if limits else self.default_limits # 计算米重偏差 weight_deviation = real_time_weight - standard_weight deviation_percentage = (weight_deviation / standard_weight) * 100 if standard_weight != 0 else 0 # 确定米重状态 if real_time_weight > upper_limit: status = "超上限" elif real_time_weight < lower_limit: status = "超下限" else: status = "正常范围" # 计算温度影响因子 - 考虑温度对挤出过程的影响 # 温度越高，物料流动性越好，相同螺杆转速下挤出量越大 temperature_factor = 1.0 # 如果提供了温度参数，计算温度影响因子 if current_screw_temperature is not None and current_head_temperature is not None: # 计算平均温度与参考温度的偏差 reference_temperature = 80.0 # 参考温度，可根据实际工艺调整 avg_temperature = (current_screw_temperature + current_head_temperature) / 2 temperature_deviation = avg_temperature - reference_temperature # 温度偏差每变化10°C，影响因子变化5% temperature_factor = 1.0 + (temperature_deviation / 10.0) * 0.05 # 限制温度影响因子的范围 temperature_factor = max(0.8, min(1.2, temperature_factor)) # 计算调整量 - 考虑温度影响因子 # 米重偏差 * 系数 * 温度因子 = 调整百分比 screw_speed_adjustment_percent = -deviation_percentage * coeffs['screw_speed'] * temperature_factor process_speed_adjustment_percent = -deviation_percentage * coeffs['process_main_speed'] * temperature_factor # 限制调整幅度，避免调整量过大 screw_speed_adjustment_percent = max( -self.max_adjustment_percentage['screw_speed'], min(self.max_adjustment_percentage['screw_speed'], screw_speed_adjustment_percent) ) process_speed_adjustment_percent = max( -self.max_adjustment_percentage['process_main_speed'], min(self.max_adjustment_percentage['process_main_speed'], process_speed_adjustment_percent) ) # 计算实际调整量 screw_speed_adjustment = (screw_speed_adjustment_percent / 100) * current_screw_speed process_speed_adjustment = (process_speed_adjustment_percent / 100) * current_process_speed # 计算调整后的参数值 new_screw_speed = current_screw_speed + screw_speed_adjustment new_process_speed = current_process_speed + process_speed_adjustment # 确保参数在合理范围内 new_screw_speed = max(param_limits['screw_speed']['min'], min(param_limits['screw_speed']['max'], new_screw_speed)) new_process_speed = max(param_limits['process_main_speed']['min'], min(param_limits['process_main_speed']['max'], new_process_speed)) # 重新计算实际调整百分比（考虑了最大调整幅度和参数上下限） screw_speed_adjust_percent = ((new_screw_speed - current_screw_speed) / current_screw_speed) * 100 if current_screw_speed != 0 else 0 process_speed_adjust_percent = ((new_process_speed - current_process_speed) / current_process_speed) * 100 if current_process_speed != 0 else 0 # 生成调整建议 - 无论米重是否在范围内，都提供调整建议，目标是让实时米重尽可能接近标准米重 recommendation = self._generate_recommendation( deviation_percentage, screw_speed_adjust_percent, new_screw_speed, process_speed_adjust_percent, new_process_speed ) return { 'status': status, 'real_time_weight': real_time_weight, 'standard_weight': standard_weight, 'upper_limit': upper_limit, 'lower_limit': lower_limit, 'deviation': weight_deviation, 'deviation_percentage': deviation_percentage, 'current_screw_speed': current_screw_speed, 'current_process_speed': current_process_speed, 'new_screw_speed': new_screw_speed, 'new_process_speed': new_process_speed, 'screw_speed_adjustment': screw_speed_adjustment, 'process_speed_adjustment': process_speed_adjustment, 'screw_speed_adjust_percent': screw_speed_adjust_percent, 'process_speed_adjust_percent': process_speed_adjust_percent, 'recommendation': recommendation } def _generate_recommendation(self, deviation_percentage, screw_speed_adjust_percent, new_screw_speed, process_speed_adjust_percent, new_process_speed): """ 生成调整建议文本 :param deviation_percentage: 米重偏差百分比 :param screw_speed_adjust_percent: 螺杆转速调整百分比 :param new_screw_speed: 调整后的螺杆转速 :param process_speed_adjust_percent: 流程主速调整百分比 :param new_process_speed: 调整后的流程主速 :return: 调整建议文本 """ # 根据偏差方向和大小生成建议 abs_deviation = abs(deviation_percentage) if abs_deviation < 0.5: # 偏差很小，接近标准值 return f"米重偏差很小 ({abs_deviation:.2f}%)，接近标准值。\n" \ f"建议微调以进一步接近标准值：\n" \ f"1. 将螺杆转速调整至 {new_screw_speed:.1f} rpm " \ f"( {'降低' if screw_speed_adjust_percent < 0 else '提高'} {abs(screw_speed_adjust_percent):.2f}% )\n" \ f"2. 将流程主速调整至 {new_process_speed:.1f} m/min " \ f"( {'降低' if process_speed_adjust_percent < 0 else '提高'} {abs(process_speed_adjust_percent):.2f}% )" elif deviation_percentage > 0: # 米重偏高，需要降低米重 return f"米重偏差 {abs_deviation:.2f}%（偏高），建议调整：\n" \ f"1. 将螺杆转速从当前值调整至 {new_screw_speed:.1f} rpm " \ f"( {'降低' if screw_speed_adjust_percent < 0 else '提高'} {abs(screw_speed_adjust_percent):.2f}% )\n" \ f"2. 将流程主速从当前值调整至 {new_process_speed:.1f} m/min " \ f"( {'降低' if process_speed_adjust_percent < 0 else '提高'} {abs(process_speed_adjust_percent):.2f}% )" else: # 米重偏低，需要提高米重 return f"米重偏差 {abs_deviation:.2f}%（偏低），建议调整：\n" \ f"1. 将螺杆转速从当前值调整至 {new_screw_speed:.1f} rpm " \ f"( {'降低' if screw_speed_adjust_percent < 0 else '提高'} {abs(screw_speed_adjust_percent):.2f}% )\n" \ f"2. 将流程主速从当前值调整至 {new_process_speed:.1f} m/min " \ f"( {'降低' if process_speed_adjust_percent < 0 else '提高'} {abs(process_speed_adjust_percent):.2f}% )" def predict_weight(self, model_info, screw_speed, head_pressure, process_speed, screw_temperature, rear_barrel_temperature, front_barrel_temperature, head_temperature): """ 使用模型预测米重 :param model_info: 包含模型和缩放器的模型信息字典 :param screw_speed: 螺杆转速 (rpm) :param head_pressure: 机头压力 (bar) :param process_speed: 流程主速 (m/min) :param screw_temperature: 螺杆温度 (°C) :param rear_barrel_temperature: 后机筒温度 (°C) :param front_barrel_temperature: 前机筒温度 (°C) :param head_temperature: 机头温度 (°C) :return: 预测的米重值 (Kg/m) """ try: # 获取模型需要的特征 required_features = model_info['features'] # 准备输入数据，保持与模型特征顺序一致 input_features = { '螺杆转速': screw_speed, '机头压力': head_pressure, '流程主速': process_speed, '螺杆温度': screw_temperature, '后机筒温度': rear_barrel_temperature, '前机筒温度': front_barrel_temperature, '机头温度': head_temperature } # 根据模型特征创建输入DataFrame input_df = pd.DataFrame([input_features])[required_features] # 初始化预测结果 predicted_weight = None # 获取模型 model = model_info['model'] # 根据模型类型执行不同的预测逻辑 if model_info['model_type'] in ['LSTM', 'GRU', 'BiLSTM']: # 深度学习模型预测 if torch is None: print("PyTorch not available, cannot predict with deep learning models") return None # 数据标准化 scaler_X = model_info['scaler_X'] scaler_y = model_info['scaler_y'] input_scaled = scaler_X.transform(input_df) # 获取序列长度 sequence_length = model_info['sequence_length'] # 为深度学习模型创建序列 input_seq = np.tile(input_scaled, (sequence_length, 1)).reshape(1, sequence_length, -1) # 转换为PyTorch张量 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') input_tensor = torch.tensor(input_seq, dtype=torch.float32).to(device) # 预测 model.eval() with torch.no_grad(): y_pred_scaled_tensor = model(input_tensor) y_pred_scaled = y_pred_scaled_tensor.cpu().numpy().ravel()[0] # 反归一化 predicted_weight = scaler_y.inverse_transform(np.array([[y_pred_scaled]]))[0][0] elif model_info['model_type'] in ['SVR', 'MLP', 'GradientBoosting']: # 需要特征缩放的模型 scaler_X = model_info['scaler_X'] scaler_y = model_info['scaler_y'] input_scaled = scaler_X.transform(input_df) # 预测 y_pred_scaled = model.predict(input_scaled)[0] # 反归一化 predicted_weight = scaler_y.inverse_transform(np.array([[y_pred_scaled]]))[0][0] else: # 其他模型（如随机森林、线性回归等） predicted_weight = model.predict(input_df)[0] return predicted_weight except Exception as e: print(f"模型预测失败: {e}") import traceback traceback.print_exc() return None def iterative_adjustment(self, initial_params, model_info, max_iterations=5, tolerance=0.5): """ 迭代调整参数，直到预测米重满足偏差要求 :param initial_params: 初始参数字典，包含: - real_time_weight: 实时米重 - standard_weight: 标准米重 - upper_limit: 米重上限 - lower_limit: 米重下限 - current_screw_speed: 当前螺杆转速 - current_process_speed: 当前流程主速 - current_screw_temperature: 螺杆温度 - current_rear_barrel_temperature: 后机筒温度 - current_front_barrel_temperature: 前机筒温度 - current_head_temperature: 机头温度 - current_head_pressure: 机头压力 :param model_info: 模型信息字典 :param max_iterations: 最大迭代次数 :param tolerance: 允许的米重偏差百分比阈值 :return: 迭代调整结果字典，包含: - final_result: 最终调整结果 - iteration_history: 每次迭代的历史记录 - converged: 是否收敛 """ iteration_history = [] converged = False # 保存初始参数，用于最终结果计算 original_params = initial_params.copy() # 初始化当前参数为初始参数 current_params = initial_params.copy() for i in range(max_iterations): # 计算当前迭代的调整建议 adjustment_result = self.calculate_adjustment( real_time_weight=current_params['real_time_weight'], standard_weight=current_params['standard_weight'], upper_limit=current_params['upper_limit'], lower_limit=current_params['lower_limit'], current_screw_speed=current_params['current_screw_speed'], current_process_speed=current_params['current_process_speed'], current_screw_temperature=current_params['current_screw_temperature'], current_rear_barrel_temperature=current_params['current_rear_barrel_temperature'], current_front_barrel_temperature=current_params['current_front_barrel_temperature'], current_head_temperature=current_params['current_head_temperature'] ) # 使用模型预测调整后的米重 predicted_weight = self.predict_weight( model_info=model_info, screw_speed=adjustment_result['new_screw_speed'], head_pressure=current_params['current_head_pressure'], process_speed=adjustment_result['new_process_speed'], screw_temperature=current_params['current_screw_temperature'], rear_barrel_temperature=current_params['current_rear_barrel_temperature'], front_barrel_temperature=current_params['current_front_barrel_temperature'], head_temperature=current_params['current_head_temperature'] ) # 检查模型预测是否成功 if predicted_weight is None: print(f"模型预测失败，终止迭代调整") iteration_history.append({ 'iteration': i + 1, 'current_screw_speed': current_params['current_screw_speed'], 'current_process_speed': current_params['current_process_speed'], 'adjusted_screw_speed': adjustment_result['new_screw_speed'], 'adjusted_process_speed': adjustment_result['new_process_speed'], 'predicted_weight': None, 'predicted_deviation': None, 'predicted_deviation_percent': None, 'screw_speed_adjustment': adjustment_result['screw_speed_adjustment'], 'process_speed_adjustment': adjustment_result['process_speed_adjustment'] }) break # 计算预测偏差 predicted_deviation = predicted_weight - current_params['standard_weight'] predicted_deviation_percent = (predicted_deviation / current_params['standard_weight']) * 100 # 保存迭代历史 iteration_history.append({ 'iteration': i + 1, 'current_screw_speed': current_params['current_screw_speed'], 'current_process_speed': current_params['current_process_speed'], 'adjusted_screw_speed': adjustment_result['new_screw_speed'], 'adjusted_process_speed': adjustment_result['new_process_speed'], 'predicted_weight': predicted_weight, 'predicted_deviation': predicted_deviation, 'predicted_deviation_percent': predicted_deviation_percent, 'screw_speed_adjustment': adjustment_result['screw_speed_adjustment'], 'process_speed_adjustment': adjustment_result['process_speed_adjustment'] }) # 检查是否收敛 if abs(predicted_deviation_percent) <= tolerance: converged = True break # 更新当前参数，准备下一次迭代 current_params.update({ 'real_time_weight': predicted_weight, 'current_screw_speed': adjustment_result['new_screw_speed'], 'current_process_speed': adjustment_result['new_process_speed'] }) # 获取最终调整后的参数 final_screw_speed = iteration_history[-1]['adjusted_screw_speed'] final_process_speed = iteration_history[-1]['adjusted_process_speed'] final_predicted_weight = iteration_history[-1]['predicted_weight'] # 计算从初始参数到最终参数的总调整 # 创建一个包含初始参数的调整结果 final_result = { 'status': '正常范围' if abs((final_predicted_weight - original_params['standard_weight']) / original_params['standard_weight'] * 100) <= tolerance else '超上限' if final_predicted_weight > original_params['upper_limit'] else '超下限', 'real_time_weight': original_params['real_time_weight'], # 初始实时米重 'standard_weight': original_params['standard_weight'], 'upper_limit': original_params['upper_limit'], 'lower_limit': original_params['lower_limit'], 'deviation': original_params['real_time_weight'] - original_params['standard_weight'], # 初始偏差 'deviation_percentage': (original_params['real_time_weight'] - original_params['standard_weight']) / original_params['standard_weight'] * 100, # 初始偏差百分比 'current_screw_speed': original_params['current_screw_speed'], # 初始螺杆转速 'current_process_speed': original_params['current_process_speed'], # 初始流程主速 'new_screw_speed': final_screw_speed, # 最终调整后的螺杆转速 'new_process_speed': final_process_speed, # 最终调整后的流程主速 'screw_speed_adjustment': final_screw_speed - original_params['current_screw_speed'], # 总调整量 'process_speed_adjustment': final_process_speed - original_params['current_process_speed'], # 总调整量 'screw_speed_adjust_percent': ((final_screw_speed - original_params['current_screw_speed']) / original_params['current_screw_speed']) * 100 if original_params['current_screw_speed'] != 0 else 0, # 总调整百分比 'process_speed_adjust_percent': ((final_process_speed - original_params['current_process_speed']) / original_params['current_process_speed']) * 100 if original_params['current_process_speed'] != 0 else 0, # 总调整百分比 'predicted_weight': final_predicted_weight # 最终预测米重 } # 生成调整建议文本 final_deviation_percent = (final_predicted_weight - original_params['standard_weight']) / original_params['standard_weight'] * 100 final_result['recommendation'] = self._generate_recommendation( final_deviation_percent, final_result['screw_speed_adjust_percent'], final_screw_speed, final_result['process_speed_adjust_percent'], final_process_speed ) # 添加温度相关参数（用于预测） final_result['current_screw_temperature'] = original_params['current_screw_temperature'] final_result['current_rear_barrel_temperature'] = original_params['current_rear_barrel_temperature'] final_result['current_front_barrel_temperature'] = original_params['current_front_barrel_temperature'] final_result['current_head_temperature'] = original_params['current_head_temperature'] # 添加最终预测相关信息 final_predicted_deviation = final_predicted_weight - original_params['standard_weight'] final_predicted_deviation_percent = (final_predicted_deviation / original_params['standard_weight']) * 100 final_result['final_predicted_deviation'] = final_predicted_deviation final_result['final_predicted_deviation_percent'] = final_predicted_deviation_percent return { 'final_result': final_result, 'iteration_history': iteration_history, 'converged': converged, 'total_iterations': len(iteration_history), 'initial_params': original_params # 保存初始参数，用于结果展示 } def analyze_historical_adjustments(self, df): """ 分析历史调整数据，优化调整系数 :param df: 包含历史调整数据的DataFrame，需要包含以下列： - real_time_weight: 实时米重 - standard_weight: 标准米重 - current_screw_speed: 当前螺杆转速 - current_process_speed: 当前流程主速 - adjusted_screw_speed: 调整后的螺杆转速 - adjusted_process_speed: 调整后的流程主速 - result_weight: 调整后的米重 :return: 优化后的系数 """ if df is None or df.empty: return self.default_coefficients try: # 计算米重偏差 df['weight_deviation'] = df['real_time_weight'] - df['standard_weight'] df['deviation_percentage'] = (df['weight_deviation'] / df['standard_weight']) * 100 # 计算参数调整百分比 df['screw_speed_adjust_percent'] = ((df['adjusted_screw_speed'] - df['current_screw_speed']) / df['current_screw_speed']) * 100 df['process_speed_adjust_percent'] = ((df['adjusted_process_speed'] - df['current_process_speed']) / df['current_process_speed']) * 100 # 计算调整效果 df['adjustment_effect'] = df['result_weight'] - df['real_time_weight'] # 使用线性回归计算优化系数 # 系数 = 参数调整百分比 / 米重偏差百分比 screw_speed_coeff = df['screw_speed_adjust_percent'].sum() / df['deviation_percentage'].sum() if df['deviation_percentage'].sum() != 0 else self.default_coefficients['screw_speed'] process_speed_coeff = df['process_speed_adjust_percent'].sum() / df['deviation_percentage'].sum() if df['deviation_percentage'].sum() != 0 else self.default_coefficients['process_main_speed'] # 取绝对值，确保方向正确 optimized_coeffs = { 'screw_speed': abs(screw_speed_coeff), 'process_main_speed': -abs(process_speed_coeff) # 流程主速与米重负相关 } return optimized_coeffs except Exception as e: print(f"分析历史调整数据失败: {e}") return self.default_coefficients