WQ_GUI/src/core/pipeline/runner.py

# -*- coding: utf-8 -*-
"""
PipelineRunner：基于 StepSpec 声明式调度 14 个 step。

设计要点：
  - StepSpec 声明 requires（ctx 字段名列表）+ produces（ctx 字段名列表）
  - 命名约定：ctx 字段名 == panel key 名 == step 形参名（全链路无翻译）
  - 保留 spec.parameter_map 字段骨架供极少数特例覆盖（默认空 dict）
  - 调度顺序：按 PIPELINE_STEPS 列表顺序，requires 缺则 skip
  - 软取消：在每个 step 前检查 ctx.is_cancelled()
  - duck-typed pipeline：runner 只调 getattr(pipeline, method_name)，不强依赖类层级
"""

from __future__ import annotations
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Sequence

from .context import PipelineContext


# ============================================================
# StepSpec 声明式描述
# ============================================================

@dataclass
class StepSpec:
    """单个 step 的元信息（声明式，避免硬编码）"""
    step_id: str
    method_name: str
    requires: List[str]                                # PipelineContext 字段名列表
    produces: List[str] = field(default_factory=list)  # 写入 ctx 的字段名列表
    enabled: bool = True
    parameter_map: Dict[str, str] = field(default_factory=dict)
    # 当 requires 中任一字段为 None 时是否跳过；默认 True（缺输入就 skip）
    skip_when_missing: bool = True
    # 备注（仅用于文档生成 / 调试输出）
    description: str = ""


# ============================================================
# 14 个 step 的声明表（顺序即调度顺序）
# 注：本表是"权威描述"，与 WorkerThread.step_method_map / 旧 run_full_pipeline 保持一致
# ============================================================

PIPELINE_STEPS: List[StepSpec] = [
    StepSpec(
        step_id="step1", method_name="step1_generate_water_mask",
        requires=["img_path"], produces=["water_mask_path"],
        description="水域掩膜生成（NDWI 或 SHP）",
    ),
    StepSpec(
        step_id="step2", method_name="step2_find_glint_area",
        requires=["img_path", "water_mask_path"], produces=["glint_mask_path"],
        description="耀斑区域检测",
    ),
    StepSpec(
        step_id="step3", method_name="step3_remove_glint",
        requires=["img_path", "water_mask_path", "glint_mask_path"],
        produces=["deglint_img_path"],
        description="耀斑去除",
    ),
    StepSpec(
        step_id="step4", method_name="step4_process_csv",
        requires=["csv_path"], produces=["processed_csv_path"],
        description="CSV 异常值清洗",
    ),
    StepSpec(
        step_id="step5", method_name="step5_extract_training_spectra",
        requires=["deglint_img_path", "processed_csv_path", "csv_path", "boundary_path", "glint_mask_path"],
        produces=["training_csv_path"],
        # processed_csv_path(step4 产物) 才是 step5 真正需要的主路径，
        # 通过 parameter_map 显式映射到形参 csv_path。
        # raw csv_path 也保留在 requires 中以备 user_config 覆盖，
        # 但用占位名 _raw_csv_ignored 注入，落到 step5 形参列表末尾的 **kwargs 兜底。
        # 这样可以避免 L2 顺序注入中"后注入的 csv_path=None 覆盖前面的 processed_csv_path"的冲突。
        parameter_map={
            "processed_csv_path": "csv_path",
            "csv_path": "_raw_csv_ignored",
        },
        skip_when_missing=False,
        description="实测样本点光谱提取",
    ),
    StepSpec(
        step_id="step5_5", method_name="step5_5_calculate_water_quality_indices",
        requires=["training_csv_path"], produces=["indices_path"],
        description="水质光谱指数计算（optional）",
    ),
    StepSpec(
        step_id="step6", method_name="step6_train_models",
        requires=["training_csv_path"], produces=["models_dir"],
        description="ML 建模（GridSearchCV / AutoML）",
    ),
    StepSpec(
        step_id="step6_5", method_name="step6_5_non_empirical_modeling",
        requires=["training_csv_path"], produces=["models_dir"],
        parameter_map={"training_csv_path": "csv_path"},
        description="非经验统计回归",
    ),
    StepSpec(
        step_id="step6_75", method_name="step6_75_custom_regression",
        requires=["indices_path"], produces=["models_dir"],
        parameter_map={"indices_path": "csv_path"},
        description="自定义回归分析",
    ),
    StepSpec(
        step_id="step7", method_name="step7_generate_sampling_points",
        requires=["deglint_img_path", "water_mask_path"], produces=["sampling_csv_path"],
        description="整景密集采样点生成 + 光谱提取",
    ),
    StepSpec(
        step_id="step8", method_name="step8_predict_water_quality",
        requires=["sampling_csv_path", "models_dir"], produces=["prediction_csv_path"],
        description="ML 模型预测（采样点）",
    ),
    StepSpec(
        step_id="step8_5", method_name="step8_5_predict_with_non_empirical_models",
        requires=["sampling_csv_path", "models_dir"], produces=["prediction_dir"],
        parameter_map={"models_dir": "non_empirical_models_dir"},
        description="非经验模型预测",
    ),
    StepSpec(
        step_id="step8_75", method_name="step8_75_predict_with_custom_regression",
        requires=["sampling_csv_path", "models_dir", "formula_csv_path"],
        produces=["prediction_dir"],
        parameter_map={"models_dir": "custom_regression_dir"},
        description="自定义回归预测",
    ),
    StepSpec(
        step_id="step9", method_name="step9_generate_distribution_map",
        requires=["prediction_csv_path", "boundary_shp_path"],
        produces=["distribution_map_path"],
        description="克里金插值成图",
    ),
]


