README_new_arch.md + _smoke_new_arch.py：端到端新架构运行文档与三层冒烟（service/view/e2e 共 54 项断言）
This commit is contained in:
83
README_new_arch.md
Normal file
83
README_new_arch.md
Normal file
@ -0,0 +1,83 @@
|
||||
# 端到端模块化新架构（src/new/）
|
||||
|
||||
## 目录结构
|
||||
|
||||
```
|
||||
src/new/
|
||||
├── __init__.py
|
||||
├── core/
|
||||
│ ├── __init__.py
|
||||
│ └── base_view.py # 基础通讯接口（继承 QWidget + dispatch_execute）
|
||||
├── services/ # 独立后端大脑
|
||||
│ ├── __init__.py
|
||||
│ ├── step1_service.py # Step 1 真实服务（execute_step1）
|
||||
│ └── placeholder_service.py # step2-step13 占位服务
|
||||
├── views/ # 独立前端皮囊
|
||||
│ ├── __init__.py
|
||||
│ ├── step1_view.py # Step 1 真实视图（继承 BaseView）
|
||||
│ └── placeholder_view.py # step2-step13 占位视图
|
||||
└── main_router.py # 路由与调度壳（QMainWindow + QThread）
|
||||
```
|
||||
|
||||
## 端到端调用链
|
||||
|
||||
```
|
||||
Step1View._on_run_clicked (绿色按钮)
|
||||
│ self.dispatch_execute("step1", self.get_config())
|
||||
▼
|
||||
BaseView.dispatch_execute (沿父链上溯)
|
||||
│ ancestor.run_single_step(step_id, config)
|
||||
▼
|
||||
MainRouter.run_single_step (查 ROUTES 表 → 注入 work_dir)
|
||||
│ TaskWorker(service_func, config).start()
|
||||
▼
|
||||
services.step1_service.execute_step1(config)
|
||||
│ 调 WaterMaskStep.run(...) → 包装成 dict 返回
|
||||
▼
|
||||
MainRouter._on_step_done (按 status 写日志)
|
||||
```
|
||||
|
||||
## 运行验证
|
||||
|
||||
### 1. 三层冒烟（推荐先跑）
|
||||
|
||||
```cmd
|
||||
cd D:\111\office\ZHLduijie\1.WQ\WQ_GUI
|
||||
python _smoke_new_arch.py
|
||||
```
|
||||
|
||||
预期输出 `汇总：54/54 通过`。
|
||||
|
||||
### 2. 启动路由主窗口
|
||||
|
||||
```cmd
|
||||
cd D:\111\office\ZHLduijie\1.WQ\WQ_GUI
|
||||
python -m src.new.main_router
|
||||
```
|
||||
|
||||
或：
|
||||
|
||||
```cmd
|
||||
python src\new\main_router.py
|
||||
```
|
||||
|
||||
启动后：
|
||||
|
||||
* 左侧 `QListWidget` 显示 13 个 step（step1 真实，其余占位）
|
||||
* 点击 `执行 Step 1: 水域掩膜` → 绿色按钮 → `dispatch_execute`
|
||||
* 底部 `QTextEdit` 实时打印 `[Router]` / `[Service]` 日志
|
||||
|
||||
## 关键设计原则
|
||||
|
||||
1. **view 零业务**：`src/new/views/*.py` 绝不 import 任何 `src/core/`、`src/services/`
|
||||
2. **service 零 PyQt**：`src/new/services/*.py` 不 import 任何 PyQt、不读写全局
|
||||
3. **唯一跨界通道**：`BaseView.dispatch_execute` 把 (step_id, config) 推给主窗口
|
||||
4. **后台执行不阻塞 UI**：`TaskWorker(QThread)` 子线程跑 service
|
||||
5. **错误兜底**：service 任何异常都被 TaskWorker 捕获并转成 `{status: "error", ...}`
|
||||
|
||||
## 当前状态
|
||||
|
||||
| step | view | service | 状态 |
|
||||
|--------|---------------------|------------------------|---------------------|
|
||||
| step1 | `Step1View` 真实 | `execute_step1` 真实 | ✅ 已迁移 |
|
||||
| step2-13 | `PlaceholderView` | `execute_placeholder` | 🚧 占位待迁移 |
|
||||
343
_smoke_new_arch.py
Normal file
343
_smoke_new_arch.py
Normal file
@ -0,0 +1,343 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
三层冒烟：端到端模块化新架构（src/new/）
|
||||
|
||||
Layer 1 service 单元：execute_step1 在各种错误路径返回正确 dict
|
||||
Layer 2 view offscreen：Step1View UI 行为符合契约（不依赖 main_router）
|
||||
Layer 3 end-to-end：MainRouter 创建 → 切到 step1 → 点击 → 日志验证
|
||||
|
||||
运行方式（项目根）::
|
||||
|
||||
python _smoke_new_arch.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
# 把项目根加进 sys.path（脚本可能直接从根目录运行）
|
||||
_PROJECT_ROOT = Path(__file__).resolve().parent
|
||||
if str(_PROJECT_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_PROJECT_ROOT))
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
# cmd.exe 默认 GBK 会让中文 / emoji 爆 UnicodeEncodeError
|
||||
# 强制 stdout / stderr 走 utf-8（仅在 stdout 有 reconfigure 接口时生效）
|
||||
for _stream in (sys.stdout, sys.stderr):
|
||||
if hasattr(_stream, "reconfigure"):
|
||||
try:
|
||||
_stream.reconfigure(encoding="utf-8", errors="replace")
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
PASS = "✅"
|
||||
FAIL = "❌"
|
||||
results: list[tuple[str, bool, str]] = []
|
||||
|
||||
|
||||
def report(layer: str, name: str, ok: bool, detail: str = ""):
|
||||
tag = PASS if ok else FAIL
|
||||
results.append((layer, name, ok, detail))
|
||||
print(f"{tag} [{layer}] {name}" + (f" —— {detail}" if detail else ""))
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# Layer 1: service 单元测试
|
||||
# =========================================================================
|
||||
def smoke_service():
|
||||
print("\n" + "=" * 70)
|
||||
print("Layer 1 —— service 单元测试（src.new.services.step1_service）")
|
||||
print("=" * 70)
|
||||
|
||||
from src.new.services.step1_service import execute_step1, _resolve_mode
|
||||
|
||||
# ---- _resolve_mode 纯逻辑 ----
|
||||
report("L1", "_resolve_mode use_ndwi=True",
|
||||
_resolve_mode({"use_ndwi": True, "mask_path": ""}) == "ndwi",
|
||||
"→ ndwi")
|
||||
report("L1", "_resolve_mode mask=.shp",
|
||||
_resolve_mode({"use_ndwi": False, "mask_path": "x.shp"}) == "shp_rasterize",
|
||||
"→ shp_rasterize")
|
||||
report("L1", "_resolve_mode mask=.dat",
|
||||
_resolve_mode({"use_ndwi": False, "mask_path": "x.dat"}) == "existing_mask",
|
||||
"→ existing_mask")
|
||||
report("L1", "_resolve_mode mask=None",
|
||||
_resolve_mode({"use_ndwi": False, "mask_path": None}) == "existing_mask",
|
||||
"→ existing_mask (默认)")
|
||||
|
||||
# ---- execute_step1 错误路径 ----
|
||||
# 路径 1：use_ndwi=True 但 img_path 不存在
|
||||
r = execute_step1({
|
||||
"use_ndwi": True,
|
||||
"img_path": "D:/__no_such_image__.dat",
|
||||
"ndwi_threshold": 0.4,
|
||||
})
|
||||
report("L1", "execute_step1 NDWI + img_path 不存在 → status=error",
|
||||
r.get("status") == "error",
|
||||
f"status={r.get('status')}, message={r.get('message')!r}")
|
||||
report("L1", "execute_step1 NDWI + img_path 不存在 → mode=ndwi",
|
||||
r.get("mode") == "ndwi",
|
||||
f"mode={r.get('mode')}")
|
||||
|
||||
# 路径 2：use_ndwi=False 但 mask_path 不存在
|
||||
r = execute_step1({
|
||||
"use_ndwi": False,
|
||||
"mask_path": "D:/__no_such_mask__.dat",
|
||||
"img_path": "D:/__no_such_image__.dat",
|
||||
})
|
||||
report("L1", "execute_step1 existing mask 不存在 → status=error",
|
||||
r.get("status") == "error",
|
||||
f"status={r.get('status')}, message={r.get('message')!r}")
|
||||
report("L1", "execute_step1 existing mask 不存在 → mode=existing_mask",
|
||||
r.get("mode") == "existing_mask",
|
||||
f"mode={r.get('mode')}")
|
||||
|
||||
# 路径 3：use_ndwi=False + shp 但 img_path 缺
|
||||
r = execute_step1({
|
||||
"use_ndwi": False,
|
||||
"mask_path": "D:/__no_such_mask__.shp",
|
||||
# 故意不传 img_path
|
||||
})
|
||||
report("L1", "execute_step1 shp 但 img_path 缺 → status=error",
|
||||
r.get("status") == "error",
|
||||
f"status={r.get('status')}, message={r.get('message')!r}")
|
||||
report("L1", "execute_step1 shp 但 img_path 缺 → mode=shp_rasterize",
|
||||
r.get("mode") == "shp_rasterize",
|
||||
f"mode={r.get('mode')}")
|
||||
|
||||
# 路径 4：work_dir 注入后不会被强制转 str 报错
|
||||
r = execute_step1({
|
||||
"use_ndwi": True,
|
||||
"img_path": "D:/__no_such_image__.dat",
|
||||
"work_dir": "D:/workspace",
|
||||
})
|
||||
report("L1", "execute_step1 接受 work_dir 注入（不会抛）",
|
||||
isinstance(r, dict) and "status" in r,
|
||||
f"status={r.get('status')}")
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# Layer 2: view offscreen 测试
|
||||
# =========================================================================
|
||||
def smoke_view():
|
||||
print("\n" + "=" * 70)
|
||||
print("Layer 2 —— view offscreen 测试（src.new.views.step1_view）")
|
||||
print("=" * 70)
|
||||
|
||||
from PyQt5.QtWidgets import QApplication
|
||||
app = QApplication.instance() or QApplication(sys.argv)
|
||||
|
||||
from src.new.views.step1_view import Step1View
|
||||
from src.new.core.base_view import BaseView
|
||||
|
||||
# ---- 类契约 ----
|
||||
report("L2", "Step1View 继承 BaseView",
|
||||
issubclass(Step1View, BaseView),
|
||||
"MRO[0] == 'Step1View', MRO[1] == 'BaseView'")
|
||||
|
||||
# ---- 实例化 ----
|
||||
view = Step1View()
|
||||
report("L2", "Step1View 实例化成功（无异常）",
|
||||
view is not None,
|
||||
f"view={view!r}")
|
||||
|
||||
# ---- 9 个子控件非空 ----
|
||||
attrs = [
|
||||
"use_existing_radio", "use_ndwi_radio",
|
||||
"mask_file", "img_file",
|
||||
"ndwi_group", "ndwi_threshold",
|
||||
"output_file", "hint" if hasattr(view, "hint") else "enable_checkbox",
|
||||
"enable_checkbox", "run_btn",
|
||||
]
|
||||
for a in attrs:
|
||||
if a == "hint" and not hasattr(view, a):
|
||||
continue
|
||||
report("L2", f"子控件 {a} 非空", getattr(view, a) is not None)
|
||||
|
||||
# ---- 默认状态：existing 模式 ----
|
||||
report("L2", "默认 existing 模式：mask_file.isHidden() == False",
|
||||
view.mask_file.isHidden() is False)
|
||||
report("L2", "默认 existing 模式：ndwi_group.isHidden() == True",
|
||||
view.ndwi_group.isHidden() is True)
|
||||
report("L2", "默认 existing 模式：output_file.isHidden() == True",
|
||||
view.output_file.isHidden() is True)
|
||||
|
||||
# ---- 切换到 NDWI 模式 ----
|
||||
view.use_ndwi_radio.setChecked(True)
|
||||
report("L2", "NDWI 模式：mask_file.isHidden() == True",
|
||||
view.mask_file.isHidden() is True)
|
||||
report("L2", "NDWI 模式：ndwi_group.isHidden() == False",
|
||||
view.ndwi_group.isHidden() is False)
|
||||
report("L2", "NDWI 模式：output_file.isHidden() == False",
|
||||
view.output_file.isHidden() is False)
|
||||
|
||||
# ---- update_work_directory 触发自动填充（NDWI 模式下）----
|
||||
tmpdir = tempfile.mkdtemp(prefix="wqgui_smoke_").replace("\\", "/")
|
||||
view.update_work_directory(tmpdir)
|
||||
expected = f"{tmpdir}/1_water_mask/water_mask_out.dat"
|
||||
actual = view.output_file.get_path()
|
||||
report("L2", "NDWI 模式 + update_work_directory 自动填输出路径",
|
||||
actual == expected,
|
||||
f"期望={expected!r}, 实际={actual!r}")
|
||||
|
||||
# ---- get_config 双模式断言 ----
|
||||
# NDWI 模式
|
||||
cfg_ndwi = view.get_config()
|
||||
report("L2", "get_config NDWI 模式 use_ndwi=True",
|
||||
cfg_ndwi.get("use_ndwi") is True)
|
||||
report("L2", "get_config NDWI 模式 mask_path is None",
|
||||
cfg_ndwi.get("mask_path") is None)
|
||||
report("L2", "get_config NDWI 模式 output_path 已填",
|
||||
cfg_ndwi.get("output_path") == expected,
|
||||
f"output_path={cfg_ndwi.get('output_path')!r}")
|
||||
|
||||
# existing 模式
|
||||
view.use_existing_radio.setChecked(True)
|
||||
cfg_existing = view.get_config()
|
||||
report("L2", "get_config existing 模式 use_ndwi=False",
|
||||
cfg_existing.get("use_ndwi") is False)
|
||||
report("L2", "get_config existing 模式 mask_path 来自 mask_file（默认空串）",
|
||||
cfg_existing.get("mask_path") == "",
|
||||
f"mask_path={cfg_existing.get('mask_path')!r}")
|
||||
report("L2", "get_config existing 模式 output_path is None",
|
||||
cfg_existing.get("output_path") is None)
|
||||
|
||||
# ---- set_config 回灌 ----
|
||||
view.set_config({
|
||||
"use_ndwi": True,
|
||||
"ndwi_threshold": 0.55,
|
||||
"img_path": "D:/test/img.dat",
|
||||
"output_path": "D:/test/mask.dat",
|
||||
})
|
||||
report("L2", "set_config 后 use_ndwi=True",
|
||||
view.use_ndwi_radio.isChecked() is True)
|
||||
report("L2", "set_config 后 ndwi_threshold=0.55",
|
||||
abs(view.ndwi_threshold.value() - 0.55) < 1e-6,
|
||||
f"实际={view.ndwi_threshold.value()}")
|
||||
report("L2", "set_config 后 img_file 路径",
|
||||
view.img_file.get_path() == "D:/test/img.dat",
|
||||
f"实际={view.img_file.get_path()!r}")
|
||||
|
||||
# ---- update_from_config 双字段 ----
|
||||
view.update_from_config(tmpdir, pipeline={"k": "v"})
|
||||
report("L2", "update_from_config work_dir 已缓存",
|
||||
view.work_dir == tmpdir,
|
||||
f"work_dir={view.work_dir!r}")
|
||||
report("L2", "update_from_config pipeline 已缓存",
|
||||
view.pipeline == {"k": "v"},
|
||||
f"pipeline={view.pipeline!r}")
|
||||
|
||||
# ---- _on_run_clicked 父链缺失时 RuntimeError 预期 ----
|
||||
# 此时 view.parent() 是 None（root 级 QWidget），必然找不到 run_single_step
|
||||
try:
|
||||
view._on_run_clicked()
|
||||
report("L2", "_on_run_clicked 父链缺失应抛 RuntimeError", False, "未抛异常")
|
||||
except RuntimeError as e:
|
||||
report("L2", "_on_run_clicked 父链缺失抛 RuntimeError", True, str(e)[:80])
|
||||
|
||||
view.deleteLater()
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# Layer 3: end-to-end 测试
|
||||
# =========================================================================
|
||||
def smoke_e2e():
|
||||
print("\n" + "=" * 70)
|
||||
print("Layer 3 —— end-to-end（MainRouter 路由 + TaskWorker 调度）")
|
||||
print("=" * 70)
|
||||
|
||||
from PyQt5.QtCore import QEventLoop, QTimer
|
||||
from PyQt5.QtWidgets import QApplication
|
||||
app = QApplication.instance() or QApplication(sys.argv)
|
||||
|
||||
from src.new.main_router import MainRouter, ROUTES
|
||||
|
||||
win = MainRouter()
|
||||
report("L3", "MainRouter 实例化成功", win is not None)
|
||||
report("L3", "ROUTES 共 13 条", len(ROUTES) == 13, f"实际={len(ROUTES)}")
|
||||
report("L3", "nav_list 项数 == 13", win.nav_list.count() == 13,
|
||||
f"count={win.nav_list.count()}")
|
||||
report("L3", "stacked 子部件数 == 13", win.stacked.count() == 13,
|
||||
f"count={win.stacked.count()}")
|
||||
report("L3", "默认 currentRow == 0（step1）", win.nav_list.currentRow() == 0)
|
||||
|
||||
# ---- 切到 step1 ----
|
||||
win.nav_list.setCurrentRow(0)
|
||||
view_step1 = win._views.get("step1")
|
||||
report("L3", "_views['step1'] 是真实 Step1View",
|
||||
type(view_step1).__name__ == "Step1View",
|
||||
f"type={type(view_step1).__name__}")
|
||||
|
||||
# ---- 模拟点击 step1.run_btn（NDWI 模式 + 不存在的 img_path → status=error）----
|
||||
view_step1.use_ndwi_radio.setChecked(True)
|
||||
view_step1.img_file.set_path("D:/__no_such_img__.dat")
|
||||
view_step1.run_btn.click()
|
||||
|
||||
# 等 TaskWorker 跑完（offscreen + QThread 信号回到主线程）
|
||||
loop = QEventLoop()
|
||||
QTimer.singleShot(2000, loop.quit)
|
||||
loop.exec_()
|
||||
|
||||
log_text = win.log_text.toPlainText()
|
||||
report("L3", "日志含 Router 收到 step1 请求",
|
||||
"收到 step1 请求" in log_text,
|
||||
"→ 找 ['Router] 收到 step1 请求']")
|
||||
report("L3", "日志含 Service✗ 错误状态",
|
||||
"[Service✗]" in log_text,
|
||||
"→ 找 ['[Service✗]']")
|
||||
report("L3", "日志含 mode=ndwi 标记",
|
||||
"mode=ndwi" in log_text,
|
||||
"→ 找 ['mode=ndwi']")
|
||||
|
||||
# ---- 切到 step2 占位 ----
|
||||
win.nav_list.setCurrentRow(1)
|
||||
view_step2 = win._views.get("step2")
|
||||
report("L3", "_views['step2'] 是占位 PlaceholderView",
|
||||
type(view_step2).__name__ == "PlaceholderView",
|
||||
f"type={type(view_step2).__name__}")
|
||||
|
||||
# 清掉旧日志，重置 worker
|
||||
win.log_text.clear()
|
||||
view_step2.run_btn.click()
|
||||
loop = QEventLoop()
|
||||
QTimer.singleShot(1500, loop.quit)
|
||||
loop.exec_()
|
||||
|
||||
log_text = win.log_text.toPlainText()
|
||||
report("L3", "step2 占位日志含 Router 收到 step2 请求",
|
||||
"收到 step2 请求" in log_text)
|
||||
report("L3", "step2 占位日志含 Service… not_implemented",
|
||||
"not_implemented" in log_text,
|
||||
f"log 片段：{log_text[-200:]!r}")
|
||||
|
||||
win.close()
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# 入口
|
||||
# =========================================================================
|
||||
def main():
|
||||
smoke_service()
|
||||
smoke_view()
|
||||
smoke_e2e()
|
||||
|
||||
print("\n" + "=" * 70)
|
||||
total = len(results)
|
||||
passed = sum(1 for r in results if r[2])
|
||||
print(f"汇总：{passed}/{total} 通过")
|
||||
if passed != total:
|
||||
print("\n失败明细：")
|
||||
for layer, name, ok, detail in results:
|
||||
if not ok:
|
||||
print(f" ❌ [{layer}] {name} —— {detail}")
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("🎉 全部通过")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user