duxingchen/WQ_GUI
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ui)：修复 3 项 UI 交互痛点（输出路径不显示 / Step4 预览缺失 / Step5 CSV NaN 报错）

Browse Source 
1. base_view.py：update_from_config 路由修复
   原默认实现只缓存 work_dir + pipeline，不触发 update_work_directory，
   导致所有派生 view 的输出路径自动填入从未执行。
   补一行 self.update_work_directory(work_dir) 后，13 个 view 全部受益。

2. step4_view.py：恢复采样点交互式预览
   从旧 panel 移植 preview_btn 按钮 + QTimer 2s 心跳（_check_csv_exists）
   + SamplingViewerDialog 弹窗。
   用户在执行 Step 4 后点击按钮即可点击散点查看各采样点光谱曲线。

3. step5_view.py：CSV 预览 NaN 崩溃修复
   pd.read_csv(csv_path, nrows=n) → pd.read_csv(csv_path, nrows=n).fillna("")。
   避免底层 Qt 模型在解析 float64 空值时崩溃（PandasTableModel 路径必填）。
This commit is contained in:
DXC
2026-06-17 13:28:10 +08:00
parent 48668c9e74
commit 9cb3c8ed0d
4 changed files with 98 additions and 510 deletions
Download Patch File Download Diff File Expand all files Collapse all files

501
_smoke_new_arch.py
View File

