import streamlit as st import plotly.express as px import plotly.graph_objects as go import pandas as pd import numpy as np import joblib import os from datetime import datetime # 尝试导入torch，如果失败则禁用深度学习模型支持 try: import torch TORCH_AVAILABLE = True except ImportError: TORCH_AVAILABLE = False # 页面函数定义 def show_metered_weight_prediction(): # 页面标题 st.title("米重统一预测") # 初始化会话状态 if 'selected_model' not in st.session_state: st.session_state['selected_model'] = None # 创建模型目录（如果不存在） model_dir = "saved_models" os.makedirs(model_dir, exist_ok=True) # 获取所有已保存的模型文件 model_files = [f for f in os.listdir(model_dir) if f.endswith('.joblib')] model_files.sort(reverse=True) # 最新的模型排在前面 # 模型选择区域 with st.expander("📁 选择模型", expanded=True): if not model_files: st.warning("尚未保存任何模型，请先训练模型并保存。") else: # 模型选择下拉框 selected_model_file = st.selectbox( "选择已保存的模型", options=model_files, help="选择要用于预测的模型文件" ) # 加载并显示模型信息 if selected_model_file: model_path = os.path.join(model_dir, selected_model_file) model_info = joblib.load(model_path) # 显示模型基本信息 st.subheader("📊 模型信息") info_cols = st.columns(2) with info_cols[0]: st.metric("模型类型", model_info['model_type']) st.metric("创建时间", model_info['created_at'].strftime('%Y-%m-%d %H:%M:%S')) st.metric("使用稳态数据", "是" if model_info.get('use_steady_data', False) else "否") with info_cols[1]: st.metric("R² 得分", f"{model_info['r2_score']:.4f}") st.metric("均方误差 (MSE)", f"{model_info['mse']:.6f}") st.metric("均方根误差 (RMSE)", f"{model_info['rmse']:.6f}") # 显示模型特征 st.write("🔑 模型使用的特征:") st.code(", ".join(model_info['features'])) # 如果是深度学习模型，显示序列长度 if 'sequence_length' in model_info: st.metric("序列长度", model_info['sequence_length']) # 保存模型信息到会话状态 st.session_state['selected_model'] = model_info st.session_state['selected_model_file'] = selected_model_file # 预测功能区域 st.subheader("🔮 米重预测") if st.session_state['selected_model']: model_info = st.session_state['selected_model'] # 获取模型需要的特征 required_features = model_info['features'] # 创建预测表单 st.write("输入特征值进行米重预测:") predict_cols = st.columns(2) input_features = {} # 显示输入表单 for i, feature in enumerate(required_features): with predict_cols[i % 2]: input_features[feature] = st.number_input( f"{feature}", key=f"pred_{feature}", value=0.0, step=0.0001, format="%.4f" ) # 预测按钮 if st.button("🚀 开始预测"): try: # 准备预测数据 input_df = pd.DataFrame([input_features]) # 根据模型类型执行不同的预测逻辑 predicted_weight = None # 获取模型 model = model_info['model'] # 检查模型类型并执行预测 if model_info['model_type'] in ['LSTM', 'GRU', 'BiLSTM']: # 深度学习模型预测 if not TORCH_AVAILABLE: st.error("PyTorch 未安装，无法使用深度学习模型进行预测。") return # 数据标准化 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张量 import torch 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']: # 支持向量机或多层感知器预测 # 数据标准化 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] # 显示预测结果 st.success(f"预测米重: {predicted_weight:.4f} Kg/m") except Exception as e: st.error(f"预测失败: {str(e)}") else: st.warning("请先选择一个模型。") # 模型管理区域 if model_files: with st.expander("🗑️ 模型管理", expanded=False): st.write("管理已保存的模型文件:") # 显示所有模型文件 for model_file in model_files: cols = st.columns([3, 1, 1]) cols[0].write(model_file) # 查看模型信息按钮 if cols[1].button("查看", key=f"view_{model_file}", help="查看模型信息"): model_path = os.path.join(model_dir, model_file) model_info = joblib.load(model_path) st.write("模型详细信息:") st.json({ 'model_type': model_info['model_type'], 'created_at': model_info['created_at'].strftime('%Y-%m-%d %H:%M:%S'), 'r2_score': f"{model_info['r2_score']:.4f}", 'mse': f"{model_info['mse']:.6f}", 'mae': f"{model_info['mae']:.6f}", 'rmse': f"{model_info['rmse']:.6f}", 'features': model_info['features'], 'use_steady_data': model_info.get('use_steady_data', False) }) # 删除模型按钮 if cols[2].button("删除", key=f"delete_{model_file}", help="删除模型文件", type="primary"): model_path = os.path.join(model_dir, model_file) os.remove(model_path) st.success(f"已删除模型: {model_file}") st.rerun() # 页面入口 if __name__ == "__main__": show_metered_weight_prediction()