From b0ace0bde85c3d131c33c20e38b7f13f32fb4dea Mon Sep 17 00:00:00 2001 From: DXC Date: Wed, 24 Jun 2026 11:39:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(step10):=20=E6=96=B0=E5=A2=9E=20WaterIndex?= =?UTF-8?q?CsvProcessor=20=E6=95=A3=E7=82=B9=E5=A4=84=E7=90=86=E5=85=A5?= =?UTF-8?q?=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../waterindex_inversion/__init__.py | 187 ++++++++++++++++++ 1 file changed, 187 insertions(+) diff --git a/src/core/algorithms/waterindex_inversion/__init__.py b/src/core/algorithms/waterindex_inversion/__init__.py index eb41cb9..0d2723e 100644 --- a/src/core/algorithms/waterindex_inversion/__init__.py +++ b/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, ``， + 可直接喂给 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, + 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']