格式统一

src/utils/water_index.py

@@ -6,6 +6,8 @@ import re
 from pathlib import Path
 from typing import Dict, List, Optional, Union
 
+from .band_math import BandMathCalculator
+
 
 def _get_resource_path(relative_path: str) -> str:
     """获取资源文件路径，兼容开发/PyInstaller onedir/onefile 三种环境。"""
@@ -88,62 +90,80 @@ class WaterQualityIndexCalculator:
         parts = [float(x.strip()) for x in s.split(",")]
         return np.array(parts)
 
-    def _band_math_all_rows(self, df: pd.DataFrame, expression: str, wavelength_offset: float = 0.0) -> pd.Series:
+    def _band_math_all_rows(self, df: pd.DataFrame, expression: str, wavelength_offset: float = 0.0,
+                             fast: bool = False) -> pd.Series:
         """
-        向量化批量计算波段表达式（2026-07-01 重写：逐行 eval → 全列 numpy）。
+        波段表达式批量计算。
 
-        原先逐行 eval + re.sub 在 11310 行 × 63 公式时超过 600 秒；
-        现在一次性解析表达式为 numpy 操作，全 DataFrame 向量化计算，
-        同等数据量下 < 1 秒。
+        Args:
+            fast=True  → 优先向量化全列 numpy（快，但极少数公式可能触发 0xC0000005）
+            fast=False → 纯逐行 eval（稳定，不会崩）
         """
-        # ── 1. 从列名提取波长列表 ──
-        wavelengths = []
-        for col in df.columns:
-            nums = re.findall(r'\d+\.?\d*', str(col))
-            wavelengths.append(float(nums[0]) if nums else None)
+        calc = BandMathCalculator.__new__(BandMathCalculator)
+        calc.df = df
+        calc.wavelengths = calc._extract_wavelengths()
+        calc.wavelength_offset = float(wavelength_offset)
 
-        # ── 2. 解析表达式中的变量 (wNNN / WNNN) → 找到对应列索引 ──
-        var_pattern = r'[wW](\d+\.?\d*)'
-        var_matches = re.findall(var_pattern, expression)
-        col_map = {}  # {原始变量文本: 列索引}
-        for var_str in var_matches:
-            target_wl = float(var_str) + wavelength_offset
-            valid = [(i, wl) for i, wl in enumerate(wavelengths) if wl is not None]
+        variables = calc._parse_expression(expression)
+
+        var_col_map = {}
+        for var_str in variables:
+            wavelength = float(var_str) + wavelength_offset
+            valid = [(i, wl) for i, wl in enumerate(calc.wavelengths) if wl is not None]
             if not valid:
-                raise ValueError("未找到有效的波长列")
-            best_idx = min(valid, key=lambda x: abs(x[1] - target_wl))[0]
-            # 同时覆盖 wNNN 和 WNNN 两种写法
-            col_map[f'w{var_str}'] = best_idx
-            col_map[f'W{var_str}'] = best_idx
+                continue
+            col_idx = min(valid, key=lambda x: abs(x[1] - wavelength))[0]
+            var_col_map[f'w{var_str}'] = col_idx
+            var_col_map[f'W{var_str}'] = col_idx
 
-        # ── 3. 构建带缓存的向量化 numpy 表达式 ──
-        # 注意：w715 这种变量在 Python 中是合法标识符，但 eval 中会当变量名；
-        # 我们直接替换为 arr[:, col_idx] 再传给 eval，确保一次性全列计算。
-        arr = df.values  # (N, M) numpy array，避免重复 .iloc 访问
-        eval_expr = expression
-        # 按变量名长度降序替换，防止短变量吞噬长变量前缀（如 w715 先于 w71）
-        for var_name in sorted(col_map.keys(), key=len, reverse=True):
-            col_idx = col_map[var_name]
-            # 替换为安全的列引用，匹配完整 token（用 \b 边界）
-            eval_expr = re.sub(
-                r'\b' + re.escape(var_name) + r'\b',
-                f'arr[:, {col_idx}]',
-                eval_expr,
-            )
+        if not var_col_map:
+            return pd.Series(np.nan, index=df.index, name=expression)
 
