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（Shredded Woven Polypropylene，切碎编织聚丙烯）改性混凝土热机械性能 # 作者： # 许可证：MIT / CC-BY 4.0 # 数据集描述： 本Jupyter Notebook依照论文所述流程，完成了数据预处理、分段感知数据增强以及机器学习建模（极限梯度提升树XGBoost）工作。 导入所需Python库： 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} → 决定系数R2: {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}")