#!/usr/bin/env python # -*- coding: utf-8 -*- """ 水质参数反演分析系统 - 图形用户界面（重构版：纯壳模式） WaterQualityGUI 只负责窗口框架（标题栏、菜单栏、状态栏、QTabWidget）， 所有业务逻辑委托给独立的 Manager 类。 已实现的 Manager： - PanelFactory → 面板懒加载与生命周期 - PipelineExecutor → Pipeline 执行/停止/回调（通过 EventBus 发布状态） - WorkspaceInitializer → 工作目录选择 + 自动回填（通过 EventBus 广播） - LogManager → 日志区 + 进度条（内部订阅 LogMessage/ProgressUpdate） - ConfigManager → 配置读写（new/load/save/get_current_config） - DialogService → 纯展示类弹窗（Pipeline状态/关于/AI设置） - TrainingModeManager → 训练模式切换（发布 TrainingModeChanged 事件） """ import os import sys # ═══════════════════════════════════════════════════════════════ # 动态推断当前 Conda 环境下的 GDAL/PROJ 数据路径 # 必须在 import gdal / import rasterio 等任何第三方库之前执行 # ═══════════════════════════════════════════════════════════════ def _find_and_set_gdal_env(): """动态推断并强设 PROJ_LIB 和 GDAL_DATA 环境变量。 解决"C:\\ITRES\\ATK\\app\\GDAL\\projlib"等旧系统 PROJ 安装的干扰： 强制指向与当前 GDAL 版本匹配的 proj.db（通常在同 Conda 环境的 Library/share/proj 下），而非系统路径中的过时版本。 搜索策略（按优先级）： 1. CONDA_PREFIX + sys.prefix 的 Library/share/ 子目录 2. 搜索 sys.path 找到 osgeo 模块所在位置，回推 Library 路径 3. CONDA_ROOT / base 环境的 Library 路径 4. PATH 中推断的 conda 安装路径 5. 已知的硬编码回退路径 """ # ── 辅助函数 ── def _check_proj_db_version(db_path): """读取 proj.db 的 SQLite 头，返回主版本号。失败返回 -1。""" try: with open(db_path, "rb") as f: header = f.read(100) # SQLite 3 文件头: "SQLite format 3\0"，之后是页大小等 # DATABASE.LAYOUT.VERSION 在 offset 60 附近的 schema 中， # 但最简单的是读整个头中 "major" / "minor" 文本 if b"SQLite format 3" not in header: return -2 # PROJ 在 proj.db 的 metadata 表存储版本，我们直接看文件大小 # 粗略判定：< 2MB → v2-v4, 2~6MB → v5-v6, > 6MB → v7+ size_mb = os.path.getsize(db_path) / (1024 * 1024) return 6 if size_mb > 3.0 else 2 # v6+ 的 proj.db 通常 > 3MB except Exception: return -1 def _find_proj_in_prefix(prefix): """在指定前缀下递归搜索 Library/share/proj/proj.db""" candidates = [] for root, dirs, files in os.walk(prefix): # 限制深度避免全盘扫描 depth = root.replace(prefix, "").count(os.sep) if depth > 6: dirs.clear() continue if "proj.db" in files: db_path = os.path.join(root, "proj.db") ver = _check_proj_db_version(db_path) if ver >= 6: candidates.append((ver, root)) # 跳过明显的非目标目录 dirs[:] = [d for d in dirs if d not in (".git", "__pycache__", "node_modules")] if candidates: candidates.sort(key=lambda x: -x[0]) # 最新版本优先 return candidates[0][1] return None def _find_gdal_in_prefix(prefix): """在指定前缀下查找 Library/share/gdal""" for sub in ("Library/share/gdal", "share/gdal", "Library\\share\\gdal", "share\\gdal"): path = os.path.join(prefix, sub) if os.path.isdir(path): for marker in ("gdalvrt.xsd", "pcs.csv", "gcs.csv"): if os.path.isfile(os.path.join(path, marker)): return path return None # ── 构建搜索前缀 ── prefixes = [] # 1) CONDA_PREFIX + sys.prefix（最优先） for src in (os.environ.get("CONDA_PREFIX", ""), sys.prefix): if src and src not in prefixes: prefixes.append(src) # 2) 从 Python 可执行文件路径回推（最可靠：python.exe 就在 {prefix}/ 下） try: py_exe = sys.executable py_prefix = os.path.dirname(py_exe) # python.exe 所在目录 if os.path.basename(py_prefix).lower() == "scripts": # Windows venv: {prefix}/Scripts/python.exe → 前缀是父目录 exe_prefix = os.path.dirname(py_prefix) else: exe_prefix = py_prefix if exe_prefix and exe_prefix not in prefixes: prefixes.insert(0, exe_prefix) # 插到最前面，最高优先级 except Exception: pass # 3) 从 sys.path 中 site-packages 位置回推所有可能的 conda/env 前缀 try: for sp in sys.path: if not sp or sp in prefixes: continue norm = os.path.normpath(sp) parts = norm.split(os.sep) # 检查 sp 是否在 site-packages 下 → 回推到 {prefix} for marker in ("site-packages", "Lib"): if marker in parts: idx = parts.index(marker) possible = os.sep.join(parts[:idx]) if os.path.isdir(possible) and possible not in prefixes: prefixes.append(possible) break except Exception: pass # 3) CONDA_ROOT / base 环境 for key in ("CONDA_ROOT", "MAMBA_ROOT_PREFIX"): val = os.environ.get(key, "") if val and val not in prefixes: prefixes.append(val) # 4) 从 PATH 推断 for d in os.environ.get("PATH", "").split(os.pathsep): if not d or d in prefixes: continue lower = d.lower() if any(kw in lower for kw in ("anaconda", "miniconda", "anconda")): # 找到 conda 根: 去掉 /Scripts 或 /condabin if os.path.basename(d).lower() in ("scripts", "condabin", "bin"): root = os.path.dirname(d) else: root = d if root and root not in prefixes: prefixes.append(root) # 5) 已知硬编码路径 for p in [ r"D:\111\changyongruanjian\anconda\envs\WQ_GUI", r"D:\111\changyongruanjian\anconda", r"C:\ProgramData\anaconda3", r"C:\ProgramData\miniconda3", ]: if os.path.isdir(p) and p not in prefixes: prefixes.append(p) # ── 额外：扫描 conda 根目录下的 envs/*/Library/share/proj ── _extra_prefixes = [] for prefix in list(prefixes): envs_dir = os.path.join(prefix, "envs") if os.path.isdir(envs_dir): try: for entry in os.listdir(envs_dir): env_path = os.path.join(envs_dir, entry) if os.path.isdir(env_path) and env_path not in prefixes: _extra_prefixes.append(env_path) except Exception: pass prefixes.extend(_extra_prefixes) # 6) pip wheel GDAL: {site-packages}/osgeo/data/proj/proj.db # (conda-forge GDAL wheel on Windows 有时把 proj.db 放这里) try: import site as _site for _sp in _site.getsitepackages(): pip_proj = os.path.join(_sp, "osgeo", "data", "proj") if os.path.isfile(os.path.join(pip_proj, "proj.db")): pip_gdal = os.path.join(_sp, "osgeo", "data", "gdal") if _sp not in prefixes: prefixes.append(_sp) except Exception: pass proj_lib = None gdal_data = None for prefix in prefixes: if not proj_lib: # 先尝试标准子目录（快速路径） for sub in ("Library/share/proj", "share/proj", "Library\\share\\proj", "share\\proj"): db_path = os.path.join(prefix, sub, "proj.db") if os.path.isfile(db_path) and _check_proj_db_version(db_path) >= 6: proj_lib = os.path.join(prefix, sub) break # 标准路径未找到 → 递归搜索（仅对 conda 环境前缀） if not proj_lib and ( os.path.basename(prefix) not in ("", "share", "bin", "Scripts") ): proj_lib = _find_proj_in_prefix(prefix) if not gdal_data: gdal_data = _find_gdal_in_prefix(prefix) if proj_lib and gdal_data: break # ── 应用环境变量 ── if proj_lib: os.environ["PROJ_LIB"] = proj_lib # PROJ_DATA: Fiona / Rasterio / GDAL ≥3.5 / PROJ ≥8 优先读取此变量 # 不设置的话 Fiona 会使用自带的旧版 proj.db (VERSION.MINOR=2)， # 导致投影解析错误 → 矢量图层错位 → 水体被错误擦除 → 有效像元为 0 os.environ["PROJ_DATA"] = proj_lib print(f"[ENV] PROJ_LIB → {proj_lib}") print(f"[ENV] PROJ_DATA → {proj_lib}") else: print("[ENV] ⚠ 未找到兼容的 proj.db (需要 PROJ v6+)，PROJ_LIB/PROJ_DATA 保持系统默认") print(f"[ENV] (已搜索 {len(prefixes)} 个前缀: {[os.path.basename(p) for p in prefixes[:5]]}...)") print("[ENV] 如果 step11 地图渲染报 PROJ 版本错误，") print("[ENV] 请手动设置: set PROJ_LIB=\\Library\\share\\proj") print("[ENV] set PROJ_DATA=<同上>") if gdal_data: os.environ["GDAL_DATA"] = gdal_data print(f"[ENV] GDAL_DATA → {gdal_data}") else: print("[ENV] ⚠ 未找到 GDAL data 目录，GDAL_DATA 保持系统默认") print(f"[ENV] (已搜索 {len(prefixes)} 个前缀)") # ── 屏蔽 Fiona 内部旧版 PROJ 路径 ── # Fiona wheels 捆绑了古老的 proj.db (VERSION.MINOR=2)， # 即使设置了 PROJ_DATA 仍然可能被某些旧版 Fiona 忽略。 # 以下环境变量作为最后防线强制覆盖。 if proj_lib: # GDAL ≥3.5 同时检查 PROJ_DATA 和 PROJ_LIB os.environ["PROJ_NETWORK"] = "OFF" # 禁用 PROJ 网络下载(避免意外行为) # 记录此信息供诊断 if "PROJ_DATA" in os.environ: print("[ENV] 已屏蔽 Fiona 内部旧版 PROJ 数据库，全进程统一使用上述 proj.db") _find_and_set_gdal_env() import ctypes import traceback import multiprocessing from datetime import datetime from PyQt5.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QTabWidget, QToolBar, QSizePolicy, QListWidget, QListWidgetItem, QGroupBox, QTextEdit, QProgressBar, QMessageBox, QFileDialog, QSpinBox, QDoubleSpinBox, QComboBox, ) from PyQt5.QtCore import Qt, QTimer, QSize from PyQt5.QtGui import QIcon, QFont, QPixmap, QColor, QTextCursor if multiprocessing.current_process().name == 'MainProcess': if not QApplication.instance(): try: QApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True) QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True) except Exception: pass _global_app = QApplication(sys.argv) def get_resource_path(relative_path: str) -> str: if hasattr(sys, '_MEIPASS'): return os.path.join(sys._MEIPASS, relative_path) return os.path.abspath( os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), relative_path) ) def global_exception_handler(exc_type, exc_value, exc_traceback): err_lines = traceback.format_exception(exc_type, exc_value, exc_traceback) err_msg = "".join(err_lines) dump_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "crash_dump.txt") try: with open(dump_path, "a", encoding="utf-8") as f: f.write(f"\n{'='*60}\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]\n") f.write(err_msg) except Exception: pass try: QMessageBox.critical(None, "程序崩溃", f"错误类型: {exc_type.__name__}\n错误信息: {exc_value}\n详细信息已写入: {dump_path}") except Exception: pass sys.excepthook = global_exception_handler class WaterQualityGUI(QMainWindow): """水质参数反演分析系统主窗口 —— 纯壳模式。 职责边界： - 窗口框架（标题栏、菜单栏、状态栏、导航栏） - QTabWidget 托管（通过 PanelFactory 懒加载） - 日志区 + 进度条（UI 控件归 shell，内容由 EventBus 驱动） - 各 Manager 的创建与 EventBus 连线 """ def __init__(self): my_appid = u'mycompany.megacube.waterquality.v1' ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(my_appid) super().__init__() icon_path = get_resource_path("data/icons-1/uitubiao.ico") self.setWindowIcon(QIcon(icon_path)) # 第一步：创建各 Manager（纯连线，不执行业务） self._init_managers() # 第二步：构建窗口壳 self._init_shell() # 第三步：订阅 EventBus 事件 → 驱动 UI 更新 self._wire_event_bus() # 第四步：应用样式 + 禁用滚轮 self._apply_stylesheet() self._disable_wheel_for_all_spinboxes() # 第五步：默认选中第一个步骤（延迟执行，确保导航列表和 Tab 均已就位） QTimer.singleShot(120, self._select_first_nav_item) # 第六步：延迟启动工作目录选择 QTimer.singleShot(200, self._workspace_initializer.run) # ================================================================ # Manager 初始化 # ================================================================ def _init_managers(self): from src.gui.core.panel_factory import PanelFactory from src.gui.core.panel_registry import PANEL_REGISTRY from src.gui.core.workspace_initializer import WorkspaceInitializer from src.gui.core.pipeline_executor import PipelineExecutor from src.gui.core.log_manager import LogManager from src.gui.core.config_manager import ConfigManager from src.gui.core.dialog_service import DialogService from src.gui.core.training_mode_manager import TrainingModeManager from src.gui.core.event_bus import global_event_bus self._panel_factory = PanelFactory( registry=PANEL_REGISTRY, main_window=self, preload_window=1, ) self._workspace_initializer = WorkspaceInitializer( panel_factory=self._panel_factory, parent=self, ) self._pipeline_executor = PipelineExecutor( panel_factory=self._panel_factory, workspace_initializer=self._workspace_initializer, parent=self, ) self._log_manager = LogManager(parent=self) self._config_manager = ConfigManager( panel_factory=self._panel_factory, parent=self, ) self._dialog_service = DialogService(parent=self) self._training_mode_manager = TrainingModeManager(parent=self) self._event_bus = global_event_bus # ================================================================ # 窗口壳构建 # ================================================================ def _init_shell(self): self.setWindowTitle("MegaCube-Water Quality V1.2.1") screen_geometry = QApplication.primaryScreen().availableGeometry() screen_width = screen_geometry.width() screen_height = screen_geometry.height() self.resize(1200, screen_height) self.move((screen_width - 1200) // 2, 0) self.setMinimumSize(600, 400) self._create_title_bar() self._create_banner() self._create_central_layout() self.statusBar().showMessage("就绪") def _create_title_bar(self): title_widget = QWidget() title_layout = QHBoxLayout() title_layout.setContentsMargins(8, 4, 8, 4) title_layout.setSpacing(0) logo_label = QLabel() logo_label.setFixedSize(180, 48) logo_label.setAlignment(Qt.AlignCenter) logo_label.setStyleSheet( "background-color: #f8f9fa;" "border-top-left-radius: 4px; border-bottom-left-radius: 4px;" ) logo_path = get_resource_path("data/icons/logo.png") logo_pixmap = QPixmap(logo_path) if not logo_pixmap.isNull(): logo_label.setPixmap(logo_pixmap.scaledToHeight(38, Qt.SmoothTransformation)) else: logo_label.setText("Logo") title_layout.addWidget(logo_label) menubar = self.menuBar() menubar.setStyleSheet(""" QMenuBar { background-color: #f8f9fa; border: none; padding: 4px 8px; } QMenuBar::item { padding: 6px 12px; font-size: 13px; } QMenuBar::item:selected { background-color: #e6f0ff; border-radius: 3px; } """) self._build_menus(menubar) title_layout.addWidget(menubar) title_widget.setLayout(title_layout) title_widget.setStyleSheet( "background-color: #f8f9fa; border-bottom: 1px solid #d0d0d0;" ) self.setMenuWidget(title_widget) def _build_menus(self, menubar): file_menu = menubar.addMenu("文件") file_menu.addAction("新建配置").triggered.connect(self._on_new_config) file_menu.addAction("打开配置").triggered.connect(self._on_load_config) file_menu.addAction("保存配置").triggered.connect(self._on_save_config) file_menu.addSeparator() file_menu.addAction("退出").triggered.connect(self.close) tools_menu = menubar.addMenu("工具") tools_menu.addAction("设置工作目录").triggered.connect( self._workspace_initializer.set_work_directory ) tools_menu.addAction("打开工作目录").triggered.connect( self._workspace_initializer.open_work_directory ) tools_menu.addSeparator() tools_menu.addAction("AI 引擎配置...").triggered.connect(self._on_ai_settings) tools_menu.addSeparator() tools_menu.addAction("自动填充所有输入路径").triggered.connect( self._workspace_initializer.auto_populate_all ) self._training_mode_action = tools_menu.addAction("有训练数据模式") self._training_mode_action.setCheckable(True) self._training_mode_action.setChecked(True) self._training_mode_action.triggered.connect(self._on_toggle_training_mode) help_menu = menubar.addMenu("帮助") help_menu.addAction("检查Pipeline状态").triggered.connect(self._on_show_pipeline_status) help_menu.addSeparator() help_menu.addAction("关于").triggered.connect(self._on_show_about) def _create_banner(self): banner_widget = QWidget() banner_layout = QHBoxLayout() banner_layout.setContentsMargins(0, 0, 0, 0) banner_layout.setSpacing(0) self._banner_label = QLabel() self._banner_label.setMinimumHeight(140) self._banner_label.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed) self._banner_label.setScaledContents(False) banner_path = get_resource_path("data/icons/Mega Water 1.0.jpg") self._banner_pixmap = QPixmap(banner_path) if not self._banner_pixmap.isNull(): QTimer.singleShot(50, self._update_banner_image) banner_layout.addWidget(self._banner_label) self._banner_title_label = QLabel("MegaCube-Water Quality V1.2.1", self._banner_label) # 现代化字体设计：去掉老旧的衬线体，换用带字间距的粗体无衬线字体，增加高级感 self._banner_title_label.setStyleSheet(""" QLabel { background: transparent; color: #FFFFFF; font-size: 42px; font-weight: bold; font-family: "Microsoft YaHei UI", "Segoe UI", sans-serif; letter-spacing: 3px; } """) self._banner_title_label.setAttribute(Qt.WA_TransparentForMouseEvents) self._banner_title_label.show() self._banner_title_label.raise_() banner_widget.setLayout(banner_layout) banner_toolbar = QToolBar() banner_toolbar.setMovable(False) banner_toolbar.setFloatable(False) banner_toolbar.addWidget(banner_widget) banner_toolbar.setStyleSheet( "QToolBar { background: white; border: none; padding: 0px; margin: 0px; }" ) self.addToolBar(Qt.TopToolBarArea, banner_toolbar) def _create_central_layout(self): from src.gui.styles import ModernStylesheet central_widget = QWidget() main_layout = QHBoxLayout() main_layout.setContentsMargins(0, 0, 0, 0) main_layout.setSpacing(0) main_layout.addWidget(self._create_navigation(), 1) right_widget = QWidget() right_layout = QVBoxLayout() right_layout.setContentsMargins(15, 15, 15, 15) right_layout.setSpacing(10) self._tab_widget = self._panel_factory.create_tab_widget(icons_dir="data/icons") self._tab_widget.tabBar().setVisible(False) right_layout.addWidget(self._tab_widget, 3) # 上一步 / 下一步 导航按钮 nav_btn_layout = QHBoxLayout() nav_btn_layout.setSpacing(12) self._prev_btn = QPushButton(" 上一步 ") self._prev_btn.setMinimumHeight(36) self._prev_btn.setMinimumWidth(100) # 上一步是次要操作，用 normal 样式 self._prev_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet('normal')) self._prev_btn.clicked.connect(self._on_prev_clicked) nav_btn_layout.addWidget(self._prev_btn) self._next_btn = QPushButton(" 下一步 ") self._next_btn.setMinimumHeight(36) self._next_btn.setMinimumWidth(100) # 下一步是主要操作，用 primary (主题蓝) 样式 self._next_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet('primary')) self._next_btn.clicked.connect(self._on_next_clicked) nav_btn_layout.addWidget(self._next_btn) nav_btn_layout.addStretch() right_layout.addLayout(nav_btn_layout) right_layout.addWidget(self._log_manager.create_log_panel(), 1) right_widget.setLayout(right_layout) main_layout.addWidget(right_widget, 4) central_widget.setLayout(main_layout) self.setCentralWidget(central_widget) def _create_navigation(self): from src.gui.core.panel_registry import build_stage_groups from src.gui.styles import ModernStylesheet nav_widget = QWidget() nav_layout = QVBoxLayout() nav_layout.setContentsMargins(0, 15, 0, 15) # 去掉左右边距，让列表通栏 nav_layout.setSpacing(10) title = QLabel("流程导航") title.setFont(QFont("Microsoft YaHei", 14, QFont.Bold)) title.setAlignment(Qt.AlignCenter) title.setStyleSheet(f"color: {ModernStylesheet.COLORS['primary']}; padding: 10px;") nav_layout.addWidget(title) self._step_list = QListWidget() # 注入纯净的现代侧边栏 QSS self._step_list.setStyleSheet(f""" QListWidget {{ background-color: {ModernStylesheet.COLORS['panel_bg']}; border: none; outline: 0; /* 去掉点击时的虚线框 */ }} QListWidget::item {{ padding: 12px 16px; border-left: 4px solid transparent; color: {ModernStylesheet.COLORS['text_primary']}; }} QListWidget::item:hover {{ background-color: {ModernStylesheet.COLORS['hover']}; }} QListWidget::item:selected {{ background-color: {ModernStylesheet.COLORS['selected']}; color: {ModernStylesheet.COLORS['primary']}; font-weight: bold; border-left: 4px solid {ModernStylesheet.COLORS['primary']}; }} """) process_stages = build_stage_groups() stage_names = list(process_stages.keys()) for stage_idx, (stage_name, steps) in enumerate(process_stages.items()): # 分类头 stage_item = QListWidgetItem(stage_name) stage_font = QFont("Microsoft YaHei", 11, QFont.Bold) stage_item.setFont(stage_font) stage_item.setForeground(QColor(ModernStylesheet.COLORS['primary'])) stage_item.setBackground(QColor(ModernStylesheet.COLORS['main_bg'])) # 给头加个浅灰底色区隔 stage_item.setFlags(stage_item.flags() & ~Qt.ItemIsSelectable & ~Qt.ItemIsEnabled) stage_item.setData(Qt.UserRole, "stage_header") self._step_list.addItem(stage_item) # 具体步骤 for step_id, step_display in steps: item = QListWidgetItem(f" {step_display}") # 增加缩进 item.setData(Qt.UserRole, step_id) item.setFont(QFont("Microsoft YaHei", 10)) self._step_list.addItem(item) # 分隔符 (不需要实体占用高度，缩小即可) if stage_idx < len(stage_names) - 1: sep = QListWidgetItem("") sep.setSizeHint(QSize(0, 10)) sep.setFlags(sep.flags() & ~Qt.ItemIsSelectable & ~Qt.ItemIsEnabled) self._step_list.addItem(sep) self._step_list.currentRowChanged.connect(self._on_step_list_changed) nav_layout.addWidget(self._step_list) # 底部按钮区 (仅保留强制停止) btn_layout = QVBoxLayout() btn_layout.setContentsMargins(15, 0, 15, 0) self._stop_btn = QPushButton("⏹ 强制停止当前任务") self._stop_btn.setEnabled(False) self._stop_btn.setMinimumHeight(38) self._stop_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet('danger')) self._stop_btn.clicked.connect(self._pipeline_executor.stop_pipeline) btn_layout.addWidget(self._stop_btn) nav_layout.addLayout(btn_layout) nav_widget.setLayout(nav_layout) nav_widget.setMaximumWidth(260) nav_widget.setStyleSheet( f"background-color: {ModernStylesheet.COLORS['panel_bg']};" f"border-right: 1px solid {ModernStylesheet.COLORS['border_light']};" ) return nav_widget # ================================================================ # EventBus 连线（UI 状态由事件驱动） # ================================================================ def _wire_event_bus(self): self._event_bus.subscribe('PipelineStarted', self._on_pipeline_started) self._event_bus.subscribe('PipelineFinished', self._on_pipeline_finished) self._event_bus.subscribe('PipelineStopped', self._on_pipeline_stopped) self._event_bus.subscribe('NavigateToTab', self._on_navigate_to_tab) self._event_bus.subscribe('WorkspaceChanged', self._on_workspace_changed) def _on_pipeline_started(self, data): self._stop_btn.setEnabled(True) self._log_manager.progress_bar.setValue(0) def _on_pipeline_finished(self, data): self._stop_btn.setEnabled(False) success = data.get('success', False) message = data.get('message', '') if success: self._log_manager.progress_bar.setValue(100) QMessageBox.information(self, "完成", "流程执行成功！\n\n请查看工作目录中的结果文件。") else: QMessageBox.critical(self, "失败", f"流程执行失败:\n\n{message[:600]}" + ("\n...(已截断)" if len(message) > 600 else "")) def _on_pipeline_stopped(self, data): self._stop_btn.setEnabled(False) def _on_navigate_to_tab(self, data): tab_index = data.get('tab_index', 0) if 0 <= tab_index < self._tab_widget.count(): self._tab_widget.setCurrentIndex(tab_index) def _on_workspace_changed(self, data): work_dir = data.get('work_dir', '') self.statusBar().showMessage(f"工作目录: {work_dir}") # 让当前激活的 Tab 立即基于新目录重算默认路径，覆盖掉旧值，绕过非空保护 from src.gui.core.panel_registry import PANEL_REGISTRY active_idx = self._tab_widget.currentIndex() if 0 <= active_idx < len(PANEL_REGISTRY): active_step_id = PANEL_REGISTRY[active_idx]['step_id'] panel = self._panel_factory.get_panel(active_step_id) if panel is not None and hasattr(panel, 'update_from_config'): try: panel.update_from_config( work_dir=self._workspace_initializer.work_dir, pipeline=None ) except Exception: pass # ================================================================ # 导航 → Tab 单向路由（左侧 List 驱动右侧 Tab，Tab 头部已隐藏） # ================================================================ def _select_first_nav_item(self): """默认选中左侧导航栏的第一个可导航步骤项（跳过阶段标题/分隔符）。""" for i in range(self._step_list.count()): item = self._step_list.item(i) if item and item.data(Qt.UserRole) not in (None, "stage_header"): self._step_list.setCurrentRow(i) return def _on_step_list_changed(self, index): if index < 0: return item = self._step_list.item(index) if not item: return item_data = item.data(Qt.UserRole) if item_data in (None, "stage_header"): return # 跳过阶段标题 from src.gui.core.panel_registry import get_tab_index, PANEL_REGISTRY tab_index = get_tab_index(item_data) if tab_index < 0: return try: # 1. 触发懒加载生成面板 panel = self._panel_factory.get_panel(item_data) # ★ 2026-07-01：每次切页时刷新面板的自动路由 # 面板首次加载时 _replay_state_to_panel 会调 update_from_config， # 但再次切回时 get_panel() 直接返回已有实例，不会重扫文件系统。 # 此处显式调用确保 Step11 等面板始终基于最新磁盘状态做文件夹自动导入。 if panel is not None and hasattr(panel, 'update_from_config'): try: panel.update_from_config( work_dir=self._workspace_initializer.work_dir, pipeline=None, ) except Exception: pass # 🚨 核心防卡死补丁：如果目标 Tab 被后台任务异常永久锁定，强制撬开！ if not self._tab_widget.isTabEnabled(tab_index): self._log_manager.info(f"检测到 {item_data} 处于异常锁定状态，已执行强制解锁。") self._tab_widget.setTabEnabled(tab_index, True) QMessageBox.warning( self, "页面已解锁", f"步骤「{item_data}」之前被异常锁定，已自动强制解锁。\n" "如果当前有流程正在运行，请勿修改此页面的参数。" ) # 【防御】：用已加载 panel 实例反查真实 Tab 索引，防止索引错位 try: for i in range(self._tab_widget.count()): scroll_area = self._tab_widget.widget(i) if scroll_area and hasattr(scroll_area, 'widget'): if scroll_area.widget() == panel: tab_index = i break except Exception as e: self._log_manager.error(f"页面索引校验失败: {str(e)}") return # 2. 强制使用注册表的固定索引进行跳转 self._tab_widget.setCurrentIndex(tab_index) # 【新增修复】：每次切页时，强制重播所有面板的输入值！ # 原理：打破 preload_window 造成的时序差，确保目标页面能拿到源页面最新填写的值 if hasattr(self, '_panel_factory'): self._panel_factory.replay_live_panel_inputs() # 3. 防撕裂回弹机制：如果由于某种原因没跳过去，把左边的蓝条强行拽回当前真实页面 if self._tab_widget.currentIndex() != tab_index: self._step_list.blockSignals(True) current_step_id = PANEL_REGISTRY[self._tab_widget.currentIndex()]['step_id'] for i in range(self._step_list.count()): list_item = self._step_list.item(i) if list_item and list_item.data(Qt.UserRole) == current_step_id: self._step_list.setCurrentRow(i) break self._step_list.blockSignals(False) except Exception as e: self._log_manager.error(f"页面跳转失败: {str(e)}") def _find_prev_step_row(self, current_row): """从 current_row 向上遍历，跳过 stage_header 和空分隔符，返回上一个有效 step 的行号。""" for i in range(current_row - 1, -1, -1): item = self._step_list.item(i) if not item: continue data = item.data(Qt.UserRole) if data and data != "stage_header": return i return None def _find_next_step_row(self, current_row): """从 current_row 向下遍历，跳过 stage_header 和空分隔符，返回下一个有效 step 的行号。""" for i in range(current_row + 1, self._step_list.count()): item = self._step_list.item(i) if not item: continue data = item.data(Qt.UserRole) if data and data != "stage_header": return i return None def _on_prev_clicked(self): """上一步按钮：跳转到上一个有效步骤。""" row = self._find_prev_step_row(self._step_list.currentRow()) if row is not None: self._step_list.setCurrentRow(row) def _on_next_clicked(self): """下一步按钮：跳转到下一个有效步骤。""" row = self._find_next_step_row(self._step_list.currentRow()) if row is not None: self._step_list.setCurrentRow(row) # ================================================================ # 横幅自适应 # ================================================================ def _update_banner_image(self): if not hasattr(self, '_banner_pixmap') or self._banner_pixmap.isNull(): return TARGET_HEIGHT = 140 target_width = self.width() orig_w = self._banner_pixmap.width() orig_h = self._banner_pixmap.height() scale_factor = max(target_width / orig_w, TARGET_HEIGHT / orig_h) new_w = int(orig_w * scale_factor) new_h = int(orig_h * scale_factor) scaled = self._banner_pixmap.scaled(new_w, new_h, Qt.IgnoreAspectRatio, Qt.SmoothTransformation) crop_x = (new_w - target_width) // 2 crop_y = (new_h - TARGET_HEIGHT) // 2 final = scaled.copy(crop_x, crop_y, target_width, TARGET_HEIGHT) self._banner_label.setFixedHeight(TARGET_HEIGHT) self._banner_label.setFixedWidth(target_width) self._banner_label.setPixmap(final) if hasattr(self, '_banner_title_label'): title_x = 160 title_y = max(0, (TARGET_HEIGHT - 60) // 2) self._banner_title_label.move(title_x, title_y) self._banner_title_label.resize(target_width - title_x - 20, 60) def resizeEvent(self, event): super().resizeEvent(event) self._update_banner_image() # ================================================================ # 样式与 UX # ================================================================ def _apply_stylesheet(self): from src.gui.styles import ModernStylesheet self.setStyleSheet(ModernStylesheet.get_main_stylesheet()) def _disable_wheel_for_all_spinboxes(self): for sb in self.findChildren(QSpinBox): sb.setFocusPolicy(Qt.StrongFocus) sb.wheelEvent = lambda e, s=sb: None for sb in self.findChildren(QDoubleSpinBox): sb.setFocusPolicy(Qt.StrongFocus) sb.wheelEvent = lambda e, s=sb: None for cb in self.findChildren(QComboBox): cb.setFocusPolicy(Qt.StrongFocus) cb.wheelEvent = lambda e, c=cb: None # ================================================================ # 菜单回调（全部委托给对应 Manager） # ================================================================ def _on_new_config(self): self._config_manager.new_config() def _on_load_config(self): self._config_manager.load_config() def _on_save_config(self): self._config_manager.save_config() def _on_ai_settings(self): self._dialog_service.show_ai_settings() def _on_toggle_training_mode(self, checked): self._training_mode_manager.toggle(checked) self._training_mode_action.setText( self._training_mode_manager.get_action_text(checked) ) def _on_show_pipeline_status(self): self._dialog_service.show_pipeline_status() def _on_show_about(self): self._dialog_service.show_about() # ================================================================ # 向后兼容属性 # ================================================================ @property def panels(self): return self._panel_factory.get_loaded_panels() @property def work_dir(self): return self._workspace_initializer.work_dir @work_dir.setter def work_dir(self, value): self._workspace_initializer.work_dir = value @property def worker(self): return self._pipeline_executor.worker def log_message(self, message, level='info'): self._event_bus.publish('LogMessage', {'message': message, 'level': level}) def update_progress(self, percentage, message): self._event_bus.publish('ProgressUpdate', {'percentage': percentage, 'message': message}) def get_current_config(self): return self._config_manager.get_current_config() # ============================================================ def main(): if multiprocessing.current_process().name != 'MainProcess': sys.exit(0) # ★ GDAL 依赖检查（移至此处以利用 global_exception_handler 展示友好弹窗） try: import osgeo # noqa: F401 from osgeo import gdal, ogr # noqa: F401 except ImportError: from PyQt5.QtWidgets import QMessageBox, QApplication as _QA _qa = _QA.instance() or _QA(sys.argv) QMessageBox.critical( None, "依赖缺失", "GDAL (osgeo) 未安装，程序无法运行。\n\n" "请运行以下命令安装：\n" " conda install -c conda-forge gdal\n" "或访问 https://gdal.org/download.html" ) sys.exit(1) app = QApplication.instance() if not app: app = QApplication(sys.argv) app.setApplicationName("Mega Water") app.setOrganizationName("WaterQuality") try: from src.auth.license_manager import verify_license from src.auth.license_dialog import LicenseDialog except ImportError: _current_dir = os.path.dirname(os.path.abspath(__file__)) _project_root = os.path.abspath(os.path.join(_current_dir, '..', '..')) if _project_root not in sys.path: sys.path.insert(0, _project_root) from src.auth.license_manager import verify_license from src.auth.license_dialog import LicenseDialog _is_license_valid, _license_msg = verify_license() if not _is_license_valid: LicenseDialog().exec_() sys.exit(0) window = WaterQualityGUI() window.show() sys.exit(app.exec_()) if __name__ == "__main__": if multiprocessing.current_process().name != 'MainProcess': sys.exit(0) try: multiprocessing.freeze_support() except Exception: pass main()