229 lines
11 KiB
Python
229 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Smoke test: step8_ml_train_panel + step9_ml_predict_panel 的新 CSV→QComboBox 联动
|
||
& update_from_config 多源文件优先回退 逻辑。
|
||
|
||
不真正 import 整个 panel（避免 osgeo 重依赖），只 import 类本身，验证：
|
||
1) Step8MlTrainPanel.feature_start 变成 QComboBox
|
||
2) Step8._on_training_csv_changed 正确读 CSV 表头 + 默认选中 374.285
|
||
3) Step8._resolve_training_csv_from_workdir 优先级（Step7 > 10_WaterIndex > Step6 > Step7 兜底）
|
||
4) Step8.get_config / set_config / get_training_params 用 QComboBox API 不崩
|
||
5) Step9._resolve_latest_wqi_test_csv 找到正确的 WQI 测试集
|
||
6) Step9.update_from_config 优先级（WQI CSV > Step4 raw）
|
||
|
||
用 offscreen 模式避免弹窗，完整可重复跑。
|
||
"""
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
# 避免 panel 顶部 osgeo 导入阻塞
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
# 让 src 目录可 import
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
|
||
import pandas as pd
|
||
import numpy as np
|
||
from PyQt5.QtWidgets import QApplication
|
||
|
||
# 必须先 create QApplication 才能 new widget
|
||
app = QApplication.instance() or QApplication(sys.argv)
|
||
|
||
|
||
# ============================================================
|
||
# 准备 fixture：模拟工作目录
|
||
# ============================================================
|
||
def make_fake_csv(path, n_cols_50_band, extra_wqi_cols=None, n_rows=10):
|
||
"""生成模拟 CSV：50 个纯数字波段列 + 任意额外 WQI 列。"""
|
||
band_cols = [374.285 + i for i in range(n_cols_50_band)]
|
||
# 浮点列名会被 pandas 自动转成 374.285 / 374.285001 / ...
|
||
cols = band_cols + (extra_wqi_cols or [])
|
||
data = np.random.RandomState(42).rand(n_rows, len(cols))
|
||
df = pd.DataFrame(data, columns=cols)
|
||
df.to_csv(path, index=False)
|
||
return df
|
||
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
work_dir = tmp
|
||
print(f"[fixture] work_dir = {work_dir}")
|
||
|
||
# 1) Step 6 原始光谱
|
||
os.makedirs(os.path.join(work_dir, "6_Spectral_Feature_Extraction"), exist_ok=True)
|
||
step6_csv = os.path.join(work_dir, "6_Spectral_Feature_Extraction", "training_spectra.csv")
|
||
make_fake_csv(step6_csv, 50)
|
||
|
||
# 2) Step 7 WQI 增强版
|
||
os.makedirs(os.path.join(work_dir, "7_Water_Quality_Indices"), exist_ok=True)
|
||
step7_csv = os.path.join(work_dir, "7_Water_Quality_Indices", "training_spectra_indices.csv")
|
||
wqi_names = [f"WQI_Chla_{i}" for i in range(10)] + [f"WQI_TP_{i}" for i in range(5)]
|
||
make_fake_csv(step7_csv, 50, wqi_names)
|
||
|
||
# 3) 10_WaterIndex_CSV 下的 training csv（较旧 mtime）
|
||
os.makedirs(os.path.join(work_dir, "10_WaterIndex_CSV"), exist_ok=True)
|
||
step10_csv = os.path.join(work_dir, "10_WaterIndex_CSV", "training_watercolor_summary.csv")
|
||
make_fake_csv(step10_csv, 50, wqi_names)
|
||
# 把 mtime 调成 3 年前，确认 7_Water_Quality_Indices 那个更新
|
||
old_time = (pd.Timestamp.now() - pd.Timedelta(days=365 * 3)).timestamp()
|
||
os.utime(step10_csv, (old_time, old_time))
|
||
|
||
# 4) Step 4 raw sampling
|
||
os.makedirs(os.path.join(work_dir, "4_sampling"), exist_ok=True)
|
||
step4_csv = os.path.join(work_dir, "4_sampling", "sampling_spectra.csv")
|
||
make_fake_csv(step4_csv, 50)
|
||
|
||
# 5) WQI 测试集（10_WaterIndex_CSV 下另一个 mtime 较新的 csv）
|
||
wqi_test_csv = os.path.join(work_dir, "10_WaterIndex_CSV", "sampling_watercolor_summary.csv")
|
||
make_fake_csv(wqi_test_csv, 50, wqi_names)
|
||
# 设为 1 小时前（mtime 比 step7_csv 新）
|
||
new_time = (pd.Timestamp.now() - pd.Timedelta(hours=1)).timestamp()
|
||
os.utime(wqi_test_csv, (new_time, new_time))
|
||
|
||
# 6) 9_ML_Prediction 输出目录
|
||
os.makedirs(os.path.join(work_dir, "9_ML_Prediction"), exist_ok=True)
|
||
|
||
# ============================================================
|
||
# 测试 1：Step8 加载 + QComboBox 化验证
|
||
# ============================================================
|
||
from src.gui.panels.step8_ml_train_panel import Step8MlTrainPanel
|
||
from PyQt5.QtWidgets import QComboBox
|
||
|
||
panel8 = Step8MlTrainPanel()
|
||
assert isinstance(panel8.feature_start, QComboBox), \
|
||
f"FAIL: feature_start 应为 QComboBox，实际为 {type(panel8.feature_start)}"
|
||
print(f"[OK] Step8.feature_start 类型: {type(panel8.feature_start).__name__}")
|
||
|
||
# ============================================================
|
||
# 测试 2：CSV 变更槽 — 模拟用户选 WQI CSV，自动填表头
|
||
# ============================================================
|
||
panel8._on_training_csv_changed(step7_csv)
|
||
assert panel8.feature_start.count() > 0
|
||
head_text = panel8.feature_start.currentText()
|
||
print(f"[OK] Step8 选中 step7_csv 后，feature_start.currentText() = {head_text!r}")
|
||
# 默认选中：第一个纯数字波段列 "374.285"
|
||
assert head_text.replace('.', '').lstrip('-').isdigit(), \
|
||
f"FAIL: 默认应选数字波段，实际 = {head_text!r}"
|
||
print(f"[OK] 默认选中数字波段列: {head_text!r}")
|
||
|
||
# 验证前 50 列已填充
|
||
n_items = panel8.feature_start.count()
|
||
assert n_items >= 50, f"FAIL: 应至少 50 项，实际 {n_items}"
|
||
print(f"[OK] Step8.feature_start 填充了 {n_items} 个候选列（前 50 列）")
|
||
|
||
# ============================================================
|
||
# 测试 3：CSV 不存在时，combo 回到占位项
|
||
# ============================================================
|
||
panel8._on_training_csv_changed("/nonexistent/path.csv")
|
||
assert panel8.feature_start.count() == 1
|
||
assert "请先选择训练 CSV" in panel8.feature_start.itemText(0)
|
||
print(f"[OK] Step8 CSV 不存在时，combo 回退到占位项: {panel8.feature_start.itemText(0)!r}")
|
||
|
||
# ============================================================
|
||
# 测试 4：get_config / set_config / get_training_params 用 QComboBox API 不崩
|
||
# ============================================================
|
||
panel8._on_training_csv_changed(step7_csv)
|
||
cfg = panel8.get_config()
|
||
assert 'feature_start_column' in cfg
|
||
assert isinstance(cfg['feature_start_column'], str)
|
||
assert cfg['feature_start_column'].replace('.', '').lstrip('-').isdigit()
|
||
print(f"[OK] Step8.get_config()['feature_start_column'] = {cfg['feature_start_column']!r}")
|
||
|
||
# set_config 试一下（先选个不存在的列名，应该走兜底 addItem 分支）
|
||
panel8.set_config({'feature_start_column': '999.999_unknown'})
|
||
# 重新触发 CSV 加载，验证回放不污染
|
||
panel8._on_training_csv_changed(step7_csv)
|
||
panel8.set_config({'feature_start_column': '374.285'})
|
||
assert panel8.feature_start.currentText() == '374.285'
|
||
print(f"[OK] Step8.set_config() 正确把 374.285 选中")
|
||
|
||
# get_training_params
|
||
params = panel8.get_training_params()
|
||
assert 'feature_start' in params
|
||
assert isinstance(params['feature_start'], float)
|
||
print(f"[OK] Step8.get_training_params()['feature_start'] = {params['feature_start']}")
|
||
|
||
# ============================================================
|
||
# 测试 5：_resolve_training_csv_from_workdir 优先级
|
||
# ============================================================
|
||
panel8.work_dir = work_dir
|
||
resolved = panel8._resolve_training_csv_from_workdir()
|
||
expected = step7_csv.replace('\\', '/')
|
||
assert resolved == expected, f"FAIL: 应优先 step7, 实际 {resolved!r}, 期望 {expected!r}"
|
||
print(f"[OK] Step8._resolve_training_csv_from_workdir() = {resolved}")
|
||
print(f" 优先级 1: Step 7 WQI 增强版 ✓")
|
||
|
||
# 模拟用户没跑过 Step 7（删掉 step7_csv 目录）— 但 10_WaterIndex_CSV 还在
|
||
import shutil
|
||
shutil.rmtree(os.path.join(work_dir, "7_Water_Quality_Indices"))
|
||
resolved = panel8._resolve_training_csv_from_workdir()
|
||
# 用户明确提到 10_WaterIndex_CSV 目录下的结果是有效来源，应作为第二优先级
|
||
expected_step10 = str(Path(step10_csv)).replace('\\', '/')
|
||
assert resolved == expected_step10, \
|
||
f"FAIL: 应取 10_WaterIndex_CSV/*training*, 实际 {resolved!r}, 期望 {expected_step10!r}"
|
||
print(f"[OK] Step 7 缺省时，回退到 10_WaterIndex_CSV/*training*: {resolved}")
|
||
print(f" 优先级 2: 10_WaterIndex_CSV/*training* ✓")
|
||
|
||
# 再删 10_WaterIndex_CSV 才回退到 Step 6
|
||
shutil.rmtree(os.path.join(work_dir, "10_WaterIndex_CSV"))
|
||
resolved = panel8._resolve_training_csv_from_workdir()
|
||
assert resolved == step6_csv.replace('\\', '/'), \
|
||
f"FAIL: 应回退 step6, 实际 {resolved!r}, 期望 {step6_csv!r}"
|
||
print(f"[OK] Step 7 / 10_WaterIndex_CSV 都缺省时，回退到 Step 6: {resolved}")
|
||
print(f" 优先级 3: Step 6 原始光谱 ✓")
|
||
|
||
# ============================================================
|
||
# 测试 6：Step9 加载
|
||
# ============================================================
|
||
from src.gui.panels.step9_ml_predict_panel import Step9MlPredictPanel
|
||
|
||
panel9 = Step9MlPredictPanel()
|
||
print(f"[OK] Step9MlPredictPanel 实例化成功")
|
||
|
||
# 重建 10_WaterIndex_CSV（WQI 测试集 fixture，test 5 中被删了）
|
||
os.makedirs(os.path.join(work_dir, "10_WaterIndex_CSV"), exist_ok=True)
|
||
wqi_test_csv = os.path.join(work_dir, "10_WaterIndex_CSV", "sampling_watercolor_summary.csv")
|
||
make_fake_csv(wqi_test_csv, 50, wqi_names)
|
||
new_time = (pd.Timestamp.now() - pd.Timedelta(hours=1)).timestamp()
|
||
os.utime(wqi_test_csv, (new_time, new_time))
|
||
|
||
# ============================================================
|
||
# 测试 7：Step9._resolve_latest_wqi_test_csv 优先 10_WaterIndex_CSV
|
||
# ============================================================
|
||
panel9.work_dir = work_dir
|
||
resolved = panel9._resolve_latest_wqi_test_csv()
|
||
# 应该是 sampling_watercolor_summary.csv（mtime 较新）
|
||
expected = wqi_test_csv.replace('\\', '/')
|
||
assert resolved == expected, \
|
||
f"FAIL: 应取 mtime 最新的 wqi 测试集, 实际 {resolved!r}, 期望 {expected!r}"
|
||
print(f"[OK] Step9._resolve_latest_wqi_test_csv() = {resolved}")
|
||
print(f" 优先级：10_WaterIndex_CSV/sampling_watercolor_summary.csv ✓")
|
||
|
||
# ============================================================
|
||
# 测试 8：Step9.update_from_config 优先级：WQI CSV > Step4 raw
|
||
# ============================================================
|
||
# 注：Step 4 回退路径依赖 main_window._panel_factory（offline 测试无 main_window，
|
||
# 故仅验证 WQI CSV 优先级主路径；Step 4 路径与原代码同源，逻辑没变）
|
||
# 有 WQI 时优先 WQI
|
||
os.makedirs(os.path.join(work_dir, "10_WaterIndex_CSV"), exist_ok=True)
|
||
wqi_test_csv2 = os.path.join(work_dir, "10_WaterIndex_CSV", "sampling_watercolor_summary.csv")
|
||
make_fake_csv(wqi_test_csv2, 50, wqi_names)
|
||
panel9.update_from_config(work_dir=work_dir, pipeline=None)
|
||
path = panel9.sampling_csv_file.get_path()
|
||
assert wqi_test_csv2.replace('\\', '/') in path, \
|
||
f"FAIL: 应优先 WQI CSV, 实际 {path!r}"
|
||
print(f"[OK] Step9.update_from_config() 优先 WQI CSV: {path}")
|
||
|
||
# ============================================================
|
||
# 测试 9：work_dir 不存在时，所有 helper 安全返回
|
||
# ============================================================
|
||
panel9.work_dir = ""
|
||
assert panel9._resolve_latest_wqi_test_csv() == ""
|
||
print(f"[OK] work_dir 空时 _resolve_latest_wqi_test_csv 安全返回空串")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("ALL SMOKE TESTS PASSED")
|
||
print("=" * 60)
|