services/step6-9：打通光谱计算与机器学习预测的核心独立服务

_smoke_new_arch.py

@@ -314,6 +314,82 @@ def smoke_e2e():
            "[Service✗]" in log_text and "execute_step2" in log_text,
            f"log 片段：{log_text[-200:]!r}")
 
+    # ---- 切到 step6（真实 view + 真实 service） ----
+    win.nav_list.setCurrentRow(5)
+    view_step6 = win._views.get("step6")
+    report("L3", "_views['step6'] 是真实 Step6View（已迁移）",
+           type(view_step6).__name__ == "Step6View",
+           f"type={type(view_step6).__name__}")
+
+    win.log_text.clear()
+    view_step6.run_btn.click()
+    loop = QEventLoop()
+    QTimer.singleShot(1500, loop.quit)
+    loop.exec_()
+
+    log_text = win.log_text.toPlainText()
+    report("L3", "step6 真实 service 已迁移：空 deglint_img_path 触发 Service✗ 错误分支",
+           "[Service✗]" in log_text and "execute_step6" in log_text,
+           f"log 片段：{log_text[-200:]!r}")
+
+    # ---- 切到 step7（真实 view + 真实 service） ----
+    win.nav_list.setCurrentRow(6)
+    view_step7 = win._views.get("step7")
+    report("L3", "_views['step7'] 是真实 Step7View（已迁移）",
+           type(view_step7).__name__ == "Step7View",
+           f"type={type(view_step7).__name__}")
+
+    win.log_text.clear()
+    view_step7.run_btn.click()
+    loop = QEventLoop()
+    QTimer.singleShot(1500, loop.quit)
+    loop.exec_()
+
+    log_text = win.log_text.toPlainText()
+    report("L3", "step7 真实 service 已迁移：空 training_csv_path 触发 Service✗ 错误分支",
+           "[Service✗]" in log_text and "execute_step7" in log_text,
+           f"log 片段：{log_text[-200:]!r}")
+
+    # ---- 切到 step8（真实 view + 真实 service） ----
+    win.nav_list.setCurrentRow(7)
+    view_step8 = win._views.get("step8")
+    report("L3", "_views['step8'] 是真实 Step8View（已迁移）",
+           type(view_step8).__name__ == "Step8View",
+           f"type={type(view_step8).__name__}")
+
+    # step8 默认 enable_checkbox=False → service 走 skipped；强制开启以触发 Service✗
+    if hasattr(view_step8, "enable_checkbox"):
+        view_step8.enable_checkbox.setChecked(True)
+
+    win.log_text.clear()
+    view_step8.run_btn.click()
+    loop = QEventLoop()
+    QTimer.singleShot(1500, loop.quit)
+    loop.exec_()
+
+    log_text = win.log_text.toPlainText()
+    report("L3", "step8 真实 service 已迁移：空 training_csv_path 触发 Service✗ 错误分支",
+           "[Service✗]" in log_text and "execute_step8" in log_text,
+           f"log 片段：{log_text[-200:]!r}")
+
+    # ---- 切到 step9（真实 view + 真实 service） ----
+    win.nav_list.setCurrentRow(8)
+    view_step9 = win._views.get("step9")
+    report("L3", "_views['step9'] 是真实 Step9View（已迁移）",
+           type(view_step9).__name__ == "Step9View",
+           f"type={type(view_step9).__name__}")
+
+    win.log_text.clear()
+    view_step9.run_btn.click()
+    loop = QEventLoop()
+    QTimer.singleShot(1500, loop.quit)
+    loop.exec_()
+
+    log_text = win.log_text.toPlainText()
+    report("L3", "step9 真实 service 已迁移：空 sampling_csv_path 触发 Service✗ 错误分支",
+           "[Service✗]" in log_text and "execute_step9" in log_text,
+           f"log 片段：{log_text[-200:]!r}")
+
     win.close()
 
 

src/new/main_view.py

