# -*- coding: utf-8 -*- """ Step3 后端计算服务（耀斑去除） ==================================== 纯计算函数——绝对不引用 PyQt、绝对不引用 main_view、绝对不读写全局变量。它只： 1. 从 ``config`` 字典读取参数； 2. 调用旧版 ``GlintRemovalStep.run`` 执行去耀斑（4 种方法：Goodman/Kutser/Hedley/SUGAR）； 3. 返回结果字典 ``{status, output_path, message, mode}``。 调用入口（由 main_view 在后台 QThread 中调用）： execute_step3({ "img_path": "D:/ref.bsq", # 输入影像（去耀斑前） "method": "goodman", # goodman/kutser/hedley/sugar "enabled": True, "interpolate_zeros": False, # 是否先做 0 值像素插值 "interpolation_method": "bilinear", "water_mask_path": "D:/mask.dat", # 水域掩膜（可选） "output_path": "D:/deglint_image.bsq", # 方法专属参数（按 method 任选一组） "nir_lower": 65, "nir_upper": 91, "goodman_A": 1.9e-5, "goodman_B": 0.1, # goodman "oxy_band": 38, "lower_oxy": 36, "upper_oxy": 49, "nir_band": 47, # kutser "hedley_nir_band": 47, # hedley "sugar_iter": 3, "sugar_sigma": 1.0, "sugar_estimate_background": True, "sugar_glint_mask_method": "cdf", "sugar_termination_thresh": 20.0, "sugar_bounds": [(1, 2)], # sugar "work_dir": "D:/workspace", # 工作目录（main_view 注入） }) 返回字典字段： * ``status`` : "completed" | "skipped" | "error" * ``output_path`` : 生成的 .bsq 去耀斑影像路径（失败时为 None） * ``message`` : 人类可读说明 * ``mode`` : 实际调用的去耀斑方法名，便于 UI 提示 """ from __future__ import annotations from pathlib import Path from typing import Any, Dict from src.core.steps.glint_removal_step import GlintRemovalStep from src.new.services._output_resolver import ( copy_to_user_path, get_user_output_path, is_user_specified, resolve_output_dir, ) def _resolve_dirs(config: Dict[str, Any], work_dir: str) -> tuple[Path, Path]: """根据 output_path / work_dir 计算 (deglint_dir, water_mask_dir) 使用共享解析器强制执行"用户优先"规则——用户指定 output_path 时用其父目录， 否则用 work_dir/3_Deglint 默认。 """ deglint_dir, _source = resolve_output_dir( config, work_dir, "3_Deglint", "output_path", "output_dir" ) water_mask_dir = Path(work_dir) / "1_water_mask" return deglint_dir, water_mask_dir def _normalize_method(method: str) -> str: """方法名标准化（与 GlintRemovalStep.run 内部规则保持一致）""" raw = str(method).lower() if "kutser" in raw: return "kutser" if "goodman" in raw: return "goodman" if "hedley" in raw: return "hedley" if "sugar" in raw: return "sugar" return raw def _build_method_kwargs(method: str, config: Dict[str, Any]) -> Dict[str, Any]: """按 method 从 config 中抽取对应的方法专属参数""" if method == "goodman": return { "nir_lower": int(config.get("nir_lower", 65)), "nir_upper": int(config.get("nir_upper", 91)), "goodman_A": float(config.get("goodman_A", 0.000019)), "goodman_B": float(config.get("goodman_B", 0.1)), } if method == "kutser": return { "oxy_band": int(config.get("oxy_band", 38)), "lower_oxy": int(config.get("lower_oxy", 36)), "upper_oxy": int(config.get("upper_oxy", 49)), "nir_band": int(config.get("nir_band", 47)), } if method == "hedley": return { "hedley_nir_band": int(config.get("hedley_nir_band", 47)), } if method == "sugar": bounds = config.get("sugar_bounds") if bounds is None or not isinstance(bounds, (list, tuple)): bounds = [(1, 2)] return { "sugar_iter": int(config.get("sugar_iter", 3)), "sugar_sigma": float(config.get("sugar_sigma", 1.0)), "sugar_estimate_background": bool(config.get("sugar_estimate_background", True)), "sugar_glint_mask_method": str(config.get("sugar_glint_mask_method", "cdf")), "sugar_termination_thresh": float(config.get("sugar_termination_thresh", 20.0)), "sugar_bounds": bounds, } return {} def execute_step3(config: Dict[str, Any]) -> Dict[str, Any]: """Step 3 后端计算入口——纯函数 Args: config: 由前端 view.get_config() 序列化、再经 main_view 注入 work_dir 的字典 Returns: 标准结果字典 ``{status, output_path, message, mode}`` """ # ---------- 入参规整 ---------- img_path = config.get("img_path") raw_method = config.get("method", "goodman") method = _normalize_method(raw_method) enabled = bool(config.get("enabled", True)) interpolate_zeros = bool(config.get("interpolate_zeros", False)) interpolation_method = str(config.get("interpolation_method", "bilinear")) water_mask_path = config.get("water_mask_path") output_path = config.get("output_path") work_dir = config.get("work_dir") or "." deglint_dir, water_mask_dir = _resolve_dirs(config, work_dir) # ---------- 提前失败检查 ---------- if not enabled: return { "status": "skipped", "output_path": img_path, "message": "用户禁用此步骤（enabled=False），保留原始影像", "mode": method, } if not img_path: return { "status": "error", "output_path": None, "message": "未提供输入影像路径（img_path）", "mode": method, } if not Path(img_path).exists(): return { "status": "error", "output_path": None, "message": f"输入影像不存在: {img_path}", "mode": method, } # ---------- 构建底层 kwargs ---------- # 注意：GlintRemovalStep.run 不接受 output_path 关键字——它只接收 deglint_dir； # 用户指定的文件名将通过下文的 copy_to_user_path 事后劫持拷贝。 method_kwargs = _build_method_kwargs(method, config) kwargs: Dict[str, Any] = { "img_path": img_path, "method": method, "water_mask": water_mask_path, "interpolate_zeros": interpolate_zeros, "interpolation_method": interpolation_method, "deglint_dir": deglint_dir, "water_mask_dir": water_mask_dir, "callback": None, # 日志由 main_view 统一接管 } kwargs.update(method_kwargs) # ---------- 执行（包一层 try/except 把异常转 dict，避免炸线程） ---------- try: result_path = GlintRemovalStep.run(**kwargs) except FileNotFoundError as e: return { "status": "error", "output_path": None, "message": f"文件不存在: {e}", "mode": method, } except ValueError as e: return { "status": "error", "output_path": None, "message": f"参数错误: {e}", "mode": method, } except Exception as e: # noqa: BLE001 —— service 层兜底捕获所有 return { "status": "error", "output_path": None, "message": f"{type(e).__name__}: {e}", "mode": method, } # ---------- 成功路径 ---------- p = Path(result_path) if not p.exists(): return { "status": "error", "output_path": None, "message": f"GlintRemovalStep.run 未生成文件: {result_path}", "mode": method, } # ---------- 事后劫持：用户指定文件名 vs 底层硬编码文件名 ---------- # 旧版 GlintRemovalStep.run 只接受 deglint_dir 不接受确切文件名； # 用户浏览指定的 .bsq 文件名被底层忽略（同时 .hdr 头文件也按硬编码名生成）。 # 这里事后把 result_path 拷贝/重命名到 user_path，copy_to_user_path # 会自动处理 .hdr / .HDR 伴随文件。 user_path = config.get("output_path") if user_path: result_path = copy_to_user_path(result_path, user_path) p = Path(result_path) return { "status": "completed", "output_path": str(p).replace("\\", "/"), "message": f"去耀斑影像已生成: {p.name}", "mode": method, }