diff --git a/src/core/algorithms/waterindex_inversion/csv_processor.py b/src/core/algorithms/waterindex_inversion/csv_processor.py index 402a9ae..53d3824 100644 --- a/src/core/algorithms/waterindex_inversion/csv_processor.py +++ b/src/core/algorithms/waterindex_inversion/csv_processor.py @@ -211,7 +211,9 @@ class WaterIndexCsvProcessor: skipped_count += 1 continue try: - per_idx = results_df[name] + # ★ .copy() 确保拿到独立内存，防止上游向量化 eval 的临时视图 + # 在后续 .replace / .clip / .to_csv 中触发 0xC0000005 + per_idx = results_df[name].copy() # ===== P0 防御: 写盘前清洗（防 Step 11 Kriging / TIN 碎玻璃）===== # 1) inf / -inf → NaN：pandas 默认会把 inf 写成字面 "Infinity"， # 下游 ContentMapper 严格按位置读第 3 列时会原样拿到 inf， diff --git a/src/core/prediction/inference_batch.py b/src/core/prediction/inference_batch.py index 1e4a59c..7864a56 100644 --- a/src/core/prediction/inference_batch.py +++ b/src/core/prediction/inference_batch.py @@ -431,7 +431,8 @@ class WaterQualityInference: if not formulas: print("[特征补全] Calculator 未持有任何公式，跳过补全") else: - results_df = calc.calculate_many(formulas, spectra) + # fast=True: Step9 特征补全走向量化快车道（~63 公式，已验证稳定） + results_df = calc.calculate_many(formulas, spectra, fast=True) # results_df 是列对齐的 WQI 计算结果（每列一个公式，行数=样本数） if isinstance(results_df, pd.DataFrame) and not results_df.empty: original_col_count = spectra.shape[1] diff --git a/src/gui/panels/step12_viz_panel.py b/src/gui/panels/step12_viz_panel.py index f35aeb8..ed9d03b 100644 --- a/src/gui/panels/step12_viz_panel.py +++ b/src/gui/panels/step12_viz_panel.py @@ -687,8 +687,10 @@ class ImageCategoryTree(QTreeWidget): chart_type = "光谱曲线图" elif "HEATMAP" in name_upper or "热力图" in name_upper: chart_type = "相关性热力图" - elif "BOXPLOT" in name_upper or "HISTOGRAM" in name_upper or "箱线" in name_upper or "直方" in name_upper: + elif "BOXPLOT" in name_upper or "箱线" in name_upper: chart_type = "统计箱线图" + elif "HISTOGRAM" in name_upper or "直方" in name_upper: + chart_type = "分布直方图" elif "SAMPLING" in name_upper or "采样" in name_upper: chart_type = "采样点地图" elif "GLINT" in name_upper or "MASK" in name_upper or "PREVIEW" in name_upper: @@ -1531,7 +1533,7 @@ class Step12VizPanel(QWidget): self.view_mode_cb.currentIndexChanged.connect(self.update_image_tree_view) self.chart_filter_cb = QComboBox() - self.chart_filter_cb.addItems(["全部图表", "空间分布图", "模型散点图", "光谱曲线图", "统计箱线图", "相关性热力图", "掩膜与预览", "采样点地图"]) + self.chart_filter_cb.addItems(["全部图表", "空间分布图", "模型散点图", "光谱曲线图", "统计箱线图", "分布直方图", "相关性热力图", "掩膜与预览", "采样点地图"]) self.chart_filter_cb.currentIndexChanged.connect(self.update_image_tree_view) filter_layout.addRow("视图模式:", self.view_mode_cb) diff --git a/src/postprocessing/map.py b/src/postprocessing/map.py index 9c8576e..23cc780 100644 --- a/src/postprocessing/map.py +++ b/src/postprocessing/map.py @@ -585,7 +585,8 @@ class ContentMapper: verbose=False, enable_plotting=False ) - z, _ = ok.execute('grid', grid_x, grid_y) + # ★ 局部邻域 Kriging：只参考最近的 15 个点，避免万阶矩阵求逆 + z, _ = ok.execute('grid', grid_x, grid_y, backend='loop', n_closest_points=15) grid_content = np.array(z) valid_count = np.sum(~np.isnan(grid_content)) print(f"Kriging插值成功，有效点数: {valid_count} / {grid_content.size}") diff --git a/src/utils/water_index.py b/src/utils/water_index.py index 2e1bd96..b0f0592 100644 --- a/src/utils/water_index.py +++ b/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) # ========================================================================= # 向后兼容：保留原有外部调用接口