@@ -106,32 +106,32 @@ ROUTES = [
         "name": "6. 光谱特征",
         "view_module": "src.new.views.step6_view",
         "view_class": "Step6View",
-        "service_module": "src.new.services.placeholder_service",
-        "service_func": "execute_placeholder",
+        "service_module": "src.new.services.step6_service",
+        "service_func": "execute_step6",
     },
     {
         "id": "step7",
         "name": "7. 水质光谱指数",
         "view_module": "src.new.views.step7_view",
         "view_class": "Step7View",
-        "service_module": "src.new.services.placeholder_service",
-        "service_func": "execute_placeholder",
+        "service_module": "src.new.services.step7_service",
+        "service_func": "execute_step7",
     },
     {
         "id": "step8",
         "name": "8. 机器学习建模",
         "view_module": "src.new.views.step8_view",
         "view_class": "Step8View",
-        "service_module": "src.new.services.placeholder_service",
-        "service_func": "execute_placeholder",
+        "service_module": "src.new.services.step8_service",
+        "service_func": "execute_step8",
     },
     {
         "id": "step9",
         "name": "9. 机器学习预测",
         "view_module": "src.new.views.step9_view",
         "view_class": "Step9View",
-        "service_module": "src.new.services.placeholder_service",
-        "service_func": "execute_placeholder",
+        "service_module": "src.new.services.step9_service",
+        "service_func": "execute_step9",
     },
     {
         "id": "step10",
@@ -233,7 +233,7 @@ class MainView(QMainWindow):
         self._build_routes()
 
         self._log("[Boot] MainView 初始化完成")
-        self._log(f"[Boot] 已注册 {len(ROUTES)} 条路由（13 个 view 全部真实，前 5 个步骤 service 已迁移）")
+        self._log(f"[Boot] 已注册 {len(ROUTES)} 条路由（13 个 view 全部真实，前 9 个步骤 service 已迁移）")
 
     # ------------------------------------------------------------------
     # UI 布局：左侧导航 + 右侧 stacked + 底部日志

src/new/services/step6_service.py

@@ -0,0 +1,171 @@
+# -*- coding: utf-8 -*-
+"""
+Step6 后端计算服务（光谱特征提取）
+====================================
+
+纯计算函数——绝对不引用 PyQt、绝对不引用 main_view、绝对不读写全局变量。它只：
+
+1. 从 ``config`` 字典读取参数；
+2. 调用旧版 ``DataPreparationStep.extract_training_spectra`` 在去耀斑影像
+   上按采样点坐标提取平均光谱；
+3. 返回结果字典 ``{status, output_path, message, mode}``。
+
+调用入口（由 main_view 在后台 QThread 中调用）：
+
+    execute_step6({
+        "deglint_img_path": "D:/deglint.bsq",   # 去耀斑影像路径
+        "csv_path": "D:/processed_data.csv",      # 采样点坐标 CSV
+        "boundary_path": "D:/water_mask.dat",     # 水域掩膜（可选，自动回退到 water_mask_path）
+        "glint_mask_path": "D:/severe_glint.dat", # 耀斑掩膜（独立运行时必填）
+        "water_mask_path": "D:/water_mask.dat",   # 别名（与 boundary_path 二选一）
+        "radius": 5,                              # 采样半径
+        "source_epsg": 4326,                      # 源坐标系
+        "enabled": True,
+        "output_path": "D:/training_spectra.csv", # 可选，未指定则用 work_dir 约定目录
+        "work_dir": "D:/workspace",               # 工作目录（main_view 注入）
+    })
+
+返回字典字段：
+
+* ``status``       : "completed" | "skipped" | "error"
+* ``output_path``  : 生成的 training_spectra.csv 路径（失败时为 None）
+* ``message``      : 人类可读说明
+* ``mode``         : "extract_spectra"（便于 UI 提示）
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict
+
+from src.core.steps.data_preparation_step import DataPreparationStep
+
+
+def _resolve_output_dir(output_path: str | None, work_dir: str) -> Path:
+    """根据 output_path / work_dir 计算 training_spectra.csv 输出目录"""
+    if output_path:
+        return Path(output_path).parent
+    return Path(work_dir) / "6_Spectral_Feature_Extraction"
+
+
+def execute_step6(config: Dict[str, Any]) -> Dict[str, Any]:
+    """Step 6 后端计算入口——纯函数
+
+    Args:
+        config: 由前端 view.get_config() 序列化、再经 main_view 注入 work_dir 的字典
+
+    Returns:
+        标准结果字典 ``{status, output_path, message, mode}``
+    """
+    # ---------- 入参规整 ----------
+    deglint_img_path = config.get("deglint_img_path")
+    csv_path = config.get("csv_path")
+    boundary_path = config.get("boundary_path")
+    glint_mask_path = config.get("glint_mask_path")
+    water_mask_path = config.get("water_mask_path")
+    radius = int(config.get("radius", 5))
+    source_epsg = int(config.get("source_epsg", 4326))
+    enabled = bool(config.get("enabled", True))
+    output_path = config.get("output_path")
+    work_dir = config.get("work_dir") or "."
+
+    output_dir = _resolve_output_dir(output_path, work_dir)
+    mode = "extract_spectra"
+
+    # ---------- 提前失败检查 ----------
+    if not enabled:
+        return {
+            "status": "skipped",
+            "output_path": None,
+            "message": "用户禁用此步骤（enabled=False）",
+            "mode": mode,
+        }
+    if not deglint_img_path:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": "未提供去耀斑影像路径（deglint_img_path）",
+            "mode": mode,
+        }
+    if not csv_path:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": "未提供采样点 CSV 路径（csv_path）",
+            "mode": mode,
+        }
+    if not Path(deglint_img_path).exists():
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"去耀斑影像不存在: {deglint_img_path}",
+            "mode": mode,
+        }
+    if not Path(csv_path).exists():
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"采样点 CSV 不存在: {csv_path}",
+            "mode": mode,
+        }
+
+    # ---------- 显式构造 output_path（如未指定）----------
+    if not output_path:
+        output_path = str(output_dir / "training_spectra.csv").replace("\\", "/")
+
+    # ---------- 智能回退：boundary_path ← water_mask_path ----------
+    final_boundary_path = boundary_path or water_mask_path
+
+    # ---------- 执行（包一层 try/except 把异常转 dict，避免炸线程） ----------
+    try:
+        print(f"[Step6 Service] 提取光谱: deglint={deglint_img_path}, csv={csv_path}")
+        print(f"[Step6 Service] 采样半径={radius}, EPSG={source_epsg}")
+        result_path = DataPreparationStep.extract_training_spectra(
+            deglint_img_path=deglint_img_path,
+            radius=radius,
+            source_epsg=source_epsg,
+            csv_path=csv_path,
+            boundary_path=final_boundary_path,
+            glint_mask_path=glint_mask_path,
+            water_mask_path=water_mask_path,
+            output_dir=output_dir,
+            callback=None,  # 日志由 main_view 统一接管
+        )
+    except FileNotFoundError as e:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"文件不存在: {e}",
+            "mode": mode,
+        }
+    except ValueError as e:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"参数错误: {e}",
+            "mode": mode,
+        }
+    except Exception as e:  # noqa: BLE001 —— service 层兜底捕获所有
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"{type(e).__name__}: {e}",
+            "mode": mode,
+        }
+
+    # ---------- 成功路径 ----------
+    p = Path(result_path)
+    if not p.exists():
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"DataPreparationStep.extract_training_spectra 未生成文件: {result_path}",
+            "mode": mode,
+        }
+
+    return {
+        "status": "completed",
+        "output_path": str(p).replace("\\", "/"),
+        "message": f"训练光谱已保存: {p.name}",
+        "mode": mode,
+    }

src/new/services/step7_service.py

@@ -0,0 +1,166 @@
+# -*- coding: utf-8 -*-
+"""
+Step7 后端计算服务（水质光谱指数计算）
+====================================
+
+纯计算函数——绝对不引用 PyQt、绝对不引用 main_view、绝对不读写全局变量。它只：
+
+1. 从 ``config`` 字典读取参数；
+2. 调用旧版 ``DataPreparationStep.calculate_water_quality_indices`` 在训练光谱
+   CSV 上批量计算指定公式（band_math 方法）；
+3. 返回结果字典 ``{status, output_path, message, mode}``。
+
+调用入口（由 main_view 在后台 QThread 中调用）：
+
+    execute_step7({
+        "training_csv_path": "D:/training_spectra.csv",         # 输入光谱 CSV
+        "formula_csv_file":  ".../src/gui/model/waterindex.csv", # 公式 CSV（必填）
+        "formula_names": ["NDCI", "Chl_Conc_NDCI", ...],         # 指定公式；None=全量
+        "enabled": True,
+        "output_file": "D:/training_spectra_indices.csv",        # 可选
+        "work_dir": "D:/workspace",                              # 工作目录（main_view 注入）
+    })
+
+返回字典字段：
+
+* ``status``       : "completed" | "skipped" | "error"
+* ``output_path``  : 生成的 indices CSV 路径（失败时为 None）
+* ``message``      : 人类可读说明
+* ``mode``         : "calc_indices"（便于 UI 提示）
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from src.core.steps.data_preparation_step import DataPreparationStep
+
+
+def _resolve_output_dir(output_file: str | None, work_dir: str) -> Path:
+    """根据 output_file / work_dir 计算 indices CSV 输出目录"""
+    if output_file:
+        return Path(output_file).parent
+    return Path(work_dir) / "7_Water_Quality_Indices"
+
+
+def execute_step7(config: Dict[str, Any]) -> Dict[str, Any]:
+    """Step 7 后端计算入口——纯函数
+
+    Args:
+        config: 由前端 view.get_config() 序列化、再经 main_view 注入 work_dir 的字典
+
+    Returns:
+        标准结果字典 ``{status, output_path, message, mode}``
+    """
+    # ---------- 入参规整 ----------
+    training_csv_path: Optional[str] = config.get("training_csv_path")
+    formula_csv_file: Optional[str] = config.get("formula_csv_file")
+    formula_names: Optional[List[str]] = config.get("formula_names")
+    enabled: bool = bool(config.get("enabled", True))
+    output_file: Optional[str] = config.get("output_file")
+    work_dir: str = config.get("work_dir") or "."
+
+    output_dir = _resolve_output_dir(output_file, work_dir)
+    mode = "calc_indices"
+
+    # ---------- 提前失败检查 ----------
+    if not enabled:
+        return {
+            "status": "skipped",
+            "output_path": None,
+            "message": "用户禁用此步骤（enabled=False）",
+            "mode": mode,
+        }
+    if not training_csv_path:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": "未提供训练光谱 CSV 路径（training_csv_path）",
+            "mode": mode,
+        }
+    if not formula_csv_file:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": "未提供公式 CSV 路径（formula_csv_file）",
+            "mode": mode,
+        }
+    if not Path(training_csv_path).exists():
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"训练光谱 CSV 不存在: {training_csv_path}",
+            "mode": mode,
+        }
+    if not Path(formula_csv_file).exists():
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"公式 CSV 不存在: {formula_csv_file}",
+            "mode": mode,
+        }
+
+    # ---------- 显式构造 output_file（如未指定）----------
+    if not output_file:
+        output_file = str(output_dir / "training_spectra_indices.csv").replace("\\", "/")
+
+    # ---------- 执行（包一层 try/except 把异常转 dict，避免炸线程） ----------
+    try:
+        formula_count = len(formula_names) if formula_names else "全量"
+        print(f"[Step7 Service] 计算水质指数: csv={training_csv_path}, formula_count={formula_count}")
+        result_path = DataPreparationStep.calculate_water_quality_indices(
+            training_csv_path=training_csv_path,
+            formula_csv_file=formula_csv_file,
+            formula_names=formula_names,
+            output_file=output_file,
+            enabled=enabled,
+            output_dir=output_dir,
+            callback=None,  # 日志由 main_view 统一接管
+        )
+    except FileNotFoundError as e:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"文件不存在: {e}",
+            "mode": mode,
+        }
+    except ValueError as e:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"参数错误: {e}",
+            "mode": mode,
+        }
+    except Exception as e:  # noqa: BLE001 —— service 层兜底捕获所有
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"{type(e).__name__}: {e}",
+            "mode": mode,
+        }
+
+    # ---------- 成功路径 ----------
+    if not result_path:
+        return {
+            "status": "skipped",
+            "output_path": None,
+            "message": "水质指数计算被跳过（enabled=False 或无产物）",
+            "mode": mode,
+        }
+
+    p = Path(result_path)
+    if not p.exists():
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"DataPreparationStep.calculate_water_quality_indices 未生成文件: {result_path}",
+            "mode": mode,
+        }
+
+    return {
+        "status": "completed",
+        "output_path": str(p).replace("\\", "/"),
+        "message": f"水质指数已保存: {p.name}",
+        "mode": mode,
+    }

