Step10 Kriging 输出路径强制 14_visualization + Step11 掩膜自动填入

src/core/water_quality_inversion_pipeline_GUI.py

@@ -1032,7 +1032,7 @@ class WaterQualityInversionPipeline:
         Args:
             prediction_csv_path: 预测结果CSV文件路径（前两列为经纬度，第三列为预测值）
             boundary_shp_path: 边界shapefile文件路径
-            output_image_path: 输出图片路径（如果为None，自动生成）
+            output_image_path: 输出图片路径（已废弃：本函数强制写入 self.visualization_dir，参数仅保留签名兼容）
             resolution: 插值网格分辨率（米）
             input_crs: 输入坐标系
             output_crs: 输出坐标系
@@ -1046,11 +1046,32 @@ class WaterQualityInversionPipeline:
             expand_ratio: 边界外扩比例（0-1之间）
 
         Returns:
-            可视化分布图文件路径
+            可视化分布图文件路径（始终位于 self.visualization_dir 下）
         """
-        if output_image_path is None:
+        # 修复：所有分布图（PNG）与底层 Kriging 输出的派生文件必须落到 14_visualization。
-            csv_name = Path(prediction_csv_path).stem
+        # 不论调用方（panel / 主流程 run_full_pipeline / 批量线程）传入什么路径，
-            output_image_path = str(self.visualization_dir / f"{csv_name}_distribution.png")
+        # 都在此强制 override 为 self.visualization_dir，规避
+        #   (a) 调用方误传 prediction_dir (11_12_13_predictions) 之类错位路径
+        #   (b) 老代码里硬编码字符串残留
+        # 若调用方传入的路径仍在 self.visualization_dir 内（子目录/不同文件名）则尊重其意图。
+        csv_name = Path(prediction_csv_path).stem
+        forced_image_path = str(self.visualization_dir / f"{csv_name}_distribution.png")
+        viz_dir_resolved = str(self.visualization_dir)
+        if output_image_path and output_image_path != forced_image_path:
+            # 判断调用方路径是否落在 visualization_dir 内（用 str.startswith 轻量检查）
+            norm_user = output_image_path.replace('\\', '/').rstrip('/')
+            norm_viz = viz_dir_resolved.replace('\\', '/').rstrip('/')
+            if not norm_user.startswith(norm_viz + '/') and norm_user != norm_viz:
+                print(
+                    f"⚠️ [step10_map] 调用方传入 output_image_path={output_image_path!r} "
+                    f"不在 {viz_dir_resolved} 下，强制重定向到 {forced_image_path}"
+                )
+                output_image_path = forced_image_path
+            else:
+                # 调用方路径已在 visualization_dir 内（如子目录），保留意图
+                output_image_path = output_image_path
+        else:
+            output_image_path = forced_image_path
 
         self._notify("started", "步骤9: 生成水质参数可视化分布图")
         result = MappingStep.generate_distribution_map(

src/gui/panels/step11_map_panel.py

@@ -566,6 +566,39 @@ class Step11MapPanel(QWidget):
                 if not existing_out or not existing_out.strip():
                     self.output_dir.set_path(output_dir)
 
+            # 4.5. 自动探测 Step1 水体掩膜（修复张冠李戴：原仅找 roi.shp，找不到时未尝试 1_water_mask）
+            #   优先调用 main_window.pipeline.get_step_output_dir('step1')（数据真实来源）
+            #   兜底走 resolve_subdir('water_mask') → <work_dir>/1_water_mask
+            #   Step1 典型产物：water_mask_from_ndwi.dat、water_mask_from_shp.dat、xxx.shp
+            if self.work_dir:
+                water_mask_dir = None
+                pipeline = None
+                try:
+                    _win = self.window()
+                    if _win is not None:
+                        pipeline = getattr(_win, 'pipeline', None)
+                except Exception:
+                    pipeline = None
+                if pipeline is not None and hasattr(pipeline, 'get_step_output_dir'):
+                    try:
+                        water_mask_dir = pipeline.get_step_output_dir('step1')
+                    except Exception as e:
+                        print(f"⚠️ [step11_map_panel] pipeline.get_step_output_dir('step1') 失败: {e}")
+                        water_mask_dir = None
+                if not water_mask_dir:
+                    water_mask_dir = resolve_subdir(self.work_dir, 'water_mask')
+
+                existing_boundary = (self.boundary_file.get_path() or "").strip()
+                if not existing_boundary and water_mask_dir and os.path.isdir(water_mask_dir):
+                    # 优先 .shp（geopandas 读矢量最稳），其次 .dat
+                    mask_candidates = (
+                        sorted(Path(water_mask_dir).glob("*.shp"))
+                        + sorted(Path(water_mask_dir).glob("*.dat"))
+                    )
+                    if mask_candidates:
+                        self.boundary_file.set_path(str(mask_candidates[0]))
+                        print(f"✅ [step11_map_panel] 自动从 Step1 掩膜目录填入: {mask_candidates[0]}")
+
             # 5. 自动探测原始矢量边界文件（.shp）作为专题图底图
             #  优先回溯 input-test/roi.shp，geopandas.read_file 仅支持矢量格式
             if self.work_dir:

tests/smoke_step10_path_override.py

@@ -0,0 +1,115 @@
+# -*- coding: utf-8 -*-
+"""
+Smoke test for: 彻底修复底层写入路径与掩膜联动
+
+验证三件事（不依赖 GUI / 不实例化 Pipeline，避免触发 osgeo/gdal 导入）：
+  1. pipeline.step10_map 内部对 output_image_path 的 override 逻辑正确：
+     - 路径不在 visualization_dir 下 → 被强制重定向
+     - 路径在 visualization_dir 下 → 保留
+     - 路径为 None/空 → 用 forced 默认值
+  2. step11_map_panel.update_from_config 含 pipeline.get_step_output_dir('step1') 调用
+  3. _step_path_resolver._FALLBACK_DIR_TABLE['water_mask'] == '1_water_mask'
+"""
+import re
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+PIPELINE_FILE = ROOT / "src" / "core" / "water_quality_inversion_pipeline_GUI.py"
+PANEL_FILE = ROOT / "src" / "gui" / "panels" / "step11_map_panel.py"
+RESOLVER_FILE = ROOT / "src" / "gui" / "panels" / "_step_path_resolver.py"
+
+
+def test_step10_map_forced_override():
+    """纯文本级检查 step10_map 是否含强制重定向逻辑。"""
+    text = PIPELINE_FILE.read_text(encoding="utf-8")
+    # 找 def step10_map( 起点；用下一个 def (8 空格缩进) 作锚点截取函数体
+    m = re.search(
+        r"def step10_map\([^\)]*\)[^\n]*:\n(.*?)(?=\n        def |\nclass |\Z)",
+        text, re.DOTALL,
+    )
+    assert m, "找不到 step10_map 函数"
+    body = m.group(1)
+
+    # 关键标记
+    assert "forced_image_path" in body, "step10_map 应计算 forced_image_path"
+    assert "强制重定向" in body, "step10_map 应有重定向提示文本"
+    assert "self.visualization_dir" in body, "step10_map 应引用 self.visualization_dir"
+    print("✅ step10_map 含强制 override 逻辑（forced_image_path + 重定向提示）")
+
+
+def test_step10_map_accepts_in_visualization_dir():
+    """模拟：当用户传入的路径已在 visualization_dir 内，应被保留（不被覆盖）。"""
+    # 走 source 的逻辑分支：在 startswith 命中 → 保留原路径
+    norm_viz = "/work/14_visualization"
+    candidates = [
+        ("/work/14_visualization/sub/foo.png", True),   # 子目录 → 保留
+        ("/work/14_visualization/foo.png", True),       # 自身 → 保留（== 情形走 else 分支）
+        ("/work/11_12_13_predictions/foo.png", False),  # 外部 → 强制重定向
+        ("/work/foo.png", False),
+    ]
+    for path, should_keep in candidates:
+        norm_user = path.replace("\\", "/").rstrip("/")
+        keep = norm_user.startswith(norm_viz + "/") or norm_user == norm_viz
+        assert keep == should_keep, f"路径 {path!r} 判断错误：keep={keep}, expect={should_keep}"
+    print("✅ step10_map 路径归属判断（startswith + ==）正确")
+
+
+def test_step11_panel_calls_pipeline_get_step_output_dir():
+    """验证 step11 panel 真的调用了 main_window.pipeline.get_step_output_dir('step1')。"""
+    text = PANEL_FILE.read_text(encoding="utf-8")
+    assert "get_step_output_dir('step1')" in text, \
+        "step11 panel 应调用 get_step_output_dir('step1')"
+    assert "getattr(_win, 'pipeline', None)" in text or "getattr(main_window, 'pipeline'" in text, \
+        "step11 panel 应安全访问 main_window.pipeline（None 守护）"
+    assert "resolve_subdir(self.work_dir, 'water_mask')" in text, \
+        "step11 panel 应有 resolve_subdir 兜底"
+    # 不应硬编码具体掩膜文件名（应通过 glob 探测）—— 排除注释行
+    code_lines = [
+        ln for ln in text.splitlines()
+        if ln.strip() and not ln.strip().startswith("#")
+    ]
+    code_only = "\n".join(code_lines)
+    assert "water_mask_from_ndwi" not in code_only, \
+        "step11 panel 不应硬编码 water_mask_from_ndwi（应通过 glob 探测）"
+    assert "water_mask_from_shp" not in code_only, \
+        "step11 panel 不应硬编码 water_mask_from_shp（应通过 glob 探测）"
+    print("✅ step11 panel 含 pipeline.get_step_output_dir('step1') 调用 + 兜底 + 安全守护")
+
+
+def test_fallback_dir_table_water_mask():
+    """验证 _FALLBACK_DIR_TABLE['water_mask'] == '1_water_mask'。"""
+    text = RESOLVER_FILE.read_text(encoding="utf-8")
+    m = re.search(r"'water_mask'\s*:\s*'([^']+)'", text)
+    assert m, "_FALLBACK_DIR_TABLE 缺 'water_mask' 键"
+    assert m.group(1) == "1_water_mask", f"应为 1_water_mask，实为 {m.group(1)}"
+    print(f"✅ _FALLBACK_DIR_TABLE['water_mask'] = {m.group(1)!r}")
+
+
+def test_panel_guard_does_not_overwrite_existing():
+    """验证 step11 panel 的新段不覆盖已有 boundary（feedback_never_overwrite_with_empty）。"""
+    text = PANEL_FILE.read_text(encoding="utf-8")
+    # 4.5 段必须先读 existing_boundary，再判断
+    m = re.search(
+        r"# 4\.5\..*?(?=\n            # 5\.)",
+        text, re.DOTALL
+    )
+    assert m, "找不到 4.5 段"
+    body = m.group(0)
+    assert "existing_boundary" in body, "4.5 段应读 existing_boundary"
+    assert "if not existing_boundary" in body, "4.5 段应仅在空时填入"
+    print("✅ step11 panel 4.5 段遵守 '非空值不覆盖' 原则")
+
+
+if __name__ == "__main__":
+    print("=" * 60)
+    print("Smoke test: 彻底修复底层写入路径与掩膜联动")
+    print("=" * 60)
+    test_step10_map_forced_override()
+    test_step10_map_accepts_in_visualization_dir()
+    test_step11_panel_calls_pipeline_get_step_output_dir()
+    test_fallback_dir_table_water_mask()
+    test_panel_guard_does_not_overwrite_existing()
+    print("=" * 60)
+    print("全部通过 ✅")
+    sys.exit(0)