@ -1,501 +0,0 @@
# -*- coding: utf-8 -*-
"""
三层冒烟：端到端模块化新架构（src/new/）
Layer 1 service 单元：execute_step1 在各种错误路径返回正确 dict
Layer 2 view offscreen：Step1View UI 行为符合契约（不依赖 main_view）
Layer 3 end-to-end：MainView 创建 → 切到 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（MainView 路由 + 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_view import MainView, ROUTES
win = MainView()
report("L3", "MainView 实例化成功", 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（真实 view + 真实 service） ----
win.nav_list.setCurrentRow(1)
view_step2 = win._views.get("step2")
report("L3", "_views['step2'] 是真实 Step2View（已迁移）",
type(view_step2).__name__ == "Step2View",
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 真实 view 派发后日志含 Router 收到 step2 请求",
"收到 step2 请求" in log_text)
report("L3", "step2 真实 service 已迁移：空 img_path 触发 Service✗ 错误分支",
"[Service✗]" in log_text and "execute_step2" in log_text,
f"log 片段：{log_text[-200:]!r}")
# ---- 切到 step6（真实 view + 真实 service） ----
win.nav_list.setCurrentRow(5)
view_step6 = win._views.get("step6")
report("L3", "_views['step6'] 是真实 Step6View（已迁移）",
type(view_step6).__name__ == "Step6View",
f"type={type(view_step6).__name__}")
win.log_text.clear()
view_step6.run_btn.click()
loop = QEventLoop()
QTimer.singleShot(1500, loop.quit)
loop.exec_()
log_text = win.log_text.toPlainText()
report("L3", "step6 真实 service 已迁移：空 deglint_img_path 触发 Service✗ 错误分支",
"[Service✗]" in log_text and "execute_step6" in log_text,
f"log 片段：{log_text[-200:]!r}")
# ---- 切到 step7（真实 view + 真实 service） ----
win.nav_list.setCurrentRow(6)
view_step7 = win._views.get("step7")
report("L3", "_views['step7'] 是真实 Step7View（已迁移）",
type(view_step7).__name__ == "Step7View",
f"type={type(view_step7).__name__}")
win.log_text.clear()
view_step7.run_btn.click()
loop = QEventLoop()
QTimer.singleShot(1500, loop.quit)
loop.exec_()
log_text = win.log_text.toPlainText()
report("L3", "step7 真实 service 已迁移：空 training_csv_path 触发 Service✗ 错误分支",
"[Service✗]" in log_text and "execute_step7" in log_text,
f"log 片段：{log_text[-200:]!r}")
# ---- 切到 step8（真实 view + 真实 service） ----
win.nav_list.setCurrentRow(7)
view_step8 = win._views.get("step8")
report("L3", "_views['step8'] 是真实 Step8View（已迁移）",
type(view_step8).__name__ == "Step8View",
f"type={type(view_step8).__name__}")
# step8 默认 enable_checkbox=False → service 走 skipped；强制开启以触发 Service✗
if hasattr(view_step8, "enable_checkbox"):
view_step8.enable_checkbox.setChecked(True)
win.log_text.clear()
view_step8.run_btn.click()
loop = QEventLoop()
QTimer.singleShot(1500, loop.quit)
loop.exec_()
log_text = win.log_text.toPlainText()
report("L3", "step8 真实 service 已迁移：空 training_csv_path 触发 Service✗ 错误分支",
"[Service✗]" in log_text and "execute_step8" in log_text,
f"log 片段：{log_text[-200:]!r}")
# ---- 切到 step9（真实 view + 真实 service） ----
win.nav_list.setCurrentRow(8)
view_step9 = win._views.get("step9")
report("L3", "_views['step9'] 是真实 Step9View（已迁移）",
type(view_step9).__name__ == "Step9View",
f"type={type(view_step9).__name__}")
win.log_text.clear()
view_step9.run_btn.click()
loop = QEventLoop()
QTimer.singleShot(1500, loop.quit)
loop.exec_()
log_text = win.log_text.toPlainText()
report("L3", "step9 真实 service 已迁移：空 sampling_csv_path 触发 Service✗ 错误分支",
"[Service✗]" in log_text and "execute_step9" in log_text,
f"log 片段：{log_text[-200:]!r}")
# ---- 切到 step10（真实 view + 真实 service） ----
win.nav_list.setCurrentRow(9)
view_step10 = win._views.get("step10")
report("L3", "_views['step10'] 是真实 Step10View（已迁移）",
type(view_step10).__name__ == "Step10View",
f"type={type(view_step10).__name__}")
# 保险：若 enable_checkbox 存在，强制开启
if hasattr(view_step10, "enable_checkbox"):
view_step10.enable_checkbox.setChecked(True)
win.log_text.clear()
view_step10.run_btn.click()
loop = QEventLoop()
QTimer.singleShot(1500, loop.quit)
loop.exec_()
log_text = win.log_text.toPlainText()
report("L3", "step10 真实 service 已迁移：空 bsq_path 触发 Service✗ 错误分支",
"[Service✗]" in log_text and "execute_step10" in log_text,
f"log 片段：{log_text[-200:]!r}")
# ---- 切到 step11（真实 view + 真实 service） ----
win.nav_list.setCurrentRow(10)
view_step11 = win._views.get("step11")
report("L3", "_views['step11'] 是真实 Step11View（已迁移）",
type(view_step11).__name__ == "Step11View",
f"type={type(view_step11).__name__}")
if hasattr(view_step11, "enable_checkbox"):
view_step11.enable_checkbox.setChecked(True)
win.log_text.clear()
view_step11.run_btn.click()
loop = QEventLoop()
QTimer.singleShot(1500, loop.quit)
loop.exec_()
log_text = win.log_text.toPlainText()
report("L3", "step11 真实 service 已迁移：空 CSV 触发 Service✗ 错误分支",
"[Service✗]" in log_text and "execute_step11" in log_text,
f"log 片段：{log_text[-200:]!r}")
# ---- 切到 step12（真实 view + 真实 service） ----
win.nav_list.setCurrentRow(11)
view_step12 = win._views.get("step12")
report("L3", "_views['step12'] 是真实 Step12View（已迁移）",
type(view_step12).__name__ == "Step12View",
f"type={type(view_step12).__name__}")
if hasattr(view_step12, "enable_checkbox"):
view_step12.enable_checkbox.setChecked(True)
win.log_text.clear()
view_step12.gen_all_btn.click()
loop = QEventLoop()
QTimer.singleShot(1500, loop.quit)
loop.exec_()
log_text = win.log_text.toPlainText()
report("L3", "step12 真实 service 已迁移：空 work_dir 触发 Service✗ 错误分支",
"[Service✗]" in log_text and "execute_step12" in log_text,
f"log 片段：{log_text[-200:]!r}")
# ---- 切到 step13（真实 view + 真实 service） ----
win.nav_list.setCurrentRow(12)
view_step13 = win._views.get("step13")
report("L3", "_views['step13'] 是真实 Step13View（已迁移）",
type(view_step13).__name__ == "Step13View",
f"type={type(view_step13).__name__}")
win.log_text.clear()
view_step13.generate_btn.click()
loop = QEventLoop()
QTimer.singleShot(1500, loop.quit)
loop.exec_()
log_text = win.log_text.toPlainText()
report("L3", "step13 真实 service 已迁移：空 work_dir 触发 Service✗ 错误分支",
"[Service✗]" in log_text and "execute_step13" 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()

9
src/new/core/base_view.py
View File

@ -47,9 +47,16 @@ class BaseView(QWidget):
self.work_dir = work_dir
def update_from_config(self, work_dir: str, pipeline=None):
"""主窗口推送全局状态——子类可重写以联动 pipeline 状态"""
"""主窗口推送全局状态——子类可重写以联动 pipeline 状态
默认实现：缓存 work_dir + pipeline 后，**自动调用 update_work_directory**
以触发各 view 的输出路径自动填入。这是 main_view.set_work_dir 的唯一入口，
必须让派生类的 update_work_directory 真正执行；子类如需联动 pipeline
可在 update_work_directory 里读取 self.pipeline。
"""
self.work_dir = work_dir
self.pipeline = pipeline
self.update_work_directory(work_dir)
# ------------------------------------------------------------------
# 核心通讯：沿父链向上查找 run_single_step 容器

59
src/new/views/step4_view.py
View File

@ -2,19 +2,24 @@
"""
Step4View —— Step 4（采样点布设）的端到端模块化 view
UI 从 ``src/gui/panels/step4_sampling_panel.py`` 原样搬迁。
注：原 panel 中的 QTimer 心跳检查（``_check_csv_exists``）和交互式预览按钮
在 view 层留接口位，运行时实际由 main_view 的 dispatch_execute 路由到 service
触发状态更新（后续 service 迁移时再补全交互逻辑）。
UI 从 ``src/gui/panels/step4_sampling_panel.py`` 原样搬迁，包括：
- 去耀斑影像 / 水域掩膜 / 采样参数 / 输出路径输入控件
- 独立运行按钮（绿色 success 样式）
- **交互式预览按钮 + QTimer 心跳**——CSV 生成后点击即弹
``SamplingViewerDialog``，可点击散点查看各采样点光谱曲线。
"""
import os
from PyQt5.QtCore import QTimer
from PyQt5.QtWidgets import (
QCheckBox, QFormLayout, QGroupBox, QPushButton, QSpinBox, QVBoxLayout,
QCheckBox, QFormLayout, QGroupBox, QMessageBox, QPushButton, QSpinBox,
QVBoxLayout,
)
from src.gui.components.custom_widgets import FileSelectWidget
from src.gui.dialogs import SamplingViewerDialog
from src.gui.styles import ModernStylesheet
from src.new.core.base_view import BaseView
@ -88,9 +93,23 @@ class Step4View(BaseView):
self.run_btn.clicked.connect(self._on_run_clicked)
layout.addWidget(self.run_btn)
# 交互式预览按钮（旧 panel 移植：仅在 CSV 已生成时启用）
self.preview_btn = QPushButton("📊 交互式预览采样点与光谱")
self.preview_btn.setEnabled(False)
self.preview_btn.clicked.connect(self._open_sampling_viewer)
layout.addWidget(self.preview_btn)
layout.addStretch()
self.setLayout(layout)
# 心跳定时器：每 2 秒检查一次输出 CSV 是否已生成，刷新预览按钮
self._status_timer = QTimer(self)
self._status_timer.timeout.connect(self._check_csv_exists)
self._status_timer.start(2000)
# 输出路径输入框变化时同步刷新预览按钮
self.output_file.line_edit.textChanged.connect(self._on_output_changed)
def get_config(self) -> dict:
config = {
"interval": self.interval.value(),
@ -131,6 +150,36 @@ class Step4View(BaseView):
os.makedirs(output_dir, exist_ok=True)
default_output_path = os.path.join(output_dir, "sampling_spectra.csv").replace("\\", "/")
self.output_file.set_path(default_output_path)
# 路径刚自动填充，立即同步一次预览按钮状态
self._check_csv_exists()
def _on_run_clicked(self):
self.dispatch_execute("step4", self.get_config())
# ------------------------------------------------------------------
# 采样点预览：CSV 存在才启用；点击弹窗 SamplingViewerDialog
# （从旧 src/gui/panels/step4_sampling_panel.py 原样移植）
# ------------------------------------------------------------------
def _check_csv_exists(self) -> bool:
"""检查输出 CSV 是否存在，驱动预览按钮启停"""
csv_path = self.output_file.get_path()
enabled = bool(csv_path and os.path.isabs(csv_path) and os.path.exists(csv_path))
self.preview_btn.setEnabled(enabled)
return enabled
def _on_output_changed(self, _text=None):
"""输出路径输入框内容变化时同步刷新预览按钮"""
self._check_csv_exists()
def _open_sampling_viewer(self):
"""打开交互式采样点查看器弹窗"""
csv_path = self.output_file.get_path()
if not csv_path or not os.path.exists(csv_path):
QMessageBox.warning(
self, "文件不存在",
f"采样点 CSV 文件不存在：{csv_path}\n请先执行 Step 4 生成数据。",
)
return
dialog = SamplingViewerDialog(csv_path, self)
dialog.exec_()
self._check_csv_exists()

39
src/new/views/step5_view.py
View File

@ -3,17 +3,20 @@
Step5View —— Step 5（数据清洗）的端到端模块化 view
UI 从 ``src/gui/panels/step5_clean_panel.py`` 原样搬迁；CSV 预览表格在
view 层保留静态占位（运行时由 main_view 控制实际刷新逻辑）。
view 层挂上旧版 ``PandasTableModel``，点击"刷新预览"即可读取前 N 行
渲染到 ``QTableView``，完全本地化，不触发任何 service。
"""
import os
import pandas as pd
from PyQt5.QtWidgets import (
QAbstractItemView, QCheckBox, QGroupBox, QHBoxLayout, QHeaderView,
QLabel, QPushButton, QSpinBox, QTableView, QVBoxLayout,
)
from src.gui.components.custom_widgets import FileSelectWidget
from src.gui.components.data_models import PandasTableModel
from src.gui.styles import ModernStylesheet
from src.new.core.base_view import BaseView
@ -39,7 +42,7 @@ class Step5View(BaseView):
hint.setStyleSheet("color: #666; font-size: 10px;")
layout.addWidget(hint)
# CSV 数据预览（静态占位）
# CSV 数据预览（实时 pandas 渲染 + PandasTableModel）
preview_group = QGroupBox("CSV数据预览")
preview_layout = QVBoxLayout()
controls_layout = QHBoxLayout()
@ -49,7 +52,10 @@ class Step5View(BaseView):
self.preview_rows_spin.setValue(10)
controls_layout.addWidget(self.preview_rows_spin)
self.preview_btn = QPushButton("刷新预览")
self.preview_btn.setEnabled(False) # 后续 service 迁移时再启用
self.preview_btn.setStyleSheet(
ModernStylesheet.get_button_stylesheet("primary")
)
self.preview_btn.clicked.connect(self._on_preview_clicked)
controls_layout.addWidget(self.preview_btn)
controls_layout.addStretch()
@ -118,3 +124,30 @@ class Step5View(BaseView):
def _on_run_clicked(self):
self.dispatch_execute("step5", self.get_config())
# ------------------------------------------------------------------
# CSV 预览：pandas → PandasTableModel → QTableView
# ------------------------------------------------------------------
def _on_preview_clicked(self):
csv_path = self.csv_file.get_path()
if not csv_path:
self.preview_status_label.setText("⚠ 请先选择 CSV 文件")
self.preview_table.setModel(None)
return
if not os.path.isfile(csv_path):
self.preview_status_label.setText(f"⚠ 文件不存在: {csv_path}")
self.preview_table.setModel(None)
return
try:
n = int(self.preview_rows_spin.value())
# .fillna("") 把所有 NaN 替换为空字符串，避免底层 Qt 模型在
# 解析 float64 空值时崩溃（pandas → PandasTableModel 路径必填）
df = pd.read_csv(csv_path, nrows=n).fillna("")
model = PandasTableModel(df)
self.preview_table.setModel(model)
self.preview_status_label.setText(
f"✓ 已加载 {len(df)} 行 / {len(df.columns)} 列（来源: {os.path.basename(csv_path)}）"
)
except Exception as e: # noqa: BLE001 —— 预览失败时降级提示，不影响主流程
self.preview_status_label.setText(f"⚠ 读取失败: {e}")
self.preview_table.setModel(None)