feat: Step1~Step14 面板单步按钮 EventBus 解耦 + Handler 补全(Step8~Step14)+ 旧上帝类删除
- 9 个面板(step1~step6/step8_ml_train/step8_qaa/step9_ml_predict/step10)单步执行按钮从 parent 链上溯改为 global_event_bus.publish('RequestRunSingleStep')
- PipelineExecutor 新增 _on_request_run_single_step 订阅
- 新增 Handler: step8_ml_train / step9_ml_predict / step10_qaa_inversion / step11_concentration / step12_kriging / step13_visualization / step14_report
- 删除旧 water_quality_inversion_pipeline_GUI.py(上帝类已肢解完毕)
This commit is contained in:
@ -26,6 +26,7 @@ Pipeline 执行器
|
||||
|
||||
import os
|
||||
import copy
|
||||
import traceback
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
@ -74,6 +75,9 @@ class PipelineExecutor(QObject):
|
||||
self._workspace_initializer = workspace_initializer
|
||||
self._worker: Optional[WorkerThread] = None
|
||||
|
||||
# 订阅面板发出的单步执行请求(解耦面板与执行器)
|
||||
global_event_bus.subscribe('RequestRunSingleStep', self._on_request_run_single_step)
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# 公开 API
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
@ -98,26 +102,60 @@ class PipelineExecutor(QObject):
|
||||
6. 获取配置 + 模式裁剪
|
||||
7. 一次性全预检 + 用户交互
|
||||
8. 确认执行 → 创建 WorkerThread → 启动
|
||||
|
||||
关键防静默失败设计:
|
||||
- 每一个 return 前必须通过 EventBus 发布 LogMessage
|
||||
- 整个方法体包裹在 try/except 中,防止 PyQt5 槽函数静默吞异常
|
||||
"""
|
||||
print("==== [探针] run_full_pipeline 方法体已进入 ====", flush=True)
|
||||
try:
|
||||
self._run_full_pipeline_impl()
|
||||
except Exception as e:
|
||||
err_detail = traceback.format_exc()
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[致命错误] run_full_pipeline 异常: {e}',
|
||||
'level': 'error',
|
||||
})
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'详细追踪:\n{err_detail}',
|
||||
'level': 'error',
|
||||
})
|
||||
QMessageBox.critical(
|
||||
self.parent(), "运行失败",
|
||||
f"启动流程时发生未预期的错误:\n\n{e}\n\n详细信息已输出到日志区。"
|
||||
)
|
||||
|
||||
def _run_full_pipeline_impl(self):
|
||||
"""run_full_pipeline 的实现体,由外层 try/except 保护。"""
|
||||
# ★ 终端即时反馈:确保即使 EventBus/日志区未就绪也能看到
|
||||
print("\n[PipelineExecutor] 收到「运行完整流程」指令,开始执行...")
|
||||
|
||||
if not PIPELINE_AVAILABLE:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '无法导入 Pipeline 模块,请检查项目文件结构!',
|
||||
'level': 'error',
|
||||
})
|
||||
# 阻断性错误仍需弹窗(用户必须知道)
|
||||
QMessageBox.critical(
|
||||
self.parent(), "错误",
|
||||
"无法导入pipeline模块,请确保water_quality_inversion_pipeline_GUI.py文件存在!"
|
||||
"无法导入 Pipeline 模块,请检查 src/core/handlers/ 目录是否完整!"
|
||||
)
|
||||
return
|
||||
|
||||
# ── 1) 获取 work_dir ──
|
||||
work_dir = self._workspace_initializer.work_dir
|
||||
if not work_dir:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '⚠ 未选择工作目录,流程中止。请先通过「工具 → 设置工作目录」选择工作目录。',
|
||||
'level': 'warning',
|
||||
})
|
||||
QMessageBox.warning(self.parent(), "警告", "未选择工作目录,请先设置工作目录。")
|
||||
return
|
||||
|
||||
work_path = Path(work_dir)
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[运行] 工作目录: {work_dir}',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ── 2) 运行前扫描 + 自动回填 ──
|
||||
global_event_bus.publish('LogMessage', {
|
||||
@ -132,11 +170,19 @@ class PipelineExecutor(QObject):
|
||||
|
||||
# ── 3) step3 波段越界预检 ──
|
||||
if not self._precheck_step3_bands():
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '⚠ 流程中止:step3 波段越界预检未通过(用户取消或波段配置无效)',
|
||||
'level': 'warning',
|
||||
})
|
||||
return
|
||||
|
||||
# ── 4) 全流程模式选择弹窗 ──
|
||||
mode_dlg = PipelineModeDialog(main_window=self.parent(), parent=self.parent())
|
||||
if mode_dlg.exec() != QDialog.Accepted:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '⚠ 流程中止:用户取消了模式选择对话框',
|
||||
'level': 'warning',
|
||||
})
|
||||
return
|
||||
selected_mode = mode_dlg.selected_mode
|
||||
global_event_bus.publish('LogMessage', {
|
||||
@ -147,8 +193,17 @@ class PipelineExecutor(QObject):
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ── 5) 获取配置 ──
|
||||
# ── 5) 获取配置(★ 先预加载所有面板,确保配置完整) ──
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '[运行] 正在收集所有步骤面板的配置...',
|
||||
'level': 'info',
|
||||
})
|
||||
self._panel_factory.preload_all()
|
||||
config = self._get_current_config()
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[运行] 已收集 {len(config)} 个步骤的配置: {list(config.keys())}',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ── 6) 模式裁剪 ──
|
||||
if selected_mode == "prediction_only":
|
||||
@ -164,9 +219,17 @@ class PipelineExecutor(QObject):
|
||||
skip_list: List[str] = []
|
||||
|
||||
if missing_items:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[预检] 发现 {len(missing_items)} 个缺失项,弹出预检对话框...',
|
||||
'level': 'warning',
|
||||
})
|
||||
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)
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[预检] 阻断性错误 ({len(critical_items)} 项):\n{lines}',
|
||||
'level': 'error',
|
||||
})
|
||||
QMessageBox.critical(
|
||||
self.parent(), "预检失败(阻断性错误)",
|
||||
f"以下为阻断性缺失,流程无法启动:\n\n{lines}\n\n请填写后重新运行。"
|
||||
@ -175,21 +238,28 @@ class PipelineExecutor(QObject):
|
||||
|
||||
dialog = PreflightDialog(missing_items, parent=self.parent())
|
||||
if dialog.exec() != QDialog.Accepted:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '⚠ 流程中止:用户取消了预检对话框',
|
||||
'level': 'warning',
|
||||
})
|
||||
return
|
||||
result = dialog.get_result()
|
||||
if result is None:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '⚠ 流程中止:预检对话框返回空结果',
|
||||
'level': 'warning',
|
||||
})
|
||||
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},已切换到对应面板。',
|
||||
'message': f'[预检] 用户选择填写 {step_id},已切换到对应面板。流程暂停,填写完成后请重新运行。',
|
||||
'level': 'info',
|
||||
})
|
||||
return
|
||||
@ -197,8 +267,13 @@ class PipelineExecutor(QObject):
|
||||
if skip_list:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[预检] 用户强制跳过 {len(skip_list)} 个步骤: {skip_list}',
|
||||
'level': 'info',
|
||||
'level': 'warning',
|
||||
})
|
||||
else:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '[预检] ✓ 所有必需项均已就绪,无需弹窗',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ── 8) 确认执行 ──
|
||||
reply = QMessageBox.question(
|
||||
@ -207,6 +282,10 @@ class PipelineExecutor(QObject):
|
||||
QMessageBox.Yes | QMessageBox.No
|
||||
)
|
||||
if reply != QMessageBox.Yes:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '⚠ 流程中止:用户取消了执行确认',
|
||||
'level': 'warning',
|
||||
})
|
||||
return
|
||||
|
||||
# ── 9) 准备 worker_config ──
|
||||
@ -222,6 +301,11 @@ class PipelineExecutor(QObject):
|
||||
if not enabled:
|
||||
worker_config.pop('step6_feature', None)
|
||||
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[运行] 最终执行配置包含 {len(worker_config)} 个步骤: {list(worker_config.keys())}',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
# ── 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)
|
||||
@ -245,17 +329,48 @@ class PipelineExecutor(QObject):
|
||||
step_name: 步骤名称(如 'step1', 'step5_clean')
|
||||
config: 步骤配置字典(可选,默认从面板获取)
|
||||
"""
|
||||
try:
|
||||
self._run_single_step_impl(step_name, config)
|
||||
except Exception as e:
|
||||
err_detail = traceback.format_exc()
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[致命错误] run_single_step 异常: {e}',
|
||||
'level': 'error',
|
||||
})
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'详细追踪:\n{err_detail}',
|
||||
'level': 'error',
|
||||
})
|
||||
QMessageBox.critical(
|
||||
self.parent(), "运行失败",
|
||||
f"启动单步执行时发生未预期的错误:\n\n{e}\n\n详细信息已输出到日志区。"
|
||||
)
|
||||
|
||||
def _run_single_step_impl(self, step_name: str, config: dict = None):
|
||||
if not PIPELINE_AVAILABLE:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '无法导入 Pipeline 模块,请检查 src/core/handlers/ 目录是否完整!',
|
||||
'level': 'error',
|
||||
})
|
||||
QMessageBox.critical(
|
||||
self.parent(), "错误",
|
||||
"无法导入pipeline模块,请确保water_quality_inversion_pipeline_GUI.py文件存在!"
|
||||
"无法导入 Pipeline 模块,请检查 src/core/handlers/ 目录是否完整!"
|
||||
)
|
||||
return
|
||||
|
||||
work_dir = self._workspace_initializer.work_dir or './work_dir'
|
||||
|
||||
if config is None:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '[运行] 正在收集所有步骤面板的配置...',
|
||||
'level': 'info',
|
||||
})
|
||||
self._panel_factory.preload_all()
|
||||
config = self._get_current_config()
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[运行] 已收集 {len(config)} 个步骤的配置',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'初始化 Pipeline,工作目录: {work_dir}',
|
||||
@ -295,6 +410,47 @@ class PipelineExecutor(QObject):
|
||||
})
|
||||
global_event_bus.publish('PipelineStopped', {})
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# EventBus 订阅回调
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
|
||||
def _on_request_run_single_step(self, data: dict):
|
||||
"""处理面板通过 EventBus 发出的单步执行请求。
|
||||
|
||||
data 格式: {'step_name': 'step1', 'config': {'step1': {...}}}
|
||||
|
||||
前置条件检查(预检/工作目录)由 run_single_step → _run_single_step_impl
|
||||
内部统一处理,此处仅做解析 + 转发 + 异常兜底。
|
||||
"""
|
||||
try:
|
||||
step_name = data.get('step_name')
|
||||
config = data.get('config')
|
||||
|
||||
if not step_name:
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': '[单步执行] 请求缺少 step_name,忽略',
|
||||
'level': 'warning',
|
||||
})
|
||||
return
|
||||
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[单步执行] 收到 {step_name} 的执行请求',
|
||||
'level': 'info',
|
||||
})
|
||||
|
||||
self.run_single_step(step_name, config)
|
||||
|
||||
except Exception as e:
|
||||
err_detail = traceback.format_exc()
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'[致命错误] _on_request_run_single_step({step_name}) 异常: {e}',
|
||||
'level': 'error',
|
||||
})
|
||||
global_event_bus.publish('LogMessage', {
|
||||
'message': f'详细追踪:\n{err_detail}',
|
||||
'level': 'error',
|
||||
})
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# WorkerThread 信号 → EventBus 事件(纯转发,零 UI 操作)
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
|
||||
@ -178,7 +178,7 @@ class VisualizationWorkerThread(QThread):
|
||||
{"task": "statistics", "output_paths": output_paths}
|
||||
)
|
||||
elif self.task == "scatter":
|
||||
from src.core.water_quality_inversion_pipeline_GUI import WaterQualityInversionPipeline
|
||||
from src.core.visualization.scatter_plot import generate_model_scatter_plots
|
||||
|
||||
training_csv_path = (self.extra.get("training_csv_path") or "").strip()
|
||||
models_dir = (self.extra.get("models_dir") or "").strip()
|
||||
@ -188,10 +188,9 @@ class VisualizationWorkerThread(QThread):
|
||||
if not models_dir or not Path(models_dir).is_dir():
|
||||
self.failed.emit("模型目录无效或不存在,请确认步骤6已生成 7_Supervised_Model_Training 下的参数子文件夹。")
|
||||
return
|
||||
pipeline = WaterQualityInversionPipeline(work_dir=str(wp))
|
||||
scatter_paths = pipeline.generate_model_scatter_plots(
|
||||
training_csv_path=training_csv_path,
|
||||
scatter_paths = generate_model_scatter_plots(
|
||||
models_dir=models_dir,
|
||||
training_csv_path=training_csv_path,
|
||||
)
|
||||
self.finished_ok.emit({"task": "scatter", "scatter_paths": scatter_paths or {}})
|
||||
elif self.task == "generate_all_selected":
|
||||
@ -205,11 +204,10 @@ class VisualizationWorkerThread(QThread):
|
||||
if training_csv.is_file():
|
||||
models_dir = wp / "7_Supervised_Model_Training"
|
||||
if models_dir.is_dir() and any(d.is_dir() for d in models_dir.iterdir()):
|
||||
from src.core.water_quality_inversion_pipeline_GUI import WaterQualityInversionPipeline
|
||||
pipeline = WaterQualityInversionPipeline(work_dir=str(wp))
|
||||
scatter_paths = pipeline.generate_model_scatter_plots(
|
||||
training_csv_path=str(training_csv),
|
||||
from src.core.visualization.scatter_plot import generate_model_scatter_plots
|
||||
scatter_paths = generate_model_scatter_plots(
|
||||
models_dir=str(models_dir),
|
||||
training_csv_path=str(training_csv),
|
||||
)
|
||||
count = len(scatter_paths) if scatter_paths else 0
|
||||
parts.append(f"散点图: {count} 个")
|
||||
|
||||
@ -54,16 +54,16 @@ def diagnose_pipeline_import_error():
|
||||
"[INFO] PyInstaller 环境:Pipeline 从程序内置包加载,跳过对仓库路径 src/core/*.py 的磁盘检查"
|
||||
)
|
||||
else:
|
||||
pipeline_file = os.path.normpath(
|
||||
os.path.join(os.path.dirname(__file__), "..", "..", "core", "water_quality_inversion_pipeline_GUI.py")
|
||||
handlers_dir = os.path.normpath(
|
||||
os.path.join(os.path.dirname(__file__), "..", "..", "core", "handlers")
|
||||
)
|
||||
if not os.path.exists(pipeline_file):
|
||||
error_info.append(f"[ERROR] Pipeline文件不存在: {pipeline_file}")
|
||||
if not os.path.isdir(handlers_dir):
|
||||
error_info.append(f"[ERROR] Handlers 目录不存在: {handlers_dir}")
|
||||
error_info.append(
|
||||
" 解决方案: 请确保项目结构完整,检查 src/core/ 下是否有 water_quality_inversion_pipeline_GUI.py"
|
||||
" 解决方案: 请确保项目结构完整,检查 src/core/handlers/ 目录是否存在"
|
||||
)
|
||||
else:
|
||||
error_info.append(f"[OK] Pipeline文件存在: {pipeline_file}")
|
||||
error_info.append(f"[OK] Handlers 目录存在: {handlers_dir}")
|
||||
|
||||
current_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
|
||||
if current_dir not in sys.path:
|
||||
@ -240,24 +240,34 @@ class WorkerThread(QThread):
|
||||
self.log_message.emit(f" [WARNING] {message}", "warning")
|
||||
|
||||
def run(self):
|
||||
"""运行 pipeline:子线程内切换 Matplotlib 为 Agg,避免 Qt5Agg 在后台线程绘图导致界面卡死。"""
|
||||
"""运行 pipeline:子线程内切换 Matplotlib 为 Agg,避免 Qt5Agg 在后台线程绘图导致界面卡死。
|
||||
|
||||
终极防崩溃设计:
|
||||
- 整个 run() 方法体包裹在单一 try/except 中
|
||||
- 任何未预期的异常都会被捕获并通过 finished 信号回报主线程
|
||||
- 确保前端永远不会面对"静默死亡"的后台线程
|
||||
"""
|
||||
import os
|
||||
# GDAL 环境变量保护(放在最前面,防止路径/编码问题)
|
||||
os.environ['GDAL_FILENAME_IS_UTF8'] = 'YES'
|
||||
os.environ['SHAPE_ENCODING'] = 'UTF-8'
|
||||
|
||||
mpl_prev = None
|
||||
try:
|
||||
import matplotlib
|
||||
mpl_prev = matplotlib.get_backend()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
import matplotlib.pyplot as plt
|
||||
plt.switch_backend("Agg")
|
||||
except Exception:
|
||||
mpl_prev = None
|
||||
try:
|
||||
# ★ 终端即时反馈
|
||||
print(f"\n[WorkerThread] 后台线程启动 (mode={self.mode}, work_dir={self.work_dir})")
|
||||
|
||||
# ── Matplotlib 后端切换(Agg 线程安全) ──
|
||||
try:
|
||||
import matplotlib
|
||||
mpl_prev = matplotlib.get_backend()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
import matplotlib.pyplot as plt
|
||||
plt.switch_backend("Agg")
|
||||
except Exception:
|
||||
mpl_prev = None
|
||||
|
||||
# ── 新架构:PipelineScheduler + Handler 注册表 ──
|
||||
scheduler = PipelineScheduler(work_dir=self.work_dir)
|
||||
scheduler.set_callback(self.pipeline_callback)
|
||||
@ -267,14 +277,17 @@ class WorkerThread(QThread):
|
||||
if self.mode == 'full':
|
||||
self.log_message.emit("开始运行完整流程 (Handler 调度模式)...", "info")
|
||||
|
||||
# ── ★ 预检已由 GUI 层 perform_preflight() 完成,此处不再重复预检 ──
|
||||
|
||||
# 过滤 skip_list 中的步骤
|
||||
active_config = {
|
||||
k: v for k, v in self.config.items()
|
||||
if k not in self.skip_list
|
||||
}
|
||||
|
||||
self.log_message.emit(
|
||||
f"[调度] 待执行步骤 ({len(active_config)} 个): {list(active_config.keys())}",
|
||||
"info"
|
||||
)
|
||||
|
||||
result = scheduler.run_full_pipeline(active_config)
|
||||
|
||||
errors = result.get('errors', {})
|
||||
@ -295,16 +308,28 @@ class WorkerThread(QThread):
|
||||
|
||||
self.progress_update.emit(100, f"步骤 {self.step_name} 执行完成")
|
||||
self.finished.emit(True, f"步骤 {self.step_name} 独立运行成功!")
|
||||
|
||||
except PipelineHalt as exc:
|
||||
# 预检失败 / 硬终止:透传清晰错误信息,不打印完整 traceback
|
||||
error_msg = str(exc)
|
||||
self.log_message.emit(f"[预检失败] {error_msg}", "error")
|
||||
self.finished.emit(False, error_msg)
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"执行失败: {str(e)}\n{traceback.format_exc()}"
|
||||
self.log_message.emit(error_msg, "error")
|
||||
self.finished.emit(False, error_msg)
|
||||
# ★ 终极捕获:任何未预期的异常都会被完整回报
|
||||
full_tb = traceback.format_exc()
|
||||
self.log_message.emit(f"[致命错误] 后台线程崩溃: {e}", "error")
|
||||
self.log_message.emit(f"详细追踪:\n{full_tb}", "error")
|
||||
# 同时 print 到终端(确保即使信号失效也能看到)
|
||||
print(f"\n{'='*60}")
|
||||
print(f"[WorkerThread 崩溃] {e}")
|
||||
print(f"{'='*60}")
|
||||
print(full_tb)
|
||||
print(f"{'='*60}\n")
|
||||
self.finished.emit(False, f"后台线程崩溃: {e}\n\n{full_tb}")
|
||||
|
||||
finally:
|
||||
# ── 恢复 Matplotlib 后端 ──
|
||||
if mpl_prev:
|
||||
try:
|
||||
import matplotlib.pyplot as plt
|
||||
|
||||
Reference in New Issue
Block a user