#!/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 osgeo # noqa: F401 from osgeo import gdal, ogr # noqa: F401 import os import sys 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, ) from PyQt5.QtCore import Qt, QTimer 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() # 第五步：延迟启动工作目录选择 QTimer.singleShot(100, 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") 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", self._banner_label) self._banner_title_label.setStyleSheet(""" QLabel { background: transparent; color: white; font-size: 48px; font-family: "Times New Roman", "Georgia", serif; } """) 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(10) self._prev_btn = QPushButton("< 上一步") self._prev_btn.setMinimumHeight(32) self._prev_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet('default')) self._prev_btn.clicked.connect(self._on_prev_clicked) nav_btn_layout.addWidget(self._prev_btn) self._next_btn = QPushButton("下一步 >") self._next_btn.setMinimumHeight(32) self._next_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet('default')) 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(10, 15, 10, 15) nav_layout.setSpacing(10) title = QLabel("流程步骤") title.setFont(QFont("Arial", 13, QFont.Bold)) title.setAlignment(Qt.AlignCenter) title.setStyleSheet(f"color: {ModernStylesheet.COLORS['text_primary']}; padding: 10px;") nav_layout.addWidget(title) self._step_list = QListWidget() self._step_list.setStyleSheet(ModernStylesheet.get_sidebar_stylesheet()) 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("Arial", 11, QFont.Bold) stage_item.setFont(stage_font) stage_item.setForeground(QColor(ModernStylesheet.COLORS.get('accent', '#0078D4'))) stage_item.setFlags(stage_item.flags() & ~Qt.ItemIsSelectable) stage_item.setFlags(stage_item.flags() & ~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) step_font = QFont("Arial", 10) item.setFont(step_font) item.setForeground(QColor(ModernStylesheet.COLORS.get('text_secondary', '#666666'))) self._step_list.addItem(item) if stage_idx < len(stage_names) - 1: sep = QListWidgetItem("") sep.setFlags(sep.flags() & ~Qt.ItemIsSelectable) sep.setFlags(sep.flags() & ~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.setSpacing(8) self._run_all_btn = QPushButton("> 运行完整流程") self._run_all_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet('success')) self._run_all_btn.setMinimumHeight(35) self._run_all_btn.clicked.connect(self._on_run_all_clicked) btn_layout.addWidget(self._run_all_btn) self._stop_btn = QPushButton("⏹ 停止") self._stop_btn.setEnabled(False) self._stop_btn.setMinimumHeight(35) 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(280) 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._run_all_btn.setEnabled(False) self._stop_btn.setEnabled(True) self._log_manager.progress_bar.setValue(0) def _on_pipeline_finished(self, data): self._run_all_btn.setEnabled(True) 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[:200]}") def _on_pipeline_stopped(self, data): self._run_all_btn.setEnabled(True) 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}") # 遍历已加载面板，调用 update_from_config 重建内存级参数和默认输出路径 for step_id, panel in self._panel_factory.get_loaded_panels().items(): if not hasattr(panel, 'update_from_config'): continue try: panel.update_from_config(work_dir=work_dir, pipeline=None) except Exception: pass def _on_run_all_clicked(self): print("==== 按钮确实被按下了 ====", flush=True) try: self._pipeline_executor.run_full_pipeline() except Exception as e: print(f"执行器调用崩溃: {e}") traceback.print_exc() # ================================================================ # 导航 → Tab 单向路由（左侧 List 驱动右侧 Tab，Tab 头部已隐藏） # ================================================================ def _on_step_list_changed(self, index): """左侧导航 → 右侧 Tab 单向路由（Tab 头部已隐藏，无反向同步）。""" if index < 0: return item = self._step_list.item(index) if not item: return item_data = item.data(Qt.UserRole) if item_data == "stage_header" or item_data is None: return from src.gui.core.panel_registry import get_tab_index tab_index = get_tab_index(item_data) if tab_index < 0: return # 先触发懒加载再切 Tab，避免 removeTab/insertTab 与导航事件重叠 self._panel_factory.get_panel(item_data) self._tab_widget.setCurrentIndex(tab_index) # 动态更新上一步/下一步按钮可用状态 self._prev_btn.setEnabled(self._find_prev_step_row(index) is not None) self._next_btn.setEnabled(self._find_next_step_row(index) is not None) 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): from PyQt5.QtWidgets import QSpinBox, QDoubleSpinBox, QComboBox 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 pipeline(self): return None @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) 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()