04f9a647d8
1. 自适应反射率量级缩放 (inference_pipeline/batch_inference/batch_inference_multi_data):
- 检测光谱列 max > 10 时自动 ÷10000 统一到 0-1 区间
- 三个推理入口全量同步保护
2. Pipeline 列精确对齐 (preprocess_spectra 两处):
- 旧: not in 盲目排除多余列 → 140个光谱列被当多余移除
- 新: 遍历 train_cols，精确字符串匹配 → math.isclose 浮点近似
匹配列重命名为训练列名，缺失列补零
最终 spectra[train_cols] 严格按训练顺序输出
- 兼容路径同步修复
3. 误导性日志修正:
- 改前: '正在应用预处理方法: D1' → 让用户以为推理端在手动做 D1
- 改后: '[模型信息] 训练预处理方法: D1 — 由 Pipeline 内部自动执行'
1707 lines
77 KiB
Python
1707 lines
77 KiB
Python
import numpy as np
|
||
import pandas as pd
|
||
import joblib
|
||
import os
|
||
import math
|
||
from pathlib import Path
|
||
from typing import List, Dict, Union, Tuple, Optional
|
||
import warnings
|
||
|
||
warnings.filterwarnings('ignore')
|
||
|
||
# 导入预处理模块 - 动态添加路径支持
|
||
import sys
|
||
import os
|
||
|
||
from src.preprocessing.spectral_Preprocessing import Preprocessing, get_preprocessing_transformer
|
||
from src.core.utils.split_methods import spxy, ks
|
||
|
||
# try:
|
||
# from modeling import WaterQualityModeling
|
||
# except ImportError:
|
||
# from src.core.modeling.modeling_batch import WaterQualityModeling
|
||
|
||
# 机器学习相关导入
|
||
from sklearn.model_selection import train_test_split
|
||
from sklearn.pipeline import Pipeline
|
||
|
||
|
||
class WaterQualityInference:
|
||
"""水质参数反演推理类"""
|
||
|
||
def __init__(self, artifacts_dir: str = "models/artifacts",
|
||
external_model=None, external_model_path=None):
|
||
"""
|
||
初始化推理类
|
||
|
||
Args:
|
||
artifacts_dir: 模型保存目录
|
||
external_model: 外部预训练模型对象（来自 GUI 导入，跳过磁盘加载）
|
||
external_model_path: 外部模型文件路径（仅用于日志）
|
||
"""
|
||
self.artifacts_dir = Path(artifacts_dir)
|
||
if not self.artifacts_dir.exists():
|
||
print(f"警告: 模型目录不存在: {artifacts_dir}，将在需要时创建")
|
||
|
||
self.best_model_info = None
|
||
self.external_model = external_model
|
||
self.external_model_path = external_model_path
|
||
|
||
# 规范化 loaded_model_data：始终为 dict，确保 ['model'] 访问不崩溃
|
||
if external_model is not None:
|
||
# ★ 外部模型可能是完整 dict（含 model + metadata + train_wavelengths），
|
||
# 也可能是裸 Pipeline 对象（旧版兼容）
|
||
if isinstance(external_model, dict) and 'model' in external_model:
|
||
self.loaded_model_data = external_model
|
||
print(f" 外部模型已规范化: dict (含 metadata)")
|
||
else:
|
||
self.loaded_model_data = {'model': external_model,
|
||
'preprocess_method': 'None'}
|
||
print(f" 外部模型已规范化: type={type(external_model).__name__}")
|
||
else:
|
||
self.loaded_model_data = None
|
||
|
||
def load_sampling_data(self, csv_path: str) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
|
||
"""
|
||
加载sampling生成的CSV数据（兼容 WQI 增强版 CSV）
|
||
|
||
Args:
|
||
csv_path: CSV文件路径
|
||
旧版：x_coord,y_coord,pixel_x,pixel_y,波长...
|
||
新版：x_coord,y_coord,WQI_...,波长...
|
||
|
||
Returns:
|
||
coords: 经纬度数据 (DataFrame, 2列)
|
||
spectra: 纯光谱数据 (DataFrame, 跳过 WQI 列)
|
||
wqi_df: WQI 指数列 (DataFrame, 0或45列)
|
||
"""
|
||
print(f"正在加载采样数据: {csv_path}")
|
||
|
||
if not os.path.exists(csv_path):
|
||
raise FileNotFoundError(f"采样数据文件不存在: {csv_path}")
|
||
|
||
# 读取CSV文件
|
||
data = pd.read_csv(csv_path)
|
||
|
||
print(f"采样数据加载完成:")
|
||
print(f" 数据形状: {data.shape}")
|
||
print(f" 列名: {list(data.columns[:5])}...") # 只显示前5列
|
||
|
||
# 检查数据列数
|
||
if data.shape[1] < 4:
|
||
raise ValueError(f"数据列数不足，期望至少4列（经度、纬度、其他列、光谱数据），实际得到{data.shape[1]}列")
|
||
|
||
# 前两列为经纬度
|
||
coords = data.iloc[:, :2].copy()
|
||
coords.columns = ['longitude', 'latitude']
|
||
|
||
# 动态识别光谱列（兼容 sampling_spectra.csv 列顺序变更）
|
||
# 列名约定：波长为纯数字字符串如 "374.285004"；WQI 为 "WQI_xxx" 前缀
|
||
# 旧版 CSV（无WQI）：x_coord,y_coord,pixel_x,pixel_y,波长... → 取 [4:]
|
||
# 新版 CSV（有WQI）：x_coord,y_coord,WQI_...,波长... → 过滤 WQI 列后取光谱
|
||
all_cols = list(data.columns)
|
||
spectral_col_indices = []
|
||
wqi_col_indices = []
|
||
for i, col in enumerate(all_cols):
|
||
col_str = str(col)
|
||
if col_str.startswith('WQI_'):
|
||
wqi_col_indices.append(i)
|
||
elif col_str.replace('.', '').lstrip('-').isdigit():
|
||
# 波长列：纯数字字符串
|
||
spectral_col_indices.append(i)
|
||
else:
|
||
# 其他元数据列（x_coord/y_coord/pixel_x/pixel_y），由 coords 接收
|
||
pass
|
||
|
||
# 光谱列 = 纯数字列（WQI 已被排除）
|
||
spectra = data.iloc[:, spectral_col_indices].copy() if spectral_col_indices else data.iloc[:, 4:].copy()
|
||
# WQI 列（用于追加到预测结果输出）
|
||
wqi_df = data.iloc[:, wqi_col_indices].copy() if wqi_col_indices else pd.DataFrame()
|
||
|
||
print(f" 经纬度数据形状: {coords.shape}")
|
||
print(f" 光谱数据形状: {spectra.shape} (自动识别波长列，排除 {len(wqi_col_indices)} 个WQI列)")
|
||
print(f" 经纬度范围: 经度[{coords['longitude'].min():.6f}, {coords['longitude'].max():.6f}], "
|
||
f"纬度[{coords['latitude'].min():.6f}, {coords['latitude'].max():.6f}]")
|
||
|
||
return coords, spectra, wqi_df
|
||
|
||
def random(self, data, label, test_ratio=0.2, random_state=123):
|
||
"""
|
||
随机划分数据集
|
||
|
||
Args:
|
||
data: shape (n_samples, n_features)
|
||
label: shape (n_sample, )
|
||
test_ratio: 测试集比例，默认: 0.2
|
||
random_state: 随机种子，默认: 123
|
||
|
||
Returns:
|
||
X_train: (n_samples, n_features)
|
||
X_test: (n_samples, n_features)
|
||
y_train: (n_sample, )
|
||
y_test: (n_sample, )
|
||
"""
|
||
X_train, X_test, y_train, y_test = train_test_split(
|
||
data, label, test_size=test_ratio, random_state=random_state
|
||
)
|
||
return X_train, X_test, y_train, y_test
|
||
|
||
def spxy(self, data, label, test_size=0.2):
|
||
"""SPXY算法划分数据集（委托至 src.core.utils.split_methods.spxy）"""
|
||
return spxy(data, label, test_size=test_size)
|
||
|
||
def ks(self, data, label, test_size=0.2):
|
||
"""Kennard-Stone算法划分数据集（委托至 src.core.utils.split_methods.ks）"""
|
||
return ks(data, label, test_size=test_size)
|
||
|
||
def split_data(self, X: np.ndarray, y: pd.Series, method: str = "random",
|
||
test_size: float = 0.2, random_state: int = 42) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
|
||
"""
|
||
根据指定方法划分数据集
|
||
|
||
Args:
|
||
X: 特征数据
|
||
y: 目标值数据
|
||
method: 划分方法 ("random", "spxy", "ks")
|
||
test_size: 测试集比例
|
||
random_state: 随机种子（仅对random方法有效）
|
||
|
||
Returns:
|
||
X_train, X_test, y_train, y_test
|
||
"""
|
||
print(f"使用 {method} 方法划分数据集")
|
||
|
||
if method == "random":
|
||
return self.random(X, y, test_ratio=test_size, random_state=random_state)
|
||
elif method == "spxy":
|
||
return self.spxy(X, y, test_size=test_size)
|
||
elif method == "ks":
|
||
return self.ks(X, y, test_size=test_size)
|
||
else:
|
||
raise ValueError(f"不支持的划分方法: {method}. 支持的方法: ['random', 'spxy', 'ks']")
|
||
|
||
def get_best_model_from_summary(self, metric: str = 'test_r2') -> Tuple[str, str]:
|
||
"""
|
||
从训练摘要中获取最佳模型信息
|
||
|
||
Args:
|
||
metric: 评估指标（默认使用test_r2，回归任务的主要指标）
|
||
|
||
Returns:
|
||
preprocess_method: 预处理方法
|
||
model_name: 模型名称
|
||
"""
|
||
# 获取当前artifacts_dir的文件夹名称（用作目标列名）
|
||
folder_name = self.artifacts_dir.name
|
||
|
||
# 尝试加载详细结果文件（使用新的命名格式）
|
||
detailed_path = self.artifacts_dir / f"{folder_name}_detailed_results.csv"
|
||
summary_path = self.artifacts_dir / f"{folder_name}_training_summary.csv"
|
||
|
||
# 备用的旧格式文件路径
|
||
old_detailed_path = self.artifacts_dir / "detailed_results.csv"
|
||
old_summary_path = self.artifacts_dir / "training_summary.csv"
|
||
|
||
summary_df = None
|
||
|
||
# 优先使用新格式的详细结果文件
|
||
if detailed_path.exists():
|
||
print(f"使用详细结果文件: {detailed_path}")
|
||
summary_df = pd.read_csv(detailed_path)
|
||
# 将中文列名映射到英文
|
||
metric_mapping = {
|
||
'test_r2': '测试集R²',
|
||
'train_r2': '训练集R²',
|
||
'test_rmse': '测试集RMSE',
|
||
'train_rmse': '训练集RMSE',
|
||
'cv_mean': 'CV均值'
|
||
}
|
||
if metric in metric_mapping and metric_mapping[metric] in summary_df.columns:
|
||
metric_col = metric_mapping[metric]
|
||
else:
|
||
metric_col = metric
|
||
elif summary_path.exists():
|
||
print(f"使用训练摘要文件: {summary_path}")
|
||
summary_df = pd.read_csv(summary_path)
|
||
metric_col = metric
|
||
elif old_detailed_path.exists():
|
||
print(f"使用旧格式详细结果文件: {old_detailed_path}")
|
||
summary_df = pd.read_csv(old_detailed_path)
|
||
# 将中文列名映射到英文
|
||
metric_mapping = {
|
||
'test_r2': '测试集R²',
|
||
'train_r2': '训练集R²',
|
||
'test_rmse': '测试集RMSE',
|
||
'train_rmse': '训练集RMSE',
|
||
'cv_mean': 'CV均值'
|
||
}
|
||
if metric in metric_mapping and metric_mapping[metric] in summary_df.columns:
|
||
metric_col = metric_mapping[metric]
|
||
else:
|
||
metric_col = metric
|
||
elif old_summary_path.exists():
|
||
print(f"使用旧格式训练摘要文件: {old_summary_path}")
|
||
summary_df = pd.read_csv(old_summary_path)
|
||
metric_col = metric
|
||
else:
|
||
raise FileNotFoundError(f"训练摘要文件不存在，尝试的路径:\n"
|
||
f" - {detailed_path}\n"
|
||
f" - {summary_path}\n"
|
||
f" - {old_detailed_path}\n"
|
||
f" - {old_summary_path}")
|
||
|
||
if summary_df.empty:
|
||
raise ValueError("训练摘要为空")
|
||
|
||
# 检查指标列是否存在
|
||
if metric_col not in summary_df.columns:
|
||
available_cols = list(summary_df.columns)
|
||
raise ValueError(f"指标 '{metric_col}' 不存在。可用列: {available_cols}")
|
||
|
||
# 获取最佳模型（对于R²等指标，值越大越好）
|
||
if 'r2' in metric.lower() or 'score' in metric.lower():
|
||
best_idx = summary_df[metric_col].idxmax()
|
||
else: # 对于RMSE、MAE等，值越小越好
|
||
best_idx = summary_df[metric_col].idxmin()
|
||
|
||
best_row = summary_df.loc[best_idx]
|
||
|
||
# 根据文件类型解析模型信息
|
||
if (detailed_path.exists() or old_detailed_path.exists()) and '划分方法' in summary_df.columns:
|
||
# 详细结果文件格式
|
||
split_method = best_row['划分方法']
|
||
preprocess_method = best_row['预处理方法']
|
||
model_name = best_row['建模方法']
|
||
|
||
# 处理 nan/NaN/None 值，转换为 "None" 字符串
|
||
if pd.isna(preprocess_method) or str(preprocess_method).lower() in ['nan', 'none', '']:
|
||
preprocess_method = "None"
|
||
|
||
best_combination = f"{split_method}_{preprocess_method}_{model_name}"
|
||
else:
|
||
# 简化结果文件格式
|
||
best_combination = best_row['combination']
|
||
# 解析组合名称（格式: split_method_preprocess_method_model_name）
|
||
parts = best_combination.split('_')
|
||
if len(parts) < 3:
|
||
raise ValueError(f"无效的模型组合名称格式: {best_combination}")
|
||
|
||
split_method = parts[0]
|
||
preprocess_method = parts[1]
|
||
model_name = '_'.join(parts[2:])
|
||
|
||
# 处理 nan/NaN/None 值，转换为 "None" 字符串
|
||
if pd.isna(preprocess_method) or str(preprocess_method).lower() in ['nan', 'none', '']:
|
||
preprocess_method = "None"
|
||
|
||
print(f"最佳模型组合: {best_combination}")
|
||
print(f" 划分方法: {split_method}")
|
||
print(f" 预处理方法: {preprocess_method}")
|
||
print(f" 模型名称: {model_name}")
|
||
print(f" {metric_col}: {best_row[metric_col]:.4f}")
|
||
|
||
self.best_model_info = {
|
||
'combination': best_combination,
|
||
'split_method': split_method,
|
||
'preprocess_method': preprocess_method,
|
||
'model_name': model_name,
|
||
'metric_value': best_row[metric_col]
|
||
}
|
||
|
||
# 返回用于加载模型的文件名格式
|
||
model_file_prefix = f"{split_method}_{preprocess_method}"
|
||
return model_file_prefix, model_name
|
||
|
||
def load_best_model(self, metric: str = 'test_r2'):
|
||
"""
|
||
加载最佳模型
|
||
|
||
Args:
|
||
metric: 评估指标
|
||
"""
|
||
model_file_prefix, model_name = self.get_best_model_from_summary(metric)
|
||
|
||
# 获取当前artifacts_dir的文件夹名称（用作目标列名）
|
||
folder_name = self.artifacts_dir.name
|
||
|
||
# 构建模型文件路径（新格式：包含目标列名）
|
||
filename = f"{folder_name}_{model_file_prefix}_{model_name}.joblib"
|
||
filepath = self.artifacts_dir / filename
|
||
|
||
# 如果新格式文件不存在，尝试旧格式
|
||
if not filepath.exists():
|
||
old_filename = f"{model_file_prefix}_{model_name}.joblib"
|
||
old_filepath = self.artifacts_dir / old_filename
|
||
if old_filepath.exists():
|
||
filepath = old_filepath
|
||
filename = old_filename
|
||
print(f"使用旧格式模型文件: {filepath}")
|
||
else:
|
||
raise FileNotFoundError(f"模型文件不存在，尝试的路径:\n"
|
||
f" - {filepath}\n"
|
||
f" - {old_filepath}")
|
||
else:
|
||
print(f"使用新格式模型文件: {filepath}")
|
||
|
||
print(f"正在加载模型: {filepath}")
|
||
|
||
# 加载模型数据
|
||
self.loaded_model_data = joblib.load(filepath)
|
||
|
||
print("模型加载完成:")
|
||
print(f" 预处理方法: {self.loaded_model_data['preprocess_method']}")
|
||
print(f" 模型名称: {self.loaded_model_data['model_name']}")
|
||
print(f" 模型类型: {type(self.loaded_model_data['model'])}")
|
||
|
||
if 'metadata' in self.loaded_model_data:
|
||
metadata = self.loaded_model_data['metadata']
|
||
print(f" 数据形状: {metadata.get('data_shape', 'Unknown')}")
|
||
print(f" 目标范围: {metadata.get('target_range', 'Unknown')}")
|
||
if 'test_r2' in metadata:
|
||
print(f" 测试集R²: {metadata['test_r2']:.4f}")
|
||
if 'test_rmse' in metadata:
|
||
print(f" 测试集RMSE: {metadata['test_rmse']:.4f}")
|
||
|
||
# ★ 加载时打印 MNF 波段选择信息（从已训练的 Pipeline 中提取）
|
||
self._print_mnf_info_if_available()
|
||
|
||
def load_specific_model(self, model_file_path: str):
|
||
"""
|
||
加载指定的模型文件
|
||
|
||
Args:
|
||
model_file_path: 模型文件路径
|
||
"""
|
||
if not os.path.exists(model_file_path):
|
||
raise FileNotFoundError(f"模型文件不存在: {model_file_path}")
|
||
|
||
print(f"正在加载指定模型: {model_file_path}")
|
||
|
||
# 加载模型数据
|
||
self.loaded_model_data = joblib.load(model_file_path)
|
||
|
||
print("模型加载完成:")
|
||
print(f" 预处理方法: {self.loaded_model_data['preprocess_method']}")
|
||
print(f" 模型名称: {self.loaded_model_data['model_name']}")
|
||
print(f" 模型类型: {type(self.loaded_model_data['model'])}")
|
||
|
||
# ★ 加载时打印 MNF 波段选择信息
|
||
self._print_mnf_info_if_available()
|
||
|
||
def _auto_detect_train_wavelengths(self):
|
||
"""自动获取训练波长：优先级 json/txt > 工作目录 CSV > None"""
|
||
import os as _os, json as _json, glob as _glob
|
||
_dir = str(self.artifacts_dir)
|
||
# 搜集所有可能的搜索根目录
|
||
_roots = [_dir, _os.path.dirname(_dir), _os.getcwd()]
|
||
# 找所有 work_dir 级别的父目录
|
||
for _r in list(_roots):
|
||
_p = _os.path.dirname(_r)
|
||
if _p and _p not in _roots:
|
||
_roots.append(_p)
|
||
|
||
# 优先级 1: 显式 wavelength 文件（JSON/TXT）
|
||
for _root in _roots:
|
||
for _fname in ('train_wavelengths.json', 'train_wavelengths.txt'):
|
||
_p = _os.path.join(_root, _fname)
|
||
if not _os.path.isfile(_p):
|
||
continue
|
||
try:
|
||
if _fname.endswith('.json'):
|
||
wl = _json.load(open(_p))
|
||
if wl:
|
||
print(f"[波长回填] 来源: {_p}")
|
||
return wl
|
||
else:
|
||
wl = [float(x) for x in open(_p).read().strip().split()]
|
||
if wl:
|
||
print(f"[波长回填] 来源: {_p}")
|
||
return wl
|
||
except Exception:
|
||
pass
|
||
|
||
# 优先级 2: training_spectra.csv / sampling_spectra.csv
|
||
for _root in _roots:
|
||
for _sub in ('4_sampling', '6_Spectral_Feature_Extraction', ''):
|
||
_d = _os.path.join(_root, _sub) if _sub else _root
|
||
for _pat in ('sampling_spectra.csv', 'training_spectra.csv'):
|
||
_p = _os.path.join(_d, _pat)
|
||
if not _os.path.isfile(_p):
|
||
continue
|
||
try:
|
||
_df = pd.read_csv(_p, nrows=0)
|
||
_wl = [float(c) for c in _df.columns
|
||
if c.replace('.','').lstrip('-').isdigit()]
|
||
if _wl:
|
||
print(f"[波长回填] 来源: {_p} ({len(_wl)} 波长)")
|
||
return _wl
|
||
except Exception:
|
||
pass
|
||
|
||
return None
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# DualStream_MNF 专用预处理：纯光谱重采样 → Pipeline 全自动
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
@staticmethod
|
||
def _check_spectral_coverage(train_wl, infer_wl):
|
||
"""光谱覆盖率智能预警：检测预测数据波长范围是否充分覆盖训练波长。
|
||
|
||
若预测波长的起止端与训练波长差距超过 15nm，说明边缘波段缺失，
|
||
np.interp 会依赖 left/right 恒定外推补齐，可能导致特征丢失。
|
||
"""
|
||
train_min, train_max = np.min(train_wl), np.max(train_wl)
|
||
infer_min, infer_max = np.min(infer_wl), np.max(infer_wl)
|
||
|
||
gap_left = infer_min - train_min
|
||
gap_right = train_max - infer_max
|
||
|
||
if gap_left > 15:
|
||
print(f"\033[93m[WARN] 预测数据起始波长 ({infer_min:.1f}nm) 晚于"
|
||
f" 训练波长 ({train_min:.1f}nm) 达 {gap_left:.0f}nm！"
|
||
f"系统将自动向左横推补齐，这可能会导致蓝端/紫外特征丢失。\033[0m")
|
||
if gap_right > 15:
|
||
print(f"\033[93m[WARN] 预测数据截止波长 ({infer_max:.1f}nm) 短于"
|
||
f" 训练波长 ({train_max:.1f}nm) 达 {gap_right:.0f}nm！"
|
||
f"系统将自动向右横推补齐，这可能会导致近红外特征丢失，影响预测精度。\033[0m")
|
||
|
||
def _preprocess_dual_stream(self, spectra: pd.DataFrame,
|
||
metadata: dict) -> np.ndarray:
|
||
"""纯光谱重采样到训练波长网格，返回后由 pipeline.predict() 全自动处理。
|
||
|
||
Pipeline 内置 FeatureUnion[PhysicalExtractor + MNFTransformer]，
|
||
会自动完成物理指数计算和 MNF 降维，无需外部补 WQI。
|
||
"""
|
||
train_wl = metadata.get('train_wavelengths', None)
|
||
if train_wl is None or len(train_wl) == 0:
|
||
raise ValueError(
|
||
"推理失败：metadata 中缺失 train_wavelengths。"
|
||
"DualStream_MNF 必须对齐训练波段以完成光谱重采样。"
|
||
"请使用包含 train_wavelengths 元数据的模型文件。"
|
||
)
|
||
|
||
# 提取纯光谱列
|
||
spec_cols = []
|
||
for c in spectra.columns:
|
||
try:
|
||
float(str(c)); spec_cols.append(c)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# np.interp 重采样：308/113/任意波段 → 模型训练波长
|
||
# ★ 边缘填充：left/right 使用当前行首尾有效值，杜绝 NaN→0.0 断崖
|
||
spec_data = spectra[spec_cols].values.astype(np.float64)
|
||
src_wl = np.array([float(c) for c in spec_cols], dtype=np.float64)
|
||
dst_wl = np.array(train_wl, dtype=np.float64)
|
||
|
||
# ★ 光谱覆盖率预警
|
||
self._check_spectral_coverage(train_wl, src_wl)
|
||
resampled = np.zeros((spec_data.shape[0], len(dst_wl)), dtype=np.float64)
|
||
for i in range(spec_data.shape[0]):
|
||
y_vals = spec_data[i]
|
||
resampled[i] = np.interp(dst_wl, src_wl, y_vals,
|
||
left=y_vals[0], right=y_vals[-1])
|
||
|
||
result = pd.DataFrame(resampled,
|
||
columns=[f'{wl:.6f}' for wl in train_wl])
|
||
print(f"[DualStream_MNF] 纯光谱重采样: {len(spec_cols)} → {len(train_wl)} 列")
|
||
print(f"[DualStream_MNF] pipeline.predict() 将自动完成 Physical+MNF 变换")
|
||
print(f"[特征对齐] 最终输入维度: {result.shape}")
|
||
return result.values
|
||
|
||
def preprocess_spectra(self, spectra: pd.DataFrame) -> np.ndarray:
|
||
"""
|
||
对光谱数据进行预处理 + 跨传感器光谱重采样。
|
||
|
||
改造要点（v2 — 光谱重采样）：
|
||
- 废除原有的「列位置截断」和「零值填充」hack
|
||
- 从 model metadata 中读取训练时的波长列表
|
||
- 如果输入传感器与训练传感器波长网格不同，使用 np.interp()
|
||
将输入光谱重采样到训练波长网格
|
||
- 非光谱列（WQI 指数等）保持不变
|
||
- 兼容旧模型（无 train_wavelengths 时回退到旧逻辑）
|
||
|
||
Args:
|
||
spectra: 原始光谱数据（DataFrame，列名为波长字符串）
|
||
|
||
Returns:
|
||
预处理后的光谱数据 (numpy ndarray)
|
||
"""
|
||
if self.loaded_model_data is None:
|
||
raise ValueError("请先加载模型")
|
||
|
||
preprocess_method = self.loaded_model_data['preprocess_method']
|
||
|
||
# 处理 nan/NaN/None 值
|
||
if pd.isna(preprocess_method) or str(preprocess_method).lower() in ['nan', 'none', '']:
|
||
preprocess_method = "None"
|
||
|
||
# 解析预处理方法（可能包含划分方法前缀）
|
||
if '_' in str(preprocess_method):
|
||
parts = str(preprocess_method).split('_')
|
||
actual_preprocess_method = '_'.join(parts[1:]) if len(parts) > 1 else parts[-1]
|
||
else:
|
||
actual_preprocess_method = str(preprocess_method)
|
||
|
||
if actual_preprocess_method.lower() in ['nan', 'none', '']:
|
||
actual_preprocess_method = "None"
|
||
|
||
model = self.loaded_model_data['model']
|
||
metadata = self.loaded_model_data.get('metadata', {})
|
||
|
||
# ★ 根据模型类型确定预处理归属
|
||
if isinstance(model, Pipeline):
|
||
print(f"[模型信息] 训练预处理方法: {actual_preprocess_method}"
|
||
f" — 由 Pipeline 内部自动执行，推理端仅做光谱重采样+列对齐")
|
||
else:
|
||
print(f"[模型信息] 训练预处理方法: {actual_preprocess_method}"
|
||
f" — 旧裸模型，推理端手动应用外部 Preprocessing")
|
||
print(f"原始光谱数据形状: {spectra.shape}")
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# ★ DualStream_MNF：纯光谱重采样 → pipeline.predict() 全自动
|
||
# ═══════════════════════════════════════════════════════════
|
||
if actual_preprocess_method == "DualStream_MNF" and isinstance(model, Pipeline):
|
||
return self._preprocess_dual_stream(spectra, metadata)
|
||
|
||
train_wavelengths = metadata.get('train_wavelengths', None)
|
||
# 旧模型无 train_wavelengths → 不做波长匹配，走下方分支 B 的 linspace 路径
|
||
train_columns = metadata.get('train_columns', None)
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# ★ 核心：光谱重采样（跨传感器适配）
|
||
# ═══════════════════════════════════════════════════════════
|
||
if train_wavelengths is not None and len(train_wavelengths) > 0:
|
||
print(f"[光谱重采样] 模型训练波长: {len(train_wavelengths)} 个, "
|
||
f"范围 {train_wavelengths[0]:.2f} ~ {train_wavelengths[-1]:.2f} nm")
|
||
|
||
# 1) 从当前 spectra 中分离光谱列和 WQI 列
|
||
target_wavelengths = []
|
||
spectral_col_indices = []
|
||
wqi_col_indices = []
|
||
wqi_col_names = []
|
||
|
||
for i, col in enumerate(spectra.columns):
|
||
col_str = str(col)
|
||
try:
|
||
wl = float(col_str)
|
||
target_wavelengths.append(wl)
|
||
spectral_col_indices.append(i)
|
||
except (ValueError, TypeError):
|
||
if col_str.startswith('WQI_'):
|
||
wqi_col_indices.append(i)
|
||
wqi_col_names.append(col_str)
|
||
|
||
print(f"[光谱重采样] 当前传感器: {len(target_wavelengths)} 个波长列, "
|
||
f"{len(wqi_col_indices)} 个 WQI 列")
|
||
|
||
if len(target_wavelengths) == 0:
|
||
print("[光谱重采样] ⚠ 未检测到波长列，跳过重采样")
|
||
else:
|
||
# 2) 提取光谱数据矩阵
|
||
spectral_data = spectra.iloc[:, spectral_col_indices].values.astype(np.float64)
|
||
|
||
# 3) 逐行 np.interp 重采样
|
||
train_wl_arr = np.array(train_wavelengths, dtype=np.float64)
|
||
target_wl_arr = np.array(target_wavelengths, dtype=np.float64)
|
||
|
||
# ★ 光谱覆盖率预警
|
||
self._check_spectral_coverage(train_wavelengths, target_wavelengths)
|
||
|
||
print(f"[光谱重采样] 执行重采样: {len(target_wavelengths)} → "
|
||
f"{len(train_wavelengths)} 个波长点 ...")
|
||
resampled = np.zeros((spectral_data.shape[0], len(train_wavelengths)),
|
||
dtype=np.float64)
|
||
for i in range(spectral_data.shape[0]):
|
||
y_vals = spectral_data[i]
|
||
resampled[i] = np.interp(
|
||
train_wl_arr, target_wl_arr, y_vals,
|
||
left=y_vals[0], right=y_vals[-1]
|
||
)
|
||
|
||
# 4) 重组为 DataFrame（列名 = 训练波长字符串）
|
||
wl_col_names = [f"{wl:.6f}" for wl in train_wavelengths]
|
||
resampled_df = pd.DataFrame(resampled, columns=wl_col_names,
|
||
index=spectra.index)
|
||
|
||
# 5) 拼接 WQI 列（如有）
|
||
if wqi_col_indices:
|
||
wqi_df = spectra.iloc[:, wqi_col_indices].copy()
|
||
wqi_df.columns = wqi_col_names
|
||
spectra = pd.concat([resampled_df, wqi_df], axis=1)
|
||
else:
|
||
spectra = resampled_df
|
||
|
||
print(f"[光谱重采样] 完成: {spectra.shape[1]} 列 "
|
||
f"(光谱 {len(train_wavelengths)} + WQI {len(wqi_col_indices)})")
|
||
|
||
else:
|
||
# ── 兼容旧模型（无 train_wavelengths）──
|
||
print("[光谱重采样] 模型无 train_wavelengths 元数据（旧模型），"
|
||
"跳过光谱重采样，进入兼容路径...")
|
||
# 旧兼容逻辑：WQI 自动补全（仅对旧裸模型生效；Pipeline 走 train_columns 对齐）
|
||
if isinstance(model, Pipeline):
|
||
train_cols = metadata.get('train_columns', None)
|
||
if train_cols is not None and len(train_cols) > 0:
|
||
print(f"[兼容+Pipeline 列对齐] 训练列数: {len(train_cols)}, 推理当前列数: {spectra.shape[1]}")
|
||
spectra_cols = [str(c) for c in spectra.columns]
|
||
mapping = {}
|
||
unmatched = []
|
||
for tc in train_cols:
|
||
tc_str = str(tc)
|
||
if tc_str in spectra_cols:
|
||
mapping[tc_str] = tc_str
|
||
else:
|
||
unmatched.append(tc_str)
|
||
still_unmatched = []
|
||
for tc_str in unmatched:
|
||
tc_float = None
|
||
try: tc_float = float(tc_str)
|
||
except (ValueError, TypeError): pass
|
||
if tc_float is not None:
|
||
found = None
|
||
for sc in spectra_cols:
|
||
if sc in mapping: continue
|
||
try:
|
||
if math.isclose(tc_float, float(sc), rel_tol=1e-4):
|
||
found = sc; break
|
||
except (ValueError, TypeError): pass
|
||
if found is not None:
|
||
mapping[found] = tc_str; continue
|
||
still_unmatched.append(tc_str)
|
||
aligned_parts = []
|
||
for sc in spectra_cols:
|
||
if sc in mapping:
|
||
col_data = spectra[sc].copy()
|
||
col_data.name = mapping[sc]
|
||
aligned_parts.append(col_data)
|
||
spectra_aligned = pd.concat(aligned_parts, axis=1) if aligned_parts else pd.DataFrame(index=spectra.index)
|
||
for tc_str in still_unmatched:
|
||
spectra_aligned[tc_str] = 0.0
|
||
spectra = spectra_aligned[[str(c) for c in train_cols]]
|
||
print(f"[兼容+Pipeline 列对齐] 完成 → {spectra.shape[1]} 列")
|
||
else:
|
||
print("[兼容+Pipeline] 无 train_columns，跳过 WQI 补全")
|
||
else:
|
||
expected_features = getattr(model, 'n_features_in_', None)
|
||
if expected_features is not None and spectra.shape[1] < expected_features:
|
||
print(f"[特征补全] 检测到特征缺口：当前 {spectra.shape[1]} 列 "
|
||
f"< 模型期望 {expected_features} 列，正在计算 WQI 指数...")
|
||
try:
|
||
from src.utils.water_index import WaterQualityIndexCalculator
|
||
calc = WaterQualityIndexCalculator()
|
||
formulas = calc.list_available()
|
||
if formulas:
|
||
results_df = calc.calculate_many(formulas, spectra, fast=True)
|
||
if isinstance(results_df, pd.DataFrame) and not results_df.empty:
|
||
original_col_count = spectra.shape[1]
|
||
spectra = pd.concat([spectra, results_df], axis=1)
|
||
print(f"[特征补全] 完成！扩充至 {spectra.shape[1]} 列 "
|
||
f"(+{spectra.shape[1] - original_col_count} WQI)")
|
||
except Exception as e:
|
||
print(f"[特征补全] 失败: {e}")
|
||
|
||
# ★ v4: 无 train_wavelengths 时，基于物理波长的经验重采样
|
||
# 从 spectra 列名中解析真实波长 → np.interp 到 400-800nm 标准网格
|
||
# 取代原基于列索引的硬截断（不同传感器列索引对应的波长完全不同）
|
||
if expected_features is not None:
|
||
n_current = spectra.shape[1]
|
||
|
||
# ── 第1步：从列名中提取波长列和非波长列 ──
|
||
_target_wl: List[float] = []
|
||
_wqi_names: List[str] = []
|
||
_other_names: List[str] = []
|
||
for _col in spectra.columns:
|
||
try:
|
||
_target_wl.append(float(str(_col)))
|
||
except (ValueError, TypeError):
|
||
if str(_col).startswith('WQI_'):
|
||
_wqi_names.append(str(_col))
|
||
else:
|
||
_other_names.append(str(_col))
|
||
|
||
# 分离出 WQI 列（保存以便后续拼接回去）
|
||
_wqi_df = spectra[_wqi_names].copy() if _wqi_names else pd.DataFrame()
|
||
|
||
if len(_target_wl) >= 2 and expected_features > 0:
|
||
# ── 第2步：有真实波长 → np.interp 重采样 ──
|
||
fallback_wl = np.linspace(400.0, 800.0, expected_features,
|
||
dtype=np.float64)
|
||
_target_arr = np.array(_target_wl, dtype=np.float64)
|
||
_spec_data = spectra.iloc[:, :len(_target_wl)].values.astype(np.float64)
|
||
# 按波长排序（以防列顺序不是严格递增）
|
||
_sort_idx = np.argsort(_target_arr)
|
||
_target_arr = _target_arr[_sort_idx]
|
||
_spec_data = _spec_data[:, _sort_idx]
|
||
|
||
_resampled = np.zeros((_spec_data.shape[0], expected_features),
|
||
dtype=np.float64)
|
||
for i in range(_spec_data.shape[0]):
|
||
_y_vals = _spec_data[i]
|
||
_resampled[i] = np.interp(
|
||
fallback_wl, _target_arr, _y_vals,
|
||
left=_y_vals[0], right=_y_vals[-1]
|
||
)
|
||
|
||
# 重组 DataFrame
|
||
_wl_cols = [f"{w:.1f}" for w in fallback_wl]
|
||
spectra = pd.DataFrame(_resampled, columns=_wl_cols,
|
||
index=spectra.index)
|
||
if not _wqi_df.empty:
|
||
spectra = pd.concat([spectra, _wqi_df.reset_index(drop=True)],
|
||
axis=1)
|
||
print(f"[兼容重采样] 列名解析到 {len(_target_wl)} 个波长 "
|
||
f"({_target_arr[0]:.1f}~{_target_arr[-1]:.1f}nm) → "
|
||
f"np.interp → {expected_features} 列 (400~800nm 标准网格)"
|
||
f"{' + ' + str(len(_wqi_names)) + ' WQI' if _wqi_names else ''}")
|
||
|
||
elif n_current > expected_features:
|
||
# ── 第3步：无法解析波长（纯字符串列名）→ 硬截断兜底 ──
|
||
spectra = spectra.iloc[:, :expected_features]
|
||
print(f"[兼容截断] 无法解析波长列名，硬截断 "
|
||
f"{n_current} → {expected_features} 列")
|
||
|
||
elif n_current < expected_features:
|
||
# ── 第4步：特征不够 → 补零 ──
|
||
for i in range(expected_features - n_current):
|
||
spectra[f'_padding_{i}'] = 0.0
|
||
print(f"[兼容填充] 特征不足，补零 "
|
||
f"{n_current} → {expected_features} 列")
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# ★ 特征列精确对齐：训练端可能包含 WQI 指数等衍生特征，
|
||
# 推理端必须精确匹配训练时的列集合，不能盲目补全。
|
||
# ═══════════════════════════════════════════════════════════
|
||
if isinstance(model, Pipeline):
|
||
# ── Pipeline 模型：用 train_columns 做精确列对齐 ──
|
||
train_cols = metadata.get('train_columns', None)
|
||
if train_cols is not None and len(train_cols) > 0:
|
||
print(f"[Pipeline 列对齐] 训练列数: {len(train_cols)}, 推理当前列数: {spectra.shape[1]}")
|
||
# 构建 推理列名 → 训练列名 的映射
|
||
spectra_cols = [str(c) for c in spectra.columns]
|
||
mapping: Dict[str, str] = {} # spectra_col_name → train_col_name
|
||
train_set = set(str(tc) for tc in train_cols)
|
||
|
||
# 第一遍：精确字符串匹配
|
||
unmatched_train = []
|
||
for tc in train_cols:
|
||
tc_str = str(tc)
|
||
if tc_str in spectra_cols:
|
||
mapping[tc_str] = tc_str
|
||
else:
|
||
unmatched_train.append(tc_str)
|
||
|
||
# 第二遍：对未匹配的训练列，尝试浮点数近似匹配
|
||
still_unmatched = []
|
||
for tc_str in unmatched_train:
|
||
tc_float = None
|
||
try:
|
||
tc_float = float(tc_str)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
if tc_float is not None:
|
||
found = None
|
||
for sc in spectra_cols:
|
||
if sc in mapping: # 已经被匹配过了
|
||
continue
|
||
try:
|
||
sc_float = float(sc)
|
||
if math.isclose(tc_float, sc_float, rel_tol=1e-4):
|
||
found = sc
|
||
break
|
||
except (ValueError, TypeError):
|
||
pass
|
||
if found is not None:
|
||
mapping[found] = tc_str
|
||
continue
|
||
still_unmatched.append(tc_str)
|
||
|
||
# 从 spectra 中挑选已匹配的列，并重命名为训练列名
|
||
aligned_parts = []
|
||
matched_train = set(mapping.values())
|
||
for sc in spectra_cols:
|
||
if sc in mapping:
|
||
col_data = spectra[sc].copy()
|
||
col_data.name = mapping[sc] # 重命名为训练列名
|
||
aligned_parts.append(col_data)
|
||
|
||
spectra_aligned = pd.concat(aligned_parts, axis=1) if aligned_parts else pd.DataFrame(index=spectra.index)
|
||
|
||
# 对仍未匹配的训练列：WQI_ 补零，其他警告
|
||
for tc_str in still_unmatched:
|
||
spectra_aligned[tc_str] = 0.0
|
||
|
||
# 最终按 train_cols 顺序排列输出
|
||
spectra = spectra_aligned[[str(c) for c in train_cols]]
|
||
matched_count = len(matched_train)
|
||
missing_count = len(still_unmatched)
|
||
print(f"[Pipeline 列对齐] 精确匹配 {matched_count - missing_count} 列, "
|
||
f"浮点近似匹配 {len(mapping) - (matched_count - missing_count)} 列, "
|
||
f"补零填充 {missing_count} 列 → 最终 {spectra.shape[1]} 列")
|
||
else:
|
||
print("[Pipeline 列对齐] 无 train_columns 元数据，跳过列对齐")
|
||
else:
|
||
# ── 旧裸模型：保留原有 WQI 自动补全逻辑 ──
|
||
expected_features = getattr(model, 'n_features_in_', None)
|
||
if expected_features is not None and spectra.shape[1] < expected_features:
|
||
print(f"[特征补全] 检测到特征缺口：当前 {spectra.shape[1]} 列 "
|
||
f"< 模型期望 {expected_features} 列，正在计算 WQI 指数...")
|
||
try:
|
||
from src.utils.water_index import WaterQualityIndexCalculator
|
||
calc = WaterQualityIndexCalculator()
|
||
formulas = calc.list_available()
|
||
if formulas:
|
||
results_df = calc.calculate_many(formulas, spectra, fast=True)
|
||
if isinstance(results_df, pd.DataFrame) and not results_df.empty:
|
||
original_col_count = spectra.shape[1]
|
||
spectra = pd.concat([spectra, results_df], axis=1)
|
||
print(f"[特征补全] 完成！扩充至 {spectra.shape[1]} 列 "
|
||
f"(+{spectra.shape[1] - original_col_count} WQI)")
|
||
except Exception as e:
|
||
print(f"[特征补全] 失败: {e}")
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# 通用清洗
|
||
# ═══════════════════════════════════════════════════════════
|
||
spectra = spectra.replace([np.inf, -np.inf], np.nan)
|
||
spectra = spectra.fillna(0)
|
||
print(f"[特征对齐] 最终输入维度: {spectra.shape}")
|
||
|
||
# ── Pipeline 化分支：模型内置 scaler → 跳过手动 Preprocessing ──
|
||
if isinstance(model, Pipeline):
|
||
print(f"[Pipeline] 模型是 sklearn Pipeline，内置预处理步骤，"
|
||
f"无需外部 Preprocessing")
|
||
return spectra.values
|
||
|
||
# ── 兼容路径：旧裸模型 + 手动 Preprocessing ──
|
||
try:
|
||
spectra_processed = Preprocessing(actual_preprocess_method, spectra)
|
||
if isinstance(spectra_processed, pd.DataFrame):
|
||
spectra_processed = spectra_processed.values
|
||
print(f" [Legacy] 旧裸模型 + 手动 Preprocessing({actual_preprocess_method}) 完成，"
|
||
f"数据形状: {spectra_processed.shape}")
|
||
return spectra_processed
|
||
except Exception as e:
|
||
print(f"预处理失败: {e}")
|
||
print("使用原始数据")
|
||
return spectra.values
|
||
|
||
def predict(self, spectra_processed: np.ndarray) -> np.ndarray:
|
||
"""
|
||
使用加载的模型进行预测
|
||
|
||
Args:
|
||
spectra_processed: 预处理后的光谱数据
|
||
|
||
Returns:
|
||
预测结果
|
||
"""
|
||
if self.loaded_model_data is None:
|
||
raise ValueError("请先加载模型")
|
||
|
||
model = self.loaded_model_data['model']
|
||
|
||
print(f"正在进行预测...")
|
||
print(f"输入数据形状: {spectra_processed.shape}")
|
||
|
||
try:
|
||
# 清洗 NaN / Inf，防止 SVR 等模型报错
|
||
spectra_clean = np.nan_to_num(spectra_processed, nan=0.0, posinf=0.0, neginf=0.0)
|
||
if np.any(np.isnan(spectra_clean)) or np.any(np.isinf(spectra_clean)):
|
||
print("警告: 清洗后数据中仍存在 NaN/Inf，已重置为 0")
|
||
spectra_clean = np.nan_to_num(spectra_clean, nan=0.0, posinf=0.0, neginf=0.0)
|
||
|
||
predictions = model.predict(spectra_clean)
|
||
print(f"预测完成，结果形状: {predictions.shape}")
|
||
print(f"预测值范围: [{np.min(predictions):.4f}, {np.max(predictions):.4f}]")
|
||
print(f"预测值统计: 均值={np.mean(predictions):.4f}, 标准差={np.std(predictions):.4f}")
|
||
|
||
# ★ 百分位裁剪：去除极端异常值，避免专题图色阶被拉爆
|
||
predictions = self._clip_outliers(predictions)
|
||
|
||
return predictions
|
||
|
||
except Exception as e:
|
||
print(f"预测失败: {e}")
|
||
raise
|
||
|
||
@staticmethod
|
||
def _clip_outliers(predictions: np.ndarray, lower_pct: float = 2.0,
|
||
upper_pct: float = 98.0) -> np.ndarray:
|
||
"""百分位裁剪：将极端异常值裁剪到合理范围。
|
||
|
||
水体边界/零值区域的光谱异常会导致模型外推到极端值
|
||
（如 BGA 预测 -86 ~ 7026），若不处理，专题图的克里金
|
||
插值色阶会被拉爆，正常空间变化完全不可见。
|
||
|
||
Parameters
|
||
----------
|
||
predictions : np.ndarray
|
||
原始预测值
|
||
lower_pct : float
|
||
下百分位（默认 2%，低于此分位数的值被裁剪）
|
||
upper_pct : float
|
||
上百分位（默认 98%，高于此分位数的值被裁剪）
|
||
|
||
Returns
|
||
-------
|
||
np.ndarray
|
||
裁剪后的预测值（副本）
|
||
"""
|
||
lo = np.percentile(predictions, lower_pct)
|
||
hi = np.percentile(predictions, upper_pct)
|
||
# 只在实际有异常值时才裁剪
|
||
if lo >= hi:
|
||
return predictions
|
||
n_clipped_lo = int(np.sum(predictions < lo))
|
||
n_clipped_hi = int(np.sum(predictions > hi))
|
||
if n_clipped_lo == 0 and n_clipped_hi == 0:
|
||
return predictions
|
||
print(f"[异常值裁剪] P{lower_pct:.0f}={lo:.4f}, P{upper_pct:.0f}={hi:.4f}, "
|
||
f"裁剪低端 {n_clipped_lo} 个, 高端 {n_clipped_hi} 个 "
|
||
f"({(n_clipped_lo + n_clipped_hi) / len(predictions) * 100:.1f}%)")
|
||
return np.clip(predictions, lo, hi)
|
||
|
||
def _print_mnf_info_if_available(self):
|
||
"""从已加载的 Pipeline 模型中提取 MNFTransformer 并打印波段选择信息。"""
|
||
if self.loaded_model_data is None:
|
||
return
|
||
model = self.loaded_model_data.get('model')
|
||
if model is None:
|
||
return
|
||
|
||
from sklearn.pipeline import Pipeline, FeatureUnion
|
||
if not isinstance(model, Pipeline):
|
||
return
|
||
|
||
# 递归查找 MNFTransformer
|
||
def _find_mnf(step):
|
||
if step.__class__.__name__ == 'MNFTransformer':
|
||
return step
|
||
if isinstance(step, FeatureUnion):
|
||
for _name, _trans in step.transformer_list:
|
||
found = _find_mnf(_trans)
|
||
if found is not None:
|
||
return found
|
||
if isinstance(step, Pipeline):
|
||
for _sub_name, _sub_step in step.steps:
|
||
found = _find_mnf(_sub_step)
|
||
if found is not None:
|
||
return found
|
||
return None
|
||
|
||
mnf = _find_mnf(model)
|
||
if mnf is not None and hasattr(mnf, 'eigvals_w_'):
|
||
target_name = self.loaded_model_data.get('target_column_name', '')
|
||
if not target_name and 'metadata' in self.loaded_model_data:
|
||
target_name = self.loaded_model_data['metadata'].get('target_column_name', '')
|
||
if target_name:
|
||
print(f"\n [MNF 波段信息] 目标指数: {target_name}")
|
||
mnf._print_band_selection()
|
||
|
||
@staticmethod
|
||
def _mask_zero_spectra_pixels(spectra: pd.DataFrame,
|
||
predictions: np.ndarray) -> np.ndarray:
|
||
"""将全零光谱行（NDWI 掩膜外的陆地像素）的预测值强制设为 NaN。
|
||
|
||
GUI 制图 / Kriging / IDW 插值天然忽略 NaN，这是最安全的掩膜继承方式。
|
||
|
||
Parameters
|
||
----------
|
||
spectra : pd.DataFrame
|
||
原始光谱 DataFrame（shape: n_pixels × n_bands）
|
||
predictions : np.ndarray
|
||
模型预测值数组（shape: n_pixels）
|
||
|
||
Returns
|
||
-------
|
||
np.ndarray
|
||
处理后的预测值数组
|
||
"""
|
||
all_zero = (spectra == 0).all(axis=1)
|
||
n_zero = int(all_zero.sum())
|
||
if n_zero == 0:
|
||
return predictions
|
||
predictions = predictions.astype(np.float64, copy=True)
|
||
predictions[all_zero.values] = np.nan
|
||
print(f"[全零拦截] 检测到 {n_zero} 个全零光谱像素 "
|
||
f"({n_zero / len(predictions) * 100:.1f}%)，预测值已设为 NaN")
|
||
return predictions
|
||
|
||
def save_predictions(self, coords: pd.DataFrame, predictions: np.ndarray,
|
||
output_path: str, prediction_column: str = 'prediction',
|
||
wqi_columns: Optional[pd.DataFrame] = None):
|
||
"""
|
||
保存预测结果
|
||
|
||
Args:
|
||
coords: 经纬度数据
|
||
predictions: 预测结果
|
||
output_path: 输出文件路径
|
||
prediction_column: 预测列名称
|
||
wqi_columns: Optional[pd.DataFrame] = None
|
||
"""
|
||
print(f"正在保存预测结果到: {output_path}")
|
||
|
||
# 创建结果DataFrame
|
||
result_df = coords.copy()
|
||
# 追加 WQI 水质指数列（如 sampling_spectra.csv 注入了 45 列指数）
|
||
if wqi_columns is not None and not wqi_columns.empty:
|
||
result_df = pd.concat([result_df, wqi_columns.reset_index(drop=True)], axis=1)
|
||
result_df[prediction_column] = predictions
|
||
|
||
# 确保输出目录存在
|
||
output_dir = os.path.dirname(output_path)
|
||
if output_dir:
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
# 根据文件扩展名选择保存格式
|
||
file_ext = Path(output_path).suffix.lower()
|
||
|
||
if file_ext == '.xls':
|
||
# 保存为Excel 97-2003格式
|
||
try:
|
||
result_df.to_excel(output_path, index=False, engine='xlwt')
|
||
print(f" 格式: Excel 97-2003 (.xls)")
|
||
except ImportError:
|
||
print("警告: xlwt库未安装，无法保存为.xls格式，改为保存CSV格式")
|
||
csv_path = output_path.replace('.xls', '.csv')
|
||
result_df.to_csv(csv_path, index=False, encoding='utf-8-sig')
|
||
output_path = csv_path
|
||
elif file_ext == '.xlsx':
|
||
# 保存为Excel 2007+格式
|
||
try:
|
||
result_df.to_excel(output_path, index=False, engine='openpyxl')
|
||
print(f" 格式: Excel 2007+ (.xlsx)")
|
||
except ImportError:
|
||
print("警告: openpyxl库未安装，无法保存为.xlsx格式，改为保存CSV格式")
|
||
csv_path = output_path.replace('.xlsx', '.csv')
|
||
result_df.to_csv(csv_path, index=False, encoding='utf-8-sig')
|
||
output_path = csv_path
|
||
else:
|
||
# 默认保存为CSV格式
|
||
result_df.to_csv(output_path, index=False, encoding='utf-8-sig')
|
||
print(f" 格式: CSV (.csv)")
|
||
|
||
print(f"预测结果保存完成:")
|
||
print(f" 输出文件: {output_path}")
|
||
print(f" 数据形状: {result_df.shape}")
|
||
print(f" 列名: {list(result_df.columns)}")
|
||
|
||
# 显示预测结果统计
|
||
print(f"\n预测结果统计:")
|
||
print(result_df[prediction_column].describe())
|
||
|
||
return result_df
|
||
|
||
def inference_pipeline(self, sampling_csv_path: str, output_csv_path: str,
|
||
metric: str = 'test_r2', prediction_column: str = 'prediction',
|
||
model_file_path: str = None):
|
||
"""
|
||
完整的推理流程
|
||
|
||
Args:
|
||
sampling_csv_path: 采样数据CSV路径
|
||
output_csv_path: 输出预测结果CSV路径
|
||
metric: 选择最佳模型的指标
|
||
prediction_column: 预测列名称
|
||
model_file_path: 指定模型文件路径（可选）
|
||
"""
|
||
print("=" * 80)
|
||
print("开始水质参数反演推理流程")
|
||
print("=" * 80)
|
||
|
||
try:
|
||
# 1. 加载模型
|
||
print("\n步骤1: 加载模型")
|
||
print("-" * 40)
|
||
if self.external_model is not None:
|
||
# 已在 __init__ 中规范化，无需重复赋值
|
||
print(f" 使用外部预训练模型: type={type(self.external_model).__name__}")
|
||
elif model_file_path:
|
||
self.load_specific_model(model_file_path)
|
||
else:
|
||
self.load_best_model(metric=metric)
|
||
|
||
# 2. 加载采样数据（coords=坐标, spectra=纯光谱, wqi_df=45个WQI指数列）
|
||
print("\n步骤2: 加载采样数据")
|
||
print("-" * 40)
|
||
coords, spectra, wqi_df = self.load_sampling_data(sampling_csv_path)
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# ★ 自适应反射率量级缩放 (Scale Alignment)
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# 不同的高光谱传感器 / 处理流程产出的反射率量级可能不同：
|
||
# - float32 0-1 物理反射率（如 result3.bsq 抽样后写入的 CSV）
|
||
# - int16 0-10000 放大反射率（如 ref_mosaic 抽样后写入的 CSV）
|
||
# 若不经缩放直接喂入 SVR，量级差异会导致预测完全失效。
|
||
# 此处在光谱列上自动检测并统一到 0-1 区间。
|
||
spec_cols = []
|
||
for c in spectra.columns:
|
||
try:
|
||
float(str(c))
|
||
spec_cols.append(c)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
if spec_cols:
|
||
max_val = spectra[spec_cols].max().max()
|
||
if max_val > 10:
|
||
print(f"\n[量级检测] 输入反射率疑似放大格式 (max={max_val:.2f})")
|
||
print("[量级检测] 自动除以 10000，缩放至 0-1 标准物理反射率区间...")
|
||
spectra[spec_cols] = spectra[spec_cols].astype(float) / 10000.0
|
||
print(f"[量级检测] 缩放完成！缩放后 max={spectra[spec_cols].max().max():.4f}")
|
||
else:
|
||
print(f"[量级检测] 输入反射率量级正常 (max={max_val:.4f})，无需缩放")
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
# 3. 数据预处理
|
||
print("\n步骤3: 数据预处理")
|
||
print("-" * 40)
|
||
spectra_processed = self.preprocess_spectra(spectra)
|
||
|
||
# 4. 模型预测
|
||
print("\n步骤4: 模型预测")
|
||
print("-" * 40)
|
||
predictions = self.predict(spectra_processed)
|
||
|
||
# ★ 全零光谱拦截：NDWI 掩膜外的陆地像素 → NaN
|
||
predictions = self._mask_zero_spectra_pixels(spectra, predictions)
|
||
|
||
# 5. 保存预测结果（透传 WQI 列至最终输出文件）
|
||
print("\n步骤5: 保存预测结果")
|
||
print("-" * 40)
|
||
result_df = self.save_predictions(coords, predictions, output_csv_path,
|
||
prediction_column, wqi_df)
|
||
|
||
print("\n" + "=" * 80)
|
||
print("推理流程完成！")
|
||
print("=" * 80)
|
||
|
||
return predictions, result_df
|
||
|
||
except Exception as e:
|
||
print(f"\n推理流程失败: {e}")
|
||
raise
|
||
|
||
def get_model_info(self) -> Dict:
|
||
"""
|
||
获取当前加载模型的信息
|
||
|
||
Returns:
|
||
模型信息字典
|
||
"""
|
||
if self.loaded_model_data is None:
|
||
return {"status": "no_model_loaded"}
|
||
|
||
info = {
|
||
"status": "model_loaded",
|
||
"preprocess_method": self.loaded_model_data.get('preprocess_method', 'Unknown'),
|
||
"model_name": self.loaded_model_data.get('model_name', type(self.external_model).__name__ if self.external_model else 'Unknown'),
|
||
"model_type": str(type(self.loaded_model_data['model'])),
|
||
"metadata": self.loaded_model_data.get('metadata', {})
|
||
}
|
||
|
||
if self.best_model_info:
|
||
info.update(self.best_model_info)
|
||
|
||
return info
|
||
|
||
def batch_inference(self, input_dir: str, output_dir: str,
|
||
metric: str = 'test_r2', prediction_column: str = 'prediction'):
|
||
"""
|
||
批量推理多个采样文件
|
||
|
||
Args:
|
||
input_dir: 输入目录，包含多个采样CSV文件
|
||
output_dir: 输出目录
|
||
metric: 选择最佳模型的指标
|
||
prediction_column: 预测列名称
|
||
"""
|
||
input_path = Path(input_dir)
|
||
output_path = Path(output_dir)
|
||
output_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 查找所有CSV文件
|
||
csv_files = list(input_path.glob("*.csv"))
|
||
|
||
if not csv_files:
|
||
print(f"在目录 {input_dir} 中未找到CSV文件")
|
||
return
|
||
|
||
print(f"找到 {len(csv_files)} 个CSV文件进行批量推理")
|
||
|
||
# 加载模型（只需加载一次）
|
||
self.load_best_model(metric=metric)
|
||
|
||
results = {}
|
||
|
||
for csv_file in csv_files:
|
||
try:
|
||
print(f"\n处理文件: {csv_file.name}")
|
||
output_file = output_path / f"prediction_{csv_file.name}"
|
||
|
||
# 执行推理
|
||
coords, spectra, wqi_df = self.load_sampling_data(str(csv_file))
|
||
# 自适应反射率量级缩放
|
||
_s_cols = []
|
||
for _c in spectra.columns:
|
||
try: float(str(_c)); _s_cols.append(_c)
|
||
except (ValueError, TypeError): pass
|
||
if _s_cols:
|
||
_mv = spectra[_s_cols].max().max()
|
||
if _mv > 10:
|
||
spectra[_s_cols] = spectra[_s_cols].astype(float) / 10000.0
|
||
spectra_processed = self.preprocess_spectra(spectra)
|
||
predictions = self.predict(spectra_processed)
|
||
predictions = self._mask_zero_spectra_pixels(spectra, predictions)
|
||
result_df = self.save_predictions(coords, predictions, str(output_file),
|
||
prediction_column, wqi_df)
|
||
|
||
results[csv_file.name] = {
|
||
'output_file': str(output_file),
|
||
'sample_count': len(predictions),
|
||
'prediction_stats': {
|
||
'mean': np.mean(predictions),
|
||
'std': np.std(predictions),
|
||
'min': np.min(predictions),
|
||
'max': np.max(predictions)
|
||
}
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"处理文件 {csv_file.name} 失败: {e}")
|
||
results[csv_file.name] = {'error': str(e)}
|
||
|
||
print(f"\n批量推理完成，共处理 {len(csv_files)} 个文件")
|
||
return results
|
||
|
||
def batch_inference_multi_models(self, models_root_dir: str, sampling_csv_path: str,
|
||
output_dir: str, metric: str = 'test_r2',
|
||
prediction_column: str = 'prediction',
|
||
output_format: str = 'csv',
|
||
external_model=None,
|
||
external_model_path=None,
|
||
external_models_dict=None):
|
||
"""
|
||
使用多个子文件夹中的模型进行批量推理
|
||
|
||
Args:
|
||
models_root_dir: 包含多个子文件夹的根目录，每个子文件夹作为artifacts_dir
|
||
sampling_csv_path: 采样数据CSV路径
|
||
output_dir: 输出目录
|
||
metric: 选择最佳模型的指标
|
||
prediction_column: 预测列名称
|
||
output_format: 输出文件格式 ('csv', 'xls', 'xlsx')
|
||
"""
|
||
models_root = Path(models_root_dir)
|
||
output_path = Path(output_dir)
|
||
output_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
all_results = {}
|
||
|
||
# 优先级 1：_external_models_dict 非空 → 直接用字典的 keys 作为 targets，不扫描磁盘
|
||
print(f"[BatchInference] 终于收到字典啦！包含模型: {list(external_models_dict.keys()) if external_models_dict else 'None'}")
|
||
if external_models_dict is not None and len(external_models_dict) > 0:
|
||
targets = list(external_models_dict.keys())
|
||
print(f"\n使用外部导入模型字典（{len(targets)} 个模型）")
|
||
print(f"检测到外部导入模型，将预测以下参数: {targets}")
|
||
elif external_model is not None:
|
||
print(f"\n使用外部预训练模型: {external_model_path or 'unknown'}")
|
||
subdirs = [d for d in models_root.iterdir() if d.is_dir()]
|
||
if not subdirs:
|
||
print(f"在目录 {models_root_dir} 中未找到子文件夹")
|
||
return {}
|
||
print(f"找到 {len(subdirs)} 个模型子文件夹进行批量推理")
|
||
targets = [d.name for d in subdirs]
|
||
else:
|
||
subdirs = [d for d in models_root.iterdir() if d.is_dir()]
|
||
if not subdirs:
|
||
print(f"在目录 {models_root_dir} 中未找到子文件夹")
|
||
return {}
|
||
print(f"找到 {len(subdirs)} 个模型子文件夹进行批量推理")
|
||
targets = [d.name for d in subdirs]
|
||
|
||
print(f"输出格式: {output_format.upper()}")
|
||
|
||
for subdir_name in targets:
|
||
try:
|
||
print(f"\n{'='*60}")
|
||
print(f"处理模型: {subdir_name}")
|
||
print(f"{'='*60}")
|
||
|
||
# 优先级：字典中该 target 的模型 > 共享单模型 > 磁盘加载
|
||
effective_model = None
|
||
if external_models_dict and subdir_name in external_models_dict:
|
||
effective_model = external_models_dict[subdir_name]
|
||
print(f" → 使用字典中模型: {type(effective_model).__name__}")
|
||
elif external_model is not None:
|
||
effective_model = external_model
|
||
print(f" → 使用共享外部模型: {type(effective_model).__name__}")
|
||
|
||
# artifacts_dir：字典模式优先用 placeholder "./"，否则用真实子目录
|
||
artifacts_dir = (
|
||
str(models_root / subdir_name)
|
||
if (models_root / subdir_name).is_dir()
|
||
else str(models_root)
|
||
)
|
||
if effective_model is not None:
|
||
model_inferencer = WaterQualityInference(
|
||
artifacts_dir,
|
||
external_model=effective_model,
|
||
external_model_path=external_model_path or "",
|
||
)
|
||
else:
|
||
model_inferencer = WaterQualityInference(artifacts_dir)
|
||
|
||
# 根据输出格式设置文件扩展名
|
||
file_ext = f".{output_format}"
|
||
output_file = output_path / f"{subdir_name}{file_ext}"
|
||
|
||
# 执行推理流程
|
||
predictions, result_df = model_inferencer.inference_pipeline(
|
||
sampling_csv_path=sampling_csv_path,
|
||
output_csv_path=str(output_file),
|
||
metric=metric,
|
||
prediction_column=prediction_column
|
||
)
|
||
|
||
# 收集结果信息
|
||
model_info = model_inferencer.get_model_info()
|
||
all_results[subdir_name] = {
|
||
'status': 'success',
|
||
'output_file': str(output_file),
|
||
'sample_count': len(predictions),
|
||
'model_info': model_info,
|
||
'prediction_stats': {
|
||
'mean': np.mean(predictions),
|
||
'std': np.std(predictions),
|
||
'min': np.min(predictions),
|
||
'max': np.max(predictions)
|
||
}
|
||
}
|
||
|
||
print(f"模型 {subdir_name} 处理完成")
|
||
|
||
except Exception as e:
|
||
print(f"处理模型 {subdir_name} 失败: {e}")
|
||
all_results[subdir_name] = {
|
||
'status': 'error',
|
||
'error': str(e)
|
||
}
|
||
|
||
print(f"\n{'='*80}")
|
||
print(f"批量推理完成，共处理 {len(subdirs)} 个模型文件夹")
|
||
print(f"{'='*80}")
|
||
|
||
# 打印汇总信息
|
||
print("\n汇总结果:")
|
||
for folder_name, result in all_results.items():
|
||
if result['status'] == 'success':
|
||
print(f" ✓ {folder_name}: {result['sample_count']} 个预测值，"
|
||
f"均值={result['prediction_stats']['mean']:.4f}")
|
||
else:
|
||
print(f" ✗ {folder_name}: 失败 - {result['error']}")
|
||
|
||
return all_results
|
||
|
||
def batch_inference_multi_data(self, artifacts_dir: str, input_dir: str,
|
||
output_dir: str, metric: str = 'test_r2',
|
||
prediction_column: str = 'prediction',
|
||
output_format: str = 'csv'):
|
||
"""
|
||
使用一个模型对多个数据文件进行批量推理，输出文件名为数据文件名（不含扩展名）
|
||
|
||
Args:
|
||
artifacts_dir: 模型目录
|
||
input_dir: 输入目录，包含多个采样CSV文件
|
||
output_dir: 输出目录
|
||
metric: 选择最佳模型的指标
|
||
prediction_column: 预测列名称
|
||
output_format: 输出文件格式 ('csv', 'xls', 'xlsx')
|
||
"""
|
||
input_path = Path(input_dir)
|
||
output_path = Path(output_dir)
|
||
output_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 查找所有CSV文件
|
||
csv_files = list(input_path.glob("*.csv"))
|
||
|
||
if not csv_files:
|
||
print(f"在目录 {input_dir} 中未找到CSV文件")
|
||
return
|
||
|
||
print(f"找到 {len(csv_files)} 个CSV文件进行批量推理")
|
||
print(f"输出格式: {output_format.upper()}")
|
||
|
||
# 初始化推理器并加载模型（只需加载一次）
|
||
self.artifacts_dir = Path(artifacts_dir)
|
||
self.load_best_model(metric=metric)
|
||
|
||
results = {}
|
||
|
||
for csv_file in csv_files:
|
||
try:
|
||
# 获取不含扩展名的文件名
|
||
file_stem = csv_file.stem
|
||
print(f"\n处理文件: {csv_file.name}")
|
||
|
||
# 根据输出格式设置文件扩展名
|
||
file_ext = f".{output_format}"
|
||
output_file = output_path / f"{file_stem}{file_ext}"
|
||
|
||
# 执行推理
|
||
coords, spectra, wqi_df = self.load_sampling_data(str(csv_file))
|
||
# 自适应反射率量级缩放
|
||
_s_cols = []
|
||
for _c in spectra.columns:
|
||
try: float(str(_c)); _s_cols.append(_c)
|
||
except (ValueError, TypeError): pass
|
||
if _s_cols:
|
||
_mv = spectra[_s_cols].max().max()
|
||
if _mv > 10:
|
||
spectra[_s_cols] = spectra[_s_cols].astype(float) / 10000.0
|
||
spectra_processed = self.preprocess_spectra(spectra)
|
||
predictions = self.predict(spectra_processed)
|
||
predictions = self._mask_zero_spectra_pixels(spectra, predictions)
|
||
result_df = self.save_predictions(coords, predictions, str(output_file),
|
||
prediction_column, wqi_df)
|
||
|
||
results[file_stem] = {
|
||
'input_file': str(csv_file),
|
||
'output_file': str(output_file),
|
||
'sample_count': len(predictions),
|
||
'prediction_stats': {
|
||
'mean': np.mean(predictions),
|
||
'std': np.std(predictions),
|
||
'min': np.min(predictions),
|
||
'max': np.max(predictions)
|
||
}
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"处理文件 {csv_file.name} 失败: {e}")
|
||
results[csv_file.stem] = {'error': str(e)}
|
||
|
||
print(f"\n批量推理完成，共处理 {len(csv_files)} 个文件")
|
||
return results
|
||
|
||
def evaluate_with_split(self, data_csv_path: str, split_method: str = "random",
|
||
test_size: float = 0.2, random_state: int = 42,
|
||
target_column: int = 11, feature_start_column: int = 13,
|
||
metric: str = 'test_r2', prediction_column: str = 'prediction'):
|
||
"""
|
||
使用训练时相同的数据分割方法进行模型评估
|
||
|
||
Args:
|
||
data_csv_path: 包含目标值的完整数据集CSV路径
|
||
split_method: 数据分割方法 ("random", "spxy", "ks")
|
||
test_size: 测试集比例
|
||
random_state: 随机种子
|
||
target_column: 目标值列索引
|
||
feature_start_column: 特征开始列索引
|
||
metric: 选择模型的评估指标
|
||
prediction_column: 预测结果列名
|
||
|
||
Returns:
|
||
评估结果字典
|
||
"""
|
||
print("=" * 80)
|
||
print("开始数据分割评估流程")
|
||
print("=" * 80)
|
||
|
||
try:
|
||
# 1. 加载完整数据集
|
||
print("\n步骤1: 加载完整数据集")
|
||
print("-" * 40)
|
||
data = pd.read_csv(data_csv_path)
|
||
|
||
# 提取目标值和特征
|
||
y = data.iloc[:, target_column]
|
||
X = data.iloc[:, feature_start_column:]
|
||
|
||
# 去除目标值为空的行
|
||
mask = ~y.isna()
|
||
data_cleaned = data[mask]
|
||
y_cleaned = data_cleaned.iloc[:, target_column]
|
||
X_cleaned = data_cleaned.iloc[:, feature_start_column:]
|
||
|
||
print(f"数据加载完成:")
|
||
print(f" 原始样本数: {len(data)}")
|
||
print(f" 清理后样本数: {len(X_cleaned)}")
|
||
print(f" 特征数量: {X_cleaned.shape[1]}")
|
||
print(f" 目标值范围: {y_cleaned.min():.4f} ~ {y_cleaned.max():.4f}")
|
||
|
||
# 2. 加载最佳模型
|
||
print("\n步骤2: 加载最佳模型")
|
||
print("-" * 40)
|
||
self.load_best_model(metric=metric)
|
||
|
||
# 3. 数据预处理
|
||
print("\n步骤3: 数据预处理")
|
||
print("-" * 40)
|
||
X_processed = self.preprocess_spectra(X_cleaned)
|
||
|
||
# 4. 数据分割
|
||
print("\n步骤4: 数据分割")
|
||
print("-" * 40)
|
||
X_train, X_test, y_train, y_test = self.split_data(
|
||
X_processed, y_cleaned, method=split_method,
|
||
test_size=test_size, random_state=random_state
|
||
)
|
||
|
||
print(f"数据分割完成:")
|
||
print(f" 训练集样本数: {X_train.shape[0]}")
|
||
print(f" 测试集样本数: {X_test.shape[0]}")
|
||
|
||
# 5. 模型预测
|
||
print("\n步骤5: 模型预测")
|
||
print("-" * 40)
|
||
|
||
# 训练集预测
|
||
y_train_pred = self.loaded_model_data['model'].predict(X_train)
|
||
|
||
# 测试集预测
|
||
y_test_pred = self.loaded_model_data['model'].predict(X_test)
|
||
|
||
# 6. 计算评估指标
|
||
print("\n步骤6: 计算评估指标")
|
||
print("-" * 40)
|
||
|
||
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
|
||
|
||
# 训练集指标
|
||
train_mse = mean_squared_error(y_train, y_train_pred)
|
||
train_mae = mean_absolute_error(y_train, y_train_pred)
|
||
train_r2 = r2_score(y_train, y_train_pred)
|
||
train_rmse = np.sqrt(train_mse)
|
||
|
||
# 测试集指标
|
||
test_mse = mean_squared_error(y_test, y_test_pred)
|
||
test_mae = mean_absolute_error(y_test, y_test_pred)
|
||
test_r2 = r2_score(y_test, y_test_pred)
|
||
test_rmse = np.sqrt(test_mse)
|
||
|
||
results = {
|
||
'split_method': split_method,
|
||
'test_size': test_size,
|
||
'train_size': len(y_train),
|
||
'test_size_actual': len(y_test),
|
||
'train_metrics': {
|
||
'mse': train_mse,
|
||
'mae': train_mae,
|
||
'rmse': train_rmse,
|
||
'r2': train_r2
|
||
},
|
||
'test_metrics': {
|
||
'mse': test_mse,
|
||
'mae': test_mae,
|
||
'rmse': test_rmse,
|
||
'r2': test_r2
|
||
},
|
||
'predictions': {
|
||
'y_train_true': y_train,
|
||
'y_train_pred': y_train_pred,
|
||
'y_test_true': y_test,
|
||
'y_test_pred': y_test_pred
|
||
}
|
||
}
|
||
|
||
print(f"评估完成:")
|
||
print(f" 训练集指标:")
|
||
print(f" R²: {train_r2:.4f}")
|
||
print(f" RMSE: {train_rmse:.4f}")
|
||
print(f" MAE: {train_mae:.4f}")
|
||
print(f" 测试集指标:")
|
||
print(f" R²: {test_r2:.4f}")
|
||
print(f" RMSE: {test_rmse:.4f}")
|
||
print(f" MAE: {test_mae:.4f}")
|
||
|
||
print("\n" + "=" * 80)
|
||
print("数据分割评估流程完成！")
|
||
print("=" * 80)
|
||
|
||
return results
|
||
|
||
except Exception as e:
|
||
print(f"\n数据分割评估失败: {e}")
|
||
raise
|
||
|
||
|
||
def main():
|
||
"""主函数示例"""
|
||
# 创建推理实例
|
||
artifacts_dir = r"E:\code\WQ\yaobao925\qvchuyaoban"
|
||
inferencer = WaterQualityInference(artifacts_dir)
|
||
|
||
# 配置文件路径
|
||
sampling_csv = r"E:\code\WQ\xiaogujia\使用腰堡模型\spectral_sampling_results.csv"
|
||
# output_csv = r"E:\code\WQ\laodao\output"
|
||
|
||
try:
|
||
# # 示例1: 单个模型单个数据文件的推理
|
||
# print("示例1: 单个模型单个数据文件的推理")
|
||
# predictions, result_df = inferencer.inference_pipeline(
|
||
# sampling_csv_path=sampling_csv,
|
||
# output_csv_path=output_csv,
|
||
# metric='test_r2', # 使用测试集R²作为选择最佳模型的指标
|
||
# prediction_column='water_quality_prediction'
|
||
# )
|
||
#
|
||
# print(f"\n推理完成，共生成 {len(predictions)} 个预测值")
|
||
#
|
||
# # 显示模型信息
|
||
# model_info = inferencer.get_model_info()
|
||
# print(f"\n使用的模型信息:")
|
||
# print(f" 组合: {model_info.get('combination', 'Unknown')}")
|
||
# print(f" 预处理: {model_info.get('preprocess_method', 'Unknown')}")
|
||
# print(f" 算法: {model_info.get('model_name', 'Unknown')}")
|
||
|
||
# 示例2: 批量推理多个模型（每个子文件夹作为不同的artifacts_dir）
|
||
print(f"\n{'='*80}")
|
||
print("示例2: 批量推理多个模型")
|
||
models_root_dir = r"E:\code\WQ\yaobao925\qvchuyaoban" # 包含多个子文件夹的根目录
|
||
output_dir = r"E:\code\WQ\xiaogujia\使用腰堡模型\predict"
|
||
|
||
all_results = inferencer.batch_inference_multi_models(
|
||
models_root_dir=models_root_dir,
|
||
sampling_csv_path=sampling_csv,
|
||
output_dir=output_dir,
|
||
metric='test_r2',
|
||
prediction_column='water_quality_prediction'
|
||
)
|
||
|
||
# 示例3: 使用数据分割方法进行模型评估（可选）
|
||
# print(f"\n{'='*80}")
|
||
# print("示例3: 数据分割评估")
|
||
# complete_data_csv = r"E:\code\WQ\laodao\data\捞刀河-浏阳河-圭塘河.csv" # 包含目标值的完整数据集
|
||
#
|
||
# # 使用SPXY方法进行数据分割评估
|
||
# eval_results = inferencer.evaluate_with_split(
|
||
# data_csv_path=complete_data_csv,
|
||
# split_method="spxy", # 可选: "random", "spxy", "ks"
|
||
# test_size=0.2,
|
||
# random_state=42,
|
||
# target_column=11, # 目标值列索引
|
||
# feature_start_column=13, # 特征开始列索引
|
||
# metric='test_r2'
|
||
# )
|
||
#
|
||
# print(f"\n数据分割评估结果:")
|
||
# print(f" 分割方法: {eval_results['split_method']}")
|
||
# print(f" 训练集R²: {eval_results['train_metrics']['r2']:.4f}")
|
||
# print(f" 测试集R²: {eval_results['test_metrics']['r2']:.4f}")
|
||
# print(f" 训练集RMSE: {eval_results['train_metrics']['rmse']:.4f}")
|
||
# print(f" 测试集RMSE: {eval_results['test_metrics']['rmse']:.4f}")
|
||
|
||
except Exception as e:
|
||
print(f"推理失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|