refactor(step10): 拆分 WaterIndexCsvProcessor 到独立子模块 + smoke test
This commit is contained in:
@ -649,184 +649,11 @@ class WaterIndexProcessor:
|
||||
# ------------------------------------------------------------------
|
||||
# 散点处理入口（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
|
||||
# WaterIndexCsvProcessor 已拆出到独立子模块 csv_processor.py，
|
||||
# 目的是让纯 CSV 计算链路不再被 __init__.py 顶部 osgeo import 拖垮。
|
||||
# 这里做一次 re-export，保留所有 `from src.core.algorithms.waterindex_inversion import WaterIndexCsvProcessor`
|
||||
# 这类已有 import 路径仍能正常工作（生产环境/打包后）。
|
||||
from src.core.algorithms.waterindex_inversion.csv_processor import WaterIndexCsvProcessor # noqa: E402,F401
|
||||
|
||||
|
||||
# 保留旧 import 路径兼容
|
||||
|
||||
220
src/core/algorithms/waterindex_inversion/csv_processor.py
Normal file
220
src/core/algorithms/waterindex_inversion/csv_processor.py
Normal file
@ -0,0 +1,220 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
水色指数反演 — 散点 CSV 模式处理器（独立子模块）。
|
||||
|
||||
设计意图
|
||||
--------
|
||||
本模块与 ``waterindex_inversion.__init__.py`` 中的 ``WaterIndexProcessor``
|
||||
(栅格 BSQ 模式) **彻底解耦**，不依赖任何 osgeo / rasterio / gdal，仅依赖
|
||||
``pandas`` 与 ``src.utils.water_index.WaterQualityIndexCalculator``。
|
||||
|
||||
**为什么独立成文件？**
|
||||
|
||||
``__init__.py`` 顶部有 ``from osgeo import gdal, osr``（用于 BSQ 栅格模式），
|
||||
这意味着任何 ``from src.core.algorithms.waterindex_inversion import X``
|
||||
都会触发 osgeo 加载——而某些验证环境（无 gdal 包的 venv）会因此 ImportError。
|
||||
|
||||
本子模块独立后，可通过
|
||||
``from src.core.algorithms.waterindex_inversion.csv_processor import WaterIndexCsvProcessor``
|
||||
直接加载，**完全不触发** ``__init__.py`` 的 osgeo import 链，便于无 gdal 环境
|
||||
做端到端 smoke test。
|
||||
|
||||
**调用入口（由 Step 10 service / panel 调用）**::
|
||||
|
||||
from src.core.algorithms.waterindex_inversion.csv_processor import WaterIndexCsvProcessor
|
||||
proc = WaterIndexCsvProcessor(waterindex_csv_path)
|
||||
out = proc.compute_indices_from_csv(
|
||||
sampling_csv_path=...,
|
||||
output_dir=...,
|
||||
selected_formulas=[...],
|
||||
progress_callback=lambda msg, pct: ...,
|
||||
)
|
||||
|
||||
输出格式
|
||||
--------
|
||||
每个公式一个 CSV，三列严格为 ``longitude, latitude, <formula_name>``。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from typing import Callable, Dict, List, Optional
|
||||
|
||||
|
||||
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: "object") -> List[str]:
|
||||
"""识别光谱列：列名是浮点数字符串（Step 4 输出的 '400.000000' 形式）"""
|
||||
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
|
||||
Reference in New Issue
Block a user