-        # ── 4. 一次性向量化求值 ──
-        try:
-            with np.errstate(divide='ignore', invalid='ignore'):
-                result = eval(eval_expr, {"__builtins__": None}, {"arr": arr, "np": np})
-        except Exception:
-            # 回退：返回全 NaN 列
-            result = np.full(len(df), np.nan)
+        # ── 快车道：向量化全列 numpy（仅 fast=True 时启用）──
+        if fast:
+            try:
+                arr = df.values.astype(np.float64, copy=False)
+                vec_expr = expression
+                for var_name in sorted(var_col_map.keys(), key=len, reverse=True):
+                    col_idx = var_col_map[var_name]
+                    if col_idx < 0 or col_idx >= arr.shape[1]:
+                        raise ValueError(f'列索引 {col_idx} 越界')
+                    vec_expr = re.sub(
+                        r'\b' + re.escape(var_name) + r'\b',
+                        f'arr[:, {col_idx}]',
+                        vec_expr,
+                    )
+                with np.errstate(divide='ignore', invalid='ignore'):
+                    result = eval(vec_expr, {'__builtins__': None}, {'arr': arr, 'np': np})
+                result = np.asarray(result, dtype=np.float64).ravel().copy()
+                if len(result) == len(df) and np.isfinite(result).sum() > 0:
+                    return pd.Series(result, index=df.index, name=expression)
+            except Exception:
+                pass  # 向量化失败 → 回退逐行
 
-        # 确保结果是一维的
-        result = np.asarray(result).ravel()
-        return pd.Series(result, index=df.index, name=expression)
+        # ── 慢车道：逐行 eval（永远不会崩）──
+        n_rows = len(df)
+        results = [np.nan] * n_rows
+        for i in range(n_rows):
+            calc_expr = expression
+            for var_pattern, col_idx in var_col_map.items():
+                value = df.iloc[i, col_idx]
+                calc_expr = re.sub(
+                    r'\b' + re.escape(var_pattern) + r'\b',
+                    f'({value})',
+                    calc_expr,
+                )
+            try:
+                with np.errstate(divide='ignore', invalid='ignore'):
+                    r = eval(calc_expr, {'__builtins__': None}, {'nan': np.nan, 'inf': np.inf, 'np': np})
+            except Exception:
+                r = np.nan
+            results[i] = r
 
-    def calculate_one(self, name: str, df: pd.DataFrame, wavelength_offset: float = 0.0) -> pd.Series:
+        return pd.Series(results, index=df.index, name=expression)
+
+    def calculate_one(self, name: str, df: pd.DataFrame, wavelength_offset: float = 0.0,
+                      fast: bool = False) -> pd.Series:
         """
         计算单个水质指数。
 
@@ -163,17 +183,22 @@ class WaterQualityIndexCalculator:
         ftype = cfg["type"]
         coeff_str = cfg["coeff"]
 
-        raw = self._band_math_all_rows(df, expr, wavelength_offset=wavelength_offset)
+        raw = self._band_math_all_rows(df, expr, wavelength_offset=wavelength_offset, fast=fast)
 
         if ftype == "concentration":
             coeff = self._parse_coeff(coeff_str)
-            result = np.polyval(coeff, raw.values)
+            # ★ .copy() 强制物化：polyval 返回的数组可能引用临时内存
+            result = np.asarray(np.polyval(coeff, raw.values), dtype=np.float64).copy()
             raw = pd.Series(result, index=df.index, name=name)
 
+        # ★ 强制物化：向量化路径产生的 Series 内部数组可能指向临时内存，
+        #   后续 csv_processor 的 .values / .replace / .clip 访问会触发 0xC0000005
+        raw = raw.copy()
         raw.name = name
         return raw
 
-    def calculate_many(self, names: List[str], df: pd.DataFrame, wavelength_offset: float = 0.0) -> pd.DataFrame:
+    def calculate_many(self, names: List[str], df: pd.DataFrame, wavelength_offset: float = 0.0,
+                       fast: bool = False) -> pd.DataFrame:
         """
         批量计算多个水质指数。
 
@@ -181,19 +206,17 @@ class WaterQualityIndexCalculator:
             names: 公式名称列表
             df: 光谱反射率 DataFrame
             wavelength_offset: 波长偏移修正量（nm）
-
-        Returns:
-            pd.DataFrame，每列对应一个公式的计算结果
+            fast: True=优先向量化（快，Step9用）；False=纯逐行（稳定，Step10用）
         """
         results = {}
         for name in names:
             try:
-                results[name] = self.calculate_one(name, df, wavelength_offset=wavelength_offset)
+                results[name] = self.calculate_one(name, df, wavelength_offset=wavelength_offset, fast=fast)
             except Exception as e:
                 print(f"⚠️ 计算 {name} 失败: {e}")
                 results[name] = pd.Series(np.nan, index=df.index, name=name)
 
-        return pd.DataFrame(results)
+        return pd.DataFrame(results).copy() if fast else pd.DataFrame(results)
 
     # =========================================================================
     # 向后兼容：保留原有外部调用接口