feat(step10): 新增 WaterIndexCsvProcessor 散点处理入口

src/core/algorithms/waterindex_inversion/__init__.py

@@ -644,3 +644,190 @@ class WaterIndexProcessor:
 
         notify("水色指数反演完成", 100)
         return results
+
+
+# ------------------------------------------------------------------
+# 散点处理入口（Step 10 重构后使用，与 Step 9 对称）
+# ------------------------------------------------------------------
+
+class WaterIndexCsvProcessor:
+    """
+    散点 CSV 驱动的水色指数反演器。
+
+    设计目的
+    --------
+    与 Step 9 (ML 预测) 完全对称的【散点处理模式】：
+
+    * 输入：Step 4 生成的 ``sampling_spectra.csv``，列结构为
+      ``x_coord, y_coord, pixel_x, pixel_y, 400.000000, 401.000000, ...``
+    * 处理：解析 ``waterindex.csv`` 中的公式，对每行采样点
+            提取对应波段数值、逐行 eval 计算水色指数
+    * 输出：每个公式一个 CSV，列严格为 ``longitude, latitude, <formula_name>``，
+            可直接喂给 Step 11 ContentMapper
+
+    输出目录
+    --------
+    默认 ``{work_dir}/10_WaterIndex_CSV/``；若用户指定 ``output_dir`` 则用其值。
+    """
+
+    COORD_RENAME_MAP = {
+        "x_coord": "longitude",
+        "y_coord": "latitude",
+        "lon": "longitude",
+        "lat": "latitude",
+    }
+
+    def __init__(self, waterindex_csv_path: Optional[str] = None):
+        if waterindex_csv_path is None:
+            candidates = [
+                os.path.join(os.path.dirname(__file__), '..', '..', 'gui', 'model', 'waterindex.csv'),
+                os.path.join(os.path.dirname(__file__), '..', '..', '..', 'gui', 'model', 'waterindex.csv'),
+            ]
+            for p in candidates:
+                if os.path.isfile(p):
+                    waterindex_csv_path = p
+                    break
+        self.waterindex_csv_path = waterindex_csv_path
+        self._index_calc = None
+
+    def _get_index_calc(self):
+        """懒加载 WaterQualityIndexCalculator（首次访问时实例化）"""
+        if self._index_calc is None and self.waterindex_csv_path:
+            from src.utils.water_index import WaterQualityIndexCalculator
+            self._index_calc = WaterQualityIndexCalculator(self.waterindex_csv_path)
+        return self._index_calc
+
+    @staticmethod
+    def _detect_wavelength_columns(df: "pd.DataFrame") -> List[str]:
+        """识别光谱列：列名是浮点数字符串（Step 4 输出的 '400.000000' 形式）"""
+        import re
+        wl_cols = []
+        for col in df.columns:
+            try:
+                float(str(col).strip())
+                wl_cols.append(col)
+            except (ValueError, TypeError):
+                continue
+        return wl_cols
+
+    @staticmethod
+    def _safe_filename(name: str) -> str:
+        """公式名 → 文件名安全字符（与旧 BSQ 输出命名习惯一致）"""
+        return re.sub(r'[^\w\u4e00-\u9fff-]', '_', name).strip('_') or 'index'
+
+    def compute_indices_from_csv(
+        self,
+        sampling_csv_path: str,
+        output_dir: str,
+        selected_formulas: Optional[List[str]] = None,
+        progress_callback: Optional[Callable[[str, float], None]] = None,
+    ) -> Dict[str, str]:
+        """
+        散点 CSV → 按指数拆分的多个 CSV。
+
+        Parameters
+        ----------
+        sampling_csv_path : str
+            Step 4 输出的 ``sampling_spectra.csv`` 路径
+        output_dir : str
+            输出目录；不存在会自动创建
+        selected_formulas : list, optional
+            要计算的公式名列表；None 或空列表 = 全部公式
+        progress_callback : callable, optional
+            进度回调 ``(msg: str, pct: float)``
+
+        Returns
+        -------
+        dict
+            ``{公式名: 输出 CSV 路径}``
+        """
+        def notify(msg: str, pct: float) -> None:
+            if progress_callback:
+                progress_callback(msg, pct)
+
+        if not os.path.isfile(sampling_csv_path):
+            raise FileNotFoundError(f"采样点 CSV 不存在: {sampling_csv_path}")
+
+        if not self.waterindex_csv_path or not os.path.isfile(self.waterindex_csv_path):
+            raise FileNotFoundError(
+                f"waterindex.csv 未配置或不存在: {self.waterindex_csv_path}"
+            )
+
+        os.makedirs(output_dir, exist_ok=True)
+
+        notify("正在读取采样点 CSV…", 5)
+        import pandas as pd
+        df = pd.read_csv(sampling_csv_path, encoding="utf-8-sig")
+        if df.empty:
+            raise ValueError(f"采样点 CSV 为空: {sampling_csv_path}")
+
+        # 坐标列重命名（x_coord → longitude, y_coord → latitude）
+        df = df.rename(columns={k: v for k, v in self.COORD_RENAME_MAP.items()
+                                if k in df.columns})
+        if "longitude" not in df.columns or "latitude" not in df.columns:
+            raise ValueError(
+                f"采样点 CSV 缺少坐标列（期望 x_coord/y_coord 或 longitude/latitude），"
+                f"实际列: {list(df.columns)}"
+            )
+
+        # 识别光谱列
+        wl_cols = self._detect_wavelength_columns(df)
+        if not wl_cols:
+            raise ValueError(
+                f"采样点 CSV 中未识别到任何光谱列（列名为数字）,"
+                f"实际列: {list(df.columns)}"
+            )
+        notify(f"识别到 {len(wl_cols)} 个光谱列, 采样点 {len(df)} 个", 15)
+
+        calc = self._get_index_calc()
+        if calc is None:
+            raise RuntimeError("WaterQualityIndexCalculator 初始化失败")
+
+        all_formula_names = calc.list_available()
+        if selected_formulas:
+            targets = [n for n in selected_formulas if n in all_formula_names]
+            missing = [n for n in selected_formulas if n not in all_formula_names]
+            if missing:
+                print(f"[WaterIndexCsvProcessor] 警告: 以下公式未在 waterindex.csv 中找到,已跳过: {missing}")
+        else:
+            targets = all_formula_names
+
+        if not targets:
+            raise ValueError("没有可计算的公式（selected_formulas 为空且 waterindex.csv 中无公式）")
+
+        # 一次性算出所有目标公式的 Series（避免重复遍历 DataFrame）
+        notify(f"开始逐行计算 {len(targets)} 个公式…", 25)
+        spectra_df = df[wl_cols]
+        try:
+            results_df = calc.calculate_many(targets, spectra_df)
+        except Exception as e:
+            raise RuntimeError(f"公式计算失败: {e}")
+
+        # 每个公式一个 CSV：longitude, latitude, <formula_name>
+        out_files: Dict[str, str] = {}
+        n_total = len(targets)
+        for i, name in enumerate(targets):
+            try:
+                per_idx = results_df[name]
+                out_df = pd.DataFrame({
+                    "longitude": df["longitude"].values,
+                    "latitude": df["latitude"].values,
+                    name: per_idx.values,
+                })
+                out_path = os.path.join(output_dir, f"{self._safe_filename(name)}.csv")
+                out_df.to_csv(out_path, index=False, float_format="%.6f", encoding="utf-8-sig")
+                out_files[name] = out_path
+                notify(
+                    f"[{i + 1}/{n_total}] {name} → {os.path.basename(out_path)}",
+                    25 + 70 * (i + 1) / n_total,
+                )
+            except Exception as e:
+                print(f"[WaterIndexCsvProcessor] 公式 '{name}' 失败: {e}")
+                continue
+
+        notify(f"完成！共输出 {len(out_files)} / {n_total} 个指数 CSV", 100)
+        return out_files
+
+
+# 保留旧 import 路径兼容
+__all__ = ['WaterIndexProcessor', 'WaterIndexCsvProcessor']