From 04f9a647d896e0a96bbab934fd938ba958dbdd45 Mon Sep 17 00:00:00 2001 From: duxin Date: Tue, 4 Aug 2026 09:10:26 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8E=A8=E7=90=86=E7=AB=AF=E8=87=AA?= =?UTF-8?q?=E9=80=82=E5=BA=94=E5=8F=8D=E5=B0=84=E7=8E=87=E7=BC=A9=E6=94=BE?= =?UTF-8?q?=20+=20Pipeline=20=E5=88=97=E7=B2=BE=E7=A1=AE=E5=AF=B9=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 内部自动执行' --- src/core/prediction/inference_batch.py | 250 ++++++++++++++++++++----- 1 file changed, 208 insertions(+), 42 deletions(-) diff --git a/src/core/prediction/inference_batch.py b/src/core/prediction/inference_batch.py index 2061429..03c4b29 100644 --- a/src/core/prediction/inference_batch.py +++ b/src/core/prediction/inference_batch.py @@ -2,6 +2,7 @@ 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 @@ -546,12 +547,18 @@ class WaterQualityInference: if actual_preprocess_method.lower() in ['nan', 'none', '']: actual_preprocess_method = "None" - print(f"正在应用预处理方法: {actual_preprocess_method}") - print(f"原始光谱数据形状: {spectra.shape}") - 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() 全自动 # ═══════════════════════════════════════════════════════════ @@ -633,24 +640,67 @@ class WaterQualityInference: # ── 兼容旧模型（无 train_wavelengths）── print("[光谱重采样] 模型无 train_wavelengths 元数据（旧模型），" "跳过光谱重采样，进入兼容路径...") - # 旧兼容逻辑：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}") + # 旧兼容逻辑：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 标准网格 @@ -720,26 +770,97 @@ class WaterQualityInference: f"{n_current} → {expected_features} 列") # ═══════════════════════════════════════════════════════════ - # ★ 特征补全：模型训练时可能包含 WQI 指数等衍生特征， - # 推理端需自动计算补齐（适用于新旧模型两条路径） + # ★ 特征列精确对齐：训练端可能包含 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}") + 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}") # ═══════════════════════════════════════════════════════════ # 通用清洗 @@ -1009,6 +1130,33 @@ class WaterQualityInference: 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) @@ -1097,12 +1245,21 @@ class WaterQualityInference: # 执行推理 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), @@ -1297,12 +1454,21 @@ class WaterQualityInference: # 执行推理 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),