# -*- coding: utf-8 -*- """ Step 路径解析器——统一消灭 panel 端的硬编码路径与"张冠李戴"跨面板引用。 提供三个公共 API： - resolve_step_widget(main_window, step_key) - get_step_output_path(main_window, step_key, work_dir=None) - STEP_DATA_SOURCE 映射表 典型使用： from src.gui.panels._step_path_resolver import ( resolve_step_widget, get_step_output_path ) # 替换前：getattr(main_window.step11_panel, 'output_file', None) # 死代码 # 替换后： widget = resolve_step_widget(main_window, 'step11_predictions') # 找到正确 widget if widget: ... """ from pathlib import Path from typing import Optional, Union # 用户口语编号 / 业务别名 → PANEL_REGISTRY 中真实 step_id 的映射 # 这是"张冠李戴"修复的核心——main_window 上已不再直接挂载 panel 属性， # 所有面板都通过 _panel_factory.get_panel(step_id) 懒加载访问。 STEP_DATA_SOURCE = { # 数据流 step 编号（用户口语） → PANEL_REGISTRY 中的 step_id 'step5_clean_output': 'step5_clean', 'step7_index_output': 'step7_index', 'step8_ml_train_output': 'step8_ml_train', 'step8_5_non_empirical': 'step8_ml_train', 'step9_ml_predict_output': 'step9_ml_predict', 'step10_watercolor_output': 'step10_watercolor', 'step11_ml_prediction': 'step9_ml_predict', # 主流程 step11 = ML 预测 'step12_regression_prediction': 'step8_ml_train', # 主流程 step12 = 非经验预测 'step13_custom_regression': 'step13_report', # 自定义回归借用 step13 报告面板 'sampling_csv': 'step4_sampling', 'training_spectra_csv': 'step5_clean', 'indices_csv': 'step7_index', 'models_dir': 'step8_ml_train', 'watercolor_dir': 'step10_watercolor', 'prediction_csv_dir': 'step9_ml_predict', # 默认从 ML 预测读 } def _read_widget_path(widget) -> str: """统一从 widget 读 path（兼容 FileSelectWidget / QLineEdit / 字符串）。""" if widget is None: return "" if hasattr(widget, 'get_path'): try: return str(widget.get_path() or "").strip() except Exception: return "" if hasattr(widget, 'text'): try: return str(widget.text() or "").strip() except Exception: return "" if isinstance(widget, str): return widget.strip() return "" def resolve_step_widget(main_window, step_key: str, widget_attr: str = 'output_file'): """通过 panel_factory 访问对应面板，并获取其实际的输入/输出控件。 解析顺序： 1. STEP_DATA_SOURCE[step_key] 找到真实 step_id 2. main_window._panel_factory.get_panel(step_id) 懒加载拿到面板 3. getattr(panel, widget_attr) 取出真实控件 Returns: widget 对象 or None（找不到时返回 None，调用方需自行兜底） """ step_id = STEP_DATA_SOURCE.get(step_key) if not step_id: return None # 从主窗口获取工厂，再由工厂通过 step_id 拿出真实面板 factory = getattr(main_window, '_panel_factory', None) if not factory: return None panel = factory.get_panel(step_id) if not panel: return None return getattr(panel, widget_attr, None) _FALLBACK_DIR_TABLE = { # pipeline key（与 _ensure_step_dir_map 对齐）→ 子目录名 'step1': '1_water_mask', 'step2': '2_Glint_Detection', 'step3': '3_deglint', 'step4_sampling': '4_sampling', 'step5_clean': '5_Data_Cleaning', 'step6_feature': '6_Spectral_Feature_Extraction', 'step7_index': '7_Water_Quality_Indices', 'step8_ml_train': '8_Supervised_Model_Training', 'step8': '8_Supervised_Model_Training', 'step9_ml_predict': '8_Non_Empirical_Regression', 'step9': '8_Non_Empirical_Regression', 'step10_watercolor': '10_WaterIndex_Images', 'step10': '10_WaterIndex_Images', 'step11_map': '14_visualization', 'step11': '11_12_13_predictions', 'step11_predictions': '11_12_13_predictions', 'step12': '13_Custom_Regression', 'step12_predictions': '11_12_13_predictions', 'step13': 'reports', 'step13_predictions': '11_12_13_predictions', 'step14': '14_visualization', 'prediction_dir': '11_12_13_predictions', 'visualization': '14_visualization', 'reports': 'reports', 'custom_regression': '13_Custom_Regression', # 扩展：覆盖 panel 内部使用的子目录别名 'water_mask': '1_water_mask', 'glint_detection': '2_Glint_Detection', 'deglint': '3_deglint', 'sampling': '4_sampling', 'data_cleaning': '5_Data_Cleaning', 'spectral_feature': '6_Spectral_Feature_Extraction', 'indices': '7_Water_Quality_Indices', 'supervised_models': '8_Supervised_Model_Training', 'non_empirical': '8_Non_Empirical_Regression', 'qaa_inversion': '8_QAA_Inversion', 'regression_modeling': '8_Regression_Modeling', 'watercolor': '10_WaterIndex_Images', 'ml_prediction': '9_ML_Prediction', 'sampling_csv_path': '4_sampling/sampling_spectra.csv', } def get_step_output_path( main_window, step_key: str, work_dir: Optional[Union[str, Path]] = None, widget_attr: str = 'output_file', fallback_key: Optional[str] = None, ) -> str: """获取 step_key 指向的输出路径（带 main_window 解析 + 兜底路径）。 解析顺序： 1. STEP_DATA_SOURCE[step_key] 找到对应 panel，从 widget 读用户填的 path 2. 若为空字符串，用 _FALLBACK_DIR_TABLE[fallback_key or step_key] + work_dir 拼兜底 3. 全失败返回 str(work_dir) 注意：不创建 pipeline 实例（避免触发 osgeo 导入），用本地子目录字典兜底。 """ wd = str(work_dir) if work_dir else "" widget = resolve_step_widget(main_window, step_key, widget_attr) p = _read_widget_path(widget) if p: if not Path(p).is_absolute() and wd: p = str(Path(wd) / p).replace('\\', '/') return p # 兜底：本地子目录字典（与 pipeline._ensure_step_dir_map 一致） key = fallback_key or step_key sub = _FALLBACK_DIR_TABLE.get(key) if sub and wd: return str(Path(wd) / sub).replace('\\', '/') return wd def resolve_subdir(work_dir, subdir_key: str) -> str: """纯子目录拼装：把 pipeline key 解析为 work_dir 下的子目录路径。 用法：resolve_subdir(self.work_dir, 'visualization') → '/14_visualization' 与 pipeline.get_step_output_dir 同源（都查同一份 _FALLBACK_DIR_TABLE 子集）。 """ wd = str(work_dir) if work_dir else "" sub = _FALLBACK_DIR_TABLE.get(subdir_key) if sub and wd: return str(Path(wd) / sub).replace('\\', '/') return wd # ═══════════════════════════════════════════════════════════════ # 文件系统扫描表 —— 定义每个上游产出类型对应的搜索策略 # ═══════════════════════════════════════════════════════════════ _SCAN_TABLE = { # output_type → (subdir_key, extensions, file_name_matcher) 'water_mask': ('water_mask', ['.dat', '.tif', '.tiff'], None), 'glint_mask': ('glint_detection', ['.dat'], 'severe_glint'), 'deglint_image': ('deglint', ['.bsq', '.dat', '.tif', '.tiff'], None), 'sampling_points': ('sampling', ['.csv'], 'sampling_spectra'), 'processed_data': ('data_cleaning', ['.csv'], 'processed_data'), 'training_spectra': ('spectral_feature',['.csv'], 'training_spectra'), 'training_spectra_indices': ('indices', ['.csv'], 'training_spectra_indices'), 'ml_models_dir': ('supervised_models', None, None), # 目录 'ml_predictions_dir': ('ml_prediction', None, None), # 目录 'water_index_csv_dir': ('watercolor', None, None), # 目录 'visualization_dir': ('visualization', None, None), # 目录 'reference_img': ('water_mask', ['.bsq', '.dat', '.tif', '.tiff'], None), } def scan_work_dir_for_input(work_dir: str, output_type: str): """基于文件系统扫描查找上游步骤的产出文件。 这是 update_from_config 重构的核心工具函数。 仅信任硬盘上真实存在的文件，彻底消除面板间 UI 控件互相读取。 Args: work_dir: 工作目录路径 output_type: 产出类型，如 'water_mask', 'deglint_image', 'sampling_points' Returns: 找到的文件/目录绝对路径字符串，未找到返回 None """ if not work_dir: return None wd = Path(work_dir) entry = _SCAN_TABLE.get(output_type) if entry is None: return None subdir_key, extensions, matcher = entry target_dir = wd / _FALLBACK_DIR_TABLE.get(subdir_key, subdir_key) if not target_dir.is_dir(): return None # 目录类型（extensions 为 None）→ 只要目录存在就返回 if extensions is None: return str(target_dir).replace('\\', '/') # 扫描目录中的文件 candidates = [] for ext in extensions: for f in target_dir.glob(f'*{ext}'): if not f.is_file(): continue candidates.append(f) # 也搜 ext 的大写变体 for f in target_dir.glob(f'*{ext.upper()}'): if not f.is_file(): continue candidates.append(f) if not candidates: return None # 按 matcher 优先级 + mtime 排序 if matcher: matched = [f for f in candidates if matcher.lower() in f.stem.lower()] if matched: matched.sort(key=lambda p: p.stat().st_mtime, reverse=True) return str(matched[0]).replace('\\', '/') # matcher 未命中时，仍返回最新文件（宽松匹配） candidates.sort(key=lambda p: p.stat().st_mtime, reverse=True) return str(candidates[0]).replace('\\', '/') # 无 matcher → 返回最新的 candidates.sort(key=lambda p: p.stat().st_mtime, reverse=True) return str(candidates[0]).replace('\\', '/') __all__ = [ 'STEP_DATA_SOURCE', 'resolve_step_widget', 'get_step_output_path', 'resolve_subdir', 'scan_work_dir_for_input', ]