fix: 推理端自适应反射率缩放 + Pipeline 列精确对齐

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 内部自动执行'
This commit is contained in:
duxin
2026-08-04 09:10:26 +08:00
parent f801b481fe
commit 04f9a647d8

View File

@ -2,6 +2,7 @@ import numpy as np
import pandas as pd import pandas as pd
import joblib import joblib
import os import os
import math
from pathlib import Path from pathlib import Path
from typing import List, Dict, Union, Tuple, Optional from typing import List, Dict, Union, Tuple, Optional
import warnings import warnings
@ -546,12 +547,18 @@ class WaterQualityInference:
if actual_preprocess_method.lower() in ['nan', 'none', '']: if actual_preprocess_method.lower() in ['nan', 'none', '']:
actual_preprocess_method = "None" actual_preprocess_method = "None"
print(f"正在应用预处理方法: {actual_preprocess_method}")
print(f"原始光谱数据形状: {spectra.shape}")
model = self.loaded_model_data['model'] model = self.loaded_model_data['model']
metadata = self.loaded_model_data.get('metadata', {}) 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() 全自动 # ★ DualStream_MNF:纯光谱重采样 → pipeline.predict() 全自动
# ═══════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════
@ -633,7 +640,50 @@ class WaterQualityInference:
# ── 兼容旧模型(无 train_wavelengths)── # ── 兼容旧模型(无 train_wavelengths)──
print("[光谱重采样] 模型无 train_wavelengths 元数据(旧模型)," print("[光谱重采样] 模型无 train_wavelengths 元数据(旧模型),"
"跳过光谱重采样,进入兼容路径...") "跳过光谱重采样,进入兼容路径...")
# 旧兼容逻辑:WQI 自动补全 # 旧兼容逻辑: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) expected_features = getattr(model, 'n_features_in_', None)
if expected_features is not None and spectra.shape[1] < expected_features: if expected_features is not None and spectra.shape[1] < expected_features:
print(f"[特征补全] 检测到特征缺口:当前 {spectra.shape[1]} 列 " print(f"[特征补全] 检测到特征缺口:当前 {spectra.shape[1]} 列 "
@ -720,9 +770,80 @@ class WaterQualityInference:
f"{n_current} → {expected_features} 列") f"{n_current} → {expected_features} 列")
# ═══════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════
# ★ 特征补全:模型训练时可能包含 WQI 指数等衍生特征, # ★ 特征列精确对齐:训练端可能包含 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) expected_features = getattr(model, 'n_features_in_', None)
if expected_features is not None and spectra.shape[1] < expected_features: if expected_features is not None and spectra.shape[1] < expected_features:
print(f"[特征补全] 检测到特征缺口:当前 {spectra.shape[1]} 列 " print(f"[特征补全] 检测到特征缺口:当前 {spectra.shape[1]} 列 "
@ -1009,6 +1130,33 @@ class WaterQualityInference:
print("-" * 40) print("-" * 40)
coords, spectra, wqi_df = self.load_sampling_data(sampling_csv_path) 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. 数据预处理 # 3. 数据预处理
print("\n步骤3: 数据预处理") print("\n步骤3: 数据预处理")
print("-" * 40) print("-" * 40)
@ -1097,6 +1245,15 @@ class WaterQualityInference:
# 执行推理 # 执行推理
coords, spectra, wqi_df = self.load_sampling_data(str(csv_file)) 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) spectra_processed = self.preprocess_spectra(spectra)
predictions = self.predict(spectra_processed) predictions = self.predict(spectra_processed)
predictions = self._mask_zero_spectra_pixels(spectra, predictions) predictions = self._mask_zero_spectra_pixels(spectra, predictions)
@ -1297,6 +1454,15 @@ class WaterQualityInference:
# 执行推理 # 执行推理
coords, spectra, wqi_df = self.load_sampling_data(str(csv_file)) 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) spectra_processed = self.preprocess_spectra(spectra)
predictions = self.predict(spectra_processed) predictions = self.predict(spectra_processed)
predictions = self._mask_zero_spectra_pixels(spectra, predictions) predictions = self._mask_zero_spectra_pixels(spectra, predictions)