Robust Design of Polypropylene Waste Modified Concrete Using Physics Guided Machine Learning

# ============================================================================= # TITLE: Thermo-Mechanical Performance of SWPP-Modified Concrete Using ML # AUTHOR: # LICENSE: MIT / CC-BY 4.0 # DESCRIPTION: # This notebook performs data preprocessing, piecewise-aware augmentation, # and machine learning modeling (XGBoost) as described in the paper. # ============================================================================= import os import numpy as np import pandas as pd import matplotlib.pyplot as plt from sklearn.model_selection import train_test_split from sklearn.metrics import r2_score, mean_squared_error from xgboost import XGBRegressor # --------------------------------------------------------- # CONFIGURATION & PATHS (Portable) # --------------------------------------------------------- # Paths are set relative to the 'src' folder to work on any machine DATA_PATH = os.path.join("..", "data", "Data.xlsx") OUT_DIR = os.path.join("..", "outputs") # Create output directory if it doesn't exist os.makedirs(OUT_DIR, exist_ok=True) print(f"Data Source: {DATA_PATH}") print(f"Output Dir : {OUT_DIR}") # ============================================================================= # STEP 1: DATA LOADING & PREPROCESSING (Wide -> Long) # ============================================================================= # 1.1 Load Data if not os.path.exists(DATA_PATH): raise FileNotFoundError(f"Data file not found at {DATA_PATH}. Please check folder structure.") df_raw = pd.read_excel(DATA_PATH) print("Data Loaded. Shape:", df_raw.shape) # 1.2 Column Mapping COL_SAMPLE = "Sample no" COL_SWPP = "SWPP (Shredded woven polypropylene)" # Helper to clean numeric columns (handles comma decimals if any) def numify(x): s = pd.Series(x) if not isinstance(x, pd.Series) else x s = s.astype(str).str.replace(",", ".", regex=False).str.strip() return pd.to_numeric(s, errors="coerce") # Clean SWPP ratio column df_raw["SWPP_ratio_pct"] = df_raw[COL_SWPP].astype(str).str.replace("%", "").str.replace(",", ".").astype(float) # 1.3 Transform WIDE to LONG (Tidy) format # Mapping: Temp -> (CS col, FS col, STS col) temp_map = [ (24, "Compressive Strength-28 day, 24 C", "Flexural Strength-28 day, 24 C", "Splitting Tensile Strength-28 day, 24 C"), (200, "Compressive Strength, after 200 C", "Flexural Strength, after 200 C", "Splitting Tensile Strength, after 200 C"), (300, "Compressive Strength, after 300 C", "Felxural Strength, after 300 C", "Splitting Tensile Strength, after 300 C"), (400, "Compressive Strength, after 400 C", "Felxural Strength, after 400 C", "Splitting Tensile Strength, after 400 C"), (600, "Compressive Strength, after 600 C", "Flexural Strength, after 600 C", "Splitting Tensile Strength, after 600 C"), ] records = [] for _, r in df_raw.iterrows(): for T, c_cs, c_fs, c_sts in temp_map: if c_cs in df_raw.columns: records.append({ "Sample": str(r[COL_SAMPLE]).strip(), "SWPP_ratio_pct": r["SWPP_ratio_pct"], "Temperature_C": T, "CS_MPa": float(numify(r[c_cs]).iloc[0]), "FS_MPa": float(numify(r[c_fs]).iloc[0]), "STS_MPa": float(numify(r[c_sts]).iloc[0]), }) df = pd.DataFrame(records) # 1.4 Define Exposure Regimes (Bins) bins = [-np.inf, 200, 300, 400, np.inf] labels = ["RT-200", "200-300", "300-400", "400-600"] df["Exposure_level"] = pd.cut(df["Temperature_C"], bins=bins, labels=labels, right=False) # Save cleaned data cleaned_path = os.path.join(OUT_DIR, "cleaned_data.xlsx") df.to_excel(cleaned_path, index=False) print(f"Data cleaned and saved to {cleaned_path}") # ============================================================================= # STEP 2: LEAKAGE-SAFE SPLITTING & AUGMENTATION # ============================================================================= # 2.1 Split by Sample to prevent data leakage train_samples, test_samples = train_test_split(df["Sample"].unique(), test_size=0.30, random_state=42) train_df = df[df["Sample"].isin(train_samples)].copy() test_df = df[df["Sample"].isin(test_samples)].copy() print(f"Train samples: {len(train_samples)} | Test samples: {len(test_samples)}") # 2.2 Piecewise-Aware Augmentation Function rng = np.random.default_rng(42) def piecewise_interpolate_regime(df_regime, n_synth, jitter=True): if len(df_regime) < 2: return pd.DataFrame() synth_rows = [] X_swpp = df_regime["SWPP_ratio_pct"].values X_temp = df_regime["Temperature_C"].values Y_mat = df_regime[["CS_MPa", "FS_MPa", "STS_MPa"]].values for _ in range(n_synth): idx = rng.choice(len(df_regime), 2, replace=False) lam = rng.uniform(0.05, 0.95) new_swpp = lam * X_swpp[idx[0]] + (1 - lam) * X_swpp[idx[1]] new_temp = lam * X_temp[idx[0]] + (1 - lam) * X_temp[idx[1]] new_y = lam * Y_mat[idx[0]] + (1 - lam) * Y_mat[idx[1]] if jitter: new_swpp += rng.normal(0, 0.5) new_y *= rng.normal(1.0, 0.02, size=3) row = df_regime.iloc[idx[0]].copy() row["Sample"] = f"SYN_{int(rng.random()*10000)}" row["SWPP_ratio_pct"] = np.clip(new_swpp, 0, 100) row["Temperature_C"] = new_temp row[["CS_MPa", "FS_MPa", "STS_MPa"]] = np.maximum(new_y, 0) row["is_synthetic"] = True synth_rows.append(row) return pd.DataFrame(synth_rows) # 2.3 Apply Augmentation aug_list = [train_df.assign(is_synthetic=False)] MULTIPLIER = 2 for regime in labels: regime_data = train_df[train_df["Exposure_level"] == regime] n_synth = int(len(regime_data) * MULTIPLIER) if n_synth > 0: synth_df = piecewise_interpolate_regime(regime_data, n_synth) aug_list.append(synth_df) train_aug = pd.concat(aug_list, ignore_index=True) aug_path = os.path.join(OUT_DIR, "train_augmented.xlsx") train_aug.to_excel(aug_path, index=False) print(f"Augmentation complete. Saved to {aug_path}") # ============================================================================= # STEP 3: ML MODELING (XGBoost) # ============================================================================= features = ["SWPP_ratio_pct", "Temperature_C"] targets = ["CS_MPa", "FS_MPa", "STS_MPa"] model = XGBRegressor(n_estimators=300, max_depth=3, learning_rate=0.05, random_state=42) print("\n--- Model Performance on Test Set ---") for target in targets: model.fit(train_aug[features], train_aug[target]) y_pred = model.predict(test_df[features]) y_true = test_df[target] r2 = r2_score(y_true, y_pred) rmse = mean_squared_error(y_true, y_pred, squared=False) print(f"{target} -> R2: {r2:.4f} | RMSE: {rmse:.4f}") # Save Plot plt.figure(figsize=(5,5)) plt.scatter(y_true, y_pred, alpha=0.7) plt.plot([y_true.min(), y_true.max()], [y_true.min(), y_true.max()], 'r--') plt.xlabel("Experimental") plt.ylabel("Predicted") plt.title(f"{target} Prediction") plt.tight_layout() plt.savefig(os.path.join(OUT_DIR, f"Parity_{target}.png"), dpi=300) plt.close() print(f"\nAll outputs saved to: {OUT_DIR}")# 标题：采用机器学习方法的SWPP改性混凝土热力力学性能研究（SWPP，Shredded Woven Polypropylene，切碎机织聚丙烯） # 作者：无 # 许可协议：MIT / CC-BY 4.0 # 数据集描述： # 本Jupyter Notebook依照发表论文中的研究框架，完成数据预处理、分段感知数据增强以及机器学习建模（XGBoost）工作。 # ============================================================================= # 导入依赖库 import os import numpy as np import pandas as pd import matplotlib.pyplot as plt from sklearn.model_selection import train_test_split from sklearn.metrics import r2_score, mean_squared_error from xgboost import XGBRegressor # --------------------------------------------------------- # 配置与路径（可移植性适配） # --------------------------------------------------------- # 路径设置基于src文件夹，以适配所有运行环境 DATA_PATH = os.path.join("..", "data", "Data.xlsx") OUT_DIR = os.path.join("..", "outputs") # 若输出目录不存在则自动创建 os.makedirs(OUT_DIR, exist_ok=True) print(f"数据源路径: {DATA_PATH}") print(f"输出目录路径: {OUT_DIR}") # ============================================================================= # 步骤1：数据加载与预处理（宽格式转长格式） # ============================================================================= # 1.1 数据加载 if not os.path.exists(DATA_PATH): raise FileNotFoundError(f"未在路径 {DATA_PATH} 找到数据文件，请检查文件夹结构。") df_raw = pd.read_excel(DATA_PATH) print("数据加载完成。数据维度:", df_raw.shape) # 1.2 列名映射 COL_SAMPLE = "样本编号" COL_SWPP = "SWPP（切碎机织聚丙烯）掺量" # 数值转换工具函数（处理逗号小数格式） def numify(x): s = pd.Series(x) if not isinstance(x, pd.Series) else x s = s.astype(str).str.replace(",", ".", regex=False).str.strip() return pd.to_numeric(s, errors="coerce") # 清理SWPP掺量列 df_raw["SWPP_ratio_pct"] = df_raw[COL_SWPP].astype(str).str.replace("%", "").str.replace(",", ".").astype(float) # 1.3 宽格式转整洁长格式 # 温度映射表：温度 -> (抗压强度列名, 抗折强度列名, 劈裂抗拉强度列名) temp_map = [ (24, "28天抗压强度，24℃", "28天抗折强度，24℃", "28天劈裂抗拉强度，24℃"), (200, "200℃受热后抗压强度", "200℃受热后抗折强度", "200℃受热后劈裂抗拉强度"), (300, "300℃受热后抗压强度", "300℃受热后抗折强度", "300℃受热后劈裂抗拉强度"), (400, "400℃受热后抗压强度", "400℃受热后抗折强度", "400℃受热后劈裂抗拉强度"), (600, "600℃受热后抗压强度", "600℃受热后抗折强度", "600℃受热后劈裂抗拉强度"), ] records = [] for _, r in df_raw.iterrows(): for T, c_cs, c_fs, c_sts in temp_map: if c_cs in df_raw.columns: records.append({ "样本编号": str(r[COL_SAMPLE]).strip(), "SWPP掺量百分比": r["SWPP_ratio_pct"], "受热温度_℃": T, "抗压强度_MPa": float(numify(r[c_cs]).iloc[0]), "抗折强度_MPa": float(numify(r[c_fs]).iloc[0]), "劈裂抗拉强度_MPa": float(numify(r[c_sts]).iloc[0]), }) df = pd.DataFrame(records) # 1.4 受热暴露区间分箱 bins = [-np.inf, 200, 300, 400, np.inf] labels = ["常温-200℃", "200-300℃", "300-400℃", "400-600℃"] df["受热暴露等级"] = pd.cut(df["受热温度_℃"], bins=bins, labels=labels, right=False) # 保存清理后的数据 cleaned_path = os.path.join(OUT_DIR, "cleaned_data.xlsx") df.to_excel(cleaned_path, index=False) print(f"数据清理完成，已保存至 {cleaned_path}") # ============================================================================= # 步骤2：防止数据泄露的数据集拆分与数据增强 # ============================================================================= # 2.1 按样本拆分数据集以避免数据泄露 train_samples, test_samples = train_test_split(df["样本编号"].unique(), test_size=0.30, random_state=42) train_df = df[df["样本编号"].isin(train_samples)].copy() test_df = df[df["样本编号"].isin(test_samples)].copy() print(f"训练集样本数: {len(train_samples)} | 测试集样本数: {len(test_samples)}") # 2.2 分段感知数据增强函数 rng = np.random.default_rng(42) def piecewise_interpolate_regime(df_regime, n_synth, jitter=True): if len(df_regime) < 2: return pd.DataFrame() synth_rows = [] X_swpp = df_regime["SWPP掺量百分比"].values X_temp = df_regime["受热温度_℃"].values Y_mat = df_regime[["抗压强度_MPa", "抗折强度_MPa", "劈裂抗拉强度_MPa"]].values for _ in range(n_synth): idx = rng.choice(len(df_regime), 2, replace=False) lam = rng.uniform(0.05, 0.95) new_swpp = lam * X_swpp[idx[0]] + (1 - lam) * X_swpp[idx[1]] new_temp = lam * X_temp[idx[0]] + (1 - lam) * X_temp[idx[1]] new_y = lam * Y_mat[idx[0]] + (1 - lam) * Y_mat[idx[1]] if jitter: new_swpp += rng.normal(0, 0.5) new_y *= rng.normal(1.0, 0.02, size=3) row = df_regime.iloc[idx[0]].copy() row["样本编号"] = f"SYN_{int(rng.random()*10000)}" row["SWPP掺量百分比"] = np.clip(new_swpp, 0, 100) row["受热温度_℃"] = new_temp row[["抗压强度_MPa", "抗折强度_MPa", "劈裂抗拉强度_MPa"]] = np.maximum(new_y, 0) row["是否为合成样本"] = True synth_rows.append(row) return pd.DataFrame(synth_rows) # 2.3 执行数据增强 aug_list = [train_df.assign(是否为合成样本=False)] MULTIPLIER = 2 for regime in labels: regime_data = train_df[train_df["受热暴露等级"] == regime] n_synth = int(len(regime_data) * MULTIPLIER) if n_synth > 0: synth_df = piecewise_interpolate_regime(regime_data, n_synth) aug_list.append(synth_df) train_aug = pd.concat(aug_list, ignore_index=True) aug_path = os.path.join(OUT_DIR, "train_augmented.xlsx") train_aug.to_excel(aug_path, index=False) print(f"数据增强完成，已保存至 {aug_path}") # ============================================================================= # 步骤3：机器学习建模（XGBoost） # ============================================================================= features = ["SWPP掺量百分比", "受热温度_℃"] targets = ["抗压强度_MPa", "抗折强度_MPa", "劈裂抗拉强度_MPa"] model = XGBRegressor(n_estimators=300, max_depth=3, learning_rate=0.05, random_state=42) print(" --- 测试集模型性能 ---") for target in targets: model.fit(train_aug[features], train_aug[target]) y_pred = model.predict(test_df[features]) y_true = test_df[target] r2 = r2_score(y_true, y_pred) rmse = mean_squared_error(y_true, y_pred, squared=False) print(f"{target} -> 决定系数R²: {r2:.4f} | 均方根误差RMSE: {rmse:.4f}") # 绘制散点图并保存 plt.figure(figsize=(5,5)) plt.scatter(y_true, y_pred, alpha=0.7) plt.plot([y_true.min(), y_true.max()], [y_true.min(), y_true.max()], 'r--') plt.xlabel("实测值") plt.ylabel("预测值") plt.title(f"{target} 预测结果") plt.tight_layout() plt.savefig(os.path.join(OUT_DIR, f"Parity_{target}.png"), dpi=300) plt.close() print(f" 所有输出结果已保存至: {OUT_DIR}")