# ============================================================
# PipelineRunner：执行者
# ============================================================

class PipelineRunner:
    """按 StepSpec 调度 14 个 step 方法，支持软取消 + 路径 ctx 注入。

    用法：
        runner = PipelineRunner(pipeline_instance)
        ctx = PipelineContext(img_path=..., ...)
        result_ctx = runner.run(ctx)
    """

    def __init__(self, pipeline, steps: Optional[Sequence[StepSpec]] = None):
        self.pipeline = pipeline
        self.steps: List[StepSpec] = list(steps) if steps else list(PIPELINE_STEPS)

    def run(self, ctx: PipelineContext) -> PipelineContext:
        """主入口：按顺序执行 14 步。软取消时已完成的 step 保留结果。"""
        ctx.pipeline_start_time = time.time()
        for spec in self.steps:
            if ctx.is_cancelled():
                ctx.append_log(f"[RUNNER] 收到取消信号，提前终止 @ {spec.step_id}")
                break
            if not spec.enabled:
                ctx.status[spec.step_id] = "skipped"
                ctx.append_log(f"[RUNNER] {spec.step_id} 标记为 disabled，跳过")
                continue
            if spec.skip_when_missing:
                missing = [k for k in spec.requires if not ctx.get(k)]
                if missing:
                    ctx.status[spec.step_id] = "skipped"
                    reason = f"缺少必要的上下文参数，自动跳过: {missing}"
                    ctx.append_log(f"[RUNNER] {spec.step_id} {reason}")
                    if hasattr(self.pipeline, "_notify"):
                        self.pipeline._notify(spec.description, "skipped", reason)
                    continue
            self._invoke(spec, ctx)
        ctx.pipeline_end_time = time.time()
        return ctx

    # ------------------------------------------------------------------
    def _invoke(self, spec: StepSpec, ctx: PipelineContext) -> None:
        """调一个 step 方法：ctx 路径 → 形参；产出 → ctx 字段。"""
        # DEBUG: 诊断"停在 step4"问题——每步打印 requires + ctx 实际数据
        # 看到 requires=[] 但 actual=[None,...] 就说明 ctx 缺料，step 会被 skip
        ctx.append_log(
            f"[DEBUG] Step {spec.step_id} requires: {spec.requires}, "
            f"actual ctx data: {[ctx.get(k) for k in spec.requires]}"
        )
        method = getattr(self.pipeline, spec.method_name, None)
        if method is None:
            ctx.append_log(f"[RUNNER] 步骤方法缺失: {spec.method_name}（跳过）")
            ctx.status[spec.step_id] = "skipped"
            return

        # 1) 把 ctx 路径作为形参注入（默认约定：去 _path 后缀）
        kwargs: Dict[str, Any] = {}
        for ctx_key in spec.requires:
            param_name = spec.parameter_map.get(ctx_key, self._default_param_name(ctx_key))
            kwargs[param_name] = ctx.get(ctx_key)

        # 2) 允许用户在 ctx.user_config[step_id] 覆盖/补充
        user_overrides = ctx.user_config.get(spec.step_id) or {}
        if isinstance(user_overrides, dict):
            for k, v in user_overrides.items():
                # ★ 关键防御：绝不用 GUI 的“空字符串”或 None 覆盖上游传来的有效路径
                if v is not None and v != "":
                    kwargs[k] = v

        # 3) 状态置 start
        ctx.append_log(
            f"[RUNNER] -> {spec.method_name}({list(kwargs.keys())})"
        )
        ctx.status[spec.step_id] = "start"
        notify = getattr(self.pipeline, "_notify", None)
        if callable(notify):
            try:
                notify(f"步骤{spec.step_id[-1]}", "start", spec.method_name)
            except Exception:
                pass

        # 4) 执行 + 捕获异常（不让单步崩溃拖垮 runner）
        t0 = time.time()
        try:
            result = method(**kwargs)
            ctx.status[spec.step_id] = "completed"
            ctx.step_timings[spec.step_id] = time.time() - t0

            # 5) 产出收割
            self._harvest(spec, result, ctx)

            if callable(notify):
                try:
                    notify(
                        f"步骤{spec.step_id[-1]}",
                        "completed",
                        str(result)[:200] if result is not None else "",
                    )
                except Exception:
                    pass
        except Exception as exc:
            ctx.status[spec.step_id] = "error"
            ctx.last_error = f"{spec.step_id}: {exc!r}"
            ctx.append_log(f"[RUNNER] {spec.step_id} 异常: {exc!r}")
            if callable(notify):
                try:
                    notify(f"步骤{spec.step_id[-1]}", "error", str(exc))
                except Exception:
                    pass

    # ------------------------------------------------------------------
    def _harvest(self, spec: StepSpec, result: Any, ctx: PipelineContext) -> None:
        """把 step 方法返回值灌入 ctx 的 produces 字段。

        规则：
          - 若 result 是 dict 且 key 匹配 produce_key：ctx.set(produce_key, result[key])
          - 若 result 非 dict 且 produces 非空：第一个 produces 字段接 result
          - 若 produces 为空：result 仅记录到 log，不写 ctx
        """
        if not spec.produces:
            return
        if isinstance(result, dict):
            for produce_key in spec.produces:
                if produce_key in result:
                    ctx.set(produce_key, result[produce_key])
        elif result is not None:
            ctx.set(spec.produces[0], result)

    # ------------------------------------------------------------------
    @staticmethod
    def _default_param_name(ctx_key: str) -> str:
        """
        废弃有毒的去 _path 后缀逻辑。
        默认原样返回 ctx 键名作为形参名。遇到特殊缩写时，由各个 step 的 parameter_map 显式处理。
        """
        return ctx_key