refactor: 实现纯壳主窗口 + 第一批 Manager（PanelFactory/PipelineExecutor/WorkspaceInitializer）
- water_quality_gui_v2.py: 纯壳主窗口（725行），依赖注入链 + EventBus 驱动，7个菜单连线 - panel_factory.py: PanelFactory 懒加载工厂（占位页替换 + 邻接预加载） - pipeline_executor.py: PipelineExecutor 核心调度（接管 run_full_pipeline/run_single_step/stop_pipeline） - workspace_initializer.py: WorkspaceInitializer 环境初始化（接管 init_workspace/set_work_directory/auto_populate_all）
This commit is contained in:
204
src/gui/core/panel_factory.py
Normal file
204
src/gui/core/panel_factory.py
Normal file
@ -0,0 +1,204 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
面板注册与装载工厂
|
||||
|
||||
按需懒加载步骤面板，替代 create_content_area 中一次性全量 new 14 个面板的做法。
|
||||
主窗口只需持有 PanelFactory 实例，通过 factory.create_tab_widget() 获取
|
||||
已挂载占位页的 QTabWidget，面板在用户首次切换到对应 Tab 时才实例化。
|
||||
|
||||
特性：
|
||||
- 懒加载：仅在 tab 首次激活时创建面板实例
|
||||
- 邻接预加载：切换 tab 时自动预加载左右邻居（可配置预加载窗口大小）
|
||||
- 注册表驱动：完全依赖 PANEL_REGISTRY，零硬编码
|
||||
- 事件总线自动接线：面板创建后自动调用 subscribe_panel_to_dependencies
|
||||
- 占位页：未加载的 tab 显示空白 QWidget，加载后原地替换为 QScrollArea(panel)
|
||||
"""
|
||||
|
||||
from PyQt5.QtWidgets import QWidget, QTabWidget, QScrollArea
|
||||
from PyQt5.QtCore import Qt
|
||||
|
||||
from src.gui.core.panel_registry import PANEL_REGISTRY
|
||||
from src.gui.core.dependency_subscriber import subscribe_panel_to_dependencies
|
||||
|
||||
|
||||
class PanelFactory:
|
||||
"""面板注册与装载工厂。
|
||||
|
||||
用法::
|
||||
|
||||
factory = PanelFactory(registry=PANEL_REGISTRY, main_window=self)
|
||||
tab_widget = factory.create_tab_widget(icons_dir="data/icons")
|
||||
# tab_widget 已包含所有占位页，可直接加入主窗口布局
|
||||
# 后续通过 factory.get_panel(step_id) 按需获取面板实例
|
||||
"""
|
||||
|
||||
def __init__(self, registry, main_window, preload_window=1):
|
||||
"""
|
||||
Args:
|
||||
registry: PANEL_REGISTRY 列表
|
||||
main_window: WaterQualityGUI 实例（用于注入 main_window 依赖）
|
||||
preload_window: 邻接预加载窗口大小。0=仅加载当前 tab；
|
||||
1=当前+左右各1个；-1=全量预加载（退化为旧行为）
|
||||
"""
|
||||
self._registry = registry
|
||||
self._main_window = main_window
|
||||
self._preload_window = preload_window
|
||||
|
||||
# step_id → panel 实例（仅已加载的）
|
||||
self._panels = {}
|
||||
# tab_index → 是否已加载
|
||||
self._loaded = set()
|
||||
# tab_index → placeholder QWidget（加载后被替换）
|
||||
self._placeholders = {}
|
||||
# 对外的 QTabWidget 引用
|
||||
self._tab_widget = None
|
||||
|
||||
# ── 公开 API ──────────────────────────────────────────────
|
||||
|
||||
def create_tab_widget(self, icons_dir="data/icons"):
|
||||
"""创建并返回已填充占位页的 QTabWidget。
|
||||
|
||||
每个 tab 初始为空白 QWidget 占位，面板在首次激活时懒加载。
|
||||
同时连接 currentChanged 信号驱动懒加载 + 邻接预加载。
|
||||
|
||||
Args:
|
||||
icons_dir: 图标目录名（相对于项目根），用于 get_resource_path
|
||||
|
||||
Returns:
|
||||
QTabWidget: 已添加所有占位 tab 的标签页控件
|
||||
"""
|
||||
from src.gui.water_quality_gui import get_resource_path
|
||||
from PyQt5.QtGui import QIcon
|
||||
|
||||
self._tab_widget = QTabWidget()
|
||||
self._tab_widget.setTabPosition(QTabWidget.North)
|
||||
self._tab_widget.setTabsClosable(False)
|
||||
|
||||
for idx, entry in enumerate(self._registry):
|
||||
step_id = entry['step_id']
|
||||
title = entry['title']
|
||||
icon_name = entry['icon']
|
||||
|
||||
# 创建占位页
|
||||
placeholder = QWidget()
|
||||
self._placeholders[idx] = placeholder
|
||||
|
||||
icon_path = get_resource_path(f"{icons_dir}/{icon_name}")
|
||||
self._tab_widget.addTab(placeholder, QIcon(icon_path), title)
|
||||
|
||||
# 连接切换信号 → 懒加载
|
||||
self._tab_widget.currentChanged.connect(self._on_tab_changed)
|
||||
|
||||
# 立即预加载首个 tab
|
||||
if self._registry:
|
||||
self._ensure_loaded(0)
|
||||
|
||||
return self._tab_widget
|
||||
|
||||
def get_panel(self, step_id):
|
||||
"""获取面板实例（若未加载则触发懒加载）。
|
||||
|
||||
Args:
|
||||
step_id: 步骤 ID，如 'step1'、'step5_clean'
|
||||
|
||||
Returns:
|
||||
QWidget 或 None: 面板实例，未找到则返回 None
|
||||
"""
|
||||
tab_index = self._step_id_to_tab_index(step_id)
|
||||
if tab_index < 0:
|
||||
return None
|
||||
self._ensure_loaded(tab_index)
|
||||
return self._panels.get(step_id)
|
||||
|
||||
def get_loaded_panels(self):
|
||||
"""返回所有已加载的面板字典 {step_id: panel}。"""
|
||||
return dict(self._panels)
|
||||
|
||||
def preload_all(self):
|
||||
"""强制加载所有面板（用于配置保存等需要遍历全部面板的场景）。"""
|
||||
for idx in range(len(self._registry)):
|
||||
self._ensure_loaded(idx)
|
||||
|
||||
def get_tab_widget(self):
|
||||
"""返回内部 QTabWidget 引用。"""
|
||||
return self._tab_widget
|
||||
|
||||
# ── 内部方法 ──────────────────────────────────────────────
|
||||
|
||||
def _on_tab_changed(self, index):
|
||||
"""Tab 切换时：加载当前 tab + 邻接预加载。"""
|
||||
if index < 0:
|
||||
return
|
||||
self._ensure_loaded(index)
|
||||
self._preload_neighbors(index)
|
||||
|
||||
def _ensure_loaded(self, tab_index):
|
||||
"""确保指定 tab 已加载；若未加载则实例化面板并替换占位页。"""
|
||||
if tab_index in self._loaded:
|
||||
return
|
||||
if tab_index < 0 or tab_index >= len(self._registry):
|
||||
return
|
||||
|
||||
entry = self._registry[tab_index]
|
||||
step_id = entry['step_id']
|
||||
cls = entry['class_ref']
|
||||
title = entry['title']
|
||||
kwargs = entry.get('constructor_kwargs')
|
||||
deps = entry.get('dependencies')
|
||||
|
||||
# 解析构造参数
|
||||
resolved_kwargs = {}
|
||||
if kwargs:
|
||||
for k in kwargs:
|
||||
if k == 'main_window':
|
||||
resolved_kwargs[k] = self._main_window
|
||||
|
||||
# 实例化面板
|
||||
panel = cls(**resolved_kwargs)
|
||||
|
||||
# 包裹到 QScrollArea
|
||||
scroll = QScrollArea()
|
||||
scroll.setWidget(panel)
|
||||
scroll.setWidgetResizable(True)
|
||||
|
||||
# 替换占位页
|
||||
placeholder = self._placeholders.get(tab_index)
|
||||
if placeholder is not None and self._tab_widget is not None:
|
||||
tab_title = self._tab_widget.tabText(tab_index)
|
||||
tab_icon = self._tab_widget.tabIcon(tab_index)
|
||||
self._tab_widget.removeTab(tab_index)
|
||||
self._tab_widget.insertTab(tab_index, scroll, tab_icon, tab_title)
|
||||
self._tab_widget.setCurrentIndex(tab_index)
|
||||
|
||||
# 注册
|
||||
self._panels[step_id] = panel
|
||||
self._loaded.add(tab_index)
|
||||
|
||||
# 事件总线自动接线
|
||||
if deps:
|
||||
subscribe_panel_to_dependencies(panel, step_id, deps)
|
||||
|
||||
def _preload_neighbors(self, index):
|
||||
"""预加载当前 tab 的邻居（根据 preload_window 配置）。"""
|
||||
if self._preload_window < 0:
|
||||
# 全量预加载
|
||||
for i in range(len(self._registry)):
|
||||
self._ensure_loaded(i)
|
||||
return
|
||||
|
||||
if self._preload_window == 0:
|
||||
return
|
||||
|
||||
start = max(0, index - self._preload_window)
|
||||
end = min(len(self._registry), index + self._preload_window + 1)
|
||||
for i in range(start, end):
|
||||
if i != index:
|
||||
self._ensure_loaded(i)
|
||||
|
||||
def _step_id_to_tab_index(self, step_id):
|
||||
"""step_id → tab_index 映射。"""
|
||||
for i, entry in enumerate(self._registry):
|
||||
if entry['step_id'] == step_id:
|
||||
return i
|
||||
return -1
|
||||
469
src/gui/core/pipeline_executor.py
Normal file
469
src/gui/core/pipeline_executor.py
Normal file
@ -0,0 +1,469 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Pipeline 执行器
|
||||
|
||||
接管 WaterQualityGUI 中所有 Pipeline 执行相关逻辑：
|
||||
- run_full_pipeline() 完整流程执行
|
||||
- run_single_step() 单步执行
|
||||
- stop_pipeline() 停止执行
|
||||
- _precheck_step3_bands() step3 波段越界预检
|
||||
|
||||
关键设计原则：
|
||||
- 所有状态变化通过 global_event_bus 发布事件，绝不直接操作 UI 控件
|
||||
- WorkerThread 的 Qt 信号连接到内部槽函数，槽函数仅做 EventBus 转发
|
||||
- 预检对话框（PreflightDialog / PipelineModeDialog / BandConfirmDialog）
|
||||
仍为模态弹窗（用户交互必需），但结果通过 EventBus 发布
|
||||
|
||||
发布的事件：
|
||||
PipelineStarted → {} 主窗口订阅：禁用运行按钮
|
||||
PipelineFinished → {success, message} 主窗口订阅：恢复按钮 + 弹窗
|
||||
PipelineStopped → {} 主窗口订阅：恢复按钮
|
||||
StepCompleted → {step_name, success, message} WorkspaceInitializer 订阅：扫描产物
|
||||
LogMessage → {message, level} LogManager 订阅：写入日志区
|
||||
ProgressUpdate → {percentage, message} LogManager 订阅：更新进度条
|
||||
"""
|
||||
|
||||
import os
|
||||
import copy
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from PyQt5.QtCore import QObject, Qt
|
||||
from PyQt5.QtWidgets import QMessageBox, QDialog
|
||||
|
||||
from src.gui.core.event_bus import global_event_bus
|
||||
from src.gui.core.worker_thread import (
|
||||
WorkerThread,
|
||||
PIPELINE_AVAILABLE,
|
||||
)
|
||||
from src.gui.core.preflight_dialog import PreflightDialog
|
||||
from src.gui.core.pipeline_mode_dialog import PipelineModeDialog
|
||||
from src.gui.dialogs import BandConfirmDialog
|
||||
from src.core.pipeline.runner import PipelineHalt
|
||||
|
||||
# pipeline step_id → panel step_id 映射
|
||||
PIPELINE_TO_PANEL_STEP = {
|
||||
'step1': 'step1',
|
||||
'step2': 'step2',
|
||||
'step3': 'step3',
|
||||
'step4': 'step5_clean',
|
||||
'step5': 'step6_feature',
|
||||
'step7': 'step7_index',
|
||||
'step8': 'step8_ml_train',
|
||||
'step9': 'step10_watercolor',
|
||||
'step10': 'step4_sampling',
|
||||
'step11_ml': 'step9_ml_predict',
|
||||
'step11': 'step11_map',
|
||||
'step14': 'step11_map',
|
||||
}
|
||||
|
||||
|
||||
class PipelineExecutor(QObject):
|
||||
"""Pipeline 执行器 —— 纯逻辑层，零 UI 直接操作。"""
|
||||
|
||||
def __init__(self, panel_factory, workspace_initializer, parent=None):
|
||||
"""
|
||||
Args:
|
||||
panel_factory: PanelFactory 实例（用于获取面板和配置）
|
||||
workspace_initializer: WorkspaceInitializer 实例（用于获取 work_dir）
|
||||
parent: 父 QObject（通常为 WaterQualityGUI）
|
||||
"""
|
||||
super().__init__(parent)
|
||||
self._panel_factory = panel_factory
|
||||
self._workspace_initializer = workspace_initializer
|
||||
self._worker: Optional[WorkerThread] = None
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# 公开 API
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
|
||||
@property
|
||||
def worker(self):
|
||||
return self._worker
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
return self._worker is not None and self._worker.isRunning()
|
||||
|
||||
def run_full_pipeline(self):
|
||||
"""运行完整流程。
|
||||
|
||||
流程：
|
||||
1. 检查 PIPELINE_AVAILABLE
|
||||
2. 获取 work_dir（从 WorkspaceInitializer）
|
||||
3. 扫描工作目录 + 自动回填
|
||||
4. step3 波段越界预检
|
||||
5. 全流程模式选择弹窗
|
||||
6. 获取配置 + 模式裁剪
|
||||
7. 一次性全预检 + 用户交互
|
||||
8. 确认执行 → 创建 WorkerThread → 启动
|
||||
"""
|
||||
if not PIPELINE_AVAILABLE:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '无法导入 Pipeline 模块，请检查项目文件结构！',
|
||||
'level': 'error',
|
||||
})
|
||||
# 阻断性错误仍需弹窗（用户必须知道）
|
||||
QMessageBox.critical(
|
||||
self.parent(), "错误",
|
||||
"无法导入pipeline模块，请确保water_quality_inversion_pipeline_GUI.py文件存在！"
|
||||
)
|
||||
return
|
||||
|
||||
# ── 1) 获取 work_dir ──
|
||||
work_dir = self._workspace_initializer.work_dir
|
||||
if not work_dir:
|
||||
QMessageBox.warning(self.parent(), "警告", "未选择工作目录，请先设置工作目录。")
|
||||
return
|
||||
|
||||
work_path = Path(work_dir)
|
||||
|
||||
# ── 2) 运行前扫描 + 自动回填 ──
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '正在进行运行前环境预检与自动扫描...',
|
||||
'level': 'info',
|
||||
})
|
||||
self._workspace_initializer.auto_populate_all()
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '✓ 预检完成：已扫描工作目录并自动回填已落盘的产物',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ── 3) step3 波段越界预检 ──
|
||||
if not self._precheck_step3_bands():
|
||||
return
|
||||
|
||||
# ── 4) 全流程模式选择弹窗 ──
|
||||
mode_dlg = PipelineModeDialog(main_window=self.parent(), parent=self.parent())
|
||||
if mode_dlg.exec() != QDialog.Accepted:
|
||||
return
|
||||
selected_mode = mode_dlg.selected_mode
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': (
|
||||
f"[模式选择] 选定模式: "
|
||||
f"{'训练新模型' if selected_mode == 'training' else '使用已有模型直接预测'}"
|
||||
),
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ── 5) 获取配置 ──
|
||||
config = self._get_current_config()
|
||||
|
||||
# ── 6) 模式裁剪 ──
|
||||
if selected_mode == "prediction_only":
|
||||
from src.core.workspace_manager import WorkspaceManager
|
||||
config = WorkspaceManager.prune_config_for_prediction_mode(config)
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '[模式选择] 已裁剪训练相关步骤（step4/5/7/8），进入仅预测模式',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ── 7) 一次性全预检 + 用户交互式决策 ──
|
||||
missing_items = PreflightDialog.build_missing_items(config)
|
||||
skip_list: List[str] = []
|
||||
|
||||
if missing_items:
|
||||
critical_items = [it for it in missing_items if it.is_critical]
|
||||
if critical_items:
|
||||
lines = "\n".join(f" - [{it.step_name}] {it.reason}" for it in critical_items)
|
||||
QMessageBox.critical(
|
||||
self.parent(), "预检失败（阻断性错误）",
|
||||
f"以下为阻断性缺失，流程无法启动：\n\n{lines}\n\n请填写后重新运行。"
|
||||
)
|
||||
return
|
||||
|
||||
dialog = PreflightDialog(missing_items, parent=self.parent())
|
||||
if dialog.exec() != QDialog.Accepted:
|
||||
return
|
||||
result = dialog.get_result()
|
||||
if result is None:
|
||||
return
|
||||
|
||||
action, *payload = result
|
||||
if action == "fill":
|
||||
_, step_id, tab_index = result
|
||||
# 发布事件：请求切换到指定 tab
|
||||
global_event_bus.publish('NavigateToTab', {
|
||||
'tab_index': tab_index,
|
||||
'step_id': step_id,
|
||||
})
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[预检] 用户选择填写 {step_id}，已切换到对应面板。',
|
||||
'level': 'info',
|
||||
})
|
||||
return
|
||||
skip_list = payload[0] if payload else []
|
||||
if skip_list:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[预检] 用户强制跳过 {len(skip_list)} 个步骤: {skip_list}',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ── 8) 确认执行 ──
|
||||
reply = QMessageBox.question(
|
||||
self.parent(), "确认",
|
||||
"是否开始执行完整流程？\n\n这可能需要较长时间，请确保配置正确。",
|
||||
QMessageBox.Yes | QMessageBox.No
|
||||
)
|
||||
if reply != QMessageBox.Yes:
|
||||
return
|
||||
|
||||
# ── 9) 准备 worker_config ──
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'初始化 Pipeline，工作目录: {work_dir}',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
worker_config = copy.deepcopy(config)
|
||||
step6_cfg = worker_config.get('step6_feature')
|
||||
if step6_cfg:
|
||||
enabled = step6_cfg.pop('enabled', True)
|
||||
if not enabled:
|
||||
worker_config.pop('step6_feature', None)
|
||||
|
||||
# ── 10) 创建 WorkerThread 并连线 ──
|
||||
self._worker = WorkerThread(work_dir, worker_config, mode='full', skip_list=skip_list)
|
||||
self._worker.log_message.connect(self._on_log_message, Qt.QueuedConnection)
|
||||
self._worker.progress_update.connect(self._on_progress_update, Qt.QueuedConnection)
|
||||
self._worker.step_completed.connect(self._on_step_completed, Qt.QueuedConnection)
|
||||
self._worker.finished.connect(self._on_finished, Qt.QueuedConnection)
|
||||
|
||||
# ── 11) 发布启动事件 → 主窗口订阅后禁用按钮 ──
|
||||
global_event_bus.publish('PipelineStarted', {})
|
||||
global_event_bus.publish('ProgressUpdate', {'percentage': 0, 'message': '准备执行...'})
|
||||
global_event_bus.publish('LogMessage', {'message': '=' * 50, 'level': 'info'})
|
||||
global_event_bus.publish('LogMessage', {'message': '开始执行完整流程...', 'level': 'info'})
|
||||
global_event_bus.publish('LogMessage', {'message': '=' * 50, 'level': 'info'})
|
||||
|
||||
self._worker.start()
|
||||
|
||||
def run_single_step(self, step_name: str, config: dict = None):
|
||||
"""运行单个步骤。
|
||||
|
||||
Args:
|
||||
step_name: 步骤名称（如 'step1', 'step5_clean'）
|
||||
config: 步骤配置字典（可选，默认从面板获取）
|
||||
"""
|
||||
if not PIPELINE_AVAILABLE:
|
||||
QMessageBox.critical(
|
||||
self.parent(), "错误",
|
||||
"无法导入pipeline模块，请确保water_quality_inversion_pipeline_GUI.py文件存在！"
|
||||
)
|
||||
return
|
||||
|
||||
work_dir = self._workspace_initializer.work_dir or './work_dir'
|
||||
|
||||
if config is None:
|
||||
config = self._get_current_config()
|
||||
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'初始化 Pipeline，工作目录: {work_dir}',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
self._worker = WorkerThread(work_dir, config, mode='single_step', step_name=step_name)
|
||||
self._worker.log_message.connect(self._on_log_message, Qt.QueuedConnection)
|
||||
self._worker.progress_update.connect(self._on_progress_update, Qt.QueuedConnection)
|
||||
self._worker.step_completed.connect(self._on_step_completed, Qt.QueuedConnection)
|
||||
self._worker.finished.connect(self._on_finished, Qt.QueuedConnection)
|
||||
|
||||
global_event_bus.publish('PipelineStarted', {})
|
||||
global_event_bus.publish('ProgressUpdate', {'percentage': 0, 'message': f'准备执行 {step_name}...'})
|
||||
global_event_bus.publish('LogMessage', {'message': '=' * 50, 'level': 'info'})
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'开始独立运行步骤 {step_name}...',
|
||||
'level': 'info',
|
||||
})
|
||||
global_event_bus.publish('LogMessage', {'message': '=' * 50, 'level': 'info'})
|
||||
|
||||
self._worker.start()
|
||||
|
||||
def stop_pipeline(self):
|
||||
"""停止当前执行的流程。"""
|
||||
if self._worker and self._worker.isRunning():
|
||||
reply = QMessageBox.question(
|
||||
self.parent(), "确认",
|
||||
"是否停止当前流程？",
|
||||
QMessageBox.Yes | QMessageBox.No
|
||||
)
|
||||
if reply == QMessageBox.Yes:
|
||||
self._worker.stop()
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '用户取消执行',
|
||||
'level': 'warning',
|
||||
})
|
||||
global_event_bus.publish('PipelineStopped', {})
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# WorkerThread 信号 → EventBus 事件（纯转发，零 UI 操作）
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
|
||||
def _on_log_message(self, message: str, level: str):
|
||||
"""WorkerThread 日志 → EventBus LogMessage 事件。"""
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': message,
|
||||
'level': level,
|
||||
})
|
||||
|
||||
def _on_progress_update(self, percentage: int, message: str):
|
||||
"""WorkerThread 进度 → EventBus ProgressUpdate 事件。"""
|
||||
global_event_bus.publish('ProgressUpdate', {
|
||||
'percentage': percentage,
|
||||
'message': message,
|
||||
})
|
||||
|
||||
def _on_step_completed(self, step_name: str, success: bool, message: str):
|
||||
"""WorkerThread 步骤完成 → EventBus StepCompleted 事件。
|
||||
|
||||
WorkspaceInitializer 订阅此事件，自动扫描产物并发布 OutputUpdated。
|
||||
"""
|
||||
global_event_bus.publish('StepCompleted', {
|
||||
'step_name': step_name,
|
||||
'success': success,
|
||||
'message': message,
|
||||
})
|
||||
|
||||
def _on_finished(self, success: bool, message: str):
|
||||
"""WorkerThread 完成 → EventBus PipelineFinished 事件。
|
||||
|
||||
主窗口订阅此事件，恢复按钮状态并弹窗。
|
||||
"""
|
||||
global_event_bus.publish('PipelineFinished', {
|
||||
'success': success,
|
||||
'message': message,
|
||||
})
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# 内部辅助
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
|
||||
def _get_current_config(self) -> dict:
|
||||
"""从所有已加载面板收集配置。
|
||||
|
||||
注意：仅收集已加载面板（懒加载模式下可能不全）。
|
||||
如需全量配置，调用方应先执行 panel_factory.preload_all()。
|
||||
"""
|
||||
config = {}
|
||||
for step_id, panel in self._panel_factory.get_loaded_panels().items():
|
||||
if hasattr(panel, 'get_config'):
|
||||
config[step_id] = panel.get_config()
|
||||
return config
|
||||
|
||||
def _precheck_step3_bands(self) -> bool:
|
||||
"""步骤 3 波段越界预检（主线程同步执行，避免多线程弹窗问题）。
|
||||
|
||||
读取 step1 影像的 RasterCount，校验 step3 面板当前方法下所有波段索引
|
||||
是否越界。若越界，弹 BandConfirmDialog（60s 倒计时）让用户调整或取消。
|
||||
|
||||
Returns:
|
||||
True: 预检通过或已自动调整，继续执行
|
||||
False: 用户点"取消运行"，应中止
|
||||
"""
|
||||
try:
|
||||
step1_panel = self._panel_factory.get_panel('step1')
|
||||
step3_panel = self._panel_factory.get_panel('step3')
|
||||
img_path = step1_panel.img_file.get_path() if step1_panel else None
|
||||
step3_cfg = step3_panel.get_config() if step3_panel else None
|
||||
step3_enabled = step3_panel.enable_checkbox.isChecked() if step3_panel else False
|
||||
except Exception as e:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'⚠ step3 波段预检：读取面板状态失败 - {e}',
|
||||
'level': 'warning',
|
||||
})
|
||||
return True
|
||||
|
||||
if not step3_enabled:
|
||||
return True
|
||||
if not img_path or not os.path.isfile(img_path):
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '⚠ step3 波段预检：未找到参考影像，跳过',
|
||||
'level': 'info',
|
||||
})
|
||||
return True
|
||||
if not step3_cfg:
|
||||
return True
|
||||
|
||||
try:
|
||||
from osgeo import gdal
|
||||
dataset = gdal.Open(img_path)
|
||||
if dataset is None:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'⚠ step3 波段预检：gdal 无法打开影像 {img_path}',
|
||||
'level': 'warning',
|
||||
})
|
||||
return True
|
||||
max_band = dataset.RasterCount
|
||||
dataset = None
|
||||
except Exception as e:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'⚠ step3 波段预检：读取 RasterCount 失败 - {e}',
|
||||
'level': 'warning',
|
||||
})
|
||||
return True
|
||||
|
||||
if max_band <= 0:
|
||||
return True
|
||||
|
||||
method = step3_cfg.get('method', 'goodman')
|
||||
if method == 'goodman':
|
||||
band_fields = [
|
||||
('nir_lower', 'nir_lower', 65, 'NIR下波段'),
|
||||
('nir_upper', 'nir_upper', 91, 'NIR上波段'),
|
||||
]
|
||||
elif method == 'kutser':
|
||||
band_fields = [
|
||||
('oxy_band', 'oxy_band', 38, '氧吸收波段'),
|
||||
('lower_oxy', 'lower_oxy', 36, '下氧吸收波段'),
|
||||
('upper_oxy', 'upper_oxy', 49, '上氧吸收波段'),
|
||||
('nir_band', 'nir_band', 47, 'NIR波段'),
|
||||
]
|
||||
elif method == 'hedley':
|
||||
band_fields = [
|
||||
('hedley_nir_band', 'hedley_nir_band', 47, 'NIR波段'),
|
||||
]
|
||||
else:
|
||||
return True
|
||||
|
||||
for cfg_key, panel_attr, recommended, label in band_fields:
|
||||
requested = step3_cfg.get(cfg_key)
|
||||
if requested is None or requested <= max_band:
|
||||
continue
|
||||
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'⚠ step3 波段越界：{label}={requested} > 影像波段数 {max_band}',
|
||||
'level': 'warning',
|
||||
})
|
||||
|
||||
dlg = BandConfirmDialog(
|
||||
self.parent(),
|
||||
requested_band=requested,
|
||||
max_band=max_band,
|
||||
recommended_band=recommended,
|
||||
method_label=label,
|
||||
)
|
||||
result = dlg.exec_()
|
||||
if result == QDialog.Rejected:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '✗ 用户取消运行（step3 波段越界未解决）',
|
||||
'level': 'warning',
|
||||
})
|
||||
return False
|
||||
|
||||
new_band = dlg.selected_band()
|
||||
try:
|
||||
spin = getattr(step3_panel, panel_attr)
|
||||
spin.setValue(new_band)
|
||||
except AttributeError:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'⚠ step3 panel 缺控件 {panel_attr}，跳过回写',
|
||||
'level': 'warning',
|
||||
})
|
||||
continue
|
||||
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'✓ {label}：{requested} → {new_band}（影像最多 {max_band} 波段）',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
return True
|
||||
234
src/gui/core/workspace_initializer.py
Normal file
234
src/gui/core/workspace_initializer.py
Normal file
@ -0,0 +1,234 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
工作空间初始化器
|
||||
|
||||
接管 WaterQualityGUI 中所有工作目录相关逻辑：
|
||||
- init_workspace() 启动时工作目录选择对话框
|
||||
- set_work_directory() 手动设置工作目录
|
||||
- open_work_directory() 在资源管理器中打开工作目录
|
||||
- auto_populate_all() 扫描工作目录并自动填充所有步骤输入路径
|
||||
- _auto_fill_output_paths() step1 输出路径自动回填
|
||||
|
||||
关键设计原则：
|
||||
- 配合 WorkspaceManager 进行目录扫描与产物发现
|
||||
- WorkspaceManager 扫描完成后自动发布 OutputUpdated 事件
|
||||
- 工作目录变更时发布 WorkspaceChanged 事件
|
||||
- 订阅 StepCompleted 事件，自动触发产物扫描
|
||||
|
||||
发布的事件：
|
||||
WorkspaceChanged → {work_dir} 主窗口订阅：更新状态栏
|
||||
|
||||
订阅的事件：
|
||||
StepCompleted → 自动调用 WorkspaceManager.update_step_outputs()
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from PyQt5.QtCore import QObject
|
||||
from PyQt5.QtWidgets import QMessageBox, QFileDialog
|
||||
|
||||
from src.gui.core.event_bus import global_event_bus
|
||||
from src.core.workspace_manager import WorkspaceManager
|
||||
|
||||
# pipeline step_id → panel step_id 映射
|
||||
PIPELINE_TO_PANEL_STEP = {
|
||||
'step1': 'step1',
|
||||
'step2': 'step2',
|
||||
'step3': 'step3',
|
||||
'step4': 'step5_clean',
|
||||
'step5': 'step6_feature',
|
||||
'step7': 'step7_index',
|
||||
'step8': 'step8_ml_train',
|
||||
'step9': 'step10_watercolor',
|
||||
'step10': 'step4_sampling',
|
||||
'step11_ml': 'step9_ml_predict',
|
||||
'step11': 'step11_map',
|
||||
'step14': 'step11_map',
|
||||
}
|
||||
|
||||
|
||||
class WorkspaceInitializer(QObject):
|
||||
"""工作空间初始化器 —— 纯逻辑层，零 UI 直接操作。"""
|
||||
|
||||
def __init__(self, panel_factory, parent=None):
|
||||
"""
|
||||
Args:
|
||||
panel_factory: PanelFactory 实例（用于获取面板和注入 work_dir）
|
||||
parent: 父 QObject（通常为 WaterQualityGUI）
|
||||
"""
|
||||
super().__init__(parent)
|
||||
self._panel_factory = panel_factory
|
||||
self._work_dir: Optional[str] = None
|
||||
|
||||
# 工作空间管理器（文件扫描、路径发现）
|
||||
self._workspace_manager = WorkspaceManager()
|
||||
self._workspace_manager.set_step_id_mapping(PIPELINE_TO_PANEL_STEP)
|
||||
|
||||
# 订阅 StepCompleted 事件 → 自动扫描产物
|
||||
global_event_bus.subscribe('StepCompleted', self._on_step_completed)
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# 公开 API
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
|
||||
@property
|
||||
def work_dir(self) -> Optional[str]:
|
||||
return self._work_dir
|
||||
|
||||
@work_dir.setter
|
||||
def work_dir(self, value: str):
|
||||
if value and value != self._work_dir:
|
||||
self._work_dir = value
|
||||
global_event_bus.publish('WorkspaceChanged', {'work_dir': value})
|
||||
|
||||
@property
|
||||
def workspace_manager(self) -> WorkspaceManager:
|
||||
return self._workspace_manager
|
||||
|
||||
def run(self):
|
||||
"""启动时工作目录选择对话框（由 QTimer.singleShot 延迟调用）。
|
||||
|
||||
若用户取消或关闭对话框，程序退出。
|
||||
"""
|
||||
msg_box = QMessageBox()
|
||||
msg_box.setIcon(QMessageBox.Information)
|
||||
msg_box.setWindowTitle("选择工作目录")
|
||||
msg_box.setText("欢迎使用Mega Water！\n\n请选择工作目录来保存所有分析结果。")
|
||||
msg_box.setInformativeText(
|
||||
"工作目录将用于存储：\n"
|
||||
"• 水域掩膜文件\n"
|
||||
"• 耀斑检测结果\n"
|
||||
"• 模型训练数据\n"
|
||||
"• 预测结果与分布图\n\n"
|
||||
"点击'确定'选择目录"
|
||||
)
|
||||
msg_box.setStandardButtons(QMessageBox.Ok | QMessageBox.Cancel)
|
||||
msg_box.setDefaultButton(QMessageBox.Ok)
|
||||
|
||||
result = msg_box.exec_()
|
||||
if result == QMessageBox.Cancel:
|
||||
QMessageBox.warning(None, "取消操作", "未选择工作目录，程序将退出。")
|
||||
sys.exit(0)
|
||||
|
||||
work_dir = QFileDialog.getExistingDirectory(
|
||||
self.parent(), "选择工作目录", "",
|
||||
QFileDialog.ShowDirsOnly | QFileDialog.DontResolveSymlinks
|
||||
)
|
||||
|
||||
if not work_dir:
|
||||
QMessageBox.critical(self.parent(), "错误", "必须选择工作目录才能使用系统！\n程序即将退出。")
|
||||
sys.exit(0)
|
||||
|
||||
self.work_dir = work_dir
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'✓ 已选择工作目录: {self._work_dir}',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# 自动填充 step1 输出路径
|
||||
self._auto_fill_output_paths()
|
||||
|
||||
def set_work_directory(self):
|
||||
"""手动设置工作目录（菜单触发）。"""
|
||||
dir_path = QFileDialog.getExistingDirectory(self.parent(), "选择工作目录")
|
||||
if dir_path:
|
||||
self.work_dir = dir_path
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'工作目录已设置: {dir_path}',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# 同步到可视化/报告面板
|
||||
for sid in ('step12_viz', 'step13_report'):
|
||||
panel = self._panel_factory.get_panel(sid)
|
||||
if panel and hasattr(panel, 'set_work_dir'):
|
||||
panel.set_work_dir(dir_path)
|
||||
|
||||
def open_work_directory(self):
|
||||
"""在资源管理器中打开工作目录。"""
|
||||
work_dir = self._work_dir or './work_dir'
|
||||
if os.path.exists(work_dir):
|
||||
os.startfile(work_dir)
|
||||
else:
|
||||
QMessageBox.warning(self.parent(), "警告", "工作目录不存在！")
|
||||
|
||||
def auto_populate_all(self):
|
||||
"""扫描工作目录并触发事件总线自动填充所有步骤的输入路径。
|
||||
|
||||
流程：
|
||||
1. WorkspaceManager.scan_work_directory_for_files() 扫描磁盘
|
||||
2. WorkspaceManager 内部自动发布 OutputUpdated 事件
|
||||
3. 各面板通过 DependencySubscriber 自动接收并填充
|
||||
4. 补充发布 reference_img（step1 的输入影像）
|
||||
"""
|
||||
work_dir = self._work_dir
|
||||
if not work_dir:
|
||||
QMessageBox.warning(self.parent(), "警告", "未设置工作目录，请先设置。")
|
||||
return
|
||||
|
||||
work_path = Path(work_dir)
|
||||
if not work_path.exists():
|
||||
QMessageBox.warning(self.parent(), "警告", f"工作目录不存在: {work_dir}\n请先设置正确的工作目录。")
|
||||
return
|
||||
|
||||
# WorkspaceManager 扫描 → 内部自动发布 OutputUpdated 事件
|
||||
self._workspace_manager.scan_work_directory_for_files(work_path)
|
||||
|
||||
# 补充发布 reference_img（step1 的输入影像，非 step1 产出但下游依赖它）
|
||||
step1_panel = self._panel_factory.get_panel('step1')
|
||||
if step1_panel and hasattr(step1_panel, 'img_file'):
|
||||
ref_img = step1_panel.img_file.get_path()
|
||||
if ref_img:
|
||||
global_event_bus.publish('OutputUpdated', {
|
||||
'step_id': 'step1',
|
||||
'output_type': 'reference_img',
|
||||
'path': ref_img,
|
||||
})
|
||||
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '✓ 工作目录扫描完成，事件总线已通知所有面板自动填充',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# EventBus 订阅回调
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
|
||||
def _on_step_completed(self, data: dict):
|
||||
"""StepCompleted 事件回调：自动扫描产物并发布 OutputUpdated。
|
||||
|
||||
当 PipelineExecutor 发布 StepCompleted 事件时，
|
||||
此方法调用 WorkspaceManager.update_step_outputs() 扫描磁盘，
|
||||
WorkspaceManager 内部自动发布 OutputUpdated 事件驱动面板填充。
|
||||
"""
|
||||
if not data.get('success'):
|
||||
return
|
||||
|
||||
step_name = data.get('step_name', '')
|
||||
work_dir = self._work_dir or './work_dir'
|
||||
work_path = Path(work_dir)
|
||||
|
||||
if step_name not in self._workspace_manager.step_outputs:
|
||||
self._workspace_manager.step_outputs[step_name] = {}
|
||||
|
||||
self._workspace_manager.update_step_outputs(step_name, work_path)
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# 内部辅助
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
|
||||
def _auto_fill_output_paths(self):
|
||||
"""根据工作目录自动填充 step1 的输出路径。
|
||||
|
||||
注意：Step1 的输出路径由 update_work_directory() 根据模式自动控制。
|
||||
"""
|
||||
if not self._work_dir:
|
||||
return
|
||||
|
||||
step1_panel = self._panel_factory.get_panel('step1')
|
||||
if step1_panel:
|
||||
step1_panel.update_work_directory(self._work_dir)
|
||||
Reference in New Issue
Block a user