src/new/services/step8_service.py

@@ -0,0 +1,147 @@
+# -*- coding: utf-8 -*-
+"""
+Step8 后端计算服务（机器学习建模训练）
+====================================
+
+纯计算函数——绝对不引用 PyQt、绝对不引用 main_view、绝对不读写全局变量。它只：
+
+1. 从 ``config`` 字典读取参数；
+2. 调用旧版 ``ModelingStep.train_models`` 在训练光谱 CSV 上做
+   GridSearchCV 多模型/多预处理/多划分方法的批量训练；
+3. 返回结果字典 ``{status, output_path, message, mode}``。
+
+调用入口（由 main_view 在后台 QThread 中调用）：
+
+    execute_step8({
+        "training_csv_path": "D:/training_spectra_indices.csv",  # 训练 CSV（必填）
+        "feature_start_column": "374.285004",                    # 特征起始列名/索引
+        "preprocessing_methods": ["None", "MMS"],                 # 预处理方法列表
+        "model_names": ["RF", "SVR", "Ridge", "Lasso"],          # 模型列表
+        "split_methods": ["spxy"],                                # 划分方法列表
+        "cv_folds": 5,                                            # 交叉验证折数
+        "enabled": True,
+        "output_path": "D:/8_Supervised_Model_Training",          # 可选，模型保存目录
+        "work_dir": "D:/workspace",                               # 工作目录（main_view 注入）
+    })
+
+返回字典字段：
+
+* ``status``       : "completed" | "skipped" | "error"
+* ``output_path``  : 模型保存目录路径（失败时为 None）
+* ``message``      : 人类可读说明
+* ``mode``         : "train_ml"（便于 UI 提示）
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from src.core.steps.modeling_step import ModelingStep
+
+
+def _resolve_output_dir(output_path: str | None, work_dir: str) -> Path:
+    """根据 output_path / work_dir 计算模型保存目录"""
+    if output_path:
+        return Path(output_path)
+    return Path(work_dir) / "8_Supervised_Model_Training"
+
+
+def execute_step8(config: Dict[str, Any]) -> Dict[str, Any]:
+    """Step 8 后端计算入口——纯函数
+
+    Args:
+        config: 由前端 view.get_config() 序列化、再经 main_view 注入 work_dir 的字典
+
+    Returns:
+        标准结果字典 ``{status, output_path, message, mode}``
+    """
+    # ---------- 入参规整 ----------
+    training_csv_path: Optional[str] = config.get("training_csv_path")
+    feature_start_column: str = str(config.get("feature_start_column", "374.285004"))
+    preprocessing_methods: Optional[List[str]] = config.get("preprocessing_methods")
+    model_names: Optional[List[str]] = config.get("model_names")
+    split_methods: Optional[List[str]] = config.get("split_methods")
+    cv_folds: int = int(config.get("cv_folds", 5))
+    enabled: bool = bool(config.get("enabled", True))
+    output_path: Optional[str] = config.get("output_path")
+    work_dir: str = config.get("work_dir") or "."
+
+    output_dir = _resolve_output_dir(output_path, work_dir)
+    mode = "train_ml"
+
+    # ---------- 提前失败检查 ----------
+    if not enabled:
+        return {
+            "status": "skipped",
+            "output_path": None,
+            "message": "用户禁用此步骤（enabled=False）",
+            "mode": mode,
+        }
+    if not training_csv_path:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": "未提供训练 CSV 路径（training_csv_path）",
+            "mode": mode,
+        }
+    if not Path(training_csv_path).exists():
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"训练 CSV 不存在: {training_csv_path}",
+            "mode": mode,
+        }
+
+    # ---------- 兜底默认值（与旧 panel 默认对齐）----------
+    if not preprocessing_methods:
+        preprocessing_methods = ["None", "MMS"]
+    if not model_names:
+        model_names = ["SVR"]
+    if not split_methods:
+        split_methods = ["spxy"]
+
+    # ---------- 执行（包一层 try/except 把异常转 dict，避免炸线程） ----------
+    try:
+        print(f"[Step8 Service] 训练 ML 模型: csv={training_csv_path}")
+        print(f"[Step8 Service] 模型={model_names}, 预处理={preprocessing_methods}, 划分={split_methods}, CV={cv_folds}")
+        result_dir = ModelingStep.train_models(
+            feature_start_column=feature_start_column,
+            preprocessing_methods=preprocessing_methods,
+            model_names=model_names,
+            split_methods=split_methods,
+            cv_folds=cv_folds,
+            training_csv_path=training_csv_path,
+            output_dir=output_dir,
+            callback=None,  # 日志由 main_view 统一接管
+        )
+    except FileNotFoundError as e:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"文件不存在: {e}",
+            "mode": mode,
+        }
+    except ValueError as e:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"参数错误: {e}",
+            "mode": mode,
+        }
+    except Exception as e:  # noqa: BLE001 —— service 层兜底捕获所有
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"{type(e).__name__}: {e}",
+            "mode": mode,
+        }
+
+    # ---------- 成功路径 ----------
+    p = Path(result_dir)
+    return {
+        "status": "completed",
+        "output_path": str(p).replace("\\", "/"),
+        "message": f"模型训练完成，结果保存在: {p.name or p}",
+        "mode": mode,
+    }

src/new/services/step9_service.py

@@ -0,0 +1,154 @@
+# -*- coding: utf-8 -*-
+"""
+Step9 后端计算服务（机器学习预测校正）
+====================================
+
+纯计算函数——绝对不引用 PyQt、绝对不引用 main_view、绝对不读写全局变量。它只：
+
+1. 从 ``config`` 字典读取参数；
+2. 调用旧版 ``PredictionStep.predict_water_quality`` 用训练好的 ML 模型
+   对采样点 CSV 做批量预测，输出每个目标参数的预测 CSV；
+3. 返回结果字典 ``{status, output_path, message, mode}``。
+
+调用入口（由 main_view 在后台 QThread 中调用）：
+
+    execute_step9({
+        "sampling_csv_path": "D:/sampling_spectra.csv",  # 采样点光谱 CSV（必填）
+        "models_dir": "D:/8_Supervised_Model_Training",  # 模型目录（必填）
+        "metric": "test_r2",                              # 模型选择指标
+        "prediction_column": "prediction",                # 预测列名
+        "enabled": True,
+        "output_path": "D:/9_ML_Prediction",              # 可选，预测输出目录
+        "work_dir": "D:/workspace",                        # 工作目录（main_view 注入）
+    })
+
+返回字典字段：
+
+* ``status``       : "completed" | "skipped" | "error"
+* ``output_path``  : 预测输出目录路径（失败时为 None）
+* ``message``      : 人类可读说明
+* ``mode``         : "predict_ml"（便于 UI 提示）
+
+注意：外部导入模型（``_external_models_dict``）模式暂未支持——该模式需要
+在主线程维护一个 dict of model objects，无法 pickle 跨 QThread 传递。
+如需启用，需在 main_view 端新增"先扫盘加载 → 转 joblib 路径 → 传 dict"
+的预处理链路。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict
+
+from src.core.steps.prediction_step import PredictionStep
+
+
+def _resolve_output_dir(output_path: str | None, work_dir: str) -> Path:
+    """根据 output_path / work_dir 计算预测结果输出目录"""
+    if output_path:
+        return Path(output_path)
+    # 与旧 pipeline 一致：prediction_dir / 9_ML_Prediction
+    return Path(work_dir) / "prediction" / "9_ML_Prediction"
+
+
+def execute_step9(config: Dict[str, Any]) -> Dict[str, Any]:
+    """Step 9 后端计算入口——纯函数
+
+    Args:
+        config: 由前端 view.get_config() 序列化、再经 main_view 注入 work_dir 的字典
+
+    Returns:
+        标准结果字典 ``{status, output_path, message, mode}``
+    """
+    # ---------- 入参规整 ----------
+    sampling_csv_path: str = config.get("sampling_csv_path") or ""
+    models_dir: str = config.get("models_dir") or ""
+    metric: str = str(config.get("metric", "test_r2"))
+    prediction_column: str = str(config.get("prediction_column", "prediction"))
+    enabled: bool = bool(config.get("enabled", True))
+    output_path: str = config.get("output_path")
+    work_dir: str = config.get("work_dir") or "."
+
+    output_dir = _resolve_output_dir(output_path, work_dir)
+    mode = "predict_ml"
+
+    # ---------- 提前失败检查 ----------
+    if not enabled:
+        return {
+            "status": "skipped",
+            "output_path": None,
+            "message": "用户禁用此步骤（enabled=False）",
+            "mode": mode,
+        }
+    if not sampling_csv_path:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": "未提供采样点 CSV 路径（sampling_csv_path）",
+            "mode": mode,
+        }
+    if not models_dir:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": "未提供模型目录（models_dir）",
+            "mode": mode,
+        }
+    if not Path(sampling_csv_path).exists():
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"采样点 CSV 不存在: {sampling_csv_path}",
+            "mode": mode,
+        }
+    if not Path(models_dir).exists():
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"模型目录不存在: {models_dir}",
+            "mode": mode,
+        }
+
+    # ---------- 执行（包一层 try/except 把异常转 dict，避免炸线程） ----------
+    try:
+        print(f"[Step9 Service] 预测水质参数: sampling={sampling_csv_path}")
+        print(f"[Step9 Service] models_dir={models_dir}, metric={metric}")
+        prediction_files = PredictionStep.predict_water_quality(
+            sampling_csv_path=sampling_csv_path,
+            models_dir=models_dir,
+            metric=metric,
+            prediction_column=prediction_column,
+            output_dir=output_dir,
+            callback=None,  # 日志由 main_view 统一接管
+        )
+    except FileNotFoundError as e:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"文件不存在: {e}",
+            "mode": mode,
+        }
+    except ValueError as e:
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"参数错误: {e}",
+            "mode": mode,
+        }
+    except Exception as e:  # noqa: BLE001 —— service 层兜底捕获所有
+        return {
+            "status": "error",
+            "output_path": None,
+            "message": f"{type(e).__name__}: {e}",
+            "mode": mode,
+        }
+
+    # ---------- 成功路径 ----------
+    p = Path(output_dir)
+    pred_count = len(prediction_files) if isinstance(prediction_files, dict) else 0
+    return {
+        "status": "completed",
+        "output_path": str(p).replace("\\", "/"),
+        "message": f"预测完成，共生成 {pred_count} 个目标参数的预测 CSV（目录: {p.name or p}）",
+        "mode": mode,
+    }