# -*- coding: utf-8 -*- """ MainRouter —— 端到端模块化新架构的路由与调度壳 ================================================= 职责 ---- 1. **路由**：根据 ``ROUTES`` 表动态加载每个 step 的 view（前端皮囊） 和 service（后端大脑），左侧 ``QListWidget`` 导航，右侧 ``QStackedWidget`` 承载 view。 2. **解耦通讯**：view 不直接持有 main_router 引用；按钮点击通过 ``BaseView.dispatch_execute`` 沿父链回到本类 ``run_single_step``。 3. **后台执行**：内置 ``TaskWorker(QThread)``，把 service 调用丢进 子线程，避免阻塞 UI。 4. **日志** :底部 ``QTextEdit`` 集中输出路由与 service 结果。 运行方式 -------- :: # 方式 A：作为模块运行（推荐） cd D:/111/office/ZHLduijie/1.WQ/WQ_GUI python -m src.new.main_router # 方式 B：直接脚本运行 python src/new/main_router.py """ from __future__ import annotations import importlib import os import sys import traceback from pathlib import Path from PyQt5.QtCore import Qt, QThread, pyqtSignal from PyQt5.QtWidgets import ( QFileDialog, QHBoxLayout, QLabel, QListWidget, QMainWindow, QPushButton, QSplitter, QStackedWidget, QTextEdit, QVBoxLayout, QWidget, ) # 让 ``python src/new/main_router.py`` 直接运行时也能 import ``src.xxx`` _HERE = Path(__file__).resolve().parent # .../src/new _SRC_ROOT = _HERE.parent # .../src if str(_SRC_ROOT) not in sys.path: sys.path.insert(0, str(_SRC_ROOT)) from src.gui.styles import ModernStylesheet from src.new.views.placeholder_view import PlaceholderView # ========================================================================= # 路由注册表：每条路由对应一对 view + service # ========================================================================= ROUTES = [ { "id": "step1", "name": "1. 水域掩膜", "view_module": "src.new.views.step1_view", "view_class": "Step1View", "service_module": "src.new.services.step1_service", "service_func": "execute_step1", }, # -------- 占位路由（step2-step13 尚未迁移） -------- {"id": "step2", "name": "2. 数据准备", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step3", "name": "3. 耀斑检测", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step4", "name": "4. 耀斑去除", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step5", "name": "5. 水色指数反演", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step6", "name": "6. 指数波段匹配", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step7", "name": "7. 水质参数计算", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step8", "name": "8. ML 建模训练", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step9", "name": "9. 模型校正", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step10", "name": "10. 浓度反演", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step11", "name": "11. 空间克里金", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step12", "name": "12. 制图渲染", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, {"id": "step13", "name": "13. 结果导出", "view_module": "_placeholder", "view_class": "_placeholder", "service_module": "src.new.services.placeholder_service", "service_func": "execute_placeholder"}, ] # ========================================================================= # TaskWorker：QThread 后台执行器 # ========================================================================= class TaskWorker(QThread): """后台执行 ``service_func(config)`` 并把结果 emit 回主线程 为什么用 QThread 而不是 Thread：service_func 可能涉及 rasterio / matplotlib 等原生扩展，主线程与子线程的 Qt 事件循环隔离更稳。 """ finished_with_result = pyqtSignal(dict) def __init__(self, service_func, config: dict): super().__init__() self.service_func = service_func self.config = config def run(self): try: result = self.service_func(self.config) except Exception as e: # noqa: BLE001 result = { "status": "error", "output_path": None, "message": f"{type(e).__name__}: {e}", "traceback": traceback.format_exc(), } self.finished_with_result.emit(result) # ========================================================================= # MainRouter：QMainWindow 路由壳 # ========================================================================= class MainRouter(QMainWindow): """新架构的轻量路由主窗口""" def __init__(self): super().__init__() self.work_dir: str = "" self._views: dict = {} self._route_map: dict = {} self.worker: TaskWorker | None = None self.setWindowTitle("WQ_GUI 新架构 · 端到端模块化路由壳") self.resize(1280, 800) self.setStyleSheet(ModernStylesheet.get_main_stylesheet()) self._build_ui() self._build_routes() self._log("[Boot] MainRouter 初始化完成") self._log(f"[Boot] 已注册 {len(ROUTES)} 条路由（step1 真实 / step2-step13 占位）") # ------------------------------------------------------------------ # UI 布局：左侧导航 + 右侧 stacked + 底部日志 # ------------------------------------------------------------------ def _build_ui(self): central = QWidget() self.setCentralWidget(central) outer = QVBoxLayout(central) outer.setContentsMargins(8, 8, 8, 8) # ----- 顶部工具条：工作目录 + 设置按钮 ----- toolbar = QHBoxLayout() toolbar.addWidget(QLabel("工作目录:")) self.work_dir_label = QLabel("(未设置)") self.work_dir_label.setStyleSheet("color: #0055D4; font-weight: bold;") toolbar.addWidget(self.work_dir_label, 1) self.set_workdir_btn = QPushButton("设置工作目录...") self.set_workdir_btn.setStyleSheet(ModernStylesheet.get_button_stylesheet("normal")) self.set_workdir_btn.clicked.connect(self._on_set_workdir) toolbar.addWidget(self.set_workdir_btn) outer.addLayout(toolbar) # ----- 中部：左导航 + 右 stacked（用 QSplitter 可拖动） ----- splitter = QSplitter(Qt.Horizontal) self.nav_list = QListWidget() self.nav_list.setStyleSheet(ModernStylesheet.get_sidebar_stylesheet()) self.nav_list.setFixedWidth(200) self.nav_list.currentRowChanged.connect(self._on_nav_changed) splitter.addWidget(self.nav_list) self.stacked = QStackedWidget() splitter.addWidget(self.stacked) splitter.setStretchFactor(0, 0) splitter.setStretchFactor(1, 1) outer.addWidget(splitter, 1) # ----- 底部日志 ----- self.log_text = QTextEdit() self.log_text.setReadOnly(True) self.log_text.setStyleSheet("background-color: #1E1E1E; color: #DCDCDC; font-family: Consolas, monospace;") self.log_text.setFixedHeight(180) outer.addWidget(self.log_text) # ------------------------------------------------------------------ # 路由注册：实例化 view、填充 nav_list 与 stacked # ------------------------------------------------------------------ def _build_routes(self): for route in ROUTES: self._route_map[route["id"]] = route view = self._instantiate_view(route) self._views[route["id"]] = view self.stacked.addWidget(view) self.nav_list.addItem(route["name"]) if ROUTES: self.nav_list.setCurrentRow(0) def _instantiate_view(self, route: dict) -> QWidget: """根据路由表实例化 view：step1 走真实类，其余走 PlaceholderView""" if route["view_module"] != "_placeholder": try: module = importlib.import_module(route["view_module"]) cls = getattr(module, route["view_class"]) return cls(self.stacked) except Exception as e: # noqa: BLE001 # 真实 view 加载失败时降级为占位，并在日志区报错 self._log(f"[Router] 加载真实 view 失败 ({route['id']}): {e}") traceback.print_exc() # 占位 view：直接构造 return PlaceholderView(route["id"], route["name"], self.stacked) # ------------------------------------------------------------------ # 导航切换 # ------------------------------------------------------------------ def _on_nav_changed(self, row: int): if row < 0 or row >= len(ROUTES): return route = ROUTES[row] view = self._views[route["id"]] self.stacked.setCurrentWidget(view) self._log(f"[Nav] 切换到 {route['id']} → {route['name']}") # ------------------------------------------------------------------ # 工作目录推送：设置后广播给所有 view # ------------------------------------------------------------------ def _on_set_workdir(self): d = QFileDialog.getExistingDirectory(self, "选择工作目录", self.work_dir or "") if not d: return self.set_work_dir(d) def set_work_dir(self, work_dir: str): """设置工作目录并广播给所有 view""" if not work_dir: return self.work_dir = work_dir.replace("\\", "/") self.work_dir_label.setText(self.work_dir) self._log(f"[Router] 工作目录已设置: {self.work_dir}") # 推送给所有 view（让它们按需自动填充路径） for view in self._views.values(): if hasattr(view, "update_from_config"): view.update_from_config(self.work_dir, pipeline=None) # ------------------------------------------------------------------ # 核心调度：run_single_step（被 view.dispatch_execute 调用） # ------------------------------------------------------------------ def run_single_step(self, step_id: str, config_dict: dict): """从 view 接收 ``(step_id, config)`` → 调度 service 后台执行 1. 查路由表； 2. 注入 ``work_dir``（避免 view 重复感知）； 3. 启动 ``TaskWorker`` 在子线程运行 service； 4. 完成后 emit 回主线程并写日志。 """ route = self._route_map.get(step_id) if route is None: self._log(f"[Router] 未知 step_id: {step_id}") return # 注入 work_dir（仅当 view 未带且 main_router 已配置时） if self.work_dir and "work_dir" not in config_dict: config_dict = {**config_dict, "work_dir": self.work_dir} service_func = self._load_callable(route["service_module"], route["service_func"]) if service_func is None: return self._log(f"[Router] 收到 {step_id} 请求，配置: {config_dict}") # 启动后台线程（覆盖旧 worker 防泄漏） if self.worker and self.worker.isRunning(): self._log("[Router] 上一个任务尚未结束，丢弃新请求") return self.worker = TaskWorker(service_func, config_dict) self.worker.finished_with_result.connect(self._on_step_done) self.worker.start() def _load_callable(self, module_name: str, attr_name: str): """动态加载 ``module_name.attr_name``，失败时写日志并返回 None""" try: module = importlib.import_module(module_name) return getattr(module, attr_name) except Exception as e: # noqa: BLE001 self._log(f"[Router] 加载 service 失败 ({module_name}.{attr_name}): {e}") traceback.print_exc() return None def _on_step_done(self, result: dict): """TaskWorker 完成回调——按 status 分支写日志""" status = result.get("status", "unknown") message = result.get("message", "") output_path = result.get("output_path") mode = result.get("mode", "") if status == "completed": self._log(f"[Service✓] {message}（mode={mode}, output={output_path}）") elif status == "skipped": self._log(f"[Service↻] {message}（已复用历史结果）") elif status == "not_implemented": self._log(f"[Service…] {message}") else: # error self._log(f"[Service✗] {message}（mode={mode or 'n/a'}）") tb = result.get("traceback") if tb: self._log(tb) # ------------------------------------------------------------------ # 日志：所有日志统一走这里 # ------------------------------------------------------------------ def _log(self, msg: str): # 控制台 + 底部日志框 print(msg) self.log_text.append(msg) # ========================================================================= # 入口 # ========================================================================= def main(): from PyQt5.QtWidgets import QApplication app = QApplication(sys.argv) win = MainRouter() win.show() sys.exit(app.exec_()) if __name__ == "__main__": main()